amadan

rastrillo / abono Public

abono: a lost event record is retried; reconcile carries on past one bad subscription

Summary

A webhook whose event record cannot be saved now answers 500 so Stripe retries, and Reconcile collects per-subscription sync errors (named by id) and keeps going; only the Stripe listing failing stops the walk. Tests use a quiet logger.
Agent notes
The gorm WARN line printed by the deliberately failing event insert in TestWebhookFailedEventRecordIs500AndRetried is gorm's own logger, not abono's.
claude 685ec1e0a43c1278976c781d0eb93143a696d9b0 parent 4a808b1
5 files changed, +93 −4
  • abono_test.go +6 −0
  • reconcile.go +8 −2
  • reconcile_test.go +42 −0
  • webhook.go +6 −2
  • webhook_test.go +31 −0

CI passed — run details

CI log 
=== vet ===
go vet ./...
=== fmt ===
=== test ===
go test ./... -count=1
ok  	amadan.net/rastrillo/abono	0.728s
ok  	amadan.net/rastrillo/abono/internal/stripeapi	0.016s
=== race ===
CGO_ENABLED=1 go test ./... -race -count=1
ok  	amadan.net/rastrillo/abono	5.098s
ok  	amadan.net/rastrillo/abono/internal/stripeapi	1.021s
=== browser ===
browser: no browser test file yet, nothing to run

Intent

full ledger: v0.1

