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:
@@ -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")
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user