[MongoDB Storage] Handle "topology is closed" (#278)

* Handle MongoDB TopologyIsClosed errors by closing the process.

* Connect to storage database on startup.

* Add changeset.

Author: rkistner

.changeset/yellow-icons-cross.md

@@ -0,0 +1,9 @@
+---
+'@powersync/service-module-mongodb-storage': patch
+'@powersync/service-errors': patch
+'@powersync/service-core': patch
+'@powersync/lib-services-framework': patch
+'@powersync/lib-service-mongodb': patch
+---
+
+[MongoDB Storage] Handle connection errors on startup

libs/lib-services/src/system/LifeCycledSystem.ts

@@ -6,7 +6,9 @@
  * A System can contain anything but should offer a `start` and `stop` operation
  */
 
+import { ServiceError } from '@powersync/service-errors';
 import { container } from '../container.js';
+import { logger } from '../logger/Logger.js';
 
 export type LifecycleCallback<T> = (singleton: T) => Promise<void> | void;
 
@@ -46,4 +48,17 @@ export class LifeCycledSystem {
       await lifecycle.stop?.(lifecycle.component);
     }
   };
+
+  stopWithError = async (error: ServiceError) => {
+    try {
+      logger.error('Stopping process due to fatal error', error);
+      await this.stop();
+    } catch (e) {
+      logger.error('Error while stopping', e);
+    } finally {
+      // Custom error code to distinguish from other common errors
+      logger.warn(`Exiting with code 151`);
+      process.exit(151);
+    }
+  };
 }

modules/module-mongodb-storage/src/storage/implementation/MongoStorageProvider.ts

@@ -1,5 +1,5 @@
 import * as lib_mongo from '@powersync/lib-service-mongodb';
-import { logger, ServiceAssertionError } from '@powersync/lib-services-framework';
+import { ErrorCode, logger, ServiceAssertionError, ServiceError } from '@powersync/lib-services-framework';
 import { storage } from '@powersync/service-core';
 import { MongoStorageConfig } from '../../types/types.js';
 import { MongoBucketStorage } from '../MongoBucketStorage.js';
@@ -26,6 +26,15 @@ export class MongoStorageProvider implements storage.BucketStorageProvider {
       maxPoolSize: resolvedConfig.storage.max_pool_size ?? 8
     });
 
+    let shuttingDown = false;
+
+    // Explicitly connect on startup.
+    // Connection errors during startup are typically not recoverable - we get topologyClosed.
+    // This helps to catch the error early, along with the cause, and before the process starts
+    // to serve API requests.
+    // Errors here will cause the process to exit.
+    await client.connect();
+
     const database = new PowerSyncMongo(client, { database: resolvedConfig.storage.database });
     const factory = new MongoBucketStorage(database, {
       // TODO currently need the entire resolved config due to this
@@ -34,12 +43,29 @@ export class MongoStorageProvider implements storage.BucketStorageProvider {
     return {
       storage: factory,
       shutDown: async () => {
+        shuttingDown = true;
         await factory[Symbol.asyncDispose]();
         await client.close();
       },
       tearDown: () => {
         logger.info(`Tearing down storage: ${database.db.namespace}...`);
         return database.db.dropDatabase();
+      },
+      onFatalError: (callback) => {
+        client.addListener('topologyClosed', () => {
+          // If we're shutting down, this is expected and we can ignore it.
+          if (!shuttingDown) {
+            // Unfortunately there is no simple way to catch the cause of this issue.
+            // It most commonly happens when the process fails to _ever_ connect - connection issues after
+            // the initial connection are usually recoverable.
+            callback(
+              new ServiceError({
+                code: ErrorCode.PSYNC_S2402,
+                description: 'MongoDB topology closed - failed to connect to MongoDB storage.'
+              })
+            );
+          }
+        });
       }
     } satisfies storage.ActiveStorage;
   }

packages/service-core/src/entry/cli-entry.ts

@@ -36,8 +36,8 @@ export function generateEntryProgram(startHandlers?: Record<utils.ServiceRunner,
       try {
         await entryProgram.parseAsync();
       } catch (e) {
-        logger.error('Fatal error', e);
-        process.exit(1);
+        logger.error('Fatal startup error - exiting with code 150.', e);
+        process.exit(150);
       }
     }
   };

packages/service-core/src/storage/StorageEngine.ts

@@ -1,14 +1,15 @@
-import { BaseObserver, logger } from '@powersync/lib-services-framework';
+import { BaseObserver, logger, ServiceError } from '@powersync/lib-services-framework';
 import { ResolvedPowerSyncConfig } from '../util/util-index.js';
-import { ActiveStorage, BucketStorageProvider } from './StorageProvider.js';
 import { BucketStorageFactory } from './BucketStorageFactory.js';
+import { ActiveStorage, BucketStorageProvider } from './StorageProvider.js';
 
 export type StorageEngineOptions = {
   configuration: ResolvedPowerSyncConfig;
 };
 
 export interface StorageEngineListener {
   storageActivated: (storage: BucketStorageFactory) => void;
+  storageFatalError: (error: ServiceError) => void;
 }
 
 export class StorageEngine extends BaseObserver<StorageEngineListener> {
@@ -47,6 +48,9 @@ export class StorageEngine extends BaseObserver<StorageEngineListener> {
       resolvedConfig: configuration
     });
     this.iterateListeners((cb) => cb.storageActivated?.(this.activeBucketStorage));
+    this.currentActiveStorage.onFatalError?.((error) => {
+      this.iterateListeners((cb) => cb.storageFatalError?.(error));
+    });
     logger.info(`Successfully activated storage: ${configuration.storage.type}.`);
     logger.info('Successfully started Storage Engine.');
   }

packages/service-core/src/storage/StorageProvider.ts

@@ -1,3 +1,4 @@
+import { ServiceError } from '@powersync/lib-services-framework';
 import * as util from '../util/util-index.js';
 import { BucketStorageFactory } from './BucketStorageFactory.js';
 
@@ -9,6 +10,8 @@ export interface ActiveStorage {
    *  Tear down / drop the storage permanently
    */
   tearDown(): Promise<boolean>;
+
+  onFatalError?(callback: (error: ServiceError) => void): void;
 }
 
 export interface GetStorageOptions {

packages/service-core/src/system/ServiceContext.ts

@@ -59,6 +59,12 @@ export class ServiceContextContainer implements ServiceContext {
     this.storageEngine = new storage.StorageEngine({
       configuration
     });
+    this.storageEngine.registerListener({
+      storageFatalError: (error) => {
+        // Propagate the error to the lifecycle engine
+        this.lifeCycleEngine.stopWithError(error);
+      }
+    });
 
     this.lifeCycleEngine.withLifecycle(this.storageEngine, {
       start: (storageEngine) => storageEngine.start(),

packages/service-errors/src/codes.ts

@@ -423,6 +423,11 @@ export enum ErrorCode {
    */
   PSYNC_S2401 = 'PSYNC_S2401',
 
+  /**
+   * Failed to connect to the MongoDB storage database.
+   */
+  PSYNC_S2402 = 'PSYNC_S2402',
+
   // ## PSYNC_S23xx: Sync API errors - Postgres Storage
 
   // ## PSYNC_S3xxx: Service configuration issues