amadan
Make the quantity worker's operation a compare-and-set the provider enforces
Review round three (Astra R1, R2, R5; Opus 1, 2, 3) found four ways the team quantity worker could still act on a world that had moved on. The policy switch stays unset; none of this runs in production. A stale proposal can no longer replace a newer operation. Recording an operation is now one transaction under the membership write lock that re-reads the billing row and recomputes the target from the database. It records nothing unless the subscription, its period and period end, the billed quantity, the target and the newest pending addition are all what the proposal was computed from, and no operation is outstanding. A process that read the provider before another process's whole cycle has nothing to send. A renewal between the provider read and the update can no longer charge an old period's seat into the new one. The request carries proration_date, the instant the operation was computed at, which the provider refuses outside the subscription's current period. The operation persists its period end and is never sent at or past it. The send window is measured from the provider read, where it previously measured from an instant taken just before the check. An operation that landed before a renewal is recorded as charged, with its own reference, and no review claims the seat went unbilled. A crash can no longer strand an operation. The mark and the acknowledgement commit in one transaction; account discovery includes unacknowledged operations; an ended subscription's operation is closed with its additions and no provider call. An operation older than the provider's idempotency memory is closed and never re-sent, and its replacement gets a fresh key and date. Reproductions, each confirmed to fail with its fix disabled: a pause before the operation is recorded; the compare-and-set case by case; a renewal between read and request; a landed operation after a renewal; an unacknowledged operation with nothing pending; an ended subscription with an operation outstanding; a stall measured through the production path; a forgotten key. The house no-op test now asserts the general audit trail, and the recovery tests assert the recorded outcome. Not verified against Stripe: that it refuses a proration_date outside the current period. Test mode is not authorized. It must be confirmed there before CONSOLE_BILLING_PRORATION is set.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
5 files changed,
+749
−131
internal/billingstripe/client.go+14 −2internal/console/billingoperator_test.go+12 −0internal/console/billingquantity.go+253 −118internal/console/billingquantity_test.go+465 −8internal/console/billingservice.go+5 −3
diff --git a/internal/billingstripe/client.go b/internal/billingstripe/client.go| index 9f4620e90..170b4e97d 100644 |
| --- a/internal/billingstripe/client.go |
| +++ b/internal/billingstripe/client.go |
| @@ -615,13 +615,19 @@ const ( |
| // proration immediately (always_invoice) rather than deferring it to the |
| // next regular invoice. |
| func (c *Client) UpdateQuantity(ctx context.Context, itemID string, quantity int64, idempotencyKey string) (*SubscriptionItem, error) { |
| - return c.UpdateQuantityWith(ctx, itemID, quantity, ProrationAlwaysInvoice, idempotencyKey) |
| + return c.UpdateQuantityWith(ctx, itemID, quantity, ProrationAlwaysInvoice, time.Time{}, idempotencyKey) |
| } |
| // UpdateQuantityWith changes one subscription item's quantity under the |
| // named proration behaviour. When the proration is collected is a product |
| // decision the caller carries; nothing here defaults it. |
| -func (c *Client) UpdateQuantityWith(ctx context.Context, itemID string, quantity int64, proration, idempotencyKey string) (*SubscriptionItem, error) { |
| +// |
| +// A non-zero prorationDate is sent as proration_date: the proration is |
| +// computed as of that instant, and the provider refuses the request when |
| +// the instant lies outside the subscription's current period. A caller that |
| +// decided the update in one period therefore cannot have it charged into |
| +// the next. |
| +func (c *Client) UpdateQuantityWith(ctx context.Context, itemID string, quantity int64, proration string, prorationDate time.Time, idempotencyKey string) (*SubscriptionItem, error) { |
| if !validID(itemID) { |
| return nil, &Error{Code: CodeInvalidID} |
| } |
| @@ -636,6 +642,12 @@ func (c *Client) UpdateQuantityWith(ctx context.Context, itemID string, quantity |
| form := url.Values{} |
| form.Set("quantity", strconv.FormatInt(quantity, 10)) |
| form.Set("proration_behavior", proration) |
| + if !prorationDate.IsZero() { |
| + if prorationDate.Unix() <= 0 { |
| + return nil, &Error{Code: CodeInvalidParams} |
| + } |
| + form.Set("proration_date", strconv.FormatInt(prorationDate.Unix(), 10)) |
| + } |
| var item SubscriptionItem |
| if err := c.do(ctx, http.MethodPost, "/v1/subscription_items/"+itemID, form, idempotencyKey, &item); err != nil { |
| return nil, err |
diff --git a/internal/console/billingoperator_test.go b/internal/console/billingoperator_test.go| index e28fd5ce5..7aea51e13 100644 |
| --- a/internal/console/billingoperator_test.go |
| +++ b/internal/console/billingoperator_test.go |
| @@ -272,4 +272,16 @@ func TestBillingOperatorHouseNoOpLeavesNoAuditTrail(t *testing.T) { |
| if count() != 0 { |
| t.Fatalf("a no-op wrote %d house change rows", count()) |
| } |
| + // Nor an entry in the account's general audit trail. |
| + var general int |
| + if err := fx.st.db.QueryRow(`SELECT COUNT(*) FROM audit WHERE account_id=? AND action LIKE 'billing-house-%'`, OpsAccountID).Scan(&general); err != nil || general != 0 { |
| + t.Fatalf("a no-op wrote %d general audit entries (%v)", general, err) |
| + } |
| + // A real change writes exactly one of each. |
| + if status, _ := postJSON(t, fx.ts.URL+"/api/billing/house", map[string]any{"account": fx.sqid, "house": true, "note": "a real change"}, token); status != http.StatusOK { |
| + t.Fatalf("set = %d", status) |
| + } |
| + if err := fx.st.db.QueryRow(`SELECT COUNT(*) FROM audit WHERE account_id=? AND action='billing-house-set'`, OpsAccountID).Scan(&general); err != nil || general != 1 || count() != 1 { |
| + t.Fatalf("a real change wrote %d general entries and %d house rows (%v), want one each", general, count(), err) |
| + } |
| } |
diff --git a/internal/console/billingquantity.go b/internal/console/billingquantity.go| index c0b0f8d9e..e494b6451 100644 |
| --- a/internal/console/billingquantity.go |
| +++ b/internal/console/billingquantity.go |
| @@ -44,17 +44,27 @@ import ( |
| // anything newer may begin. |
| // 2. Read the subscription from the provider and check it is the one the |
| // console bills, active, in the period the console recorded. |
| -// 3. Record the operation (target, generation, behaviour, idempotency |
| -// key), then update the item to the target under that key. |
| -// 4. Reconcile the subscription: the authoritative read that moves |
| +// 3. Record the operation. This is a compare-and-set, in one transaction |
| +// under the membership write lock, on everything the target was |
| +// computed from: the subscription, its period, the billed quantity, the |
| +// newest pending addition and the target itself, recomputed from the |
| +// database. A process that read an older state finds it changed and |
| +// records nothing, so it has nothing to send. |
| +// 4. Update the item to the target under the operation's idempotency key, |
| +// with the proration dated at the operation's own instant. The provider |
| +// refuses a proration date outside the subscription's current period, |
| +// so an update that arrives after a renewal is refused there, not |
| +// charged into the new period. |
| +// 5. Reconcile the subscription: the authoritative read that moves |
| // billing_accounts.quantity and appends the assignment revision. |
| -// 5. Mark the additions processed, in a transaction that re-checks the |
| -// subscription, its period and that the billed quantity covers them. |
| -// 6. Acknowledge the operation. |
| +// 6. Mark the additions processed and acknowledge the operation, in one |
| +// transaction that re-checks the subscription, its period and that the |
| +// billed quantity covers them. |
| // |
| -// A crash anywhere after 3 leaves the operation outstanding and step 1 |
| -// finishes it with the same key. A provider quantity that is neither what |
| -// the console bills nor the target is filed for review and nothing is sent. |
| +// A crash anywhere after 3 leaves the operation outstanding, and step 1 |
| +// finishes it with the same key and the same parameters. A provider |
| +// quantity that is neither what the console bills nor the target is filed |
| +// for review and nothing is sent. |
| // |
| // Not done here: lowering the quantity at renewal for members who left. |
| // Until that exists a renewal bills the old quantity, which must be fixed |
| @@ -86,10 +96,13 @@ func (cfg BillingConfig) prorationBehavior() (behavior, outcome string, ok bool) |
| return "", "", false |
| } |
| -// pendingQuantityAccounts lists accounts with an unprocessed addition bound |
| -// to a subscription, in account order. |
| +// pendingQuantityAccounts lists, in account order, every account the pass |
| +// has work for: an unprocessed addition bound to a subscription, or an |
| +// operation left unacknowledged (a crash between the mark and the |
| +// acknowledgement must not strand it). |
| func (s *Store) pendingQuantityAccounts() ([]int64, error) { |
| - rows, err := s.db.Query(`SELECT DISTINCT account_id FROM billing_quantity_events WHERE change=? AND processed_at=0 AND subscription<>'' ORDER BY account_id`, quantityEventAddition) |
| + rows, err := s.db.Query(`SELECT account_id FROM billing_quantity_events WHERE change=? AND processed_at=0 AND subscription<>'' |
| + UNION SELECT account_id FROM billing_quantity_operations WHERE acked_at=0 ORDER BY account_id`, quantityEventAddition) |
| if err != nil { |
| return nil, err |
| } |
| @@ -142,13 +155,29 @@ func billingSubscriptionOver(state billingState) bool { |
| return state.Subscription == "" || state.Status == "canceled" || state.Status == "incomplete_expired" |
| } |
| +// quantityTarget is the quantity the pending additions call for: the peak |
| +// the current period's additions asked for, and never less than the seats |
| +// the account occupies now. A member added in an earlier period who is |
| +// still here occupies a seat in this one, prorated from now, not from a |
| +// period that is over. Both the worker and the transaction that records an |
| +// operation compute it, from the same rows, so they cannot disagree. |
| +func quantityTarget(pending quantityPending, members int64, terms, plan string) (int64, error) { |
| + def, err := plans.Lookup(terms, plan) |
| + if err != nil { |
| + return 0, err |
| + } |
| + return max(pending.Peak, max(members, def.MinQuantity)), nil |
| +} |
| + |
| // closeEndedQuantityAdditions marks, without any provider call, every |
| // unprocessed addition that can never be charged: one bound to a |
| // subscription that is not the account's current one, and, when the current |
| // subscription is over, every one. Left pending they would hold the |
| -// account's freeze for good, including under a later subscription. It |
| -// re-reads the billing row inside the transaction, under the membership |
| -// write lock, and returns whether the current subscription is over. |
| +// account's freeze for good, including under a later subscription. When the |
| +// subscription is over, an operation still outstanding is acknowledged too: |
| +// nothing may be sent for it. It re-reads the billing row inside the |
| +// transaction, under the membership write lock, and returns whether the |
| +// current subscription is over. |
| func (s *Store) closeEndedQuantityAdditions(accountID int64, now time.Time) (over bool, err error) { |
| tx, err := s.db.Begin() |
| if err != nil { |
| @@ -169,6 +198,11 @@ func (s *Store) closeEndedQuantityAdditions(accountID int64, now time.Time) (ove |
| if err != nil { |
| return false, err |
| } |
| + if over { |
| + if _, err := tx.Exec(`UPDATE billing_quantity_operations SET acked_at=? WHERE account_id=? AND acked_at=0`, max(now.Unix(), 1), accountID); err != nil { |
| + return false, err |
| + } |
| + } |
| if n, err := res.RowsAffected(); err != nil { |
| return false, err |
| } else if n > 0 { |
| @@ -180,14 +214,15 @@ func (s *Store) closeEndedQuantityAdditions(accountID int64, now time.Time) (ove |
| } |
| // markQuantityAdditionsProcessed marks the unprocessed additions of |
| -// subscription up to maxGeneration. Current-period additions are marked |
| -// with outcome and reference once the billed quantity covers their peak. |
| -// Additions from a period that has since renewed are marked as not |
| -// charged, and a review is filed in the same transaction so a person sees |
| -// that seats went unbilled in that period. It refuses with |
| +// subscription up to maxGeneration, and acknowledges the operation ackKey |
| +// (when not empty) in the same transaction, so a crash cannot leave one |
| +// done and the other not. Current-period additions are marked with outcome |
| +// and reference once the billed quantity covers their peak. Additions from |
| +// a period that has since renewed are marked as not charged, and a review |
| +// is filed so a person sees that seats went unbilled. It refuses with |
| // errBillingPending when the subscription or its period moved, or when the |
| // reconcile that raises the billed quantity has not landed. |
| -func (s *Store) markQuantityAdditionsProcessed(accountID int64, subscription string, periodStart, maxGeneration int64, outcome, reference string, now time.Time) error { |
| +func (s *Store) markQuantityAdditionsProcessed(accountID int64, subscription string, periodStart, maxGeneration int64, outcome, reference, ackKey string, now time.Time) error { |
| tx, err := s.db.Begin() |
| if err != nil { |
| return err |
| @@ -198,13 +233,24 @@ func (s *Store) markQuantityAdditionsProcessed(accountID int64, subscription str |
| if _, err := lockMembershipWritesTx(tx); err != nil { |
| return err |
| } |
| - pending, err := pendingQuantityAdditions(tx, accountID, subscription, periodStart, maxGeneration) |
| - if err != nil { |
| + ack := func() error { |
| + if ackKey == "" { |
| + return nil |
| + } |
| + _, err := tx.Exec(`UPDATE billing_quantity_operations SET acked_at=? WHERE account_id=? AND idempotency_key=? AND acked_at=0`, max(now.Unix(), 1), accountID, ackKey) |
| return err |
| } |
| - if pending.Current == 0 && pending.Stale == 0 { |
| + if maxGeneration <= 0 { |
| + // Nothing was pending when the step was decided. |
| + if err := ack(); err != nil { |
| + return err |
| + } |
| return tx.Commit() |
| } |
| + pending, err := pendingQuantityAdditions(tx, accountID, subscription, periodStart, maxGeneration) |
| + if err != nil { |
| + return err |
| + } |
| var billed, currentStart int64 |
| var current string |
| if err := tx.QueryRow(`SELECT quantity,subscription,period_start FROM billing_accounts WHERE account_id=?`, accountID).Scan(&billed, ¤t, ¤tStart); err != nil { |
| @@ -230,80 +276,124 @@ func (s *Store) markQuantityAdditionsProcessed(accountID int64, subscription str |
| return err |
| } |
| } |
| - if err := refreshQuantityChangeTx(tx, accountID, now); err != nil { |
| + if err := ack(); err != nil { |
| return err |
| } |
| + if pending.Current > 0 || pending.Stale > 0 { |
| + if err := refreshQuantityChangeTx(tx, accountID, now); err != nil { |
| + return err |
| + } |
| + } |
| return tx.Commit() |
| } |
| // quantityOperation is the account's one provider update in flight, |
| // persisted BEFORE the request is sent. Consoles are several processes and |
| -// the per-account lock is process-local, so without it two of them could |
| -// each read the provider, and the slower could then send an older, lower |
| -// target after the faster had raised, reconciled and marked a newer one. |
| -// With it there is at most one outstanding operation per account across |
| -// every process: a newer target cannot be started until the outstanding |
| -// one is acknowledged, whoever finishes it sends the SAME idempotency key |
| -// (which the provider replays instead of executing twice), and a process |
| -// that was paused past quantityOperationSendWindow does not send at all. |
| +// the per-account lock is process-local. One outstanding operation per |
| +// account, recorded by compare-and-set, is what orders them. |
| type quantityOperation struct { |
| - AccountID int64 |
| - Subscription, Item string |
| - PeriodStart int64 |
| - FromQuantity, Target int64 |
| - Generation int64 |
| - Behavior, Key string |
| - CreatedAt, AckedAt int64 |
| + AccountID int64 |
| + Subscription, Item string |
| + PeriodStart, PeriodEnd int64 |
| + FromQuantity, Target int64 |
| + Generation int64 |
| + Behavior, Key string |
| + CreatedAt, AckedAt int64 |
| } |
| -const ( |
| - // A process sends an operation's request only within this long of |
| - // writing its row: longer means it was paused, and someone else may |
| - // have finished the operation and begun another. |
| - quantityOperationSendWindow = 2 * time.Minute |
| - // Past this age an unacknowledged operation is never re-sent: the |
| - // provider has forgotten its idempotency key, so a send would be a new |
| - // request, not a replay. |
| - quantityOperationReplayLimit = 22 * time.Hour |
| -) |
| +// A process sends an operation's request only within this long of reading |
| +// the provider for it: longer means it was paused, and the world it read |
| +// may be gone. A variable so a test can shorten it. |
| +var quantityOperationSendWindow = 2 * time.Minute |
| + |
| +// Past this age an unacknowledged operation is never re-sent: the provider |
| +// has forgotten its idempotency key, so a send would be a new request, not |
| +// a replay. |
| +const quantityOperationReplayLimit = 22 * time.Hour |
| func (s *Store) migrateBillingQuantityOperations() error { |
| - _, err := s.db.Exec(`CREATE TABLE IF NOT EXISTS billing_quantity_operations ( |
| - account_id INTEGER PRIMARY KEY, subscription TEXT NOT NULL, period_start INTEGER NOT NULL, |
| + if _, err := s.db.Exec(`CREATE TABLE IF NOT EXISTS billing_quantity_operations ( |
| + account_id INTEGER PRIMARY KEY, subscription TEXT NOT NULL, period_start INTEGER NOT NULL, period_end INTEGER NOT NULL DEFAULT 0, |
| item TEXT NOT NULL, from_quantity INTEGER NOT NULL, target INTEGER NOT NULL, |
| generation INTEGER NOT NULL, behavior TEXT NOT NULL, idempotency_key TEXT NOT NULL UNIQUE, |
| created_at INTEGER NOT NULL, acked_at INTEGER NOT NULL DEFAULT 0 |
| - )`) |
| - return err |
| + )`); err != nil { |
| + return err |
| + } |
| + return s.addColumnIfMissing("billing_quantity_operations", "period_end", `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,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 |
| FROM billing_quantity_operations WHERE account_id=?`, accountID). |
| - Scan(&op.AccountID, &op.Subscription, &op.PeriodStart, &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) |
| if errors.Is(err, sql.ErrNoRows) { |
| return op, false, nil |
| } |
| return op, err == nil, err |
| } |
| -// beginQuantityOperation records op as the account's operation in flight. |
| -// started is false when another operation is still unacknowledged: only |
| -// one may be outstanding, across every process. |
| +// beginQuantityOperation records op as the account's operation in flight, |
| +// if and only if everything op was computed from is still true. In one |
| +// 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 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). |
| +// |
| +// started is false when any of these fails. The caller then has nothing to |
| +// send, and the next pass recomputes from the present. This is what stops a |
| +// process that read an older state from recording, and then sending, a |
| +// lower target after another process has finished a newer one. |
| func (s *Store) beginQuantityOperation(op quantityOperation) (started bool, err error) { |
| - res, err := s.db.Exec(`INSERT INTO billing_quantity_operations(account_id,subscription,period_start,item,from_quantity,target,generation,behavior,idempotency_key,created_at,acked_at) |
| - VALUES(?,?,?,?,?,?,?,?,?,?,0) |
| - ON CONFLICT(account_id) DO UPDATE SET subscription=excluded.subscription,period_start=excluded.period_start,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 |
| - WHERE billing_quantity_operations.acked_at<>0`, |
| - op.AccountID, op.Subscription, op.PeriodStart, op.Item, op.FromQuantity, op.Target, op.Generation, op.Behavior, op.Key, op.CreatedAt) |
| + tx, err := s.db.Begin() |
| + if err != nil { |
| + return false, err |
| + } |
| + defer tx.Rollback() |
| + if _, err := lockMembershipWritesTx(tx); err != nil { |
| + return false, err |
| + } |
| + var outstanding bool |
| + if err := tx.QueryRow(`SELECT EXISTS(SELECT 1 FROM billing_quantity_operations WHERE account_id=? AND acked_at=0)`, op.AccountID).Scan(&outstanding); err != nil { |
| + return false, err |
| + } |
| + if outstanding { |
| + return false, nil |
| + } |
| + state, err := scanBilling(tx.QueryRow(`SELECT `+billingColumns+` FROM billing_accounts WHERE account_id=?`, op.AccountID), op.AccountID) |
| + if err != nil { |
| + return false, err |
| + } |
| + if 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 |
| + } |
| + pending, err := pendingQuantityAdditions(tx, op.AccountID, op.Subscription, op.PeriodStart, 0) |
| + if err != nil { |
| + return false, err |
| + } |
| + members, err := memberCountTx(tx, op.AccountID) |
| if err != nil { |
| return false, err |
| } |
| - n, err := res.RowsAffected() |
| - return n == 1, err |
| + target, err := quantityTarget(pending, members, state.Terms, state.Plan) |
| + 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) |
| + 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 { |
| + return false, err |
| + } |
| + return true, tx.Commit() |
| } |
| func (s *Store) ackQuantityOperation(accountID int64, key string, now time.Time) error { |
| @@ -311,6 +401,38 @@ 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 |
| +// 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() |
| + if err != nil { |
| + return err |
| + } |
| + defer tx.Rollback() |
| + if _, err := lockMembershipWritesTx(tx); err != nil { |
| + return err |
| + } |
| + res, err := tx.Exec(`UPDATE billing_quantity_events SET processed_at=max(?,occurred_at_ns/1000000000+1), processing_outcome=?, processing_reference=? |
| + WHERE account_id=? AND change=? AND processed_at=0 AND subscription=? AND period_start=? AND generation<=?`, |
| + now.Unix(), outcome, reference, op.AccountID, quantityEventAddition, op.Subscription, op.PeriodStart, op.Generation) |
| + if err != nil { |
| + return err |
| + } |
| + if _, err := tx.Exec(`UPDATE billing_quantity_operations SET acked_at=? WHERE account_id=? AND idempotency_key=? AND acked_at=0`, max(now.Unix(), 1), op.AccountID, op.Key); err != nil { |
| + return err |
| + } |
| + if n, err := res.RowsAffected(); err != nil { |
| + return err |
| + } else if n > 0 { |
| + if err := refreshQuantityChangeTx(tx, op.AccountID, now); err != nil { |
| + return err |
| + } |
| + } |
| + return tx.Commit() |
| +} |
| + |
| // billingQuantityPass processes every account with pending additions or an |
| // operation in flight. With no proration policy configured it returns at |
| // once. |
| @@ -346,6 +468,10 @@ func quantityOutcomeFor(behavior string) string { |
| return quantityOutcomeIncluded |
| } |
| +func quantityReference(item string, target int64, behavior string) string { |
| + return item + ":quantity=" + strconv.FormatInt(target, 10) + ":" + behavior |
| +} |
| + |
| func (s *Server) processQuantityAdditions(ctx context.Context, accountID int64, behavior string, now time.Time) error { |
| acct, err := s.store.AccountByID(accountID) |
| if err != nil { |
| @@ -355,7 +481,7 @@ func (s *Server) processQuantityAdditions(ctx context.Context, accountID int64, |
| return nil |
| } |
| // Additions nothing can charge any more are closed first, with no |
| - // provider call. |
| + // provider call; so is an operation of a subscription that is over. |
| over, err := s.store.closeEndedQuantityAdditions(accountID, now) |
| if err != nil || over { |
| return err |
| @@ -373,13 +499,7 @@ func (s *Server) processQuantityAdditions(ctx context.Context, accountID int64, |
| } |
| outcome, reference = quantityOutcomeFor(step.behavior), step.reference |
| } |
| - if err := s.store.markQuantityAdditionsProcessed(accountID, step.state.Subscription, step.state.PeriodStart, step.generation, outcome, reference, now); err != nil { |
| - return err |
| - } |
| - if step.key != "" { |
| - return s.store.ackQuantityOperation(accountID, step.key, now) |
| - } |
| - return nil |
| + return s.store.markQuantityAdditionsProcessed(accountID, step.state.Subscription, step.state.PeriodStart, step.generation, outcome, reference, step.key, now) |
| } |
| // quantityStep is what raiseQuantity decided under the account's lock. |
| @@ -388,11 +508,11 @@ type quantityStep struct { |
| generation int64 // additions up to this generation are covered |
| behavior string // the behaviour the provider request carried |
| reference string // non-empty: the provider carries the target, reconcile next |
| - key string // non-empty: the operation to acknowledge after the mark |
| + key string // non-empty: the operation to acknowledge with the mark |
| done bool // nothing to do, or not in a state to act |
| } |
| -// raiseQuantity is steps 1 and 2 under the account's billing lock. |
| +// raiseQuantity is steps 1 to 4 under the account's billing lock. |
| func (s *Server) raiseQuantity(ctx context.Context, acct *Account, behavior string, now time.Time) (quantityStep, error) { |
| b := s.billing |
| defer b.locks.lock(acct.Sqid, "billing")() |
| @@ -415,17 +535,13 @@ func (s *Server) raiseQuantity(ctx context.Context, acct *Account, behavior stri |
| if pending.Current == 0 && pending.Stale == 0 { |
| return quantityStep{done: true}, nil |
| } |
| - // The target is the peak the current period's additions asked for, and |
| - // never less than the seats the account occupies right now: a member |
| - // added in an earlier period who is still here occupies a seat in this |
| - // one, prorated from now, not from a period that is over. |
| - target := pending.Peak |
| - if def, lerr := plans.Lookup(state.Terms, state.Plan); lerr == nil { |
| - members, merr := s.store.Members(acct.ID) |
| - if merr != nil { |
| - return quantityStep{}, merr |
| - } |
| - target = max(target, max(int64(len(members)), def.MinQuantity)) |
| + members, err := s.store.Members(acct.ID) |
| + if err != nil { |
| + return quantityStep{}, err |
| + } |
| + target, err := quantityTarget(pending, int64(len(members)), state.Terms, state.Plan) |
| + if err != nil { |
| + return quantityStep{}, err |
| } |
| step := quantityStep{state: state, generation: pending.Newest, behavior: behavior} |
| if target <= state.Quantity { |
| @@ -437,6 +553,7 @@ func (s *Server) raiseQuantity(ctx context.Context, acct *Account, behavior stri |
| if state.Status != "active" { |
| return quantityStep{done: true}, nil |
| } |
| + read := time.Now() |
| item, ok, err := s.quantityProviderItem(ctx, acct, state) |
| if err != nil || !ok { |
| return quantityStep{done: !ok}, err |
| @@ -444,28 +561,35 @@ func (s *Server) raiseQuantity(ctx context.Context, acct *Account, behavior stri |
| switch item.Quantity { |
| case target: |
| // Already there, by a path that left no operation (an operator's |
| - // hand, or a request from before operations were recorded). |
| - step.reference = item.ID + ":quantity=" + strconv.FormatInt(target, 10) + ":observed" |
| + // hand). Nothing is sent; the reconcile records it. |
| + step.reference = quantityReference(item.ID, target, "observed") |
| return step, nil |
| case state.Quantity: |
| default: |
| return s.quantityMismatch(acct, state, now) |
| } |
| - // The key names everything the request depends on, the behaviour |
| - // included, so a changed policy is a new request and never a replay |
| - // with different parameters. |
| - op := quantityOperation{AccountID: acct.ID, Subscription: state.Subscription, PeriodStart: state.PeriodStart, Item: item.ID, |
| - FromQuantity: state.Quantity, Target: target, Generation: pending.Newest, Behavior: behavior, CreatedAt: now.Unix(), |
| - Key: fmt.Sprintf("carloku-qty-%s-%s-%d-%d-%d-%s", acct.Sqid, state.Subscription, state.PeriodStart, pending.Newest, target, behavior)} |
| + if hook := b.quantityBeforeBegin; hook != nil { |
| + hook() |
| + } |
| + // The key names everything the request carries, the behaviour and the |
| + // 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() |
| + 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, |
| + 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 { |
| return quantityStep{}, err |
| } |
| if !started { |
| - // Another process began one between our read and our write. |
| + // What this process read is no longer true: another process acted, |
| + // a member came or went, or the subscription moved. Nothing was |
| + // recorded and nothing is sent; the next pass starts from now. |
| return quantityStep{done: true}, nil |
| } |
| - return s.sendQuantityOperation(ctx, state, op, time.Now()) |
| + return s.sendQuantityOperation(ctx, state, op, read) |
| } |
| // quantityProviderItem reads the subscription from the provider and checks |
| @@ -489,18 +613,22 @@ func (s *Server) quantityProviderItem(ctx context.Context, acct *Account, state |
| return sub.Items[0], true, nil |
| } |
| -// sendQuantityOperation sends op's request. began is when this process |
| -// wrote, or picked up, the operation: a process paused past the send |
| -// window does not send. |
| -func (s *Server) sendQuantityOperation(ctx context.Context, state billingState, op quantityOperation, began time.Time) (quantityStep, error) { |
| +// sendQuantityOperation sends op's request. read is when this process read |
| +// the provider for it. Three things keep the request inside the period it |
| +// was computed for: a process paused past the send window does not send; |
| +// nothing is sent at or after the recorded period end; and the request |
| +// dates its proration at the operation's own instant, which the provider |
| +// refuses when that instant is outside the subscription's current period. |
| +// The last is the one that holds across a renewal this process cannot see. |
| +func (s *Server) sendQuantityOperation(ctx context.Context, state billingState, op quantityOperation, read time.Time) (quantityStep, error) { |
| b := s.billing |
| if hook := b.quantityBeforeSend; hook != nil { |
| hook() |
| } |
| - if time.Since(began) >= quantityOperationSendWindow { |
| + if time.Since(read) >= quantityOperationSendWindow || (op.PeriodEnd > 0 && time.Now().Unix() >= op.PeriodEnd) { |
| return quantityStep{done: true}, nil |
| } |
| - updated, err := b.client.UpdateQuantityWith(ctx, op.Item, op.Target, op.Behavior, op.Key) |
| + updated, err := b.client.UpdateQuantityWith(ctx, op.Item, op.Target, op.Behavior, time.Unix(op.CreatedAt, 0), op.Key) |
| if err != nil { |
| return quantityStep{}, err |
| } |
| @@ -508,7 +636,7 @@ func (s *Server) sendQuantityOperation(ctx context.Context, state billingState, |
| return quantityStep{}, errBillingIdentity |
| } |
| return quantityStep{state: state, generation: op.Generation, behavior: op.Behavior, key: op.Key, |
| - reference: op.Item + ":quantity=" + strconv.FormatInt(op.Target, 10) + ":" + op.Behavior}, nil |
| + reference: quantityReference(op.Item, op.Target, op.Behavior)}, nil |
| } |
| // resolveQuantityOperation finishes an operation some process left |
| @@ -524,34 +652,41 @@ func (s *Server) resolveQuantityOperation(ctx context.Context, acct *Account, st |
| if op.Subscription != state.Subscription { |
| return abandon() |
| } |
| + read := time.Now() |
| item, ok, err := s.quantityProviderItem(ctx, acct, state) |
| if err != nil { |
| return quantityStep{}, err |
| } |
| - landed := quantityStep{state: state, generation: op.Generation, behavior: op.Behavior, key: op.Key, |
| - reference: op.Item + ":quantity=" + strconv.FormatInt(op.Target, 10) + ":" + op.Behavior} |
| switch { |
| case !ok: |
| - // Not active, or the period moved: nothing may be sent. Wait for |
| - // the reconcile; the operation is looked at again next pass. |
| + // Not active, or the console has not caught up with the provider's |
| + // period: nothing may be sent. The operation is looked at again |
| + // next pass. |
| return quantityStep{done: true}, nil |
| case item.ID != op.Item: |
| return s.quantityMismatch(acct, state, now) |
| - case item.Quantity == op.Target && op.PeriodStart == state.PeriodStart: |
| - return landed, nil |
| case op.PeriodStart != state.PeriodStart: |
| // The subscription renewed while the operation was outstanding. |
| - // Nothing is sent for a period that is over. The operation is |
| - // closed, and the old period's additions are then closed as not |
| - // charged with a review (markQuantityAdditionsProcessed), which is |
| - // what a person should look at if this ever happens. |
| + // 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. |
| + 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 abandon() |
| + case item.Quantity == op.Target: |
| + return quantityStep{state: state, generation: op.Generation, behavior: op.Behavior, key: op.Key, |
| + reference: quantityReference(op.Item, op.Target, op.Behavior)}, nil |
| case item.Quantity != op.FromQuantity: |
| return s.quantityMismatch(acct, state, now) |
| - case now.Unix()-op.CreatedAt >= int64(quantityOperationReplayLimit.Seconds()), now.Unix() < op.CreatedAt: |
| + case read.Unix()-op.CreatedAt >= int64(quantityOperationReplayLimit.Seconds()), read.Unix() < op.CreatedAt: |
| + // Too old for the provider to still remember its key (or dated in |
| + // the future: a clock went backwards). The provider read above |
| + // shows it did not land, so it is closed and never re-sent. |
| return abandon() |
| } |
| - return s.sendQuantityOperation(ctx, state, op, time.Now()) |
| + return s.sendQuantityOperation(ctx, state, op, read) |
| } |
| // quantityMismatch files the review for a provider quantity that is |
diff --git a/internal/console/billingquantity_test.go b/internal/console/billingquantity_test.go| index ad0e5011b..114a2ae67 100644 |
| --- a/internal/console/billingquantity_test.go |
| +++ b/internal/console/billingquantity_test.go |
| @@ -24,15 +24,17 @@ type quantityStripe struct { |
| quantity int64 |
| status string |
| gets int |
| - posts []map[string]string // quantity, proration_behavior, key |
| + posts []map[string]string // quantity, proration_behavior, proration_date, key |
| failPost bool |
| start time.Time |
| failAfter bool // accept the POST, then answer 500: a lost response |
| // byKey is the provider's idempotency memory: a key it has seen is |
| // answered with what it answered then, and changes nothing. |
| byKey map[string]int64 |
| + params map[string]string // key -> the parameters it was first sent with |
| replays int |
| lowered bool // the quantity ever went down |
| + refused int // requests refused for a proration date outside the period |
| } |
| func (q *quantityStripe) serve(w http.ResponseWriter, r *http.Request) { |
| @@ -57,16 +59,32 @@ func (q *quantityStripe) serve(w http.ResponseWriter, r *http.Request) { |
| } |
| n, _ := strconv.ParseInt(r.PostForm.Get("quantity"), 10, 64) |
| key := r.Header.Get("Idempotency-Key") |
| + // The provider refuses a proration date outside the subscription's |
| + // current period. This is what makes the period a precondition of |
| + // the request and not a check the console makes beforehand. |
| + date, _ := strconv.ParseInt(r.PostForm.Get("proration_date"), 10, 64) |
| + 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"}}`)) |
| + return |
| + } |
| if prev, seen := q.byKey[key]; seen { |
| + // A key replayed with different parameters is refused. |
| + if q.params[key] != r.PostForm.Encode() { |
| + w.WriteHeader(http.StatusBadRequest) |
| + _, _ = w.Write([]byte(`{"error":{"type":"idempotency_error","message":"Keys for idempotent requests can only be used with the same parameters"}}`)) |
| + return |
| + } |
| q.replays++ |
| _ = json.NewEncoder(w).Encode(map[string]any{"id": "si_team", "price": map[string]string{"id": "price_community_A"}, "quantity": prev}) |
| return |
| } |
| if q.byKey == nil { |
| - q.byKey = map[string]int64{} |
| + q.byKey, q.params = map[string]int64{}, map[string]string{} |
| } |
| - q.byKey[key] = n |
| - q.posts = append(q.posts, map[string]string{"quantity": r.PostForm.Get("quantity"), "proration_behavior": r.PostForm.Get("proration_behavior"), "key": key}) |
| + q.byKey[key], q.params[key] = n, r.PostForm.Encode() |
| + q.posts = append(q.posts, map[string]string{"quantity": r.PostForm.Get("quantity"), "proration_behavior": r.PostForm.Get("proration_behavior"), "proration_date": r.PostForm.Get("proration_date"), "key": key}) |
| if n < q.quantity { |
| q.lowered = true |
| } |
| @@ -285,6 +303,28 @@ func TestQuantityWorkerRecoversALostResponse(t *testing.T) { |
| if state, _ := fx.st.billingAccount(OpsAccountID); state.Quantity != 11 || quantityHeld(t, fx) { |
| t.Fatalf("not recovered: quantity %d held %v", state.Quantity, quantityHeld(t, fx)) |
| } |
| + // The recovered addition is recorded as what it was: charged, by this |
| + // operation. Not as a seat that happened to be billed already. |
| + assertLastAddition(t, fx, quantityOutcomeCharged, "si_team:quantity=11:always_invoice") |
| +} |
| + |
| +// assertLastAddition checks the newest addition was processed with outcome |
| +// and reference. |
| +func assertLastAddition(t *testing.T, fx *consoleFixture, outcome, reference string) { |
| + t.Helper() |
| + events, err := fx.st.billingQuantityEvents(OpsAccountID) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + for i := len(events) - 1; i >= 0; i-- { |
| + if e := events[i]; e.Change == quantityEventAddition { |
| + if e.ProcessedAt == 0 || e.ProcessingOutcome != outcome || e.ProcessingReference != reference { |
| + t.Fatalf("addition processed=%d outcome=%q reference=%q, want %q %q", e.ProcessedAt, e.ProcessingOutcome, e.ProcessingReference, outcome, reference) |
| + } |
| + return |
| + } |
| + } |
| + t.Fatal("no addition recorded") |
| } |
| // A provider quantity the console neither billed nor wants is a person's |
| @@ -327,7 +367,8 @@ func TestQuantityWorkerLeavesAPastDueSubscriptionAlone(t *testing.T) { |
| // renew moves the account's subscription into its next yearly period, on |
| // the provider and through the console's own state path, at the quantity it |
| -// currently bills. |
| +// currently bills. The new period starts at the renewal instant, so the |
| +// wall clock the worker dates its requests with lies inside it. |
| func renewQuantitySubscription(t *testing.T, fx *consoleFixture, q *quantityStripe, at time.Time) { |
| t.Helper() |
| state, err := fx.st.billingAccount(OpsAccountID) |
| @@ -335,7 +376,7 @@ func renewQuantitySubscription(t *testing.T, fx *consoleFixture, q *quantityStri |
| t.Fatal(err) |
| } |
| next := state |
| - next.PeriodStart, next.PeriodEnd, next.LiveMode = state.PeriodEnd, time.Unix(state.PeriodEnd, 0).AddDate(1, 0, 0).Unix(), true |
| + next.PeriodStart, next.PeriodEnd, next.LiveMode = at.Unix(), at.AddDate(1, 0, 0).Unix(), true |
| if err := fx.st.applyBillingState(next, state.Revision, "evt_quantity_renew", at); err != nil { |
| t.Fatal(err) |
| } |
| @@ -490,18 +531,433 @@ func TestQuantityWorkerCannotLowerAQuantityAcrossProcesses(t *testing.T) { |
| } |
| } |
| -// A process paused past the send window does not send at all. |
| +// A process paused past the send window, measured from its provider read, |
| +// does not send at all; nor does one whose period has ended by its own |
| +// clock. |
| func TestQuantityWorkerDoesNotSendAfterALongPause(t *testing.T) { |
| fx, q, now := newQuantityFixture(t, billingProrationImmediate) |
| state, err := fx.st.billingAccount(OpsAccountID) |
| if err != nil { |
| t.Fatal(err) |
| } |
| - op := quantityOperation{AccountID: OpsAccountID, Subscription: "sub_team", PeriodStart: state.PeriodStart, Item: "si_team", FromQuantity: 10, Target: 11, Generation: 1, Behavior: billingstripe.ProrationAlwaysInvoice, Key: "carloku-qty-paused", CreatedAt: now.Unix()} |
| + op := quantityOperation{AccountID: OpsAccountID, Subscription: "sub_team", PeriodStart: state.PeriodStart, PeriodEnd: state.PeriodEnd, Item: "si_team", FromQuantity: 10, Target: 11, Generation: 1, Behavior: billingstripe.ProrationAlwaysInvoice, Key: "carloku-qty-paused", CreatedAt: now.Unix()} |
| step, err := fx.srv.sendQuantityOperation(context.Background(), state, op, time.Now().Add(-quantityOperationSendWindow-time.Second)) |
| if err != nil || !step.done || len(q.posts) != 0 { |
| t.Fatalf("a paused process sent: step=%+v err=%v posts=%v", step, err, q.posts) |
| } |
| + op.PeriodEnd = time.Now().Unix() - 1 |
| + step, err = fx.srv.sendQuantityOperation(context.Background(), state, op, time.Now()) |
| + if err != nil || !step.done || len(q.posts) != 0 { |
| + t.Fatalf("sent after the period ended: step=%+v err=%v posts=%v", step, err, q.posts) |
| + } |
| +} |
| + |
| +// The same pause through the production path: the window is measured from |
| +// the provider read, so a process that stalls between the read and the |
| +// request sends nothing. The operation stays outstanding for the next pass. |
| +func TestQuantityWorkerMeasuresThePauseFromTheProviderRead(t *testing.T) { |
| + 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) |
| + } |
| + prev := quantityOperationSendWindow |
| + quantityOperationSendWindow = 20 * time.Millisecond |
| + t.Cleanup(func() { quantityOperationSendWindow = prev }) |
| + fx.srv.billing.quantityBeforeSend = func() { time.Sleep(60 * time.Millisecond) } |
| + if err := fx.srv.billingQuantityPass(context.Background(), time.Now().UTC()); err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(q.posts) != 0 || len(pendingAdditions(t, fx)) != 1 { |
| + t.Fatalf("a stalled process sent: %+v", q.posts) |
| + } |
| + // Unstalled, the next pass finishes the same operation. |
| + quantityOperationSendWindow = prev |
| + fx.srv.billing.quantityBeforeSend = nil |
| + if err := fx.srv.billingQuantityPass(context.Background(), time.Now().UTC()); err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(q.posts) != 1 || q.posts[0]["quantity"] != "11" || quantityHeld(t, fx) { |
| + t.Fatalf("not finished afterwards: %+v", q.posts) |
| + } |
| +} |
| + |
| +// Two console processes. A reads the provider at 10, computes a target of |
| +// 11 and is paused BEFORE it records its operation. Meanwhile B runs a |
| +// whole cycle for a twelfth member: begin, send 12, reconcile, mark, |
| +// acknowledge. A wakes with a proposal computed from a world that is gone. |
| +// Recording it is a compare-and-set on that world, so it records nothing |
| +// and sends nothing: the quantity is never lowered to 11 and B's |
| +// acknowledged operation is not overwritten. |
| +func TestQuantityWorkerRefusesAStaleProposal(t *testing.T) { |
| + fx, q, now := newQuantityFixture(t, billingProrationImmediate) |
| + ctx := context.Background() |
| + if err := fx.st.AddMember(OpsAccountID, "eleventh@example.test", RoleMember, now.Add(time.Minute)); err != nil { |
| + t.Fatal(err) |
| + } |
| + procB := &Server{cfg: fx.srv.cfg, store: fx.srv.store, billing: &billingRuntime{locks: newStoreWalkLocks(), client: fx.srv.billing.client, cfg: fx.srv.billing.cfg}} |
| + paused := false |
| + fx.srv.billing.quantityBeforeBegin = func() { |
| + if paused { |
| + return |
| + } |
| + paused = true |
| + if err := fx.st.AddMember(OpsAccountID, "twelfth@example.test", RoleMember, time.Now().UTC()); err != nil { |
| + t.Error(err) |
| + } |
| + if err := procB.billingQuantityPass(ctx, time.Now().UTC()); err != nil { |
| + t.Errorf("process B: %v", err) |
| + } |
| + if q.quantity != 12 { |
| + t.Errorf("after process B the provider quantity is %d, want 12", q.quantity) |
| + } |
| + } |
| + if err := fx.srv.billingQuantityPass(ctx, time.Now().UTC()); err != nil { |
| + t.Fatal(err) |
| + } |
| + if !paused { |
| + t.Fatal("the seam never ran") |
| + } |
| + if q.lowered || q.quantity != 12 || len(q.posts) != 1 || q.posts[0]["quantity"] != "12" { |
| + t.Fatalf("provider quantity %d lowered=%v posts=%+v: the stale process sent", q.quantity, q.lowered, q.posts) |
| + } |
| + op, ok, err := fx.st.quantityOperationFor(OpsAccountID) |
| + if err != nil || !ok || op.Target != 12 || op.AckedAt == 0 { |
| + t.Fatalf("the newer operation was overwritten: %+v %v %v", op, ok, err) |
| + } |
| + if state, _ := fx.st.billingAccount(OpsAccountID); state.Quantity != 12 { |
| + t.Fatalf("billed quantity %d, want 12", state.Quantity) |
| + } |
| + if len(pendingAdditions(t, fx)) != 0 || quantityHeld(t, fx) { |
| + t.Fatal("additions left pending or the freeze still held") |
| + } |
| +} |
| + |
| +// The compare-and-set itself, case by case: an operation is recorded only |
| +// when the subscription, period, billed quantity, target and newest pending |
| +// addition are all still what the proposal was computed from. |
| +func TestBeginQuantityOperationIsACompareAndSet(t *testing.T) { |
| + fx, _, now := newQuantityFixture(t, billingProrationImmediate) |
| + if err := fx.st.AddMember(OpsAccountID, "eleventh@example.test", RoleMember, now.Add(time.Minute)); err != nil { |
| + t.Fatal(err) |
| + } |
| + state, err := fx.st.billingAccount(OpsAccountID) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + pending, err := pendingQuantityAdditions(fx.st.db, OpsAccountID, "sub_team", state.PeriodStart, 0) |
| + if err != nil || pending.Current != 1 { |
| + 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()} |
| + 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-- }, |
| + } { |
| + op := good |
| + op.Key = "k-" + name |
| + mutate(&op) |
| + if started, err := fx.st.beginQuantityOperation(op); err != nil || started { |
| + t.Fatalf("%s: started=%v err=%v, want refused", name, started, err) |
| + } |
| + } |
| + if _, ok, _ := fx.st.quantityOperationFor(OpsAccountID); ok { |
| + t.Fatal("a refused proposal left a row") |
| + } |
| + if started, err := fx.st.beginQuantityOperation(good); err != nil || !started { |
| + t.Fatalf("the true proposal: started=%v err=%v", started, err) |
| + } |
| + // One outstanding operation at a time. |
| + again := good |
| + again.Key = "k-again" |
| + if started, err := fx.st.beginQuantityOperation(again); err != nil || started { |
| + t.Fatalf("a second operation over an outstanding one: started=%v err=%v", started, err) |
| + } |
| + if op, _, _ := fx.st.quantityOperationFor(OpsAccountID); op.Key != "k-good" { |
| + t.Fatalf("the outstanding operation was replaced: %+v", op) |
| + } |
| +} |
| + |
| +// The subscription renews between the worker's provider read and its |
| +// request. The request dates its proration inside the period it was |
| +// computed for, the provider refuses that date in the new period, and |
| +// nothing is charged there. Once the console has seen the renewal the |
| +// operation is closed as never having run. |
| +func TestQuantityWorkerCannotChargeAcrossARenewal(t *testing.T) { |
| + fx, q, now := newQuantityFixture(t, billingProrationImmediate) |
| + ctx := context.Background() |
| + if err := fx.st.AddMember(OpsAccountID, "brief@example.test", RoleMember, now.Add(time.Minute)); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := fx.st.RemoveMember(OpsAccountID, "brief@example.test"); err != nil { |
| + t.Fatal(err) |
| + } |
| + fx.srv.billing.quantityBeforeSend = func() { |
| + // The provider renews: its current period now starts after the |
| + // instant the operation was computed at. |
| + q.mu.Lock() |
| + q.start = time.Now().Add(time.Hour).UTC() |
| + q.mu.Unlock() |
| + } |
| + if err := fx.srv.billingQuantityPass(ctx, time.Now().UTC()); err == nil { |
| + t.Fatal("the provider's refusal must surface") |
| + } |
| + 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) |
| + } |
| + 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 { |
| + t.Fatal(err) |
| + } |
| + if q.refused != 1 || len(q.posts) != 0 { |
| + t.Fatalf("re-sent before the renewal was reconciled: refused=%d posts=%+v", q.refused, q.posts) |
| + } |
| + // The console learns of the renewal. The operation never ran: it is |
| + // closed, the old period's addition is closed as not charged, and the |
| + // renewed subscription stays at ten. |
| + state, _ := fx.st.billingAccount(OpsAccountID) |
| + next := state |
| + next.PeriodStart, next.PeriodEnd, next.LiveMode = q.start.Unix(), q.start.AddDate(1, 0, 0).Unix(), true |
| + if err := fx.st.applyBillingState(next, state.Revision, "evt_quantity_renew", time.Now().UTC()); err != nil { |
| + t.Fatal(err) |
| + } |
| + publishQuantityEntitlement(t, fx, time.Now().UTC()) |
| + for i := 0; i < 2; i++ { |
| + if err := fx.srv.billingQuantityPass(ctx, time.Now().UTC()); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + if len(q.posts) != 0 || q.quantity != 10 { |
| + t.Fatalf("posts=%+v quantity=%d", q.posts, q.quantity) |
| + } |
| + if op, ok, _ := fx.st.quantityOperationFor(OpsAccountID); !ok || op.AckedAt == 0 { |
| + t.Fatalf("operation left outstanding: %+v", op) |
| + } |
| + if len(pendingAdditions(t, fx)) != 0 { |
| + t.Fatal("the old period's addition is still pending") |
| + } |
| + if rev, err := fx.st.billingReview(OpsAccountID, "sub_team", billingQuantityUnchargedReason); err != nil || rev.ResolvedAt != 0 { |
| + t.Fatalf("no open review for the unbilled seat: %+v %v", rev, err) |
| + } |
| +} |
| + |
| +// Every request carries the proration date of its operation, inside the |
| +// period, and a recovery re-sends the same date under the same key. |
| +func TestQuantityWorkerDatesTheProrationInsideThePeriod(t *testing.T) { |
| + 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") |
| + } |
| + op, ok, _ := fx.st.quantityOperationFor(OpsAccountID) |
| + if !ok || op.AckedAt != 0 || op.PeriodEnd == 0 { |
| + t.Fatalf("operation %+v", op) |
| + } |
| + q.failPost = false |
| + 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) || q.posts[0]["key"] != op.Key { |
| + t.Fatalf("posts %+v, want one dated %d under %s", q.posts, op.CreatedAt, op.Key) |
| + } |
| + if op.CreatedAt < op.PeriodStart || op.CreatedAt >= op.PeriodEnd { |
| + t.Fatalf("proration date %d outside %d..%d", op.CreatedAt, op.PeriodStart, op.PeriodEnd) |
| + } |
| +} |
| + |
| +// 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. |
| +func TestQuantityWorkerRecordsALandedOperationAfterARenewal(t *testing.T) { |
| + fx, q, now := newQuantityFixture(t, billingProrationImmediate) |
| + ctx := context.Background() |
| + if err := fx.st.AddMember(OpsAccountID, "eleventh@example.test", RoleMember, now.Add(time.Minute)); err != nil { |
| + t.Fatal(err) |
| + } |
| + q.failAfter = true |
| + if err := fx.srv.billingQuantityPass(ctx, time.Now().UTC()); err == nil { |
| + t.Fatal("a lost response must surface as an error") |
| + } |
| + if q.quantity != 11 { |
| + t.Fatalf("provider quantity %d, want 11", q.quantity) |
| + } |
| + // The renewal, reconciled by the console at the provider's quantity. |
| + state, _ := fx.st.billingAccount(OpsAccountID) |
| + next := state |
| + at := time.Now().UTC() |
| + next.Quantity, next.PeriodStart, next.PeriodEnd, next.LiveMode = 11, at.Unix(), at.AddDate(1, 0, 0).Unix(), true |
| + if err := fx.st.applyBillingState(next, state.Revision, "evt_quantity_renew", time.Now().UTC()); err != nil { |
| + t.Fatal(err) |
| + } |
| + q.mu.Lock() |
| + q.start = time.Unix(next.PeriodStart, 0).UTC() |
| + q.mu.Unlock() |
| + publishQuantityEntitlement(t, fx, time.Now().UTC()) |
| + for i := 0; i < 2; i++ { |
| + if err := fx.srv.billingQuantityPass(ctx, time.Now().UTC()); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + 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") |
| + 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) |
| + } |
| + if op, _, _ := fx.st.quantityOperationFor(OpsAccountID); op.AckedAt == 0 { |
| + t.Fatal("operation left outstanding") |
| + } |
| + 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. |
| +func TestQuantityWorkerFindsAnOperationWithNothingPending(t *testing.T) { |
| + fx, q, now := newQuantityFixture(t, billingProrationImmediate) |
| + ctx := context.Background() |
| + if err := fx.st.AddMember(OpsAccountID, "eleventh@example.test", RoleMember, now.Add(time.Minute)); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := fx.srv.billingQuantityPass(ctx, time.Now().UTC()); err != nil { |
| + t.Fatal(err) |
| + } |
| + // The mark and the acknowledgement commit together. |
| + if op, _, _ := fx.st.quantityOperationFor(OpsAccountID); op.AckedAt == 0 || len(pendingAdditions(t, fx)) != 0 { |
| + t.Fatalf("mark and acknowledgement not committed together: %+v", op) |
| + } |
| + if _, err := fx.st.db.Exec(`UPDATE billing_quantity_operations SET acked_at=0 WHERE account_id=?`, OpsAccountID); err != nil { |
| + t.Fatal(err) |
| + } |
| + ids, err := fx.st.pendingQuantityAccounts() |
| + if err != nil || len(ids) != 1 || ids[0] != OpsAccountID { |
| + t.Fatalf("an unacknowledged operation is not discovered: %v %v", ids, err) |
| + } |
| + if err := fx.srv.billingQuantityPass(ctx, time.Now().UTC()); err != nil { |
| + t.Fatal(err) |
| + } |
| + if op, _, _ := fx.st.quantityOperationFor(OpsAccountID); op.AckedAt == 0 { |
| + t.Fatal("operation still outstanding") |
| + } |
| + if len(q.posts) != 1 { |
| + t.Fatalf("finishing it sent again: %+v", q.posts) |
| + } |
| +} |
| + |
| +// A subscription that ended with an operation outstanding: the operation |
| +// is closed with its additions, and the provider is never called. |
| +func TestQuantityWorkerClosesAnOperationOfAnEndedSubscription(t *testing.T) { |
| + fx, q, now := newQuantityFixture(t, billingProrationImmediate) |
| + ctx := context.Background() |
| + 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(ctx, time.Now().UTC()); err == nil { |
| + t.Fatal("a failed request must surface") |
| + } |
| + q.failPost = false |
| + if _, err := fx.st.db.Exec(`UPDATE billing_accounts SET status='canceled' WHERE account_id=?`, OpsAccountID); err != nil { |
| + t.Fatal(err) |
| + } |
| + gets := q.gets |
| + if err := fx.srv.billingQuantityPass(ctx, time.Now().UTC()); err != nil { |
| + t.Fatal(err) |
| + } |
| + if q.gets != gets || len(q.posts) != 0 { |
| + t.Fatalf("an ended subscription reached the provider: gets %d -> %d posts=%v", gets, q.gets, q.posts) |
| + } |
| + if op, ok, _ := fx.st.quantityOperationFor(OpsAccountID); !ok || op.AckedAt == 0 { |
| + t.Fatalf("operation left outstanding: %+v", op) |
| + } |
| + if ids, _ := fx.st.pendingQuantityAccounts(); len(ids) != 0 { |
| + t.Fatalf("still discovered: %v", ids) |
| + } |
| + assertLastAddition(t, fx, quantityOutcomeIncluded, "subscription_ended:sub_team") |
| +} |
| + |
| +// An operation older than the provider's idempotency memory is never |
| +// re-sent: it would be a new request, not a replay. |
| +func TestQuantityWorkerNeverReplaysAForgottenKey(t *testing.T) { |
| + fx, q, now := newQuantityFixture(t, billingProrationImmediate) |
| + ctx := context.Background() |
| + 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(ctx, time.Now().UTC()); err == nil { |
| + t.Fatal("a failed request must surface") |
| + } |
| + q.failPost = false |
| + // The period began before the operation did, so the aged operation is |
| + // still inside it. |
| + aged := time.Now().Add(-quantityOperationReplayLimit - time.Minute).Unix() |
| + if _, err := fx.st.db.Exec(`UPDATE billing_quantity_operations SET created_at=?, idempotency_key='carloku-qty-aged' WHERE account_id=?`, aged, OpsAccountID); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := fx.srv.billingQuantityPass(ctx, time.Now().UTC()); err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(q.posts) != 0 { |
| + t.Fatalf("an expired operation was re-sent: %+v", q.posts) |
| + } |
| + // The addition is still owed: a fresh operation, under a fresh key. |
| + if err := fx.srv.billingQuantityPass(ctx, time.Now().UTC()); err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(q.posts) != 1 || q.posts[0]["quantity"] != "11" || q.posts[0]["key"] == "carloku-qty-aged" || q.posts[0]["proration_date"] == strconv.FormatInt(aged, 10) { |
| + t.Fatalf("posts %+v: want one fresh request under a fresh key and date", q.posts) |
| + } |
| +} |
| + |
| +// A mismatch review is filed once and left alone while it is open. Once an |
| +// operator resolves it, a mismatch that is still there files it again: the |
| +// resolution said it was dealt with, and it was not. |
| +func TestQuantityMismatchReviewIsFiledOncePerResolution(t *testing.T) { |
| + fx, q, now := newQuantityFixture(t, billingProrationImmediate) |
| + ctx := context.Background() |
| + if err := fx.st.AddMember(OpsAccountID, "eleventh@example.test", RoleMember, now.Add(time.Minute)); err != nil { |
| + t.Fatal(err) |
| + } |
| + q.mu.Lock() |
| + q.quantity = 13 |
| + q.mu.Unlock() |
| + for i := 0; i < 3; i++ { |
| + if err := fx.srv.billingQuantityPass(ctx, time.Now().UTC()); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + rev, err := fx.st.billingReview(OpsAccountID, "sub_team", billingQuantityMismatchReason) |
| + if err != nil || rev.ResolvedAt != 0 || rev.Occurrences != 1 { |
| + t.Fatalf("three passes over one open mismatch: %+v %v, want one occurrence", rev, err) |
| + } |
| + if _, err := fx.st.db.Exec(`UPDATE billing_reviews SET resolved_at=? WHERE account_id=? AND reason=?`, time.Now().Unix(), OpsAccountID, billingQuantityMismatchReason); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := fx.srv.billingQuantityPass(ctx, time.Now().UTC()); err != nil { |
| + t.Fatal(err) |
| + } |
| + rev, err = fx.st.billingReview(OpsAccountID, "sub_team", billingQuantityMismatchReason) |
| + if err != nil || rev.ResolvedAt != 0 || rev.Occurrences != 2 { |
| + t.Fatalf("a mismatch that outlived its resolution: %+v %v, want reopened with two occurrences", rev, err) |
| + } |
| + if len(q.posts) != 0 { |
| + t.Fatalf("acted on a mismatch: %+v", q.posts) |
| + } |
| } |
| // An operation left outstanding by a crash after the provider accepted it |
| @@ -529,4 +985,5 @@ func TestQuantityWorkerFinishesAnOutstandingOperation(t *testing.T) { |
| if len(q.posts) != 1 || quantityHeld(t, fx) { |
| t.Fatalf("posts %+v held %v", q.posts, quantityHeld(t, fx)) |
| } |
| + assertLastAddition(t, fx, quantityOutcomeCharged, "si_team:quantity=11:always_invoice") |
| } |
diff --git a/internal/console/billingservice.go b/internal/console/billingservice.go| index 2be2748cf..3d36c1e46 100644 |
| --- a/internal/console/billingservice.go |
| +++ b/internal/console/billingservice.go |
| @@ -42,9 +42,11 @@ type billingRuntime struct { |
| // invoice is the monthly overage engine's per-process state |
| // (billinginvoice.go): its lease owner id and, in tests only, its seams. |
| invoice billingInvoiceState |
| - // quantityBeforeSend is a test seam only: it runs immediately before |
| - // the quantity worker sends a provider update (billingquantity.go). |
| - quantityBeforeSend func() |
| + // quantityBeforeBegin and quantityBeforeSend are test seams only: they |
| + // run immediately before the quantity worker records an operation and |
| + // before it sends the provider update (billingquantity.go). |
| + quantityBeforeBegin func() |
| + quantityBeforeSend func() |
| } |
| var ( |