mirror of
https://github.com/PerpetualSoftware/pad.git
synced 2026-10-26 06:19:49 +00:00
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
This commit is contained in:
@@ -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",
|
||||
}
|
||||
|
||||
@@ -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,
|
||||
})
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -0,0 +1,4 @@
|
||||
{
|
||||
"completed": true,
|
||||
"event_id": "<string>"
|
||||
}
|
||||
@@ -1,5 +1,6 @@
|
||||
{
|
||||
"already_processed": false,
|
||||
"event_id": "<string>",
|
||||
"processed_at": "<string>"
|
||||
"processed_at": "<string>",
|
||||
"state": "<string>"
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
@@ -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",
|
||||
|
||||
@@ -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;
|
||||
+102
-47
@@ -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,
|
||||
|
||||
Reference in New Issue
Block a user