From 3201342d3d68c134a22f1bdb993b48fb36612bce Mon Sep 17 00:00:00 2001 From: xarmian Date: Sun, 11 Oct 2026 00:02:50 -0400 Subject: [PATCH] feat(cloud): a Stripe event marker is a claim with a lease, so a handler that dies does not lose the event (BUG-3561, pad half) (#2043) * feat(cloud): a Stripe event marker is a claim with a lease, so a handler that dies does not lose the event (BUG-3561, pad half) pad-cloud marked a webhook event processed BEFORE its handler ran, and the mark was final. A sidecar crash, OOM or redeploy mid-handler, or a failed unmark, left the event 'processed' forever: every Stripe retry was acknowledged as a duplicate. On an annual plan cancelled early, the user kept Pro until the period end. The marker now has two states (migration 136 / PG 108; existing rows read as 'done', which is what they meant): - ClaimStripeEvent inserts a new event 'processing' with a 5-minute lease (claimed). A 'done' event is a duplicate. A 'processing' event with a live lease is in_progress. One with an EXPIRED lease is taken over by one conditional UPDATE with a new token: of several racing retries exactly one wins. - POST /api/v1/admin/stripe-event-complete {event_id, processed_at} marks it done after the handler succeeds. It is token-scoped, so a handler whose lease was taken over cannot complete it for the one that took it. - POST /admin/stripe-event-processed keeps already_processed with its old meaning (anything but a fresh claim) and adds 'state', which a newer sidecar feature-detects (lead ruling). A sidecar that predates this never completes; its rows stay 'processing', and a redelivery after the lease re-runs a handler that re-syncs from current Stripe state, which is harmless. Wired like its siblings: cloudAdminPaths, requireCloudMode, the cloud-admin rate limiter, BUG-3361's cloud-only route list, and the cloud contract goldens (processed gains 'state'; complete is new). The NUL census records the two new server-written columns. Test: claim, in-progress, done, an expired lease taken over exactly once by 4 racing claimers, and a stale-token complete as a no-op. Without the takeover, 0 of 4 claim. Claude-Session: https://claude.ai/code/session_01Q9WssCjFuHjSfLABoW4tmS * test(cloud): the contract test gets a roomy cloud-admin limiter (BUG-3561) It calls every sidecar route once from one address to pin response shapes, and with billing-users and stripe-event-complete the list outgrew the real limiter's burst of 10 (the 11th call answered 429). Rate limiting is not what it tests. Claude-Session: https://claude.ai/code/session_01Q9WssCjFuHjSfLABoW4tmS --- internal/server/bug3361_admin_pat_test.go | 1 + internal/server/cloud_contract_test.go | 16 ++ internal/server/handlers_cloud.go | 47 +++++- internal/server/middleware_ratelimit.go | 2 +- internal/server/server.go | 1 + .../admin_stripe_event_complete.json | 4 + .../admin_stripe_event_processed.json | 3 +- internal/store/bug3561_event_lease_test.go | 85 ++++++++++ .../migrations/136_stripe_event_lease.sql | 9 ++ internal/store/nulcolumns.go | 2 + .../pgmigrations/108_stripe_event_lease.sql | 9 ++ internal/store/stripe_events.go | 149 ++++++++++++------ 12 files changed, 278 insertions(+), 50 deletions(-) create mode 100644 internal/server/testdata/cloud_contract/admin_stripe_event_complete.json create mode 100644 internal/store/bug3561_event_lease_test.go create mode 100644 internal/store/migrations/136_stripe_event_lease.sql create mode 100644 internal/store/pgmigrations/108_stripe_event_lease.sql diff --git a/internal/server/bug3361_admin_pat_test.go b/internal/server/bug3361_admin_pat_test.go index 5c6c4b8c8..96513f2f8 100644 --- a/internal/server/bug3361_admin_pat_test.go +++ b/internal/server/bug3361_admin_pat_test.go @@ -167,6 +167,7 @@ var cloudOnlyAdminRoutes = []string{ "POST /api/v1/admin/payment-failed", "POST /api/v1/admin/plan", "POST /api/v1/admin/stripe-customer-id", + "POST /api/v1/admin/stripe-event-complete", "POST /api/v1/admin/stripe-event-processed", "POST /api/v1/admin/stripe-event-unmark", } diff --git a/internal/server/cloud_contract_test.go b/internal/server/cloud_contract_test.go index b389db8d7..5fecfa14d 100644 --- a/internal/server/cloud_contract_test.go +++ b/internal/server/cloud_contract_test.go @@ -11,6 +11,7 @@ import ( "testing" "github.com/PerpetualSoftware/pad/internal/models" + "golang.org/x/time/rate" ) // The responses the pad-cloud sidecar DECODES, captured from the real handlers @@ -68,6 +69,12 @@ func TestCloudContract(t *testing.T) { srv := testServer(t) bootstrapFirstUser(t, srv, "admin@example.com", "Admin") srv.SetCloudMode(cloudContractSecret) + // This test pins response SHAPES and calls every sidecar route once, from + // one address. The real cloud-admin limiter (burst 10) is not what it + // tests, and the route list outgrew it (BUG-3561), so it gets a roomy one. + srv.rateLimiters.CloudAdmin.Stop() + srv.rateLimiters.CloudAdmin = newIPRateLimiter(rateLimitConfig{Rate: rate.Limit(100), Burst: 100}) + t.Cleanup(srv.rateLimiters.CloudAdmin.Stop) u, err := srv.store.CreateUser(models.UserCreate{Email: "buyer@example.com", Name: "Buyer", Password: "pw-contract-x"}) if err != nil { t.Fatal(err) @@ -127,6 +134,15 @@ func TestCloudContract(t *testing.T) { } processed := shapes["admin_stripe_event_processed"] unmarkAt, _ := processed["processed_at"].(string) + // BUG-3561: the claim is completed after the handler succeeds. A second + // event is claimed for that, since unmark below removes the first. + claimed := call("POST", "/api/v1/admin/stripe-event-processed", map[string]any{ + "event_id": "evt_contract_complete", "cloud_secret": cloudContractSecret, + }) + claimedAt, _ := claimed["processed_at"].(string) + shapes["admin_stripe_event_complete"] = call("POST", "/api/v1/admin/stripe-event-complete", map[string]any{ + "event_id": "evt_contract_complete", "processed_at": claimedAt, "cloud_secret": cloudContractSecret, + }) shapes["admin_stripe_event_unmark"] = call("POST", "/api/v1/admin/stripe-event-unmark", map[string]any{ "event_id": "evt_contract", "processed_at": unmarkAt, "cloud_secret": cloudContractSecret, }) diff --git a/internal/server/handlers_cloud.go b/internal/server/handlers_cloud.go index cb3373a77..e69760c65 100644 --- a/internal/server/handlers_cloud.go +++ b/internal/server/handlers_cloud.go @@ -130,6 +130,7 @@ var cloudAdminPaths = map[string]struct{}{ "/api/v1/admin/billing-users": {}, "/api/v1/admin/stripe-event-processed": {}, "/api/v1/admin/stripe-event-unmark": {}, + "/api/v1/admin/stripe-event-complete": {}, "/api/v1/admin/payment-failed": {}, } @@ -1151,7 +1152,13 @@ func (s *Server) handleStripeEventProcessed(w http.ResponseWriter, r *http.Reque } // 3. Atomically record-or-detect-duplicate - alreadyProcessed, processedAt, err := s.store.MarkStripeEventProcessed(input.EventID) + // BUG-3561: a claim with a lease. already_processed keeps its meaning for a + // sidecar that predates the lease (anything but a fresh claim); `state` + // tells a newer one apart: claimed (run, then complete), done (a + // duplicate), in_progress (another handler holds the lease; fail the + // delivery so Stripe retries after it). + state, processedAt, err := s.store.ClaimStripeEvent(input.EventID, store.StripeEventLease) + alreadyProcessed := state != store.StripeEventClaimed if err != nil { writeInternalError(w, err) return @@ -1179,9 +1186,47 @@ func (s *Server) handleStripeEventProcessed(w http.ResponseWriter, r *http.Reque "event_id": input.EventID, "already_processed": alreadyProcessed, "processed_at": processedAt, + "state": state, }) } +// handleStripeEventComplete handles POST /api/v1/admin/stripe-event-complete +// (BUG-3561): the sidecar reports that the handler for a claimed event +// succeeded, so later deliveries are duplicates. Token-scoped like unmark: a +// handler whose lease was taken over cannot complete the event for the one +// that took it (completed: false). A sidecar that predates the lease never +// calls this; its rows stay 'processing' and a redelivery after the lease +// re-runs the handler, which is harmless because every handler re-syncs from +// current Stripe state. +func (s *Server) handleStripeEventComplete(w http.ResponseWriter, r *http.Request) { + var input struct { + EventID string `json:"event_id"` + ProcessedAt string `json:"processed_at"` + CloudSecret string `json:"cloud_secret"` + } + if err := decodeJSON(r, &input); err != nil { + writeError(w, http.StatusBadRequest, "bad_request", "Invalid request body") + return + } + user := currentUser(r) + isAdmin := user != nil && user.Role == "admin" + if !isAdmin { + if !s.validateCloudSecret(input.CloudSecret, w) { + return + } + } + if input.EventID == "" || input.ProcessedAt == "" { + writeError(w, http.StatusBadRequest, "bad_request", "event_id and processed_at are required") + return + } + completed, err := s.store.CompleteStripeEvent(input.EventID, input.ProcessedAt) + if err != nil { + writeInternalError(w, err) + return + } + writeJSON(w, http.StatusOK, map[string]any{"event_id": input.EventID, "completed": completed}) +} + // handleStripeEventUnmark handles POST /api/v1/admin/stripe-event-unmark. // Called by the pad-cloud sidecar as a best-effort rollback when a webhook // handler fails AFTER /stripe-event-processed marked the event as seen diff --git a/internal/server/middleware_ratelimit.go b/internal/server/middleware_ratelimit.go index 0ee0a4209..6dfa60232 100644 --- a/internal/server/middleware_ratelimit.go +++ b/internal/server/middleware_ratelimit.go @@ -741,7 +741,7 @@ func (s *Server) RateLimit(next http.Handler) http.Handler { // Cloud admin endpoints (sidecar → pad): plan changes, Stripe mapping, user lookup if strings.HasPrefix(path, "/api/v1/admin/") { switch path { - case "/api/v1/admin/plan", "/api/v1/admin/stripe-customer-id", "/api/v1/admin/user-by-customer", "/api/v1/admin/user-by-id", "/api/v1/admin/billing-users", "/api/v1/admin/stripe-event-processed", "/api/v1/admin/stripe-event-unmark", "/api/v1/admin/payment-failed": + case "/api/v1/admin/plan", "/api/v1/admin/stripe-customer-id", "/api/v1/admin/user-by-customer", "/api/v1/admin/user-by-id", "/api/v1/admin/billing-users", "/api/v1/admin/stripe-event-processed", "/api/v1/admin/stripe-event-unmark", "/api/v1/admin/stripe-event-complete", "/api/v1/admin/payment-failed": if !s.rateLimiters.CloudAdmin.allow(addr) { slog.Warn("rate limited", "ip", ip, "path", path, "limiter", "cloud_admin") writeRateLimitResponse(w, s.rateLimiters.CloudAdmin.config) diff --git a/internal/server/server.go b/internal/server/server.go index cd3efc8cf..f59fd0af9 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -1844,6 +1844,7 @@ func (s *Server) setupRouter() { r.Get("/billing-users", s.handleListBillingUsers) // Cloud: sidecar reconcile sweep pages billing users (TASK-3564) r.Post("/stripe-event-processed", s.handleStripeEventProcessed) // Cloud: sidecar webhook idempotency (TASK-696) r.Post("/stripe-event-unmark", s.handleStripeEventUnmark) // Cloud: sidecar handler-failure rollback (TASK-736) + r.Post("/stripe-event-complete", s.handleStripeEventComplete) // Cloud: sidecar reports a claimed event handled (BUG-3561) r.Post("/payment-failed", s.handlePaymentFailed) // Cloud: sidecar forwards invoice.payment_failed to trigger email (TASK-712) // Admin Billing dashboard data (TASK-827 / PLAN-825). Proxies diff --git a/internal/server/testdata/cloud_contract/admin_stripe_event_complete.json b/internal/server/testdata/cloud_contract/admin_stripe_event_complete.json new file mode 100644 index 000000000..1e69fee54 --- /dev/null +++ b/internal/server/testdata/cloud_contract/admin_stripe_event_complete.json @@ -0,0 +1,4 @@ +{ + "completed": true, + "event_id": "" +} diff --git a/internal/server/testdata/cloud_contract/admin_stripe_event_processed.json b/internal/server/testdata/cloud_contract/admin_stripe_event_processed.json index 279e0256a..2cc78586f 100644 --- a/internal/server/testdata/cloud_contract/admin_stripe_event_processed.json +++ b/internal/server/testdata/cloud_contract/admin_stripe_event_processed.json @@ -1,5 +1,6 @@ { "already_processed": false, "event_id": "", - "processed_at": "" + "processed_at": "", + "state": "" } diff --git a/internal/store/bug3561_event_lease_test.go b/internal/store/bug3561_event_lease_test.go new file mode 100644 index 000000000..21b084f22 --- /dev/null +++ b/internal/store/bug3561_event_lease_test.go @@ -0,0 +1,85 @@ +package store_test + +import ( + "sync" + "testing" + "time" + + "github.com/PerpetualSoftware/pad/internal/store" +) + +// BUG-3561: the processed-event marker is a claim with a lease, not a final +// mark. A handler that dies after its claim no longer loses the event: the +// next delivery after the lease takes it over. Both dialects, because the +// takeover is a conditional UPDATE racing others and the lease compares as +// text. +func TestStripeEventClaimLease(t *testing.T) { + for _, b := range contentBackends() { + t.Run(b.name, func(t *testing.T) { + s := b.open(t) + + // A new event is claimed; a second delivery inside the lease is in + // progress (which an old sidecar still reads as "already processed"). + st, tok, err := s.ClaimStripeEvent("evt_live", store.StripeEventLease) + if err != nil || st != store.StripeEventClaimed || tok == "" { + t.Fatalf("first claim: %q %q %v", st, tok, err) + } + if st, _, _ := s.ClaimStripeEvent("evt_live", store.StripeEventLease); st != store.StripeEventInProgress { + t.Fatalf("a delivery inside the lease = %q, want in_progress", st) + } + if already, _, _ := s.MarkStripeEventProcessed("evt_live"); !already { + t.Fatal("the old-sidecar answer for an event in progress must be already_processed") + } + + // Completed, it is a duplicate. + if ok, err := s.CompleteStripeEvent("evt_live", tok); err != nil || !ok { + t.Fatalf("complete: %v %v", ok, err) + } + if st, _, _ := s.ClaimStripeEvent("evt_live", store.StripeEventLease); st != store.StripeEventDone { + t.Fatalf("after complete = %q, want done", st) + } + + // THE BUG: a handler claims, then dies (no complete, no unmark). The + // lease runs out, so the next delivery takes the event over, once, + // with a new token, even with several deliveries racing. + _, deadTok, err := s.ClaimStripeEvent("evt_crash", -time.Minute) // a lease already expired + if err != nil { + t.Fatal(err) + } + time.Sleep(1100 * time.Millisecond) // a new second, so the takeover's token differs + var wg sync.WaitGroup + results := make(chan string, 4) + for i := 0; i < 4; i++ { + wg.Add(1) + go func() { + defer wg.Done() + st, _, err := s.ClaimStripeEvent("evt_crash", store.StripeEventLease) + if err != nil { + t.Error(err) + } + results <- st + }() + } + wg.Wait() + close(results) + claimed := 0 + for st := range results { + if st == store.StripeEventClaimed { + claimed++ + } + } + if claimed != 1 { + t.Fatalf("racing deliveries after an expired lease: %d claimed, want exactly 1", claimed) + } + + // The dead handler's late complete holds the old token: a no-op, + // so it cannot complete the event for the handler that took over. + if ok, err := s.CompleteStripeEvent("evt_crash", deadTok); err != nil || ok { + t.Fatalf("a stale-token complete changed the row: %v %v", ok, err) + } + if st, _, _ := s.ClaimStripeEvent("evt_crash", store.StripeEventLease); st != store.StripeEventInProgress { + t.Fatalf("after a stale complete the event must still be in progress under the new lease; got %q", st) + } + }) + } +} diff --git a/internal/store/migrations/136_stripe_event_lease.sql b/internal/store/migrations/136_stripe_event_lease.sql new file mode 100644 index 000000000..6363321de --- /dev/null +++ b/internal/store/migrations/136_stripe_event_lease.sql @@ -0,0 +1,9 @@ +-- BUG-3561: a processed-event marker written BEFORE the sidecar's handler +-- ran was final, so a sidecar crash or redeploy mid-handler (or a failed +-- unmark) lost the event for good. The marker now has two states: +-- 'processing' with a lease while a handler runs, 'done' once the sidecar +-- reports success. A retry finding an expired lease takes the event over. +-- Existing rows were written by sidecars that treated the mark as final, so +-- they are 'done'. +ALTER TABLE stripe_processed_events ADD COLUMN state TEXT NOT NULL DEFAULT 'done'; +ALTER TABLE stripe_processed_events ADD COLUMN lease_until TEXT; diff --git a/internal/store/nulcolumns.go b/internal/store/nulcolumns.go index a6e515aaf..3209f6b1b 100644 --- a/internal/store/nulcolumns.go +++ b/internal/store/nulcolumns.go @@ -402,6 +402,8 @@ var nulExcluded = map[string]string{ "item_builtin_origin.created_at": "server timestamp from now() (TASK-3462)", "item_builtin_origin.builtin_key": "refused by models.BuiltinOrigin.Validate unless it matches a lowercase [a-z0-9-/] path, so no NUL can be stored, on every write including import (TASK-3462)", "item_builtin_origin.seed_hash": "refused by models.BuiltinOrigin.Validate unless it is 64 lowercase hex digits, on every write including import (TASK-3462)", + "stripe_processed_events.state": "fixed server constant 'processing' | 'done', written only by ClaimStripeEvent and CompleteStripeEvent (BUG-3561, migration 136); never request-derived", + "stripe_processed_events.lease_until": "server timestamp, time.Now().UTC().Add(lease) in RFC3339, written only by ClaimStripeEvent (BUG-3561, migration 136); never request-derived", "activities.action": "fixed enum, models.ValidActions", "collections.source": "server enum 'web' | 'cli' | 'mcp' from actorFromRequest's auth shape, never the body (BUG-3447, migration 126); a bundle import carries the bundle's value only after models.ValidCollectionSource accepts it, else ''", "event_outbox.last_error": "Go error string, server-composed", diff --git a/internal/store/pgmigrations/108_stripe_event_lease.sql b/internal/store/pgmigrations/108_stripe_event_lease.sql new file mode 100644 index 000000000..6363321de --- /dev/null +++ b/internal/store/pgmigrations/108_stripe_event_lease.sql @@ -0,0 +1,9 @@ +-- BUG-3561: a processed-event marker written BEFORE the sidecar's handler +-- ran was final, so a sidecar crash or redeploy mid-handler (or a failed +-- unmark) lost the event for good. The marker now has two states: +-- 'processing' with a lease while a handler runs, 'done' once the sidecar +-- reports success. A retry finding an expired lease takes the event over. +-- Existing rows were written by sidecars that treated the mark as final, so +-- they are 'done'. +ALTER TABLE stripe_processed_events ADD COLUMN state TEXT NOT NULL DEFAULT 'done'; +ALTER TABLE stripe_processed_events ADD COLUMN lease_until TEXT; diff --git a/internal/store/stripe_events.go b/internal/store/stripe_events.go index 78bb3fb94..5dd64d391 100644 --- a/internal/store/stripe_events.go +++ b/internal/store/stripe_events.go @@ -7,63 +7,118 @@ import ( "time" ) -// MarkStripeEventProcessed records a Stripe webhook event ID in -// stripe_processed_events if it hasn't been seen before. Returns -// alreadyProcessed=true when the event was ALREADY recorded (i.e. the caller -// is seeing a duplicate Stripe retry and should skip re-processing). Returns -// false for first-time events (caller should run its handler logic). -// -// Always returns processedAt — either the timestamp we just inserted (on a -// fresh mark) or the existing row's timestamp (on a duplicate). Sidecars -// MUST persist this string and pass it back to UnmarkStripeEventProcessed if -// a best-effort rollback is later needed (TASK-736): the unmark uses -// (event_id, processed_at) as a composite key, so a stale unmark from an -// earlier attempt can't delete the fresh marker left by a successful retry. -// -// Implementation uses INSERT ... ON CONFLICT DO NOTHING. On fresh insert we -// return the wall-clock we just wrote; on conflict we SELECT the existing -// row to recover its timestamp. Both cases are success — only real DB -// errors propagate. +// MarkStripeEventProcessed is ClaimStripeEvent with the pre-BUG-3561 answer: +// alreadyProcessed is false only for a claim, and true for a done event and for +// one in progress under another lease, which is what a sidecar that predates +// the lease must keep seeing. processedAt is the claim's token (or the stored +// one); pass it to CompleteStripeEvent, or to UnmarkStripeEventProcessed on a +// handler failure (TASK-736). func (s *Store) MarkStripeEventProcessed(eventID string) (alreadyProcessed bool, processedAt string, err error) { - if eventID == "" { - return false, "", fmt.Errorf("mark stripe event: event_id is required") - } + state, processedAt, err := s.ClaimStripeEvent(eventID, StripeEventLease) + return state != StripeEventClaimed, processedAt, err +} + +// StripeEventLease is how long a claimed event belongs to the sidecar that +// claimed it (BUG-3561). Well above a handler's worst case (the sidecar bounds +// each handler with a shorter deadline) and well below Stripe's retry spacing. +const StripeEventLease = 5 * time.Minute + +// Outcomes of ClaimStripeEvent. +const ( + // StripeEventClaimed: run the handler, then CompleteStripeEvent with the token. + StripeEventClaimed = "claimed" + // StripeEventDone: a handler already succeeded; acknowledge the duplicate. + StripeEventDone = "done" + // StripeEventInProgress: another handler holds a live lease. Do not run; + // fail the delivery so Stripe retries after the lease. + StripeEventInProgress = "in_progress" +) + +// ClaimStripeEvent takes a Stripe event for handling (BUG-3561). The marker +// used to be final the moment it was written, BEFORE the handler ran, so a +// sidecar that died mid-handler (or whose unmark failed) left the event +// "processed" forever and every retry was acknowledged as a duplicate. Now: +// +// - a new event is inserted 'processing' with a lease, and is claimed; +// - a 'done' event is a duplicate; +// - a 'processing' event whose lease has EXPIRED is taken over by one +// conditional UPDATE (a new token and a new lease), so of two concurrent +// retries exactly one wins, and the earlier handler's late complete or +// unmark, holding the old token, changes nothing; +// - a 'processing' event with a live lease is in progress elsewhere. +// +// processedAt is the token: CompleteStripeEvent and UnmarkStripeEventProcessed +// act only on the row still carrying it. Timestamps are UTC RFC3339 at second +// resolution, a fixed width, so the lease compares lexically on both dialects. +func (s *Store) ClaimStripeEvent(eventID string, lease time.Duration) (state, processedAt string, err error) { + if eventID == "" { + return "", "", fmt.Errorf("claim stripe event: event_id is required") + } + t := time.Now().UTC() + ts := t.Format(time.RFC3339) + until := t.Add(lease).Format(time.RFC3339) - ts := now() res, err := s.db.Exec( - s.q(`INSERT INTO stripe_processed_events (event_id, processed_at) VALUES (?, ?) ON CONFLICT (event_id) DO NOTHING`), - eventID, ts, + s.q(`INSERT INTO stripe_processed_events (event_id, processed_at, state, lease_until) VALUES (?, ?, 'processing', ?) ON CONFLICT (event_id) DO NOTHING`), + eventID, ts, until, ) if err != nil { - return false, "", fmt.Errorf("mark stripe event %s: %w", eventID, err) + return "", "", fmt.Errorf("claim stripe event %s: %w", eventID, err) + } + if n, rerr := res.RowsAffected(); rerr == nil && n == 1 { + return StripeEventClaimed, ts, nil } - n, rowsErr := res.RowsAffected() - if rowsErr == nil && n == 1 { - // Fresh insert — ts is authoritative and the unmark token. - return false, ts, nil + // Take over an expired lease. The token must change, or the earlier + // handler could still complete or unmark the row it no longer owns; a + // takeover inside the same second as the stored token is impossible, + // since the lease is minutes long. + res, err = s.db.Exec( + s.q(`UPDATE stripe_processed_events SET processed_at = ?, lease_until = ? WHERE event_id = ? AND state = 'processing' AND (lease_until IS NULL OR lease_until < ?)`), + ts, until, eventID, ts, + ) + if err != nil { + return "", "", fmt.Errorf("take over stripe event %s: %w", eventID, err) + } + if n, rerr := res.RowsAffected(); rerr == nil && n == 1 { + return StripeEventClaimed, ts, nil } - // Conflict (or driver couldn't report RowsAffected). Recover the - // existing row's processed_at so the caller has a token matching the - // row currently in the table — without this a retried-and-successful - // handler has no way for its future unmark call to target the right - // marker. - var existing string - selErr := s.db.QueryRow( - s.q(`SELECT processed_at FROM stripe_processed_events WHERE event_id = ?`), - eventID, - ).Scan(&existing) - if selErr != nil { - // The row really isn't there (possible under a driver that didn't - // report RowsAffected AND a concurrent delete) — treat as fresh - // so the handler can still run. processedAt=ts may not match the - // actual DB row in this unusual case; that's acceptable because - // the sidecar handler is idempotent at the application level (set - // plan is an UPSERT, webhook handlers are defensively safe). - return false, ts, nil + var st, existing string + if err := s.db.QueryRow( + s.q(`SELECT state, processed_at FROM stripe_processed_events WHERE event_id = ?`), eventID, + ).Scan(&st, &existing); err != nil { + // The row vanished between the statements (a concurrent unmark or + // prune). Claim it as new: every handler re-syncs from current Stripe + // state, so a rerun is safe. + return StripeEventClaimed, ts, nil } - return true, existing, nil + if st == "done" { + return StripeEventDone, existing, nil + } + return StripeEventInProgress, existing, nil +} + +// CompleteStripeEvent marks a claimed event done after its handler succeeded +// (BUG-3561). It acts only on the row still carrying the claim's token, so a +// handler whose lease was taken over cannot complete the event for the one +// that took it. Returns whether the row changed. +func (s *Store) CompleteStripeEvent(eventID, processedAt string) (bool, error) { + if eventID == "" || processedAt == "" { + return false, fmt.Errorf("complete stripe event: event_id and processed_at are required") + } + res, err := s.db.Exec( + s.q(`UPDATE stripe_processed_events SET state = 'done', lease_until = NULL WHERE event_id = ? AND processed_at = ? AND state = 'processing'`), + eventID, processedAt, + ) + if err != nil { + return false, fmt.Errorf("complete stripe event %s: %w", eventID, err) + } + n, err := res.RowsAffected() + if err != nil { + return false, nil + } + return n == 1, nil } // UnmarkStripeEventProcessed deletes a row from stripe_processed_events,