amadan

carlos / platform Public

Associate durability attempts with their admitted requests

claude 987f20c29173a4bcbd03d0fc784d51efbc8f176b parent 0e5267f
6 files changed, +394 −7
  • internal/lease/requestmeter.go +18 −1
  • internal/lease/requestmeter_admission_test.go +123 −0
  • internal/ops/litestream_admission.go +2 −0
  • internal/ops/litestream_metered_test.go +9 −5
  • internal/usage/durabilityattemptassociation_test.go +204 −0
  • internal/usage/durabilityledger.go +38 −1
diff --git a/internal/lease/requestmeter.go b/internal/lease/requestmeter.go
index 319daf928..f95ce7366 100644
--- a/internal/lease/requestmeter.go
+++ b/internal/lease/requestmeter.go
@@ -150,7 +150,7 @@ func meterAttempt(ctx context.Context, in middleware.DeserializeInput, next midd
// request path. The participant is never synthesised while its worker
// can still run.
token, recordErr, ran := a.call(slots, func() (uint64, error) {
- token, err := recorder.Start(operation, started)
+ token, err := startAdmittedAttempt(recorder, adm, operation, started)
if err == nil {
adm.RecorderStarted(started)
}
@@ -292,3 +292,20 @@ func (a *meteredAttempt) call(slots chan struct{}, fn func() (uint64, error), se
return value.token, value.err, true
}
}
+
+// A gate-backed attempt must carry its admission into the same durable Start.
+// An older recorder remains usable without a gate; with one, missing support
+// is a recording failure and holds the day through the existing meter path.
+func startAdmittedAttempt(recorder requestRecorder, admission Admission, operation string, at time.Time) (uint64, error) {
+ named, ok := admission.(interface{ AdmissionToken() string })
+ if !ok {
+ return recorder.Start(operation, at)
+ }
+ linked, ok := recorder.(interface {
+ StartAdmitted(string, time.Time, string) (uint64, error)
+ })
+ if !ok || named.AdmissionToken() == "" {
+ return 0, errRequestMeterRequired
+ }
+ return linked.StartAdmitted(operation, at, named.AdmissionToken())
+}
diff --git a/internal/lease/requestmeter_admission_test.go b/internal/lease/requestmeter_admission_test.go
new file mode 100644
index 000000000..f40403fcc
--- /dev/null
+++ b/internal/lease/requestmeter_admission_test.go
@@ -0,0 +1,123 @@
+package lease
+
+import (
+ "context"
+ "database/sql"
+ "net/http"
+ "path/filepath"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/usage"
+)
+
+type admissionRecorder struct{ *usage.DurabilityClient }
+
+func (r admissionRecorder) Finish(token uint64, out requestOutcome) error {
+ return r.DurabilityClient.Finish(token, usage.RequestOutcome{At: out.At, Status: out.Status, Answered: out.Answered, Incompletes: out.Incompletes})
+}
+
+type journalAdmission struct {
+ t *testing.T
+ l *usage.DurabilityLedger
+ token string
+}
+
+func (a journalAdmission) AdmissionToken() string { return a.token }
+func (a journalAdmission) RecorderStarted(at time.Time) {
+ a.t.Helper()
+ if err := a.l.RecorderStarted(a.token, at); err != nil {
+ a.t.Error(err)
+ }
+}
+func (a journalAdmission) RecorderDone(at time.Time, how string) {
+ a.t.Helper()
+ if err := a.l.RecorderDone(a.token, at, how); err != nil {
+ a.t.Error(err)
+ }
+}
+func (a journalAdmission) TransportDone(at time.Time) {
+ a.t.Helper()
+ if err := a.l.TransportDone(a.token, at, false); err != nil {
+ a.t.Error(err)
+ }
+}
+
+func TestRequestMeterAdmissionAssociationDurableBeforeSend(t *testing.T) {
+ for _, gateFirst := range []bool{false, true} {
+ name := "meter-option-first"
+ if gateFirst {
+ name = "gate-option-first"
+ }
+ t.Run(name, func(t *testing.T) {
+ dir := t.TempDir()
+ l, err := usage.OpenDurabilityLedger(dir, "box1", time.Now, usage.DurabilityLedgerOptions{})
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer l.Close(context.Background(), nil)
+ recorder, err := l.Bind(usage.ReplicationAttribution{Account: "bac", App: "one", Host: "one.example.com", Purpose: usage.PurposeReplicate})
+ if err != nil {
+ t.Fatal(err)
+ }
+ db, err := sql.Open("sqlite", filepath.Join(dir, "ledger.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer db.Close()
+ var token string
+ client := gatedTestClient(t, func(w http.ResponseWriter, r *http.Request) {
+ var n int
+ err := db.QueryRow(`SELECT COUNT(*) FROM admission_attempts link JOIN attempts a ON a.seq = link.attempt_seq JOIN admissions d ON d.token = link.admission_token WHERE d.token = ? AND a.operation = 'PutObject' AND d.account = 'bac' AND d.app = 'one'`, token).Scan(&n)
+ if err != nil || n != 1 {
+ t.Errorf("association before transport: count=%d error=%v", n, err)
+ }
+ w.WriteHeader(http.StatusOK)
+ }, func(context.Context) (Admission, error) {
+ var err error
+ token, err = usage.NewAdmissionToken()
+ if err != nil {
+ return nil, err
+ }
+ if err = l.Admit(token, "fleet1", "bac", "one", time.Now()); err != nil {
+ return nil, err
+ }
+ return journalAdmission{t, l, token}, nil
+ }, admissionRecorder{recorder}, gateFirst)
+ if err := meterPut(context.Background(), client); err != nil {
+ t.Fatal(err)
+ }
+ row, err := l.ReadAdmission(token)
+ if err != nil || !row.Settled() || row.How != usage.AdmissionFinished {
+ t.Fatalf("admission after transport: %+v %v", row, err)
+ }
+ })
+ }
+}
+
+func TestRequestMeterMissingAdmissionAssociationHoldsDay(t *testing.T) {
+ dir := t.TempDir()
+ l, err := usage.OpenDurabilityLedger(dir, "box1", time.Now, usage.DurabilityLedgerOptions{})
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer l.Close(context.Background(), nil)
+ token, _ := usage.NewAdmissionToken()
+ if err := l.Admit(token, "fleet1", "bac", "one", time.Now()); err != nil {
+ t.Fatal(err)
+ }
+ recorder := &memoryRequestRecorder{}
+ client := gatedTestClient(t, func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) }, func(context.Context) (Admission, error) { return journalAdmission{t, l, token}, nil }, recorder, true)
+ if err := meterPut(context.Background(), client); err != nil {
+ t.Fatal(err)
+ }
+ recorder.mu.Lock()
+ defer recorder.mu.Unlock()
+ if len(recorder.requests) != 0 || recorder.incomplete == 0 {
+ t.Fatalf("unsupported recorder lost the failure: rows=%d incomplete=%d", len(recorder.requests), recorder.incomplete)
+ }
+ a, err := l.ReadAdmission(token)
+ if err != nil || !a.Settled() || a.How != usage.AdmissionUnrecorded {
+ t.Fatalf("unrecorded outcome: %+v %v", a, err)
+ }
+}
diff --git a/internal/ops/litestream_admission.go b/internal/ops/litestream_admission.go
index 32f6963f6..3ad6e6189 100644
--- a/internal/ops/litestream_admission.go
+++ b/internal/ops/litestream_admission.go
@@ -51,6 +51,8 @@ type fleetAdmission struct {
done atomic.Bool // the transport finalised
}
+func (a *fleetAdmission) AdmissionToken() string { return a.token }
+
func (a *fleetAdmission) TransportDone(at time.Time) {
if !a.done.CompareAndSwap(false, true) {
return
diff --git a/internal/ops/litestream_metered_test.go b/internal/ops/litestream_metered_test.go
index a205ed98f..c907ed7bf 100644
--- a/internal/ops/litestream_metered_test.go
+++ b/internal/ops/litestream_metered_test.go
@@ -27,11 +27,12 @@ import (
type nopRecorder struct{}
-func (nopRecorder) Start(string, time.Time) (uint64, error) { return 1, nil }
-func (nopRecorder) Finish(uint64, lease.RequestOutcome) error { return nil }
-func (nopRecorder) Abandon(uint64, time.Time) {}
-func (nopRecorder) Incomplete(time.Time) {}
-func (nopRecorder) Hold(time.Time) func() { return func() {} }
+func (nopRecorder) Start(string, time.Time) (uint64, error) { return 1, nil }
+func (nopRecorder) StartAdmitted(string, time.Time, string) (uint64, error) { return 1, nil }
+func (nopRecorder) Finish(uint64, lease.RequestOutcome) error { return nil }
+func (nopRecorder) Abandon(uint64, time.Time) {}
+func (nopRecorder) Incomplete(time.Time) {}
+func (nopRecorder) Hold(time.Time) func() { return func() {} }
// countingRecorder counts what it is asked to start and finish and
// remembers each day it is told is incomplete.
@@ -48,6 +49,9 @@ func (c *countingRecorder) Start(string, time.Time) (uint64, error) {
c.started++
return uint64(c.started), nil
}
+func (c *countingRecorder) StartAdmitted(op string, at time.Time, admission string) (uint64, error) {
+ return c.Start(op, at)
+}
func (c *countingRecorder) Finish(uint64, lease.RequestOutcome) error {
c.mu.Lock()
defer c.mu.Unlock()
diff --git a/internal/usage/durabilityattemptassociation_test.go b/internal/usage/durabilityattemptassociation_test.go
new file mode 100644
index 000000000..88570f60a
--- /dev/null
+++ b/internal/usage/durabilityattemptassociation_test.go
@@ -0,0 +1,204 @@
+package usage
+
+import (
+ "context"
+ "errors"
+ "testing"
+ "time"
+)
+
+func attemptAssociationCounts(t *testing.T, l *DurabilityLedger) (attempts, associations int) {
+ t.Helper()
+ if err := l.db.QueryRow(`SELECT count(*) FROM attempts`).Scan(&attempts); err != nil {
+ t.Fatal(err)
+ }
+ if err := l.db.QueryRow(`SELECT count(*) FROM admission_attempts`).Scan(&associations); err != nil {
+ t.Fatal(err)
+ }
+ return attempts, associations
+}
+
+func ledgerNextSeq(t *testing.T, l *DurabilityLedger) int64 {
+ t.Helper()
+ var seq int64
+ if err := l.db.QueryRow(`SELECT next_seq FROM ledger WHERE id = 1`).Scan(&seq); err != nil {
+ t.Fatal(err)
+ }
+ return seq
+}
+
+func TestDurabilityAttemptAssociationSurvivesRestart(t *testing.T) {
+ dir := t.TempDir()
+ clock := newLedgerClock(ledgerDay0.Add(9 * time.Hour))
+ l := openTestLedger(t, dir, "box", clock, true)
+ c, err := l.Bind(tenantAttr)
+ if err != nil {
+ t.Fatal(err)
+ }
+ admission := mustAdmit(t, l, "fleet", tenantAttr.Account, tenantAttr.App, clock.now())
+ attempt, err := c.StartAdmitted("PutObject", clock.now(), admission)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := c.Finish(attempt, RequestOutcome{At: clock.now(), Status: 200, Answered: true}); err != nil {
+ t.Fatal(err)
+ }
+ if err := l.RecorderStarted(admission, clock.now()); err != nil {
+ t.Fatal(err)
+ }
+ if err := l.RecorderDone(admission, clock.now(), AdmissionFinished); err != nil {
+ t.Fatal(err)
+ }
+ if err := l.TransportDone(admission, clock.now(), false); err != nil {
+ t.Fatal(err)
+ }
+ if err := l.Close(context.Background(), nil); err != nil {
+ t.Fatal(err)
+ }
+ handoff := l.HandoffToken()
+ clock.add(time.Minute)
+ l2 := openTestLedgerWith(t, dir, "box", clock, DurabilityLedgerOptions{AllClientsInstrumented: true, HandoffToken: handoff})
+ defer l2.Close(context.Background(), nil)
+ var gotAttempt int64
+ if err := l2.db.QueryRow(`SELECT attempt_seq FROM admission_attempts WHERE admission_token = ?`, admission).Scan(&gotAttempt); err != nil {
+ t.Fatal(err)
+ }
+ if gotAttempt != int64(attempt) {
+ t.Fatalf("association attempt = %d, want %d", gotAttempt, attempt)
+ }
+ var finished int64
+ if err := l2.db.QueryRow(`SELECT finished_at FROM attempts WHERE seq = ?`, gotAttempt).Scan(&finished); err != nil {
+ t.Fatal(err)
+ }
+ if finished == 0 {
+ t.Fatal("associated attempt reopened unfinished")
+ }
+ a, err := l2.ReadAdmission(admission)
+ if err != nil || !a.Settled() || a.How != AdmissionFinished {
+ t.Fatalf("reopened admission = %+v, %v", a, err)
+ }
+}
+
+func TestDurabilityLegacyStartHasNoAdmissionAssociation(t *testing.T) {
+ l, _ := openAdmissionLedger(t)
+ c, err := l.Bind(tenantAttr)
+ if err != nil {
+ t.Fatal(err)
+ }
+ first, err := c.Start("PutObject", l.now())
+ if err != nil {
+ t.Fatal(err)
+ }
+ if attempts, associations := attemptAssociationCounts(t, l); attempts != 1 || associations != 0 {
+ t.Fatalf("legacy Start: attempts=%d associations=%d", attempts, associations)
+ }
+ if err := c.Finish(first, RequestOutcome{At: l.now(), Status: 200, Answered: true}); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := c.Start("HeadObject", l.now()); err != nil {
+ t.Fatalf("legacy Start after Finish: %v", err)
+ }
+}
+
+func TestDurabilityStartAdmittedRefusalsAddNoAttempt(t *testing.T) {
+ l, _ := openAdmissionLedger(t)
+ c, err := l.Bind(tenantAttr)
+ if err != nil {
+ t.Fatal(err)
+ }
+ admitted := l.now()
+ crossOwner := mustAdmit(t, l, "fleet", "other", tenantAttr.App, admitted)
+ crossApp := mustAdmit(t, l, "fleet", tenantAttr.Account, "other", admitted)
+ backdated := mustAdmit(t, l, "fleet", tenantAttr.Account, tenantAttr.App, admitted)
+ settled := mustAdmit(t, l, "fleet", tenantAttr.Account, tenantAttr.App, admitted)
+ if err := l.Unissued(settled, admitted); err != nil {
+ t.Fatal(err)
+ }
+ for _, tc := range []struct {
+ name string
+ token string
+ at time.Time
+ want error
+ }{
+ {name: "nonexistent", token: "missing", at: admitted, want: ErrAdmissionUnknown},
+ {name: "cross owner", token: crossOwner, at: admitted, want: ErrAdmissionEdge},
+ {name: "cross app", token: crossApp, at: admitted, want: ErrAdmissionEdge},
+ {name: "backdated", token: backdated, at: admitted.Add(-time.Second), want: ErrAdmissionBackdated},
+ {name: "settled", token: settled, at: admitted, want: ErrAdmissionSettled},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ before := ledgerNextSeq(t, l)
+ if _, err := c.StartAdmitted("PutObject", tc.at, tc.token); !errors.Is(err, tc.want) {
+ t.Fatalf("StartAdmitted error = %v, want %v", err, tc.want)
+ }
+ if attempts, associations := attemptAssociationCounts(t, l); attempts != 0 || associations != 0 {
+ t.Fatalf("refusal wrote attempts=%d associations=%d", attempts, associations)
+ }
+ if after := ledgerNextSeq(t, l); after != before {
+ t.Fatalf("refusal advanced next_seq from %d to %d", before, after)
+ }
+ })
+ }
+}
+
+func TestDurabilityAttemptAssociationFailureRollsBackAttemptAndSequence(t *testing.T) {
+ l, _ := openAdmissionLedger(t)
+ c, err := l.Bind(tenantAttr)
+ if err != nil {
+ t.Fatal(err)
+ }
+ admission := mustAdmit(t, l, "fleet", tenantAttr.Account, tenantAttr.App, l.now())
+ before := ledgerNextSeq(t, l)
+ if _, err := l.db.Exec(`CREATE TRIGGER refuse_attempt_association BEFORE INSERT ON admission_attempts BEGIN SELECT RAISE(ABORT, 'forced association failure'); END`); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := c.StartAdmitted("PutObject", l.now(), admission); err == nil {
+ t.Fatal("StartAdmitted succeeded through the association trigger")
+ }
+ if attempts, associations := attemptAssociationCounts(t, l); attempts != 0 || associations != 0 {
+ t.Fatalf("failed association left attempts=%d associations=%d", attempts, associations)
+ }
+ if after := ledgerNextSeq(t, l); after != before {
+ t.Fatalf("failed association advanced next_seq from %d to %d", before, after)
+ }
+ if _, err := l.db.Exec(`DROP TRIGGER refuse_attempt_association`); err != nil {
+ t.Fatal(err)
+ }
+ attempt, err := c.StartAdmitted("PutObject", l.now(), admission)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if int64(attempt) != before+1 {
+ t.Fatalf("attempt after rollback = %d, want next sequence %d", attempt, before+1)
+ }
+}
+
+func TestDurabilityAdmissionTokenCannotAssociateWithTwoAttempts(t *testing.T) {
+ l, _ := openAdmissionLedger(t)
+ c, err := l.Bind(tenantAttr)
+ if err != nil {
+ t.Fatal(err)
+ }
+ admission := mustAdmit(t, l, "fleet", tenantAttr.Account, tenantAttr.App, l.now())
+ first, err := c.StartAdmitted("PutObject", l.now(), admission)
+ if err != nil {
+ t.Fatal(err)
+ }
+ before := ledgerNextSeq(t, l)
+ if _, err := c.StartAdmitted("UploadPart", l.now(), admission); err == nil {
+ t.Fatal("one admission token associated with a second attempt")
+ }
+ if attempts, associations := attemptAssociationCounts(t, l); attempts != 1 || associations != 1 {
+ t.Fatalf("duplicate association left attempts=%d associations=%d", attempts, associations)
+ }
+ if after := ledgerNextSeq(t, l); after != before {
+ t.Fatalf("duplicate association advanced next_seq from %d to %d", before, after)
+ }
+ var got int64
+ if err := l.db.QueryRow(`SELECT attempt_seq FROM admission_attempts WHERE admission_token = ?`, admission).Scan(&got); err != nil {
+ t.Fatal(err)
+ }
+ if got != int64(first) {
+ t.Fatalf("duplicate association replaced attempt %d with %d", first, got)
+ }
+}
diff --git a/internal/usage/durabilityledger.go b/internal/usage/durabilityledger.go
index 0d5ff581f..f22ee0d14 100644
--- a/internal/usage/durabilityledger.go
+++ b/internal/usage/durabilityledger.go
@@ -577,6 +577,10 @@ var ledgerSchema = []string{
abandoned INTEGER NOT NULL DEFAULT 0,
late INTEGER NOT NULL DEFAULT 0
)`,
+ `CREATE TABLE IF NOT EXISTS admission_attempts (
+ admission_token TEXT PRIMARY KEY,
+ attempt_seq INTEGER NOT NULL UNIQUE
+ )`,
`CREATE INDEX IF NOT EXISTS attempts_date ON attempts(date)`,
`CREATE INDEX IF NOT EXISTS attempts_open ON attempts(finished_at) WHERE finished_at IS NULL`,
`CREATE TABLE IF NOT EXISTS failures (
@@ -1009,6 +1013,18 @@ func (c *DurabilityClient) Attribution() ReplicationAttribution { return c.attr
// is NOT on disk — the caller marks the request incomplete and sends it
// anyway. After Close it is refused and the dirty marker is written.
func (c *DurabilityClient) Start(operation string, at time.Time) (uint64, error) {
+ return c.start(operation, at, "")
+}
+
+// StartAdmitted commits the attempt and its admission association together.
+func (c *DurabilityClient) StartAdmitted(operation string, at time.Time, admission string) (uint64, error) {
+ if admission == "" {
+ return 0, ErrAdmissionUnknown
+ }
+ return c.start(operation, at, admission)
+}
+
+func (c *DurabilityClient) start(operation string, at time.Time, admission string) (uint64, error) {
if !durabilityOperationRE.MatchString(operation) {
return 0, fmt.Errorf("usage ledger: %q is not an SDK operation name", operation)
}
@@ -1030,7 +1046,22 @@ func (c *DurabilityClient) Start(operation string, at time.Time) (uint64, error)
return 0, err
}
defer tx.Rollback() //nolint:errcheck
- before := l.state // restored if the transaction fails
+ if admission != "" {
+ a, err := readAdmission(tx, admission)
+ if err != nil {
+ return 0, err
+ }
+ if a.Settled() {
+ return 0, ErrAdmissionSettled
+ }
+ if a.Account != c.attr.Account || a.App != c.attr.App {
+ return 0, ErrAdmissionEdge
+ }
+ if at.Before(a.AdmittedAt) {
+ return 0, ErrAdmissionBackdated
+ }
+ }
+ before := l.state // restored if the transaction fails
date := DayOf(at)
var closed int
if err := tx.QueryRow(`SELECT closed FROM days WHERE date = ?`, date).Scan(&closed); err != nil && !errors.Is(err, sql.ErrNoRows) {
@@ -1043,6 +1074,12 @@ func (c *DurabilityClient) Start(operation string, at time.Time) (uint64, error)
l.state = before
return 0, err
}
+ if admission != "" {
+ if _, err := tx.Exec(`INSERT INTO admission_attempts (admission_token, attempt_seq) VALUES (?, ?)`, admission, seq); err != nil {
+ l.state = before
+ return 0, err
+ }
+ }
if err := touchDay(tx, date, seq); err != nil {
l.state = before
return 0, err