amazon: reconnect the LISTEN connection on failure instead of dying

A dropped connection, a Postgres restart, or any other error on the
dedicated LISTEN connection previously propagated all the way out of
ProcessEvents, and main.go's top-level shutdown logic treats that
error channel firing the same as a fatal server error - taking down
the entire application over a hiccup on one background connection that
has nothing to do with serving HTTP traffic. This matters more with
eleven other platforms already sharing the same trigger+notify shape
in the migrations with no Go processor yet.

Adds (*Mocks).reconnectOrStop: on a real failure (not an ordinary
shutdown), logs a warning and re-establishes LISTEN after a backoff
that starts at 1s, caps at 30s, doubles on repeated immediate
failures, and resets once a reconnect actually succeeds.
ProcessEvents' select no longer returns on a LISTEN error - it loops
back in with fresh channels instead.

Verified against a genuinely killed connection (pg_terminate_backend,
targeting the backend via pg_stat_activity matched on its LISTEN
query text), not a simulated one - both in a manual check and in the
new TestProcessEvents_ReconnectsAfterListenConnectionDrops test.

Closes out all four insights from the domains/amazon design review.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XEDaCB7C2NEBgyvqEtZuxY
This commit is contained in:
2026-08-19 23:20:15 -06:00
co-authored by Claude Sonnet 5
parent d579a3b8cf
commit bf454ea949
3 changed files with 166 additions and 2 deletions
+84
View File
@@ -180,6 +180,32 @@ func cleanupShop(t *testing.T, pool *pgxpool.Pool, shopID string) {
})
}
// terminateListenConnection finds the backend holding this package's LISTEN
// registration (identified by its last query text, which Postgres keeps
// showing while the connection sits idle waiting for notifications) and
// forcibly kills it - the same failure mode a dropped connection or a
// Postgres restart produces, so tests can exercise real reconnect behavior
// instead of a simulated one.
func terminateListenConnection(t *testing.T, pool *pgxpool.Pool) {
t.Helper()
ctx := context.Background()
var pid int
err := pool.QueryRow(ctx, `
SELECT pid FROM pg_stat_activity
WHERE query = 'LISTEN ' || $1
ORDER BY backend_start DESC
LIMIT 1
`, eventChannelName).Scan(&pid)
if err != nil {
t.Fatalf("failed to find the LISTEN connection's backend pid: %v", err)
}
if _, err := pool.Exec(ctx, `SELECT pg_terminate_backend($1)`, pid); err != nil {
t.Fatalf("failed to terminate backend %d: %v", pid, err)
}
}
func TestProcessUnprocessedEvents_ProcessesAllEventsAcrossBatches(t *testing.T) {
pool := testdb.Pool(t)
spy := newNotifySpy()
@@ -426,3 +452,61 @@ func TestProcessEvents_ShutsDownOnContextCancel(t *testing.T) {
t.Fatal("ProcessEvents() did not return within 5s of context cancellation")
}
}
// TestProcessEvents_ReconnectsAfterListenConnectionDrops is insight #4's
// fix: forcibly kills the real backend connection ProcessEvents is
// LISTEN-ing on (the same failure mode a dropped connection or a Postgres
// restart produces) and confirms it reconnects and keeps working on its
// own, rather than the error propagating out of ProcessEvents entirely.
func TestProcessEvents_ReconnectsAfterListenConnectionDrops(t *testing.T) {
pool := testdb.Pool(t)
spy := newNotifySpy()
m := NewMocks(testdb.Logger(), pool).SetListener(spy)
shopID := "test-shop-" + uuid.NewString()
cleanupShop(t, pool, shopID)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
errCh := make(chan error, 1)
go func() {
errCh <- m.ProcessEvents(ctx)
}()
select {
case <-m.Ready():
case <-time.After(5 * time.Second):
t.Fatal("ProcessEvents() did not become ready (LISTEN registered) within 5s")
}
terminateListenConnection(t, pool)
// ProcessEvents should still be running, just reconnecting (initial
// backoff is 1s) - it must not have returned because of this.
select {
case err := <-errCh:
t.Fatalf("ProcessEvents() returned (err = %v) after its LISTEN connection was killed, want it to reconnect and keep running", err)
case <-time.After(2 * time.Second):
}
insertRawAmazonEvent(t, pool, shopID, "evt-1")
deadline := time.Now().Add(5 * time.Second)
for !isProcessed(t, pool, shopID, "evt-1") {
if time.Now().After(deadline) {
t.Fatal("event was not processed within 5s of insertion after the LISTEN connection was forcibly dropped - reconnection did not restore the reactive path")
}
time.Sleep(20 * time.Millisecond)
}
cancel()
select {
case err := <-errCh:
if err != nil {
t.Fatalf("ProcessEvents() returned error = %v after context cancellation, want nil", err)
}
case <-time.After(5 * time.Second):
t.Fatal("ProcessEvents() did not return within 5s of context cancellation")
}
}