amadan
Start a seat's allowance when its charge starts; bound and retire the proration date
Review round four on the quantity worker (Astra R1, R4; Opus N1, N3, N5). The policy switch stays unset; none of this runs in production. A seat's charge and its allowance now start at the same instant. The provider prorates a raised quantity from the operation's instant, but the assignment was dated when the console observed the change, so a request retried hours later charged the customer for hours whose allowance they never received. A quantity change that is the account's outstanding operation landing now takes effect at the operation's instant, whichever path sees it first, the worker's reconcile or a webhook. When that instant cannot be the assignment's (later history, a frozen month), the assignment is dated when observed and a review holds the account. The operation's instant is held inside its period: never before the period start a faster clock recorded, and nothing is recorded at or past the period end. The compare-and-set refuses a date outside the period. An operation the provider refuses over its date did not execute and never will with that date, so it is retired at once instead of being replayed for 22 hours. The compare-and-set also carries the billing row's revision, which covers plan, band and terms at once. Each clause is now pinned on its own, the subscription clause included. An operation whose target the provider carries after a renewal is recorded as landed_unverified, a new outcome the audit guard admits: a matching quantity is not proof this operation set it. The send-window test now stalls between the provider read and the recording of the operation, and fails when the window is measured from any later instant. Not verified against Stripe: how it treats a proration date outside the period, at its start, or ahead of its clock, and whether its refusal names the parameter. Test mode is not authorized.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
7 files changed,
+425
−48
internal/billingstripe/catalog.go+12 −0internal/billingstripe/client.go+17 −6internal/console/billingquantity.go+66 −20internal/console/billingquantity_test.go+240 −17internal/console/billingquantityaudit.go+8 −4internal/console/billingservice.go+19 −1internal/console/billingstore.go+63 −0
diff --git a/internal/billingstripe/catalog.go b/internal/billingstripe/catalog.go| index cc96a90a4..2e21a08e5 100644 |
| --- a/internal/billingstripe/catalog.go |
| +++ b/internal/billingstripe/catalog.go |
| @@ -18,6 +18,18 @@ func safeErrorEnum(value string) string { |
| } |
| } |
| +// safeErrorParam is safeErrorEnum for the error's "param": only the |
| +// parameter names a caller acts on are kept, so nothing else from the |
| +// provider's response body travels with the error. |
| +func safeErrorParam(value string) string { |
| + switch value { |
| + case "proration_date", "quantity": |
| + return value |
| + default: |
| + return "" |
| + } |
| +} |
| + |
| type Price struct { |
| ProductID string |
| CurrencyOptions map[string]json.RawMessage |
diff --git a/internal/billingstripe/client.go b/internal/billingstripe/client.go| index 170b4e97d..eed378538 100644 |
| --- a/internal/billingstripe/client.go |
| +++ b/internal/billingstripe/client.go |
| @@ -65,6 +65,15 @@ type Error struct { |
| HTTPStatus int // 0 when the request never reached Stripe |
| StripeType string // Stripe's error "type" field, present only for CodeAPIError |
| StripeCode string // Stripe's error "code" field, present only for CodeAPIError |
| + // StripeParam is Stripe's error "param" field: the request parameter |
| + // the refusal is about. Present only for CodeAPIError. |
| + StripeParam string |
| +} |
| + |
| +// RefusedParam reports whether Stripe refused the request outright as |
| +// invalid because of the named parameter. Such a request was not executed. |
| +func (e *Error) RefusedParam(name string) bool { |
| + return e != nil && e.Code == CodeAPIError && e.HTTPStatus == 400 && e.StripeType == "invalid_request_error" && e.StripeParam == name |
| } |
| func (e *Error) Error() string { |
| @@ -140,8 +149,9 @@ func validID(id string) bool { return validIDRe.MatchString(id) } |
| type stripeErrorEnvelope struct { |
| Error struct { |
| - Type string `json:"type"` |
| - Code string `json:"code"` |
| + Type string `json:"type"` |
| + Code string `json:"code"` |
| + Param string `json:"param"` |
| // message and param are deliberately not decoded: Stripe's message |
| // text can interpolate request parameters (card numbers' last four, |
| // email addresses, free-text fields), and this package promises |
| @@ -204,10 +214,11 @@ func (c *Client) do(ctx context.Context, method, path string, form url.Values, i |
| var env stripeErrorEnvelope |
| _ = json.Unmarshal(respBody, &env) // best-effort; body itself never leaves this function |
| return &Error{ |
| - Code: CodeAPIError, |
| - HTTPStatus: resp.StatusCode, |
| - StripeType: safeErrorEnum(env.Error.Type), |
| - StripeCode: safeErrorEnum(env.Error.Code), |
| + Code: CodeAPIError, |
| + HTTPStatus: resp.StatusCode, |
| + StripeType: safeErrorEnum(env.Error.Type), |
| + StripeCode: safeErrorEnum(env.Error.Code), |
| + StripeParam: safeErrorParam(env.Error.Param), |
| } |
| } |
diff --git a/internal/console/billingquantity.go b/internal/console/billingquantity.go| index e494b6451..17a834bc8 100644 |
| --- a/internal/console/billingquantity.go |
| +++ b/internal/console/billingquantity.go |
| @@ -298,7 +298,15 @@ type quantityOperation struct { |
| FromQuantity, Target int64 |
| Generation int64 |
| Behavior, Key string |
| - CreatedAt, AckedAt int64 |
| + // CreatedAt is the instant the operation was computed at. It is the |
| + // proration date the provider is given, and the instant the resulting |
| + // assignment takes effect at (billingservice.go), so what the customer |
| + // pays for and what their allowance covers start together. |
| + CreatedAt, AckedAt int64 |
| + // Revision is billing_accounts.revision as read when the target was |
| + // computed: any change to the billing row since (plan, band, terms, |
| + // status, period, quantity) is a different world. |
| + Revision int64 |
| } |
| // A process sends an operation's request only within this long of reading |
| @@ -320,14 +328,17 @@ func (s *Store) migrateBillingQuantityOperations() error { |
| )`); err != nil { |
| return err |
| } |
| - return s.addColumnIfMissing("billing_quantity_operations", "period_end", `INTEGER NOT NULL DEFAULT 0`) |
| + if err := s.addColumnIfMissing("billing_quantity_operations", "period_end", `INTEGER NOT NULL DEFAULT 0`); err != nil { |
| + return err |
| + } |
| + return s.addColumnIfMissing("billing_quantity_operations", "revision", `INTEGER NOT NULL DEFAULT 0`) |
| } |
| func (s *Store) quantityOperationFor(accountID int64) (quantityOperation, bool, error) { |
| var op quantityOperation |
| - err := s.db.QueryRow(`SELECT account_id,subscription,period_start,period_end,item,from_quantity,target,generation,behavior,idempotency_key,created_at,acked_at |
| + err := s.db.QueryRow(`SELECT account_id,subscription,period_start,period_end,item,from_quantity,target,generation,behavior,idempotency_key,created_at,acked_at,revision |
| FROM billing_quantity_operations WHERE account_id=?`, accountID). |
| - Scan(&op.AccountID, &op.Subscription, &op.PeriodStart, &op.PeriodEnd, &op.Item, &op.FromQuantity, &op.Target, &op.Generation, &op.Behavior, &op.Key, &op.CreatedAt, &op.AckedAt) |
| + Scan(&op.AccountID, &op.Subscription, &op.PeriodStart, &op.PeriodEnd, &op.Item, &op.FromQuantity, &op.Target, &op.Generation, &op.Behavior, &op.Key, &op.CreatedAt, &op.AckedAt, &op.Revision) |
| if errors.Is(err, sql.ErrNoRows) { |
| return op, false, nil |
| } |
| @@ -339,8 +350,11 @@ func (s *Store) quantityOperationFor(accountID int64) (quantityOperation, bool, |
| // transaction, under the membership write lock, it requires that: |
| // |
| // - no operation is outstanding for the account; |
| -// - the billing row is still the subscription, period, billed quantity |
| -// and active status op was read from; |
| +// - the billing row is the very revision op was read from (which covers |
| +// every column at once: plan, band, terms, customer), and is still the |
| +// subscription, period, billed quantity and active status op names; |
| +// - op's instant lies inside that period, so the proration date the |
| +// provider will be given is one it can accept; |
| // - the pending additions and the member count still call for exactly |
| // op.Target, with op.Generation the newest pending addition; |
| // - op.Target is above the billed quantity (an operation only raises). |
| @@ -369,10 +383,13 @@ func (s *Store) beginQuantityOperation(op quantityOperation) (started bool, err |
| if err != nil { |
| return false, err |
| } |
| - if state.Subscription != op.Subscription || state.PeriodStart != op.PeriodStart || state.PeriodEnd != op.PeriodEnd || |
| + if state.Revision != op.Revision || state.Subscription != op.Subscription || state.PeriodStart != op.PeriodStart || state.PeriodEnd != op.PeriodEnd || |
| state.Quantity != op.FromQuantity || state.Status != "active" || op.Target <= state.Quantity { |
| return false, nil |
| } |
| + if op.CreatedAt < state.PeriodStart || op.CreatedAt >= state.PeriodEnd { |
| + return false, nil |
| + } |
| pending, err := pendingQuantityAdditions(tx, op.AccountID, op.Subscription, op.PeriodStart, 0) |
| if err != nil { |
| return false, err |
| @@ -385,12 +402,12 @@ func (s *Store) beginQuantityOperation(op quantityOperation) (started bool, err |
| if err != nil || target != op.Target || pending.Newest != op.Generation { |
| return false, nil |
| } |
| - if _, err := tx.Exec(`INSERT INTO billing_quantity_operations(account_id,subscription,period_start,period_end,item,from_quantity,target,generation,behavior,idempotency_key,created_at,acked_at) |
| - VALUES(?,?,?,?,?,?,?,?,?,?,?,0) |
| + if _, err := tx.Exec(`INSERT INTO billing_quantity_operations(account_id,subscription,period_start,period_end,item,from_quantity,target,generation,behavior,idempotency_key,created_at,acked_at,revision) |
| + VALUES(?,?,?,?,?,?,?,?,?,?,?,0,?) |
| ON CONFLICT(account_id) DO UPDATE SET subscription=excluded.subscription,period_start=excluded.period_start,period_end=excluded.period_end,item=excluded.item, |
| from_quantity=excluded.from_quantity,target=excluded.target,generation=excluded.generation,behavior=excluded.behavior, |
| - idempotency_key=excluded.idempotency_key,created_at=excluded.created_at,acked_at=0`, |
| - op.AccountID, op.Subscription, op.PeriodStart, op.PeriodEnd, op.Item, op.FromQuantity, op.Target, op.Generation, op.Behavior, op.Key, op.CreatedAt); err != nil { |
| + idempotency_key=excluded.idempotency_key,created_at=excluded.created_at,acked_at=0,revision=excluded.revision`, |
| + op.AccountID, op.Subscription, op.PeriodStart, op.PeriodEnd, op.Item, op.FromQuantity, op.Target, op.Generation, op.Behavior, op.Key, op.CreatedAt, op.Revision); err != nil { |
| return false, err |
| } |
| return true, tx.Commit() |
| @@ -401,9 +418,9 @@ func (s *Store) ackQuantityOperation(accountID int64, key string, now time.Time) |
| return err |
| } |
| -// markOperationLanded records an operation that reached the provider in a |
| -// period that has since renewed: the additions it covered, in the period |
| -// it was for, were charged by it and are marked with its own outcome and |
| +// markOperationLanded records an operation whose target the provider |
| +// carries in a period that has since renewed: the additions it covered, in |
| +// the period it was for, are marked with outcome and the operation's own |
| // reference, never as uncharged, and the operation is acknowledged. |
| func (s *Store) markOperationLanded(op quantityOperation, outcome, reference string, now time.Time) error { |
| tx, err := s.db.Begin() |
| @@ -461,6 +478,11 @@ func (s *Server) billingQuantityPass(ctx context.Context, now time.Time) error { |
| return errors.Join(failures...) |
| } |
| +// quantityOutcomeUnverified marks additions whose operation's target the |
| +// provider carries after a renewal. Whether this operation or a person put |
| +// it there is not established. |
| +const quantityOutcomeUnverified = "landed_unverified" |
| + |
| func quantityOutcomeFor(behavior string) string { |
| if behavior == billingstripe.ProrationAlwaysInvoice { |
| return quantityOutcomeCharged |
| @@ -575,9 +597,18 @@ func (s *Server) raiseQuantity(ctx context.Context, acct *Account, behavior stri |
| // proration date included, so a changed policy or a later attempt is a |
| // new request and never a replay with different parameters. A recovery |
| // re-sends the persisted operation, key and date unchanged. |
| - created := time.Now().Unix() |
| + // |
| + // The instant is this process's clock, held inside the period: a clock |
| + // a few seconds behind the one that reconciled a renewal must not |
| + // produce a date before the period it is in, which the provider would |
| + // refuse on every replay. At or past the period end nothing is |
| + // recorded at all. |
| + created := max(time.Now().Unix(), state.PeriodStart) |
| + if created >= state.PeriodEnd { |
| + return quantityStep{done: true}, nil |
| + } |
| op := quantityOperation{AccountID: acct.ID, Subscription: state.Subscription, PeriodStart: state.PeriodStart, PeriodEnd: state.PeriodEnd, Item: item.ID, |
| - FromQuantity: state.Quantity, Target: target, Generation: pending.Newest, Behavior: behavior, CreatedAt: created, |
| + FromQuantity: state.Quantity, Target: target, Generation: pending.Newest, Behavior: behavior, CreatedAt: created, Revision: state.Revision, |
| Key: fmt.Sprintf("carloku-qty-%s-%s-%d-%d-%d-%s-%d", acct.Sqid, state.Subscription, state.PeriodStart, pending.Newest, target, behavior, created)} |
| started, err := s.store.beginQuantityOperation(op) |
| if err != nil { |
| @@ -630,6 +661,17 @@ func (s *Server) sendQuantityOperation(ctx context.Context, state billingState, |
| } |
| updated, err := b.client.UpdateQuantityWith(ctx, op.Item, op.Target, op.Behavior, time.Unix(op.CreatedAt, 0), op.Key) |
| if err != nil { |
| + var refused *billingstripe.Error |
| + if errors.As(err, &refused) && refused.RefusedParam("proration_date") { |
| + // The provider refused the request outright over its date: it |
| + // did not execute, and the same date will be refused on every |
| + // replay (the period moved, or a clock disagreed). The |
| + // operation is retired now, not replayed until it expires. The |
| + // additions stay pending and the next pass computes afresh. |
| + if aerr := s.store.ackQuantityOperation(op.AccountID, op.Key, time.Now()); aerr != nil { |
| + return quantityStep{}, errors.Join(err, aerr) |
| + } |
| + } |
| return quantityStep{}, err |
| } |
| if updated.ID != op.Item || updated.Quantity != op.Target { |
| @@ -668,11 +710,15 @@ func (s *Server) resolveQuantityOperation(ctx context.Context, acct *Account, st |
| case op.PeriodStart != state.PeriodStart: |
| // The subscription renewed while the operation was outstanding. |
| // Nothing is sent for a period that is over. If the provider |
| - // carries the operation's target, it landed before the renewal: |
| - // the additions it covered were charged and are recorded so, never |
| - // as uncharged. Otherwise it never ran and is closed. |
| + // carries the operation's target, it most likely landed before the |
| + // renewal: the additions it covered are recorded against it, never |
| + // as uncharged. But a matching quantity is not proof this operation |
| + // put it there (a person could have set the same number by hand), |
| + // and this code does not read the provider's invoice items, so the |
| + // outcome says exactly that: landed, unverified. Otherwise it never |
| + // ran and is closed. |
| if item.Quantity == op.Target { |
| - return quantityStep{done: true}, s.store.markOperationLanded(op, quantityOutcomeFor(op.Behavior), quantityReference(op.Item, op.Target, op.Behavior), now) |
| + return quantityStep{done: true}, s.store.markOperationLanded(op, quantityOutcomeUnverified, quantityReference(op.Item, op.Target, op.Behavior), now) |
| } |
| return abandon() |
| case item.Quantity == op.Target: |
diff --git a/internal/console/billingquantity_test.go b/internal/console/billingquantity_test.go| index 114a2ae67..55a3dea23 100644 |
| --- a/internal/console/billingquantity_test.go |
| +++ b/internal/console/billingquantity_test.go |
| @@ -66,7 +66,7 @@ func (q *quantityStripe) serve(w http.ResponseWriter, r *http.Request) { |
| if date != 0 && (date < q.start.Unix() || date >= q.start.AddDate(1, 0, 0).Unix()) { |
| q.refused++ |
| w.WriteHeader(http.StatusBadRequest) |
| - _, _ = w.Write([]byte(`{"error":{"type":"invalid_request_error","message":"Cannot specify proration date outside the subscription's current period"}}`)) |
| + _, _ = w.Write([]byte(`{"error":{"type":"invalid_request_error","param":"proration_date","message":"Cannot specify proration date outside the subscription's current period"}}`)) |
| return |
| } |
| if prev, seen := q.byKey[key]; seen { |
| @@ -100,12 +100,16 @@ func (q *quantityStripe) serve(w http.ResponseWriter, r *http.Request) { |
| } |
| } |
| +// quantityFixtureAge is how long before the test the fixture's subscription |
| +// period began. A test that needs an older period sets it and restores it. |
| +var quantityFixtureAge = time.Hour |
| + |
| // newQuantityFixture is a live Community account at its floor of ten |
| // members, billed for ten, with its entitlement published. |
| func newQuantityFixture(t *testing.T, proration string) (*consoleFixture, *quantityStripe, time.Time) { |
| t.Helper() |
| fx := newConsoleFixture(t) |
| - now := time.Now().UTC().Add(-time.Hour) |
| + now := time.Now().UTC().Add(-quantityFixtureAge) |
| q := &quantityStripe{sqid: fx.sqid, quantity: 10, status: "active", start: now} |
| ts := httptest.NewServer(http.HandlerFunc(q.serve)) |
| t.Cleanup(ts.Close) |
| @@ -563,7 +567,9 @@ func TestQuantityWorkerMeasuresThePauseFromTheProviderRead(t *testing.T) { |
| prev := quantityOperationSendWindow |
| quantityOperationSendWindow = 20 * time.Millisecond |
| t.Cleanup(func() { quantityOperationSendWindow = prev }) |
| - fx.srv.billing.quantityBeforeSend = func() { time.Sleep(60 * time.Millisecond) } |
| + // The stall is between the provider read and the recording of the |
| + // operation: a window measured from any later instant would not see it. |
| + fx.srv.billing.quantityBeforeBegin = func() { time.Sleep(60 * time.Millisecond) } |
| if err := fx.srv.billingQuantityPass(context.Background(), time.Now().UTC()); err != nil { |
| t.Fatal(err) |
| } |
| @@ -572,7 +578,7 @@ func TestQuantityWorkerMeasuresThePauseFromTheProviderRead(t *testing.T) { |
| } |
| // Unstalled, the next pass finishes the same operation. |
| quantityOperationSendWindow = prev |
| - fx.srv.billing.quantityBeforeSend = nil |
| + fx.srv.billing.quantityBeforeBegin = nil |
| if err := fx.srv.billingQuantityPass(context.Background(), time.Now().UTC()); err != nil { |
| t.Fatal(err) |
| } |
| @@ -649,15 +655,18 @@ func TestBeginQuantityOperationIsACompareAndSet(t *testing.T) { |
| t.Fatalf("pending %+v %v", pending, err) |
| } |
| good := quantityOperation{AccountID: OpsAccountID, Subscription: "sub_team", PeriodStart: state.PeriodStart, PeriodEnd: state.PeriodEnd, Item: "si_team", |
| - FromQuantity: 10, Target: 11, Generation: pending.Newest, Behavior: billingstripe.ProrationAlwaysInvoice, Key: "k-good", CreatedAt: time.Now().Unix()} |
| + FromQuantity: 10, Target: 11, Generation: pending.Newest, Behavior: billingstripe.ProrationAlwaysInvoice, Key: "k-good", CreatedAt: time.Now().Unix(), Revision: state.Revision} |
| for name, mutate := range map[string]func(*quantityOperation){ |
| - "another subscription": func(op *quantityOperation) { op.Subscription = "sub_other" }, |
| - "another period": func(op *quantityOperation) { op.PeriodStart-- }, |
| - "another period end": func(op *quantityOperation) { op.PeriodEnd++ }, |
| - "a stale billed quantity": func(op *quantityOperation) { op.FromQuantity = 9 }, |
| - "a lower target": func(op *quantityOperation) { op.Target = 10 }, |
| - "a higher target": func(op *quantityOperation) { op.Target = 12 }, |
| - "an older generation": func(op *quantityOperation) { op.Generation-- }, |
| + "another subscription": func(op *quantityOperation) { op.Subscription = "sub_other" }, |
| + "another period": func(op *quantityOperation) { op.PeriodStart-- }, |
| + "another period end": func(op *quantityOperation) { op.PeriodEnd++ }, |
| + "a stale billed quantity": func(op *quantityOperation) { op.FromQuantity = 9 }, |
| + "a lower target": func(op *quantityOperation) { op.Target = 10 }, |
| + "a higher target": func(op *quantityOperation) { op.Target = 12 }, |
| + "an older generation": func(op *quantityOperation) { op.Generation-- }, |
| + "another revision": func(op *quantityOperation) { op.Revision-- }, |
| + "a date before the period": func(op *quantityOperation) { op.CreatedAt = state.PeriodStart - 1 }, |
| + "a date at the period end": func(op *quantityOperation) { op.CreatedAt = state.PeriodEnd }, |
| } { |
| op := good |
| op.Key = "k-" + name |
| @@ -669,6 +678,44 @@ func TestBeginQuantityOperationIsACompareAndSet(t *testing.T) { |
| if _, ok, _ := fx.st.quantityOperationFor(OpsAccountID); ok { |
| t.Fatal("a refused proposal left a row") |
| } |
| + // The subscription clause on its own. The pending addition is rebound |
| + // to another subscription, so a proposal naming that subscription |
| + // agrees with the pending rows on generation and target and with the |
| + // billing row on revision, period and quantity. Only the subscription |
| + // differs from what the account bills, and that alone must refuse it. |
| + // (The audit guard refuses to rewrite an event, rightly. It is taken |
| + // off this test's database so the fixture can be built.) |
| + rows, err := fx.st.db.Query(`SELECT name FROM sqlite_master WHERE type='trigger' AND tbl_name='billing_quantity_events'`) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + var guards []string |
| + for rows.Next() { |
| + var name string |
| + if err := rows.Scan(&name); err != nil { |
| + t.Fatal(err) |
| + } |
| + guards = append(guards, name) |
| + } |
| + rows.Close() |
| + for _, name := range guards { |
| + if _, err := fx.st.db.Exec(`DROP TRIGGER "` + name + `"`); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + rebind := func(sub string) { |
| + t.Helper() |
| + if _, err := fx.st.db.Exec(`UPDATE billing_quantity_events SET subscription=? WHERE account_id=? AND processed_at=0`, sub, OpsAccountID); err != nil { |
| + t.Fatalf("rebind the pending addition: %v", err) |
| + } |
| + } |
| + rebind("sub_other") |
| + other := good |
| + other.Subscription, other.Key = "sub_other", "k-other-subscription" |
| + if started, err := fx.st.beginQuantityOperation(other); err != nil || started { |
| + t.Fatalf("a proposal for a subscription the account does not bill: started=%v err=%v, want refused", started, err) |
| + } |
| + rebind("sub_team") |
| if started, err := fx.st.beginQuantityOperation(good); err != nil || !started { |
| t.Fatalf("the true proposal: started=%v err=%v", started, err) |
| } |
| @@ -710,6 +757,11 @@ func TestQuantityWorkerCannotChargeAcrossARenewal(t *testing.T) { |
| if q.refused != 1 || len(q.posts) != 0 || q.quantity != 10 { |
| t.Fatalf("refused=%d posts=%+v quantity=%d: an old period's seat was charged into the new one", q.refused, q.posts, q.quantity) |
| } |
| + // A request refused over its date did not execute and never will with |
| + // that date: the operation is retired at once, not replayed for a day. |
| + if op, ok, _ := fx.st.quantityOperationFor(OpsAccountID); !ok || op.AckedAt == 0 { |
| + t.Fatalf("the refused operation is still outstanding: %+v", op) |
| + } |
| fx.srv.billing.quantityBeforeSend = nil |
| // Until the console has reconciled the renewal, nothing more is sent. |
| if err := fx.srv.billingQuantityPass(ctx, time.Now().UTC()); err != nil { |
| @@ -775,10 +827,9 @@ func TestQuantityWorkerDatesTheProrationInsideThePeriod(t *testing.T) { |
| } |
| // The update landed but its response was lost, and the subscription then |
| -// renewed before the next pass. The additions the operation covered were |
| -// charged by it, in their own period. They are recorded as charged with |
| -// the operation's reference; no review claims the seat went unbilled, and |
| -// nothing more is sent. |
| +// renewed before the next pass. The additions the operation covered are |
| +// recorded against it, with its reference, as landed and unverified; no |
| +// review claims the seat went unbilled, and nothing more is sent. |
| func TestQuantityWorkerRecordsALandedOperationAfterARenewal(t *testing.T) { |
| fx, q, now := newQuantityFixture(t, billingProrationImmediate) |
| ctx := context.Background() |
| @@ -812,7 +863,9 @@ func TestQuantityWorkerRecordsALandedOperationAfterARenewal(t *testing.T) { |
| if len(q.posts) != 1 { |
| t.Fatalf("sent again after the renewal: %+v", q.posts) |
| } |
| - assertLastAddition(t, fx, quantityOutcomeCharged, "si_team:quantity=11:always_invoice") |
| + // Landed, unverified: the quantity matches the operation's target, and |
| + // nothing here proves the operation is what set it. |
| + assertLastAddition(t, fx, quantityOutcomeUnverified, "si_team:quantity=11:always_invoice") |
| if rev, err := fx.st.billingReview(OpsAccountID, "sub_team", billingQuantityUnchargedReason); err == nil && rev.Occurrences > 0 { |
| t.Fatalf("a charged seat was reported unbilled: %+v", rev) |
| } |
| @@ -824,6 +877,176 @@ func TestQuantityWorkerRecordsALandedOperationAfterARenewal(t *testing.T) { |
| } |
| } |
| +// quantityAssignment is the account's newest recorded assignment. |
| +func quantityAssignment(t *testing.T, fx *consoleFixture) billingAssignmentRow { |
| + t.Helper() |
| + history, err := fx.st.billingInvoiceHistory(OpsAccountID) |
| + if err != nil || len(history) == 0 { |
| + t.Fatalf("assignment history: %v %v", history, err) |
| + } |
| + return history[len(history)-1] |
| +} |
| + |
| +// agedQuantityFixture is a fixture whose period began thirty hours ago, |
| +// with one addition whose operation was recorded twenty hours ago and has |
| +// not reached the provider. |
| +func agedQuantityFixture(t *testing.T) (*consoleFixture, *quantityStripe, quantityOperation) { |
| + t.Helper() |
| + prev := quantityFixtureAge |
| + quantityFixtureAge = 30 * time.Hour |
| + t.Cleanup(func() { quantityFixtureAge = prev }) |
| + fx, q, now := newQuantityFixture(t, billingProrationImmediate) |
| + if err := fx.st.AddMember(OpsAccountID, "eleventh@example.test", RoleMember, now.Add(time.Minute)); err != nil { |
| + t.Fatal(err) |
| + } |
| + q.failPost = true |
| + if err := fx.srv.billingQuantityPass(context.Background(), time.Now().UTC()); err == nil { |
| + t.Fatal("a failed request must surface") |
| + } |
| + q.failPost = false |
| + if _, err := fx.st.db.Exec(`UPDATE billing_quantity_operations SET created_at=? WHERE account_id=?`, time.Now().Add(-20*time.Hour).Unix(), OpsAccountID); err != nil { |
| + t.Fatal(err) |
| + } |
| + op, ok, err := fx.st.quantityOperationFor(OpsAccountID) |
| + if err != nil || !ok || op.AckedAt != 0 { |
| + t.Fatalf("fixture operation: %+v %v %v", op, ok, err) |
| + } |
| + return fx, q, op |
| +} |
| + |
| +// The request failed, and the retry got through twenty hours later. The |
| +// provider prorates the seat from the operation's instant, so the customer |
| +// pays for it from then. The assignment must take effect at that same |
| +// instant, not when the retry was observed: otherwise the customer pays |
| +// for twenty hours of a seat whose allowance they never get. |
| +func TestQuantityRetryDatesTheAllowanceAtTheProrationInstant(t *testing.T) { |
| + fx, q, op := agedQuantityFixture(t) |
| + if err := fx.srv.billingQuantityPass(context.Background(), time.Now().UTC()); err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(q.posts) != 1 || q.posts[0]["proration_date"] != strconv.FormatInt(op.CreatedAt, 10) { |
| + t.Fatalf("posts %+v, want one dated %d", q.posts, op.CreatedAt) |
| + } |
| + got := quantityAssignment(t, fx) |
| + if got.Quantity != 11 || got.EffectiveAt.Unix() != op.CreatedAt { |
| + t.Fatalf("the eleven-seat assignment takes effect at %d, want the proration instant %d (observed twenty hours later)", got.EffectiveAt.Unix(), op.CreatedAt) |
| + } |
| + if rev, err := fx.st.billingReview(OpsAccountID, "sub_team", billingQuantityDateReviewReason(op.CreatedAt)); err == nil && rev.Occurrences > 0 { |
| + t.Fatalf("a datable operation filed a review: %+v", rev) |
| + } |
| + // The allowance the month's bill is set against includes the seat for |
| + // those twenty hours. |
| + history, _ := fx.st.billingInvoiceHistory(OpsAccountID) |
| + var hist []plans.Assignment |
| + for _, h := range history { |
| + hist = append(hist, h.Assignment) |
| + } |
| + var seatSeconds int64 |
| + for i, h := range hist { |
| + end := time.Now().UTC() |
| + if i+1 < len(hist) { |
| + end = hist[i+1].EffectiveAt |
| + } |
| + seatSeconds += h.Quantity * int64(end.Sub(h.EffectiveAt).Seconds()) |
| + } |
| + observedLate := 10*int64(30*3600) + 1*0 // the seat dated when the retry was seen |
| + if seatSeconds < observedLate+19*3600 { |
| + t.Fatalf("seat-seconds recorded %d: the eleventh seat's twenty hours are missing", seatSeconds) |
| + } |
| +} |
| + |
| +// The same landing, seen first by a webhook's reconcile and not by the |
| +// worker's: the assignment is dated at the operation's instant all the same. |
| +func TestQuantityLandingSeenByAWebhookIsDatedAtTheOperation(t *testing.T) { |
| + fx, q, op := agedQuantityFixture(t) |
| + // The provider now carries the target (the request landed; only its |
| + // response was lost), and a subscription-updated webhook arrives. |
| + q.mu.Lock() |
| + q.quantity = 11 |
| + q.mu.Unlock() |
| + if err := fx.srv.reconcileBillingSubscription(context.Background(), "sub_team", "evt_quantity_webhook", time.Now().UTC()); err != nil { |
| + t.Fatal(err) |
| + } |
| + got := quantityAssignment(t, fx) |
| + if got.Quantity != 11 || got.EffectiveAt.Unix() != op.CreatedAt { |
| + t.Fatalf("webhook-observed assignment takes effect at %d, want %d", got.EffectiveAt.Unix(), op.CreatedAt) |
| + } |
| + // The worker then finishes its operation without sending. |
| + if err := fx.srv.billingQuantityPass(context.Background(), time.Now().UTC()); err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(q.posts) != 0 || quantityHeld(t, fx) { |
| + t.Fatalf("posts %+v held %v", q.posts, quantityHeld(t, fx)) |
| + } |
| +} |
| + |
| +// When the operation's instant cannot be the assignment's (something else |
| +// was recorded after it), the assignment is dated when observed and a |
| +// review holds the account: the provider charged from one instant and the |
| +// allowance starts at another, and nobody is billed on that unseen. |
| +func TestQuantityLandingThatCannotBeDatedFilesAReview(t *testing.T) { |
| + fx, q, op := agedQuantityFixture(t) |
| + // A payment failure and recovery, both recorded after the operation's |
| + // instant. |
| + for i, status := range []string{"past_due", "active"} { |
| + state, _ := fx.st.billingAccount(OpsAccountID) |
| + next := state |
| + next.Status, next.LiveMode = status, true |
| + if err := fx.st.applyBillingState(next, state.Revision, "evt_flap_"+status, time.Now().UTC().Add(time.Duration(i)*time.Second)); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + publishQuantityEntitlement(t, fx, time.Now().UTC()) |
| + q.mu.Lock() |
| + q.quantity = 11 |
| + q.mu.Unlock() |
| + observed := time.Now().UTC().Add(5 * time.Second) |
| + if err := fx.srv.reconcileBillingSubscription(context.Background(), "sub_team", "evt_quantity_late", observed); err != nil { |
| + t.Fatal(err) |
| + } |
| + got := quantityAssignment(t, fx) |
| + if got.Quantity != 11 || got.EffectiveAt.Unix() == op.CreatedAt { |
| + t.Fatalf("assignment %+v: dated behind later history", got.Assignment) |
| + } |
| + rev, err := fx.st.billingReview(OpsAccountID, "sub_team", billingQuantityDateReviewReason(op.CreatedAt)) |
| + if err != nil || rev.ResolvedAt != 0 || rev.Occurrences != 1 { |
| + t.Fatalf("no open review for the undatable operation: %+v %v", rev, err) |
| + } |
| +} |
| + |
| +// This console's clock is a few seconds behind the one that reconciled the |
| +// renewal: the period it reads from the database starts in its future. The |
| +// operation's instant is held inside the period, so the provider accepts |
| +// it, and the assignment is dated there without a review. |
| +func TestQuantityOperationIsDatedInsideThePeriodUnderClockSkew(t *testing.T) { |
| + fx, q, _ := newQuantityFixture(t, billingProrationImmediate) |
| + ahead := time.Now().Add(3 * time.Second).Unix() |
| + if _, err := fx.st.db.Exec(`UPDATE billing_accounts SET period_start=?, period_end=? WHERE account_id=?`, ahead, time.Unix(ahead, 0).UTC().AddDate(1, 0, 0).Unix(), OpsAccountID); err != nil { |
| + t.Fatal(err) |
| + } |
| + q.mu.Lock() |
| + q.start = time.Unix(ahead, 0).UTC() |
| + q.mu.Unlock() |
| + if err := fx.st.AddMember(OpsAccountID, "eleventh@example.test", RoleMember, time.Now().UTC()); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := fx.srv.billingQuantityPass(context.Background(), time.Now().UTC()); err != nil { |
| + t.Fatal(err) |
| + } |
| + if q.refused != 0 || len(q.posts) != 1 || q.posts[0]["proration_date"] != strconv.FormatInt(ahead, 10) { |
| + t.Fatalf("refused=%d posts=%+v, want one request dated at the period start %d", q.refused, q.posts, ahead) |
| + } |
| + if got := quantityAssignment(t, fx); got.Quantity != 11 || got.EffectiveAt.Unix() != ahead { |
| + t.Fatalf("assignment %+v, want eleven seats from %d", got.Assignment, ahead) |
| + } |
| + if rev, err := fx.st.billingReview(OpsAccountID, "sub_team", billingQuantityDateReviewReason(ahead)); err == nil && rev.Occurrences > 0 { |
| + t.Fatalf("seconds of clock skew filed a review: %+v", rev) |
| + } |
| + if quantityHeld(t, fx) { |
| + t.Fatal("still held") |
| + } |
| +} |
| + |
| // An operation left unacknowledged with no addition pending (the state an |
| // older build's crash between the mark and the acknowledgement left) is |
| // still found and finished. Nothing is sent. |
diff --git a/internal/console/billingquantityaudit.go b/internal/console/billingquantityaudit.go| index e3767b9e3..070bd34e9 100644 |
| --- a/internal/console/billingquantityaudit.go |
| +++ b/internal/console/billingquantityaudit.go |
| @@ -15,13 +15,17 @@ func (s *Store) migrateQuantityHistoryGuards() error { |
| OR (NEW.change='removal' AND NEW.effect='at_renewal' AND NEW.members_after=NEW.members_before-1)) |
| OR EXISTS(SELECT 1 FROM billing_quantity_events WHERE seq=NEW.seq OR (account_id=NEW.account_id AND generation=NEW.generation)) |
| BEGIN SELECT RAISE(ABORT,'billing_quantity_events_append_only'); END`, |
| - `CREATE TRIGGER IF NOT EXISTS billing_quantity_events_processing_guard BEFORE UPDATE ON billing_quantity_events |
| + // Dropped and recreated, not IF NOT EXISTS: the set of outcomes the |
| + // guard admits is part of its body, and a database created before |
| + // an outcome was added would otherwise keep refusing it. |
| + `DROP TRIGGER IF EXISTS billing_quantity_events_processing_guard`, |
| + `CREATE TRIGGER billing_quantity_events_processing_guard BEFORE UPDATE ON billing_quantity_events |
| WHEN NEW.processed_at<=0 OR NEW.processed_at<OLD.occurred_at_ns/1000000000 |
| - OR NEW.processing_outcome NOT IN ('charged','included','checkout_expired','checkout_abandoned') |
| + OR NEW.processing_outcome NOT IN ('charged','included','landed_unverified','checkout_expired','checkout_abandoned') |
| OR length(trim(NEW.processing_reference))=0 OR length(NEW.processing_reference)>256 |
| BEGIN SELECT RAISE(ABORT,'billing_quantity_processing_invalid'); END`, |
| `UPDATE billing_quantity_changes SET history_gap=1,state='history_unproven' |
| - WHERE EXISTS(SELECT 1 FROM billing_quantity_events e WHERE e.account_id=billing_quantity_changes.account_id AND e.processed_at<>0 AND (e.processed_at<e.occurred_at_ns/1000000000 OR e.processing_outcome NOT IN ('charged','included','checkout_expired','checkout_abandoned') OR length(trim(e.processing_reference))=0 OR length(e.processing_reference)>256))`, |
| + WHERE EXISTS(SELECT 1 FROM billing_quantity_events e WHERE e.account_id=billing_quantity_changes.account_id AND e.processed_at<>0 AND (e.processed_at<e.occurred_at_ns/1000000000 OR e.processing_outcome NOT IN ('charged','included','landed_unverified','checkout_expired','checkout_abandoned') OR length(trim(e.processing_reference))=0 OR length(e.processing_reference)>256))`, |
| } { |
| if _, err := s.db.Exec(statement); err != nil { |
| return err |
| @@ -32,7 +36,7 @@ func (s *Store) migrateQuantityHistoryGuards() error { |
| func quantityObligationTx(tx *sql.Tx, accountID int64) (bool, error) { |
| var held bool |
| - err := tx.QueryRow(`SELECT EXISTS(SELECT 1 FROM billing_quantity_events WHERE account_id=? AND ((change='addition' AND processed_at=0) OR processed_at<0 OR (processed_at<>0 AND (processed_at<occurred_at_ns/1000000000 OR processing_outcome NOT IN ('charged','included','checkout_expired','checkout_abandoned') OR length(trim(processing_reference))=0 OR length(processing_reference)>256)))) |
| + err := tx.QueryRow(`SELECT EXISTS(SELECT 1 FROM billing_quantity_events WHERE account_id=? AND ((change='addition' AND processed_at=0) OR processed_at<0 OR (processed_at<>0 AND (processed_at<occurred_at_ns/1000000000 OR processing_outcome NOT IN ('charged','included','landed_unverified','checkout_expired','checkout_abandoned') OR length(trim(processing_reference))=0 OR length(processing_reference)>256)))) |
| OR EXISTS(SELECT 1 FROM billing_quantity_changes WHERE account_id=? AND history_gap<>0)`, accountID, accountID).Scan(&held) |
| return held, err |
| } |
diff --git a/internal/console/billingservice.go b/internal/console/billingservice.go| index 3d36c1e46..46c75ad70 100644 |
| --- a/internal/console/billingservice.go |
| +++ b/internal/console/billingservice.go |
| @@ -481,7 +481,25 @@ func (s *Server) reconcileBillingSubscription(ctx context.Context, id, eventID s |
| } |
| renewalStart = sub.CurrentPeriodStart |
| } |
| - return s.store.applyBillingStateAt(next, state.Revision, eventID, effective, renewalStart) |
| + // A quantity change that is the account's own outstanding operation |
| + // landing takes effect at the operation's instant, the proration date |
| + // the provider was given, whichever path observes it first (the worker's |
| + // reconcile or a webhook) and however long after that instant the |
| + // request finally went through. The customer pays for the seats from |
| + // that instant, so their allowance includes them from it too. |
| + var quantityStart time.Time |
| + quantityKey := "" |
| + if state.Subscription == id && next.Quantity != state.Quantity { |
| + op, ok, err := s.store.quantityOperationFor(acct.ID) |
| + if err != nil { |
| + return err |
| + } |
| + if ok && op.AckedAt == 0 && op.Subscription == id && op.PeriodStart == state.PeriodStart && op.PeriodStart == next.PeriodStart && |
| + op.FromQuantity == state.Quantity && op.Target == next.Quantity { |
| + quantityStart, quantityKey = time.Unix(op.CreatedAt, 0).UTC(), op.Key |
| + } |
| + } |
| + return s.store.applyBillingStateDated(next, state.Revision, eventID, effective, renewalStart, quantityStart, quantityKey) |
| } |
| // provableRenewal reports whether current -> next is exactly one uninterrupted |
diff --git a/internal/console/billingstore.go b/internal/console/billingstore.go| index ca1a6ab9a..c8624afe7 100644 |
| --- a/internal/console/billingstore.go |
| +++ b/internal/console/billingstore.go |
| @@ -7,6 +7,7 @@ import ( |
| "encoding/hex" |
| "errors" |
| "regexp" |
| + "strconv" |
| "time" |
| "unicode/utf8" |
| @@ -352,6 +353,35 @@ func (s *Store) applyBillingState(next billingState, expectedRevision int64, eve |
| // effective as usual, and a live change files a durable billing review: its |
| // month's allowance is then refused as incomplete rather than guessed. |
| func (s *Store) applyBillingStateAt(next billingState, expectedRevision int64, eventID string, effective, renewalStart time.Time) error { |
| + return s.applyBillingStateDated(next, expectedRevision, eventID, effective, renewalStart, time.Time{}, "") |
| +} |
| + |
| +// billingClockTolerance is how far ahead of this process's clock an |
| +// instant recorded by another console process may be and still be taken as |
| +// the same moment seen by two clocks. |
| +const billingClockTolerance = 5 * time.Minute |
| + |
| +// billingQuantityDateReviewReason names the durable hold for a quantity |
| +// operation whose assignment could not be dated at the instant the provider |
| +// prorated it from. |
| +func billingQuantityDateReviewReason(at int64) string { |
| + return "billing_quantity_date_unproven:" + strconv.FormatInt(at, 10) |
| +} |
| + |
| +// applyBillingStateDated is applyBillingStateAt with a second candidate |
| +// instant: quantityStart, the proration date of the account's outstanding |
| +// quantity operation quantityKey, when the caller believes this change is |
| +// that operation landing. The new assignment is dated there only when, in |
| +// this transaction, the operation is still outstanding with this target, |
| +// the change is the quantity and nothing else on an active subscription in |
| +// the same period, and the instant lies inside the period, is not ahead of |
| +// the observation by more than clock skew, does not precede the latest recorded assignment and |
| +// does not fall in or before a month already frozen. Otherwise the |
| +// assignment is dated at effective as usual and a live change files a |
| +// durable review: the provider charged the seats from one instant and the |
| +// allowance would start at another, and nobody is billed on that until a |
| +// person has looked. |
| +func (s *Store) applyBillingStateDated(next billingState, expectedRevision int64, eventID string, effective, renewalStart, quantityStart time.Time, quantityKey string) error { |
| if next.AccountID <= 0 || next.Customer == "" || next.Subscription == "" || next.Quantity <= 0 || next.Plan == "" || next.Terms == "" || next.Band == "" || next.PeriodEnd <= next.PeriodStart { |
| return errors.New("billing_state_invalid") |
| } |
| @@ -412,6 +442,34 @@ func (s *Store) applyBillingStateAt(next billingState, expectedRevision int64, e |
| } |
| } |
| } |
| + quantityUnproven := false |
| + if !quantityStart.IsZero() && !dated { |
| + var outstanding bool |
| + if err := tx.QueryRow(`SELECT EXISTS(SELECT 1 FROM billing_quantity_operations WHERE account_id=? AND idempotency_key=? AND acked_at=0 AND subscription=? AND period_start=? AND from_quantity=? AND target=? AND created_at=?)`, |
| + next.AccountID, quantityKey, next.Subscription, next.PeriodStart, current.Quantity, next.Quantity, quantityStart.Unix()).Scan(&outstanding); err != nil { |
| + return err |
| + } |
| + onlyQuantity := currentLive == next.LiveMode && current.Customer == next.Customer && current.Subscription == next.Subscription && |
| + current.Status == "active" && next.Status == "active" && current.Plan == next.Plan && current.Terms == next.Terms && current.Band == next.Band && |
| + current.PeriodStart == next.PeriodStart && current.PeriodEnd == next.PeriodEnd && current.Quantity != next.Quantity |
| + // The instant may sit a little ahead of this process's clock: the |
| + // operation is dated on the clock of the process that recorded it, |
| + // held inside a period another process may have reconciled, and |
| + // consoles differ by seconds. More than billingClockTolerance ahead |
| + // is not skew, and is not dated. |
| + if outstanding && onlyQuantity && quantityStart.Unix() <= effective.Unix()+int64(billingClockTolerance.Seconds()) && quantityStart.Unix() >= next.PeriodStart && quantityStart.Unix() < next.PeriodEnd { |
| + var frozen bool |
| + if err := tx.QueryRow(`SELECT EXISTS(SELECT 1 FROM billing_months WHERE account_id=? AND month>=?)`, next.AccountID, quantityStart.UTC().Format("2006-01")).Scan(&frozen); err != nil { |
| + return err |
| + } |
| + if latest <= quantityStart.Unix() && !frozen { |
| + assignedAt, dated = quantityStart, true |
| + } |
| + } |
| + // Not datable, and the operation is this change: the hold is filed |
| + // below, once the change is known to be a real one. |
| + quantityUnproven = !dated && outstanding && next.LiveMode |
| + } |
| if next.ReviewReason != "" { |
| if err := fileBillingReviewTx(tx, next.AccountID, next.Subscription, next.ReviewReason, effective); err != nil { |
| return err |
| @@ -455,6 +513,11 @@ func (s *Store) applyBillingStateAt(next billingState, expectedRevision int64, e |
| // same transaction to hold every month for the account until it is |
| // resolved. This runs only once a real revision change is committed to, |
| // so a no-op replay of unchanged state can never file a spurious review. |
| + if quantityUnproven { |
| + if err := fileBillingReviewTx(tx, next.AccountID, next.Subscription, billingQuantityDateReviewReason(quantityStart.Unix()), effective); err != nil { |
| + return err |
| + } |
| + } |
| if !dated && assignedAt.Unix() < latest { |
| assignedAt = time.Unix(latest, 0).UTC() |
| if next.LiveMode { |