feat(db): split replication lag into transport vs replay

apps/web/src/app/api/cron/db-replication-health/route.test.ts

@@ -77,6 +77,8 @@ describe('GET /api/cron/db-replication-health', () => {
             status: 'lagging',
             in_recovery: true,
             replay_lsn: '1EE6/65035140',
+            receive_lsn: '1EE6/65035140',
+            receive_replay_gap_bytes: '0',
             last_xact_replay_timestamp: '2026-07-22 10:41:00+00',
             replay_delay_seconds: 691200,
             error: null,
@@ -93,6 +95,43 @@ describe('GET /api/cron/db-replication-health', () => {
     expect(mockCaptureException).toHaveBeenCalledTimes(1);
   });
 
+  it('emits one walsender line per replica so lag can be split into transport vs replay', async () => {
+    mockCollect.mockResolvedValue(
+      report({
+        walSenders: [
+          {
+            application_name: 'main',
+            client_addr: '54.153.89.218/32',
+            state: 'streaming',
+            sync_state: 'async',
+            sent_lag_bytes: '0',
+            flush_lag_bytes: '690368',
+            replay_lag_bytes: '135858008',
+            write_lag_seconds: 0.15,
+            flush_lag_seconds: 0.15,
+            replay_lag_seconds: 199,
+          },
+        ],
+      })
+    );
+
+    await GET(createRequest({ authorization: 'Bearer cron-secret' }));
+
+    const logged = jest
+      .mocked(console.log)
+      .mock.calls.map(([line]) => JSON.parse(String(line)))
+      .filter(entry => entry.type === 'db_replication_wal_sender');
+
+    expect(logged).toEqual([
+      expect.objectContaining({
+        client_addr: '54.153.89.218/32',
+        flush_lag_seconds: 0.15,
+        replay_lag_seconds: 199,
+        timestamp: '2026-07-30T09:39:06.000Z',
+      }),
+    ]);
+  });
+
   it('alerts when a slot is at risk', async () => {
     mockCollect.mockResolvedValue(
       report({

apps/web/src/app/api/cron/db-replication-health/route.ts

@@ -13,6 +13,10 @@ import { collectReplicationHealth } from '@/lib/replication-health';
  * metric has a documented history of returning no data, and — like the primary's
  * `pg_stat_replication` — cannot see a replica whose walreceiver has died. The
  * per-replica probe is the authoritative signal.
+ *
+ * Runs every minute (see `vercel.json`). The us-west replica loses ~1-3 minutes
+ * of replay a few times a day; at the previous 5-minute cadence each episode was
+ * caught by roughly a single sample, so its onset and duration were unmeasurable.
  */
 export async function GET(request: Request) {
   const authHeader = request.headers.get('authorization');
@@ -28,6 +32,19 @@ export async function GET(request: Request) {
       JSON.stringify({ type: 'db_replication_health', ...replica, timestamp: report.timestamp })
     );
   }
+  // The primary's view of each replica. Only the replica probe can see a replica
+  // that stopped streaming, but only the primary can attribute lag to getting the
+  // WAL there (write/flush) versus applying it (replay), which is what the
+  // per-replica probe cannot distinguish on its own.
+  for (const walSender of report.walSenders) {
+    console.log(
+      JSON.stringify({
+        type: 'db_replication_wal_sender',
+        ...walSender,
+        timestamp: report.timestamp,
+      })
+    );
+  }
   for (const slot of report.slots) {
     console.log(
       JSON.stringify({ type: 'db_replication_slot', ...slot, timestamp: report.timestamp })

apps/web/src/app/api/internal/db/replication-lag/route.test.ts

@@ -56,6 +56,8 @@ describe('GET /api/internal/db/replication-lag', () => {
             status: 'unreachable',
             in_recovery: null,
             replay_lsn: null,
+            receive_lsn: null,
+            receive_replay_gap_bytes: null,
             last_xact_replay_timestamp: null,
             replay_delay_seconds: null,
             error: 'connection timeout',

apps/web/src/lib/replication-health.test.ts

@@ -27,7 +27,11 @@ function walSenderRow(overrides: Record<string, unknown> = {}) {
     client_addr: '18.153.166.51/32',
     state: 'streaming',
     sync_state: 'async',
+    sent_lag_bytes: '0',
+    flush_lag_bytes: '0',
     replay_lag_bytes: '0',
+    write_lag_seconds: 0.002,
+    flush_lag_seconds: 0.002,
     replay_lag_seconds: 0.002,
     ...overrides,
   };
@@ -50,6 +54,8 @@ function okProbe(name: string): ReplicaHealth {
     status: 'ok',
     in_recovery: true,
     replay_lsn: '2008/2F522000',
+    receive_lsn: '2008/2F522000',
+    receive_replay_gap_bytes: '0',
     last_xact_replay_timestamp: '2026-07-30 09:39:06+00',
     replay_delay_seconds: 0.1,
     error: null,
@@ -87,6 +93,8 @@ describe('classifyReplicaRow', () => {
       classifyReplicaRow({
         in_recovery: false,
         replay_lsn: null,
+        receive_lsn: null,
+        receive_replay_gap_bytes: null,
         last_xact_replay_timestamp: null,
         replay_delay_seconds: 0,
       })
@@ -98,6 +106,8 @@ describe('classifyReplicaRow', () => {
       classifyReplicaRow({
         in_recovery: true,
         replay_lsn: '1/1',
+        receive_lsn: '1/1',
+        receive_replay_gap_bytes: '0',
         last_xact_replay_timestamp: '2026-07-22 10:41:00+00',
         replay_delay_seconds: REPLICA_LAG_ALERT_SECONDS + 1,
       })
@@ -109,6 +119,8 @@ describe('classifyReplicaRow', () => {
       classifyReplicaRow({
         in_recovery: true,
         replay_lsn: '1/1',
+        receive_lsn: '1/1',
+        receive_replay_gap_bytes: '0',
         last_xact_replay_timestamp: '2026-07-30 00:00:00+00',
         replay_delay_seconds: 0.2,
       })
@@ -195,6 +207,8 @@ describe('collectReplicationHealth', () => {
         status: 'unreachable',
         in_recovery: null,
         replay_lsn: null,
+        receive_lsn: null,
+        receive_replay_gap_bytes: null,
         last_xact_replay_timestamp: null,
         replay_delay_seconds: null,
         error: 'connection timeout',

apps/web/src/lib/replication-health.ts

@@ -89,6 +89,29 @@ export type ReplicaHealth = {
   status: ReplicaHealthStatus;
   in_recovery: boolean | null;
   replay_lsn: string | null;
+  /**
+   * How far the WAL *receiver* has got, versus `replay_lsn` (how far replay has
+   * got). Together these split lag into its two possible causes: a large
+   * `receive_replay_gap_bytes` means the WAL arrived and replay is the
+   * bottleneck, while a gap near zero alongside a high `replay_delay_seconds`
+   * means the WAL has not arrived yet, i.e. the bottleneck is transport.
+   *
+   * Compare across samples, never in isolation. `pg_last_wal_receive_lsn()` is
+   * null only while streaming has not started in the current recovery session;
+   * a walreceiver that *dies* does not null it, it freezes it at the last LSN
+   * received. So the dead-walreceiver failure this module exists to catch shows
+   * up as a large and roughly constant gap — which in any single sample is
+   * indistinguishable from a slow replay. A `receive_lsn` that does not advance
+   * between samples is the signature.
+   *
+   * A negative gap is legitimate: a replica recovering from the WAL archive
+   * replays past the position streaming reached.
+   *
+   * Both fields come from one snapshot of the LSN functions, so they are always
+   * consistent with each other and with `replay_lsn` (see `probeReplica`).
+   */
+  receive_lsn: string | null;
+  receive_replay_gap_bytes: string | null;
   last_xact_replay_timestamp: string | null;
   replay_delay_seconds: number | null;
   error: string | null;
@@ -97,17 +120,34 @@ export type ReplicaHealth = {
 type ReplicaProbeRow = {
   in_recovery: boolean;
   replay_lsn: string | null;
+  receive_lsn: string | null;
+  receive_replay_gap_bytes: string | null;
   last_xact_replay_timestamp: string | null;
   replay_delay_seconds: number | null;
 };
 
-/** Currently-connected walsender, as seen on the primary. */
+/**
+ * Currently-connected walsender, as seen on the primary.
+ *
+ * The three read replicas all connect with `application_name = 'main'`, so
+ * `client_addr` is the only way to tell them apart.
+ *
+ * `write`/`flush`/`replay` are reported separately for the same reason the
+ * replica probe reports receive and replay separately: write and flush lag
+ * cover getting the WAL to the standby's disk, replay lag covers applying it.
+ * A spike in all three points at the link; a spike in replay alone points at
+ * the standby.
+ */
 export type WalSender = {
   application_name: string | null;
   client_addr: string | null;
   state: string | null;
   sync_state: string | null;
+  sent_lag_bytes: string;
+  flush_lag_bytes: string;
   replay_lag_bytes: string;
+  write_lag_seconds: number | null;
+  flush_lag_seconds: number | null;
   replay_lag_seconds: number | null;
 };
 
@@ -172,13 +212,31 @@ export async function probeReplica(target: ReplicaTarget): Promise<ReplicaHealth
     // process, so attach a no-op like the long-lived pools in lib/drizzle.ts do.
     client.pool.on('error', () => {});
 
+    // Every one of these LSN functions is volatile and reads live shared memory,
+    // so each call site is evaluated independently: calling them once per output
+    // column would let `receive_replay_gap_bytes` disagree with the `receive_lsn`
+    // and `replay_lsn` logged beside it, and could even report a negative gap
+    // from replay advancing mid-row. Reading them once in a subquery and
+    // deriving the outputs from those columns keeps the row self-consistent —
+    // Postgres will not flatten a subquery whose target list is volatile, so the
+    // values really are reused rather than re-read.
     const { rows } = await client.db.execute<ReplicaProbeRow>(sql`
       SELECT
-        pg_is_in_recovery() AS in_recovery,
-        pg_last_wal_replay_lsn()::text AS replay_lsn,
-        pg_last_xact_replay_timestamp()::text AS last_xact_replay_timestamp,
-        EXTRACT(EPOCH FROM (now() - pg_last_xact_replay_timestamp()))::double precision
+        snapshot.in_recovery,
+        snapshot.replay_lsn::text AS replay_lsn,
+        snapshot.receive_lsn::text AS receive_lsn,
+        pg_wal_lsn_diff(snapshot.receive_lsn, snapshot.replay_lsn)::text
+          AS receive_replay_gap_bytes,
+        snapshot.last_xact_replay_timestamp::text AS last_xact_replay_timestamp,
+        EXTRACT(EPOCH FROM (now() - snapshot.last_xact_replay_timestamp))::double precision
           AS replay_delay_seconds
+      FROM (
+        SELECT
+          pg_is_in_recovery() AS in_recovery,
+          pg_last_wal_replay_lsn() AS replay_lsn,
+          pg_last_wal_receive_lsn() AS receive_lsn,
+          pg_last_xact_replay_timestamp() AS last_xact_replay_timestamp
+      ) AS snapshot
     `);
 
     const row = rows[0];
@@ -188,6 +246,8 @@ export async function probeReplica(target: ReplicaTarget): Promise<ReplicaHealth
         status: 'unreachable',
         in_recovery: null,
         replay_lsn: null,
+        receive_lsn: null,
+        receive_replay_gap_bytes: null,
         last_xact_replay_timestamp: null,
         replay_delay_seconds: null,
         error: 'probe returned no rows',
@@ -199,6 +259,8 @@ export async function probeReplica(target: ReplicaTarget): Promise<ReplicaHealth
       status: classifyReplicaRow(row),
       in_recovery: row.in_recovery,
       replay_lsn: row.replay_lsn,
+      receive_lsn: row.receive_lsn,
+      receive_replay_gap_bytes: row.receive_replay_gap_bytes,
       last_xact_replay_timestamp: row.last_xact_replay_timestamp,
       replay_delay_seconds: row.replay_delay_seconds,
       error: null,
@@ -209,6 +271,8 @@ export async function probeReplica(target: ReplicaTarget): Promise<ReplicaHealth
       status: 'unreachable',
       in_recovery: null,
       replay_lsn: null,
+      receive_lsn: null,
+      receive_replay_gap_bytes: null,
       last_xact_replay_timestamp: null,
       replay_delay_seconds: null,
       error: errorMessage(error),
@@ -225,7 +289,11 @@ async function getWalSenders(): Promise<WalSender[]> {
       client_addr::text AS client_addr,
       state,
       sync_state,
+      COALESCE(pg_wal_lsn_diff(pg_current_wal_lsn(), sent_lsn), 0)::text AS sent_lag_bytes,
+      COALESCE(pg_wal_lsn_diff(pg_current_wal_lsn(), flush_lsn), 0)::text AS flush_lag_bytes,
       COALESCE(pg_wal_lsn_diff(pg_current_wal_lsn(), replay_lsn), 0)::text AS replay_lag_bytes,
+      EXTRACT(EPOCH FROM write_lag)::double precision AS write_lag_seconds,
+      EXTRACT(EPOCH FROM flush_lag)::double precision AS flush_lag_seconds,
       EXTRACT(EPOCH FROM replay_lag)::double precision AS replay_lag_seconds
     FROM pg_stat_replication
     ORDER BY application_name, client_addr
@@ -291,6 +359,8 @@ export async function collectReplicationHealth(options?: {
             status: 'unreachable',
             in_recovery: null,
             replay_lsn: null,
+            receive_lsn: null,
+            receive_replay_gap_bytes: null,
             last_xact_replay_timestamp: null,
             replay_delay_seconds: null,
             error: errorMessage(error),

apps/web/vercel.json

@@ -66,7 +66,7 @@
     },
     {
       "path": "/api/cron/db-replication-health",
-      "schedule": "*/5 * * * *"
+      "schedule": "* * * * *"
     },
     {
       "path": "/api/cron/dispatch-affiliate-events",