Merge pull request #8 from journeyapps-platform/feat/rework-teardown

Implement powersync instance teardown functionality

Author: Rentacookie

modules/module-postgres/src/module/PostgresModule.ts

@@ -1,11 +1,20 @@
-import { api, auth, ConfigurationFileSyncRulesProvider, replication, system } from '@powersync/service-core';
+import {
+  api,
+  auth,
+  ConfigurationFileSyncRulesProvider,
+  replication,
+  system,
+  TearDownOptions
+} from '@powersync/service-core';
 import * as jpgwire from '@powersync/service-jpgwire';
 import * as types from '../types/types.js';
 import { PostgresRouteAPIAdapter } from '../api/PostgresRouteAPIAdapter.js';
 import { SupabaseKeyCollector } from '../auth/SupabaseKeyCollector.js';
 import { WalStreamReplicator } from '../replication/WalStreamReplicator.js';
 import { ConnectionManagerFactory } from '../replication/ConnectionManagerFactory.js';
 import { PostgresErrorRateLimiter } from '../replication/PostgresErrorRateLimiter.js';
+import { cleanUpReplicationSlot } from '../replication/replication-utils.js';
+import { PgManager } from '../replication/PgManager.js';
 
 export class PostgresModule extends replication.ReplicationModule<types.PostgresConnectionConfig> {
   constructor() {
@@ -24,29 +33,28 @@ export class PostgresModule extends replication.ReplicationModule<types.Postgres
       this.registerSupabaseAuth(context);
     }
 
-    jpgwire.setMetricsRecorder({
-      addBytesRead(bytes) {
-        context.metrics.data_replicated_bytes.add(bytes);
-      }
-    });
+    if (context.metrics) {
+      jpgwire.setMetricsRecorder({
+        addBytesRead(bytes) {
+          context.metrics!.data_replicated_bytes.add(bytes);
+        }
+      });
+    }
   }
 