diff --git a/abono_test.go b/abono_test.go
index 5928684..521bce9 100644
--- a/abono_test.go
+++ b/abono_test.go
@@ -1,8 +1,10 @@
package abono_test
import (
+ "bytes"
"context"
"fmt"
+ "log/slog"
"strings"
"testing"
"time"
@@ -27,6 +29,10 @@ func newService(t *testing.T, now *time.Time, cfg abono.Config) (*abono.Service,
t.Fatalf("migrate.Apply: %v", err)
}
cfg.DB = d.G
+ if cfg.Logger == nil {
+ // Quiet: a test that reads the log passes its own.
+ cfg.Logger = slog.New(slog.NewTextHandler(&bytes.Buffer{}, nil))
+ }
cfg.Now = func() time.Time { return *now }
s, err := abono.New(cfg)
if err != nil {
diff --git a/reconcile.go b/reconcile.go
index 01ee2ee..45b974d 100644
--- a/reconcile.go
+++ b/reconcile.go
@@ -20,7 +20,8 @@ type Report struct {
// Reconcile is the daily tick's work. It lists every subscription on the
// Stripe account and syncs each with no hint, which makes up for a webhook
-// Stripe gave up on; it refreshes every price's interval, amount and
+// Stripe gave up on (a subscription that fails to sync is reported, by id,
+// and the rest carry on; only the listing itself failing stops the walk); it refreshes every price's interval, amount and
// currency; and it runs OnChange for every distinct subject that holds any
// subscription row or comp, each in its own transaction. That last step is
// how a comp or a paid period that ran out is told to the app, which no
@@ -38,7 +39,12 @@ func (s *Service) Reconcile(ctx context.Context) (Report, error) {
if s.On() {
err := s.stripe.ListSubscriptions(ctx, func(sub stripeapi.Subscription) error {
rep.Subscriptions++
- return s.sync(ctx, sub, Hint{})
+ // 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, Hint{}); err != nil {
+ errs = append(errs, fmt.Errorf("abono: reconcile: subscription %s: %w", sub.ID, err))
+ }
+ return nil
})
if err != nil {
s.noteRefusedKey(err)
diff --git a/reconcile_test.go b/reconcile_test.go
index b7e91bd..8b00980 100644
--- a/reconcile_test.go
+++ b/reconcile_test.go
@@ -2,9 +2,13 @@ package abono_test
import (
"context"
+ "errors"
+ "strings"
"testing"
"time"
+ "gorm.io/gorm"
+
"amadan.net/rastrillo/abono"
)
@@ -154,3 +158,41 @@ func TestReconcileWithStripeOffStillSweepsComps(t *testing.T) {
t.Errorf("OnChange ran %v", changes)
}
}
+
+// One subscription the app cannot take (Holder fails for it) does not stop
+// the ones listed after it, and is named in the error.
+func TestReconcileCarriesOnPastOneBadSubscription(t *testing.T) {
+ now := t0
+ d := newStripeDouble(t)
+ s, db := d.service(t, &now, abono.Config{
+ Holder: func(_ context.Context, _ *gorm.DB, email, _ string) (string, error) {
+ if email == "bad@example.com" {
+ return "", errors.New("the app refuses " + email)
+ }
+ return "bo", nil
+ },
+ })
+ seedPlans(t, s)
+ // Listed in id order: the failing one first.
+ d.put(t, subSpec{ID: "sub_1", Status: "active", Price: "price_A", Customer: "cus_1", Email: "bad@example.com", End: t0.Add(30 * day)})
+ d.put(t, subSpec{ID: "sub_2", Status: "active", Price: "price_M", Customer: "cus_2", Email: "bo@example.com", End: t0.Add(30 * day)})
+ d.price("price_M", "month", 700, "usd")
+
+ rep, err := s.Reconcile(context.Background())
+ if err == nil || !strings.Contains(err.Error(), "sub_1") {
+ t.Fatalf("Reconcile error = %v, want one naming sub_1", err)
+ }
+ if strings.Contains(err.Error(), "sub_2") {
+ t.Errorf("the error blames the good subscription: %v", err)
+ }
+ if rep.Subscriptions != 2 || rep.Subjects != 1 {
+ t.Errorf("Report = %+v, want 2 subscriptions seen and 1 subject told", rep)
+ }
+ if got := db.dump(subsQuery); got != "sub_2|bo|monthly|active|bo@example.com\n" {
+ t.Errorf("subscriptions:\n%s", got)
+ }
+ // The later steps ran too.
+ if got := db.dump(`SELECT amount FROM abono_prices WHERE stripe_price_id = 'price_M'`); got != "700\n" {
+ t.Errorf("price_M amount = %q", got)
+ }
+}
diff --git a/webhook.go b/webhook.go
index 8f568a3..92c57b5 100644
--- a/webhook.go
+++ b/webhook.go
@@ -165,10 +165,14 @@ func (s *Service) ServeWebhook(w http.ResponseWriter, r *http.Request) {
}
// Recorded after the work, so a failure above is retried in full. Two
- // deliveries of one event at once both sync, which is harmless.
+ // deliveries of one event at once both sync, which is harmless. A record
+ // that cannot be saved is a failure too: answering 200 would stop Stripe
+ // retrying with the deduplication unwritten. Sync is idempotent, so the
+ // retry redoing the work is safe.
if err := s.cfg.DB.WithContext(ctx).Clauses(clause.OnConflict{DoNothing: true}).
Create(&eventRow{ID: e.ID, Type: e.Type, ReceivedAt: s.now()}).Error; err != nil {
- s.cfg.Logger.Warn("abono: record event", "event", e.ID, "err", err)
+ s.fail(w, "webhook: record event", err, "event", e.ID)
+ return
}
w.WriteHeader(http.StatusOK)
}
diff --git a/webhook_test.go b/webhook_test.go
index bf27476..3df88e6 100644
--- a/webhook_test.go
+++ b/webhook_test.go
@@ -377,3 +377,34 @@ func TestSyncFailureIs500(t *testing.T) {
s.ServeSync(w, r)
wantStatus(t, w, http.StatusInternalServerError)
}
+
+// A deduplication record that cannot be saved is a failed delivery: Stripe
+// must retry, and the retry (sync being idempotent) succeeds.
+func TestWebhookFailedEventRecordIs500AndRetried(t *testing.T) {
+ now := t0
+ d := newStripeDouble(t)
+ rec := &recorder{}
+ s, db := d.service(t, &now, rec.config())
+ seedPlans(t, s)
+ d.put(t, subSpec{ID: "sub_1", Status: "active", Price: "price_A", Customer: "cus_1", Email: "ana@example.com", End: t0.Add(30 * day),
+ Metadata: map[string]string{"abono_subject": "ana"}})
+ body := event("evt_1", "customer.subscription.created", map[string]any{"id": "sub_1"})
+
+ db.exec(`CREATE TRIGGER no_events BEFORE INSERT ON abono_events BEGIN SELECT RAISE(ABORT, 'disk full'); END`)
+ wantStatus(t, deliver(t, s, body), http.StatusInternalServerError)
+ if got := db.dump(eventsQuery); got != "" {
+ t.Fatalf("events:\n%s", got)
+ }
+
+ db.exec(`DROP TRIGGER no_events`)
+ wantStatus(t, deliver(t, s, body), http.StatusOK)
+ if got := db.dump(eventsQuery); got != "evt_1|customer.subscription.created\n" {
+ t.Errorf("events after the retry:\n%s", got)
+ }
+ if got := db.dump(subsQuery); got != "sub_1|ana|annual|active|ana@example.com\n" {
+ t.Errorf("subscriptions:\n%s", got)
+ }
+ if _, changes := rec.take(); len(changes) != 1 {
+ t.Errorf("OnChange ran %v, want once (the retry found nothing to change)", changes)
+ }
+}