amadan

carlos / platform Public

Confirm a quantity operation by the provider's answer, and never retire one under its sender

Review round five on the quantity worker (Astra F2, F9; Opus N5, N8).
The policy switch stays unset; none of this runs in production.

An operation is confirmed only by the provider's success answer to its
own request. Only a confirmed operation dates the assignment at its
instant and is recorded as charged. A quantity that merely matches an
unconfirmed operation's target (a lost response, or someone's hand) is
recorded as landed_unverified, dated when observed, with a review.
Nothing is sent to find out which it was. A young unconfirmed operation
makes a reconcile wait, because its sender may be about to record the
answer.

An operation dated ahead of the observing clock is left outstanding. It
used to be retired, which let a replacement be recorded and completed
while the original sender, still inside its send window with a key the
provider had never seen, sent a lower target. A sender now re-reads its
operation immediately before sending and does not send one that is no
longer the outstanding one.

The operation's instant is never before the newest addition it charges
for, so a clock behind the one that recorded the member cannot bill the
seat for seconds before the member existed.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude 8936244e0d0dfa2c04a8febb6f02f7b57c1b8c4f parent 0a4c00e
4 files changed, +416 −58
  • internal/console/billingquantity.go +90 −16
  • internal/console/billingquantity_test.go +288 −34
  • internal/console/billingservice.go +21 −3
  • internal/console/billingstore.go +17 −5