-  protected createRouteAPIAdapter(decodedConfig: types.PostgresConnectionConfig): api.RouteAPI {
-    return new PostgresRouteAPIAdapter(this.resolveConfig(decodedConfig));
+  protected createRouteAPIAdapter(): api.RouteAPI {
+    return new PostgresRouteAPIAdapter(this.resolveConfig(this.decodedConfig!));
   }
 
-  protected createReplicator(
-    decodedConfig: types.PostgresConnectionConfig,
-    context: system.ServiceContext
-  ): replication.AbstractReplicator {
-    const normalisedConfig = this.resolveConfig(decodedConfig);
-    const connectionFactory = new ConnectionManagerFactory(normalisedConfig);
+  protected createReplicator(context: system.ServiceContext): replication.AbstractReplicator {
+    const normalisedConfig = this.resolveConfig(this.decodedConfig!);
     const syncRuleProvider = new ConfigurationFileSyncRulesProvider(context.configuration.sync_rules);
+    const connectionFactory = new ConnectionManagerFactory(normalisedConfig);
 
     return new WalStreamReplicator({
       id: this.getDefaultId(normalisedConfig.database),
       syncRuleProvider: syncRuleProvider,
-      storageEngine: context.storage,
+      storageEngine: context.storageEngine,
       connectionFactory: connectionFactory,
       rateLimiter: new PostgresErrorRateLimiter()
     });
@@ -62,7 +70,29 @@ export class PostgresModule extends replication.ReplicationModule<types.Postgres
     };
   }
 
-  async teardown(): Promise<void> {}
+  async teardown(options: TearDownOptions): Promise<void> {
+    const normalisedConfig = this.resolveConfig(this.decodedConfig!);
+    const connectionManager = new PgManager(normalisedConfig, {
+      idleTimeout: 30_000,
+      maxSize: 1
+    });
+
+    try {
+      if (options.syncRules) {
+        // TODO: In the future, once we have more replication types, we will need to check if these syncRules are for Postgres
+        for (let syncRules of options.syncRules) {
+          try {
+            await cleanUpReplicationSlot(syncRules.slot_name, connectionManager.pool);
+          } catch (e) {
+            // Not really much we can do here for failures, most likely the database is no longer accessible
+            this.logger.warn(`Failed to fully clean up Postgres replication slot: ${syncRules.slot_name}`, e);
+          }
+        }
+      }
+    } finally {
+      await connectionManager.end();
+    }
+  }
 
   // TODO: This should rather be done by registering the key collector in some kind of auth engine
   private registerSupabaseAuth(context: system.ServiceContextContainer) {

modules/module-postgres/src/replication/WalStreamReplicationJob.ts

@@ -4,14 +4,15 @@ import { PgManager } from './PgManager.js';
 
 import { replication } from '@powersync/service-core';
 import { ConnectionManagerFactory } from './ConnectionManagerFactory.js';
+import { cleanUpReplicationSlot } from './replication-utils.js';
 
 export interface WalStreamReplicationJobOptions extends replication.AbstractReplicationJobOptions {
   connectionFactory: ConnectionManagerFactory;
 }
 
 export class WalStreamReplicationJob extends replication.AbstractReplicationJob {
   private connectionFactory: ConnectionManagerFactory;
-  private connectionManager: PgManager;
+  private readonly connectionManager: PgManager;
 
   constructor(options: WalStreamReplicationJobOptions) {
     super(options);
@@ -24,15 +25,14 @@ export class WalStreamReplicationJob extends replication.AbstractReplicationJob
   }
 
   async cleanUp(): Promise<void> {
-    this.logger.info(`Cleaning up replication slot: ${this.slot_name}`);
-
+    const connectionManager = this.connectionFactory.create({
+      idleTimeout: 30_000,
+      maxSize: 1
+    });
     try {
-      await this.connectionManager.pool.query({
-        statement: 'SELECT pg_drop_replication_slot(slot_name) FROM pg_replication_slots WHERE slot_name = $1',
-        params: [{ type: 'varchar', value: this.slot_name }]
-      });
+      await cleanUpReplicationSlot(this.slotName, connectionManager.pool);
     } finally {
-      await this.connectionManager.end();
+      await connectionManager.end();
     }
   }
 
@@ -56,7 +56,7 @@ export class WalStreamReplicationJob extends replication.AbstractReplicationJob
     }
   }
 
-  get slot_name() {
+  get slotName() {
     return this.options.storage.slot_name;
   }
 
@@ -67,14 +67,14 @@ export class WalStreamReplicationJob extends replication.AbstractReplicationJob
       // Fatal exception
       container.reporter.captureException(e, {
         metadata: {
-          replication_slot: this.slot_name
+          replication_slot: this.slotName
         }
       });
-      this.logger.error(`Replication failed on ${this.slot_name}`, e);
+      this.logger.error(`Replication failed on ${this.slotName}`, e);
 
       if (e instanceof MissingReplicationSlotError) {
         // This stops replication on this slot, and creates a new slot
-        await this.options.storage.factory.slotRemoved(this.slot_name);
+        await this.options.storage.factory.slotRemoved(this.slotName);
       }
     } finally {
       this.abortController.abort();
@@ -108,7 +108,7 @@ export class WalStreamReplicationJob extends replication.AbstractReplicationJob
       const stream = new WalStream({
         abort_signal: this.abortController.signal,
         storage: this.options.storage,
-        connections: this.connectionManager
+        connections: connectionManager
       });
       await stream.replicate();
     } catch (e) {
@@ -142,7 +142,7 @@ export class WalStreamReplicationJob extends replication.AbstractReplicationJob
         // Report the error if relevant, before retrying
         container.reporter.captureException(e, {
           metadata: {
-            replication_slot: this.slot_name
+            replication_slot: this.slotName
           }
         });
         // This sets the retry delay

modules/module-postgres/src/replication/WalStreamReplicator.ts

@@ -1,8 +1,9 @@
-import { AbstractReplicatorOptions, replication } from '@powersync/service-core';
+import { storage, replication } from '@powersync/service-core';
 import { WalStreamReplicationJob } from './WalStreamReplicationJob.js';
 import { ConnectionManagerFactory } from './ConnectionManagerFactory.js';
+import { cleanUpReplicationSlot } from './replication-utils.js';
 
-export interface WalStreamReplicatorOptions extends AbstractReplicatorOptions {
+export interface WalStreamReplicatorOptions extends replication.AbstractReplicatorOptions {
   connectionFactory: ConnectionManagerFactory;
 }
 
@@ -23,6 +24,19 @@ export class WalStreamReplicator extends replication.AbstractReplicator<WalStrea
     });
   }
 
