fix(webapp): stop replication services re-acking an older commit LSN

Stops the runs, sessions and webhook-deliveries replication services
from reporting an older replication position after a keepalive. On an
idle database this made `pg_stat_replication` lag appear to grow without
bound, even though the slot itself was current.

The acknowledge loop compared the keepalive LSN (unpadded) with the last
commit LSN (zero-padded) as strings, so they never matched and the older
commit LSN was re-sent. LSNs are now compared numerically, and an ack is
skipped when the last acknowledged LSN is already at or past the latest
commit:

```ts
if (
  runtime.lastAcknowledgedLsn &&
  lsnToUInt64(runtime.lastAcknowledgedLsn) >= lsnToUInt64(runtime.latestCommitEndLsn)
) {
  return;
}
```

Each service has an integration test against real Postgres that fails
before the fix (flush LSN moves backwards) and passes after.

Mono-RevId: 3cbe8f510438e51171aa6577edafb525b30bef1a

Author: d-cs

apps/webapp/app/services/runsReplicationService.server.ts

@@ -832,7 +832,11 @@ export class RunsReplicationService {
       return;
     }
 
-    if (runtime.lastAcknowledgedLsn === runtime.latestCommitEndLsn) {
+    // Keepalive acks can be ahead of the last commit; re-acking the commit would move flush back.
+    if (
+      runtime.lastAcknowledgedLsn &&
+      lsnToUInt64(runtime.lastAcknowledgedLsn) >= lsnToUInt64(runtime.latestCommitEndLsn)
+    ) {
       return;
     }
 

apps/webapp/app/services/sessionsReplicationService.server.ts

@@ -512,7 +512,11 @@ export class SessionsReplicationService {
       return;
     }
 
-    if (this._lastAcknowledgedLsn === this._latestCommitEndLsn) {
+    // Keepalive acks can be ahead of the last commit; re-acking the commit would move flush back.
+    if (
+      this._lastAcknowledgedLsn &&
+      lsnToUInt64(this._lastAcknowledgedLsn) >= lsnToUInt64(this._latestCommitEndLsn)
+    ) {
       return;
     }
 

apps/webapp/app/services/webhookDeliveriesReplicationService.server.ts

@@ -527,7 +527,11 @@ export class WebhookDeliveriesReplicationService {
       return;
     }
 
-    if (this._lastAcknowledgedLsn === this._latestCommitEndLsn) {
+    // Keepalive acks can be ahead of the last commit; re-acking the commit would move flush back.
+    if (
+      this._lastAcknowledgedLsn &&
+      lsnToUInt64(this._lastAcknowledgedLsn) >= lsnToUInt64(this._latestCommitEndLsn)
+    ) {
       return;
     }
 

apps/webapp/test/runsReplicationService.part5.test.ts

@@ -2,6 +2,10 @@ import { ClickHouse } from "@internal/clickhouse";
 import { replicationContainerTest } from "@internal/testcontainers";
 import { z } from "zod";
 import { RunsReplicationService } from "~/services/runsReplicationService.server";
+import {
+  expectFlushLsnHoldsPastIdleWal,
+  withShortWalSenderTimeout,
+} from "./utils/replicationUtils";
 import { TestReplicationClickhouseFactory } from "./utils/testReplicationClickhouseFactory";
 
 vi.setConfig({ testTimeout: 60_000 });
@@ -144,4 +148,79 @@ describe("RunsReplicationService (part 5/7)", () => {
       await runsReplicationService.stop();
     }
   );
+  // Catches: the 1s ack loop re-acking an older commit LSN after a keepalive ack, which moves
+  // pg_stat_replication flush_lsn backwards and inflates reported lag on an idle database.
+  replicationContainerTest(
+    "never acknowledges an older commit LSN after a keepalive acknowledgement",
+    async ({ clickhouseContainer, redisOptions, postgresContainer, prisma }) => {
+      const clickhouse = new ClickHouse({
+        url: clickhouseContainer.getConnectionUrl(),
+        name: "runs-replication-keepalive-ack",
+        logLevel: "warn",
+      });
+
+      const slotName = "task_runs_keepalive_ack_v1";
+      const runsReplicationService = new RunsReplicationService({
+        clickhouseFactory: new TestReplicationClickhouseFactory(clickhouse),
+        pgConnectionUrl: withShortWalSenderTimeout(postgresContainer.getConnectionUri()),
+        serviceName: "runs-replication-keepalive-ack",
+        slotName,
+        publicationName: "task_runs_keepalive_ack_v1_publication",
+        redisOptions,
+        maxFlushConcurrency: 1,
+        flushIntervalMs: 100,
+        flushBatchSize: 1,
+        leaderLockTimeoutMs: 5000,
+        leaderLockExtendIntervalMs: 1000,
+        logLevel: "warn",
+      });
+
+      await runsReplicationService.start();
+      try {
+        const organization = await prisma.organization.create({
+          data: { title: "keepalive-ack", slug: "keepalive-ack" },
+        });
+        const project = await prisma.project.create({
+          data: {
+            name: "keepalive-ack",
+            slug: "keepalive-ack",
+            organizationId: organization.id,
+            externalRef: "keepalive-ack",
+          },
+        });
+        const runtimeEnvironment = await prisma.runtimeEnvironment.create({
+          data: {
+            slug: "keepalive-ack",
+            type: "DEVELOPMENT",
+            projectId: project.id,
+            organizationId: organization.id,
+            apiKey: "keepalive-ack",
+            pkApiKey: "keepalive-ack",
+            shortcode: "keepalive-ack",
+          },
+        });
+        await prisma.taskRun.create({
+          data: {
+            friendlyId: "run_keepalive_ack",
+            taskIdentifier: "my-task",
+            payload: "{}",
+            payloadType: "application/json",
+            traceId: "keepalive-ack",
+            spanId: "keepalive-ack",
+            queue: "keepalive-ack",
+            runtimeEnvironmentId: runtimeEnvironment.id,
+            projectId: project.id,
+            organizationId: organization.id,
+            environmentType: "DEVELOPMENT",
+            engine: "V2",
+            status: "PENDING",
+          },
+        });
+
+        await expectFlushLsnHoldsPastIdleWal(prisma, slotName);
+      } finally {
+        await runsReplicationService.stop();
+      }
+    }
+  );
 });

apps/webapp/test/sessionsReplicationService.test.ts

@@ -3,6 +3,10 @@ import { replicationContainerTest } from "@internal/testcontainers";
 import { setTimeout } from "node:timers/promises";
 import { z } from "zod";
 import { SessionsReplicationService } from "~/services/sessionsReplicationService.server";
+import {
+  expectFlushLsnHoldsPastIdleWal,
+  withShortWalSenderTimeout,
+} from "./utils/replicationUtils";
 import { TestReplicationClickhouseFactory } from "./utils/testReplicationClickhouseFactory";
 
 vi.setConfig({ testTimeout: 60_000 });
@@ -212,4 +216,75 @@ describe("SessionsReplicationService", () => {
       await service.stop();
     }
   );
+
+  // Catches: the 1s ack loop re-acking an older commit LSN after a keepalive ack (flush_lsn regresses).
+  replicationContainerTest(
+    "never acknowledges an older commit LSN after a keepalive acknowledgement",
+    async ({ clickhouseContainer, redisOptions, postgresContainer, prisma }) => {
+      const clickhouse = new ClickHouse({
+        url: clickhouseContainer.getConnectionUrl(),
+        name: "sessions-replication",
+        logLevel: "warn",
+      });
+
+      const slotName = "sessions_keepalive_ack_v1";
+      const service = new SessionsReplicationService({
+        clickhouseFactory: new TestReplicationClickhouseFactory(clickhouse),
+        pgConnectionUrl: withShortWalSenderTimeout(postgresContainer.getConnectionUri()),
+        serviceName: "sessions-replication",
+        slotName,
+        publicationName: "sessions_keepalive_ack_v1_publication",
+        redisOptions,
+        maxFlushConcurrency: 1,
+        flushIntervalMs: 100,
+        flushBatchSize: 1,
+        leaderLockTimeoutMs: 5000,
+        leaderLockExtendIntervalMs: 1000,
+        logLevel: "warn",
+      });
+
+      await service.start();
+      try {
+        const organization = await prisma.organization.create({
+          data: { title: "test", slug: "test" },
+        });
+        const project = await prisma.project.create({
+          data: {
+            name: "test",
+            slug: "test",
+            organizationId: organization.id,
+            externalRef: "test",
+          },
+        });
+        const environment = await prisma.runtimeEnvironment.create({
+          data: {
+            slug: "test",
+            type: "DEVELOPMENT",
+            projectId: project.id,
+            organizationId: organization.id,
+            apiKey: "test",
+            pkApiKey: "test",
+            shortcode: "test",
+          },
+        });
+        await prisma.session.create({
+          data: {
+            id: "session_test_keepalive_ack",
+            friendlyId: "session_keepalive_ack",
+            type: "chat.agent",
+            projectId: project.id,
+            runtimeEnvironmentId: environment.id,
+            environmentType: "DEVELOPMENT",
+            organizationId: organization.id,
+            taskIdentifier: "my-agent",
+            triggerConfig: { basePayload: { messages: [], trigger: "preload" } },
+          },
+        });
+
+        await expectFlushLsnHoldsPastIdleWal(prisma, slotName);
+      } finally {
+        await service.stop();
+      }
+    }
+  );
 });

