From c00d03eee704e76969d3887aa8d2855d93074322 Mon Sep 17 00:00:00 2001 From: Bretton Date: Sat, 15 Aug 2026 19:55:55 -0700 Subject: [PATCH] fix(outbound): count only parks that applied, and close the review's paper cuts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Second-opinion findings: a park whose fence refused (stale claim, terminal row) still counted in tidepool_outbound_parked — the one signal an operator sizes a held queue by — and the bounced claim's +1 is exactly the abandoned-claim leak, invisible precisely during a degraded-DB incident. park/parkCausal now settle through one countPark tail: applied parks count, bounced ones log at Warn and do not. Also: the release statements are built once at package init (no per-call Sprintf, no future %-in-SQL edge); models.go documents Attempts' refunded- counter semantics; FOLLOWUPS' abandoned-claim entry no longer calls a permanent budget cost "cosmetic" nor describes the counter backwards; two guard tests pin settleLater's deliberate Release (settle-holds stay charged) and the state-half of the fence via a poisoned row with a live token. Co-Authored-By: Claude Fable 5 --- DEPLOY.md | 4 +- FOLLOWUPS.md | 14 +++- internal/outbound/park_bounce_test.go | 83 +++++++++++++++++++ internal/outbound/park_budget_test.go | 16 +++- internal/outbound/purge_held_vote_test.go | 11 ++- internal/outbound/worker.go | 30 ++++++- internal/store/models.go | 7 +- internal/store/outbound_deliveries.go | 21 +++-- internal/store/outbound_delivery_park_test.go | 42 ++++++++++ 9 files changed, 206 insertions(+), 22 deletions(-) create mode 100644 internal/outbound/park_bounce_test.go diff --git a/DEPLOY.md b/DEPLOY.md index 5e8886b..8c4d8a3 100644 --- a/DEPLOY.md +++ b/DEPLOY.md @@ -145,8 +145,8 @@ failure is a retry, not a poison. `DefaultMaxDeliveryAttempts = 8` ⚠️ **It is a continuous write load, though.** A parked head is re-claimed and re-parked every `parkDelay = 5 * time.Second` (`worker.go:560`), so a held switch costs one claim plus one `UPDATE` per parked ordering-key head per ~5s, -for as long as it is engaged — indefinitely, since nothing now ends the cycle on -its own. That churn is the remaining argument for **preferring +for as long as it is engaged — indefinitely, since nothing ends the cycle on its +own. That churn is the remaining argument for **preferring `OUTBOUND_WORKERS=0` to park everything** — see [Rollback](#rollback) — and it is a cost argument, not a safety one. diff --git a/FOLLOWUPS.md b/FOLLOWUPS.md index 4e2bb6a..3ffc727 100644 --- a/FOLLOWUPS.md +++ b/FOLLOWUPS.md @@ -87,9 +87,17 @@ task documents and git history rather than this list. *Two residual limits, both pre-existing and neither closed by this.* - **An abandoned claim still leaks its `+1` forever.** A lapsed lease, or a crash between `ClaimNext` and any settle, leaves an increment with nobody - holding the fence to hand it back. Shared with the real-failure path and - bounded by the lease, so it is cosmetic — but it is why `attempts` is a - count of claims that settled, not of POSTs. + holding the fence to hand it back — so `attempts` counts claims CHARGED AND + NEVER HANDED BACK: deliveries genuinely tried, plus abandoned claims. Each + leak permanently costs the delivery one retry of real poison budget; the + lease bounds only how soon the row is re-claimable, not the loss. Rare and + bounded at **at most one lost retry per abandoned claim**, and shared with + the real-failure path, so it is not a park defect — but it is the reason a + poisoned row can read above `maxAttempts`. A park that the fence REFUSED now + logs at Warn (`park did not apply: claim lost or row terminal`, + `internal/outbound/worker.go`) and is not counted in + `tidepool_outbound_parked`, which is what makes such a leak diagnosable + rather than merely visible in the column. - **A park writes `last_status_code = 0`,** clobbering a real status a PRIOR failed attempt recorded. The divergence sweep reads that column as its answered-or-silent discriminator (`COALESCE(d.last_status_code, 0) > 0`, diff --git a/internal/outbound/park_bounce_test.go b/internal/outbound/park_bounce_test.go new file mode 100644 index 0000000..7d8b1cb --- /dev/null +++ b/internal/outbound/park_bounce_test.go @@ -0,0 +1,83 @@ +package outbound + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "tidepool/internal/store" +) + +// A park that the fence REFUSED is not a park. Both park helpers ask the store +// to hold the delivery and the store answers applied=false when the caller no +// longer owns the claim — and an operator reading tidepool_outbound_parked has +// no other way to tell a queue that is genuinely held from a stale worker +// shouting at a row somebody else moved on. +// +// The counter is the only observable here on purpose: the ROW is already +// correct in this scenario (the fence protects it, as the store tests pin), so +// the bounce is invisible everywhere except the metric. That is exactly the kind +// of miscount that makes a parked-delivery dashboard lie during an incident, +// which is when it is read. +func TestWorker_BouncedParkIsNotCounted(t *testing.T) { + for _, tc := range []struct { + name string + class string + park func(w *Worker, ctx context.Context, d *store.OutboundDelivery) error + }{ + { + name: "switch park", + class: "switch_parked", + park: func(w *Worker, ctx context.Context, d *store.OutboundDelivery) error { + return w.park(ctx, d, "switch_parked", "outbound kill switch engaged") + }, + }, + { + name: "causal park", + class: "parent_pending", + park: func(w *Worker, ctx context.Context, d *store.OutboundDelivery) error { + return w.parkCausal(ctx, d, "parent_pending", "waiting for bridge-origin parent") + }, + }, + } { + t.Run(tc.name, func(t *testing.T) { + conn := workerTestDB(t) + ctx := context.Background() + seedWorkerActor(t, conn, true, false) + id := seedDelivery(t, conn, "Create", "", createPayload("x")) + deliveries := store.NewOutboundDeliveries(conn) + + // A worker claims the delivery and then wedges. + stale, err := deliveries.ClaimNext(ctx, time.Minute) + require.NoError(t, err) + require.NotNil(t, stale.ClaimedUntil) + require.Equal(t, 1, stale.Attempts) + + // Its lease lapses and a second worker takes the row. + _, err = conn.ExecContext(ctx, + `UPDATE outbound_deliveries SET claimed_until = now() - interval '1 minute' WHERE activity_id = $1`, id) + require.NoError(t, err) + reclaimed, err := deliveries.ClaimNext(ctx, time.Minute) + require.NoError(t, err) + require.Equal(t, 2, reclaimed.Attempts, "the second worker owns the claim now") + + // The wedged worker wakes up holding a token nobody honours and + // tries to park what it thinks is still its delivery. + w := newWorker(t, conn, &fakeSender{}, nil) + before := metricParked.Value() + require.NoError(t, tc.park(w, ctx, stale)) + + assert.Equal(t, before, metricParked.Value(), + "a park the fence REFUSED must not be counted: tidepool_outbound_parked is how an "+ + "operator sizes the held queue, and a stale worker bouncing off the fence held nothing") + + got := getDelivery(t, conn, id) + assert.Equal(t, 2, got.Attempts, + "and the bounced park changes nothing about the row — the current claim's attempt stands") + assert.Equal(t, store.DeliveryStatePending, got.State) + }) + } +} diff --git a/internal/outbound/park_budget_test.go b/internal/outbound/park_budget_test.go index a26b2c0..d2232a0 100644 --- a/internal/outbound/park_budget_test.go +++ b/internal/outbound/park_budget_test.go @@ -32,9 +32,15 @@ import ( // failure this test is here to catch. func clearParkDelay(t *testing.T, conn *sql.DB, activityID string) { t.Helper() - _, err := conn.ExecContext(context.Background(), + result, err := conn.ExecContext(context.Background(), `UPDATE outbound_deliveries SET next_attempt_at = now() WHERE activity_id = $1`, activityID) require.NoError(t, err) + // A helper that silently matches no row would turn every caller's loop into + // a claim that never happens, failing somewhere far from the drift that + // caused it. + affected, err := result.RowsAffected() + require.NoError(t, err) + require.Equal(t, int64(1), affected, "clearParkDelay must rewind exactly the delivery under test") } func TestWorker_KillSwitchDoesNotSpendPoisonBudget(t *testing.T) { @@ -52,9 +58,11 @@ func TestWorker_KillSwitchDoesNotSpendPoisonBudget(t *testing.T) { } return nil }} - // MaxAttempts 3 is the package default; BackoffBase must be REAL time here - // (the helper's 1ms default would put next_attempt_at in the past by the - // time we read it, making the reschedule assertion vacuous). + // Both of these are spelled out rather than inherited: MaxAttempts 3 is the + // TEST HELPER's default (the package default is 8), and a helper default is + // not a contract this test should silently depend on. BackoffBase must be + // REAL time here — the helper's 1ms would put next_attempt_at in the past by + // the time we read it, making the reschedule assertion vacuous. w := newWorker(t, conn, sender, func(o *WorkerOptions) { o.Switches = switches o.MaxAttempts = 3 diff --git a/internal/outbound/purge_held_vote_test.go b/internal/outbound/purge_held_vote_test.go index 3daccea..9f7739a 100644 --- a/internal/outbound/purge_held_vote_test.go +++ b/internal/outbound/purge_held_vote_test.go @@ -120,6 +120,11 @@ func newHeldVoteWorld(t *testing.T) *heldVoteWorld { require.Equal(t, store.DeliveryStatePending, held.State) require.Equal(t, store.DeliveryHeldForSettlement, held.LastErrorClass, "precondition: the delivery is held for settlement, so the worker WILL come back to it") + require.Equal(t, 1, held.Attempts, + "a settlement hold KEEPS its attempt, unlike a park. The asymmetry is deliberate: a park "+ + "is a hold nobody tried, while this one already POSTed and its retries are spaced by a "+ + "backoff computed from this very counter — hand the attempt back and a persistently "+ + "failing local write spins at the base delay forever") stored, err := votes.GetByATURI(ctx, pvVoteATURI) require.NoError(t, err) require.Equal(t, store.DeliveredStatePending, stored.DeliveredState, @@ -149,9 +154,13 @@ func (w *heldVoteWorld) settle(t *testing.T) { w.faulty.failSet = false _, err := w.worker.DeliverNext(context.Background()) require.NoError(t, err) - require.Equal(t, store.DeliveryStateDelivered, getDelivery(t, w.conn, w.likeID).State, + settled := getDelivery(t, w.conn, w.likeID) + require.Equal(t, store.DeliveryStateDelivered, settled.State, "precondition: the held settlement completed and the delivery reached its terminal "+ "state — this is the recovery working, not a fault") + require.Equal(t, 2, settled.Attempts, + "and the ledger counted BOTH claims: settlement retries accumulate, so each one waits "+ + "longer than the last") } func TestPurge_DoesNotLeaveAHeldVoteStandingOnAPeer(t *testing.T) { diff --git a/internal/outbound/worker.go b/internal/outbound/worker.go index c4472c6..3862440 100644 --- a/internal/outbound/worker.go +++ b/internal/outbound/worker.go @@ -571,14 +571,36 @@ const parkDelay = 5 * time.Second // attempt cap only through attempts it actually made. func (w *Worker) park(ctx context.Context, delivery *store.OutboundDelivery, class, reason string) error { next := time.Now().Add(parkDelay) - _, _, err := w.deliveries.ReleaseParked(ctx, delivery.ActivityID, delivery.TargetInbox, class, reason, 0, next, *delivery.ClaimedUntil) + _, applied, err := w.deliveries.ReleaseParked(ctx, delivery.ActivityID, delivery.TargetInbox, class, reason, 0, next, *delivery.ClaimedUntil) if err != nil { return fmt.Errorf("park delivery %s: %w", delivery.ActivityID, err) } - metricParked.Add(1) + w.countPark(delivery, class, applied) return nil } +// countPark records a park THE FENCE ACCEPTED, and only that one. +// +// tidepool_outbound_parked is how an operator sizes a held queue, and it is the +// only signal that can: a parked row still reads `pending`, exactly like one +// merely waiting its turn. A stale worker bouncing off a claim somebody else +// owns held nothing, so counting it inflates the one number an incident is read +// through. +// +// The bounce is logged rather than passed over in silence. The ROW is safe — the +// fence saw to that — but the wedged claim's own +1 stays on the ledger with +// nobody left to hand it back (the abandoned-claim leak in FOLLOWUPS.md), and +// this line is the only thing that connects an operator's "attempts climbing +// while the switch is held" to a lapsed lease rather than to the park path. +func (w *Worker) countPark(delivery *store.OutboundDelivery, class string, applied bool) { + if !applied { + w.logger.Warn("park did not apply: claim lost or row terminal", + "activity", delivery.ActivityID, "inbox", delivery.TargetInbox, "class", class) + return + } + metricParked.Add(1) +} + // parkCausal holds a causally-ineligible delivery WITHOUT a future delay: a held // child must become claimable the instant its bridge-origin parent is accepted // (in practice the parent, a lower-seq delivery on the same serial line, is @@ -590,11 +612,11 @@ func (w *Worker) park(ctx context.Context, delivery *store.OutboundDelivery, cla // causalStatus poisons on the CausalWaitBudget deadline, so however many times a // child cycles through this hold, what ends the wait is elapsed time. func (w *Worker) parkCausal(ctx context.Context, delivery *store.OutboundDelivery, class, reason string) error { - _, _, err := w.deliveries.ReleaseParked(ctx, delivery.ActivityID, delivery.TargetInbox, class, reason, 0, time.Now(), *delivery.ClaimedUntil) + _, applied, err := w.deliveries.ReleaseParked(ctx, delivery.ActivityID, delivery.TargetInbox, class, reason, 0, time.Now(), *delivery.ClaimedUntil) if err != nil { return fmt.Errorf("park (causal) delivery %s: %w", delivery.ActivityID, err) } - metricParked.Add(1) + w.countPark(delivery, class, applied) return nil } diff --git a/internal/store/models.go b/internal/store/models.go index 99e4c81..218ef02 100644 --- a/internal/store/models.go +++ b/internal/store/models.go @@ -479,7 +479,12 @@ type OutboundDelivery struct { OrderingKey string // State is the delivery's fate (pending until terminal). State DeliveryState - // Attempts counts how many times a worker claimed this delivery. + // Attempts is a REFUNDED counter, not a claim count: ClaimNext charges one + // on every claim, and a park that the fence applied hands its own charge + // back (ReleaseParked). What is left standing is claims charged and never + // handed back — deliveries genuinely TRIED, plus abandoned claims whose + // worker died or lost its lease before settling. The attempt cap counts + // this, so it counts attempts made rather than holds endured. Attempts int // NextAttemptAt is the retry-backoff schedule; claimable when <= now. NextAttemptAt time.Time diff --git a/internal/store/outbound_deliveries.go b/internal/store/outbound_deliveries.go index 1514df6..5362cbd 100644 --- a/internal/store/outbound_deliveries.go +++ b/internal/store/outbound_deliveries.go @@ -220,7 +220,11 @@ func (r *postgresOutboundDeliveries) MarkDelivered(ctx context.Context, activity // gaining retries it already used, visible nowhere until a poison budget that // should have stopped never does. // -// The slot is filled from a CONSTANT below and never from input. +// The slot is filled from a CONSTANT below and never from input, and both +// expansions happen ONCE at package init (releaseQuery / releaseParkedQuery +// under it) rather than per call — so there are exactly two finished statements +// in this package, neither of which can be handed a runtime string, and no +// future `%` written into this SQL can be mangled by a formatting pass. const releaseStatement = ` WITH updated AS ( UPDATE outbound_deliveries @@ -243,6 +247,13 @@ const releaseStatement = ` const handBackClaimAttempt = `, attempts = GREATEST(attempts - 1, 0)` +// The two finished statements, expanded once at init: a failure keeps its +// attempt, a park hands its own back. +var ( + releaseQuery = fmt.Sprintf(releaseStatement, "") + releaseParkedQuery = fmt.Sprintf(releaseStatement, handBackClaimAttempt) +) + func (r *postgresOutboundDeliveries) Release(ctx context.Context, activityID, targetInbox, errorClass, excerpt string, lastStatusCode int, nextAttempt, claimToken time.Time) (bool, bool, error) { // Fencing + non-terminal guard: only the current claim holder reschedules // (claimed_until == claimToken, state = 'pending'), so a stale worker's late @@ -250,9 +261,7 @@ func (r *postgresOutboundDeliveries) Release(ctx context.Context, activityID, ta // terminal state. The row stays pending with the lease cleared so a retry // can re-claim after the backoff. The attempt ClaimNext charged stays // charged: this delivery was TRIED. - query := fmt.Sprintf(releaseStatement, "") - - return r.markResult(ctx, "release", query, + return r.markResult(ctx, "release", releaseQuery, activityID, targetInbox, nextAttempt.UTC(), errorClass, excerpt, lastStatusCode, claimToken.UTC()) } @@ -274,9 +283,7 @@ func (r *postgresOutboundDeliveries) Release(ctx context.Context, activityID, ta // short-circuits before re-POSTing it, so it can never re-poison and its growing // attempt count only widens the backoff between bookkeeping retries. func (r *postgresOutboundDeliveries) ReleaseParked(ctx context.Context, activityID, targetInbox, errorClass, excerpt string, lastStatusCode int, nextAttempt, claimToken time.Time) (bool, bool, error) { - query := fmt.Sprintf(releaseStatement, handBackClaimAttempt) - - return r.markResult(ctx, "release parked", query, + return r.markResult(ctx, "release parked", releaseParkedQuery, activityID, targetInbox, nextAttempt.UTC(), errorClass, excerpt, lastStatusCode, claimToken.UTC()) } diff --git a/internal/store/outbound_delivery_park_test.go b/internal/store/outbound_delivery_park_test.go index 2c6ee99..2df3fc5 100644 --- a/internal/store/outbound_delivery_park_test.go +++ b/internal/store/outbound_delivery_park_test.go @@ -117,6 +117,48 @@ func TestOutboundDeliveries_ReleaseParkedFencing(t *testing.T) { "the holder's park hands back its OWN claim only: the earlier, genuinely-spent attempt remains") }) + // The fence is two predicates, and the case above only exercises one of + // them: every real terminal writer nulls claimed_until on its way out, so a + // terminal row and a live token never coexist in production and the + // state='pending' half is unreachable by ordinary means. Raw SQL is the only + // way to hold that half up on its own — without it, deleting `state = + // 'pending'` from ReleaseParked's WHERE clause would leave every test green + // while the method happily un-poisoned a decided delivery. + t.Run("poisoned row with a live token is untouched", func(t *testing.T) { + database := deliveryTestDB(t) + activities := NewOutboundActivities(database) + repo := NewOutboundDeliveries(database) + ctx := context.Background() + + seedActivity(t, activities, testActivity()) + _, err := repo.Enqueue(ctx, testDelivery()) + require.NoError(t, err) + + claimed, err := repo.ClaimNext(ctx, time.Minute) + require.NoError(t, err) + require.NotNil(t, claimed.ClaimedUntil) + token := *claimed.ClaimedUntil + + // state only — the lease is left exactly as the claim stamped it, so + // the token below is genuinely CURRENT and only terminality can refuse. + _, err = database.ExecContext(ctx, + `UPDATE outbound_deliveries SET state = 'poisoned' WHERE activity_id = $1`, delActivityID) + require.NoError(t, err) + + _, applied, err := repo.ReleaseParked(ctx, delActivityID, delTargetInbox, + "switch_parked", "kill switch", 0, time.Now().Add(5*time.Second), token) + require.NoError(t, err) + assert.False(t, applied, + "a park must not apply to a POISONED row even when its token is current — terminal is terminal") + + got, err := repo.Get(ctx, delActivityID, delTargetInbox) + require.NoError(t, err) + assert.Equal(t, DeliveryStatePoisoned, got.State, + "the park must never lift a delivery back out of poisoned and into the live queue") + assert.Equal(t, 1, got.Attempts, + "nor rewind the ledger of a row whose outcome is already decided") + }) + t.Run("terminal row is untouched", func(t *testing.T) { database := deliveryTestDB(t) activities := NewOutboundActivities(database) -- 2.51.2