amadan
abono: one lock per subscription from Stripe fetch to commit, so an older snapshot never lands last
Summary
Adds subLocks, an in-process lock per subscription id that honours the context. The webhook, Finish, Adopt and Reconcile each take it before fetching the subscription from Stripe and hold it until their write commits. Reconcile now fetches each listed subscription again inside its lock rather than writing the listed snapshot, which was read outside any lock.
Agent notes
Final review I2, ruling R35. Cost: Reconcile makes one extra GET per subscription per run. In-process only: a deployment with several processes on one database would need a database-level check, as the ruling says. Regression: TestAnOlderSnapshotNeverOverwritesANewerOne, six pairs over Finish, webhook, Reconcile and Adopt; in each, Stripe cancels the subscription right after the first entry point fetched it as active, and the first one waits for the second before writing. All six failed (row left active) before the lock and pass after.
7 files changed,
+215
−17
abono.go+1 −0adopt.go+6 −0checkout.go+7 −0locks.go+58 −0reconcile.go+26 −4sync_test.go+90 −0webhook.go+27 −13
CI passed — run details
CI log
=== vet === go vet ./... === fmt === === test === go test ./... -count=1 ok amadan.net/rastrillo/abono 2.543s ok amadan.net/rastrillo/abono/abonotest 0.051s ok amadan.net/rastrillo/abono/cmd/abono 0.608s ok amadan.net/rastrillo/abono/internal/fakestripe 0.017s ok amadan.net/rastrillo/abono/internal/stripeapi 0.006s ok example 0.116s === race === CGO_ENABLED=1 go test ./... -race -count=1 ok amadan.net/rastrillo/abono 7.652s ok amadan.net/rastrillo/abono/abonotest 1.380s ok amadan.net/rastrillo/abono/cmd/abono 1.829s ok amadan.net/rastrillo/abono/internal/fakestripe 1.059s ok amadan.net/rastrillo/abono/internal/stripeapi 1.018s ok example 1.730s === browser === ok example 0.718s
Intent
diff --git a/abono.go b/abono.go| index 663b1be..9de55d9 100644 |
| --- a/abono.go |
| +++ b/abono.go |
| @@ -82,6 +82,7 @@ type Service struct { |
| cfg Config |
| stripe *stripeapi.Client |
| sandbox *fakestripe.Fake // set with Config.Sandbox |
| + subs subLocks // one per subscription, fetch to commit |
| } |
| // New validates cfg and returns a ready *Service. |
diff --git a/adopt.go b/adopt.go| index 6b223d8..080d47a 100644 |
| --- a/adopt.go |
| +++ b/adopt.go |
| @@ -26,6 +26,12 @@ func (s *Service) Adopt(ctx context.Context, subscriptionID, subject, plan strin |
| if subscriptionID == "" || plan == "" { |
| return errors.New("abono: Adopt needs a subscription and a plan") |
| } |
| + // The subscription's lock from the fetch to the commit (ruling R35). |
| + unlock, err := s.subs.lock(ctx, subscriptionID) |
| + if err != nil { |
| + return err |
| + } |
| + defer unlock() |
| sub, err := s.stripe.GetSubscription(ctx, subscriptionID) |
| if err != nil { |
| return err |
diff --git a/checkout.go b/checkout.go| index 464865c..def1862 100644 |
| --- a/checkout.go |
| +++ b/checkout.go |
| @@ -262,6 +262,13 @@ func (s *Service) Finish(ctx context.Context, sessionID string) (Finished, error |
| if cs.Status != "complete" || cs.Subscription == "" { |
| return f, nil |
| } |
| + // The subscription's lock from the fetch to the commit (ruling R35): |
| + // a webhook for it waits, so whichever read Stripe later writes later. |
| + unlock, err := s.subs.lock(ctx, cs.Subscription) |
| + if err != nil { |
| + return Finished{}, err |
| + } |
| + defer unlock() |
| sub, err := s.stripe.GetSubscription(ctx, cs.Subscription) |
| if err != nil { |
| return Finished{}, err |
diff --git a/locks.go b/locks.go| new file mode 100644 |
| index 0000000..fbbc79a |
| --- /dev/null |
| +++ b/locks.go |
| @@ -0,0 +1,58 @@ |
| +package abono |
| + |
| +import ( |
| + "context" |
| + "sync" |
| +) |
| + |
| +// subLocks serialises the work on one Stripe subscription across every |
| +// entry point that fetches it and writes what it read: the webhook, |
| +// Finish, Reconcile and Adopt (ruling R35). Each holds the subscription's |
| +// lock from before its fetch until its write has committed, so the |
| +// snapshots are written in the order they were read and an older one can |
| +// never overwrite a newer one. SQLite's single writer orders the writes, |
| +// not the fetches, which is why the database alone is not enough. |
| +// |
| +// The lock is in this process: a rastrillo app is one process on one |
| +// SQLite file. Several processes writing one database would need a check |
| +// in the database instead. |
| +type subLocks struct { |
| + mu sync.Mutex |
| + held map[string]*subLock |
| +} |
| + |
| +type subLock struct { |
| + ch chan struct{} // a token while held |
| + users int // holders and waiters, so an idle entry is dropped |
| +} |
| + |
| +// lock waits for id's lock and returns the function that releases it, or |
| +// ctx's error if ctx ends first. |
| +func (l *subLocks) lock(ctx context.Context, id string) (unlock func(), err error) { |
| + l.mu.Lock() |
| + if l.held == nil { |
| + l.held = map[string]*subLock{} |
| + } |
| + e := l.held[id] |
| + if e == nil { |
| + e = &subLock{ch: make(chan struct{}, 1)} |
| + l.held[id] = e |
| + } |
| + e.users++ |
| + l.mu.Unlock() |
| + |
| + done := func() { |
| + l.mu.Lock() |
| + if e.users--; e.users == 0 { |
| + delete(l.held, id) |
| + } |
| + l.mu.Unlock() |
| + } |
| + select { |
| + case e.ch <- struct{}{}: |
| + return func() { <-e.ch; done() }, nil |
| + case <-ctx.Done(): |
| + done() |
| + return nil, ctx.Err() |
| + } |
| +} |
diff --git a/reconcile.go b/reconcile.go| index 3e5ec92..c57686c 100644 |
| --- a/reconcile.go |
| +++ b/reconcile.go |
| @@ -19,7 +19,10 @@ type Report struct { |
| } |
| // Reconcile is the daily tick's work. It lists every subscription on the |
| -// Stripe account and syncs each with what the subscription itself says of |
| +// Stripe account and, for each, takes its lock, fetches it again and syncs |
| +// it (ruling R35: a listed snapshot was read outside the lock, and a |
| +// webhook may have written a newer one since), with what the subscription |
| +// itself says of |
| // its buyer (Checkout's abono_subject metadata, then the customer's |
| // address, as a subscription event does), which makes up for a webhook |
| // Stripe gave up on (a subscription that fails to sync is reported, by id, |
| @@ -39,12 +42,13 @@ func (s *Service) Reconcile(ctx context.Context) (Report, error) { |
| var rep Report |
| var errs []error |
| if s.On() { |
| - err := s.stripe.ListSubscriptions(ctx, func(sub stripeapi.Subscription) error { |
| + err := s.stripe.ListSubscriptions(ctx, func(listed stripeapi.Subscription) error { |
| rep.Subscriptions++ |
| // One subscription failing (the app's Holder or OnChange, say) |
| // must not stop the rest being refreshed on every tick after. |
| - if err := s.sync(ctx, sub, subscriptionHint(sub)); err != nil { |
| - errs = append(errs, fmt.Errorf("abono: reconcile: subscription %s: %w", sub.ID, err)) |
| + if err := s.reconcileOne(ctx, listed.ID); err != nil { |
| + s.noteRefusedKey(err) |
| + errs = append(errs, fmt.Errorf("abono: reconcile: subscription %s: %w", listed.ID, err)) |
| } |
| return nil |
| }) |
| @@ -71,6 +75,24 @@ func (s *Service) Reconcile(ctx context.Context) (Report, error) { |
| return rep, errors.Join(errs...) |
| } |
| +// reconcileOne fetches subscription id again, inside its lock, and syncs |
| +// it. One deleted since the listing is skipped. |
| +func (s *Service) reconcileOne(ctx context.Context, id string) error { |
| + unlock, err := s.subs.lock(ctx, id) |
| + if err != nil { |
| + return err |
| + } |
| + defer unlock() |
| + sub, err := s.stripe.GetSubscription(ctx, id) |
| + switch { |
| + case stripeapi.NotFound(err): |
| + return nil |
| + case err != nil: |
| + return err |
| + } |
| + return s.sync(ctx, sub, subscriptionHint(sub)) |
| +} |
| + |
| // refreshPrices reads every known price again. A price that cannot be read |
| // is reported and the rest carry on. |
| func (s *Service) refreshPrices(ctx context.Context) []error { |
diff --git a/sync_test.go b/sync_test.go| index 7ee8af5..ab5c277 100644 |
| --- a/sync_test.go |
| +++ b/sync_test.go |
| @@ -2,9 +2,12 @@ package abono_test |
| import ( |
| "context" |
| + "fmt" |
| + "net/http" |
| "reflect" |
| "strings" |
| "sync" |
| + "sync/atomic" |
| "testing" |
| "time" |
| @@ -678,3 +681,90 @@ func TestConsecutiveFailedRenewalsKeepTheFirstUnpaidDay(t *testing.T) { |
| t.Error("a new failure after paying does not get its own 30 days") |
| } |
| } |
| + |
| +// TestAnOlderSnapshotNeverOverwritesANewerOne is final review I2 (ruling |
| +// R35). Every entry point fetches the subscription from Stripe and then |
| +// writes it. When Stripe changes it between two such fetches, the one |
| +// that fetched first must not write last: A fetches active, Stripe cancels |
| +// the subscription, B fetches canceled; A then waits for B (as long as B |
| +// can get on) before writing. Unserialised, B writes canceled and A |
| +// overwrites it with active, restoring access. One lock per subscription, |
| +// held from the fetch to the commit, makes B wait for A instead, so the |
| +// row ends canceled whichever entry point is A and which is B. |
| +func TestAnOlderSnapshotNeverOverwritesANewerOne(t *testing.T) { |
| + type entry struct { |
| + name string |
| + run func(t *testing.T, s *abono.Service, n int) error |
| + } |
| + finish := entry{"Finish", func(t *testing.T, s *abono.Service, _ int) error { |
| + _, err := s.Finish(context.Background(), "cs_1") |
| + return err |
| + }} |
| + webhook := entry{"the webhook", func(t *testing.T, s *abono.Service, n int) error { |
| + w := deliver(t, s, event(fmt.Sprintf("evt_%d", n), "customer.subscription.updated", map[string]any{"id": "sub_1"})) |
| + if w.Code != http.StatusOK { |
| + return fmt.Errorf("webhook: %d %s", w.Code, w.Body.String()) |
| + } |
| + return nil |
| + }} |
| + reconcile := entry{"Reconcile", func(t *testing.T, s *abono.Service, _ int) error { |
| + _, err := s.Reconcile(context.Background()) |
| + return err |
| + }} |
| + adopt := entry{"Adopt", func(t *testing.T, s *abono.Service, _ int) error { |
| + return s.Adopt(context.Background(), "sub_1", "", "annual") |
| + }} |
| + for _, pair := range [][2]entry{ |
| + {finish, webhook}, {webhook, finish}, {reconcile, webhook}, {webhook, reconcile}, {adopt, webhook}, {webhook, adopt}, |
| + } { |
| + a, b := pair[0], pair[1] |
| + t.Run(a.name+" fetches first, then "+b.name, func(t *testing.T) { |
| + now := t0 |
| + d := newStripeDouble(t) |
| + rec := &recorder{subject: "42"} |
| + s, db := d.service(t, &now, rec.config()) |
| + seedPlans(t, s) |
| + active := subSpec{ID: "sub_1", Status: "active", Price: "price_A", Customer: "cus_1", Email: "ana@example.com", End: t0.Add(30 * day)} |
| + d.put(t, active) |
| + d.session("cs_1", "complete", "sub_1", "", "ana@example.com") |
| + |
| + var bErr error |
| + bDone := make(chan struct{}) |
| + var first atomic.Bool |
| + d.OnSubscriptionFetch(func() { |
| + // Only A's fetch, the first; B's must not wait on it. |
| + if !first.CompareAndSwap(false, true) { |
| + return |
| + } |
| + // A has its snapshot (active). Stripe cancels now, and B |
| + // starts; A carries on once B is done, or once it is clear |
| + // B is waiting for A. |
| + canceled := active |
| + canceled.Status = "canceled" |
| + d.put(t, canceled) |
| + go func() { defer close(bDone); bErr = b.run(t, s, 2) }() |
| + select { |
| + case <-bDone: |
| + case <-time.After(300 * time.Millisecond): |
| + } |
| + }) |
| + if err := a.run(t, s, 1); err != nil { |
| + t.Fatalf("%s: %v", a.name, err) |
| + } |
| + select { |
| + case <-bDone: |
| + case <-time.After(10 * time.Second): |
| + t.Fatalf("%s never fetched sub_1 itself, so nothing made B run: its snapshot came from somewhere else", a.name) |
| + } |
| + if bErr != nil { |
| + t.Fatalf("%s: %v", b.name, bErr) |
| + } |
| + if got := db.dump(`SELECT status FROM abono_subscriptions WHERE stripe_subscription_id = 'sub_1'`); got != "canceled\n" { |
| + t.Errorf("status = %q, want canceled: the older snapshot was written last", got) |
| + } |
| + if hasPlan(t, s, "42", "annual") { |
| + t.Error("a canceled subscription still grants") |
| + } |
| + }) |
| + } |
| +} |
diff --git a/webhook.go b/webhook.go| index 7b14508..e1a4a49 100644 |
| --- a/webhook.go |
| +++ b/webhook.go |
| @@ -165,19 +165,8 @@ func (s *Service) receive(ctx context.Context, body []byte) (int, string) { |
| } |
| if id := e.subscriptionID(); id != "" { |
| - sub, err := s.stripe.GetSubscription(ctx, id) |
| - switch { |
| - case stripeapi.NotFound(err): |
| - // Gone from Stripe (deleted with its account data, or the |
| - // delivery is from another mode). Retrying cannot help. |
| - s.cfg.Logger.Warn("abono: webhook: subscription no longer exists", "event", e.ID, "type", e.Type, "subscription", id) |
| - case err != nil: |
| - return s.fail("webhook: fetch subscription", err, "event", e.ID, "subscription", id) |
| - default: |
| - if err := s.sync(ctx, sub, e.hint(sub)); err != nil { |
| - return s.fail("webhook: sync", err, "event", e.ID, "subscription", id) |
| - } |
| - s.cfg.Logger.Info("abono: synced", "event", e.ID, "type", e.Type, "subscription", id, "status", sub.Status) |
| + if status, msg := s.receiveSubscription(ctx, e, id); status != http.StatusOK { |
| + return status, msg |
| } |
| } |
| @@ -193,6 +182,31 @@ func (s *Service) receive(ctx context.Context, body []byte) (int, string) { |
| return http.StatusOK, "" |
| } |
| +// receiveSubscription fetches subscription id again and syncs it, holding |
| +// its lock from the fetch to the commit (ruling R35). |
| +func (s *Service) receiveSubscription(ctx context.Context, e event, id string) (int, string) { |
| + unlock, err := s.subs.lock(ctx, id) |
| + if err != nil { |
| + return s.fail("webhook: wait for subscription", err, "event", e.ID, "subscription", id) |
| + } |
| + defer unlock() |
| + sub, err := s.stripe.GetSubscription(ctx, id) |
| + switch { |
| + case stripeapi.NotFound(err): |
| + // Gone from Stripe (deleted with its account data, or the |
| + // delivery is from another mode). Retrying cannot help. |
| + s.cfg.Logger.Warn("abono: webhook: subscription no longer exists", "event", e.ID, "type", e.Type, "subscription", id) |
| + case err != nil: |
| + return s.fail("webhook: fetch subscription", err, "event", e.ID, "subscription", id) |
| + default: |
| + if err := s.sync(ctx, sub, e.hint(sub)); err != nil { |
| + return s.fail("webhook: sync", err, "event", e.ID, "subscription", id) |
| + } |
| + s.cfg.Logger.Info("abono: synced", "event", e.ID, "type", e.Type, "subscription", id, "status", sub.Status) |
| + } |
| + return http.StatusOK, "" |
| +} |
| + |
| // ServeSync is the daily tick, POST SyncPath: Reconcile, inside the request |
| // so the instance stays awake for it. Only carlos's bearer token gets in |
| // (403 otherwise). It answers 204, or 500 when Reconcile failed. It is |