+  async cleanUp(syncRulesStorage: storage.SyncRulesBucketStorage): Promise<void> {
+    const connectionManager = this.connectionFactory.create({
+      idleTimeout: 30_000,
+      maxSize: 1
+    });
+    try {
+      // TODO: Slot_name will likely have to come from a different source in the future
+      await cleanUpReplicationSlot(syncRulesStorage.slot_name, connectionManager.pool);
+    } finally {
+      await connectionManager.end();
+    }
+  }
+
   async stop(): Promise<void> {
     await super.stop();
     await this.connectionFactory.shutdown();

modules/module-postgres/src/replication/replication-utils.ts

@@ -7,6 +7,7 @@ import * as sync_rules from '@powersync/service-sync-rules';
 import * as service_types from '@powersync/service-types';
 import * as pg_utils from '../utils/pgwire_utils.js';
 import * as util from '../utils/pgwire_utils.js';
+import { logger } from '@powersync/lib-services-framework';
 
 export interface ReplicaIdentityResult {
   replicationColumns: storage.ColumnDescriptor[];
@@ -317,3 +318,12 @@ export async function getDebugTableInfo(options: GetDebugTableInfoOptions): Prom
     ) as service_types.ReplicationError[]
   };
 }
+
+export async function cleanUpReplicationSlot(slotName: string, db: pgwire.PgClient): Promise<void> {
+  logger.info(`Cleaning up Postgres replication slot: ${slotName}...`);
+
+  await db.query({
+    statement: 'SELECT pg_drop_replication_slot(slot_name) FROM pg_replication_slots WHERE slot_name = $1',
+    params: [{ type: 'varchar', value: slotName }]
+  });
+}

packages/service-core/src/api/diagnostics.ts

@@ -3,7 +3,7 @@ import { DEFAULT_TAG, SourceTableInterface, SqlSyncRules } from '@powersync/serv
 import { SyncRulesStatus, TableInfo } from '@powersync/service-types';
 
 import * as storage from '../storage/storage-index.js';
