From 7b6e7df9e2cb882b0cf7070d9c5078083846a8be Mon Sep 17 00:00:00 2001
From: sravan27 <sridharsravan@live.com>
Date: Thu, 9 Jul 2026 20:55:26 +0530
Subject: [PATCH] Add configurable Postgres replication socket timeout

---
 .changeset/thin-chefs-juggle.md               |  5 +++
 .../src/replication/PgManager.ts              |  8 ++++-
 modules/module-postgres/src/types/types.ts    | 14 ++++++--
 .../module-postgres/test/src/config.test.ts   | 35 +++++++++++++++++++
 4 files changed, 58 insertions(+), 4 deletions(-)
 create mode 100644 .changeset/thin-chefs-juggle.md
 create mode 100644 modules/module-postgres/test/src/config.test.ts

diff --git a/.changeset/thin-chefs-juggle.md b/.changeset/thin-chefs-juggle.md
new file mode 100644
index 000000000..8bc48db5d
--- /dev/null
+++ b/.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.
diff --git a/modules/module-postgres/src/replication/PgManager.ts b/modules/module-postgres/src/replication/PgManager.ts
index 612d631ac..83e6bf2c1 100644
--- a/modules/module-postgres/src/replication/PgManager.ts
+++ b/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;
   }
 
   /**
diff --git a/modules/module-postgres/src/types/types.ts b/modules/module-postgres/src/types/types.ts
index 4de2ac0ed..e769541e1 100644
--- a/modules/module-postgres/src/types/types.ts
+++ b/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;
 }
diff --git a/modules/module-postgres/test/src/config.test.ts b/modules/module-postgres/test/src/config.test.ts
new file mode 100644
index 000000000..f38d8d1d9
--- /dev/null
+++ b/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();
+    });
+  });
+});