apps/webapp/test/utils/replicationUtils.ts

@@ -3,7 +3,47 @@ import type { RedisOptions } from "@internal/redis";
 import type { PrismaClient } from "~/db.server";
 import { RunsReplicationService } from "~/services/runsReplicationService.server";
 import { TestReplicationClickhouseFactory } from "./testReplicationClickhouseFactory";
-import { afterEach } from "vitest";
+import { afterEach, expect } from "vitest";
+import { setTimeout } from "node:timers/promises";
+
+// Keepalives with shouldRespond every 1s instead of 30s, on this connection's walsender only.
+export function withShortWalSenderTimeout(pgConnectionUrl: string) {
+  const url = new URL(pgConnectionUrl);
+  url.searchParams.set("options", "-c wal_sender_timeout=2s");
+  return url.toString();
+}
+
+// Writes WAL that only keepalives report, then asserts flush_lsn reaches it and never falls back.
+export async function expectFlushLsnHoldsPastIdleWal(prisma: PrismaClient, slotName: string) {
+  const [{ lsn: target }] = await prisma.$queryRaw<{ lsn: string }[]>`
+    SELECT pg_logical_emit_message(false, 'replication-ack-test', 'idle')::text AS lsn`;
+
+  const flushAhead = async () => {
+    const rows = await prisma.$queryRaw<{ ahead: number }[]>`
+      SELECT (r.flush_lsn - ${target}::pg_lsn)::float8 AS ahead
+      FROM pg_stat_replication r
+      JOIN pg_replication_slots s ON s.active_pid = r.pid
+      WHERE s.slot_name = ${slotName}`;
+    return rows[0]?.ahead ?? null;
+  };
+
+  const deadline = performance.now() + 15_000;
+  let ahead = await flushAhead();
+  while ((ahead === null || ahead < 0) && performance.now() < deadline) {
+    await setTimeout(50);
+    ahead = await flushAhead();
+  }
+  expect(ahead, `flush_lsn never reached keepalive position ${target}`).toBeGreaterThanOrEqual(0);
+
+  const samples: Array<number | null> = [];
+  const observeUntil = performance.now() + 3_000;
+  while (performance.now() < observeUntil) {
+    samples.push(await flushAhead());
+    await setTimeout(50);
+  }
+  const regressed = samples.filter((s) => s === null || s < 0);
+  expect(regressed, `flush_lsn moved back below ${target}`).toEqual([]);
+}
 
 export async function setupClickhouseReplication({
   prisma,

apps/webapp/test/webhookDeliveriesReplicationService.test.ts

@@ -4,6 +4,10 @@ import type { PrismaClient } from "@trigger.dev/database";
 import { setTimeout } from "node:timers/promises";
 import { z } from "zod";
 import { WebhookDeliveriesReplicationService } from "~/services/webhookDeliveriesReplicationService.server";
+import {
+  expectFlushLsnHoldsPastIdleWal,
+  withShortWalSenderTimeout,
+} from "./utils/replicationUtils";
 import { TestReplicationClickhouseFactory } from "./utils/testReplicationClickhouseFactory";
 
 vi.setConfig({ testTimeout: 60_000 });
@@ -220,4 +224,29 @@ describe("WebhookDeliveriesReplicationService", () => {
       }
     }
   );
+
+  // Catches: the 1s ack loop re-acking an older commit LSN after a keepalive ack (flush_lsn regresses).
+  replicationContainerTest(
+    "never acknowledges an older commit LSN after a keepalive acknowledgement",
+    async ({ clickhouseContainer, redisOptions, postgresContainer, prisma }) => {
+      const clickhouse = new ClickHouse({
+        url: clickhouseContainer.getConnectionUrl(),
+        name: "webhook-deliveries-replication",
+        logLevel: "warn",
+      });
+      const service = buildService(
+        clickhouse,
+        withShortWalSenderTimeout(postgresContainer.getConnectionUri()),
+        redisOptions
+      );
+      await service.start();
+
+      try {
+        await createEndpoint(prisma, await seedEnvironment(prisma));
+        await expectFlushLsnHoldsPastIdleWal(prisma, "webhook_deliveries_to_clickhouse_v1");
+      } finally {
+        await service.stop();
+      }
+    }
+  );
 });