-import * as system from '../system/system-index.js';
+import { RouteAPI } from './RouteAPI.js';
 
 export interface DiagnosticsOptions {
   /**
@@ -27,7 +27,8 @@ export interface DiagnosticsOptions {
 export const DEFAULT_DATASOURCE_ID = 'default';
 
 export async function getSyncRulesStatus(
-  serviceContext: system.ServiceContext,
+  bucketStorage: storage.BucketStorageFactory,
+  apiHandler: RouteAPI,
   sync_rules: storage.PersistedSyncRulesContent | null,
   options: DiagnosticsOptions
 ): Promise<SyncRulesStatus | undefined> {
@@ -52,24 +53,15 @@ export async function getSyncRulesStatus(
     };
   }
 
-  const {
-    storage: { activeBucketStorage }
-  } = serviceContext;
-  const api = serviceContext.routerEngine.getAPI();
-
-  const systemStorage = live_status ? activeBucketStorage.getInstance(persisted) : undefined;
+  const systemStorage = live_status ? bucketStorage.getInstance(persisted) : undefined;
   const status = await systemStorage?.getStatus();
   let replication_lag_bytes: number | undefined = undefined;
 
   let tables_flat: TableInfo[] = [];
 
   if (check_connection) {
-    if (!api) {
-      throw new Error('No connection configured');
-    }
-
     const source_table_patterns = rules.getSourceTables();
-    const resolved_tables = await api.getDebugTablesInfo(source_table_patterns, rules);
+    const resolved_tables = await apiHandler.getDebugTablesInfo(source_table_patterns, rules);
     tables_flat = resolved_tables.flatMap((info) => {
       if (info.table) {
         return [info.table];
@@ -82,7 +74,7 @@ export async function getSyncRulesStatus(
 
     if (systemStorage) {
       try {
-        replication_lag_bytes = await api.getReplicationLag(systemStorage.slot_name);
+        replication_lag_bytes = await apiHandler.getReplicationLag(systemStorage.slot_name);
       } catch (e) {
         // Ignore
         logger.warn(`Unable to get replication lag`, e);
@@ -136,15 +128,15 @@ export async function getSyncRulesStatus(
     })
   );
 
-  const sourceConfig = await api?.getSourceConfig();
-  const tag = sourceConfig?.tag ?? DEFAULT_TAG;
+  const sourceConfig = await apiHandler.getSourceConfig();
+  const tag = sourceConfig.tag ?? DEFAULT_TAG;
 
   return {
     content: include_content ? sync_rules.sync_rules_content : undefined,
     connections: [
       {
-        id: sourceConfig?.id ?? DEFAULT_DATASOURCE_ID,
-        tag: sourceConfig?.tag ?? DEFAULT_TAG,
+        id: sourceConfig.id ?? DEFAULT_DATASOURCE_ID,
+        tag: sourceConfig.tag ?? DEFAULT_TAG,
         slot_name: sync_rules.slot_name,
         initial_replication_done: status?.snapshot_done ?? false,
         // TODO: Rename?

packages/service-core/src/api/schema.ts

@@ -2,7 +2,7 @@ import { internal_routes } from '@powersync/service-types';
 
 import * as api from '../api/api-index.js';
 
-export async function getConnectionsSchema(api: api.RouteAPI | null): Promise<internal_routes.GetSchemaResponse> {
+export async function getConnectionsSchema(api: api.RouteAPI): Promise<internal_routes.GetSchemaResponse> {
   if (!api) {
     return {
       connections: []

packages/service-core/src/entry/commands/teardown-action.ts

@@ -12,7 +12,7 @@ export function registerTearDownAction(program: Command) {
 
   return teardownCommand
     .argument('[ack]', 'Type `TEARDOWN` to confirm teardown should occur')
-    .description('Terminate all replicating sync rules, deleting the replication slots')
+    .description('Terminate all replicating sync rules, clear remote configuration and remove all data')
     .action(async (ack, options) => {
       if (ack !== 'TEARDOWN') {
         throw new Error('TEARDOWN was not acknowledged.');

packages/service-core/src/modules/AbstractModule.ts

@@ -1,6 +1,14 @@
 import { ServiceContextContainer } from '../system/ServiceContext.js';
 import { logger } from '@powersync/lib-services-framework';
 import winston from 'winston';
+import { PersistedSyncRulesContent } from '../storage/BucketStorage.js';
+
+export interface TearDownOptions {
+  /**
+   *  If required, tear down any configuration/state for the specific sync rules
+   */
+  syncRules?: PersistedSyncRulesContent[];
+}
 
 export interface AbstractModuleOptions {
   name: string;
@@ -19,9 +27,9 @@ export abstract class AbstractModule {
   public abstract initialize(context: ServiceContextContainer): Promise<void>;
 
   /**
-   *  Terminate and clean up any resources managed by the module right away
+   *  Permanently clean up and dispose of any configuration or state for this module.
    */
-  public abstract teardown(): Promise<void>;
+  public abstract teardown(options: TearDownOptions): Promise<void>;
 
   public get name() {
     return this.options.name;

packages/service-core/src/modules/ModuleManager.ts

@@ -1,8 +1,8 @@
 import { logger } from '@powersync/lib-services-framework';
 import * as system from '../system/system-index.js';
-import { AbstractModule } from './AbstractModule.js';
+import { AbstractModule, TearDownOptions } from './AbstractModule.js';
 /**
- *  The module manager is responsible for managing the lifecycle of all modules in the system.
+ *  The module manager keeps track of activated modules
  */
 export class ModuleManager {
   private readonly modules: Map<string, AbstractModule> = new Map();
@@ -26,9 +26,9 @@ export class ModuleManager {
     logger.info(`Successfully Initialized modules.`);
   }
 
-  async tearDown() {
+  async tearDown(options: TearDownOptions) {
     for (const module of this.modules.values()) {
-      await module.teardown();
+      await module.teardown(options);
     }
   }
 }