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

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)
 

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")
+	}
+}