feat(module-postgres): configurable snapshot socket timeout

.changeset/snapshot-socket-timeout.md

@@ -0,0 +1,5 @@
+---
+'@powersync/service-module-postgres': minor
+---
+
+Add a `snapshot_socket_timeout` connection option for the idle timeout of snapshot connection sockets. Defaults to the previous fixed 30 seconds. When storage flushes stall the snapshot loop for longer than the timeout (for example when replicating while an active sync rules instance is streaming on the same storage), the source connection is killed mid-snapshot; raising the timeout avoids the reconnect cycle.

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

@@ -11,6 +11,8 @@ export interface PgManagerOptions extends pgwire.PgPoolOptions {}
 
 /**
  * Shorter timeout for snapshot connections than for replication connections.
+ *
+ * Default, can be overridden with the `snapshot_socket_timeout` connection option.
  */
 const SNAPSHOT_SOCKET_TIMEOUT = 30_000;
 
@@ -74,7 +76,7 @@ export class PgManager extends BaseObserver<PgManagerListener> {
     // Since we are constantly using the connection, we don't need any
     // custom keepalives.
 
-    (connection as any)._socket.setTimeout(SNAPSHOT_SOCKET_TIMEOUT);
+    (connection as any)._socket.setTimeout(this.options.snapshot_socket_timeout_ms ?? SNAPSHOT_SOCKET_TIMEOUT);
 
     // Disable statement timeout for snapshot queries.
     // On Supabase, the default is 2 minutes.

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

@@ -6,13 +6,24 @@ import * as t from 'ts-codec';
 // Maintain backwards compatibility by exporting these
 export const validatePort = lib_postgres.validatePort;
 export const baseUri = lib_postgres.baseUri;
-export type NormalizedPostgresConnectionConfig = lib_postgres.NormalizedBasePostgresConnectionConfig;
+export interface NormalizedPostgresConnectionConfig extends lib_postgres.NormalizedBasePostgresConnectionConfig {
+  snapshot_socket_timeout_ms?: number | undefined;
+}
 export const POSTGRES_CONNECTION_TYPE = lib_postgres.POSTGRES_CONNECTION_TYPE;
 
 export const PostgresConnectionConfig = service_types.configFile.DataSourceConfig.and(
   lib_postgres.BasePostgresConnectionConfig
 ).and(
   t.object({
+    /**
+     * Idle timeout in seconds for snapshot connection sockets.
+     *
+     * Defaults to 30 seconds. When the storage cannot keep up with the snapshot,
+     * a storage flush can stall the snapshot loop for longer than this, killing
+     * the source connection mid-snapshot. Raising the timeout gives the source
+     * connection more slack under storage backpressure.
+     */
+    snapshot_socket_timeout: t.number.optional(),
     /**
      * Interval in seconds between source connection heartbeats. Null or omitted uses the default.
      */
@@ -47,6 +58,7 @@ export function isPostgresConfig(
 export function normalizeConnectionConfig(options: PostgresConnectionConfig) {
   return {
     ...lib_postgres.normalizeConnectionConfig(options),
+    snapshot_socket_timeout_ms: lib_postgres.parseConnectTimeout(options.snapshot_socket_timeout, undefined),
     heartbeat_interval_seconds: normalizeHeartbeatInterval(options.heartbeat_interval_seconds)
   } satisfies NormalizedPostgresConnectionConfig & { heartbeat_interval_seconds: number };
 }

modules/module-postgres/test/src/config.test.ts

@@ -32,4 +32,26 @@ describe('Postgres connection config', () => {
       );
     }
   );
+
+  describe('snapshot_socket_timeout', () => {
+    test('normalizes snapshot socket timeout from seconds to milliseconds', () => {
+      const normalized = normalizeConnectionConfig({ ...BASE_CONFIG, snapshot_socket_timeout: 90 });
+
+      expect(normalized.snapshot_socket_timeout_ms).equals(90_000);
+    });
+
+    test('leaves snapshot socket timeout unset by default', () => {
+      const normalized = normalizeConnectionConfig(BASE_CONFIG);
+
+      expect(normalized.snapshot_socket_timeout_ms).toBeUndefined();
+    });
+
+    test('ignores invalid snapshot socket timeout values', () => {
+      for (const invalid of [0, -5, NaN, Infinity]) {
+        const normalized = normalizeConnectionConfig({ ...BASE_CONFIG, snapshot_socket_timeout: invalid });
+
+        expect(normalized.snapshot_socket_timeout_ms, `value ${invalid}`).toBeUndefined();
+      }
+    });
+  });
 });