amadan
Fix the findings of review round two and of the Opus production review
Astra round 2 rejected (seven P1, one P2) and the Opus production review accepted with fixes (two P1, three P2, three P3). All were checked against the code and accepted; the dispositions file lists each fix and its reproduction. Quantity worker: one persisted operation per account, so two console processes cannot send conflicting targets and a stale, lower one cannot land after a newer one (the same idempotency key finishes an outstanding operation; a process paused past two minutes does not send). Only additions of the subscription's current period raise it; additions of a period that has renewed are closed uncharged with a review; additions of an ended subscription are closed without a provider call. House marker: the checkout intent transaction re-reads the marker and refuses a coupon that disagrees with it; the door takes the account's billing lock and audits only a real change. Bucket ownership: completeness needs a grant block positively recorded as minting, no delivered credentials and a probe showing the bucket does not exist yet. Teardown closes the interval only after the credentials are revoked. The month evidence projection carries the completeness boundary. Report classifier: any non-usage line with a cost on a transfer usage type holds. Durability fold: a box another source expected must be expected by the durability source for the same span. Month catch-up: walks back to where metering began, checks every account's statement, and logs a month held more than six. Verified: go build, go vet, gofmt and go test ./... plus the vendored Litestream S3 package pass outside the sandbox (60 packages ok). Three of the new reproductions were confirmed to fail with their fixes disabled. Round three of the reviews has not run yet.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
22 files changed,
+1244
−182
internal/awsbilling/classify.go+14 −4internal/awsbilling/classify_test.go+16 −1internal/console/billingoperator.go+32 −21internal/console/billingoperator_test.go+67 −0internal/console/billingquantity.go+399 −96internal/console/billingquantity_test.go+226 −1internal/console/billingservice.go+3 −0internal/console/billingstore.go+10 −0internal/console/billingstore_test.go+5 −0internal/console/durabilityfold.go+19 −7internal/console/economicspass.go+63 −25internal/console/meterfolds_test.go+92 −0internal/console/templates/billing.html+1 −1internal/objstore/aws.go+16 −0internal/objstore/memprovisioner.go+12 −0internal/objstore/objstore.go+6 −0internal/objstore/ownership.go+4 −2internal/objstore/ownership_test.go+154 −0internal/objstore/provisioner.go+10 −0internal/objstore/walk.go+37 −17internal/usage/billable_test.go+41 −0internal/usage/evidence_month.go+17 −7
diff --git a/internal/awsbilling/classify.go b/internal/awsbilling/classify.go| index d7822e4cf..8dea26c05 100644 |
| --- a/internal/awsbilling/classify.go |
| +++ b/internal/awsbilling/classify.go |
| @@ -36,9 +36,14 @@ const ( |
| // is a hold, because AWS charged something nobody has been shown to |
| // owe. |
| ReasonUnexpectedCharge = "unexpected_transfer_charge" |
| - ReasonNoResource = "no_bucket_resource_id" |
| - ReasonNotHourly = "not_an_hourly_line" |
| - ReasonBadAmount = "non_positive_amount" |
| + // A refund, credit, fee or other non-usage line on an S3 transfer usage |
| + // type with a cost that is not zero. AWS changed what it charged for |
| + // transfer in the month and the usage lines alone no longer say what |
| + // anyone owes, so the month holds until a person has reconciled it. |
| + ReasonTransferAdjustment = "transfer_adjustment" |
| + ReasonNoResource = "no_bucket_resource_id" |
| + ReasonNotHourly = "not_an_hourly_line" |
| + ReasonBadAmount = "non_positive_amount" |
| ) |
| // IsHold reports whether a reason describes a CHARGED line the pipeline |
| @@ -48,7 +53,7 @@ const ( |
| // absorbing it. |
| func IsHold(reason string) bool { |
| switch reason { |
| - case ReasonUnknownTransfer, ReasonUnexpectedCharge, ReasonNoResource, ReasonNotHourly, ReasonBadAmount: |
| + case ReasonUnknownTransfer, ReasonUnexpectedCharge, ReasonTransferAdjustment, ReasonNoResource, ReasonNotHourly, ReasonBadAmount: |
| return true |
| } |
| return false |
| @@ -113,6 +118,11 @@ func ChargedS3Transfer(l Line) (bucket string, bytes int64, charged bool, reason |
| return "", 0, false, ReasonNotS3 |
| } |
| if !l.IsUsage() { |
| + if l.UsageType != "" && l.Cost != nil && l.Cost.Sign() != 0 { |
| + if out, why := classifyUsageType(l.UsageType); out || why != ReasonNotTransfer { |
| + return "", 0, false, ReasonTransferAdjustment |
| + } |
| + } |
| return "", 0, false, ReasonNotUsage |
| } |
| out, why := classifyUsageType(l.UsageType) |
diff --git a/internal/awsbilling/classify_test.go b/internal/awsbilling/classify_test.go| index 2cc5e5a1d..0b4781abd 100644 |
| --- a/internal/awsbilling/classify_test.go |
| +++ b/internal/awsbilling/classify_test.go |
| @@ -68,10 +68,25 @@ func TestChargedS3TransferTable(t *testing.T) { |
| l.ServiceCode = "AmazonEC2" |
| return l |
| }, want{false, 0, ReasonNotS3}}, |
| - {"credit line", func() Line { |
| + {"a credit on transfer holds: AWS changed what it charged", func() Line { |
| l := s3Line(t, "EUW1-DataTransfer-Out-Bytes", "0", "-0.0135") |
| l.LineItemType = "Credit" |
| return l |
| + }, want{false, 0, ReasonTransferAdjustment}}, |
| + {"a refund on transfer holds", func() Line { |
| + l := s3Line(t, "EUW1-DataTransfer-Out-Bytes", "1", "-0.09") |
| + l.LineItemType = "Refund" |
| + return l |
| + }, want{false, 0, ReasonTransferAdjustment}}, |
| + {"a credit on storage is not transfer", func() Line { |
| + l := s3Line(t, "EUW1-TimedStorage-ByteHrs", "0", "-0.01") |
| + l.LineItemType = "Credit" |
| + return l |
| + }, want{false, 0, ReasonNotUsage}}, |
| + {"a zero-cost credit on transfer changes nothing", func() Line { |
| + l := s3Line(t, "EUW1-DataTransfer-Out-Bytes", "0", "0") |
| + l.LineItemType = "Credit" |
| + return l |
| }, want{false, 0, ReasonNotUsage}}, |
| {"tax line", func() Line { |
| l := s3Line(t, "", "0", "0.05") |
diff --git a/internal/console/billingoperator.go b/internal/console/billingoperator.go| index 9de084bcd..37df56ab2 100644 |
| --- a/internal/console/billingoperator.go |
| +++ b/internal/console/billingoperator.go |
| @@ -78,59 +78,59 @@ func (s *Store) openBillingReviews() ([]billingReview, error) { |
| // row, in one transaction. Asking for the state already in place writes |
| // nothing. It refuses while a subscription that is not over, or any |
| // checkout intent, exists for the account. |
| -func (s *Store) setBillingHouse(accountID int64, house bool, actor, note string, now time.Time) error { |
| +func (s *Store) setBillingHouse(accountID int64, house bool, actor, note string, now time.Time) (changed bool, err error) { |
| if accountID <= 0 || now.Unix() <= 0 || !validReviewText(actor, billingReviewActorMax) || !validReviewText(note, billingReviewNoteMax) { |
| - return errBillingHouseInvalid |
| + return false, errBillingHouseInvalid |
| } |
| tx, err := s.db.Begin() |
| if err != nil { |
| - return err |
| + return false, err |
| } |
| defer tx.Rollback() |
| // Reserve the writer before the reads, as the other billing writers do. |
| if _, err := tx.Exec(`UPDATE billing_accounts SET house=house WHERE account_id=?`, accountID); err != nil { |
| - return err |
| + return false, err |
| } |
| var operator bool |
| if err := tx.QueryRow(`SELECT EXISTS(SELECT 1 FROM users WHERE address=? AND operator<>0)`, actor).Scan(&operator); err != nil { |
| - return err |
| + return false, err |
| } |
| if !operator { |
| - return errBillingReviewActor |
| + return false, errBillingReviewActor |
| } |
| var exists bool |
| if err := tx.QueryRow(`SELECT EXISTS(SELECT 1 FROM accounts WHERE id=?)`, accountID).Scan(&exists); err != nil { |
| - return err |
| + return false, err |
| } |
| if !exists { |
| - return errBillingHouseInvalid |
| + return false, errBillingHouseInvalid |
| } |
| var current bool |
| var subscription, status string |
| err = tx.QueryRow(`SELECT house,subscription,status FROM billing_accounts WHERE account_id=?`, accountID).Scan(¤t, &subscription, &status) |
| if err != nil && !errors.Is(err, sql.ErrNoRows) { |
| - return err |
| + return false, err |
| } |
| if current == house { |
| - return nil |
| + return false, nil |
| } |
| if subscription != "" && status != "canceled" && status != "incomplete_expired" { |
| - return errBillingHouseSubscribed |
| + return false, errBillingHouseSubscribed |
| } |
| var intent bool |
| if err := tx.QueryRow(`SELECT EXISTS(SELECT 1 FROM billing_checkout_intents WHERE account_id=?)`, accountID).Scan(&intent); err != nil { |
| - return err |
| + return false, err |
| } |
| if intent { |
| - return errBillingHouseCheckout |
| + return false, errBillingHouseCheckout |
| } |
| if _, err := tx.Exec(`INSERT INTO billing_accounts(account_id,house) VALUES(?,?) ON CONFLICT(account_id) DO UPDATE SET house=excluded.house`, accountID, house); err != nil { |
| - return err |
| + return false, err |
| } |
| if _, err := tx.Exec(`INSERT INTO billing_house_changes(account_id,house,actor,note,changed_at) VALUES(?,?,?,?,?)`, accountID, house, actor, note, now.Unix()); err != nil { |
| - return err |
| + return false, err |
| } |
| - return tx.Commit() |
| + return true, tx.Commit() |
| } |
| func (s *Store) billingHouseAccounts() ([]int64, error) { |
| @@ -276,14 +276,25 @@ func (s *Server) handleAPIBillingHouseSet(w http.ResponseWriter, r *http.Request |
| jsonError(w, http.StatusBadRequest, "account, house and a note are required") |
| return |
| } |
| - err = s.store.setBillingHouse(acct.ID, *req.House, addr, strings.TrimSpace(req.Note), time.Now()) |
| + // The same per-account lock the checkout and reconcile paths hold, so |
| + // the marker never moves while one of them is between its read and its |
| + // write. (beginBillingCheckout re-checks it in its own transaction as |
| + // well; the lock covers the consoles that share this process.) |
| + if s.billing != nil { |
| + defer s.billing.locks.lock(acct.Sqid, "billing")() |
| + } |
| + changed, err := s.store.setBillingHouse(acct.ID, *req.House, addr, strings.TrimSpace(req.Note), time.Now()) |
| switch { |
| case err == nil: |
| - action := "billing-house-set" |
| - if !*req.House { |
| - action = "billing-house-clear" |
| + // A request for the state already in place changed nothing and |
| + // leaves no trail claiming it did. |
| + if changed { |
| + action := "billing-house-set" |
| + if !*req.House { |
| + action = "billing-house-clear" |
| + } |
| + s.audit(acct, "", addr, action, strings.TrimSpace(req.Note)) |
| } |
| - s.audit(acct, "", addr, action, strings.TrimSpace(req.Note)) |
| writeJSON(w, http.StatusOK, map[string]any{"account": acct.Sqid, "house": *req.House}) |
| case errors.Is(err, errBillingHouseInvalid): |
| jsonError(w, http.StatusBadRequest, "account, house and a note are required") |
diff --git a/internal/console/billingoperator_test.go b/internal/console/billingoperator_test.go| index 9286f53ea..e28fd5ce5 100644 |
| --- a/internal/console/billingoperator_test.go |
| +++ b/internal/console/billingoperator_test.go |
| @@ -3,9 +3,12 @@ package console |
| import ( |
| "bytes" |
| "encoding/json" |
| + "errors" |
| "net/http" |
| "testing" |
| "time" |
| + |
| + "github.com/carlosframework/platform/internal/plans" |
| ) |
| // rawStatus issues a request and returns only the status: a refused door |
| @@ -206,3 +209,67 @@ func TestBillingOperatorHouseMarker(t *testing.T) { |
| t.Fatalf("unknown account = %d, want 400", status) |
| } |
| } |
| + |
| +// The coupon is chosen from the house marker checkout read several provider |
| +// round trips earlier. If the marker moved in between, the intent must not |
| +// be created: a house account would be charged in full, or a paying one |
| +// handed the house coupon. |
| +func TestCheckoutIntentRefusesAStaleHouseSnapshot(t *testing.T) { |
| + fx := newConsoleFixture(t) |
| + now := time.Now() |
| + want := billingIntent{AccountID: OpsAccountID, Plan: plans.PlanPersonal, Terms: plans.CurrentTermsVersion, Band: "A", Price: "price_x", Quantity: 1, Country: "IE", LocDigest: "d", Members: 1} |
| + // The request is good in every other respect: with the marker as |
| + // checkout read it, it begins. Only then are the refusals below about |
| + // the marker and nothing else. |
| + if _, err := fx.st.beginBillingCheckout(want, now); err != nil { |
| + t.Fatalf("fixture: a consistent request was refused: %v", err) |
| + } |
| + if _, err := fx.st.db.Exec(`DELETE FROM billing_checkout_intents WHERE account_id=?`, OpsAccountID); err != nil { |
| + t.Fatal(err) |
| + } |
| + // Checkout read house=false and chose no coupon; the operator then |
| + // marked the account. |
| + if _, err := fx.st.db.Exec(`INSERT INTO billing_accounts(account_id,house) VALUES(?,1) ON CONFLICT(account_id) DO UPDATE SET house=1`, OpsAccountID); err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := fx.st.beginBillingCheckout(want, now); !errors.Is(err, errBillingConflict) || errors.Is(err, errBillingMembershipChanged) { |
| + t.Fatalf("a house account got an intent with no coupon: %v", err) |
| + } |
| + // The reverse: checkout read house=true and chose the coupon; the |
| + // marker was then cleared. |
| + if _, err := fx.st.db.Exec(`UPDATE billing_accounts SET house=0 WHERE account_id=?`, OpsAccountID); err != nil { |
| + t.Fatal(err) |
| + } |
| + want.Coupon = "coupon_house" |
| + if _, err := fx.st.beginBillingCheckout(want, now); !errors.Is(err, errBillingConflict) { |
| + t.Fatalf("a paying account got an intent with the house coupon: %v", err) |
| + } |
| + var intents int |
| + if err := fx.st.db.QueryRow(`SELECT COUNT(*) FROM billing_checkout_intents WHERE account_id=?`, OpsAccountID).Scan(&intents); err != nil || intents != 0 { |
| + t.Fatalf("a refused intent was written: %d %v", intents, err) |
| + } |
| +} |
| + |
| +// A request for the state already in place changes nothing and must not |
| +// leave an audit entry saying it did. |
| +func TestBillingOperatorHouseNoOpLeavesNoAuditTrail(t *testing.T) { |
| + fx := newConsoleFixture(t) |
| + token := billingOperatorToken(t, fx, "ops@keymail.dev", "code-billing-op-noop") |
| + count := func() int { |
| + var n int |
| + if err := fx.st.db.QueryRow(`SELECT COUNT(*) FROM billing_house_changes WHERE account_id=?`, OpsAccountID).Scan(&n); err != nil { |
| + t.Fatal(err) |
| + } |
| + return n |
| + } |
| + changed, err := fx.st.setBillingHouse(OpsAccountID, false, "ops@keymail.dev", "already so", time.Now()) |
| + if err != nil || changed { |
| + t.Fatalf("clearing an unset marker: changed=%v err=%v", changed, err) |
| + } |
| + if status, _ := postJSON(t, fx.ts.URL+"/api/billing/house", map[string]any{"account": fx.sqid, "house": false, "note": "already so"}, token); status != http.StatusOK { |
| + t.Fatalf("no-op = %d", status) |
| + } |
| + if count() != 0 { |
| + t.Fatalf("a no-op wrote %d house change rows", count()) |
| + } |
| +} |
diff --git a/internal/console/billingquantity.go b/internal/console/billingquantity.go| index f0200eed1..c0b0f8d9e 100644 |
| --- a/internal/console/billingquantity.go |
| +++ b/internal/console/billingquantity.go |
| @@ -9,6 +9,7 @@ import ( |
| "time" |
| "github.com/carlosframework/platform/internal/billingstripe" |
| + "github.com/carlosframework/platform/internal/plans" |
| ) |
| // The team quantity worker (pricing spec, "Member changes update the |
| @@ -28,25 +29,32 @@ import ( |
| // |
| // What quantity: removals apply at renewal, so inside a period the billed |
| // quantity only ever rises. The target is the PEAK desired quantity over |
| -// the unprocessed additions, not today's member count. A member added on |
| -// Monday and removed on Tuesday still occupied a seat that is paid for |
| -// until renewal. |
| +// the unprocessed additions of the subscription's CURRENT period, and never |
| +// less than the seats occupied now. A member added on Monday and removed on |
| +// Tuesday still occupied a seat that is paid for until renewal. An addition |
| +// from a period that has since renewed never raises anything: it is closed |
| +// as not charged and filed for review. An addition bound to a subscription |
| +// that is over, or that is no longer the account's, is closed the same way |
| +// without a review: there is nothing left to charge it to. |
| // |
| -// Order, and why it is crash safe: |
| +// Order, and why it is safe against a crash and against a second console |
| +// process: |
| // |
| -// 1. Read the subscription from the provider and check it is the one the |
| -// console bills. If it already carries the target, an earlier attempt |
| -// landed and nothing is sent. |
| -// 2. Otherwise update the item to the target, under an idempotency key |
| -// fixed by the account, subscription, newest addition and target. |
| -// 3. Reconcile the subscription: the authoritative read that moves |
| +// 1. Finish any operation left unacknowledged (quantityOperation) before |
| +// 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 |
| // billing_accounts.quantity and appends the assignment revision. |
| -// 4. Mark the additions processed, in a transaction that re-checks the |
| -// billed quantity covers them. |
| +// 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. |
| // |
| -// A crash after 2 is repaired by 1 on the next pass; after 3, by 4 alone. |
| -// 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. 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 |
| @@ -60,6 +68,9 @@ const ( |
| quantityOutcomeIncluded = "included" |
| billingQuantityMismatchReason = "billing_quantity_mismatch" |
| + // Filed when additions from a period that has since renewed are closed |
| + // without a charge: a person should know seats went unbilled. |
| + billingQuantityUnchargedReason = "billing_quantity_uncharged_period" |
| ) |
| // prorationBehavior maps the configured policy to the provider's behaviour |
| @@ -94,23 +105,89 @@ func (s *Store) pendingQuantityAccounts() ([]int64, error) { |
| return out, rows.Err() |
| } |
| -// pendingQuantityAdditions summarises the unprocessed additions recorded |
| -// against subscription: how many, the newest generation, and the peak |
| -// desired quantity among them. |
| -func pendingQuantityAdditions(q interface { |
| +// quantityPending is the account's unprocessed additions bound to its |
| +// current subscription, split by the period they were recorded in. |
| +// |
| +// Only an addition recorded in the subscription's CURRENT period may raise |
| +// it: the seat it asks for is a seat in this period. An addition from a |
| +// period that has since renewed (Stale) asks for a seat in a period that is |
| +// over; raising the subscription now would charge a prorated seat in a |
| +// period that member may never have belonged to. |
| +type quantityPending struct { |
| + Current, Stale int64 // how many |
| + Newest int64 // newest generation among both |
| + Peak int64 // peak desired quantity among the current-period ones |
| +} |
| + |
| +type quantityRowQuerier interface { |
| QueryRow(string, ...any) *sql.Row |
| -}, accountID int64, subscription string, maxGeneration int64) (count, newest, peak int64, err error) { |
| - err = q.QueryRow(`SELECT COUNT(*), COALESCE(MAX(generation),0), COALESCE(MAX(desired_quantity),0) FROM billing_quantity_events |
| +} |
| + |
| +// pendingQuantityAdditions reads quantityPending for subscription, whose |
| +// current period starts at periodStart. maxGeneration of zero means no |
| +// upper bound. |
| +func pendingQuantityAdditions(q quantityRowQuerier, accountID int64, subscription string, periodStart, maxGeneration int64) (quantityPending, error) { |
| + 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) |
| + FROM billing_quantity_events |
| WHERE account_id=? AND change=? AND processed_at=0 AND subscription=? AND (?=0 OR generation<=?)`, |
| - accountID, quantityEventAddition, subscription, maxGeneration, maxGeneration).Scan(&count, &newest, &peak) |
| - return count, newest, peak, err |
| + periodStart, periodStart, periodStart, accountID, quantityEventAddition, subscription, maxGeneration, maxGeneration). |
| + Scan(&x.Current, &x.Stale, &x.Newest, &x.Peak) |
| + return x, err |
| +} |
| + |
| +func billingSubscriptionOver(state billingState) bool { |
| + return state.Subscription == "" || state.Status == "canceled" || state.Status == "incomplete_expired" |
| +} |
| + |
| +// 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. |
| +func (s *Store) closeEndedQuantityAdditions(accountID int64, now time.Time) (over bool, err error) { |
| + tx, err := s.db.Begin() |
| + if err != nil { |
| + return false, err |
| + } |
| + defer tx.Rollback() |
| + if _, err := lockMembershipWritesTx(tx); err != nil { |
| + return false, err |
| + } |
| + state, err := scanBilling(tx.QueryRow(`SELECT `+billingColumns+` FROM billing_accounts WHERE account_id=?`, accountID), accountID) |
| + if err != nil { |
| + return false, err |
| + } |
| + over = billingSubscriptionOver(state) |
| + res, err := tx.Exec(`UPDATE billing_quantity_events SET processed_at=max(?,occurred_at_ns/1000000000+1), processing_outcome=?, processing_reference='subscription_ended:'||subscription |
| + WHERE account_id=? AND change=? AND processed_at=0 AND subscription<>'' AND (subscription<>? OR ?)`, |
| + now.Unix(), quantityOutcomeIncluded, accountID, quantityEventAddition, state.Subscription, over) |
| + if err != nil { |
| + return false, err |
| + } |
| + if n, err := res.RowsAffected(); err != nil { |
| + return false, err |
| + } else if n > 0 { |
| + if err := refreshQuantityChangeTx(tx, accountID, now); err != nil { |
| + return false, err |
| + } |
| + } |
| + return over, tx.Commit() |
| } |
| // markQuantityAdditionsProcessed marks the unprocessed additions of |
| -// subscription up to maxGeneration, once the billed quantity covers them. |
| -// It refuses with errBillingPending when the reconcile that moves the |
| -// billed quantity has not landed. |
| -func (s *Store) markQuantityAdditionsProcessed(accountID int64, subscription string, maxGeneration int64, outcome, reference string, now time.Time) error { |
| +// 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 |
| +// 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 { |
| tx, err := s.db.Begin() |
| if err != nil { |
| return err |
| @@ -121,25 +198,37 @@ func (s *Store) markQuantityAdditionsProcessed(accountID int64, subscription str |
| if _, err := lockMembershipWritesTx(tx); err != nil { |
| return err |
| } |
| - count, _, peak, err := pendingQuantityAdditions(tx, accountID, subscription, maxGeneration) |
| + pending, err := pendingQuantityAdditions(tx, accountID, subscription, periodStart, maxGeneration) |
| if err != nil { |
| return err |
| } |
| - if count == 0 { |
| + if pending.Current == 0 && pending.Stale == 0 { |
| return tx.Commit() |
| } |
| - var billed int64 |
| + var billed, currentStart int64 |
| var current string |
| - if err := tx.QueryRow(`SELECT quantity,subscription FROM billing_accounts WHERE account_id=?`, accountID).Scan(&billed, ¤t); err != nil { |
| + if err := tx.QueryRow(`SELECT quantity,subscription,period_start FROM billing_accounts WHERE account_id=?`, accountID).Scan(&billed, ¤t, ¤tStart); err != nil { |
| return err |
| } |
| - if current != subscription || billed < peak { |
| + if current != subscription || currentStart != periodStart || billed < pending.Peak { |
| return errBillingPending |
| } |
| - if _, 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 generation<=?`, |
| - now.Unix(), outcome, reference, accountID, quantityEventAddition, subscription, maxGeneration); err != nil { |
| - return err |
| + if pending.Stale > 0 { |
| + if _, err := tx.Exec(`UPDATE billing_quantity_events SET processed_at=max(?,occurred_at_ns/1000000000+1), processing_outcome=?, processing_reference='period_renewed_uncharged:'||period_start |
| + WHERE account_id=? AND change=? AND processed_at=0 AND subscription=? AND period_start<>? AND generation<=?`, |
| + now.Unix(), quantityOutcomeIncluded, accountID, quantityEventAddition, subscription, periodStart, maxGeneration); err != nil { |
| + return err |
| + } |
| + if err := fileBillingReviewTx(tx, accountID, subscription, billingQuantityUnchargedReason, now); err != nil { |
| + return err |
| + } |
| + } |
| + if pending.Current > 0 { |
| + if _, 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, accountID, quantityEventAddition, subscription, periodStart, maxGeneration); err != nil { |
| + return err |
| + } |
| } |
| if err := refreshQuantityChangeTx(tx, accountID, now); err != nil { |
| return err |
| @@ -147,14 +236,90 @@ func (s *Store) markQuantityAdditionsProcessed(accountID int64, subscription str |
| return tx.Commit() |
| } |
| -// billingQuantityPass processes every account with pending additions. With |
| -// no proration policy configured it returns at once. |
| +// 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. |
| +type quantityOperation struct { |
| + AccountID int64 |
| + Subscription, Item string |
| + PeriodStart 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 |
| +) |
| + |
| +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, |
| + 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 |
| +} |
| + |
| +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 |
| + 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) |
| + 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. |
| +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) |
| + if err != nil { |
| + return false, err |
| + } |
| + n, err := res.RowsAffected() |
| + return n == 1, 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 |
| +} |
| + |
| +// billingQuantityPass processes every account with pending additions or an |
| +// operation in flight. With no proration policy configured it returns at |
| +// once. |
| func (s *Server) billingQuantityPass(ctx context.Context, now time.Time) error { |
| b := s.billing |
| if b == nil { |
| return nil |
| } |
| - behavior, outcome, ok := b.cfg.prorationBehavior() |
| + behavior, _, ok := b.cfg.prorationBehavior() |
| if !ok { |
| return nil |
| } |
| @@ -167,14 +332,21 @@ func (s *Server) billingQuantityPass(ctx context.Context, now time.Time) error { |
| if ctx.Err() != nil { |
| return errors.Join(append(failures, ctx.Err())...) |
| } |
| - if err := s.processQuantityAdditions(ctx, id, behavior, outcome, now); err != nil { |
| + if err := s.processQuantityAdditions(ctx, id, behavior, now); err != nil { |
| failures = append(failures, err) |
| } |
| } |
| return errors.Join(failures...) |
| } |
| -func (s *Server) processQuantityAdditions(ctx context.Context, accountID int64, behavior, outcome string, now time.Time) error { |
| +func quantityOutcomeFor(behavior string) string { |
| + if behavior == billingstripe.ProrationAlwaysInvoice { |
| + return quantityOutcomeCharged |
| + } |
| + return quantityOutcomeIncluded |
| +} |
| + |
| +func (s *Server) processQuantityAdditions(ctx context.Context, accountID int64, behavior string, now time.Time) error { |
| acct, err := s.store.AccountByID(accountID) |
| if err != nil { |
| return err |
| @@ -182,98 +354,229 @@ func (s *Server) processQuantityAdditions(ctx context.Context, accountID int64, |
| if acct == nil { |
| return nil |
| } |
| - state, newest, reference, done, err := s.raiseQuantity(ctx, acct, behavior, now) |
| - if err != nil || done { |
| + // Additions nothing can charge any more are closed first, with no |
| + // provider call. |
| + over, err := s.store.closeEndedQuantityAdditions(accountID, now) |
| + if err != nil || over { |
| return err |
| } |
| - if reference != "" { |
| + step, err := s.raiseQuantity(ctx, acct, behavior, now) |
| + if err != nil || step.done { |
| + return err |
| + } |
| + outcome, reference := quantityOutcomeIncluded, "quantity_already_billed:"+strconv.FormatInt(step.state.Quantity, 10) |
| + if step.reference != "" { |
| // The provider carries the target. The reconcile is the read that |
| // moves billing_accounts.quantity and records the assignment. |
| - if err := s.reconcileBillingSubscription(ctx, state.Subscription, "", now); err != nil { |
| + if err := s.reconcileBillingSubscription(ctx, step.state.Subscription, "", now); err != nil { |
| return err |
| } |
| - } else { |
| - // The quantity already billed covers every addition (a team below |
| - // its minimum, or a repair after a crash): nothing was sent now. |
| - outcome = quantityOutcomeIncluded |
| - reference = "quantity_already_billed:" + strconv.FormatInt(state.Quantity, 10) |
| + 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 s.store.markQuantityAdditionsProcessed(accountID, state.Subscription, newest, outcome, reference, now) |
| + return nil |
| +} |
| + |
| +// quantityStep is what raiseQuantity decided under the account's lock. |
| +type quantityStep struct { |
| + state billingState |
| + 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 |
| + done bool // nothing to do, or not in a state to act |
| } |
| -// raiseQuantity is steps 1 and 2 under the account's billing lock. done is |
| -// true when there is nothing to process or the subscription is not in a |
| -// state to be raised; reference is non-empty when the provider carries the |
| -// target and a reconcile must follow. |
| -func (s *Server) raiseQuantity(ctx context.Context, acct *Account, behavior string, now time.Time) (state billingState, newest int64, reference string, done bool, err error) { |
| +// raiseQuantity is steps 1 and 2 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")() |
| - state, err = s.store.billingAccount(acct.ID) |
| + state, err := s.store.billingAccount(acct.ID) |
| if err != nil { |
| - return state, 0, "", false, err |
| + return quantityStep{}, err |
| + } |
| + if billingSubscriptionOver(state) { |
| + return quantityStep{done: true}, nil |
| } |
| - if state.Subscription == "" { |
| - return state, 0, "", true, nil |
| + if op, ok, err := s.store.quantityOperationFor(acct.ID); err != nil { |
| + return quantityStep{}, err |
| + } else if ok && op.AckedAt == 0 { |
| + return s.resolveQuantityOperation(ctx, acct, state, op, now) |
| } |
| - count, newest, peak, err := pendingQuantityAdditions(s.store.db, acct.ID, state.Subscription, 0) |
| + pending, err := pendingQuantityAdditions(s.store.db, acct.ID, state.Subscription, state.PeriodStart, 0) |
| if err != nil { |
| - return state, 0, "", false, err |
| + return quantityStep{}, err |
| + } |
| + if pending.Current == 0 && pending.Stale == 0 { |
| + return quantityStep{done: true}, nil |
| } |
| - if count == 0 { |
| - return state, 0, "", 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)) |
| } |
| - if peak <= state.Quantity { |
| - return state, newest, "", false, nil |
| + step := quantityStep{state: state, generation: pending.Newest, behavior: behavior} |
| + if target <= state.Quantity { |
| + return step, nil |
| } |
| // Only a healthy subscription is raised: adding a charge to one whose |
| - // last payment failed, or that is over, helps nobody. The additions |
| - // stay pending and hold. |
| + // last payment failed helps nobody. The additions stay pending and |
| + // hold. |
| if state.Status != "active" { |
| - return state, 0, "", true, nil |
| + return quantityStep{done: true}, nil |
| + } |
| + item, ok, err := s.quantityProviderItem(ctx, acct, state) |
| + if err != nil || !ok { |
| + return quantityStep{done: !ok}, err |
| + } |
| + 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" |
| + 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)} |
| + started, err := s.store.beginQuantityOperation(op) |
| + if err != nil { |
| + return quantityStep{}, err |
| + } |
| + if !started { |
| + // Another process began one between our read and our write. |
| + return quantityStep{done: true}, nil |
| } |
| + return s.sendQuantityOperation(ctx, state, op, time.Now()) |
| +} |
| + |
| +// quantityProviderItem reads the subscription from the provider and checks |
| +// it is the one the console bills, active, and in the period the console |
| +// recorded. ok is false when it is not in a state to act on (the next |
| +// reconcile catches the console up). |
| +func (s *Server) quantityProviderItem(ctx context.Context, acct *Account, state billingState) (billingstripe.SubscriptionItem, bool, error) { |
| + b := s.billing |
| sub, err := b.client.GetSubscription(ctx, state.Subscription) |
| if err != nil { |
| - return state, 0, "", false, err |
| + return billingstripe.SubscriptionItem{}, false, err |
| } |
| if sub.ID != state.Subscription || sub.LiveMode != b.cfg.LiveMode || sub.CustomerID != state.Customer || |
| sub.Metadata["account_sqid"] != acct.Sqid || len(sub.Items) != 1 || sub.ItemsHasMore || |
| sub.Items[0].PriceID != b.cfg.Prices[billingPriceKey(state.Terms, state.Plan, state.Band)] { |
| - return state, 0, "", false, errBillingIdentity |
| + return billingstripe.SubscriptionItem{}, false, errBillingIdentity |
| } |
| - if sub.Status != "active" { |
| - return state, 0, "", true, nil |
| + if sub.Status != "active" || sub.CurrentPeriodStart.Unix() != state.PeriodStart { |
| + return billingstripe.SubscriptionItem{}, false, nil |
| } |
| - item := sub.Items[0] |
| - switch item.Quantity { |
| - case peak: |
| - // An earlier attempt landed and was not recorded. |
| - case state.Quantity: |
| - key := fmt.Sprintf("carloku-qty-%s-%s-%d-%d", acct.Sqid, state.Subscription, newest, peak) |
| - updated, err := b.client.UpdateQuantityWith(ctx, item.ID, peak, behavior, key) |
| - if err != nil { |
| - return state, 0, "", false, err |
| - } |
| - if updated.ID != item.ID || updated.Quantity != peak { |
| - return state, 0, "", false, errBillingIdentity |
| - } |
| - default: |
| - // Neither what the console bills nor what it wants: someone else |
| - // changed the subscription. A person decides. |
| - if ferr := s.store.fileBillingReview(acct.ID, state.Subscription, billingQuantityMismatchReason, now); ferr != nil { |
| - return state, 0, "", false, ferr |
| - } |
| - return state, 0, "", true, nil |
| + 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) { |
| + b := s.billing |
| + if hook := b.quantityBeforeSend; hook != nil { |
| + hook() |
| } |
| - return state, newest, item.ID + ":quantity=" + strconv.FormatInt(peak, 10) + ":" + behavior, false, nil |
| + if time.Since(began) >= quantityOperationSendWindow { |
| + return quantityStep{done: true}, nil |
| + } |
| + updated, err := b.client.UpdateQuantityWith(ctx, op.Item, op.Target, op.Behavior, op.Key) |
| + if err != nil { |
| + return quantityStep{}, err |
| + } |
| + 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, |
| + reference: op.Item + ":quantity=" + strconv.FormatInt(op.Target, 10) + ":" + op.Behavior}, nil |
| } |
| -// fileBillingReview files one review occurrence in its own transaction. |
| -func (s *Store) fileBillingReview(accountID int64, subscription, reason string, observed time.Time) error { |
| +// resolveQuantityOperation finishes an operation some process left |
| +// unacknowledged, before anything newer may begin. It never sends a |
| +// request for a period or subscription that has moved on, and never |
| +// re-sends one the provider has forgotten the key of. |
| +func (s *Server) resolveQuantityOperation(ctx context.Context, acct *Account, state billingState, op quantityOperation, now time.Time) (quantityStep, error) { |
| + abandon := func() (quantityStep, error) { |
| + // Provably not executed, and no longer wanted as it stands: the |
| + // additions stay pending and the next pass starts afresh. |
| + return quantityStep{done: true}, s.store.ackQuantityOperation(acct.ID, op.Key, now) |
| + } |
| + if op.Subscription != state.Subscription { |
| + return abandon() |
| + } |
| + 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. |
| + 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. |
| + return abandon() |
| + case item.Quantity != op.FromQuantity: |
| + return s.quantityMismatch(acct, state, now) |
| + case now.Unix()-op.CreatedAt >= int64(quantityOperationReplayLimit.Seconds()), now.Unix() < op.CreatedAt: |
| + return abandon() |
| + } |
| + return s.sendQuantityOperation(ctx, state, op, time.Now()) |
| +} |
| + |
| +// quantityMismatch files the review for a provider quantity that is |
| +// neither what the console bills nor what it wants. It is filed once and |
| +// left alone while open, so its occurrence count does not move under the |
| +// operator reading it. |
| +func (s *Server) quantityMismatch(acct *Account, state billingState, now time.Time) (quantityStep, error) { |
| + return quantityStep{done: true}, s.store.fileBillingReviewOnce(acct.ID, state.Subscription, billingQuantityMismatchReason, now) |
| +} |
| + |
| +// fileBillingReviewOnce files a review occurrence unless one is already |
| +// open for the same account, subscription and reason. |
| +func (s *Store) fileBillingReviewOnce(accountID int64, subscription, reason string, observed time.Time) error { |
| tx, err := s.db.Begin() |
| if err != nil { |
| return err |
| } |
| defer tx.Rollback() |
| + var open bool |
| + if err := tx.QueryRow(`SELECT EXISTS(SELECT 1 FROM billing_reviews WHERE account_id=? AND subscription=? AND reason=? AND resolved_at=0)`, accountID, subscription, reason).Scan(&open); err != nil { |
| + return err |
| + } |
| + if open { |
| + return nil |
| + } |
| if err := fileBillingReviewTx(tx, accountID, subscription, reason, observed); err != nil { |
| return err |
| } |
diff --git a/internal/console/billingquantity_test.go b/internal/console/billingquantity_test.go| index 8e15238fb..ad0e5011b 100644 |
| --- a/internal/console/billingquantity_test.go |
| +++ b/internal/console/billingquantity_test.go |
| @@ -7,6 +7,7 @@ import ( |
| "net/http" |
| "net/http/httptest" |
| "strconv" |
| + "strings" |
| "sync" |
| "testing" |
| "time" |
| @@ -27,6 +28,11 @@ type quantityStripe struct { |
| 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 |
| + replays int |
| + lowered bool // the quantity ever went down |
| } |
| func (q *quantityStripe) serve(w http.ResponseWriter, r *http.Request) { |
| @@ -50,7 +56,20 @@ func (q *quantityStripe) serve(w http.ResponseWriter, r *http.Request) { |
| return |
| } |
| n, _ := strconv.ParseInt(r.PostForm.Get("quantity"), 10, 64) |
| - q.posts = append(q.posts, map[string]string{"quantity": r.PostForm.Get("quantity"), "proration_behavior": r.PostForm.Get("proration_behavior"), "key": r.Header.Get("Idempotency-Key")}) |
| + key := r.Header.Get("Idempotency-Key") |
| + if prev, seen := q.byKey[key]; seen { |
| + 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[key] = n |
| + q.posts = append(q.posts, map[string]string{"quantity": r.PostForm.Get("quantity"), "proration_behavior": r.PostForm.Get("proration_behavior"), "key": key}) |
| + if n < q.quantity { |
| + q.lowered = true |
| + } |
| q.quantity = n |
| if q.failAfter { |
| q.failAfter = false |
| @@ -305,3 +324,209 @@ func TestQuantityWorkerLeavesAPastDueSubscriptionAlone(t *testing.T) { |
| t.Fatalf("raised a past-due subscription: gets=%d posts=%v", q.gets, q.posts) |
| } |
| } |
| + |
| +// 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. |
| +func renewQuantitySubscription(t *testing.T, fx *consoleFixture, q *quantityStripe, at time.Time) { |
| + t.Helper() |
| + state, err := fx.st.billingAccount(OpsAccountID) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + next := state |
| + next.PeriodStart, next.PeriodEnd, next.LiveMode = state.PeriodEnd, time.Unix(state.PeriodEnd, 0).AddDate(1, 0, 0).Unix(), true |
| + if err := fx.st.applyBillingState(next, state.Revision, "evt_quantity_renew", at); err != nil { |
| + t.Fatal(err) |
| + } |
| + q.mu.Lock() |
| + q.start = time.Unix(next.PeriodStart, 0).UTC() |
| + q.mu.Unlock() |
| + publishQuantityEntitlement(t, fx, at) |
| +} |
| + |
| +// An addition from a period that has since renewed must never raise the |
| +// renewed subscription. The member joined and left before the renewal; the |
| +// policy was switched on afterwards. Nothing is sent, the addition is |
| +// closed as not charged, and a review says seats went unbilled. |
| +func TestQuantityWorkerNeverChargesAnExpiredPeriodsAddition(t *testing.T) { |
| + fx, q, now := newQuantityFixture(t, "") |
| + 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) |
| + } |
| + renewQuantitySubscription(t, fx, q, time.Now().UTC()) |
| + fx.srv.billing.cfg.Proration = billingProrationImmediate |
| + if err := fx.srv.billingQuantityPass(context.Background(), time.Now().UTC()); err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(q.posts) != 0 { |
| + t.Fatalf("an expired period's addition raised the renewed subscription: %+v", q.posts) |
| + } |
| + events, _ := fx.st.billingQuantityEvents(OpsAccountID) |
| + var add billingQuantityEvent |
| + for _, e := range events { |
| + if e.Change == quantityEventAddition { |
| + add = e |
| + } |
| + } |
| + if add.ProcessedAt == 0 || add.ProcessingOutcome != quantityOutcomeIncluded || !strings.HasPrefix(add.ProcessingReference, "period_renewed_uncharged:") { |
| + t.Fatalf("stale addition not closed as uncharged: %+v", add) |
| + } |
| + if rev, err := fx.st.billingReview(OpsAccountID, "sub_team", billingQuantityUnchargedReason); err != nil || rev.ResolvedAt != 0 { |
| + t.Fatalf("no open review for the unbilled period: %+v %v", rev, err) |
| + } |
| + if state, _ := fx.st.billingAccount(OpsAccountID); state.Quantity != 10 { |
| + t.Fatalf("billed quantity %d, want 10", state.Quantity) |
| + } |
| +} |
| + |
| +// The same, but the member is still here after the renewal. They occupy a |
| +// seat in the new period, so the subscription is raised once, from now, and |
| +// the old period's addition is still closed as not charged. |
| +func TestQuantityWorkerSeatsAMemberWhoOutlivedTheirPeriod(t *testing.T) { |
| + fx, q, now := newQuantityFixture(t, "") |
| + if err := fx.st.AddMember(OpsAccountID, "eleventh@example.test", RoleMember, now.Add(time.Minute)); err != nil { |
| + t.Fatal(err) |
| + } |
| + renewQuantitySubscription(t, fx, q, time.Now().UTC()) |
| + fx.srv.billing.cfg.Proration = billingProrationImmediate |
| + 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" { |
| + t.Fatalf("provider update: %+v", q.posts) |
| + } |
| + if state, _ := fx.st.billingAccount(OpsAccountID); state.Quantity != 11 { |
| + t.Fatalf("billed quantity %d, want 11", state.Quantity) |
| + } |
| + 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 period: %+v %v", rev, err) |
| + } |
| +} |
| + |
| +// An addition bound to a subscription that is over can never be charged. It |
| +// is closed with no provider call, so it cannot hold this account's freeze, |
| +// or a later subscription's, for good. |
| +func TestQuantityWorkerClosesAdditionsOfAnEndedSubscription(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) |
| + } |
| + if _, err := fx.st.db.Exec(`UPDATE billing_accounts SET status='canceled' WHERE account_id=?`, OpsAccountID); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := fx.srv.billingQuantityPass(context.Background(), time.Now().UTC()); err != nil { |
| + t.Fatal(err) |
| + } |
| + if q.gets != 0 || len(q.posts) != 0 { |
| + t.Fatalf("an ended subscription reached the provider: gets=%d posts=%v", q.gets, q.posts) |
| + } |
| + events, _ := fx.st.billingQuantityEvents(OpsAccountID) |
| + last := events[len(events)-1] |
| + if last.ProcessedAt == 0 || last.ProcessingReference != "subscription_ended:sub_team" || last.ProcessingOutcome != quantityOutcomeIncluded { |
| + t.Fatalf("addition not closed: %+v", last) |
| + } |
| + if len(pendingAdditions(t, fx)) != 0 { |
| + t.Fatal("still pending") |
| + } |
| +} |
| + |
| +// 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. |
| +func TestQuantityWorkerCannotLowerAQuantityAcrossProcesses(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) |
| + } |
| + // Process B: the same database and provider, its own in-process locks. |
| + 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 |
| + if err := fx.st.AddMember(OpsAccountID, "twelfth@example.test", RoleMember, time.Now().UTC()); err != nil { |
| + t.Error(err) |
| + } |
| + // B's first pass finishes A's outstanding operation; its second |
| + // raises for the new member. |
| + for i := 0; i < 2; i++ { |
| + if err := procB.billingQuantityPass(ctx, time.Now().UTC()); err != nil { |
| + t.Errorf("process B pass %d: %v", i, 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 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 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") |
| + } |
| + if op, ok, err := fx.st.quantityOperationFor(OpsAccountID); err != nil || !ok || op.AckedAt == 0 || op.Target != 12 { |
| + t.Fatalf("operation left outstanding: %+v %v %v", op, ok, err) |
| + } |
| +} |
| + |
| +// A process paused past the send window does not send at all. |
| +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()} |
| + 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) |
| + } |
| +} |
| + |
| +// An operation left outstanding by a crash after the provider accepted it |
| +// is finished by the next pass with the same key, and only then |
| +// acknowledged. |
| +func TestQuantityWorkerFinishesAnOutstandingOperation(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 op, ok, _ := fx.st.quantityOperationFor(OpsAccountID); !ok || op.AckedAt != 0 || op.Target != 11 { |
| + t.Fatalf("operation not left outstanding: %+v %v", op, ok) |
| + } |
| + 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 after recovery") |
| + } |
| + if len(q.posts) != 1 || quantityHeld(t, fx) { |
| + t.Fatalf("posts %+v held %v", q.posts, quantityHeld(t, fx)) |
| + } |
| +} |
diff --git a/internal/console/billingservice.go b/internal/console/billingservice.go| index bc8018b83..2be2748cf 100644 |
| --- a/internal/console/billingservice.go |
| +++ b/internal/console/billingservice.go |
| @@ -42,6 +42,9 @@ 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() |
| } |
| var ( |
diff --git a/internal/console/billingstore.go b/internal/console/billingstore.go| index be20682a3..ca1a6ab9a 100644 |
| --- a/internal/console/billingstore.go |
| +++ b/internal/console/billingstore.go |
| @@ -154,6 +154,9 @@ func (s *Store) migrateBilling() error { |
| if err := s.migrateBillingOperator(); err != nil { |
| return err |
| } |
| + if err := s.migrateBillingQuantityOperations(); err != nil { |
| + return err |
| + } |
| return s.migrateEntitlements() |
| } |
| @@ -222,6 +225,13 @@ func (s *Store) beginBillingCheckout(want billingIntent, now time.Time) (billing |
| if current.Subscription != "" && current.Status != "canceled" && current.Status != "incomplete_expired" { |
| return billingIntent{}, errBillingConflict |
| } |
| + // The caller chose the coupon from the house marker it read several |
| + // provider round trips ago. The marker is re-read here, in the |
| + // transaction that creates the intent: a house account never checks |
| + // out without its coupon, and a paying account never with one. |
| + if current.House != (want.Coupon != "") { |
| + return billingIntent{}, errBillingConflict |
| + } |
| if err := checkIntentMembershipTx(tx, want); err != nil { |
| return billingIntent{}, err |
| } |
diff --git a/internal/console/billingstore_test.go b/internal/console/billingstore_test.go| index 834639a08..305b65b31 100644 |
| --- a/internal/console/billingstore_test.go |
| +++ b/internal/console/billingstore_test.go |
| @@ -210,6 +210,11 @@ func TestBillingMonthFreezesCalculationAndSingleInvoice(t *testing.T) { |
| func TestBillingIntentPinsCustomerAndCoupon(t *testing.T) { |
| s, want, now := billingStoreFixture(t) |
| want.Customer, want.Coupon = "cus_original", "coupon_house" |
| + // A coupon is only ever attached for a house account, and the intent |
| + // transaction now checks the marker agrees with the coupon. |
| + if _, err := s.db.Exec(`INSERT INTO billing_accounts(account_id,house) VALUES(?,1) ON CONFLICT(account_id) DO UPDATE SET house=1`, want.AccountID); err != nil { |
| + t.Fatal(err) |
| + } |
| first, err := s.beginBillingCheckout(want, now) |
| if err != nil { |
| t.Fatal(err) |
diff --git a/internal/console/durabilityfold.go b/internal/console/durabilityfold.go| index 7767a7d51..86c40d695 100644 |
| --- a/internal/console/durabilityfold.go |
| +++ b/internal/console/durabilityfold.go |
| @@ -38,23 +38,35 @@ import ( |
| // |
| // active is every box another per-box source knows (metrics, edge), keyed |
| // by that source. A box that was alive on the day, by its own metrics or |
| -// edge expectation, and has no ledger directory at all is a box whose |
| -// ledger never opened: its replication ran unmetered and it must not be |
| -// left out of the question. It holds the day. |
| +// edge expectation, must be expected by the durability source for the same |
| +// span. A box with no ledger directory, or whose ledger registered later |
| +// than the box began serving, ran its replication unmetered for the |
| +// difference and must not be left out of the question. It holds the day. |
| func (s *Server) foldDurabilityDay(ctx context.Context, cands []sourceCandidate, active map[string][]sourceCandidate, d time.Time, date string, lifecycleOK bool, sweep *usageSweep, touch func(string) map[string]float64) bool { |
| // No candidate proves nothing: a deployment that has never published a |
| // ledger has no evidence that nothing replicated. |
| complete := lifecycleOK && len(cands) > 0 |
| - ledgered := map[string]bool{} |
| + // Every box another per-box source expected that day must be expected |
| + // by the durability source too, for at least the same span. A box with |
| + // no ledger directory, or one whose ledger was first registered after |
| + // the box was already serving, replicated unmetered for the difference. |
| + ledgers := map[string]sourceCandidate{} |
| for _, cand := range cands { |
| - ledgered[cand.Box] = true |
| + ledgers[cand.Box] = cand |
| } |
| for source, others := range active { |
| for _, other := range others { |
| - if ledgered[other.Box] { |
| + from, to, expected := other.window(source, d) |
| + if !expected { |
| continue |
| } |
| - if _, _, expected := other.window(source, d); expected { |
| + cand, ok := ledgers[other.Box] |
| + if !ok { |
| + complete = false |
| + continue |
| + } |
| + dFrom, dTo, dExpected := cand.window(usage.SourceDurability, d) |
| + if !dExpected || dFrom.After(from) || dTo.Before(to) { |
| complete = false |
| } |
| } |
diff --git a/internal/console/economicspass.go b/internal/console/economicspass.go| index 54983b465..1f6a8f68c 100644 |
| --- a/internal/console/economicspass.go |
| +++ b/internal/console/economicspass.go |
| @@ -1213,13 +1213,14 @@ const statementCloseDayGrace = usage.StatementCloseGraceDays |
| // failure is logged and does not skip the other, matching billPass and |
| // economicsPass's stage-degrades-independently posture. |
| // |
| -// Closing has no catch-up: a month is only ever closed by a tick landing in |
| -// the window between day 4 and the end of the FOLLOWING month, because the |
| -// previous-month period is the only one that ever closes. A console down for |
| -// a whole calendar month leaves the month before it permanently |
| -// Closed: false — still correct, still recomputed, but silently un-billable, |
| -// since billing reads only closed statements. Nothing notices; someone has |
| -// to. |
| +// Without a store-transfer report configured, closing has no catch-up: a |
| +// month is only ever closed by a tick landing in the window between day 4 |
| +// and the end of the FOLLOWING month, because the previous-month period is |
| +// the only one that ever closes. A console down for a whole calendar month |
| +// leaves the month before it permanently Closed: false — still correct, |
| +// still recomputed, but silently un-billable, since billing reads only |
| +// closed statements. With a report configured, the catch-up at the end of |
| +// this function revisits every earlier month still open. |
| func (s *Server) statementsPass(ctx context.Context, now time.Time) error { |
| reg, err := s.openPassRegistry() |
| if err != nil { |
| @@ -1255,12 +1256,24 @@ func (s *Server) statementsPass(ctx context.Context, now time.Time) error { |
| // close. Only a deployment with a report configured can be in that |
| // state, so only it pays for the extra reads. |
| if s.cfg.StoreEgressReportPrefix != "" { |
| + accounts, err := s.store.Accounts() |
| + if err != nil { |
| + log.Printf("console: statements pass: catch-up: %v", err) |
| + return nil |
| + } |
| for back := 2; back <= statementCatchUpMonths; back++ { |
| start := firstOfThisMonth.AddDate(0, -back, 0) |
| month := usage.MonthOf(start) |
| - if !s.statementStillOpen(ctx, month) { |
| + exists, open := s.statementMonthState(ctx, month, accounts) |
| + if !exists { |
| + break // before this deployment metered anything |
| + } |
| + if !open { |
| continue |
| } |
| + if back > statementCatchUpWarnMonths { |
| + log.Printf("console: statements pass: %s has been held open for %d months and cannot be billed; someone must repair what holds it", month, back) |
| + } |
| if err := s.statementsForMonth(ctx, month, start, start.AddDate(0, 1, -1), nowUTC, true, routes); err != nil { |
| log.Printf("console: statements pass: %s (catch-up): %v", month, err) |
| } |
| @@ -1269,25 +1282,50 @@ func (s *Server) statementsPass(ctx context.Context, now time.Time) error { |
| return nil |
| } |
| -// statementCatchUpMonths bounds how far back statementsPass looks for a |
| -// month held open. Six months is far longer than AWS takes to invoice and |
| -// long enough for an ownership hole to be noticed and repaired. |
| -const statementCatchUpMonths = 6 |
| - |
| -// statementStillOpen reports whether the platform account's statement for |
| -// month exists and is not closed. Every pass writes that account's |
| -// statement with the rest, so it stands for the month. A month with no |
| -// statement at all predates the deployment, or metering, and is left alone. |
| -func (s *Server) statementStillOpen(ctx context.Context, month string) bool { |
| - raw, _, err := s.cfg.Releases.Get(ctx, usage.StatementKey(release.OpsAccount, month)) |
| - if err != nil { |
| - return false |
| +// statementCatchUpMonths bounds how far back statementsPass walks looking |
| +// for a month still open. The walk also stops at the first month with no |
| +// platform statement, which is where metering began. Three years is far |
| +// beyond any invoice delay; a month held longer than |
| +// statementCatchUpWarnMonths is logged on every pass. |
| +const ( |
| + statementCatchUpMonths = 36 |
| + statementCatchUpWarnMonths = 6 |
| +) |
| + |
| +// statementMonthState reports whether the month has statements at all |
| +// (the platform account's is written by every pass, so it stands for |
| +// "this month was metered"), and whether any account's statement for it is |
| +// still open. Every account is checked, not only the platform's: a closing |
| +// pass that wrote the platform statement and failed on one account's |
| +// leaves that account's month open, and it must be retried. |
| +func (s *Server) statementMonthState(ctx context.Context, month string, accounts []Account) (exists, open bool) { |
| + read := func(acct string) (found, closed bool) { |
| + raw, _, err := s.cfg.Releases.Get(ctx, usage.StatementKey(acct, month)) |
| + if err != nil { |
| + return false, false |
| + } |
| + var st usage.Statement |
| + if json.Unmarshal(raw, &st) != nil { |
| + return false, false |
| + } |
| + return true, st.Closed |
| } |
| - var st usage.Statement |
| - if json.Unmarshal(raw, &st) != nil { |
| - return false |
| + found, closed := read(release.OpsAccount) |
| + if !found { |
| + return false, false |
| + } |
| + if !closed { |
| + return true, true |
| + } |
| + for _, a := range accounts { |
| + if a.Sqid == release.OpsAccount { |
| + continue |
| + } |
| + if f, c := read(a.Sqid); f && !c { |
| + return true, true |
| + } |
| } |
| - return !st.Closed |
| + return true, false |
| } |
| // statementsForMonth folds one month's metrics, edge and storage days (plus |
diff --git a/internal/console/meterfolds_test.go b/internal/console/meterfolds_test.go| index de5b4b77d..d44c122c3 100644 |
| --- a/internal/console/meterfolds_test.go |
| +++ b/internal/console/meterfolds_test.go |
| @@ -457,3 +457,95 @@ func TestStoreEgressHeldMonthIsRevisitedAfterTheNextMonthEnds(t *testing.T) { |
| t.Fatalf("September after a November repair: closed=%v days=%d", a.Closed, a.SourceDays[usage.SourceStoreEgress]) |
| } |
| } |
| + |
| +// A box whose ledger registered later than the box began serving replicated |
| +// unmetered for the difference. Its ledger directory exists, and its own |
| +// durability window says "not expected yet", but another source saw it |
| +// alive: the day holds. |
| +func TestDurabilityFoldHoldsBeforeABoxRegisteredItsLedger(t *testing.T) { |
| + ctx := context.Background() |
| + day := time.Date(2026, 9, 17, 0, 0, 0, 0, time.UTC) |
| + run := func(durabilityFrom time.Time) int { |
| + fx := newConsoleFixture(t) |
| + // The box has served edge traffic since the first of the month. |
| + if _, err := usage.RegisterBoxSource(ctx, fx.rel, "box1", usage.SourceEdge, time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC)); err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := usage.RegisterBoxSource(ctx, fx.rel, "box1", usage.SourceDurability, durabilityFrom); err != nil { |
| + t.Fatal(err) |
| + } |
| + putStoreJSON(t, fx.rel, usage.EdgeDayKey("box1", "2026-09-17"), watchedEdgeDay("box1", "2026-09-17")) |
| + publishDurabilityDay(t, fx, "box1", true, tenantWrites(t)) |
| + if err := fx.srv.statementsPass(ctx, time.Date(2026, 9, 20, 6, 0, 0, 0, time.UTC)); err != nil { |
| + t.Fatal(err) |
| + } |
| + return septemberStatement(t, fx, foldAcctA).SourceDays[usage.SourceDurability] |
| + } |
| + if got := run(time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC)); got != 1 { |
| + t.Fatalf("control: ledger registered with the box: durability source days = %d, want 1", got) |
| + } |
| + // Registered at 10:00 on the day itself: the first ten hours of the |
| + // 17th were served with no ledger. |
| + if got := run(day.Add(10 * time.Hour)); got != 0 { |
| + t.Errorf("durability source days = %d, want 0: the ledger registered after the box began serving", got) |
| + } |
| +} |
| + |
| +// One account's statement failed to write in the pass that closed the |
| +// month. The platform statement is closed, that account's is not. The |
| +// catch-up must see it and retry, including long after six months. |
| +func TestStatementCatchUpRetriesAnAccountLeftOpen(t *testing.T) { |
| + fx := newConsoleFixture(t) |
| + ctx := context.Background() |
| + fx.srv.cfg.StoreEgressReportPrefix = foldReportPrefix |
| + openOwner(t, fx, foldBucketA, foldAcctA, "shop") |
| + putCURDelivery(t, fx, "exec-1", curLine("l1", "INV-1", "EUW1-DataTransfer-Out-Bytes", foldBucketA, "2026-09-03T10:00:00Z", "2", "0.09", "0.18")) |
| + if err := fx.srv.statementsPass(ctx, time.Date(2026, 10, 5, 6, 0, 0, 0, time.UTC)); err != nil { |
| + t.Fatal(err) |
| + } |
| + if !septemberStatement(t, fx, fixtureAcct).Closed || !septemberStatement(t, fx, foldAcctA).Closed { |
| + t.Fatal("fixture: September should have closed for both accounts") |
| + } |
| + // Put the tenant's statement back to open, as a failed write in the |
| + // closing pass would have left it, and make it a console account so |
| + // the catch-up knows to look for it. |
| + if err := fx.st.CreateAccount(testAccountID(), foldAcctA, "Tenant", "owner@example.test", time.Now()); err != nil { |
| + t.Fatal(err) |
| + } |
| + st := septemberStatement(t, fx, foldAcctA) |
| + st.Closed = false |
| + raw, _ := json.Marshal(st) |
| + _, etag, err := fx.rel.Get(ctx, usage.StatementKey(foldAcctA, "2026-09")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := fx.rel.Put(ctx, usage.StatementKey(foldAcctA, "2026-09"), raw, etag); err != nil { |
| + t.Fatal(err) |
| + } |
| + // Eight months on: far past the old six-month horizon, and every month |
| + // in between has a platform statement to walk back through. |
| + for m := 0; m < 8; m++ { |
| + at := time.Date(2026, time.Month(11+m), 5, 6, 0, 0, 0, time.UTC) |
| + if err := fx.srv.statementsPass(ctx, at); err != nil { |
| + t.Fatal(err) |
| + } |
| + if m == 0 { |
| + if !septemberStatement(t, fx, foldAcctA).Closed { |
| + t.Fatal("the account left open was not retried by the next catch-up") |
| + } |
| + // Open it again: the later passes must still find it. |
| + st := septemberStatement(t, fx, foldAcctA) |
| + st.Closed = false |
| + raw, _ := json.Marshal(st) |
| + _, etag, _ := fx.rel.Get(ctx, usage.StatementKey(foldAcctA, "2026-09")) |
| + if _, err := fx.rel.Put(ctx, usage.StatementKey(foldAcctA, "2026-09"), raw, etag); err != nil { |
| + t.Fatal(err) |
| + } |
| + // Hold it for now by withdrawing the delivery's owner proof is |
| + // not needed: later passes retry it on their own. |
| + } |
| + } |
| + if !septemberStatement(t, fx, foldAcctA).Closed { |
| + t.Fatal("September was forgotten once it fell out of the old six-month window") |
| + } |
| +} |
diff --git a/internal/console/templates/billing.html b/internal/console/templates/billing.html| index f3167979f..5a9f1a507 100644 |
| --- a/internal/console/templates/billing.html |
| +++ b/internal/console/templates/billing.html |
| @@ -61,7 +61,7 @@ |
| {{if eq .Plan "personal"}}<!-- billing.plan.personal.name -->Personal{{else if eq .Plan "community"}}<!-- billing.plan.community.name -->Community{{else if eq .Plan "business"}}<!-- billing.plan.business.name -->Business{{end}}</label></td> |
| <td>{{if .Team}}<strong>{{.PerMember}}</strong> <!-- billing.price.permember -->per member a year |
| <br><span class="small">{{.Price}} <!-- billing.price.peryear -->a year</span>{{else}}<strong>{{.Price}}</strong> <!-- billing.price.peryear -->a year{{end}}</td> |
| - <td class="small">{{if eq .Plan "personal"}}<!-- billing.plan.personal.blurb -->One member. Each month: 85 awake hours, 2 GB storage, 2 GB transfer.{{else if eq .Plan "community"}}<!-- billing.plan.community.blurb -->Per member, minimum 10. Each member adds, per month: 17 awake hours, 0.5 GB storage, 0.4 GB transfer.{{else if eq .Plan "business"}}<!-- billing.plan.business.blurb -->Per member, minimum 10. Each member adds, per month: 170 awake hours, 5 GB storage, 3 GB transfer, 30,000 replication writes. Priority support.{{end}}{{if .Unavailable}}<br><!-- billing.plan.personal.unavailable -->Personal is for one member. This account has {{$.Billing.Members}}.{{end}}</td> |
| + <td class="small">{{if eq .Plan "personal"}}<!-- billing.plan.personal.blurb -->One member. Each month: 85 awake hours, 2 GB storage, 2 GB transfer.{{else if eq .Plan "community"}}<!-- billing.plan.community.blurb -->Per member, minimum 10. Each member adds, per month: 17 awake hours, 0.5 GB storage, 0.4 GB transfer.{{else if eq .Plan "business"}}<!-- billing.plan.business.blurb -->Per member, minimum 10. Each member adds, per month: 170 awake hours, 5 GB storage, 3 GB transfer, 30,000 replication writes. Priority support.{{end}}{{if and .Unavailable (eq .Plan "personal")}}<br><!-- billing.plan.personal.unavailable -->Personal is for one member. This account has {{$.Billing.Members}}.{{end}}</td> |
| </tr> |
| {{end}} |
| </table> |
diff --git a/internal/objstore/aws.go b/internal/objstore/aws.go| index 931b713b5..91fa09894 100644 |
| --- a/internal/objstore/aws.go |
| +++ b/internal/objstore/aws.go |
| @@ -258,6 +258,22 @@ func isNoSuchBucket(err error) bool { |
| // ---- S3 ------------------------------------------------------------------- |
| +// BucketExists reports whether name exists, without creating it |
| +// (BucketProber). It lists one key: S3 answers NoSuchBucket for a name |
| +// nobody holds, and anything else, including a refusal because another |
| +// account holds the name, means the bucket is there. An error that is not |
| +// NoSuchBucket is returned as an error, and the caller claims nothing. |
| +func (p *awsProvisioner) BucketExists(ctx context.Context, name string) (bool, error) { |
| + _, err := p.s3.ListObjectsV2(ctx, &awss3.ListObjectsV2Input{Bucket: aws.String(name), MaxKeys: aws.Int32(1)}) |
| + if err == nil { |
| + return true, nil |
| + } |
| + if isNoSuchBucket(err) { |
| + return false, nil |
| + } |
| + return false, fmt.Errorf("objstore: probe bucket %q: %w", name, err) |
| +} |
| + |
| // EnsureBucket creates name if absent and answers the region it was |
| // provisioned in. AWS returns BucketAlreadyOwnedByYou (not us-east-1) or a |
| // bare success (us-east-1's historical quirk) when the caller's own prior |
diff --git a/internal/objstore/memprovisioner.go b/internal/objstore/memprovisioner.go| index 8b393d06c..20f4e99ef 100644 |
| --- a/internal/objstore/memprovisioner.go |
| +++ b/internal/objstore/memprovisioner.go |
| @@ -194,3 +194,15 @@ func (p *MemProvisioner) DeleteUser(ctx context.Context, user string) error { |
| delete(p.Users, user) |
| return nil |
| } |
| + |
| +// BucketExists reports whether the bucket has been created (BucketProber). |
| +func (m *MemProvisioner) BucketExists(_ context.Context, name string) (bool, error) { |
| + m.mu.Lock() |
| + defer m.mu.Unlock() |
| + if err := m.FailNext["BucketExists"]; err != nil { |
| + delete(m.FailNext, "BucketExists") |
| + return false, err |
| + } |
| + _, ok := m.Buckets[name] |
| + return ok, nil |
| +} |
diff --git a/internal/objstore/objstore.go b/internal/objstore/objstore.go| index 6299c2abf..a113d6b4c 100644 |
| --- a/internal/objstore/objstore.go |
| +++ b/internal/objstore/objstore.go |
| @@ -55,6 +55,12 @@ type GrantBlock struct { |
| Source string `json:"source"` |
| At time.Time `json:"at"` |
| By string `json:"by"` |
| + // Minted is a positive record, written with the block, that this grant |
| + // chose a bucket name of its own and did not adopt one. A block written |
| + // before the field existed, or by an adopting grant, reads false. The |
| + // ownership ledger may claim completeness only for a minted bucket |
| + // (ownership.go), so the absence of the field must never read as minted. |
| + Minted bool `json:"minted,omitempty"` |
| } |
| type Provisioned struct { |
diff --git a/internal/objstore/ownership.go b/internal/objstore/ownership.go| index 43c74e8b5..211fdf503 100644 |
| --- a/internal/objstore/ownership.go |
| +++ b/internal/objstore/ownership.go |
| @@ -39,8 +39,10 @@ import ( |
| // change was computed from (the internal/usage/lifecycle.go discipline). |
| // - Opened BEFORE credentials are delivered (Grant), from the grant |
| // block's own time, so no tenant can move a byte out of a bucket before |
| -// the ledger names them. Closed AFTER the bucket is deleted (Teardown), |
| -// so no byte can leave it after the ledger stops naming them. |
| +// the ledger names them. Closed AFTER the bucket is deleted and its |
| +// credentials are revoked (Teardown), so no byte can leave a bucket of |
| +// that name, even one created again, after the ledger stops naming |
| +// them. |
| // - OwnerAt answers only for a ledger created by the grant that minted |
| // the bucket (CompleteSince), and only when every interval touching |
| // the hour names the same account and app. Two owners in one hour, |
diff --git a/internal/objstore/ownership_test.go b/internal/objstore/ownership_test.go| index e7112f684..5354df31f 100644 |
| --- a/internal/objstore/ownership_test.go |
| +++ b/internal/objstore/ownership_test.go |
| @@ -496,3 +496,157 @@ func TestGrantStopsWhenLedgerWriteFails(t *testing.T) { |
| } |
| mustLedger(t, s, ledgerBucket) |
| } |
| + |
| +// A bucket that already exists under the name this grant would mint may be |
| +// reachable by someone the ledger never saw (an operator adopted it for |
| +// another app before ledgers existed). The first grant must not claim the |
| +// ledger is complete. |
| +func TestGrantClaimsNothingForABucketThatAlreadyExists(t *testing.T) { |
| + ctx := context.Background() |
| + s := release.NewDirStore(t.TempDir()) |
| + _, pub := testSealKeypair(t) |
| + p := NewMemProvisioner("eu-west-1") |
| + if _, err := p.EnsureBucket(ctx, ledgerBucket); err != nil { |
| + t.Fatal(err) |
| + } |
| + declare(t, s, "bab", "titogo", "TITO_S3") |
| + if _, err := Grant(ctx, s, p, testDeployment(), pub, "bab", "titogo", "op", GrantSourceOperator, "", t0); err != nil { |
| + t.Fatal(err) |
| + } |
| + b := mustLedger(t, s, ledgerBucket) |
| + if b.CompleteSince != nil { |
| + t.Fatalf("claimed completeness for a bucket that existed before the grant: %v", b.CompleteSince) |
| + } |
| + if acct, _, ok := b.OwnerAt(t0.Add(24 * time.Hour)); ok { |
| + t.Fatalf("resolved %q for a bucket with unknown earlier access", acct) |
| + } |
| +} |
| + |
| +// A probe that fails claims nothing either. |
| +func TestGrantClaimsNothingWhenTheProbeFails(t *testing.T) { |
| + ctx := context.Background() |
| + s := release.NewDirStore(t.TempDir()) |
| + _, pub := testSealKeypair(t) |
| + p := NewMemProvisioner("eu-west-1") |
| + p.FailNext = map[string]error{"BucketExists": errors.New("probe refused")} |
| + declare(t, s, "bab", "titogo", "TITO_S3") |
| + if _, err := Grant(ctx, s, p, testDeployment(), pub, "bab", "titogo", "op", GrantSourceOperator, "", t0); err != nil { |
| + t.Fatal(err) |
| + } |
| + if b := mustLedger(t, s, ledgerBucket); b.CompleteSince != nil { |
| + t.Fatalf("claimed completeness without a probe answer: %v", b.CompleteSince) |
| + } |
| +} |
| + |
| +// Whether a grant minted its bucket is read from the grant block, not from |
| +// the arguments of whichever call happens to be running. A block written by |
| +// an adopting grant that crashed is re-run without the adopt argument (a |
| +// re-run never re-chooses the bucket); it must still read as adopted. |
| +func TestReRunOfAnAdoptingGrantIsStillAdopted(t *testing.T) { |
| + ctx := context.Background() |
| + s := release.NewDirStore(t.TempDir()) |
| + _, pub := testSealKeypair(t) |
| + p := NewMemProvisioner("eu-west-1") |
| + declare(t, s, "bab", "titogo", "TITO_S3") |
| + // The adopting grant dies at its first provisioner call, after the |
| + // grant block is written. |
| + p.FailNext = map[string]error{"EnsureBucket": errors.New("crash"), "EnsureUser": errors.New("crash")} |
| + if _, err := Grant(ctx, s, p, testDeployment(), pub, "bab", "titogo", "op", GrantSourceOperator, ledgerBucket, t0); err == nil { |
| + t.Fatal("the injected crash did not stop the grant") |
| + } |
| + rec, _, err := Get(ctx, s, "bab", "titogo") |
| + if err != nil || rec.Grant == nil || rec.Grant.Minted { |
| + t.Fatalf("grant block after an adopting grant: %+v %v", rec.Grant, err) |
| + } |
| + if err := s.Delete(ctx, BucketOwnershipKey(ledgerBucket)); err != nil { |
| + t.Fatal(err) |
| + } |
| + p.FailNext = nil |
| + if _, err := Grant(ctx, s, p, testDeployment(), pub, "bab", "titogo", "op", GrantSourceOperator, "", t0.Add(time.Hour)); err != nil { |
| + t.Fatal(err) |
| + } |
| + if b := mustLedger(t, s, ledgerBucket); b.CompleteSince != nil { |
| + t.Fatalf("a re-run without the adopt argument claimed the adopted bucket was minted: %v", b.CompleteSince) |
| + } |
| +} |
| + |
| +// A key that survives a failed DeleteUser can reach a bucket of the same |
| +// name created again. The interval must stay open until the credentials are |
| +// gone, so anyone else granted the name shares its hours and nobody is |
| +// billed for them. |
| +func TestTeardownKeepsOwnershipOpenUntilCredentialsAreRevoked(t *testing.T) { |
| + ctx := context.Background() |
| + s := release.NewDirStore(t.TempDir()) |
| + _, pub := testSealKeypair(t) |
| + p := NewMemProvisioner("eu-west-1") |
| + declare(t, s, "bab", "titogo", "TITO_S3") |
| + if _, err := Grant(ctx, s, p, testDeployment(), pub, "bab", "titogo", "op", GrantSourceOperator, "", t0); err != nil { |
| + t.Fatal(err) |
| + } |
| + p.FailNext = map[string]error{"DeleteUser": errors.New("iam had a bad moment")} |
| + if err := Teardown(ctx, s, p, nil, "bab", "titogo"); err == nil { |
| + t.Fatal("the injected DeleteUser failure did not stop the teardown") |
| + } |
| + if _, exists := p.Buckets[ledgerBucket]; exists { |
| + t.Fatal("fixture: the bucket should be gone by the time DeleteUser fails") |
| + } |
| + b := mustLedger(t, s, ledgerBucket) |
| + if b.Intervals[0].Until != nil { |
| + t.Fatal("ownership was closed while the credentials still exist") |
| + } |
| + // Another app adopts the name while those credentials live: the hours |
| + // are shared, and resolve to nobody. |
| + declare(t, s, "bac", "other", "OTHER_S3") |
| + adoptedAt := t0.Add(48 * time.Hour) |
| + if _, err := Grant(ctx, s, p, testDeployment(), pub, "bac", "other", "op", GrantSourceOperator, ledgerBucket, adoptedAt); err != nil { |
| + t.Fatal(err) |
| + } |
| + if acct, _, ok := mustLedger(t, s, ledgerBucket).OwnerAt(adoptedAt.Add(24 * time.Hour)); ok { |
| + t.Fatalf("resolved %q while an earlier grantee's credentials survive", acct) |
| + } |
| + // The re-run revokes the credentials and only then closes the interval. |
| + p.FailNext = nil |
| + old := ownershipNow |
| + closedAt := t0.Add(96 * time.Hour) |
| + ownershipNow = func() time.Time { return closedAt } |
| + defer func() { ownershipNow = old }() |
| + if err := Teardown(ctx, s, p, nil, "bab", "titogo"); err != nil { |
| + t.Fatal(err) |
| + } |
| + b = mustLedger(t, s, ledgerBucket) |
| + if b.Intervals[0].Until == nil || !b.Intervals[0].Until.Equal(closedAt) { |
| + t.Fatalf("interval after the re-run: %+v", b.Intervals[0]) |
| + } |
| +} |
| + |
| +// The reviewed scenario in full: two apps were given the same adopted |
| +// bucket before ledgers existed. Re-granting only one of them must not make |
| +// it the owner of the other's transfer. |
| +func TestPartialBackfillOfASharedBucketResolvesNoOwner(t *testing.T) { |
| + ctx := context.Background() |
| + s := release.NewDirStore(t.TempDir()) |
| + _, pub := testSealKeypair(t) |
| + p := NewMemProvisioner("eu-west-1") |
| + const shared = "carlos-store-shared-archive" |
| + declare(t, s, "bab", "titogo", "TITO_S3") |
| + declare(t, s, "bac", "other", "OTHER_S3") |
| + for _, g := range []struct{ acct, app string }{{"bab", "titogo"}, {"bac", "other"}} { |
| + if _, err := Grant(ctx, s, p, testDeployment(), pub, g.acct, g.app, "op", GrantSourceOperator, shared, t0); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + // As it was before ledgers: both hold credentials, nothing is recorded. |
| + if err := s.Delete(ctx, BucketOwnershipKey(shared)); err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := Grant(ctx, s, p, testDeployment(), pub, "bab", "titogo", "op", GrantSourceOperator, "", t0.Add(30*24*time.Hour)); err != nil { |
| + t.Fatal(err) |
| + } |
| + b := mustLedger(t, s, shared) |
| + if len(b.Intervals) != 1 || b.Intervals[0].Account != "bab" { |
| + t.Fatalf("fixture: want only the re-granted app recorded, got %+v", b.Intervals) |
| + } |
| + if acct, _, ok := b.OwnerAt(t0.Add(31 * 24 * time.Hour)); ok { |
| + t.Fatalf("the re-granted app %q was made owner of a bucket its mate can still reach", acct) |
| + } |
| +} |
diff --git a/internal/objstore/provisioner.go b/internal/objstore/provisioner.go| index d96d02272..8f2e7293b 100644 |
| --- a/internal/objstore/provisioner.go |
| +++ b/internal/objstore/provisioner.go |
| @@ -31,3 +31,13 @@ type Provisioner interface { |
| // DeleteUser removes the user's keys and inline policy first. |
| DeleteUser(ctx context.Context, user string) error |
| } |
| + |
| +// BucketProber is implemented by a Provisioner that can say whether a bucket |
| +// already exists without creating it. The grant walk asks it before it |
| +// claims that a bucket's ownership ledger is complete: a bucket that was |
| +// already there may already have been reachable by someone the ledger has |
| +// never heard of. A Provisioner that cannot answer is treated as "may |
| +// exist", which claims nothing. |
| +type BucketProber interface { |
| + BucketExists(ctx context.Context, name string) (bool, error) |
| +} |
diff --git a/internal/objstore/walk.go b/internal/objstore/walk.go| index d606b216d..5342f83e7 100644 |
| --- a/internal/objstore/walk.go |
| +++ b/internal/objstore/walk.go |
| @@ -167,7 +167,7 @@ func Grant(ctx context.Context, s release.Store, p Provisioner, d Deployment, |
| return nil, fmt.Errorf("objstore: invalid bucket name %q", bucket) |
| } |
| if rec.Grant == nil { |
| - rec.Grant = &GrantBlock{Backend: d.Backend, Bucket: bucket, Source: source, At: now, By: actor} |
| + rec.Grant = &GrantBlock{Backend: d.Backend, Bucket: bucket, Source: source, At: now, By: actor, Minted: adoptBucket == ""} |
| if etag, err = Put(ctx, s, acct, app, rec, etag); err != nil { |
| return nil, err |
| } |
| @@ -180,11 +180,27 @@ func Grant(ctx context.Context, s release.Store, p Provisioner, d Deployment, |
| // minted below). Idempotent: an open interval for |
| // this app is left untouched. A failure stops the walk before anything |
| // is created; re-running it repairs. |
| - // The ledger may claim completeness only when this grant mints the |
| - // bucket and nothing has been delivered yet: an adopted bucket, or a |
| - // store that already held credentials before its ledger existed, may |
| - // have been reachable by someone the ledger never saw. |
| - minted := bucket == MintedBucket(d, acct, app) && adoptBucket == "" && rec.Provisioned == nil |
| + // The ledger may claim completeness only when all of these hold, each |
| + // read from what is recorded or observed, never from this call's |
| + // arguments: |
| + // |
| + // - the grant block says the grant minted its bucket (GrantBlock.Minted, |
| + // a positive record; an adopting grant or an older block reads false); |
| + // - nothing has been delivered yet (no Provisioned block); |
| + // - the bucket does not exist yet. A bucket that is already there, under |
| + // any owner, may already be reachable by someone the ledger never saw. |
| + // |
| + // A provisioner that cannot say whether the bucket exists, or a probe |
| + // that fails, claims nothing. A ledger that already exists is left as it |
| + // is whatever this computes. |
| + minted := false |
| + if rec.Grant.Minted && rec.Provisioned == nil { |
| + if prober, ok := p.(BucketProber); ok { |
| + if exists, perr := prober.BucketExists(ctx, bucket); perr == nil && !exists { |
| + minted = true |
| + } |
| + } |
| + } |
| if err := OpenBucketOwner(ctx, s, bucket, acct, app, grantOwnershipFrom(rec.Grant, now), minted); err != nil { |
| return nil, fmt.Errorf("objstore: record ownership of %s: %w", bucket, err) |
| } |
| @@ -548,17 +564,6 @@ func Teardown(ctx context.Context, s release.Store, p Provisioner, sc Scanner, a |
| if err := p.DeleteBucket(ctx, bucket); err != nil { |
| return fmt.Errorf("objstore: delete bucket %s: %w", bucket, err) |
| } |
| - // Close this app's ownership interval (ownership.go) only now that |
| - // the bucket is gone, so no byte can leave it after the ledger |
| - // stops naming the app. Before the record goes, so a failure here |
| - // leaves the record standing and a re-run closes it. A name Grant |
| - // would refuse can never have had an interval opened, so there is |
| - // nothing to close for one. |
| - if ValidBucketName(bucket) { |
| - if err := CloseBucketOwner(ctx, s, bucket, acct, app, ownershipNow()); err != nil { |
| - return fmt.Errorf("objstore: close ownership of %s: %w", bucket, err) |
| - } |
| - } |
| } |
| // DeleteUser reaps the keys and the inline policy, so the credential dies |
| // with the estate it was scoped to — no orphan principal keeps a key that |
| @@ -566,6 +571,21 @@ func Teardown(ctx context.Context, s release.Store, p Provisioner, sc Scanner, a |
| if err := p.DeleteUser(ctx, user); err != nil { |
| return fmt.Errorf("objstore: delete user %s: %w", user, err) |
| } |
| + // Close this app's ownership interval (ownership.go) only now that the |
| + // bucket is gone AND its credentials are dead. Closing after the bucket |
| + // alone is not enough: a bucket name can be created again, and a key |
| + // that survived a failed DeleteUser would reach the new bucket while |
| + // the ledger named only its new owner. While the interval stays open, |
| + // anyone else granted the name shares its hours with this app, and a |
| + // shared hour resolves to nobody. Before the record goes, so a failure |
| + // here leaves the record standing and a re-run closes it. A name Grant |
| + // would refuse can never have had an interval opened, so there is |
| + // nothing to close for one. |
| + if bucket != "" && ValidBucketName(bucket) { |
| + if err := CloseBucketOwner(ctx, s, bucket, acct, app, ownershipNow()); err != nil { |
| + return fmt.Errorf("objstore: close ownership of %s: %w", bucket, err) |
| + } |
| + } |
| // The delivered set goes even though the credentials behind it are already |
| // dead: leaving them would hand the app's next render a bucket name and a |
| // key that authenticate against nothing, which is harder to debug than an |
diff --git a/internal/usage/billable_test.go b/internal/usage/billable_test.go| index 4972b6efd..69385afac 100644 |
| --- a/internal/usage/billable_test.go |
| +++ b/internal/usage/billable_test.go |
| @@ -107,3 +107,44 @@ func TestBucketOwnershipEvidenceIsPinnedByItsMonthProjection(t *testing.T) { |
| t.Fatalf("a ledger naming another bucket must be refused, got %v", err) |
| } |
| } |
| + |
| +// The completeness boundary decides which hours resolve an owner, so moving |
| +// it inside the month is changed evidence, and moving it before the month, |
| +// where it changes nothing the month relied on, is not. |
| +func TestBucketOwnershipEvidenceKeepsTheCompletenessBoundary(t *testing.T) { |
| + ctx := context.Background() |
| + raw := release.NewDirStore(t.TempDir()) |
| + key := BucketOwnershipPrefix + "carlos-store-acme-shop.json" |
| + from := time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC) |
| + put := func(since time.Time) { |
| + t.Helper() |
| + b, _ := json.Marshal(map[string]any{"v": 1, "bucket": "carlos-store-acme-shop", "complete_since": since, |
| + "intervals": []any{map[string]any{"account_sqid": "acme", "app": "shop", "from": from}}}) |
| + _, etag, err := raw.Get(ctx, key) |
| + if err != nil { |
| + etag = "" |
| + } |
| + if _, err := raw.Put(ctx, key, b, etag); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + put(from) |
| + es, err := NewMonthEvidenceStore(raw, "2026-09") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, _, err := es.Get(ctx, key); err != nil { |
| + t.Fatal(err) |
| + } |
| + ev, err := es.Snapshot() |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := ev.Verify(ctx, raw); err != nil { |
| + t.Fatal(err) |
| + } |
| + put(time.Date(2026, 9, 20, 0, 0, 0, 0, time.UTC)) |
| + if err := ev.Verify(ctx, raw); !errors.Is(err, ErrEvidenceChanged) { |
| + t.Fatalf("a completeness boundary moved inside the month must read as changed evidence, got %v", err) |
| + } |
| +} |
diff --git a/internal/usage/evidence_month.go b/internal/usage/evidence_month.go| index 687decdde..d77961706 100644 |
| --- a/internal/usage/evidence_month.go |
| +++ b/internal/usage/evidence_month.go |
| @@ -99,14 +99,24 @@ func projectBucketOwnershipMonth(key string, raw []byte, month string) ([]byte, |
| spans = append(spans, span{iv.Account, iv.App, lo, hi}) |
| } |
| } |
| - // Whether the ledger vouched for the month is part of what it proved. |
| - complete := rec.CompleteSince != nil && rec.CompleteSince.Before(end) |
| + // From when the ledger vouched for the month is part of what it |
| + // proved, to the instant: an hour before that boundary resolves no |
| + // owner. It is clipped to the month like the intervals, so a boundary |
| + // before the month reads as the month's start and one after it as none. |
| + var completeFrom *time.Time |
| + if rec.CompleteSince != nil && rec.CompleteSince.Before(end) { |
| + at := rec.CompleteSince.UTC() |
| + if at.Before(start) { |
| + at = start |
| + } |
| + completeFrom = &at |
| + } |
| return json.Marshal(struct { |
| - Bucket string `json:"bucket"` |
| - Month string `json:"month"` |
| - Complete bool `json:"complete"` |
| - Spans []span `json:"spans"` |
| - }{bucket, month, complete, spans}) |
| + Bucket string `json:"bucket"` |
| + Month string `json:"month"` |
| + CompleteFrom *time.Time `json:"complete_from"` |
| + Spans []span `json:"spans"` |
| + }{bucket, month, completeFrom, spans}) |
| } |
| func projectLifecycleMonth(key string, raw []byte, month string) ([]byte, error) { |