Add configurable Postgres replication socket timeout

.changeset/thin-chefs-juggle.md

@@ -0,0 +1,5 @@
+---
+'@powersync/service-module-postgres': patch
+---
+
+Add an optional `replication_socket_timeout` setting for Postgres replication connections.

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

@@ -48,7 +48,13 @@ export class PgManager extends BaseObserver<PgManagerListener> {
   async replicationConnection(): Promise<pgwire.PgConnection> {
     const p = pgwire.connectPgWire(this.options, { type: 'replication', applicationName: getApplicationName() });
     this.connectionPromises.push(p);
-    return await p;
+    const connection = await p;
+
+    if (this.options.replication_socket_timeout_ms != null) {
+      (connection as any)._socket.setTimeout(this.options.replication_socket_timeout_ms);
+    }
+
+    return connection;
   }
 
   /**

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

@@ -5,14 +5,21 @@ 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 {
+  replication_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({
-    // Add any replication connection specific config here in future
+    /**
+     * Idle timeout in seconds for the logical replication socket.
+     * When set, a half-open or idle replication stream is destroyed after this timeout so the
+     * existing replication retry path can reconnect.
+     */
+    replication_socket_timeout: t.number.optional()
   })
 );
 
@@ -39,6 +46,7 @@ export function isPostgresConfig(
  */
 export function normalizeConnectionConfig(options: PostgresConnectionConfig) {
   return {
-    ...lib_postgres.normalizeConnectionConfig(options)
+    ...lib_postgres.normalizeConnectionConfig(options),
+    replication_socket_timeout_ms: lib_postgres.parseConnectTimeout(options.replication_socket_timeout, undefined)
   } satisfies NormalizedPostgresConnectionConfig;
 }

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

@@ -0,0 +1,35 @@
+import { describe, expect, test } from 'vitest';
+import { normalizeConnectionConfig } from '../../src/types/types.js';
+
+describe('config', () => {
+  describe('replication_socket_timeout', () => {
+    test('normalizes replication socket timeout from seconds to milliseconds', () => {
+      const normalized = normalizeConnectionConfig({
+        type: 'postgresql',
+        uri: 'postgresql://postgres:postgres@localhost:4321/powersync_test',
+        replication_socket_timeout: 45
+      });
+
+      expect(normalized.replication_socket_timeout_ms).equals(45_000);
+    });
+
+    test('leaves replication socket timeout unset by default', () => {
+      const normalized = normalizeConnectionConfig({
+        type: 'postgresql',
+        uri: 'postgresql://postgres:postgres@localhost:4321/powersync_test'
+      });
+
+      expect(normalized.replication_socket_timeout_ms).toBeUndefined();
+    });
+
+    test('ignores invalid replication socket timeout values', () => {
+      const normalized = normalizeConnectionConfig({
+        type: 'postgresql',
+        uri: 'postgresql://postgres:postgres@localhost:4321/powersync_test',
+        replication_socket_timeout: 0
+      });
+
+      expect(normalized.replication_socket_timeout_ms).toBeUndefined();
+    });
+  });
+});