diff --git a/domains/amazon/mock.go b/domains/amazon/mock.go index 283ea7e..6300efb 100644 --- a/domains/amazon/mock.go +++ b/domains/amazon/mock.go @@ -56,6 +56,13 @@ const ( defaultNotifyRetryAfter = 30 * time.Second defaultPollInterval = time.Minute + + // backoff for reconnecting the LISTEN connection after it fails for a + // reason other than shutdown (e.g. a dropped connection, a Postgres + // restart). Doubles on each consecutive failure, capped, and resets + // once a reconnect actually succeeds. + initialListenReconnectBackoff = 1 * time.Second + maxListenReconnectBackoff = 30 * time.Second ) func NewMocks(log *logging.Logger, db *pgxpool.Pool) *Mocks { @@ -99,6 +106,7 @@ func (m *Mocks) Ready() <-chan struct{} { func (m *Mocks) ProcessEvents(ctx context.Context) error { notifCh, errCh := m.listenForNotifications(ctx) + backoff := initialListenReconnectBackoff for { m.log.Debug("processing events") @@ -112,19 +120,65 @@ func (m *Mocks) ProcessEvents(ctx context.Context) error { select { case <-ctx.Done(): return nil + case err := <-errCh: - return err + var stop bool + if notifCh, errCh, backoff, stop = m.reconnectOrStop(ctx, err, backoff); stop { + return nil + } + case _, ok := <-notifCh: if !ok { - return <-errCh + var stop bool + if notifCh, errCh, backoff, stop = m.reconnectOrStop(ctx, <-errCh, backoff); stop { + return nil + } + continue } m.log.Debug("woke up: notification received") + case <-time.After(m.pollInterval): m.log.Debug("woke up: poll interval elapsed") } } } +// reconnectOrStop handles a LISTEN-connection failure. A nil listenErr (or +// ctx already being done) means this is an ordinary shutdown, not a +// failure - stop is true and the caller should return. Otherwise it logs a +// warning, waits out backoff, and re-establishes LISTEN: on success the +// backoff resets to its initial value for next time; on immediate failure +// (e.g. the pool itself is unreachable) it doubles, capped, so repeated +// failures back off rather than hot-looping. +func (m *Mocks) reconnectOrStop( + ctx context.Context, + listenErr error, + backoff time.Duration, +) (notifCh <-chan struct{}, errCh <-chan error, nextBackoff time.Duration, stop bool) { + if listenErr == nil || ctx.Err() != nil { + return nil, nil, backoff, true + } + + m.log.Warn("lost connection while listening for notifications; reconnecting", "error", listenErr, "retry_in", backoff) + + select { + case <-ctx.Done(): + return nil, nil, backoff, true + case <-time.After(backoff): + } + + notifCh, errCh = m.listenForNotifications(ctx) + if notifCh != nil { + return notifCh, errCh, initialListenReconnectBackoff, false + } + + nextBackoff = backoff * 2 + if nextBackoff > maxListenReconnectBackoff { + nextBackoff = maxListenReconnectBackoff + } + return notifCh, errCh, nextBackoff, false +} + func (m *Mocks) listenForNotifications(ctx context.Context) (<-chan struct{}, <-chan error) { errCh := make(chan error, 1) diff --git a/domains/amazon/mock_test.go b/domains/amazon/mock_test.go index 6d1b405..4536412 100644 --- a/domains/amazon/mock_test.go +++ b/domains/amazon/mock_test.go @@ -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") + } +} diff --git a/work-summaries/work-summary-Claude-2026-08-05-2155.md b/work-summaries/work-summary-Claude-2026-08-05-2155.md new file mode 100644 index 0000000..bef9991 --- /dev/null +++ b/work-summaries/work-summary-Claude-2026-08-05-2155.md @@ -0,0 +1,26 @@ +# Work Summary — 2026-08-05 21:55 + +## Task +Insight #4 from the `domains/amazon` design review, and the one the user flagged as most consequential: a LISTEN-connection error (dropped connection, Postgres restart, network blip) currently propagates all the way out of `ProcessEvents`, and `main.go`'s top-level shutdown logic treats that the same as a fatal server error - taking down the *entire application*, not just the Amazon event processor. Significant because the same trigger+notify shape is already migrated (SQL-side) for eleven other platforms with no Go processor yet. + +## Design confirmed with user before implementing +Of three options presented (reconnect/retry around LISTEN; decouple event-processing failure from app shutdown in `main.go`; both), the user chose reconnect/retry only, with a warning logged - not touching `main.go`'s shutdown behavior at all. + +## Changes +`domains/amazon/mock.go`: +- New constants `initialListenReconnectBackoff` (1s) and `maxListenReconnectBackoff` (30s). +- New `(*Mocks).reconnectOrStop`: given the error that came off `errCh`, distinguishes an ordinary shutdown (nil error, or `ctx` already done - returns `stop=true`) from a real failure. For a real failure: logs `Warn("lost connection while listening for notifications; reconnecting", "error", ..., "retry_in", backoff)`, waits out the backoff (still respecting `ctx` cancellation), and calls `listenForNotifications` again. Backoff resets to its initial value on a successful reconnect and doubles (capped) on an immediate repeat failure (e.g. the pool itself being unreachable), so a persistently-down DB backs off rather than hot-looping. +- `ProcessEvents`'s `select` no longer returns on `errCh`/closed-`notifCh` directly - both paths now go through `reconnectOrStop`, looping back into the main loop with fresh channels instead of exiting. + +## Verification +- Before writing the real test, manually confirmed the mechanism end-to-end with a throwaway program (scratchpad, not committed): started `ProcessEvents`, looked up its LISTEN connection's backend PID via `pg_stat_activity` (matching on `query = 'LISTEN mock_shop_amazon_event_inserted'`, which Postgres keeps showing while a connection sits idle), and killed it with `pg_terminate_backend`. Logs showed the exact expected sequence: the real connection error, the warning with `retry_in=1s`, then processing resuming on schedule. +- New test `TestProcessEvents_ReconnectsAfterListenConnectionDrops`, using that same real-kill technique (via a new `terminateListenConnection` test helper) rather than a simulated failure: confirms `ProcessEvents` does *not* return after the connection is killed, and that the reactive path (insert → dispatch) still works afterward, proving the reconnect actually restored a working LISTEN. +- `go build ./...` / `go vet ./...` clean. +- `go test ./domains/amazon/... -v -race`: all 7 tests pass (6 existing + 1 new). +- 10x repeated runs (`-count=1 -race`) with no flakes, ~3.5-3.7s each (the new test alone takes ~2s, waiting out the real 1s backoff plus recovery time). +- `make test`: full suite green. +- Deliberately **skipped** `make test-against-dev-db` this round: a live `go run .` process was found running against the dev DB at verification time, and this test kills a LISTEN connection by matching on query text - safe against the dedicated test DB, but running it against dev risked hitting that live process's own connection instead of (or alongside) the test's. It would have recovered gracefully (that's the entire point of this fix), but there was no need to disrupt a possibly-in-use process just to re-prove what the test DB run already confirmed. Flagged to the user rather than done silently. + +## Follow-ups / not done here +- This closes all four insights from the original `domains/amazon` design review (2026-08-04). `domains/reports` test coverage remains open, deferred by the user to later. +- Worth being aware of for next time: `terminateListenConnection`'s PID lookup matches purely on query text, with no way to scope it to "this specific test's connection" if multiple `Mocks` instances are ever LISTEN-ing concurrently against the same database (e.g. a real dev server running at the same time as `test-against-dev-db`). Not fixed - just something to check for before running this specific test against a shared/live database.