[kbn/optimizer] throw errors into stream on invalid completion

packages/kbn-optimizer/src/common/rxjs_helpers.ts

@@ -24,9 +24,9 @@ type Operator<T1, T2> = (source: Rx.Observable<T1>) => Rx.Observable<T2>;
 type MapFn<T1, T2> = (item: T1, index: number) => T2;
 
 /**
- * Wrap an operator chain in a closure so that is can have some local
- * state. The `fn` is called each time the final observable is
- * subscribed so the pipeline/closure is setup for each subscription.
+ * Wrap an operator chain in a closure so that it can have some local
+ * state. The `fn` is called each time the returned observable is
+ * subscribed; the closure is recreated for each subscription.
  */
 export const pipeClosure = <T1, T2>(fn: Operator<T1, T2>): Operator<T1, T2> => {
   return (source: Rx.Observable<T1>) => {

packages/kbn-optimizer/src/optimizer/handle_optimizer_completion.test.ts

@@ -0,0 +1,104 @@
+/*
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import * as Rx from 'rxjs';
+import { REPO_ROOT } from '@kbn/dev-utils';
+
+import { Update } from '../common';
+
+import { OptimizerState } from './optimizer_reducer';
+import { OptimizerConfig } from './optimizer_config';
+import { handleOptimizerCompletion } from './handle_optimizer_completion';
+import { toArray } from 'rxjs/operators';
+
+const createUpdate$ = (phase: OptimizerState['phase']) =>
+  Rx.of<Update<any, OptimizerState>>({
+    state: {
+      phase,
+      compilerStates: [],
+      durSec: 0,
+      offlineBundles: [],
+      onlineBundles: [],
+      startTime: Date.now(),
+    },
+  });
+
+const config = (watch?: boolean) =>
+  OptimizerConfig.create({
+    repoRoot: REPO_ROOT,
+    watch,
+  });
+const collect = <T>(stream: Rx.Observable<T>): Promise<T[]> => stream.pipe(toArray()).toPromise();
+
+it('errors if the optimizer completes when in watch mode', async () => {
+  const update$ = createUpdate$('success');
+
+  await expect(
+    collect(update$.pipe(handleOptimizerCompletion(config(true))))
+  ).rejects.toThrowErrorMatchingInlineSnapshot(
+    `"optimizer unexpectedly completed when in watch mode"`
+  );
+});
+
+it('errors if the optimizer completes in phase "issue"', async () => {
+  const update$ = createUpdate$('issue');
+
+  await expect(
+    collect(update$.pipe(handleOptimizerCompletion(config())))
+  ).rejects.toThrowErrorMatchingInlineSnapshot(`"webpack issue"`);
+});
+
+it('errors if the optimizer completes in phase "initializing"', async () => {
+  const update$ = createUpdate$('initializing');
+
+  await expect(
+    collect(update$.pipe(handleOptimizerCompletion(config())))
+  ).rejects.toThrowErrorMatchingInlineSnapshot(
+    `"optimizer unexpectedly exit in phase \\"initializing\\""`
+  );
+});
+
+it('errors if the optimizer completes in phase "reallocating"', async () => {
+  const update$ = createUpdate$('reallocating');
+
+  await expect(
+    collect(update$.pipe(handleOptimizerCompletion(config())))
+  ).rejects.toThrowErrorMatchingInlineSnapshot(
+    `"optimizer unexpectedly exit in phase \\"reallocating\\""`
+  );
+});
+
+it('errors if the optimizer completes in phase "running"', async () => {
+  const update$ = createUpdate$('running');
+
+  await expect(
+    collect(update$.pipe(handleOptimizerCompletion(config())))
+  ).rejects.toThrowErrorMatchingInlineSnapshot(
+    `"optimizer unexpectedly exit in phase \\"running\\""`
+  );
+});
+
+it('passes through errors on the source stream', async () => {
+  const error = new Error('foo');
+  const update$ = Rx.throwError(error);
+
+  await expect(collect(update$.pipe(handleOptimizerCompletion(config())))).rejects.toThrowError(
+    error
+  );
+});

packages/kbn-optimizer/src/optimizer/handle_optimizer_completion.ts

@@ -0,0 +1,56 @@
+/*
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import * as Rx from 'rxjs';
+import { tap } from 'rxjs/operators';
+import { createFailError } from '@kbn/dev-utils';
+
+import { pipeClosure, Update } from '../common';
+
+import { OptimizerState } from './optimizer_reducer';
+import { OptimizerConfig } from './optimizer_config';
+
+export function handleOptimizerCompletion(config: OptimizerConfig) {
+  return pipeClosure((source$: Rx.Observable<Update<any, OptimizerState>>) => {
+    let prevState: OptimizerState | undefined;
+
+    return source$.pipe(
+      tap({
+        next: update => {
+          prevState = update.state;
+        },
+        complete: () => {
+          if (config.watch) {
+            throw new Error('optimizer unexpectedly completed when in watch mode');
+          }
+
+          if (prevState?.phase === 'success') {
+            return;
+          }
+
+          if (prevState?.phase === 'issue') {
+            throw createFailError('webpack issue');
+          }
+
+          throw new Error(`optimizer unexpectedly exit in phase "${prevState?.phase}"`);
+        },
+      })
+    );
+  });
+}

packages/kbn-optimizer/src/optimizer/index.ts

@@ -24,3 +24,4 @@ export * from './cache_keys';
 export * from './watch_bundles_for_changes';
 export * from './run_workers';
 export * from './bundle_cache';
+export * from './handle_optimizer_completion';

packages/kbn-optimizer/src/run_optimizer.ts

@@ -32,6 +32,7 @@ import {
   runWorkers,
   OptimizerInitializedEvent,
   createOptimizerReducer,
+  handleOptimizerCompletion,
 } from './optimizer';
 
 export type OptimizerUpdate = Update<OptimizerEvent, OptimizerState>;
@@ -77,6 +78,7 @@ export function runOptimizer(config: OptimizerConfig) {
         },
         createOptimizerReducer(config)
       );
-    })
+    }),
+    handleOptimizerCompletion(config)
   );
 }