Handle plugin errors

packages/core/src/Source.ts

@@ -9,6 +9,7 @@ import type {
   IVolumeImmutable,
   MutableData,
   Plugin,
+  PluginErrors,
   PluginModuleDefinition,
   Serialiser,
   SerialiserModuleDefinition,
@@ -36,6 +37,7 @@ export default class Source {
   #ignorePages: string[];
   #workflows: SourceWorkflow[] = [];
   #schedule: SourceSchedule;
+  #pluginErrors: any = {};
 
   config: MutableData<Record<string, unknown>>;
   serialiser: Serialiser;
@@ -188,12 +190,15 @@ export default class Source {
 
   async start() {
     this.serialiser = await bindSerialiser(this.#serialisers);
-    this.#pluginApi = await bindPluginMethods(this.#plugins);
+    this.#pluginApi = await bindPluginMethods(this.#plugins, this.trackPluginErrors.bind(this));
     this.#worker = this.#createWorker();
     this.#worker.on(EVENT.UPDATE, async ({ data: { pages, symlinks, data } }) => {
       this.config = createConfig(data);
       this.#emitter.emit(EVENT.UPDATE, { pages, symlinks, data });
     });
+    this.#worker.on(EVENT.TRACK, async ({ data: { errors, lifecycleMethod } }) => {
+      this.#pluginErrors[`${lifecycleMethod}`] = errors;
+    });
   }
 
   async isOwner(filePath: string) {
@@ -230,7 +235,12 @@ export default class Source {
   }
 
   async restart() {
+    console.log(this.#pluginErrors);
     this.stop();
     await this.start();
   }
+
+  trackPluginErrors(errors: PluginErrors, lifecycleMethod: string) {
+    this.#pluginErrors[`${lifecycleMethod}`] = errors;
+  }
 }

packages/core/src/WorkerSubscription.ts

@@ -7,7 +7,8 @@ export enum EVENT {
   ERROR = 'ERROR',
   EXIT = 'EXIT',
   START = 'START',
-  UPDATE = 'UPDATE'
+  UPDATE = 'UPDATE',
+  TRACK = 'TRACK'
 }
 
 const textDecoder = new TextDecoder('utf8');
@@ -52,9 +53,11 @@ export default class WorkerSubscription {
     return () => this.#emitter.off(type, handler);
   }
 
-  #onNext = ({ type, data }: { data: Uint8Array; type: 'init' | 'message' }) => {
+  #onNext = ({ type, data }: { data: Uint8Array; type: 'init' | 'message' | 'track' }) => {
     if (type === 'message') {
       this.#emitter.emit(EVENT.UPDATE, { data: JSON.parse(textDecoder.decode(data)) });
+    } else if (type === 'track') {
+      this.#emitter.emit(EVENT.TRACK, { data: JSON.parse(textDecoder.decode(data)) });
     } else if (type === 'init') {
       this.#emitter.emit(EVENT.START);
       if (data) {

packages/core/src/plugin/createPluginAPI.ts

@@ -4,8 +4,10 @@ import type {
   Page,
   Plugin,
   PluginModuleDefinition,
-  Serialiser
+  Serialiser,
+  TrackPluginErrorCallback
 } from '@jpmorganchase/mosaic-types';
+import PluginError from '@jpmorganchase/mosaic-plugins/PluginError';
 
 import loadDefinitionModules from './loadDefinitionModules.js';
 import pluginRunner from './pluginRunner.js';
@@ -43,7 +45,8 @@ function createProxyBaseAPI<ConfigData>(): Plugin<Page, ConfigData> {
 }
 
 export default async function createPluginAPI<PluginInput, ConfigData = Record<string, unknown>>(
-  plugins: PluginModuleDefinition[]
+  plugins: PluginModuleDefinition[],
+  track?: TrackPluginErrorCallback
 ): Promise<Plugin<Page, ConfigData>> {
   const loadedPlugins: LoadedPlugin[] | LoadedSerialiser[] = await loadDefinitionModules(plugins);
   const baseObj = createProxyBaseAPI<ConfigData>();
@@ -64,6 +67,9 @@ export default async function createPluginAPI<PluginInput, ConfigData = Record<s
               ...args
             );
           } catch (e) {
+            if (e instanceof PluginError) {
+              throw e;
+            }
             throw new Error(e);
           }
           return result;
@@ -73,14 +79,19 @@ export default async function createPluginAPI<PluginInput, ConfigData = Record<s
         return async (...args: [PluginInput, Record<string, unknown>]) => {
           let result;
           try {
-            result = await pluginRunner(
+            const { result: pluginResult, errors } = await pluginRunner(
               {
                 loadedPlugins: loadedPlugins as LoadedPlugin[],
                 lifecycleName: String(propOrLifecycleName)
               },
               ...args
             );
+            track?.(errors, String(propOrLifecycleName));
+            result = pluginResult;
           } catch (e) {
+            if (e instanceof PluginError) {
+              throw e;
+            }
             throw new Error(e);
           }
           return result;

packages/core/src/plugin/index.ts

@@ -5,7 +5,8 @@ import type {
   Page,
   Plugin,
   PluginModuleDefinition,
-  Serialiser
+  Serialiser,
+  TrackPluginErrorCallback
 } from '@jpmorganchase/mosaic-types';
 
 import loadPlugins from './createPluginAPI.js';
@@ -35,58 +36,27 @@ export async function bindSerialiser(serialisers): Promise<Serialiser> {
   };
 }
 
-export async function bindPluginMethods(plugins: PluginModuleDefinition[]): Promise<Plugin> {
-  const pluginApi = await loadPlugins<Page[] | TDataOut | IVolumeMutable>(plugins);
+export async function bindPluginMethods(
+  plugins: PluginModuleDefinition[],
+  track?: TrackPluginErrorCallback
+): Promise<Plugin> {
+  const pluginApi = await loadPlugins<Page[] | TDataOut | IVolumeMutable>(plugins, track);
 
   return {
     async $afterSource(pages, args) {
-      let result;
-      try {
-        result = await pluginApi.$afterSource(pages, args);
-      } catch (e) {
-        throw new Error(e);
-      }
+      const result = await pluginApi.$afterSource(pages, args);
       return result;
     },
     async shouldClearCache(lastAfterUpdateReturn, args) {
-      let result;
-      try {
-        result = await pluginApi.shouldClearCache(lastAfterUpdateReturn, args);
-      } catch (e) {
-        throw new Error(e);
-      }
+      const result = await pluginApi.shouldClearCache(lastAfterUpdateReturn, args);
       return result;
     },
     async afterUpdate(mutableFilesystem, args) {
-      let result;
-      try {
-        result = await pluginApi.afterUpdate(mutableFilesystem, args);
-      } catch (e) {
-        throw new Error(e);
-      }
+      const result = await pluginApi.afterUpdate(mutableFilesystem, args);
       return result;
     },
     async $beforeSend(mutableFilesystem, args) {
-      let result;
-      try {
-        result = await pluginApi.$beforeSend(mutableFilesystem, args);
-      } catch (e) {
-        throw new Error(e);
-      }
-      return result;
-    },
-    async saveContent(
-      filePath: string,
-      data: unknown,
-      sourceOptions: Record<string, unknown>,
-      args
-    ) {
-      let result;
-      try {
-        result = await pluginApi.saveContent(filePath, data, sourceOptions, args);
-      } catch (e) {
-        throw new Error(e);
-      }
+      const result = await pluginApi.$beforeSend(mutableFilesystem, args);
       return result;
     }
   };

packages/core/src/plugin/pluginRunner.ts

@@ -1,13 +1,14 @@
-import type { LoadedPlugin } from '@jpmorganchase/mosaic-types';
-
-import PluginError from '../PluginError.js';
+import path from 'node:path';
+import type { LoadedPlugin, PluginErrors } from '@jpmorganchase/mosaic-types';
+import PluginError from '@jpmorganchase/mosaic-plugins/PluginError';
 
 export default async function pluginRunner(
   { loadedPlugins, lifecycleName }: { loadedPlugins: LoadedPlugin[]; lifecycleName: string },
   input,
   ...args
 ) {
   let transformedInput = input;
+  const pluginErrors: PluginErrors = [];
 
   for (const plugin of loadedPlugins) {
     try {
@@ -20,48 +21,47 @@ export default async function pluginRunner(
         continue;
       }
 
-      // console.debug(
-      //   `[Mosaic] Applying plugin method \`${lifecycleName}\`${lifecycleName.startsWith('$') ? ' (in a child worker)' : ''} for '${plugin.modulePath}'.`
-      // );
-
       // eslint-disable-next-line no-await-in-loop
       const result = await plugin[lifecycleName](
         lifecycleName === '$afterSource' ? transformedInput : input,
         ...args,
         plugin.options
       );
 
-      if (
-        result &&
-        lifecycleName !== '$afterSource' &&
-        lifecycleName !== 'shouldClearCache' &&
-        lifecycleName !== 'saveContent'
-      ) {
+      if (result && lifecycleName !== '$afterSource' && lifecycleName !== 'shouldClearCache') {
         console.warn(
           `[Mosaic] \`${lifecycleName}\` plugin should not return a value - this lifecycle phase expects mutation to occur directly on the filesystem instance. This will be ignored.`
         );
       }
 
-      if (lifecycleName === 'saveContent' && !result) {
-        console.warn(
-          `[Mosaic] \`${lifecycleName}\` plugin returned a falsy value - this result has been discarded.`
-        );
-        // eslint-disable-next-line no-continue
-        continue;
-      }
-
       transformedInput = result;
-    } catch (e) {
-      // This check will stop nested errors from ending up with multiple 'Plugin X threw an exception' headers from
-      // being prefixed to the messages
-      if (e instanceof PluginError) {
-        throw e;
-      }
-      throw new PluginError(
-        `Plugin '${plugin.modulePath}' threw an exception running \`${lifecycleName}\`. See below:
-  ${e.stack}`
+    } catch (exception) {
+      const pluginName = path.posix.basename(
+        plugin.modulePath,
+        path.posix.extname(plugin.modulePath)
+      );
+      console.log(
+        `[Mosaic][Plugin] '${pluginName}' threw an exception running \`${lifecycleName}\``
       );
+      console.log(`[Mosaic][Plugin] stack: ${exception.stack}`);
+
+      if (exception instanceof PluginError) {
+        exception.lifecycleMethod = lifecycleName;
+        exception.pluginModulePath = plugin.modulePath;
+        exception.pluginName = pluginName;
+        pluginErrors.push(exception);
+      } else {
+        // create a new plugin error
+        const pluginError = new PluginError(exception.message);
+        pluginError.pluginModulePath = plugin.modulePath;
+        pluginError.lifecycleMethod = lifecycleName;
+        pluginError.pluginName = pluginName;
+        pluginErrors.push(pluginError);
+      }
+
+      // process this lifecycle for the rest of the loaded plugins
+      continue;
     }
   }
-  return transformedInput;
+  return { result: transformedInput, errors: pluginErrors };
 }

packages/core/src/plugin/serialiserRunner.ts

@@ -1,7 +1,6 @@
 import type { LoadedSerialiser } from '@jpmorganchase/mosaic-types';
 import path from 'node:path';
-
-import PluginError from '../PluginError.js';
+import PluginError from '@jpmorganchase/mosaic-plugins/PluginError';
 
 const isPathWithExtension = (filePath: string) => {
   return typeof filePath === 'string' && path.extname(filePath).length > 1;
@@ -31,17 +30,18 @@ export default async function serialiserRunner(
       const result = await serialiser[serialiserMethod](fullPath, ...args, serialiser.options);
       return result;
     } catch (e) {
-      // This check will stop nested errors from ending up with multiple 'Plugin X threw an exception' headers from
-      // being prefixed to the messages
       if (e instanceof PluginError) {
+        e.lifecycleMethod = serialiserMethod;
+        e.pluginModulePath = serialiser.modulePath;
         throw e;
       }
-      throw new PluginError(
-        `Serialiser '${serialiser.modulePath}' threw an exception calling \`serialiser.${serialiserMethod}\` on '${fullPath}'.
-
-  Is this the correct serialiser to use for this file type? See error below:
-  ${e.stack}`
-      );
+
+      // create a new plugin error
+      const pluginError = new PluginError(e.message);
+      pluginError.stack = e.stack;
+      pluginError.pluginModulePath = serialiser.modulePath;
+      pluginError.lifecycleMethod = serialiserMethod;
+      throw pluginError;
     }
   }
   throw new Error(`Could not find a suitable serialiser for file '${fullPath}'.`);

packages/core/src/worker/Source.worker.ts

@@ -5,14 +5,22 @@ import fs from 'fs';
 import { DirectoryJSON, Volume } from 'memfs';
 import { switchMap, tap } from 'rxjs';
 
-import type { Page, WorkerData } from '@jpmorganchase/mosaic-types';
+import type { Page, PluginErrors, WorkerData } from '@jpmorganchase/mosaic-types';
 
 import FileAccess from '../filesystems/FileAccess.js';
 import MutableVolume from '../filesystems/MutableVolume.js';
 import createConfig from '../helpers/createConfig.js';
 import { bindPluginMethods, bindSerialiser } from '../plugin/index.js';
 import createSourceObservable from './helpers/createSourceObservable.js';
 
+const trackPluginErrors = (errors: PluginErrors, lifecycleMethod: string) => {
+  const errorsBuffer = Buffer.from(JSON.stringify({ errors, lifecycleMethod }));
+  parentPort.postMessage({
+    type: 'track',
+    data: errorsBuffer
+  });
+};
+
 const workerData: WorkerData<{ cache: boolean }> = unTypedWorkerData;
 
 if (isMainThread) {
@@ -22,7 +30,7 @@ if (isMainThread) {
   let config;
   let filesystem;
   const serialiser = await bindSerialiser(workerData.serialisers);
-  const pluginApi = await bindPluginMethods(workerData.plugins);
+  const pluginApi = await bindPluginMethods(workerData.plugins, trackPluginErrors);
   const cachePath = path.join(process.cwd(), '.tmp', '.cache', `${workerData.name}.json`);
 
   if (workerData.options.cache !== false) {

packages/types/src/Plugin.ts

@@ -12,6 +12,36 @@ import type { MutableData, ImmutableData } from './MutableData.js';
 export type LoadedPlugin = Partial<Plugin<Page>> & PluginModuleDefinition;
 export type LifecycleMethod = keyof Plugin<Page>;
 
+export type PluginErrorDescriptor = {
+  /**
+   * The plugin lifecycle method running when the error occurred
+   */
+  lifecycleMethod: string;
+  /**
+   * The path to the loaded plugin module
+   */
+  pluginModulePath: string;
+  /**
+   * The name of the plugin
+   */
+  pluginName: string;
+  /**
+   * The name of the error
+   */
+  name: string;
+  /**
+   * The path of the file that triggered the plugin error
+   */
+  fullPath?: string;
+};
+
+export type PluginErrors = Array<PluginErrorDescriptor>;
+
+export type TrackPluginErrorCallback = (
+  errors: Array<PluginErrorDescriptor>,
+  lifecycleMethod: string
+) => void;
+
 /**
  * Plugins are lifecycle-based hooks that are called on every source at different stages.
  * Consumers will never need to invoke a lifecycle method; but for technical clarity - when a lifecycle method is called,