diff --git a/internal/console/billingquantity.go b/internal/console/billingquantity.go
index 17a834bc8..e8cf1c806 100644
--- a/internal/console/billingquantity.go
+++ b/internal/console/billingquantity.go
@@ -130,6 +130,9 @@ type quantityPending struct {
Current, Stale int64 // how many
Newest int64 // newest generation among both
Peak int64 // peak desired quantity among the current-period ones
+ // NewestAt is the unix second by which every current-period addition
+ // had been recorded: the latest of them, rounded up.
+ NewestAt int64
}
type quantityRowQuerier interface {
@@ -143,11 +146,12 @@ func pendingQuantityAdditions(q quantityRowQuerier, accountID int64, subscriptio
var x quantityPending
err := q.QueryRow(`SELECT
COALESCE(SUM(period_start=?),0), COALESCE(SUM(period_start<>?),0), COALESCE(MAX(generation),0),
- COALESCE(MAX(CASE WHEN period_start=? THEN desired_quantity END),0)
+ COALESCE(MAX(CASE WHEN period_start=? THEN desired_quantity END),0),
+ COALESCE(MAX(CASE WHEN period_start=? THEN (occurred_at_ns+999999999)/1000000000 END),0)
FROM billing_quantity_events
WHERE account_id=? AND change=? AND processed_at=0 AND subscription=? AND (?=0 OR generation<=?)`,
- periodStart, periodStart, periodStart, accountID, quantityEventAddition, subscription, maxGeneration, maxGeneration).
- Scan(&x.Current, &x.Stale, &x.Newest, &x.Peak)
+ periodStart, periodStart, periodStart, periodStart, accountID, quantityEventAddition, subscription, maxGeneration, maxGeneration).
+ Scan(&x.Current, &x.Stale, &x.Newest, &x.Peak, &x.NewestAt)
return x, err
}
@@ -303,6 +307,11 @@ type quantityOperation struct {
// assignment takes effect at (billingservice.go), so what the customer
// pays for and what their allowance covers start together.
CreatedAt, AckedAt int64
+ // ConfirmedAt is set when the provider answered this operation's own
+ // request with success: the one piece of evidence that the quantity
+ // the provider carries was put there by this operation, at CreatedAt.
+ // Without it a matching quantity is only a matching quantity.
+ ConfirmedAt 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.
@@ -331,14 +340,17 @@ func (s *Store) migrateBillingQuantityOperations() error {
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`)
+ if err := s.addColumnIfMissing("billing_quantity_operations", "revision", `INTEGER NOT NULL DEFAULT 0`); err != nil {
+ return err
+ }
+ return s.addColumnIfMissing("billing_quantity_operations", "confirmed_at", `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,revision
+ 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,confirmed_at
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, &op.Revision)
+ 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, &op.ConfirmedAt)
if errors.Is(err, sql.ErrNoRows) {
return op, false, nil
}
@@ -390,6 +402,15 @@ func (s *Store) beginQuantityOperation(op quantityOperation) (started bool, err
if op.CreatedAt < state.PeriodStart || op.CreatedAt >= state.PeriodEnd {
return false, nil
}
+ // Nor may it be dated before an addition it charges for: the provider
+ // prorates the seat from this instant, and a clock behind the one that
+ // recorded the member would bill the seat for seconds before the
+ // member existed.
+ if covered, err := pendingQuantityAdditions(tx, op.AccountID, op.Subscription, op.PeriodStart, op.Generation); err != nil {
+ return false, err
+ } else if op.CreatedAt < covered.NewestAt {
+ return false, nil
+ }
pending, err := pendingQuantityAdditions(tx, op.AccountID, op.Subscription, op.PeriodStart, 0)
if err != nil {
return false, err
@@ -406,13 +427,20 @@ func (s *Store) beginQuantityOperation(op quantityOperation) (started bool, err
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,revision=excluded.revision`,
+ idempotency_key=excluded.idempotency_key,created_at=excluded.created_at,acked_at=0,revision=excluded.revision,confirmed_at=0`,
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()
}
+// confirmQuantityOperation records that the provider answered the
+// operation's own request with success.
+func (s *Store) confirmQuantityOperation(accountID int64, key string, now time.Time) error {
+ _, err := s.db.Exec(`UPDATE billing_quantity_operations SET confirmed_at=? WHERE account_id=? AND idempotency_key=? AND acked_at=0 AND confirmed_at=0`, max(now.Unix(), 1), accountID, key)
+ return err
+}
+
func (s *Store) ackQuantityOperation(accountID int64, key string, now time.Time) error {
_, err := s.db.Exec(`UPDATE billing_quantity_operations SET acked_at=? WHERE account_id=? AND idempotency_key=? AND acked_at=0`, max(now.Unix(), 1), accountID, key)
return err
@@ -519,7 +547,10 @@ func (s *Server) processQuantityAdditions(ctx context.Context, accountID int64,
if err := s.reconcileBillingSubscription(ctx, step.state.Subscription, "", now); err != nil {
return err
}
- outcome, reference = quantityOutcomeFor(step.behavior), step.reference
+ outcome, reference = step.outcome, step.reference
+ if outcome == "" {
+ outcome = quantityOutcomeUnverified
+ }
}
return s.store.markQuantityAdditionsProcessed(accountID, step.state.Subscription, step.state.PeriodStart, step.generation, outcome, reference, step.key, now)
}
@@ -530,6 +561,7 @@ 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
+ outcome string // with reference: how the additions are recorded
key string // non-empty: the operation to acknowledge with the mark
done bool // nothing to do, or not in a state to act
}
@@ -584,7 +616,7 @@ func (s *Server) raiseQuantity(ctx context.Context, acct *Account, behavior stri
case target:
// Already there, by a path that left no operation (an operator's
// hand). Nothing is sent; the reconcile records it.
- step.reference = quantityReference(item.ID, target, "observed")
+ step.reference, step.outcome = quantityReference(item.ID, target, "observed"), quantityOutcomeIncluded
return step, nil
case state.Quantity:
default:
@@ -603,7 +635,8 @@ func (s *Server) raiseQuantity(ctx context.Context, acct *Account, behavior stri
// 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)
+ // And never before the newest addition it charges for.
+ created := max(time.Now().Unix(), state.PeriodStart, pending.NewestAt)
if created >= state.PeriodEnd {
return quantityStep{done: true}, nil
}
@@ -659,6 +692,16 @@ func (s *Server) sendQuantityOperation(ctx context.Context, state billingState,
if time.Since(read) >= quantityOperationSendWindow || (op.PeriodEnd > 0 && time.Now().Unix() >= op.PeriodEnd) {
return quantityStep{done: true}, nil
}
+ // The operation must still be the account's outstanding one. Another
+ // process may have retired it while this one was between its read and
+ // here, and a retired operation's request, never having executed, is
+ // not protected by its idempotency key: sent now it would run for the
+ // first time, over whatever has been done since.
+ if cur, ok, err := s.store.quantityOperationFor(op.AccountID); err != nil {
+ return quantityStep{}, err
+ } else if !ok || cur.Key != op.Key || cur.AckedAt != 0 {
+ return quantityStep{done: true}, nil
+ }
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
@@ -677,7 +720,17 @@ func (s *Server) sendQuantityOperation(ctx context.Context, state billingState,
if updated.ID != op.Item || updated.Quantity != op.Target {
return quantityStep{}, errBillingIdentity
}
- return quantityStep{state: state, generation: op.Generation, behavior: op.Behavior, key: op.Key,
+ // The provider answered this operation's request with success. That is
+ // what lets the reconcile date the assignment at the operation's
+ // instant, so it is recorded before anything else can observe the new
+ // quantity through this process.
+ if err := s.store.confirmQuantityOperation(op.AccountID, op.Key, time.Now()); err != nil {
+ return quantityStep{}, err
+ }
+ if hook := b.quantityAfterSend; hook != nil {
+ hook()
+ }
+ return quantityStep{state: state, generation: op.Generation, behavior: op.Behavior, key: op.Key, outcome: quantityOutcomeFor(op.Behavior),
reference: quantityReference(op.Item, op.Target, op.Behavior)}, nil
}
@@ -722,14 +775,35 @@ func (s *Server) resolveQuantityOperation(ctx context.Context, acct *Account, st
}
return abandon()
case item.Quantity == op.Target:
- return quantityStep{state: state, generation: op.Generation, behavior: op.Behavior, key: op.Key,
+ // The provider carries the target. If this operation's request was
+ // answered with success (ConfirmedAt), the operation put it there.
+ // If not, the request's response was lost, or the request never
+ // ran and someone set the same number by hand: this code cannot
+ // tell which, and it does not send to find out (a request that
+ // never ran would run now, over whatever else has changed). The
+ // additions are recorded against the operation as landed and
+ // unverified, and the reconcile dates the assignment when it was
+ // observed and files a review.
+ outcome := quantityOutcomeUnverified
+ if op.ConfirmedAt != 0 {
+ outcome = quantityOutcomeFor(op.Behavior)
+ }
+ return quantityStep{state: state, generation: op.Generation, behavior: op.Behavior, key: op.Key, outcome: outcome,
reference: quantityReference(op.Item, op.Target, op.Behavior)}, nil
case item.Quantity != op.FromQuantity:
return s.quantityMismatch(acct, state, now)
- 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.
+ case read.Unix() < op.CreatedAt:
+ // Dated ahead of this process's clock. That is another console's
+ // clock, a few seconds on, or a period start this one has not
+ // reached. The process that recorded it may be about to send it,
+ // so it is NOT retired: retiring it would let that process's
+ // request run for the first time after a replacement had been
+ // made. It is left outstanding and looked at again next pass.
+ return quantityStep{done: true}, nil
+ case read.Unix()-op.CreatedAt >= int64(quantityOperationReplayLimit.Seconds()):
+ // Too old for the provider to still remember its key. The provider
+ // read above shows it did not land, and no sender can still be
+ // inside its send window, so it is closed and never re-sent.
return abandon()
}
return s.sendQuantityOperation(ctx, state, op, read)
diff --git a/internal/console/billingquantity_test.go b/internal/console/billingquantity_test.go
index 3289285ae..8edd687f2 100644
--- a/internal/console/billingquantity_test.go
+++ b/internal/console/billingquantity_test.go
@@ -3,6 +3,7 @@ package console
import (
"context"
"encoding/json"
+ "errors"
"fmt"
"net/http"
"net/http/httptest"
@@ -284,8 +285,13 @@ func TestQuantityWorkerChargesTheTransientSeat(t *testing.T) {
}
}
-// The update landed but its response was lost. The next pass reads the
-// provider, sees the target already there, sends nothing and finishes.
+// The update landed but its response was lost. While the operation is
+// young another process could still be mid-flight with it, so nothing is
+// concluded. After that, the next pass reads the provider, sees the target
+// there, sends nothing and finishes. It cannot know whether its lost
+// request or someone's hand set that quantity, so the additions are
+// recorded as landed and unverified, the assignment is dated when it was
+// observed, and a review holds the account until a person has looked.
func TestQuantityWorkerRecoversALostResponse(t *testing.T) {
fx, q, now := newQuantityFixture(t, billingProrationImmediate)
ctx := context.Background()
@@ -299,7 +305,15 @@ func TestQuantityWorkerRecoversALostResponse(t *testing.T) {
if len(pendingAdditions(t, fx)) != 1 || !quantityHeld(t, fx) {
t.Fatal("an unconfirmed update was marked processed")
}
- if err := fx.srv.billingQuantityPass(ctx, now.Add(3*time.Minute)); err != nil {
+ // At once: the operation is young and unconfirmed. Nothing is recorded.
+ if err := fx.srv.billingQuantityPass(ctx, time.Now().UTC()); !errors.Is(err, errBillingPending) {
+ t.Fatalf("a pass over a young unconfirmed operation: %v, want pending", err)
+ }
+ if state, _ := fx.st.billingAccount(OpsAccountID); state.Quantity != 10 || len(pendingAdditions(t, fx)) != 1 {
+ t.Fatalf("a young unconfirmed operation was settled: quantity %d", state.Quantity)
+ }
+ op := ageQuantityOperation(t, fx, 3*time.Minute)
+ if err := fx.srv.billingQuantityPass(ctx, time.Now().UTC()); err != nil {
t.Fatal(err)
}
if len(q.posts) != 1 {
@@ -308,9 +322,27 @@ 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(t, fx, quantityOutcomeUnverified, "si_team:quantity=11:always_invoice")
+ if got := quantityAssignment(t, fx); got.Quantity != 11 || got.EffectiveAt.Unix() == op.CreatedAt {
+ t.Fatalf("an unconfirmed landing was dated at the operation's instant: %+v", got.Assignment)
+ }
+ if rev, err := fx.st.billingReview(OpsAccountID, "sub_team", billingQuantityDateReviewReason(op.CreatedAt)); err != nil || rev.ResolvedAt != 0 {
+ t.Fatalf("no open review for the unverified landing: %+v %v", rev, err)
+ }
+}
+
+// ageQuantityOperation moves the account's outstanding operation back in
+// time, past the send window: no process can still be mid-flight with it.
+func ageQuantityOperation(t *testing.T, fx *consoleFixture, by time.Duration) quantityOperation {
+ t.Helper()
+ if _, err := fx.st.db.Exec(`UPDATE billing_quantity_operations SET created_at=created_at-? WHERE account_id=? AND acked_at=0`, int64(by.Seconds()), OpsAccountID); err != nil {
+ t.Fatal(err)
+ }
+ op, ok, err := fx.st.quantityOperationFor(OpsAccountID)
+ if err != nil || !ok {
+ t.Fatalf("no operation to age: %v", err)
+ }
+ return op
}
// assertLastAddition checks the newest addition was processed with outcome
@@ -485,9 +517,9 @@ func TestQuantityWorkerClosesAdditionsOfAnEndedSubscription(t *testing.T) {
// Two console processes. A reads a target of 11 and is paused before its
// request. B finishes A's operation with A's own key, then raises to 12 for
-// a member who joined meanwhile. A wakes and sends: the provider replays
-// A's key and changes nothing. The quantity never goes down, and every
-// addition is covered.
+// a member who joined meanwhile. A wakes, finds its operation is no longer
+// the outstanding one, and sends nothing. The quantity never goes down,
+// and every addition is covered.
func TestQuantityWorkerCannotLowerAQuantityAcrossProcesses(t *testing.T) {
fx, q, now := newQuantityFixture(t, billingProrationImmediate)
ctx := context.Background()
@@ -522,8 +554,8 @@ func TestQuantityWorkerCannotLowerAQuantityAcrossProcesses(t *testing.T) {
if q.lowered || q.quantity != 12 {
t.Fatalf("provider quantity %d, lowered=%v: a stale process lowered it", q.quantity, q.lowered)
}
- if q.replays != 1 || len(q.posts) != 2 || q.posts[0]["quantity"] != "11" || q.posts[1]["quantity"] != "12" {
- t.Fatalf("posts %+v replays %d: want 11 then 12 executed once each and one replay", q.posts, q.replays)
+ if q.replays != 0 || len(q.posts) != 2 || q.posts[0]["quantity"] != "11" || q.posts[1]["quantity"] != "12" {
+ t.Fatalf("posts %+v replays %d: want 11 then 12 executed once each, and nothing from the process that woke late", q.posts, q.replays)
}
if state, _ := fx.st.billingAccount(OpsAccountID); state.Quantity != 12 {
t.Fatalf("billed quantity %d, want 12", state.Quantity)
@@ -658,16 +690,17 @@ func TestBeginQuantityOperationIsACompareAndSet(t *testing.T) {
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(), 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 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 },
+ "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 },
+ "a date before the addition it charges for": func(op *quantityOperation) { op.CreatedAt = state.PeriodStart + 30 },
} {
op := good
op.Key = "k-" + name
@@ -956,28 +989,92 @@ func TestQuantityRetryDatesTheAllowanceAtTheProrationInstant(t *testing.T) {
}
}
-// 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.
+// The landing seen first by a webhook's reconcile, between the provider's
+// answer to the worker and the worker's own reconcile. The operation is
+// confirmed by then, so the webhook dates the assignment 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.
+ webhook := false
+ fx.srv.billing.quantityAfterSend = func() {
+ webhook = true
+ // The account lock is held by the worker in this process; the
+ // webhook lands in another.
+ other := &Server{cfg: fx.srv.cfg, store: fx.srv.store, billing: &billingRuntime{locks: newStoreWalkLocks(), client: fx.srv.billing.client, cfg: fx.srv.billing.cfg}}
+ if err := other.reconcileBillingSubscription(context.Background(), "sub_team", "evt_quantity_webhook", time.Now().UTC()); err != nil {
+ t.Errorf("webhook reconcile: %v", err)
+ }
+ }
+ if err := fx.srv.billingQuantityPass(context.Background(), time.Now().UTC()); err != nil {
+ t.Fatal(err)
+ }
+ if !webhook || len(q.posts) != 1 {
+ t.Fatalf("webhook ran=%v posts=%+v", webhook, q.posts)
+ }
+ 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)
+ }
+ assertLastAddition(t, fx, quantityOutcomeCharged, "si_team:quantity=11:always_invoice")
+ if quantityHeld(t, fx) {
+ t.Fatal("still held")
+ }
+}
+
+// A coincidence. The operation (10 to 11) was recorded twenty hours ago and
+// never executed. Someone then sets the subscription to 11 by hand, and a
+// webhook reconciles it. The quantity matches the operation's target, and
+// nothing shows the operation put it there: the provider prorated from the
+// hand's instant, not the operation's. The assignment must NOT be dated
+// back to the operation. It is dated when observed, a review holds the
+// account, and the worker records the additions as landed and unverified
+// without sending anything.
+func TestQuantityCoincidentalChangeIsNotBackdated(t *testing.T) {
+ fx, q, op := agedQuantityFixture(t)
+ if op.ConfirmedAt != 0 {
+ t.Fatal("fixture: the operation never executed and must be unconfirmed")
+ }
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 {
+ observed := time.Now().UTC()
+ if err := fx.srv.reconcileBillingSubscription(context.Background(), "sub_team", "evt_quantity_by_hand", observed); 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)
+ if got.Quantity != 11 || got.EffectiveAt.Unix() == op.CreatedAt || got.EffectiveAt.Unix() < observed.Unix()-5 {
+ t.Fatalf("a quantity someone else set was dated at the operation's instant %d: %+v", op.CreatedAt, 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 unproven charge date: %+v %v", rev, err)
}
- // 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))
+ if len(q.posts) != 0 {
+ t.Fatalf("the worker sent over a quantity someone else set: %+v", q.posts)
+ }
+ assertLastAddition(t, fx, quantityOutcomeUnverified, "si_team:quantity=11:always_invoice")
+}
+
+// A webhook that observes the target while the operation is young and
+// unconfirmed records nothing: the sender may be between the provider's
+// answer and writing it down.
+func TestQuantityWebhookOverAYoungUnconfirmedOperationWaits(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.failAfter = true
+ if err := fx.srv.billingQuantityPass(context.Background(), time.Now().UTC()); err == nil {
+ t.Fatal("a lost response must surface as an error")
+ }
+ if err := fx.srv.reconcileBillingSubscription(context.Background(), "sub_team", "evt_quantity_young", time.Now().UTC()); !errors.Is(err, errBillingPending) {
+ t.Fatalf("webhook over a young unconfirmed operation: %v, want pending", err)
+ }
+ if state, _ := fx.st.billingAccount(OpsAccountID); state.Quantity != 10 {
+ t.Fatalf("billed quantity %d: the change was recorded", state.Quantity)
}
}
@@ -987,6 +1084,11 @@ func TestQuantityLandingSeenByAWebhookIsDatedAtTheOperation(t *testing.T) {
// allowance starts at another, and nobody is billed on that unseen.
func TestQuantityLandingThatCannotBeDatedFilesAReview(t *testing.T) {
fx, q, op := agedQuantityFixture(t)
+ // The provider answered the operation's request with success, so the
+ // operation is known to be what set the quantity.
+ if _, err := fx.st.db.Exec(`UPDATE billing_quantity_operations SET confirmed_at=? WHERE account_id=?`, time.Now().Unix(), OpsAccountID); err != nil {
+ t.Fatal(err)
+ }
// A payment failure and recovery, both recorded after the operation's
// instant.
for i, status := range []string{"past_due", "active"} {
@@ -1015,6 +1117,48 @@ func TestQuantityLandingThatCannotBeDatedFilesAReview(t *testing.T) {
}
}
+// The two other guards on dating a confirmed operation's assignment at its
+// instant. An instant further ahead of the observing clock than clock skew
+// is not dated. An instant in a month already frozen is not dated. Each is
+// recorded when observed, with a review.
+func TestQuantityConfirmedLandingIsNotDatedAheadOfTheClockOrIntoAFrozenMonth(t *testing.T) {
+ for _, c := range []struct {
+ name string
+ prepare func(t *testing.T, fx *consoleFixture, op quantityOperation) time.Time // returns the observation instant
+ }{
+ {"ten minutes ahead of the observing clock", func(t *testing.T, fx *consoleFixture, op quantityOperation) time.Time {
+ return time.Unix(op.CreatedAt, 0).Add(-10 * time.Minute).UTC()
+ }},
+ {"in a month already frozen", func(t *testing.T, fx *consoleFixture, op quantityOperation) time.Time {
+ month := time.Unix(op.CreatedAt, 0).UTC().Format("2006-01")
+ if _, err := fx.st.db.Exec(`INSERT INTO billing_months(account_id,month,calculation,digest,state) VALUES(?,?,x'00','d','frozen')`, OpsAccountID, month); err != nil {
+ t.Fatal(err)
+ }
+ return time.Now().UTC()
+ }},
+ } {
+ t.Run(c.name, func(t *testing.T) {
+ fx, q, op := agedQuantityFixture(t)
+ if _, err := fx.st.db.Exec(`UPDATE billing_quantity_operations SET confirmed_at=? WHERE account_id=?`, time.Now().Unix(), OpsAccountID); err != nil {
+ t.Fatal(err)
+ }
+ observed := c.prepare(t, fx, op)
+ q.mu.Lock()
+ q.quantity = 11
+ q.mu.Unlock()
+ if err := fx.srv.reconcileBillingSubscription(context.Background(), "sub_team", "evt_quantity_guard", observed); err != nil {
+ t.Fatal(err)
+ }
+ if got := quantityAssignment(t, fx); got.Quantity != 11 || got.EffectiveAt.Unix() == op.CreatedAt {
+ t.Fatalf("dated at the operation's instant: %+v", got.Assignment)
+ }
+ if rev, err := fx.st.billingReview(OpsAccountID, "sub_team", billingQuantityDateReviewReason(op.CreatedAt)); err != nil || rev.ResolvedAt != 0 {
+ t.Fatalf("no open review: %+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
@@ -1048,6 +1192,115 @@ func TestQuantityOperationIsDatedInsideThePeriodUnderClockSkew(t *testing.T) {
}
}
+// The interleaving of review round five. The stored period starts three
+// seconds ahead of this process's clock. Worker A records a 10-to-11
+// operation, dated at that period start, and pauses before sending. Worker
+// B reads the provider at 10 and sees an operation dated ahead of its own
+// clock. If B retires it, a replacement can be recorded and completed
+// (10 to 12 for another member), and A, waking inside its send window with
+// a key the provider has never seen, sends 11 and LOWERS the quantity.
+//
+// B must leave it alone. A's operation stays the outstanding one, nothing
+// else can begin, A sends 11, and the twelfth seat follows in order.
+func TestQuantityFutureDatedOperationIsNotRetiredUnderItsSender(t *testing.T) {
+ fx, q, _ := newQuantityFixture(t, billingProrationImmediate)
+ ctx := context.Background()
+ 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)
+ }
+ 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.quantityBeforeSend = func() {
+ if paused {
+ return
+ }
+ paused = true
+ mine, _, _ := fx.st.quantityOperationFor(OpsAccountID)
+ if mine.CreatedAt <= time.Now().Unix() {
+ t.Errorf("fixture: the operation is not dated ahead of the clock (%d)", mine.CreatedAt)
+ }
+ if err := procB.billingQuantityPass(ctx, time.Now().UTC()); err != nil {
+ t.Errorf("process B: %v", err)
+ }
+ 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, second pass: %v", err)
+ }
+ after, ok, _ := fx.st.quantityOperationFor(OpsAccountID)
+ if !ok || after.Key != mine.Key || after.AckedAt != 0 {
+ t.Errorf("the future-dated operation was retired or replaced under its sender: %+v", after)
+ }
+ if len(q.posts) != 0 {
+ t.Errorf("process B sent while another process held the operation: %+v", q.posts)
+ }
+ }
+ if err := fx.srv.billingQuantityPass(ctx, time.Now().UTC()); err != nil {
+ t.Fatal(err)
+ }
+ if !paused {
+ t.Fatal("the seam never ran")
+ }
+ fx.srv.billing.quantityBeforeSend = nil
+ // The twelfth seat, in order.
+ time.Sleep(time.Until(time.Unix(ahead, 0)) + 1100*time.Millisecond)
+ if err := fx.srv.billingQuantityPass(ctx, time.Now().UTC()); err != nil {
+ t.Fatal(err)
+ }
+ if q.lowered || q.quantity != 12 || len(q.posts) != 2 || q.posts[0]["quantity"] != "11" || q.posts[1]["quantity"] != "12" {
+ t.Fatalf("provider quantity %d lowered=%v posts=%+v", q.quantity, q.lowered, q.posts)
+ }
+}
+
+// A sender whose operation was retired while it was paused does not send:
+// its request has never executed, so its key protects nothing.
+func TestQuantitySenderDoesNotSendARetiredOperation(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)
+ }
+ fx.srv.billing.quantityBeforeSend = func() {
+ if _, err := fx.st.db.Exec(`UPDATE billing_quantity_operations SET acked_at=? WHERE account_id=?`, time.Now().Unix(), OpsAccountID); err != nil {
+ t.Error(err)
+ }
+ }
+ 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 retired operation was sent: %+v", q.posts)
+ }
+}
+
+// The member was recorded by a console whose clock is five seconds ahead
+// of this one. The proration is dated no earlier than the addition, so the
+// customer is not charged a seat for seconds before the member existed.
+func TestQuantityOperationIsNotDatedBeforeTheAdditionItChargesFor(t *testing.T) {
+ fx, q, _ := newQuantityFixture(t, billingProrationImmediate)
+ added := time.Now().UTC().Add(5 * time.Second)
+ if err := fx.st.AddMember(OpsAccountID, "eleventh@example.test", RoleMember, added); err != nil {
+ t.Fatal(err)
+ }
+ if err := fx.srv.billingQuantityPass(context.Background(), time.Now().UTC()); err != nil {
+ t.Fatal(err)
+ }
+ if len(q.posts) != 1 {
+ t.Fatalf("posts %+v", q.posts)
+ }
+ date, _ := strconv.ParseInt(q.posts[0]["proration_date"], 10, 64)
+ if date < added.Unix() {
+ t.Fatalf("proration dated %d, before the member was added at %d", date, added.Unix())
+ }
+}
+
// 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.
@@ -1197,9 +1450,10 @@ func TestQuantityWorkerFinishesAnOutstandingOperation(t *testing.T) {
if err := fx.srv.billingQuantityPass(ctx, time.Now().UTC()); err == nil {
t.Fatal("a lost response must surface as an error")
}
- if op, ok, _ := fx.st.quantityOperationFor(OpsAccountID); !ok || op.AckedAt != 0 || op.Target != 11 {
- t.Fatalf("operation not left outstanding: %+v %v", op, ok)
+ if op, ok, _ := fx.st.quantityOperationFor(OpsAccountID); !ok || op.AckedAt != 0 || op.Target != 11 || op.ConfirmedAt != 0 {
+ t.Fatalf("operation not left outstanding and unconfirmed: %+v %v", op, ok)
}
+ ageQuantityOperation(t, fx, 3*time.Minute)
if err := fx.srv.billingQuantityPass(ctx, time.Now().UTC()); err != nil {
t.Fatal(err)
}
@@ -1209,7 +1463,7 @@ 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")
+ assertLastAddition(t, fx, quantityOutcomeUnverified, "si_team:quantity=11:always_invoice")
}
// The processing guard names the outcomes it admits. A database whose guard
diff --git a/internal/console/billingservice.go b/internal/console/billingservice.go
index 46c75ad70..6b5770a81 100644
--- a/internal/console/billingservice.go
+++ b/internal/console/billingservice.go
@@ -47,6 +47,9 @@ type billingRuntime struct {
// before it sends the provider update (billingquantity.go).
quantityBeforeBegin func()
quantityBeforeSend func()
+ // quantityAfterSend runs after the provider accepted an update and the
+ // operation was confirmed, before the worker's own reconcile.
+ quantityAfterSend func()
}
var (
@@ -488,7 +491,7 @@ func (s *Server) reconcileBillingSubscription(ctx context.Context, id, eventID s
// 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 := ""
+ quantityKey, quantityConfirmed := "", false
if state.Subscription == id && next.Quantity != state.Quantity {
op, ok, err := s.store.quantityOperationFor(acct.ID)
if err != nil {
@@ -496,10 +499,25 @@ func (s *Server) reconcileBillingSubscription(ctx context.Context, id, eventID s
}
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
+ // The quantity the provider now carries is the target of the
+ // account's outstanding operation. Only the provider's own
+ // success answer to that operation (ConfirmedAt) shows the
+ // operation is what changed it. Without that answer:
+ //
+ // - a young operation may be mid-flight in another process,
+ // its answer not yet recorded. Nothing is recorded now; the
+ // event is retried and the worker's next pass settles it;
+ // - an older one is ambiguous for good (a lost response, or
+ // someone set the same number by hand). The change is
+ // recorded when observed, and the store files a review,
+ // because the instant the provider charged from is unknown.
+ if op.ConfirmedAt == 0 && time.Now().Unix()-op.CreatedAt < int64(quantityOperationSendWindow.Seconds()) {
+ return errBillingPending
+ }
+ quantityStart, quantityKey, quantityConfirmed = time.Unix(op.CreatedAt, 0).UTC(), op.Key, op.ConfirmedAt != 0
}
}
- return s.store.applyBillingStateDated(next, state.Revision, eventID, effective, renewalStart, quantityStart, quantityKey)
+ return s.store.applyBillingStateDated(next, state.Revision, eventID, effective, renewalStart, quantityStart, quantityKey, quantityConfirmed)
}
// provableRenewal reports whether current -> next is exactly one uninterrupted
diff --git a/internal/console/billingstore.go b/internal/console/billingstore.go
index c8624afe7..16953ff83 100644
--- a/internal/console/billingstore.go
+++ b/internal/console/billingstore.go
@@ -353,7 +353,7 @@ 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{}, "")
+ return s.applyBillingStateDated(next, expectedRevision, eventID, effective, renewalStart, time.Time{}, "", false)
}
// billingClockTolerance is how far ahead of this process's clock an
@@ -381,7 +381,12 @@ func billingQuantityDateReviewReason(at int64) string {
// 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 {
+//
+// confirmed says the provider answered that operation's own request with
+// success. Without it the quantity only matches the operation's target,
+// which someone else could have set at another time, so the instant is
+// never used: the assignment is dated when observed and the review filed.
+func (s *Store) applyBillingStateDated(next billingState, expectedRevision int64, eventID string, effective, renewalStart, quantityStart time.Time, quantityKey string, confirmed bool) 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")
}
@@ -445,9 +450,16 @@ func (s *Store) applyBillingStateDated(next billingState, expectedRevision int64
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 {
+ var confirmedAt int64
+ err := tx.QueryRow(`SELECT confirmed_at 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(&confirmedAt)
+ switch {
+ case errors.Is(err, sql.ErrNoRows):
+ case err != nil:
return err
+ default:
+ outstanding = true
+ confirmed = confirmed && confirmedAt != 0
}
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 &&
@@ -457,7 +469,7 @@ func (s *Store) applyBillingStateDated(next billingState, expectedRevision int64
// 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 {
+ if outstanding && confirmed && 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