fix(webapp): auto-recover replication services after stream errors#3613
fix(webapp): auto-recover replication services after stream errors#3613ericallam wants to merge 5 commits into
Conversation
|
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
WalkthroughThis PR adds configurable error recovery for the runs and sessions replication services. When a logical replication stream fails (e.g., during a database failover), the system can reconnect with exponential backoff, exit to let an external supervisor restart the host, or remain stopped with logging. Environment variables control per-service strategy selection and tuning. The implementation integrates into both services' lifecycle (on error, stream start, and shutdown) and is validated through containerized integration tests that force replication stream failures. Estimated code review effort🎯 3 (Moderate) | ⏱️ ~25 minutes 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
ericallam
force-pushed
the
fix/replication-auto-recover-on-stream-error
branch
from
6f8cc24 to
5ba46ff
Compare
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
ericallam
force-pushed
the
fix/replication-auto-recover-on-stream-error
branch
2 times, most recently
from
bc57072 to
969dbdb
Compare
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
ericallam
force-pushed
the
fix/replication-auto-recover-on-stream-error
branch
from
969dbdb to
964b7e4
Compare
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
ericallam
force-pushed
the
fix/replication-auto-recover-on-stream-error
branch
from
964b7e4 to
a2eaf3e
Compare
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
When the underlying logical-replication client errored (e.g. after a Postgres failover), the runs and sessions replication services logged the error and left the stream stopped. The host process kept running, the WAL backed up, and ClickHouse silently fell behind. Both services now run a configurable recovery strategy on stream errors, defaulting to in-process reconnect with exponential backoff so a fresh self-hosted setup heals on its own: - "reconnect" (default) re-subscribes via the existing subscribe(lastLsn) path with exponential backoff (1s -> 60s cap, unlimited attempts), which re-validates the publication, re-acquires the leader lock, and resumes from the last acknowledged LSN. - "exit" calls process.exit after a short flush window so a host's supervisor (Docker restart=always, systemd, k8s, etc.) can replace the process. - "log" preserves the historical behaviour. Per-service strategy + exit knobs are env-driven via RUN_REPLICATION_ERROR_STRATEGY / SESSION_REPLICATION_ERROR_STRATEGY plus matching *_EXIT_DELAY_MS / *_EXIT_CODE. Reconnect tuning is shared across both services via REPLICATION_RECONNECT_INITIAL_DELAY_MS / _MAX_DELAY_MS / _MAX_ATTEMPTS (0 = unlimited).
Addresses PR review feedback:
- LogicalReplicationClient.subscribe() can throw before its internal
"error" listener is wired up (notably when pg client.connect() fails
mid-failover). The reconnect strategy's catch block only logged, so
recovery silently stopped. Now also calls scheduleReconnect(err) — the
pendingReconnect guard makes it idempotent if an error event was also
emitted.
- Reject negative values for the new replication-recovery env vars and
cap exit codes at 255.
- Convert the new ReplicationErrorRecovery{Deps,} interfaces to type
aliases to match the repo's TypeScript style.
- Tighten the reconnect dep comment to drop a stale "lastAcknowledgedLsn"
reference (the wrapper-tracked resume LSN is what callers actually pass).
- Restore process.exit after service.shutdown() in the exit-strategy
test so a delayed exit timer can't terminate the test worker.
LogicalReplicationClient.subscribe() can resolve without throwing or emitting an "error" event when leader-lock acquisition fails — it just calls this.stop() and returns. The reconnect callback now checks isStopped after subscribe() and throws so the recovery handler can schedule the next attempt instead of silently giving up.
…rough handle() The previous post-subscribe() isStopped check was always true on the happy path: subscribe() calls stop() up front (setting _isStopped=true) and only resets the flag inside the replicationStart event, which fires asynchronously after subscribe() returns. So the check threw on every successful reconnect, the catch rescheduled, the next attempt tore down the just-built client, and the cycle continued — replication briefly worked between teardowns, which is why the integration test passed. Replace it with the correct nudge: subscribe to leaderElection and call the recovery handler on isLeader=false. That's the only subscribe() exit path that doesn't either throw or emit an "error" event (the other silent-return paths emit "error" first via createPublication/createSlot failures).
The previous commit routed leaderElection(false) through handle(), which under the exit strategy schedules process.exit. In a multi-instance deployment that turns lost leader election — a normal operational state — into a restart loop: exit, supervisor restarts, election fails again, exit, and so on. Add a dedicated notifyLeaderElectionLost() on ReplicationErrorRecovery that the reconnect strategy treats as another retry trigger, while exit and log strategies no-op. Wire the wrapper services through the new method.
ericallam
force-pushed
the
fix/replication-auto-recover-on-stream-error
branch
from
499060e to
5944ff6
Compare
| const exitSpy = vi | ||
| .spyOn(process, "exit") | ||
| .mockImplementation(((code?: number) => undefined as never) as typeof process.exit); |
There was a problem hiding this comment.
🟡 Test uses vi.spyOn/mockImplementation in violation of repository testing rules
The exit strategy test uses vi.spyOn(process, "exit").mockImplementation(...) at line 79-81. The repository's testing guidelines in ai/references/tests.md:78-81 explicitly state "Do not mock anything", "Do not use mocks in tests", "Do not use spies in tests", and "Do not use stubs in tests". CLAUDE.md reinforces this with "Never mock anything - use testcontainers instead." The existing sibling replication tests (test/runsReplicationService.part1.test.ts, test/runsReplicationService.part2.test.ts) follow this convention and use no mocks.
Pragmatic context
Mocking process.exit is arguably necessary here since the alternative (letting the process actually exit) would kill the test runner. There are also ~20 existing violations of this rule elsewhere in the webapp test suite. One option would be to restructure the exit strategy to call an injectable callback instead of process.exit directly, making it testable without mocks.
Prompt for agents
The exit strategy test spies on process.exit and uses mockImplementation, violating the repository rule against mocks/spies in tests (see ai/references/tests.md). To make this testable without mocks, consider refactoring the exit strategy in replicationErrorRecovery.server.ts to accept an injectable exit function (defaulting to process.exit). The test can then pass a no-op or tracking function instead of spying on the global. This would keep the test aligned with the existing replication service test conventions in runsReplicationService.part1.test.ts and part2.test.ts which use no mocks.
Was this helpful? React with 👍 or 👎 to provide feedback.
| errorRecovery: strategyFromEnv({ | ||
| strategy: env.SESSION_REPLICATION_ERROR_STRATEGY, | ||
| reconnectInitialDelayMs: env.REPLICATION_RECONNECT_INITIAL_DELAY_MS, | ||
| reconnectMaxDelayMs: env.REPLICATION_RECONNECT_MAX_DELAY_MS, | ||
| reconnectMaxAttempts: env.REPLICATION_RECONNECT_MAX_ATTEMPTS, | ||
| exitDelayMs: env.SESSION_REPLICATION_EXIT_DELAY_MS, | ||
| exitCode: env.SESSION_REPLICATION_EXIT_CODE, | ||
| }), | ||
| }); |
There was a problem hiding this comment.
🚩 Sessions replication service is created with error recovery but never started in the instance file
The sessionsReplicationInstance.server.ts creates the SessionsReplicationService with full error recovery configuration but never calls service.start() — unlike runsReplicationInstance.server.ts:83-93 which conditionally starts and registers signal handlers. The sessions instance is reference-held in adminWorker.server.ts:12 (void sessionsReplicationInstance) with a comment claiming it "subscribes to the logical replication slot, wires signal handlers" — but neither happens. The error recovery added by this PR would only become effective if/when start() is called externally. This is a pre-existing issue, not introduced by this PR.
Was this helpful? React with 👍 or 👎 to provide feedback.
| }), | ||
| }); | ||
|
|
||
| if (env.RUN_REPLICATION_ENABLED === "1") { |
There was a problem hiding this comment.
🚩 Initial start() failure from pg connection error bypasses error recovery
If start() → subscribe() throws due to client.connect() failing (at internal-packages/replication/src/client.ts:284), the error propagates to the instance's .catch() handler at runsReplicationInstance.server.ts:89-93 which only logs. No error event is emitted by the LogicalReplicationClient for connect() failures, so the error recovery never triggers. The service is left dead. This is a pre-existing gap — the PR's error recovery correctly handles mid-stream failures and leader election losses, but the initial connection failure path remains unprotected. The reconnect strategy DOES handle this case during subsequent reconnect attempts (the catch block at replicationErrorRecovery.server.ts:98-110 re-schedules), so only the very first start() call is affected.
(Refers to lines 83-93)
Was this helpful? React with 👍 or 👎 to provide feedback.
1 participant
Summary
When the logical-replication stream errored (most commonly after a Postgres failover), the runs and sessions replication services logged the error and left the underlying client stopped. The host process kept running, the WAL backed up, and ClickHouse silently fell behind.
Fix
Both services now run a configurable recovery strategy on stream errors, defaulting to in-process reconnect with exponential backoff so a fresh self-hosted setup heals on its own.
reconnect(default) — re-subscribe with exponential backoff (1s → 60s cap, unlimited attempts).LogicalReplicationClient.subscribe(lastLsn)re-validates the publication, re-acquires the leader lock, and resumes from the last acknowledged LSN.exit—process.exit(1)after a short flush window so a host supervisor (Dockerrestart=always, systemd, k8s) can replace the process.log— preserves the old behaviour.Per-service strategy + exit knobs are env-driven (
RUN_REPLICATION_ERROR_STRATEGY/SESSION_REPLICATION_ERROR_STRATEGY+*_EXIT_DELAY_MS,*_EXIT_CODE). Reconnect tuning is shared across both services (REPLICATION_RECONNECT_INITIAL_DELAY_MS,_MAX_DELAY_MS,_MAX_ATTEMPTS;MAX_ATTEMPTS=0means unlimited).Test plan
Integration tests cover all three strategies by simulating a failover with
pg_terminate_backendagainst the WAL sender:reconnect— kill the backend, insert a new row, assert it lands in ClickHouseexit— kill the backend, assertprocess.exit(1)is calledlog— kill the backend, insert a new row, assert it does not land in ClickHousepnpm --filter webapp test --run runsReplicationService.errorRecovery