amadan

carlos / platform Public

Add inert resource budgets with consent-bound reservations and evidence holds

claude 8fe67311f84ce0ec1787a2b5a4ac9a63737a5b5f parent 2a40fed
11 files changed, +2772 −0
  • internal/console/resourcebudget_preference_test.go +108 −0
  • internal/resourcebudget/amount.go +154 −0
  • internal/resourcebudget/amount_test.go +22 −0
  • internal/resourcebudget/basis.go +198 −0
  • internal/resourcebudget/core.go +305 −0
  • internal/resourcebudget/core_test.go +726 −0
  • internal/resourcebudget/recovery_test.go +254 −0
  • internal/resourcebudget/reservation.go +271 −0
  • internal/resourcebudget/schema.go +40 −0
  • internal/resourcebudget/settlement.go +375 −0
  • internal/resourcebudget/verify.go +319 −0

CI passed — run details

diff --git a/internal/console/resourcebudget_preference_test.go b/internal/console/resourcebudget_preference_test.go
new file mode 100644
index 000000000..88edc123f
--- /dev/null
+++ b/internal/console/resourcebudget_preference_test.go
@@ -0,0 +1,108 @@
+package console
+
+import (
+ "database/sql"
+ rb "github.com/carlosframework/platform/internal/resourcebudget"
+ "path/filepath"
+ "testing"
+ "time"
+)
+
+func TestResourceBudgetPreferenceOrdering(t *testing.T) {
+ for _, order := range []string{"reserve-first", "off-first", "resave-before-claim"} {
+ t.Run(order, func(t *testing.T) {
+ path := filepath.Join(t.TempDir(), "console.db")
+ s := newOverageStore(t, path)
+ now := time.Date(2026, 10, 3, 12, 0, 0, 0, time.UTC)
+ seedOverageAccount(t, s, 42, overageOwner, now)
+ // Production remains NORMAL. Both test handles pin their own FULL connection.
+ if _, e := s.db.Exec(`PRAGMA synchronous=FULL`); e != nil {
+ t.Fatal(e)
+ }
+ other := newOverageStore(t, path)
+ if _, e := other.db.Exec(`PRAGMA synchronous=FULL`); e != nil {
+ t.Fatal(e)
+ }
+ if _, e := s.changeOveragePreference(overageChange(42, overageOwner, 0, true, 1), now); e != nil {
+ t.Fatal(e)
+ }
+ ev := overageEvidenceForTest()
+ identity := rb.Identity{Terms: ev.TermsVersion, Book: ev.PriceBook, Currency: ev.Currency}
+ loader := rb.GateFunc(func(tx *sql.Tx, a, m, subject string) (rb.Gate, error) {
+ p, e := loadOveragePreferenceTx(tx, 42)
+ return rb.Gate{Epoch: 1, Eligible: a == "42" && subject == "app", On: p.RequestedOn, Consent: p.Revision, Approved: true, Offer: ev.OfferDigest, StoredOffer: p.Evidence.OfferDigest, Identity: identity, Supported: map[rb.Resource]bool{rb.Awake: true}, Basis: map[rb.Resource]rb.BasisFact{rb.Awake: {Revision: 1, Digest: "synthetic", Identity: identity}}}, e
+ })
+ req := func(id string) rb.Request {
+ return rb.Request{ID: id, Account: "42", Month: "2026-10", Subject: "app", Epoch: 1}
+ }
+ tx, e := s.db.Begin()
+ if e != nil {
+ t.Fatal(e)
+ }
+ if e = rb.Migrate(tx); e != nil {
+ t.Fatal(e)
+ }
+ if _, e = rb.InstallBasis(tx, loader, rb.Basis{Request: req("basis"), Revision: 1, Identity: identity, Digest: "synthetic", Lines: []rb.Line{{Resource: rb.Awake, Amount: "0"}}}, now); e != nil {
+ t.Fatal(e)
+ }
+ if _, e = rb.InstallPaidLimit(tx, loader, rb.PaidLimit{Request: req("limit"), Resource: rb.Awake, Consent: 1, Offer: ev.OfferDigest, Identity: identity, Ceiling: "10", From: now, Until: now.Add(time.Hour)}, now); e != nil {
+ t.Fatal(e)
+ }
+ if e = tx.Commit(); e != nil {
+ t.Fatal(e)
+ }
+ r := rb.ReserveRequest{Request: req("reserve"), Reservation: "res", Lines: []rb.Line{{Resource: rb.Awake, Amount: "1"}}, BasisRevision: 1, Consent: 1, AllowPaid: true, From: now, Until: now.Add(time.Minute), ClaimBy: now.Add(time.Second)}
+ if order == "off-first" {
+ if _, e = s.changeOveragePreference(overageChange(42, overageOwner, 1, false, 2), now); e != nil {
+ t.Fatal(e)
+ }
+ tx, _ = other.db.Begin()
+ defer tx.Rollback()
+ if _, e = rb.Reserve(tx, loader, r, now); e == nil {
+ t.Fatal("OFF admitted paid reserve")
+ }
+ return
+ }
+ tx, _ = s.db.Begin()
+ if _, e = rb.Reserve(tx, loader, r, now); e != nil {
+ t.Fatal(e)
+ }
+ if order == "reserve-first" {
+ started := make(chan struct{})
+ done := make(chan error, 1)
+ go func() {
+ close(started)
+ _, err := other.changeOveragePreference(overageChange(42, overageOwner, 1, false, 2), now)
+ done <- err
+ }()
+ <-started
+ if e = tx.Commit(); e != nil {
+ t.Fatal(e)
+ }
+ select {
+ case e = <-done:
+ if e != nil {
+ t.Fatal(e)
+ }
+ case <-time.After(6 * time.Second):
+ t.Fatal("OFF did not complete")
+ }
+ } else {
+ if e = tx.Commit(); e != nil {
+ t.Fatal(e)
+ }
+ if _, e = s.changeOveragePreference(overageChange(42, overageOwner, 1, true, 2), now); e != nil {
+ t.Fatal(e)
+ }
+ }
+ tx, _ = s.db.Begin()
+ defer tx.Rollback()
+ if _, e = rb.Claim(tx, loader, rb.ClaimRequest{Request: req("claim"), Reservation: "res", Producer: "producer", Attempt: "attempt", ProducerEpoch: 1}, now); e == nil {
+ t.Fatal("stale consent claim")
+ }
+ if e = rb.Verify(tx); e != nil {
+ t.Fatal(e)
+ }
+ })
+ }
+}
diff --git a/internal/resourcebudget/amount.go b/internal/resourcebudget/amount.go
new file mode 100644
index 000000000..fa3cc2dce
--- /dev/null
+++ b/internal/resourcebudget/amount.go
@@ -0,0 +1,154 @@
+package resourcebudget
+
+import (
+ "errors"
+ "math/big"
+)
+
+var ErrInvalid = errors.New("resourcebudget: invalid")
+var ErrDenied = errors.New("resourcebudget: denied")
+var ErrConflict = errors.New("resourcebudget: conflict")
+var ErrCorrupt = errors.New("resourcebudget: corrupt")
+
+type Resource string
+
+const (
+ Awake Resource = "awake_ns"
+ Transfer Resource = "transfer_bytes"
+ Writes Resource = "replication_writes"
+)
+
+func (r Resource) valid() bool { return r == Awake || r == Transfer || r == Writes }
+
+type Amount string
+
+const Zero Amount = "0"
+
+var maximum = new(big.Int).Sub(new(big.Int).Lsh(big.NewInt(1), 127), big.NewInt(1))
+
+func number(a Amount) (*big.Int, error) {
+ if len(a) == 0 || len(a) > 39 || (len(a) > 1 && a[0] == '0') {
+ return nil, ErrInvalid
+ }
+ for _, c := range a {
+ if c < '0' || c > '9' {
+ return nil, ErrInvalid
+ }
+ }
+ n, ok := new(big.Int).SetString(string(a), 10)
+ if !ok || n.Cmp(maximum) > 0 {
+ return nil, ErrInvalid
+ }
+ return n, nil
+}
+func ParseAmount(s string) (Amount, error) { a := Amount(s); _, e := number(a); return a, e }
+func add(a, b Amount) (Amount, error) {
+ x, e := number(a)
+ if e != nil {
+ return "", e
+ }
+ y, e := number(b)
+ if e != nil {
+ return "", e
+ }
+ x.Add(x, y)
+ if x.Cmp(maximum) > 0 {
+ return "", ErrInvalid
+ }
+ return Amount(x.String()), nil
+}
+func sub(a, b Amount) (Amount, error) {
+ x, e := number(a)
+ if e != nil {
+ return "", e
+ }
+ y, e := number(b)
+ if e != nil {
+ return "", e
+ }
+ x.Sub(x, y)
+ if x.Sign() < 0 {
+ return "", ErrCorrupt
+ }
+ return Amount(x.String()), nil
+}
+func cmp(a, b Amount) int { x, _ := number(a); y, _ := number(b); return x.Cmp(y) }
+func min(a, b Amount) Amount {
+ if cmp(a, b) < 0 {
+ return a
+ }
+ return b
+}
+
+func validAmounts(values ...Amount) bool {
+ for _, v := range values {
+ if _, e := number(v); e != nil {
+ return false
+ }
+ }
+ return true
+}
+func balanced(cap, available, settled, outstanding Amount) bool {
+ sum, e := add(available, settled)
+ if e != nil {
+ return false
+ }
+ sum, e = add(sum, outstanding)
+ return e == nil && sum == cap
+}
+func validateStored(table, id string, v any) error {
+ switch p := v.(type) {
+ case *Pool:
+ if !p.Resource.valid() || p.Account == "" || p.Revision <= 0 || p.Digest == "" || poolKey(p.Account, p.Month, p.Resource) != id || !validAmounts(p.Capacity, p.Available, p.Settled, p.Outstanding, p.Excess) || !balanced(p.Capacity, p.Available, p.Settled, p.Outstanding) {
+ return ErrCorrupt
+ }
+ case *PaidLimit:
+ if validRequest(p.Request) != nil || !p.Resource.valid() || p.Consent <= 0 || p.Offer == "" || limitKey(p.Account, p.Month, p.Resource, p.Consent, p.Subject) != id || !window(p.Month, p.From, p.Until) || !validAmounts(p.Ceiling, p.Unused, p.Settled, p.Outstanding) || !balanced(p.Ceiling, p.Unused, p.Settled, p.Outstanding) {
+ return ErrCorrupt
+ }
+ case *Reservation:
+ if validRequest(p.Request) != nil || p.Reservation != id || validateLines(p.Lines) != nil || len(p.Allocations) != len(p.Lines) || !window(p.Month, p.From, p.Until) {
+ return ErrCorrupt
+ }
+ switch p.State {
+ case "reserved", "closed-unused":
+ if p.Attempt != "" || p.Producer != "" || p.ProducerEpoch != 0 {
+ return ErrCorrupt
+ }
+ case "claimed", "uncertain", "closed":
+ if p.Attempt == "" || p.Producer == "" || p.ProducerEpoch <= 0 {
+ return ErrCorrupt
+ }
+ default:
+ return ErrCorrupt
+ }
+ for i, a := range p.Allocations {
+ if a.Resource != p.Lines[i].Resource || a.Maximum != p.Lines[i].Amount || !validAmounts(a.Maximum, a.Included, a.Paid, a.SettledIncluded, a.SettledPaid, a.Excess, a.Total) {
+ return ErrCorrupt
+ }
+ sum, e := add(a.Included, a.Paid)
+ if e != nil || sum != a.Maximum {
+ return ErrCorrupt
+ }
+ b := min(a.Total, a.Maximum)
+ inc := min(b, a.Included)
+ paid, _ := sub(b, inc)
+ excess, _ := sub(a.Total, b)
+ if inc != a.SettledIncluded || paid != a.SettledPaid || excess != a.Excess {
+ return ErrCorrupt
+ }
+ if a.Paid != Zero && a.Limit != limitKey(p.Account, p.Month, a.Resource, p.Consent, p.Subject) {
+ return ErrCorrupt
+ }
+ }
+ case *Evidence:
+ if validEvidence(*p) != nil || p.ID != id {
+ return ErrCorrupt
+ }
+ case *FenceRequest:
+ if validRequest(p.Request) != nil || p.Producer == "" || p.ProducerEpoch <= 0 || key(p.Account, p.Producer) != id {
+ return ErrCorrupt
+ }
+ }
+ return nil
+}
diff --git a/internal/resourcebudget/amount_test.go b/internal/resourcebudget/amount_test.go
new file mode 100644
index 000000000..5f909610c
--- /dev/null
+++ b/internal/resourcebudget/amount_test.go
@@ -0,0 +1,22 @@
+package resourcebudget
+
+import "testing"
+
+func TestAmounts(t *testing.T) {
+ for _, s := range []string{"", "00", "-1", "+1", "1.0", "170141183460469231731687303715884105728"} {
+ if _, e := ParseAmount(s); e == nil {
+ t.Fatal(s)
+ }
+ }
+ for _, s := range []string{"0", "360000000000000000000", "170141183460469231731687303715884105727"} {
+ if _, e := ParseAmount(s); e != nil {
+ t.Fatal(s, e)
+ }
+ }
+ if _, e := add(Amount(maximum.String()), "1"); e == nil {
+ t.Fatal("overflow")
+ }
+ if validateLines([]Line{{Resource("storage_byte_ns"), "1"}}) == nil || validateLines([]Line{{Awake, "1"}, {Awake, "2"}}) == nil {
+ t.Fatal("resource validation")
+ }
+}
diff --git a/internal/resourcebudget/basis.go b/internal/resourcebudget/basis.go
new file mode 100644
index 000000000..a5b33bd1e
--- /dev/null
+++ b/internal/resourcebudget/basis.go
@@ -0,0 +1,198 @@
+package resourcebudget
+
+import (
+ "database/sql"
+ "time"
+)
+
+type Line struct {
+ Resource Resource
+ Amount Amount
+}
+type Basis struct {
+ Request
+ Revision int64
+ Identity Identity
+ Digest string
+ Lines []Line
+}
+type Pool struct {
+ Account, Month string
+ Resource Resource
+ Capacity, Available, Settled, Outstanding, Excess Amount
+ Revision int64
+ Identity Identity
+ Digest string
+}
+type PaidLimit struct {
+ Request
+ Resource Resource
+ Consent int64
+ Offer string
+ Identity Identity
+ Ceiling, Unused, Settled, Outstanding Amount
+ From, Until time.Time
+}
+
+func poolKey(a, m string, r Resource) string { return key(a, m, r) }
+func limitKey(a, m string, r Resource, c int64, s string) string { return key(a, m, r, c, s) }
+func validateLines(ls []Line) error {
+ if len(ls) == 0 || len(ls) > 3 {
+ return ErrInvalid
+ }
+ var previous Resource
+ for _, l := range ls {
+ if !l.Resource.valid() || l.Resource <= previous {
+ return ErrInvalid
+ }
+ if _, e := number(l.Amount); e != nil {
+ return e
+ }
+ previous = l.Resource
+ }
+ return nil
+}
+func InstallBasis(tx *sql.Tx, l GateLoader, b Basis, now time.Time) (Outcome, error) {
+ return basis(tx, l, b, now, false)
+}
+func IncreaseBasis(tx *sql.Tx, l GateLoader, b Basis, now time.Time) (Outcome, error) {
+ return basis(tx, l, b, now, true)
+}
+func basis(tx *sql.Tx, l GateLoader, b Basis, now time.Time, increase bool) (Outcome, error) {
+ if validateLines(b.Lines) != nil || b.Revision <= 0 || b.Digest == "" {
+ return Outcome{}, ErrInvalid
+ }
+ kind := "InstallBasis"
+ if increase {
+ kind = "IncreaseBasis"
+ }
+ o, re, e := begin(tx, l, b.Request, kind, b, now)
+ if e != nil {
+ return Outcome{}, e
+ }
+ if re != nil {
+ return *re, nil
+ }
+ pools := make([]Pool, len(b.Lines))
+ conflict := false
+ rows, e := tx.Query(`SELECT body FROM rb_pools`)
+ if e != nil {
+ return Outcome{}, e
+ }
+ count := 0
+ affected := map[Resource]bool{}
+ for rows.Next() {
+ var raw string
+ if e = rows.Scan(&raw); e != nil {
+ rows.Close()
+ return Outcome{}, e
+ }
+ var p Pool
+ if e = decode(raw, &p); e != nil {
+ rows.Close()
+ return Outcome{}, e
+ }
+ if p.Account == b.Account && p.Month == b.Month {
+ count++
+ affected[p.Resource] = true
+ }
+ }
+ e = rows.Err()
+ rows.Close()
+ if e != nil {
+ return Outcome{}, e
+ }
+ if (increase && count != len(b.Lines)) || (!increase && count != 0) {
+ conflict = true
+ }
+ for i, line := range b.Lines {
+ var p Pool
+ e = get(tx, "pools", poolKey(b.Account, b.Month, line.Resource), &p)
+ if e == nil {
+ if !increase || p.Identity != b.Identity || b.Revision <= p.Revision || cmp(line.Amount, p.Capacity) < 0 {
+ conflict = true
+ }
+ } else if !missing(e) {
+ return Outcome{}, e
+ } else if increase {
+ conflict = true
+ } else {
+ p = Pool{Account: b.Account, Month: b.Month, Resource: line.Resource, Capacity: Zero, Available: Zero, Settled: Zero, Outstanding: Zero, Excess: Zero}
+ }
+ pools[i] = p
+ }
+ if conflict {
+ for _, line := range b.Lines {
+ affected[line.Resource] = true
+ }
+ for resource := range affected {
+ if e = o.hold(b.Month, resource, "basis_conflict"); e != nil {
+ return Outcome{}, e
+ }
+ }
+ if e = o.put("basis", key("conflict", b.ID), b, true); e != nil {
+ return Outcome{}, e
+ }
+ return o.finish("held:basis_conflict")
+ }
+ for i, p := range pools {
+ delta, e := sub(b.Lines[i].Amount, p.Capacity)
+ if e != nil {
+ return Outcome{}, e
+ }
+ p.Available, e = add(p.Available, delta)
+ if e != nil {
+ return Outcome{}, e
+ }
+ p.Capacity = b.Lines[i].Amount
+ p.Revision = b.Revision
+ p.Identity = b.Identity
+ p.Digest = b.Digest
+ if e = o.put("pools", poolKey(b.Account, b.Month, p.Resource), p, false); e != nil {
+ return Outcome{}, e
+ }
+ }
+ if e = o.put("basis", key("installed", b.ID), b, true); e != nil {
+ return Outcome{}, e
+ }
+ return o.finish("installed")
+}
+func InstallPaidLimit(tx *sql.Tx, l GateLoader, p PaidLimit, now time.Time) (Outcome, error) {
+ if !p.Resource.valid() || p.Consent <= 0 || p.Offer == "" || !window(p.Month, p.From, p.Until) || p.Unused != "" || p.Settled != "" || p.Outstanding != "" {
+ return Outcome{}, ErrInvalid
+ }
+ if _, e := number(p.Ceiling); e != nil {
+ return Outcome{}, e
+ }
+ o, re, e := begin(tx, l, p.Request, "InstallPaidLimit", p, now)
+ if e != nil {
+ return Outcome{}, e
+ }
+ if re != nil {
+ return *re, nil
+ }
+ id := limitKey(p.Account, p.Month, p.Resource, p.Consent, p.Subject)
+ var old PaidLimit
+ e = get(tx, "limits", id, &old)
+ if e == nil {
+ return o.anomaly(p.Resource, "limit_conflict")
+ }
+ if !missing(e) {
+ return Outcome{}, e
+ }
+ if !paidGate(o.gate, p.Consent, p.Offer, p.Identity) {
+ return Outcome{}, ErrDenied
+ }
+ p.Unused, p.Settled, p.Outstanding = p.Ceiling, Zero, Zero
+ if e = o.put("limits", id, p, false); e != nil {
+ return Outcome{}, e
+ }
+ return o.finish("installed")
+}
+func paidGate(g Gate, c int64, offer string, i Identity) bool {
+ return g.Eligible && g.On && g.Approved && g.Consent == c && g.Offer != "" && g.Offer == offer && g.StoredOffer == offer && g.Identity == i
+}
+func window(m string, a, b time.Time) bool {
+ s, e, err := monthBounds(m)
+ return err == nil && !a.Before(s) && !b.After(e) && a.Before(b)
+}
diff --git a/internal/resourcebudget/core.go b/internal/resourcebudget/core.go
new file mode 100644
index 000000000..75852b137
--- /dev/null
+++ b/internal/resourcebudget/core.go
@@ -0,0 +1,305 @@
+// Package resourcebudget provides inert, caller-transaction resource accounting.
+// Every returned outcome is provisional until the caller commits. No outcome is
+// executable permission; production consumers and a durable-commit wrapper are absent.
+// Any error requires the caller to roll back the transaction, including earlier
+// successful operations in that transaction. Refused IDs have no receipt and may
+// succeed on a later attempt. A future wrapper must enforce rollback on error.
+package resourcebudget
+
+import (
+ "crypto/sha256"
+ "database/sql"
+ "encoding/hex"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "reflect"
+ "strconv"
+ "strings"
+ "time"
+ "unicode/utf8"
+)
+
+type Identity struct {
+ Terms, Book, Currency string
+ House bool
+}
+type BasisFact struct {
+ Revision int64
+ Digest string
+ Identity Identity
+}
+type Gate struct {
+ Epoch int64
+ Eligible, On, Approved bool
+ Consent int64
+ Offer, StoredOffer string
+ Identity Identity
+ Supported map[Resource]bool
+ Basis map[Resource]BasisFact
+}
+type GateLoader interface {
+ Load(*sql.Tx, string, string, string) (Gate, error)
+}
+type GateFunc func(*sql.Tx, string, string, string) (Gate, error)
+
+func (f GateFunc) Load(t *sql.Tx, a, m, s string) (Gate, error) { return f(t, a, m, s) }
+
+type Request struct {
+ ID, Account, Month, Subject string
+ Epoch int64
+}
+type Receipt struct {
+ ID, Kind, Account, Subject, Outcome string
+ Epoch int64
+}
+type Outcome struct {
+ Receipt Receipt
+ Replayed bool
+}
+type ClaimResult interface{ claimResult() }
+type NewClaimProvisional struct{ Receipt Receipt }
+
+func (NewClaimProvisional) claimResult() {}
+
+type ReplayedClaim struct{ Receipt Receipt }
+
+func (ReplayedClaim) claimResult() {}
+
+type operation struct {
+ tx *sql.Tx
+ r Request
+ kind, digest string
+ gate Gate
+ payload any
+}
+
+func key(parts ...any) string {
+ var out strings.Builder
+ for _, part := range parts {
+ var s string
+ switch v := part.(type) {
+ case string:
+ s = v
+ case Resource:
+ s = string(v)
+ case int64:
+ s = strconv.FormatInt(v, 10)
+ case int:
+ s = strconv.Itoa(v)
+ default:
+ panic(ErrInvalid)
+ }
+ out.WriteString(strconv.Itoa(len(s)))
+ out.WriteByte(':')
+ out.WriteString(s)
+ }
+ return out.String()
+}
+func encode(v any) ([]byte, error) {
+ var check func(reflect.Value) bool
+ check = func(v reflect.Value) bool {
+ if !v.IsValid() {
+ return true
+ }
+ switch v.Kind() {
+ case reflect.String:
+ return utf8.ValidString(v.String())
+ case reflect.Pointer, reflect.Interface:
+ if !v.IsNil() {
+ return check(v.Elem())
+ }
+ case reflect.Struct:
+ for i := 0; i < v.NumField(); i++ {
+ if v.Type().Field(i).PkgPath == "" && !check(v.Field(i)) {
+ return false
+ }
+ }
+ case reflect.Slice, reflect.Array:
+ for i := 0; i < v.Len(); i++ {
+ if !check(v.Index(i)) {
+ return false
+ }
+ }
+ case reflect.Map:
+ for _, k := range v.MapKeys() {
+ if !check(k) || !check(v.MapIndex(k)) {
+ return false
+ }
+ }
+ }
+ return true
+ }
+ if !check(reflect.ValueOf(v)) {
+ return nil, ErrInvalid
+ }
+ return json.Marshal(v)
+}
+func digest(v any) (string, error) {
+ b, e := encode(v)
+ if e != nil {
+ return "", e
+ }
+ h := sha256.Sum256(b)
+ return hex.EncodeToString(h[:]), nil
+}
+func monthBounds(m string) (time.Time, time.Time, error) {
+ s, e := time.Parse("2006-01", m)
+ if e != nil || s.Format("2006-01") != m {
+ return time.Time{}, time.Time{}, ErrInvalid
+ }
+ return s, s.AddDate(0, 1, 0), nil
+}
+func validRequest(r Request) error {
+ if r.ID == "" || r.Account == "" || r.Subject == "" || r.Epoch <= 0 {
+ return ErrInvalid
+ }
+ _, _, e := monthBounds(r.Month)
+ return e
+}
+func begin(tx *sql.Tx, l GateLoader, r Request, kind string, payload any, now time.Time) (*operation, *Outcome, error) {
+ if tx == nil || l == nil || validRequest(r) != nil || now.IsZero() {
+ return nil, nil, ErrInvalid
+ }
+ requestDigest, err := digest(struct {
+ Kind string
+ Payload any
+ }{kind, payload})
+ if err != nil {
+ return nil, nil, err
+ }
+ if !time.Unix(0, now.UnixNano()).Equal(now) {
+ return nil, nil, ErrInvalid
+ }
+ var sync int
+ if e := tx.QueryRow(`PRAGMA synchronous`).Scan(&sync); e != nil {
+ return nil, nil, e
+ }
+ if sync < 2 {
+ return nil, nil, ErrDenied
+ }
+ if _, e := tx.Exec(`INSERT INTO rb_authority VALUES(?,1,?,0) ON CONFLICT(account) DO UPDATE SET serial=serial`, r.Account, r.Epoch); e != nil {
+ return nil, nil, e
+ }
+ op := &operation{tx: tx, r: r, kind: kind, payload: payload, digest: requestDigest}
+ var d, b string
+ e := tx.QueryRow(`SELECT digest,body FROM rb_receipts WHERE id=?`, r.ID).Scan(&d, &b)
+ if e == nil {
+ if d != op.digest {
+ return nil, nil, ErrConflict
+ }
+ var receipt Receipt
+ if json.Unmarshal([]byte(b), &receipt) != nil {
+ return nil, nil, ErrCorrupt
+ }
+ return nil, &Outcome{receipt, true}, nil
+ }
+ if !errors.Is(e, sql.ErrNoRows) {
+ return nil, nil, e
+ }
+ var epoch, latest int64
+ if e = tx.QueryRow(`SELECT epoch,latest FROM rb_authority WHERE account=?`, r.Account).Scan(&epoch, &latest); e != nil {
+ return nil, nil, e
+ }
+ if epoch != r.Epoch || now.UnixNano() < latest {
+ return nil, nil, ErrDenied
+ }
+ op.gate, e = l.Load(tx, r.Account, r.Month, r.Subject)
+ if e != nil {
+ return nil, nil, e
+ }
+ if _, e = encode(op.gate); e != nil {
+ return nil, nil, e
+ }
+ if op.gate.Epoch != r.Epoch {
+ return nil, nil, ErrDenied
+ }
+ if _, e = tx.Exec(`UPDATE rb_authority SET latest=? WHERE account=?`, now.UnixNano(), r.Account); e != nil {
+ return nil, nil, e
+ }
+ return op, nil, nil
+}
+func (o *operation) finish(out string) (Outcome, error) {
+ r := Receipt{o.r.ID, o.kind, o.r.Account, o.r.Subject, out, o.r.Epoch}
+ b, e := encode(r)
+ if e != nil {
+ return Outcome{}, e
+ }
+ _, e = o.tx.Exec(`INSERT INTO rb_receipts VALUES(?,?,?)`, o.r.ID, o.digest, string(b))
+ return Outcome{Receipt: r}, e
+}
+func get(tx *sql.Tx, table, id string, v any) error {
+ var b string
+ if e := tx.QueryRow(`SELECT body FROM rb_`+table+` WHERE id=?`, id).Scan(&b); e != nil {
+ return e
+ }
+ if json.Unmarshal([]byte(b), v) != nil {
+ return ErrCorrupt
+ }
+ if _, e := encode(v); e != nil {
+ return ErrCorrupt
+ }
+ return validateStored(table, id, v)
+}
+func (o *operation) put(table, id string, v any, immutable bool) error {
+ b, e := encode(v)
+ if e != nil {
+ return e
+ }
+ q := `INSERT INTO rb_` + table + ` VALUES(?,?)`
+ if !immutable {
+ q += ` ON CONFLICT(id) DO UPDATE SET body=excluded.body`
+ }
+ if _, e = o.tx.Exec(q, id, string(b)); e != nil {
+ return e
+ }
+ _, e = o.tx.Exec(`INSERT INTO rb_events(kind,id,body,epoch,operation_id) VALUES(?,?,?,?,?)`, table, id, string(b), o.r.Epoch, o.r.ID)
+ return e
+}
+func (o *operation) hold(month string, r Resource, reason string) error {
+ id := key(o.r.ID, month, r, reason)
+ _, e := o.tx.Exec(`INSERT INTO rb_holds VALUES(?,?,?,?,?,?)`, id, o.r.Account, month, r, reason, o.r.Epoch)
+ return e
+}
+func (o *operation) held(r Resource) (bool, error) {
+ var n int
+ e := o.tx.QueryRow(`SELECT count(*) FROM rb_holds WHERE account=? AND month=? AND resource=?`, o.r.Account, o.r.Month, r).Scan(&n)
+ return n > 0, e
+}
+func (o *operation) admit(r Resource) error {
+ h, e := o.held(r)
+ if e != nil {
+ return e
+ }
+ if h || !o.gate.Eligible || !o.gate.Supported[r] {
+ return ErrDenied
+ }
+ return nil
+}
+func (o *operation) anomaly(r Resource, reason string, months ...string) (Outcome, error) {
+ if len(months) == 0 {
+ months = []string{o.r.Month}
+ }
+ seen := map[string]bool{}
+ for _, m := range months {
+ if !seen[m] {
+ seen[m] = true
+ if e := o.hold(m, r, reason); e != nil {
+ return Outcome{}, e
+ }
+ }
+ }
+ if e := o.put("basis", key("anomaly", o.r.ID), struct {
+ Reason string
+ Resource Resource
+ Months []string
+ Payload any
+ }{reason, r, months, o.payload}, true); e != nil {
+ return Outcome{}, e
+ }
+ return o.finish("held:" + reason)
+}
+func missing(e error) bool { return errors.Is(e, sql.ErrNoRows) }
+func problem(format string, args ...any) error {
+ return fmt.Errorf("%w: "+format, append([]any{ErrCorrupt}, args...)...)
+}
diff --git a/internal/resourcebudget/core_test.go b/internal/resourcebudget/core_test.go
new file mode 100644
index 000000000..2587c5114
--- /dev/null
+++ b/internal/resourcebudget/core_test.go
@@ -0,0 +1,726 @@
+package resourcebudget
+
+import (
+ "database/sql"
+ "errors"
+ "fmt"
+ _ "modernc.org/sqlite"
+ "path/filepath"
+ "strings"
+ "testing"
+ "time"
+)
+
+type fixture struct {
+ t *testing.T
+ db *sql.DB
+ dsn string
+ gate Gate
+ now time.Time
+ n int
+}
+
+func fresh(t *testing.T) *fixture {
+ t.Helper()
+ dir := t.TempDir()
+ dsn := filepath.Join(dir, "budget.db") + "?_pragma=synchronous(2)&_pragma=busy_timeout(5000)&_pragma=journal_mode(WAL)"
+ db, e := sql.Open("sqlite", dsn)
+ if e != nil {
+ t.Fatal(e)
+ }
+ db.SetMaxOpenConns(4)
+ t.Cleanup(func() { db.Close() })
+ f := &fixture{t: t, db: db, dsn: dsn, now: time.Date(2026, 10, 3, 12, 0, 0, 0, time.UTC), gate: Gate{Epoch: 1, Eligible: true, On: true, Approved: true, Consent: 1, Offer: "approved", StoredOffer: "approved", Identity: Identity{Terms: "synthetic", Book: "approved-test", Currency: "eur"}, Supported: map[Resource]bool{Awake: true, Transfer: true, Writes: true}}}
+ f.gate.Basis = map[Resource]BasisFact{Awake: {Revision: 1, Digest: "basis", Identity: f.gate.Identity}}
+ tx, e := db.Begin()
+ if e != nil {
+ t.Fatal(e)
+ }
+ if e = Migrate(tx); e != nil {
+ t.Fatal(e)
+ }
+ if e = tx.Commit(); e != nil {
+ t.Fatal(e)
+ }
+ return f
+}
+func (f *fixture) Load(*sql.Tx, string, string, string) (Gate, error) { return f.gate, nil }
+func (f *fixture) req() Request {
+ f.n++
+ return Request{ID: fmt.Sprint("op", f.n), Account: "account", Month: "2026-10", Subject: "app", Epoch: 1}
+}
+func (f *fixture) call(commit bool, fn func(*sql.Tx) (Outcome, error)) (Outcome, error) {
+ f.t.Helper()
+ tx, e := f.db.Begin()
+ if e != nil {
+ f.t.Fatal(e)
+ }
+ defer tx.Rollback()
+ o, e := fn(tx)
+ if e != nil {
+ return o, e
+ }
+ if e = Verify(tx); e != nil {
+ f.t.Fatal("verify", e)
+ }
+ if commit {
+ e = tx.Commit()
+ }
+ return o, e
+}
+func (f *fixture) ok(fn func(*sql.Tx) (Outcome, error)) Outcome {
+ f.t.Helper()
+ o, e := f.call(true, fn)
+ if e != nil {
+ f.t.Fatal(e)
+ }
+ return o
+}
+func (f *fixture) basis(amount Amount) Basis {
+ b := Basis{Request: f.req(), Revision: 1, Identity: f.gate.Identity, Digest: "basis", Lines: []Line{{Awake, amount}}}
+ f.ok(func(tx *sql.Tx) (Outcome, error) { return InstallBasis(tx, f, b, f.now) })
+ return b
+}
+func (f *fixture) limit() PaidLimit {
+ p := PaidLimit{Request: f.req(), Resource: Awake, Consent: 1, Offer: "approved", Identity: f.gate.Identity, Ceiling: "20", From: f.now.Add(-time.Hour), Until: f.now.Add(time.Hour)}
+ f.ok(func(tx *sql.Tx) (Outcome, error) { return InstallPaidLimit(tx, f, p, f.now) })
+ return p
+}
+func (f *fixture) reserve(amount Amount) ReserveRequest {
+ r := ReserveRequest{Request: f.req(), Reservation: "res", Lines: []Line{{Awake, amount}}, BasisRevision: 1, Consent: 1, AllowPaid: true, From: f.now, Until: f.now.Add(time.Minute), ClaimBy: f.now.Add(time.Second)}
+ f.ok(func(tx *sql.Tx) (Outcome, error) { return Reserve(tx, f, r, f.now) })
+ return r
+}
+func (f *fixture) claim(r ReserveRequest) ClaimRequest {
+ q := ClaimRequest{Request: f.req(), Reservation: r.Reservation, Producer: "producer", Attempt: "attempt", ProducerEpoch: 1}
+ f.ok(func(tx *sql.Tx) (Outcome, error) {
+ v, e := Claim(tx, f, q, f.now)
+ if e != nil {
+ return Outcome{}, e
+ }
+ n, ok := v.(NewClaimProvisional)
+ if !ok {
+ f.t.Fatal("not new")
+ }
+ return Outcome{Receipt: n.Receipt}, nil
+ })
+ return q
+}
+
+type checks struct{ fail bool }
+
+func (c checks) Measurement(Evidence) error {
+ if c.fail {
+ return ErrDenied
+ }
+ return nil
+}
+func (c checks) Terminal(Terminal) error {
+ if c.fail {
+ return ErrDenied
+ }
+ return nil
+}
+func (f *fixture) evidence(total Amount) Evidence {
+ return Evidence{ID: "meter", Account: "account", Reservation: "res", Subject: "app", Attempt: "attempt", Producer: "producer", ProducerEpoch: 1, Resource: Awake, Total: total, From: f.now, Until: f.now.Add(time.Second)}
+}
+func (f *fixture) finish(e Evidence) FinishRequest {
+ return FinishRequest{Request: f.req(), Reservation: "res", Evidence: []Evidence{e}, Terminal: Terminal{ID: "terminal", Account: e.Account, Reservation: e.Reservation, Subject: e.Subject, Attempt: e.Attempt, Producer: e.Producer, ProducerEpoch: e.ProducerEpoch}}
+}
+func (f *fixture) pool() Pool {
+ f.t.Helper()
+ tx, e := f.db.Begin()
+ if e != nil {
+ f.t.Fatal(e)
+ }
+ defer tx.Rollback()
+ var p Pool
+ if e = get(tx, "pools", poolKey("account", "2026-10", Awake), &p); e != nil {
+ f.t.Fatal(e)
+ }
+ return p
+}
+func (f *fixture) held() int {
+ var n int
+ if e := f.db.QueryRow(`SELECT count(*) FROM rb_holds`).Scan(&n); e != nil {
+ f.t.Fatal(e)
+ }
+ return n
+}
+
+func TestReserveClaimFinishAndReplay(t *testing.T) {
+ f := fresh(t)
+ f.basis("10")
+ f.limit()
+ r := f.reserve("15")
+ q := f.claim(r)
+ tx, _ := f.db.Begin()
+ v, e := Claim(tx, f, q, f.now.Add(time.Hour))
+ if e != nil {
+ t.Fatal(e)
+ }
+ if _, ok := v.(ReplayedClaim); !ok {
+ t.Fatalf("%T", v)
+ }
+ tx.Rollback()
+ q.ID = "different"
+ tx, _ = f.db.Begin()
+ if _, e = Claim(tx, f, q, f.now); e == nil {
+ t.Fatal("distinct claim")
+ }
+ tx.Rollback()
+ evi := f.evidence("12")
+ fin := f.finish(evi)
+ f.ok(func(tx *sql.Tx) (Outcome, error) { return Finish(tx, f, checks{}, fin, f.now) })
+ p := f.pool()
+ if p.Settled != "10" || p.Outstanding != Zero {
+ t.Fatal(p)
+ }
+ var raw string
+ f.db.QueryRow(`SELECT body FROM rb_limits`).Scan(&raw)
+ var limit PaidLimit
+ decode(raw, &limit)
+ if limit.Settled != "2" || limit.Outstanding != Zero || limit.Unused != "18" {
+ t.Fatal(limit)
+ }
+ out := f.ok(func(tx *sql.Tx) (Outcome, error) { return Finish(tx, f, checks{}, fin, f.now.Add(time.Hour)) })
+ if !out.Replayed {
+ t.Fatal("not replayed")
+ }
+}
+func TestProvisionalRollbackAndRejectedID(t *testing.T) {
+ f := fresh(t)
+ b := Basis{Request: f.req(), Revision: 1, Identity: f.gate.Identity, Digest: "b", Lines: []Line{{Awake, "10"}}}
+ f.gate.Basis[Awake] = BasisFact{1, "b", f.gate.Identity}
+ if _, e := f.call(false, func(tx *sql.Tx) (Outcome, error) { return InstallBasis(tx, f, b, f.now) }); e != nil {
+ t.Fatal(e)
+ }
+ var n int
+ f.db.QueryRow(`SELECT count(*) FROM rb_receipts`).Scan(&n)
+ if n != 0 {
+ t.Fatal(n)
+ }
+ f.ok(func(tx *sql.Tx) (Outcome, error) { return InstallBasis(tx, f, b, f.now) })
+ r := ReserveRequest{Request: f.req(), Reservation: "res", Lines: []Line{{Awake, "5"}}, BasisRevision: 1, From: f.now, Until: f.now.Add(time.Minute), ClaimBy: f.now.Add(time.Second)}
+ f.gate.Eligible = false
+ if _, e := f.call(true, func(tx *sql.Tx) (Outcome, error) { return Reserve(tx, f, r, f.now) }); e == nil {
+ t.Fatal("denied")
+ }
+ f.gate.Eligible = true
+ f.ok(func(tx *sql.Tx) (Outcome, error) { return Reserve(tx, f, r, f.now) })
+ q := ClaimRequest{Request: f.req(), Reservation: "res", Producer: "p", Attempt: "a", ProducerEpoch: 1}
+ tx, _ := f.db.Begin()
+ if _, e := Claim(tx, f, q, f.now); e != nil {
+ t.Fatal(e)
+ }
+ tx.Rollback()
+ tx, _ = f.db.Begin()
+ var rr Reservation
+ get(tx, "reservations", "res", &rr)
+ tx.Rollback()
+ if rr.State != "reserved" {
+ t.Fatal(rr.State)
+ }
+}
+func TestBasisAndLimitConflicts(t *testing.T) {
+ for _, kind := range []string{"same", "shrink", "identity", "revision", "limit"} {
+ for _, commit := range []bool{false, true} {
+ t.Run(fmt.Sprint(kind, commit), func(t *testing.T) {
+ f := fresh(t)
+ b := f.basis("10")
+ p := f.limit()
+ b.Request = f.req()
+ b.Revision = 2
+ switch kind {
+ case "same":
+ b.Revision = 1
+ case "shrink":
+ b.Lines[0].Amount = "9"
+ case "identity":
+ b.Identity.House = true
+ case "revision":
+ b.Revision = 1
+ case "limit":
+ p.Request = f.req()
+ }
+ out, e := f.call(commit, func(tx *sql.Tx) (Outcome, error) {
+ if kind == "limit" {
+ return InstallPaidLimit(tx, f, p, f.now)
+ }
+ if kind == "same" {
+ return InstallBasis(tx, f, b, f.now)
+ }
+ return IncreaseBasis(tx, f, b, f.now)
+ })
+ if e != nil || !strings.HasPrefix(out.Receipt.Outcome, "held:") {
+ t.Fatal(out, e)
+ }
+ if (f.held() > 0) != commit || f.pool().Capacity != "10" {
+ t.Fatal("conflict persistence")
+ }
+ })
+ }
+ }
+}
+func TestPaidGateAndWindow(t *testing.T) {
+ for _, kind := range []string{"revision", "off", "offer", "draft", "subject", "window", "capability"} {
+ t.Run(kind, func(t *testing.T) {
+ f := fresh(t)
+ f.basis("0")
+ f.limit()
+ r := ReserveRequest{Request: f.req(), Reservation: "res", Lines: []Line{{Awake, "5"}}, BasisRevision: 1, Consent: 1, AllowPaid: true, From: f.now, Until: f.now.Add(time.Minute), ClaimBy: f.now.Add(time.Second)}
+ switch kind {
+ case "revision":
+ f.gate.Consent++
+ case "off":
+ f.gate.On = false
+ case "offer":
+ f.gate.StoredOffer = "other"
+ case "draft":
+ f.gate.Approved = false
+ case "subject":
+ r.Subject = "other"
+ case "window":
+ r.Until = f.now.Add(2 * time.Hour)
+ case "capability":
+ f.gate.Supported[Awake] = false
+ }
+ if _, e := f.call(true, func(tx *sql.Tx) (Outcome, error) { return Reserve(tx, f, r, f.now) }); e == nil {
+ t.Fatal("accepted")
+ }
+ })
+ }
+}
+func TestClaimConsentResaveAndExpiredCancellation(t *testing.T) {
+ f := fresh(t)
+ f.basis("0")
+ f.limit()
+ r := f.reserve("5")
+ f.gate.Consent++
+ q := ClaimRequest{Request: f.req(), Reservation: r.Reservation, Producer: "p", Attempt: "a", ProducerEpoch: 1}
+ tx, _ := f.db.Begin()
+ if _, e := Claim(tx, f, q, f.now); e == nil {
+ t.Fatal("old consent")
+ }
+ tx.Rollback()
+ f.gate.On = false
+ request := ReservationRequest{f.req(), r.Reservation}
+ f.ok(func(tx *sql.Tx) (Outcome, error) { return CancelUnclaimed(tx, f, request, f.now.Add(time.Hour)) })
+}
+func TestHoldsPermitCloseout(t *testing.T) {
+ f := fresh(t)
+ b := f.basis("10")
+ r := f.reserve("8")
+ f.claim(r)
+ b.Request = f.req()
+ f.ok(func(tx *sql.Tx) (Outcome, error) { return InstallBasis(tx, f, b, f.now) })
+ f.gate.On = false
+ f.gate.Eligible = false
+ req := ReservationRequest{f.req(), r.Reservation}
+ f.ok(func(tx *sql.Tx) (Outcome, error) { return MarkUncertain(tx, f, req, f.now.Add(time.Hour)) })
+ if f.pool().Outstanding != "8" {
+ t.Fatal("expiry refund")
+ }
+ e := f.evidence("3")
+ ob := ObserveRequest{f.req(), e}
+ f.ok(func(tx *sql.Tx) (Outcome, error) { return Observe(tx, f, checks{}, ob, f.now.Add(time.Hour)) })
+ fin := f.finish(e)
+ f.ok(func(tx *sql.Tx) (Outcome, error) { return Finish(tx, f, checks{}, fin, f.now.Add(time.Hour)) })
+ if f.pool().Available != "7" {
+ t.Fatal(f.pool())
+ }
+ fr := FenceRequest{f.req(), "producer", 1}
+ f.ok(func(tx *sql.Tx) (Outcome, error) { return FenceProducer(tx, f, fr, f.now.Add(time.Hour)) })
+}
+func TestEvidenceAnomaliesCommitRollback(t *testing.T) {
+ for _, kind := range []string{"regression", "conflict", "window", "crossmonth", "overshoot", "finish"} {
+ for _, commit := range []bool{false, true} {
+ t.Run(fmt.Sprint(kind, commit), func(t *testing.T) {
+ f := fresh(t)
+ f.basis("10")
+ r := f.reserve("8")
+ f.claim(r)
+ first := f.evidence("3")
+ ob := ObserveRequest{f.req(), first}
+ f.ok(func(tx *sql.Tx) (Outcome, error) { return Observe(tx, f, checks{}, ob, f.now) })
+ if kind == "finish" {
+ fin := f.finish(first)
+ f.ok(func(tx *sql.Tx) (Outcome, error) { return Finish(tx, f, checks{}, fin, f.now) })
+ }
+ ev := first
+ ev.ID = "next"
+ switch kind {
+ case "regression":
+ ev.Total = "2"
+ case "conflict":
+ ev.ID = first.ID
+ ev.Total = "4"
+ case "window":
+ ev.Until = f.now.Add(time.Hour)
+ case "crossmonth":
+ ev.Until = time.Date(2026, 11, 2, 0, 0, 0, 0, time.UTC)
+ case "overshoot":
+ ev.Total = "12"
+ case "finish":
+ ev.Total = "4"
+ }
+ ob = ObserveRequest{f.req(), ev}
+ out, e := f.call(commit, func(tx *sql.Tx) (Outcome, error) {
+ if kind == "finish" {
+ fin := FinishRequest{Request: ob.Request, Reservation: "res", Evidence: []Evidence{ev}, Terminal: Terminal{ID: "end", Account: ev.Account, Reservation: ev.Reservation, Subject: ev.Subject, Attempt: ev.Attempt, Producer: ev.Producer, ProducerEpoch: ev.ProducerEpoch}}
+ return Finish(tx, f, checks{}, fin, f.now)
+ }
+ return Observe(tx, f, checks{}, ob, f.now)
+ })
+ if e != nil || !strings.HasPrefix(out.Receipt.Outcome, "held:") {
+ t.Fatal(out, e)
+ }
+ if (f.held() > 0) != commit {
+ t.Fatal("hold persistence")
+ }
+ var receipts int
+ if err := f.db.QueryRow(`SELECT count(*) FROM rb_receipts WHERE id=?`, ob.ID).Scan(&receipts); err != nil {
+ t.Fatal(err)
+ }
+ wantReceipts := 0
+ if commit {
+ wantReceipts = 1
+ }
+ if receipts != wantReceipts {
+ t.Fatal("anomaly receipt persistence", receipts)
+ }
+
+ p := f.pool()
+ if kind == "overshoot" && commit {
+ if p.Excess != "4" || p.Settled != "8" {
+ t.Fatal(p)
+ }
+ } else if p.Settled != "3" {
+ t.Fatal(p)
+ }
+ if kind == "crossmonth" && commit {
+ b := Basis{Request: f.req(), Revision: 1, Identity: f.gate.Identity, Digest: "nov", Lines: []Line{{Awake, "10"}}}
+ b.Month = "2026-11"
+ f.ok(func(tx *sql.Tx) (Outcome, error) { return InstallBasis(tx, f, b, f.now) })
+ var n int
+ f.db.QueryRow(`SELECT count(*) FROM rb_holds WHERE month='2026-11'`).Scan(&n)
+ if n != 1 {
+ t.Fatal(n)
+ }
+ }
+ })
+ }
+ }
+}
+func TestEvidenceRefusalsRollback(t *testing.T) {
+ for _, kind := range []string{"malformed", "checker", "unknown", "account", "reservation", "subject", "clock", "database"} {
+ t.Run(kind, func(t *testing.T) {
+ f := fresh(t)
+ f.basis("10")
+ r := f.reserve("8")
+ f.claim(r)
+ ev := f.evidence("3")
+ c := checks{}
+ now := f.now
+ switch kind {
+ case "malformed":
+ ev.Total = "-1"
+ case "checker":
+ c.fail = true
+ case "unknown":
+ ev.Resource = "unknown"
+ case "account":
+ ev.Account = "other"
+ case "reservation":
+ ev.Reservation = "other"
+ case "subject":
+ ev.Subject = "other"
+ case "clock":
+ now = now.Add(-time.Second)
+ }
+ ob := ObserveRequest{f.req(), ev}
+ tx, _ := f.db.Begin()
+ if kind == "database" {
+ tx.Rollback()
+ }
+ _, e := Observe(tx, f, c, ob, now)
+ if e == nil {
+ t.Fatal("accepted")
+ }
+ tx.Rollback()
+ var n int
+ f.db.QueryRow(`SELECT count(*) FROM rb_receipts WHERE id=?`, ob.ID).Scan(&n)
+ if n != 0 || f.held() != 0 || f.pool().Settled != Zero {
+ t.Fatal("refusal changed accounting")
+ }
+ })
+ }
+}
+func TestReceiptConflict(t *testing.T) {
+ f := fresh(t)
+ b := f.basis("10")
+ b.Account = "other"
+ if _, e := f.call(true, func(tx *sql.Tx) (Outcome, error) { return InstallBasis(tx, f, b, f.now) }); !errors.Is(e, ErrConflict) {
+ t.Fatal(e)
+ }
+}
+
+func TestReplayRejectsInvalidUTF8AndTime(t *testing.T) {
+ f := fresh(t)
+ b := Basis{Request: f.req(), Revision: 1, Identity: f.gate.Identity, Digest: "\ufffd", Lines: []Line{{Awake, "10"}}}
+ f.ok(func(tx *sql.Tx) (Outcome, error) { return InstallBasis(tx, f, b, f.now) })
+ for _, bad := range []string{string([]byte{0xff}), string([]byte{0xfe})} {
+ badBasis := b
+ badBasis.Digest = bad
+ if _, e := f.call(true, func(tx *sql.Tx) (Outcome, error) { return InstallBasis(tx, f, badBasis, f.now) }); e == nil {
+ t.Fatal("invalid UTF-8 replay alias")
+ }
+ }
+ p := PaidLimit{Request: f.req(), Resource: Awake, Consent: 1, Offer: "approved", Identity: f.gate.Identity, Ceiling: "1", From: f.now, Until: time.Date(10000, 1, 1, 0, 0, 0, 0, time.UTC)}
+ if _, e := f.call(true, func(tx *sql.Tx) (Outcome, error) { return InstallPaidLimit(tx, f, p, f.now) }); e == nil {
+ t.Fatal("invalid time")
+ }
+ for _, v := range []any{Request{ID: string([]byte{0xff})}, Evidence{Subject: string([]byte{0xff})}, Identity{Terms: string([]byte{0xff})}, time.Date(10000, 1, 1, 0, 0, 0, 0, time.UTC)} {
+ if _, e := digest(v); e == nil {
+ t.Fatal("invalid digest input")
+ }
+ }
+}
+
+func TestClaimSubjectAndGlobalAttempt(t *testing.T) {
+ f := fresh(t)
+ f.basis("10")
+ r := f.reserve("1")
+ q := ClaimRequest{Request: f.req(), Reservation: r.Reservation, Producer: "producer", Attempt: "attempt", ProducerEpoch: 1}
+ q.Subject = "wrong"
+ tx, _ := f.db.Begin()
+ if _, e := Claim(tx, f, q, f.now); e == nil {
+ t.Fatal("wrong subject")
+ }
+ tx.Rollback()
+ f.claim(r)
+ r.Request = f.req()
+ r.Reservation = "second"
+ f.ok(func(tx *sql.Tx) (Outcome, error) { return Reserve(tx, f, r, f.now) })
+ q.Request = f.req()
+ q.Reservation = "second"
+ tx, _ = f.db.Begin()
+ if _, e := Claim(tx, f, q, f.now); e == nil {
+ t.Fatal("duplicate attempt")
+ }
+ tx.Rollback()
+}
+
+func TestAbsentPoolHoldActuallyBarsLaterReserve(t *testing.T) {
+ f := fresh(t)
+ f.basis("10")
+ r := f.reserve("1")
+ f.claim(r)
+ ev := f.evidence("0")
+ ev.Until = time.Date(2026, 11, 2, 0, 0, 0, 0, time.UTC)
+ ob := ObserveRequest{f.req(), ev}
+ f.ok(func(tx *sql.Tx) (Outcome, error) { return Observe(tx, f, checks{}, ob, f.now) })
+ b := Basis{Request: f.req(), Revision: 1, Identity: f.gate.Identity, Digest: "nov", Lines: []Line{{Awake, "10"}}}
+ b.Month = "2026-11"
+ f.ok(func(tx *sql.Tx) (Outcome, error) { return InstallBasis(tx, f, b, f.now) })
+ r.Request = f.req()
+ r.Month = "2026-11"
+ r.Reservation = "nov"
+ r.From = time.Date(2026, 11, 1, 0, 0, 0, 0, time.UTC)
+ r.Until = r.From.Add(time.Hour)
+ r.ClaimBy = r.Until
+ if _, e := f.call(true, func(tx *sql.Tx) (Outcome, error) { return Reserve(tx, f, r, r.From) }); e == nil {
+ t.Fatal("absent-pool hold ignored")
+ }
+}
+
+func TestMultiLineRollbackAndProofRefusal(t *testing.T) {
+ f := fresh(t)
+ b := Basis{Request: f.req(), Revision: 1, Identity: f.gate.Identity, Digest: "two", Lines: []Line{{Awake, "10"}, {Transfer, "0"}}}
+ f.gate.Basis = map[Resource]BasisFact{Awake: {1, "two", f.gate.Identity}, Transfer: {1, "two", f.gate.Identity}}
+ f.ok(func(tx *sql.Tx) (Outcome, error) { return InstallBasis(tx, f, b, f.now) })
+ r := ReserveRequest{Request: f.req(), Reservation: "res", Lines: []Line{{Awake, "2"}, {Transfer, "1"}}, BasisRevision: 1, From: f.now, Until: f.now.Add(time.Minute), ClaimBy: f.now.Add(time.Second)}
+ if _, e := f.call(true, func(tx *sql.Tx) (Outcome, error) { return Reserve(tx, f, r, f.now) }); e == nil {
+ t.Fatal("partial reservation")
+ }
+ if f.pool().Available != "10" {
+ t.Fatal(f.pool())
+ }
+ r.Lines = r.Lines[:1]
+ f.ok(func(tx *sql.Tx) (Outcome, error) { return Reserve(tx, f, r, f.now) })
+ f.claim(r)
+ fin := f.finish(f.evidence("1"))
+ if _, e := f.call(true, func(tx *sql.Tx) (Outcome, error) { return Finish(tx, f, checks{fail: true}, fin, f.now) }); e == nil {
+ t.Fatal("unproven finish")
+ }
+ if f.pool().Outstanding != "2" {
+ t.Fatal("released without proof")
+ }
+}
+
+func TestFinishRejectsDuplicateEvidenceBeforeWrites(t *testing.T) {
+ f := fresh(t)
+ f.gate.Basis = map[Resource]BasisFact{Awake: {1, "two", f.gate.Identity}, Transfer: {1, "two", f.gate.Identity}}
+ b := Basis{Request: f.req(), Revision: 1, Identity: f.gate.Identity, Digest: "two", Lines: []Line{{Awake, "10"}, {Transfer, "10"}}}
+ f.ok(func(tx *sql.Tx) (Outcome, error) { return InstallBasis(tx, f, b, f.now) })
+ r := ReserveRequest{Request: f.req(), Reservation: "res", Lines: []Line{{Awake, "8"}, {Transfer, "8"}}, BasisRevision: 1, From: f.now, Until: f.now.Add(time.Minute), ClaimBy: f.now.Add(time.Second)}
+ f.ok(func(tx *sql.Tx) (Outcome, error) { return Reserve(tx, f, r, f.now) })
+ f.claim(r)
+ awake := f.evidence("3")
+ transfer := awake
+ transfer.Resource = Transfer
+ transfer.Total = "5"
+ fin := f.finish(awake)
+ fin.Evidence = append(fin.Evidence, transfer)
+ tx, _ := f.db.Begin()
+ defer tx.Rollback()
+ var beforeEvents int
+ tx.QueryRow(`SELECT count(*) FROM rb_events`).Scan(&beforeEvents)
+ if _, e := Finish(tx, f, checks{}, fin, f.now); !errors.Is(e, ErrInvalid) {
+ t.Fatalf("duplicate ID: %v", e)
+ }
+ var events, evidence, receipts int
+ tx.QueryRow(`SELECT count(*) FROM rb_events`).Scan(&events)
+ tx.QueryRow(`SELECT count(*) FROM rb_evidence`).Scan(&evidence)
+ tx.QueryRow(`SELECT count(*) FROM rb_receipts WHERE id=?`, fin.ID).Scan(&receipts)
+ if events != beforeEvents || evidence != 0 || receipts != 0 {
+ t.Fatal(events, beforeEvents, evidence, receipts)
+ }
+ for _, resource := range []Resource{Awake, Transfer} {
+ var p Pool
+ if e := get(tx, "pools", poolKey("account", "2026-10", resource), &p); e != nil {
+ t.Fatal(e)
+ }
+ if p.Outstanding != "8" || p.Settled != Zero {
+ t.Fatal(p)
+ }
+ }
+}
+
+func TestSettleDefensivelyRejectsConflictingEvidence(t *testing.T) {
+ f := fresh(t)
+ f.basis("10")
+ r := f.reserve("8")
+ f.claim(r)
+ ev := f.evidence("3")
+ ob := ObserveRequest{f.req(), ev}
+ f.ok(func(tx *sql.Tx) (Outcome, error) { return Observe(tx, f, checks{}, ob, f.now) })
+ tx, _ := f.db.Begin()
+ defer tx.Rollback()
+ var rr Reservation
+ if e := get(tx, "reservations", "res", &rr); e != nil {
+ t.Fatal(e)
+ }
+ o := &operation{tx: tx, r: f.req()}
+ ev.Total = "4"
+ if _, e := o.settle(&rr, ev); !errors.Is(e, ErrCorrupt) {
+ t.Fatal(e)
+ }
+ var p Pool
+ get(tx, "pools", poolKey("account", "2026-10", Awake), &p)
+ if p.Settled != "3" {
+ t.Fatal(p)
+ }
+}
+
+func TestEvidenceAffectedMonthsUseUTC(t *testing.T) {
+ for _, commit := range []bool{false, true} {
+ t.Run(fmt.Sprint(commit), func(t *testing.T) {
+ f := fresh(t)
+ f.basis("10")
+ r := f.reserve("8")
+ f.claim(r)
+ ev := f.evidence("3")
+ ev.From = time.Date(2026, 10, 1, 0, 30, 0, 0, time.FixedZone("x", 3600))
+ ev.Until = time.Date(2026, 10, 1, 0, 10, 0, 0, time.UTC)
+ ob := ObserveRequest{f.req(), ev}
+ out, e := f.call(commit, func(tx *sql.Tx) (Outcome, error) { return Observe(tx, f, checks{}, ob, f.now) })
+ if e != nil || out.Receipt.Outcome != "held:evidence_window" {
+ t.Fatal(out, e)
+ }
+ for _, month := range []string{"2026-09", "2026-10"} {
+ var n int
+ f.db.QueryRow(`SELECT count(*) FROM rb_holds WHERE month=?`, month).Scan(&n)
+ want := 0
+ if commit {
+ want = 1
+ }
+ if n != want {
+ t.Fatal(month, n, want)
+ }
+ }
+ var n int
+ f.db.QueryRow(`SELECT count(*) FROM rb_receipts WHERE id=?`, ob.ID).Scan(&n)
+ want := 0
+ if commit {
+ want = 1
+ }
+ if n != want || f.pool().Settled != Zero {
+ t.Fatal(n, f.pool())
+ }
+ })
+ }
+}
+
+func TestFreshBasisFactsRequiredForReserveAndClaim(t *testing.T) {
+ for _, op := range []string{"Reserve", "Claim"} {
+ for _, kind := range []string{"missing", "revision", "digest", "identity"} {
+ t.Run(op+kind, func(t *testing.T) {
+ f := fresh(t)
+ f.basis("10")
+ r := ReserveRequest{Request: f.req(), Reservation: "res", Lines: []Line{{Awake, "1"}}, BasisRevision: 1, From: f.now, Until: f.now.Add(time.Minute), ClaimBy: f.now.Add(time.Second)}
+ if op == "Claim" {
+ f.ok(func(tx *sql.Tx) (Outcome, error) { return Reserve(tx, f, r, f.now) })
+ }
+ fact := f.gate.Basis[Awake]
+ switch kind {
+ case "missing":
+ delete(f.gate.Basis, Awake)
+ case "revision":
+ fact.Revision++
+ f.gate.Basis[Awake] = fact
+ case "digest":
+ fact.Digest = "new-authoritative-digest"
+ f.gate.Basis[Awake] = fact
+ case "identity":
+ fact.Identity.Terms = "new-authoritative-terms"
+ f.gate.Basis[Awake] = fact
+ }
+ tx, _ := f.db.Begin()
+ defer tx.Rollback()
+ var e error
+ if op == "Reserve" {
+ _, e = Reserve(tx, f, r, f.now)
+ } else {
+ _, e = Claim(tx, f, ClaimRequest{Request: f.req(), Reservation: "res", Producer: "p", Attempt: "a", ProducerEpoch: 1}, f.now)
+ }
+ if !errors.Is(e, ErrDenied) {
+ t.Fatal(e)
+ }
+ })
+ }
+ }
+}
+
+func TestClaimReturnsNilOnReceiptFailure(t *testing.T) {
+ f := fresh(t)
+ f.basis("10")
+ r := f.reserve("1")
+ q := ClaimRequest{Request: f.req(), Reservation: r.Reservation, Producer: "p", Attempt: "a", ProducerEpoch: 1}
+ if _, e := f.db.Exec(`CREATE TRIGGER fail_claim_receipt BEFORE INSERT ON rb_receipts WHEN json_extract(NEW.body,'$.Kind')='Claim' BEGIN SELECT RAISE(ABORT,'injected'); END`); e != nil {
+ t.Fatal(e)
+ }
+ tx, _ := f.db.Begin()
+ v, e := Claim(tx, f, q, f.now)
+ tx.Rollback()
+ if e == nil || v != nil {
+ t.Fatal(v, e)
+ }
+ tx, _ = f.db.Begin()
+ defer tx.Rollback()
+ var rr Reservation
+ get(tx, "reservations", "res", &rr)
+ if rr.State != "reserved" {
+ t.Fatal(rr.State)
+ }
+}
diff --git a/internal/resourcebudget/recovery_test.go b/internal/resourcebudget/recovery_test.go
new file mode 100644
index 000000000..0f0bc7b23
--- /dev/null
+++ b/internal/resourcebudget/recovery_test.go
@@ -0,0 +1,254 @@
+package resourcebudget
+
+import (
+ "database/sql"
+ "errors"
+ "fmt"
+ "strings"
+ "testing"
+ "time"
+)
+
+func TestDurabilityAndReopen(t *testing.T) {
+ for _, level := range []int{0, 1, 2, 3} {
+ t.Run(fmt.Sprint(level), func(t *testing.T) {
+ f := fresh(t)
+ db, e := sql.Open("sqlite", strings.Split(f.dsn, "?")[0]+fmt.Sprintf("?_pragma=synchronous(%d)&_pragma=busy_timeout(5000)", level))
+ if e != nil {
+ t.Fatal(e)
+ }
+ defer db.Close()
+ db.SetMaxOpenConns(1)
+ tx, _ := db.Begin()
+ b := Basis{Request: f.req(), Revision: 1, Identity: f.gate.Identity, Digest: "b", Lines: []Line{{Awake, "10"}}}
+ _, e = InstallBasis(tx, f, b, f.now)
+ if (e == nil) != (level >= 2) {
+ t.Fatal(e)
+ }
+ if e == nil {
+ if e = tx.Commit(); e != nil {
+ t.Fatal(e)
+ }
+ } else {
+ tx.Rollback()
+ }
+ tx, _ = f.db.Begin()
+ defer tx.Rollback()
+ if e = Verify(tx); e != nil {
+ t.Fatal(e)
+ }
+ })
+ }
+}
+func TestSeparateHandlesContestAndReplay(t *testing.T) {
+ f := fresh(t)
+ f.basis("1")
+ db, e := sql.Open("sqlite", f.dsn)
+ if e != nil {
+ t.Fatal(e)
+ }
+ defer db.Close()
+ db.SetMaxOpenConns(1)
+ if e = db.Ping(); e != nil {
+ t.Fatal(e)
+ }
+ r := ReserveRequest{Request: f.req(), Reservation: "one", Lines: []Line{{Awake, "1"}}, BasisRevision: 1, From: f.now, Until: f.now.Add(time.Minute), ClaimBy: f.now.Add(time.Second)}
+ tx, _ := f.db.Begin()
+ if _, e = Reserve(tx, f, r, f.now); e != nil {
+ t.Fatal(e)
+ }
+ second := r
+ second.Request = f.req()
+ second.Reservation = "two"
+ second.Subject = "other-app"
+ started := make(chan struct{})
+ done := make(chan error, 1)
+ go func() {
+ other, err := db.Begin()
+ if err != nil {
+ done <- err
+ return
+ }
+ defer other.Rollback()
+ close(started)
+ _, err = Reserve(other, f, second, f.now)
+ if err == nil {
+ err = other.Commit()
+ }
+ done <- err
+ }()
+ <-started
+ if e = tx.Commit(); e != nil {
+ t.Fatal(e)
+ }
+ select {
+ case e = <-done:
+ if !errors.Is(e, ErrDenied) {
+ t.Fatal("expected capacity denial", e)
+ }
+ case <-time.After(6 * time.Second):
+ t.Fatal("worker blocked")
+ }
+ tx, _ = db.Begin()
+ o, e := Reserve(tx, f, r, f.now.Add(time.Hour))
+ if e != nil || !o.Replayed {
+ t.Fatal(o, e)
+ }
+ tx.Rollback()
+ tx, _ = db.Begin()
+ defer tx.Rollback()
+ if e = Verify(tx); e != nil {
+ t.Fatal(e)
+ }
+}
+func TestImmutableHistoryAndVerify(t *testing.T) {
+ f := fresh(t)
+ f.basis("10")
+ r := f.reserve("8")
+ f.claim(r)
+ ob := ObserveRequest{f.req(), f.evidence("2")}
+ f.ok(func(tx *sql.Tx) (Outcome, error) { return Observe(tx, f, checks{}, ob, f.now) })
+ for _, table := range []string{"receipts", "events", "basis", "evidence"} {
+ for _, q := range []string{`DELETE FROM rb_` + table, `UPDATE rb_` + table + ` SET ` + map[string]string{"receipts": "body=body", "events": "body=body", "basis": "body=body", "evidence": "body=body"}[table], `INSERT OR REPLACE INTO rb_` + table + ` SELECT * FROM rb_` + table + ` LIMIT 1`} {
+ if _, e := f.db.Exec(q); e == nil {
+ t.Fatalf("immutable %s", q)
+ }
+ }
+ }
+ if _, e := f.db.Exec(`UPDATE rb_pools SET body='{}'`); e != nil {
+ t.Fatal(e)
+ }
+ tx, _ := f.db.Begin()
+ defer tx.Rollback()
+ if e := Verify(tx); e == nil {
+ t.Fatal("corruption unnoticed")
+ }
+}
+func TestMonotoneIncreaseAndHoldAdmission(t *testing.T) {
+ f := fresh(t)
+ b := f.basis("10")
+ b.Request = f.req()
+ b.Revision = 2
+ b.Lines[0].Amount = "12"
+ f.gate.Basis[Awake] = BasisFact{2, "basis", f.gate.Identity}
+ f.ok(func(tx *sql.Tx) (Outcome, error) { return IncreaseBasis(tx, f, b, f.now) })
+ if f.pool().Available != "12" {
+ t.Fatal(f.pool())
+ }
+ r := ReserveRequest{Request: f.req(), Reservation: "res", Lines: []Line{{Awake, "1"}}, BasisRevision: 2, From: f.now, Until: f.now.Add(time.Minute), ClaimBy: f.now.Add(time.Second)}
+ f.ok(func(tx *sql.Tx) (Outcome, error) { return Reserve(tx, f, r, f.now) })
+ b.Request = f.req()
+ f.ok(func(tx *sql.Tx) (Outcome, error) { return IncreaseBasis(tx, f, b, f.now) })
+ q := ClaimRequest{Request: f.req(), Reservation: "res", Producer: "p", Attempt: "a", ProducerEpoch: 1}
+ tx, _ := f.db.Begin()
+ if _, e := Claim(tx, f, q, f.now); e == nil {
+ t.Fatal("claim under hold")
+ }
+ tx.Rollback()
+ r.Request = f.req()
+ r.Reservation = "another"
+ if _, e := f.call(true, func(tx *sql.Tx) (Outcome, error) { return Reserve(tx, f, r, f.now) }); e == nil {
+ t.Fatal("reserve under hold")
+ }
+ cancel := ReservationRequest{f.req(), "res"}
+ f.ok(func(tx *sql.Tx) (Outcome, error) { return CancelUnclaimed(tx, f, cancel, f.now) })
+}
+
+func TestVerifyRejectsHistoryDespiteMatchingCurrentRows(t *testing.T) {
+ f := fresh(t)
+ f.basis("10")
+ r := f.reserve("2")
+ q := f.claim(r)
+ // Simulate a defective writer, not an attacker: matching row/event payloads
+ // cannot turn a Claim transition into a cancellation transition.
+ if _, e := f.db.Exec(`DROP TRIGGER rb_receipts_UPDATE`); e != nil {
+ t.Fatal(e)
+ }
+ if _, e := f.db.Exec(`UPDATE rb_receipts SET body=json_set(body,'$.Kind','CancelUnclaimed') WHERE id=?`, q.ID); e != nil {
+ t.Fatal(e)
+ }
+ tx, _ := f.db.Begin()
+ defer tx.Rollback()
+ if e := Verify(tx); e == nil {
+ t.Fatal("invalid history accepted")
+ }
+}
+
+func TestPersistedMalformedAmountsRefuseWithoutPanic(t *testing.T) {
+ for _, bad := range []string{"-1", "01", "not-a-number", "170141183460469231731687303715884105728"} {
+ t.Run(bad, func(t *testing.T) {
+ f := fresh(t)
+ f.basis("10")
+ if _, e := f.db.Exec(`UPDATE rb_pools SET body=json_set(body,'$.Available',?)`, bad); e != nil {
+ t.Fatal(e)
+ }
+ r := ReserveRequest{Request: f.req(), Reservation: "res", Lines: []Line{{Awake, "1"}}, BasisRevision: 1, From: f.now, Until: f.now.Add(time.Minute), ClaimBy: f.now.Add(time.Second)}
+ tx, _ := f.db.Begin()
+ _, e := Reserve(tx, f, r, f.now)
+ tx.Rollback()
+ if !errors.Is(e, ErrCorrupt) {
+ t.Fatal(e)
+ }
+ var n int
+ f.db.QueryRow(`SELECT count(*) FROM rb_receipts WHERE id=?`, r.ID).Scan(&n)
+ if n != 0 {
+ t.Fatal("corrupt state wrote receipt")
+ }
+ })
+ }
+
+}
+
+func TestVerifyValidatesEveryHistoricalSnapshot(t *testing.T) {
+ for _, table := range []string{"pools", "limits", "reservations"} {
+ t.Run(table, func(t *testing.T) {
+ f := fresh(t)
+ f.basis("10")
+ f.limit()
+ f.reserve("1")
+ // A defective writer can emit two snapshots in one operation. The first
+ // must be validated even though the operation's final snapshot is valid.
+ var id, body string
+ if e := f.db.QueryRow(`SELECT id,body FROM rb_`+table+` LIMIT 1`).Scan(&id, &body); e != nil {
+ t.Fatal(e)
+ }
+ tx, _ := f.db.Begin()
+ defer tx.Rollback()
+ if table == "pools" {
+ var next string
+ if e := tx.QueryRow(`SELECT json_set(?,'$.Month','2026-11')`, body).Scan(&next); e != nil {
+ t.Fatal(e)
+ }
+ body = next
+ id = poolKey("account", "2026-11", Awake)
+ }
+ var bad string
+ field := map[string]string{"pools": "$.Capacity", "limits": "$.Unused", "reservations": "$.Allocations[0].Total"}[table]
+ if e := tx.QueryRow(`SELECT json_set(?,?,'bad')`, body, field).Scan(&bad); e != nil {
+ t.Fatal(e)
+ }
+ receipt := Receipt{ID: "bad-history", Kind: "InstallBasis", Account: "account", Subject: "app", Epoch: 1}
+ if table == "limits" {
+ receipt.Kind = "InstallPaidLimit"
+ }
+ if table == "reservations" {
+ receipt.Kind = "Reserve"
+ }
+ raw, e := encode(receipt)
+ if e != nil {
+ t.Fatal(e)
+ }
+ if _, e = tx.Exec(`INSERT INTO rb_receipts VALUES(?,?,?)`, receipt.ID, "synthetic", string(raw)); e != nil {
+ t.Fatal(e)
+ }
+ for _, b := range []string{bad, body} {
+ if _, e = tx.Exec(`INSERT INTO rb_events(kind,id,body,epoch,operation_id) VALUES(?,?,?,1,?)`, table, id, b, receipt.ID); e != nil {
+ t.Fatal(e)
+ }
+ }
+ if e = Verify(tx); !errors.Is(e, ErrCorrupt) {
+ t.Fatal(e)
+ }
+ })
+ }
+}
diff --git a/internal/resourcebudget/reservation.go b/internal/resourcebudget/reservation.go
new file mode 100644
index 000000000..f0666934b
--- /dev/null
+++ b/internal/resourcebudget/reservation.go
@@ -0,0 +1,271 @@
+package resourcebudget
+
+import (
+ "database/sql"
+ "encoding/json"
+ "time"
+)
+
+func decode(s string, v any) error {
+ if json.Unmarshal([]byte(s), v) != nil {
+ return ErrCorrupt
+ }
+ return nil
+}
+
+type ReserveRequest struct {
+ Request
+ Reservation string
+ Lines []Line
+ BasisRevision int64
+ Consent int64
+ AllowPaid bool
+ From, Until, ClaimBy time.Time
+}
+type Allocation struct {
+ Resource Resource
+ Maximum, Included, Paid, SettledIncluded, SettledPaid, Excess, Total Amount
+ Limit string
+}
+type Reservation struct {
+ ReserveRequest
+ State, Producer, Attempt string
+ ProducerEpoch int64
+ Allocations []Allocation
+ TerminalProof *Terminal
+}
+type ClaimRequest struct {
+ Request
+ Reservation, Producer, Attempt string
+ ProducerEpoch int64
+}
+type ReservationRequest struct {
+ Request
+ Reservation string
+}
+type FenceRequest struct {
+ Request
+ Producer string
+ ProducerEpoch int64
+}
+
+func Reserve(tx *sql.Tx, l GateLoader, r ReserveRequest, now time.Time) (Outcome, error) {
+ if validateLines(r.Lines) != nil || r.Reservation == "" || r.BasisRevision <= 0 || !window(r.Month, r.From, r.Until) || r.ClaimBy.IsZero() || r.ClaimBy.After(r.Until) {
+ return Outcome{}, ErrInvalid
+ }
+ for _, line := range r.Lines {
+ if line.Amount == Zero {
+ return Outcome{}, ErrInvalid
+ }
+ }
+ o, re, e := begin(tx, l, r.Request, "Reserve", r, now)
+ if e != nil {
+ return Outcome{}, e
+ }
+ if re != nil {
+ return *re, nil
+ }
+ if !now.Before(r.ClaimBy) {
+ return Outcome{}, ErrDenied
+ }
+ var old Reservation
+ e = get(tx, "reservations", r.Reservation, &old)
+ if e == nil {
+ return Outcome{}, ErrConflict
+ }
+ if !missing(e) {
+ return Outcome{}, e
+ }
+ rr := Reservation{ReserveRequest: r, State: "reserved"}
+ for _, line := range r.Lines {
+ if e = o.admit(line.Resource); e != nil {
+ return Outcome{}, e
+ }
+ var p Pool
+ if e = get(tx, "pools", poolKey(r.Account, r.Month, line.Resource), &p); e != nil {
+ return Outcome{}, e
+ }
+ if p.Revision != r.BasisRevision || !currentBasis(o.gate, p) {
+ return Outcome{}, ErrDenied
+ }
+ inc := min(line.Amount, p.Available)
+ paid, _ := sub(line.Amount, inc)
+ a := Allocation{Resource: line.Resource, Maximum: line.Amount, Included: inc, Paid: paid, SettledIncluded: Zero, SettledPaid: Zero, Excess: Zero, Total: Zero}
+ if paid != Zero {
+ if !r.AllowPaid || r.Consent <= 0 {
+ return Outcome{}, ErrDenied
+ }
+ a.Limit = limitKey(r.Account, r.Month, line.Resource, r.Consent, r.Subject)
+ var q PaidLimit
+ if e = get(tx, "limits", a.Limit, &q); e != nil {
+ return Outcome{}, e
+ }
+ if !paidGate(o.gate, q.Consent, q.Offer, q.Identity) || r.From.Before(q.From) || r.Until.After(q.Until) || cmp(q.Unused, paid) < 0 {
+ return Outcome{}, ErrDenied
+ }
+ q.Unused, _ = sub(q.Unused, paid)
+ q.Outstanding, e = add(q.Outstanding, paid)
+ if e != nil {
+ return Outcome{}, e
+ }
+ if e = o.put("limits", a.Limit, q, false); e != nil {
+ return Outcome{}, e
+ }
+ }
+ p.Available, _ = sub(p.Available, inc)
+ p.Outstanding, e = add(p.Outstanding, inc)
+ if e != nil {
+ return Outcome{}, e
+ }
+ if e = o.put("pools", poolKey(r.Account, r.Month, line.Resource), p, false); e != nil {
+ return Outcome{}, e
+ }
+ rr.Allocations = append(rr.Allocations, a)
+ }
+ if e = o.put("reservations", r.Reservation, rr, false); e != nil {
+ return Outcome{}, e
+ }
+ return o.finish("reserved")
+}
+func bound(o *operation, id string) (Reservation, error) {
+ var r Reservation
+ e := get(o.tx, "reservations", id, &r)
+ if e != nil {
+ return r, e
+ }
+ if r.Account != o.r.Account || r.Month != o.r.Month || r.Subject != o.r.Subject || r.Epoch != o.r.Epoch {
+ return r, ErrDenied
+ }
+ return r, nil
+}
+func Claim(tx *sql.Tx, l GateLoader, r ClaimRequest, now time.Time) (ClaimResult, error) {
+ if r.Reservation == "" || r.Producer == "" || r.Attempt == "" || r.ProducerEpoch <= 0 {
+ return nil, ErrInvalid
+ }
+ o, re, e := begin(tx, l, r.Request, "Claim", r, now)
+ if e != nil {
+ return nil, e
+ }
+ if re != nil {
+ return ReplayedClaim{re.Receipt}, nil
+ }
+ rr, e := bound(o, r.Reservation)
+ if e != nil {
+ return nil, e
+ }
+ if rr.State != "reserved" || !now.Before(rr.ClaimBy) {
+ return nil, ErrDenied
+ }
+ var fence FenceRequest
+ e = get(tx, "fences", key(r.Account, r.Producer), &fence)
+ if e != nil && !missing(e) {
+ return nil, e
+ }
+ if e == nil && r.ProducerEpoch <= fence.ProducerEpoch {
+ return nil, ErrDenied
+ }
+ for _, a := range rr.Allocations {
+ if e = o.admit(a.Resource); e != nil {
+ return nil, e
+ }
+ var p Pool
+ if e = get(tx, "pools", poolKey(r.Account, r.Month, a.Resource), &p); e != nil {
+ return nil, e
+ }
+ if p.Revision != rr.BasisRevision || !currentBasis(o.gate, p) {
+ return nil, ErrDenied
+ }
+ if a.Paid != Zero {
+ var q PaidLimit
+ if e = get(tx, "limits", a.Limit, &q); e != nil {
+ return nil, e
+ }
+ if !paidGate(o.gate, q.Consent, q.Offer, q.Identity) {
+ return nil, ErrDenied
+ }
+ }
+ }
+ rr.State, rr.Producer, rr.Attempt, rr.ProducerEpoch = "claimed", r.Producer, r.Attempt, r.ProducerEpoch
+ if e = o.put("reservations", r.Reservation, rr, false); e != nil {
+ return nil, e
+ }
+ out, e := o.finish("claimed")
+ if e != nil {
+ return nil, e
+ }
+ return NewClaimProvisional{out.Receipt}, nil
+}
+func CancelUnclaimed(tx *sql.Tx, l GateLoader, r ReservationRequest, now time.Time) (Outcome, error) {
+ o, re, e := begin(tx, l, r.Request, "CancelUnclaimed", r, now)
+ if e != nil {
+ return Outcome{}, e
+ }
+ if re != nil {
+ return *re, nil
+ }
+ rr, e := bound(o, r.Reservation)
+ if e != nil {
+ return Outcome{}, e
+ }
+ if rr.State != "reserved" {
+ return Outcome{}, ErrDenied
+ }
+ if e = release(o, &rr); e != nil {
+ return Outcome{}, e
+ }
+ rr.State = "closed-unused"
+ if e = o.put("reservations", r.Reservation, rr, false); e != nil {
+ return Outcome{}, e
+ }
+ return o.finish("closed-unused")
+}
+func MarkUncertain(tx *sql.Tx, l GateLoader, r ReservationRequest, now time.Time) (Outcome, error) {
+ o, re, e := begin(tx, l, r.Request, "MarkUncertain", r, now)
+ if e != nil {
+ return Outcome{}, e
+ }
+ if re != nil {
+ return *re, nil
+ }
+ rr, e := bound(o, r.Reservation)
+ if e != nil {
+ return Outcome{}, e
+ }
+ if rr.State != "claimed" && rr.State != "uncertain" {
+ return Outcome{}, ErrDenied
+ }
+ rr.State = "uncertain"
+ if e = o.put("reservations", r.Reservation, rr, false); e != nil {
+ return Outcome{}, e
+ }
+ return o.finish("uncertain")
+}
+func FenceProducer(tx *sql.Tx, l GateLoader, r FenceRequest, now time.Time) (Outcome, error) {
+ if r.Producer == "" || r.ProducerEpoch <= 0 {
+ return Outcome{}, ErrInvalid
+ }
+ o, re, e := begin(tx, l, r.Request, "FenceProducer", r, now)
+ if e != nil {
+ return Outcome{}, e
+ }
+ if re != nil {
+ return *re, nil
+ }
+ var old FenceRequest
+ e = get(tx, "fences", key(r.Account, r.Producer), &old)
+ if e != nil && !missing(e) {
+ return Outcome{}, e
+ }
+ if e == nil && r.ProducerEpoch <= old.ProducerEpoch {
+ return Outcome{}, ErrDenied
+ }
+ if e = o.put("fences", key(r.Account, r.Producer), r, false); e != nil {
+ return Outcome{}, e
+ }
+ return o.finish("fenced")
+}
+
+func currentBasis(g Gate, p Pool) bool {
+ b, ok := g.Basis[p.Resource]
+ return ok && b.Revision == p.Revision && b.Digest == p.Digest && b.Identity == p.Identity
+}
diff --git a/internal/resourcebudget/schema.go b/internal/resourcebudget/schema.go
new file mode 100644
index 000000000..ef3bc9ecb
--- /dev/null
+++ b/internal/resourcebudget/schema.go
@@ -0,0 +1,40 @@
+package resourcebudget
+
+import "database/sql"
+
+// Migrate is not registered with console startup. Callers own its transaction.
+func Migrate(tx *sql.Tx) error {
+ for _, s := range []string{
+ `CREATE TABLE IF NOT EXISTS rb_authority(account TEXT PRIMARY KEY, serial INTEGER NOT NULL, epoch INTEGER NOT NULL, latest INTEGER NOT NULL)`,
+ `CREATE TABLE IF NOT EXISTS rb_receipts(id TEXT PRIMARY KEY, digest TEXT NOT NULL, body TEXT NOT NULL)`,
+ `CREATE TABLE IF NOT EXISTS rb_events(seq INTEGER PRIMARY KEY, kind TEXT NOT NULL, id TEXT NOT NULL, body TEXT NOT NULL, epoch INTEGER NOT NULL, operation_id TEXT NOT NULL)`,
+ `CREATE TABLE IF NOT EXISTS rb_holds(id TEXT PRIMARY KEY, account TEXT NOT NULL, month TEXT NOT NULL, resource TEXT NOT NULL, reason TEXT NOT NULL, epoch INTEGER NOT NULL)`,
+ } {
+ if _, e := tx.Exec(s); e != nil {
+ return e
+ }
+ }
+ for _, table := range []string{"pools", "limits", "reservations", "fences", "evidence", "basis"} {
+ if _, e := tx.Exec(`CREATE TABLE IF NOT EXISTS rb_` + table + `(id TEXT PRIMARY KEY, body TEXT NOT NULL)`); e != nil {
+ return e
+ }
+ }
+ if _, e := tx.Exec(`CREATE UNIQUE INDEX IF NOT EXISTS rb_attempt_unique ON rb_reservations(json_extract(body,'$.Account'),json_extract(body,'$.Producer'),json_extract(body,'$.ProducerEpoch'),json_extract(body,'$.Attempt')) WHERE json_extract(body,'$.Attempt')<>''`); e != nil {
+ return e
+ }
+ for _, table := range []string{"receipts", "events", "holds", "evidence", "basis"} {
+ key := "id"
+ if table == "events" {
+ key = "seq"
+ }
+ for _, op := range []string{"UPDATE", "DELETE"} {
+ if _, e := tx.Exec(`CREATE TRIGGER IF NOT EXISTS rb_` + table + `_` + op + ` BEFORE ` + op + ` ON rb_` + table + ` BEGIN SELECT RAISE(ABORT,'resourcebudget: immutable'); END`); e != nil {
+ return e
+ }
+ }
+ if _, e := tx.Exec(`CREATE TRIGGER IF NOT EXISTS rb_` + table + `_replace BEFORE INSERT ON rb_` + table + ` WHEN EXISTS(SELECT 1 FROM rb_` + table + ` WHERE ` + key + `=NEW.` + key + `) BEGIN SELECT RAISE(ABORT,'resourcebudget: immutable'); END`); e != nil {
+ return e
+ }
+ }
+ return nil
+}
diff --git a/internal/resourcebudget/settlement.go b/internal/resourcebudget/settlement.go
new file mode 100644
index 000000000..d4ea87956
--- /dev/null
+++ b/internal/resourcebudget/settlement.go
@@ -0,0 +1,375 @@
+package resourcebudget
+
+import (
+ "database/sql"
+ "time"
+)
+
+type Evidence struct {
+ ID, Account, Reservation, Subject, Attempt, Producer string
+ ProducerEpoch int64
+ Resource Resource
+ Total Amount
+ From, Until time.Time
+}
+type Terminal struct {
+ ID, Account, Reservation, Subject, Attempt, Producer string
+ ProducerEpoch int64
+}
+type Checker interface {
+ Measurement(Evidence) error
+ Terminal(Terminal) error
+}
+type ObserveRequest struct {
+ Request
+ Evidence Evidence
+}
+type FinishRequest struct {
+ Request
+ Reservation string
+ Evidence []Evidence
+ Terminal Terminal
+}
+
+func validEvidence(e Evidence) error {
+ if e.ID == "" || e.Account == "" || e.Reservation == "" || e.Subject == "" || e.Attempt == "" || e.Producer == "" || e.ProducerEpoch <= 0 || !e.Resource.valid() || !e.From.Before(e.Until) {
+ return ErrInvalid
+ }
+ _, err := number(e.Total)
+ return err
+}
+func evidenceBound(r Reservation, e Evidence) bool {
+ return e.Account == r.Account && e.Reservation == r.Reservation && e.Subject == r.Subject && e.Attempt == r.Attempt && e.Producer == r.Producer && e.ProducerEpoch == r.ProducerEpoch
+}
+func affected(r Reservation, e Evidence) []string {
+ m := []string{r.Month}
+ from := e.From.UTC()
+ a := time.Date(from.Year(), from.Month(), 1, 0, 0, 0, 0, time.UTC)
+ end := e.Until.Add(-time.Nanosecond)
+ for !a.After(end) {
+ m = append(m, a.Format("2006-01"))
+ a = a.AddDate(0, 1, 0)
+ }
+ return m
+}
+func (o *operation) checkEvidence(r Reservation, e Evidence) (string, error) {
+ if !evidenceBound(r, e) {
+ return "", ErrDenied
+ }
+ found := false
+ for _, a := range r.Allocations {
+ if a.Resource == e.Resource {
+ found = true
+ }
+ }
+ if !found {
+ return "", ErrDenied
+ }
+ var old Evidence
+ err := get(o.tx, "evidence", e.ID, &old)
+ oldDigest, de := digest(old)
+ if de != nil {
+ return "", de
+ }
+ newDigest, de := digest(e)
+ if de != nil {
+ return "", de
+ }
+ if err == nil && oldDigest != newDigest {
+ return "evidence_conflict", nil
+ }
+ if err != nil && !missing(err) {
+ return "", err
+ }
+ if e.From.Before(r.From) || e.Until.After(r.Until) || !window(r.Month, e.From, e.Until) {
+ return "evidence_window", nil
+ }
+ for _, a := range r.Allocations {
+ if a.Resource == e.Resource && cmp(e.Total, a.Total) < 0 {
+ return "regression", nil
+ }
+ }
+ return "", nil
+}
+func Observe(tx *sql.Tx, l GateLoader, c Checker, r ObserveRequest, now time.Time) (Outcome, error) {
+ if c == nil || validEvidence(r.Evidence) != nil {
+ return Outcome{}, ErrInvalid
+ }
+ o, re, e := begin(tx, l, r.Request, "Observe", r, now)
+ if e != nil {
+ return Outcome{}, e
+ }
+ if re != nil {
+ return *re, nil
+ }
+ if e := c.Measurement(r.Evidence); e != nil {
+ return Outcome{}, e
+ }
+ rr, e := bound(o, r.Evidence.Reservation)
+ if e != nil {
+ return Outcome{}, e
+ }
+ if !evidenceBound(rr, r.Evidence) {
+ return Outcome{}, ErrDenied
+ }
+ reason, e := o.checkEvidence(rr, r.Evidence)
+ if e != nil {
+ return Outcome{}, e
+ }
+ if reason != "" {
+ return o.anomaly(r.Evidence.Resource, reason, affected(rr, r.Evidence)...)
+ }
+ if rr.State == "closed" {
+ for _, a := range rr.Allocations {
+ if a.Resource == r.Evidence.Resource && a.Total != r.Evidence.Total {
+ return o.anomaly(a.Resource, "finish_contradiction")
+ }
+ }
+ return o.finish("closed")
+ }
+ if rr.State != "claimed" && rr.State != "uncertain" {
+ return Outcome{}, ErrDenied
+ }
+ over, e := o.settle(&rr, r.Evidence)
+ if e != nil {
+ return Outcome{}, e
+ }
+ if e = o.put("reservations", rr.Reservation, rr, false); e != nil {
+ return Outcome{}, e
+ }
+ if over {
+ return o.anomaly(r.Evidence.Resource, "overshoot")
+ }
+ return o.finish("observed")
+}
+func (o *operation) settle(r *Reservation, e Evidence) (bool, error) {
+ var prior Evidence
+ err := get(o.tx, "evidence", e.ID, &prior)
+ if missing(err) {
+ if err = o.put("evidence", e.ID, e, true); err != nil {
+ return false, err
+ }
+ } else if err != nil {
+ return false, err
+ } else {
+ oldDigest, err := digest(prior)
+ if err != nil {
+ return false, err
+ }
+ newDigest, err := digest(e)
+ if err != nil {
+ return false, err
+ }
+ if oldDigest != newDigest {
+ return false, ErrCorrupt
+ }
+ }
+ for i, a := range r.Allocations {
+ if a.Resource != e.Resource {
+ continue
+ }
+ bounded := min(e.Total, a.Maximum)
+ inc := min(bounded, a.Included)
+ paid, _ := sub(bounded, inc)
+ excess, _ := sub(e.Total, bounded)
+ di, _ := sub(inc, a.SettledIncluded)
+ dp, _ := sub(paid, a.SettledPaid)
+ dx, _ := sub(excess, a.Excess)
+ var p Pool
+ if err = get(o.tx, "pools", poolKey(r.Account, r.Month, a.Resource), &p); err != nil {
+ return false, err
+ }
+ p.Outstanding, err = sub(p.Outstanding, di)
+ if err != nil {
+ return false, err
+ }
+ p.Settled, err = add(p.Settled, di)
+ if err != nil {
+ return false, err
+ }
+ p.Excess, err = add(p.Excess, dx)
+ if err != nil {
+ return false, err
+ }
+ if err = o.put("pools", poolKey(r.Account, r.Month, a.Resource), p, false); err != nil {
+ return false, err
+ }
+ if a.Limit != "" {
+ var q PaidLimit
+ if err = get(o.tx, "limits", a.Limit, &q); err != nil {
+ return false, err
+ }
+ q.Outstanding, err = sub(q.Outstanding, dp)
+ if err != nil {
+ return false, err
+ }
+ q.Settled, err = add(q.Settled, dp)
+ if err != nil {
+ return false, err
+ }
+ if err = o.put("limits", a.Limit, q, false); err != nil {
+ return false, err
+ }
+ }
+ a.SettledIncluded, a.SettledPaid, a.Total, a.Excess = inc, paid, e.Total, excess
+ r.Allocations[i] = a
+ return excess != Zero, nil
+ }
+ return false, ErrDenied
+}
+func Finish(tx *sql.Tx, l GateLoader, c Checker, r FinishRequest, now time.Time) (Outcome, error) {
+ if c == nil || r.Reservation == "" || len(r.Evidence) == 0 || len(r.Evidence) > 3 || r.Terminal.ID == "" {
+ return Outcome{}, ErrInvalid
+ }
+ var last Resource
+ seen := map[string]bool{}
+ for _, e := range r.Evidence {
+ if seen[e.ID] {
+ return Outcome{}, ErrInvalid
+ }
+ seen[e.ID] = true
+ if validEvidence(e) != nil || e.Resource <= last {
+ return Outcome{}, ErrInvalid
+ }
+ last = e.Resource
+ }
+ o, re, err := begin(tx, l, r.Request, "Finish", r, now)
+ if err != nil {
+ return Outcome{}, err
+ }
+ if re != nil {
+ return *re, nil
+ }
+ if e := c.Terminal(r.Terminal); e != nil {
+ return Outcome{}, e
+ }
+ for _, v := range r.Evidence {
+ if e := c.Measurement(v); e != nil {
+ return Outcome{}, e
+ }
+ }
+ rr, err := bound(o, r.Reservation)
+ if err != nil {
+ return Outcome{}, err
+ }
+ t := r.Terminal
+ if t.Account != rr.Account || t.Reservation != rr.Reservation || t.Subject != rr.Subject || t.Attempt != rr.Attempt || t.Producer != rr.Producer || t.ProducerEpoch != rr.ProducerEpoch {
+ return Outcome{}, ErrDenied
+ }
+ if len(r.Evidence) != len(rr.Allocations) {
+ return Outcome{}, ErrInvalid
+ }
+ reason := ""
+ for i, e := range r.Evidence {
+ if e.Resource != rr.Allocations[i].Resource {
+ return Outcome{}, ErrInvalid
+ }
+ why, err := o.checkEvidence(rr, e)
+ if err != nil {
+ return Outcome{}, err
+ }
+ if why != "" {
+ reason = why
+ }
+ if rr.State == "closed" && e.Total != rr.Allocations[i].Total {
+ reason = "finish_contradiction"
+ }
+ }
+ if reason != "" {
+ seen := map[string]bool{}
+ for _, e := range r.Evidence {
+ for _, m := range affected(rr, e) {
+ k := key(m, e.Resource)
+ if !seen[k] {
+ seen[k] = true
+ if err = o.hold(m, e.Resource, reason); err != nil {
+ return Outcome{}, err
+ }
+ }
+ }
+ }
+ if err = o.put("basis", key("anomaly", r.ID), r, true); err != nil {
+ return Outcome{}, err
+ }
+ return o.finish("held:" + reason)
+ }
+ if rr.State == "closed" {
+ return o.finish("closed")
+ }
+ if rr.State != "claimed" && rr.State != "uncertain" {
+ return Outcome{}, ErrDenied
+ }
+ over := false
+ for _, e := range r.Evidence {
+ v, err := o.settle(&rr, e)
+ if err != nil {
+ return Outcome{}, err
+ }
+ if v {
+ over = true
+ if err = o.hold(rr.Month, e.Resource, "overshoot"); err != nil {
+ return Outcome{}, err
+ }
+ }
+ }
+ if err = release(o, &rr); err != nil {
+ return Outcome{}, err
+ }
+ rr.State = "closed"
+ rr.TerminalProof = &r.Terminal
+ if err = o.put("reservations", rr.Reservation, rr, false); err != nil {
+ return Outcome{}, err
+ }
+ if over {
+ if err = o.put("basis", key("anomaly", r.ID), r, true); err != nil {
+ return Outcome{}, err
+ }
+ return o.finish("held:overshoot")
+ }
+ return o.finish("closed")
+}
+func release(o *operation, r *Reservation) error {
+ for _, a := range r.Allocations {
+ di, err := sub(a.Included, a.SettledIncluded)
+ if err != nil {
+ return err
+ }
+ dp, err := sub(a.Paid, a.SettledPaid)
+ if err != nil {
+ return err
+ }
+ var p Pool
+ if err = get(o.tx, "pools", poolKey(r.Account, r.Month, a.Resource), &p); err != nil {
+ return err
+ }
+ p.Outstanding, err = sub(p.Outstanding, di)
+ if err != nil {
+ return err
+ }
+ p.Available, err = add(p.Available, di)
+ if err != nil {
+ return err
+ }
+ if err = o.put("pools", poolKey(r.Account, r.Month, a.Resource), p, false); err != nil {
+ return err
+ }
+ if a.Limit != "" {
+ var q PaidLimit
+ if err = get(o.tx, "limits", a.Limit, &q); err != nil {
+ return err
+ }
+ q.Outstanding, err = sub(q.Outstanding, dp)
+ if err != nil {
+ return err
+ }
+ q.Unused, err = add(q.Unused, dp)
+ if err != nil {
+ return err
+ }
+ if err = o.put("limits", a.Limit, q, false); err != nil {
+ return err
+ }
+ }
+ }
+ return nil
+}
diff --git a/internal/resourcebudget/verify.go b/internal/resourcebudget/verify.go
new file mode 100644
index 000000000..1a9160ef4
--- /dev/null
+++ b/internal/resourcebudget/verify.go
@@ -0,0 +1,319 @@
+package resourcebudget
+
+import (
+ "database/sql"
+ "reflect"
+)
+
+// Verify replays event groups, checks each operation boundary and compares live rows.
+func Verify(tx *sql.Tx) error {
+ receipts := map[string]Receipt{}
+ rows, e := tx.Query(`SELECT id,body FROM rb_receipts`)
+ if e != nil {
+ return e
+ }
+ for rows.Next() {
+ var id, b string
+ if e = rows.Scan(&id, &b); e != nil {
+ rows.Close()
+ return e
+ }
+ var r Receipt
+ if decode(b, &r) != nil || r.ID != id {
+ rows.Close()
+ return ErrCorrupt
+ }
+ receipts[id] = r
+ }
+ e = rows.Err()
+ rows.Close()
+ if e != nil {
+ return e
+ }
+ latest := map[string]string{}
+ pools := map[string]Pool{}
+ limits := map[string]PaidLimit{}
+ rs := map[string]Reservation{}
+ closedOps := map[string]bool{}
+ previous := ""
+ check := func() error {
+ var list []Reservation
+ for _, r := range rs {
+ list = append(list, r)
+ }
+ return balances(pools, limits, list)
+ }
+ rows, e = tx.Query(`SELECT kind,id,body,epoch,operation_id FROM rb_events ORDER BY seq`)
+ if e != nil {
+ return e
+ }
+ for rows.Next() {
+ var table, id, b, op string
+ var epoch int64
+ if e = rows.Scan(&table, &id, &b, &epoch, &op); e != nil {
+ rows.Close()
+ return e
+ }
+ receipt, ok := receipts[op]
+ if !ok || receipt.Epoch != epoch {
+ rows.Close()
+ return ErrCorrupt
+ }
+ if op != previous {
+ if closedOps[op] {
+ rows.Close()
+ return ErrCorrupt
+ }
+ if e = check(); e != nil {
+ rows.Close()
+ return e
+ }
+ if previous != "" {
+ closedOps[previous] = true
+ }
+ previous = op
+ }
+ switch table {
+ case "pools":
+ var p Pool
+ if decode(b, &p) != nil || validateStored(table, id, &p) != nil || p.Account != receipt.Account || poolKey(p.Account, p.Month, p.Resource) != id {
+ rows.Close()
+ return ErrCorrupt
+ }
+ if old, ok := pools[id]; ok {
+ if old.Identity != p.Identity || old.Account != p.Account || old.Month != p.Month || old.Resource != p.Resource {
+ rows.Close()
+ return ErrCorrupt
+ }
+ if _, err := number(p.Capacity); err != nil {
+ rows.Close()
+ return err
+ }
+ if cmp(p.Capacity, old.Capacity) < 0 || p.Revision < old.Revision {
+ rows.Close()
+ return ErrCorrupt
+ }
+ if (p.Capacity != old.Capacity || p.Revision != old.Revision || p.Digest != old.Digest) && receipt.Kind != "IncreaseBasis" {
+ rows.Close()
+ return ErrCorrupt
+ }
+ } else if receipt.Kind != "InstallBasis" {
+ rows.Close()
+ return ErrCorrupt
+ }
+ pools[id] = p
+ case "limits":
+ var p PaidLimit
+ if decode(b, &p) != nil || validateStored(table, id, &p) != nil || p.Account != receipt.Account || limitKey(p.Account, p.Month, p.Resource, p.Consent, p.Subject) != id {
+ rows.Close()
+ return ErrCorrupt
+ }
+ if old, ok := limits[id]; ok {
+ old.Unused, old.Settled, old.Outstanding = p.Unused, p.Settled, p.Outstanding
+ if !reflect.DeepEqual(old, p) {
+ rows.Close()
+ return ErrCorrupt
+ }
+ } else if receipt.Kind != "InstallPaidLimit" {
+ rows.Close()
+ return ErrCorrupt
+ }
+ limits[id] = p
+ case "reservations":
+ var r Reservation
+ if decode(b, &r) != nil || validateStored(table, id, &r) != nil || r.Account != receipt.Account || r.Subject != receipt.Subject || r.Reservation != id {
+ rows.Close()
+ return ErrCorrupt
+ }
+ if old, ok := rs[id]; ok {
+ if !reflect.DeepEqual(old.ReserveRequest, r.ReserveRequest) || len(old.Allocations) != len(r.Allocations) {
+ rows.Close()
+ return ErrCorrupt
+ }
+ for i, a := range r.Allocations {
+ o := old.Allocations[i]
+ if _, e := number(a.Total); e != nil {
+ rows.Close()
+ return e
+ }
+ if cmp(a.Total, o.Total) < 0 {
+ rows.Close()
+ return ErrCorrupt
+ }
+
+ a.SettledIncluded, a.SettledPaid, a.Excess, a.Total = o.SettledIncluded, o.SettledPaid, o.Excess, o.Total
+ if a != o {
+ rows.Close()
+ return ErrCorrupt
+ }
+ }
+ valid := false
+ switch receipt.Kind {
+ case "Claim":
+ valid = old.State == "reserved" && r.State == "claimed"
+ case "CancelUnclaimed":
+ valid = old.State == "reserved" && r.State == "closed-unused"
+ case "MarkUncertain":
+ valid = (old.State == "claimed" || old.State == "uncertain") && r.State == "uncertain"
+ case "Observe":
+ valid = (old.State == "claimed" || old.State == "uncertain") && r.State == old.State
+ case "Finish":
+ valid = (old.State == "claimed" || old.State == "uncertain") && r.State == "closed"
+ }
+ if !valid {
+ rows.Close()
+ return ErrCorrupt
+ }
+ if receipt.Kind != "Claim" && (r.Producer != old.Producer || r.Attempt != old.Attempt || r.ProducerEpoch != old.ProducerEpoch) {
+ rows.Close()
+ return ErrCorrupt
+ }
+ } else if receipt.Kind != "Reserve" || r.State != "reserved" {
+ rows.Close()
+ return ErrCorrupt
+ }
+ rs[id] = r
+ case "basis", "evidence", "fences":
+ default:
+ rows.Close()
+ return ErrCorrupt
+ }
+ latest[key(table, id)] = b
+ }
+ e = rows.Err()
+ rows.Close()
+ if e != nil {
+ return e
+ }
+ if e = check(); e != nil {
+ return e
+ }
+ for _, table := range []string{"pools", "limits", "reservations", "fences", "evidence", "basis"} {
+ rows, e = tx.Query(`SELECT id,body FROM rb_` + table)
+ if e != nil {
+ return e
+ }
+ for rows.Next() {
+ var id, b string
+ if e = rows.Scan(&id, &b); e != nil {
+ rows.Close()
+ return e
+ }
+ k := key(table, id)
+ if latest[k] != b {
+ rows.Close()
+ return ErrCorrupt
+ }
+ delete(latest, k)
+ }
+ e = rows.Err()
+ rows.Close()
+ if e != nil {
+ return e
+ }
+ }
+ if len(latest) != 0 {
+ return ErrCorrupt
+ }
+ return nil
+}
+func balances(pools map[string]Pool, limits map[string]PaidLimit, reservations []Reservation) error {
+ type sums struct{ s, o, x Amount }
+ ps, ls := map[string]sums{}, map[string]sums{}
+ accum := func(m map[string]sums, k string, s, o, x Amount) error {
+ v, ok := m[k]
+ if !ok {
+ v = sums{Zero, Zero, Zero}
+ }
+ var err error
+ v.s, err = add(v.s, s)
+ if err != nil {
+ return err
+ }
+ v.o, err = add(v.o, o)
+ if err != nil {
+ return err
+ }
+ v.x, err = add(v.x, x)
+ if err != nil {
+ return err
+ }
+ m[k] = v
+ return nil
+ }
+ for _, r := range reservations {
+ for _, a := range r.Allocations {
+ for _, n := range []Amount{a.Maximum, a.Included, a.Paid, a.SettledIncluded, a.SettledPaid, a.Excess, a.Total} {
+ if _, e := number(n); e != nil {
+ return ErrCorrupt
+ }
+ }
+ bounded := min(a.Total, a.Maximum)
+ included := min(bounded, a.Included)
+ paid, _ := sub(bounded, included)
+ excess, _ := sub(a.Total, bounded)
+ if a.SettledIncluded != included || a.SettledPaid != paid || a.Excess != excess {
+ return ErrCorrupt
+ }
+ sum, e := add(a.Included, a.Paid)
+ if e != nil || sum != a.Maximum {
+ return ErrCorrupt
+ }
+ di, e := sub(a.Included, a.SettledIncluded)
+ if e != nil {
+ return e
+ }
+ dp, e := sub(a.Paid, a.SettledPaid)
+ if e != nil {
+ return e
+ }
+ if r.State == "closed" || r.State == "closed-unused" {
+ di, dp = Zero, Zero
+ }
+ pk := poolKey(r.Account, r.Month, a.Resource)
+ if _, ok := pools[pk]; !ok {
+ return ErrCorrupt
+ }
+ if e := accum(ps, pk, a.SettledIncluded, di, a.Excess); e != nil {
+ return e
+ }
+ if a.Limit != "" {
+ q, ok := limits[a.Limit]
+ if !ok || q.Account != r.Account || q.Month != r.Month || q.Subject != r.Subject || q.Resource != a.Resource || q.Consent != r.Consent {
+ return ErrCorrupt
+ }
+ if e := accum(ls, a.Limit, a.SettledPaid, dp, Zero); e != nil {
+ return e
+ }
+ }
+ }
+ }
+ compare := func(cap, avail, settled, outstanding, excess Amount, s sums) error {
+ if s.s == "" {
+ s = sums{Zero, Zero, Zero}
+ }
+ v, e := add(avail, settled)
+ if e != nil {
+ return e
+ }
+ v, e = add(v, outstanding)
+ if e != nil {
+ return e
+ }
+ if v != cap || settled != s.s || outstanding != s.o || excess != s.x {
+ return ErrCorrupt
+ }
+ return nil
+ }
+ for k, p := range pools {
+ if e := compare(p.Capacity, p.Available, p.Settled, p.Outstanding, p.Excess, ps[k]); e != nil {
+ return e
+ }
+ }
+ for k, p := range limits {
+ if e := compare(p.Ceiling, p.Unused, p.Settled, p.Outstanding, Zero, ls[k]); e != nil {
+ return e
+ }
+ }
+ return nil
+}