amadan

carlos / platform Public

Add snapshot-only admission attempt observation

claude bbeefbc51c071e166a67fbe5db62f24871a2dd98 parent ed7aa59
2 files changed, +633 −0
  • internal/usage/durabilityattemptobservation.go +189 −0
  • internal/usage/durabilityattemptobservation_test.go +444 −0

CI passed — run details

diff --git a/internal/usage/durabilityattemptobservation.go b/internal/usage/durabilityattemptobservation.go
new file mode 100644
index 000000000..f963036d5
--- /dev/null
+++ b/internal/usage/durabilityattemptobservation.go
@@ -0,0 +1,189 @@
+package usage
+
+import (
+ "context"
+ "database/sql"
+ "errors"
+ "slices"
+ "time"
+)
+
+var (
+ ErrObservationEpoch = errors.New("usage: observation_epoch")
+ ErrAdmissionObservationMalformed = errors.New("usage: admission_observation_malformed")
+)
+
+type AdmissionAttemptObservation struct {
+ Box, Epoch string
+ CurrentRun, Watermark int64
+ Admission AdmissionFacts
+ Fleet *FleetFacts
+ Attempt *AttemptFacts
+}
+
+type AdmissionFacts struct {
+ Token, FleetID, Account, App string
+ AdmittedAt, TransportDoneAt, RecorderStartedAt, RecorderDoneAt, SettledAt time.Time
+ How string
+ ParticipantTimesFilledByTerminal bool
+}
+
+type FleetFacts struct {
+ Box string
+ WitnessGeneration int64
+ Owners []string
+ OpenedAt time.Time
+ State, Predecessor, AcknowledgedBy string
+}
+
+type AttemptFacts struct {
+ Seq, Run, Client int64
+ Host, Purpose, Operation, Date, Hour string
+ StartedAt, FinishedAt time.Time
+ Answered bool
+ Status int
+ Abandoned, Late bool
+}
+
+// ObserveAdmissionAttempt reads persisted facts in one snapshot. It cannot prove
+// completeness, zero usage, or billability. ErrAdmissionUnknown means absent in
+// this snapshot; a late Admit writer may still land the token. A nil Attempt
+// means no association in this snapshot; until the row settles, StartAdmitted
+// may still add one. A nil Fleet means no journal row.
+//
+// It does not observe ops pending registrations or retry queues, l.pending,
+// meter holds, the failure or discovery latch, the lost-transport flag, or
+// per-attempt incompleteness. Consumers need Snapshot/Publish for the UTC dates
+// touched by these facts. Terminal-filled participant times are not callbacks;
+// the filled flag applies to the row. On recovered rows, which individual
+// participant times were filled is not recorded. A backwards response time
+// is preserved just as Finish recorded it.
+func (l *DurabilityLedger) ObserveAdmissionAttempt(epoch, token string) (AdmissionAttemptObservation, error) {
+ return l.observeAdmissionAttempt(epoch, token, nil)
+}
+
+func (l *DurabilityLedger) observeAdmissionAttempt(epoch, token string, observePause func()) (AdmissionAttemptObservation, error) {
+ var zero AdmissionAttemptObservation
+ if !durabilityEpochRE.MatchString(epoch) || epoch != l.epoch {
+ return zero, ErrObservationEpoch
+ }
+ if token == "" || len(token) > 128 {
+ return zero, ErrAdmissionObservationMalformed
+ }
+ l.mu.Lock()
+ closed := l.closed
+ l.mu.Unlock()
+ if closed {
+ return zero, errLedgerClosed
+ }
+ tx, err := l.rdb.BeginTx(context.Background(), &sql.TxOptions{ReadOnly: true})
+ if err != nil {
+ return zero, err
+ }
+ defer tx.Rollback() //nolint:errcheck
+ var out AdmissionAttemptObservation
+ if err = tx.QueryRow(`SELECT box, epoch, run, next_seq FROM ledger WHERE id = 1`).Scan(&out.Box, &out.Epoch, &out.CurrentRun, &out.Watermark); err != nil {
+ return zero, err
+ }
+ if observePause != nil {
+ observePause()
+ }
+ if out.Box != l.box || out.Epoch != epoch || out.CurrentRun < 1 || out.Watermark < 0 {
+ return zero, ErrAdmissionObservationMalformed
+ }
+ a, err := readAdmission(tx, token)
+ if err != nil {
+ return zero, err
+ }
+ if !observationAdmissionValid(a) {
+ return zero, ErrAdmissionObservationMalformed
+ }
+ out.Admission = AdmissionFacts{Token: a.Token, FleetID: a.FleetID, Account: a.Account, App: a.App, AdmittedAt: a.AdmittedAt, TransportDoneAt: a.TransportDoneAt, RecorderStartedAt: a.RecorderStartedAt, RecorderDoneAt: a.RecorderDoneAt, SettledAt: a.SettledAt, How: a.How, ParticipantTimesFilledByTerminal: a.How == AdmissionUnissued || a.How == AdmissionRecoveredUnknown}
+ var seq int64
+ err = tx.QueryRow(`SELECT attempt_seq FROM admission_attempts WHERE admission_token = ?`, token).Scan(&seq)
+ if err != nil && !errors.Is(err, sql.ErrNoRows) {
+ return zero, err
+ }
+ if err == nil {
+ var at AttemptFacts
+ var start int64
+ var finish, status, answered sql.NullInt64
+ var abandoned, late int
+ err = tx.QueryRow(`SELECT seq, run, client, operation, date, hour, started_at, finished_at, status, answered, abandoned, late FROM attempts WHERE seq = ?`, seq).Scan(&at.Seq, &at.Run, &at.Client, &at.Operation, &at.Date, &at.Hour, &start, &finish, &status, &answered, &abandoned, &late)
+ if errors.Is(err, sql.ErrNoRows) {
+ return zero, ErrAdmissionObservationMalformed
+ }
+ if err != nil {
+ return zero, err
+ }
+ var clientRun int64
+ var attr ReplicationAttribution
+ err = tx.QueryRow(`SELECT run, account, app, host, purpose FROM clients WHERE id = ?`, at.Client).Scan(&clientRun, &attr.Account, &attr.App, &attr.Host, &attr.Purpose)
+ if errors.Is(err, sql.ErrNoRows) {
+ return zero, ErrAdmissionObservationMalformed
+ }
+ if err != nil {
+ return zero, err
+ }
+ at.Host, at.Purpose = attr.Host, string(attr.Purpose)
+ at.StartedAt = time.Unix(0, start).UTC()
+ at.Abandoned, at.Late = abandoned == 1, late == 1
+ if attr.Validate() != nil || attr.Account != a.Account || attr.App != a.App || at.Run != clientRun || at.Run < 1 || at.Run > out.CurrentRun || at.Seq <= 0 || at.Seq > out.Watermark || !durabilityOperationRE.MatchString(at.Operation) || at.StartedAt.Before(a.AdmittedAt) || at.Date != DayOf(at.StartedAt) || at.Hour != HourLabel(at.StartedAt) || abandoned < 0 || abandoned > 1 || late < 0 || late > 1 {
+ return zero, ErrAdmissionObservationMalformed
+ }
+ if finish.Valid {
+ if at.Abandoned || !answered.Valid || !status.Valid || answered.Int64 < 0 || answered.Int64 > 1 || (answered.Int64 == 1 && (status.Int64 < 100 || status.Int64 > 599)) || (answered.Int64 == 0 && status.Int64 != 0) {
+ return zero, ErrAdmissionObservationMalformed
+ }
+ at.FinishedAt = time.Unix(0, finish.Int64).UTC()
+ at.Answered, at.Status = answered.Int64 == 1, int(status.Int64)
+ } else if status.Valid || answered.Valid || (at.Run < out.CurrentRun && !at.Abandoned) {
+ return zero, ErrAdmissionObservationMalformed
+ }
+ out.Attempt = &at
+ }
+ if (a.How == AdmissionUnissued && (out.Attempt != nil || !a.RecorderStartedAt.IsZero())) || (!a.RecorderStartedAt.IsZero() && out.Attempt == nil) || (a.How == AdmissionFinished && (out.Attempt == nil || out.Attempt.FinishedAt.IsZero())) {
+ return zero, ErrAdmissionObservationMalformed
+ }
+ f, err := readFleet(tx, a.FleetID)
+ if err != nil && !errors.Is(err, ErrFleetUnknown) {
+ return zero, err
+ }
+ if err == nil {
+ if f.Box != out.Box || !slices.Contains(f.Owners, a.Account+"/"+a.App) {
+ return zero, ErrAdmissionObservationMalformed
+ }
+ out.Fleet = &FleetFacts{Box: f.Box, WitnessGeneration: f.WitnessGeneration, Owners: f.Owners, OpenedAt: f.OpenedAt, State: f.State, Predecessor: f.Predecessor, AcknowledgedBy: f.AcknowledgedBy}
+ }
+ return out, nil
+}
+
+func observationAdmissionValid(a Admission) bool {
+ switch a.How {
+ case "", AdmissionFinished, AdmissionAbandoned, AdmissionUnrecorded, AdmissionUnissued, AdmissionRecoveredUnknown:
+ default:
+ return false
+ }
+ terminal := a.How == AdmissionUnissued || a.How == AdmissionRecoveredUnknown
+ if a.SettledAt.IsZero() {
+ if terminal || (a.RecorderDoneAt.IsZero() != (a.How == "")) {
+ return false
+ }
+ } else if a.TransportDoneAt.IsZero() || a.RecorderDoneAt.IsZero() || a.How == "" || a.SettledAt.Before(a.TransportDoneAt) || a.SettledAt.Before(a.RecorderDoneAt) {
+ return false
+ }
+ for _, at := range []time.Time{a.RecorderStartedAt, a.SettledAt} {
+ if !at.IsZero() && at.Before(a.AdmittedAt) {
+ return false
+ }
+ }
+ // Recovery fills missing completion times from its clock, which can go back.
+ if a.How != AdmissionRecoveredUnknown {
+ for _, at := range []time.Time{a.TransportDoneAt, a.RecorderDoneAt} {
+ if !at.IsZero() && at.Before(a.AdmittedAt) {
+ return false
+ }
+ }
+ }
+ return true
+}
diff --git a/internal/usage/durabilityattemptobservation_test.go b/internal/usage/durabilityattemptobservation_test.go
new file mode 100644
index 000000000..dcde996eb
--- /dev/null
+++ b/internal/usage/durabilityattemptobservation_test.go
@@ -0,0 +1,444 @@
+package usage
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "reflect"
+ "sort"
+ "strings"
+ "testing"
+ "time"
+)
+
+func observationSetup(t *testing.T) (*DurabilityLedger, *DurabilityClient, string, uint64) {
+ t.Helper()
+ l, _ := openAdmissionLedger(t)
+ c, err := l.Bind(tenantAttr)
+ if err != nil {
+ t.Fatal(err)
+ }
+ token := mustAdmit(t, l, "fleet", tenantAttr.Account, tenantAttr.App, l.now())
+ seq, err := c.StartAdmitted("PutObject", l.now(), token)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return l, c, token, seq
+}
+func observationRead(t *testing.T, l *DurabilityLedger, token string) AdmissionAttemptObservation {
+ t.Helper()
+ o, e := l.ObserveAdmissionAttempt(l.Epoch(), token)
+ if e != nil {
+ t.Fatal(e)
+ }
+ return o
+}
+func observationOK(t *testing.T, err error) {
+ t.Helper()
+ if err != nil {
+ t.Fatal(err)
+ }
+}
+
+func TestAdmissionObservationLifecycle(t *testing.T) {
+ l, _ := openAdmissionLedger(t)
+ token := "unknown"
+ o, e := l.ObserveAdmissionAttempt(l.Epoch(), token)
+ if !errors.Is(e, ErrAdmissionUnknown) || !reflect.DeepEqual(o, AdmissionAttemptObservation{}) {
+ t.Fatalf("unknown: %+v %v", o, e)
+ }
+ observationOK(t, l.Admit(token, "fleet", tenantAttr.Account, tenantAttr.App, l.now()))
+ o = observationRead(t, l, token)
+ if o.Attempt != nil || o.Fleet != nil || !o.Admission.SettledAt.IsZero() {
+ t.Fatalf("admit: %+v", o)
+ }
+ observationOK(t, l.Unissued(token, l.now()))
+ o = observationRead(t, l, token)
+ if o.Attempt != nil || o.Admission.How != AdmissionUnissued || !o.Admission.ParticipantTimesFilledByTerminal {
+ t.Fatalf("unissued: %+v", o)
+ }
+ for _, associated := range []bool{false, true} {
+ t.Run(fmt.Sprint(associated), func(t *testing.T) {
+ token := mustAdmit(t, l, "fleet", tenantAttr.Account, tenantAttr.App, l.now())
+ if associated {
+ c, e := l.Bind(tenantAttr)
+ observationOK(t, e)
+ _, e = c.StartAdmitted("PutObject", l.now(), token)
+ observationOK(t, e)
+ }
+ observationOK(t, l.RecorderDone(token, l.now(), AdmissionUnrecorded))
+ observationOK(t, l.TransportDone(token, l.now(), false))
+ o := observationRead(t, l, token)
+ if (o.Attempt != nil) != associated || o.Admission.SettledAt.IsZero() || o.Admission.ParticipantTimesFilledByTerminal || o.Admission.TransportDoneAt != l.now() {
+ t.Fatalf("unrecorded: %+v", o)
+ }
+ })
+ }
+}
+
+func TestAdmissionObservationParticipants(t *testing.T) {
+ for _, transportFirst := range []bool{false, true} {
+ t.Run(fmt.Sprint(transportFirst), func(t *testing.T) {
+ l, _, token, _ := observationSetup(t)
+ o := observationRead(t, l, token)
+ if o.Attempt == nil || !o.Admission.RecorderStartedAt.IsZero() {
+ t.Fatalf("start: %+v", o)
+ }
+ observationOK(t, l.RecorderStarted(token, l.now()))
+ transport := func() { observationOK(t, l.TransportDone(token, l.now(), false)) }
+ recorder := func() { observationOK(t, l.RecorderDone(token, l.now(), AdmissionUnrecorded)) }
+ if transportFirst {
+ transport()
+ } else {
+ recorder()
+ }
+ o = observationRead(t, l, token)
+ if !o.Admission.SettledAt.IsZero() || o.Admission.TransportDoneAt.IsZero() == transportFirst || o.Admission.RecorderDoneAt.IsZero() != transportFirst {
+ t.Fatalf("mid: %+v", o)
+ }
+ if transportFirst {
+ recorder()
+ } else {
+ transport()
+ }
+ if observationRead(t, l, token).Admission.SettledAt.IsZero() {
+ t.Fatal("not settled")
+ }
+ })
+ }
+}
+
+func TestAdmissionObservationSnapshot(t *testing.T) {
+ l, c, token, seq := observationSetup(t)
+ before := observationRead(t, l, token)
+ after, e := l.observeAdmissionAttempt(l.Epoch(), token, func() {
+ observationOK(t, c.Finish(seq, RequestOutcome{At: l.now(), Answered: true, Status: 200}))
+ observationOK(t, l.RecorderStarted(token, l.now()))
+ observationOK(t, l.RecorderDone(token, l.now(), AdmissionFinished))
+ observationOK(t, l.TransportDone(token, l.now(), false))
+ })
+ observationOK(t, e)
+ if !reflect.DeepEqual(before, after) {
+ t.Fatalf("snapshot changed: before %+v after %+v", before, after)
+ }
+ if observationRead(t, l, token).Attempt.FinishedAt.IsZero() {
+ t.Fatal("writer did not commit")
+ }
+}
+
+func TestAdmissionObservationRecovery(t *testing.T) {
+ for _, backwards := range []bool{false, true} {
+ t.Run(fmt.Sprint(backwards), func(t *testing.T) {
+ dir := t.TempDir()
+ clock := newLedgerClock(ledgerDay0.Add(9 * time.Hour))
+ l := openTestLedger(t, dir, "box", clock, true)
+ c, e := l.Bind(tenantAttr)
+ observationOK(t, e)
+ token := mustAdmit(t, l, "fleet", tenantAttr.Account, tenantAttr.App, clock.now())
+ _, e = c.StartAdmitted("PutObject", clock.now(), token)
+ observationOK(t, e)
+ epoch := l.Epoch()
+ l.abandon()
+ if backwards {
+ clock.add(-time.Hour)
+ } else {
+ clock.add(time.Hour)
+ }
+ l = openTestLedger(t, dir, "box", clock, true)
+ defer l.Close(context.Background(), nil)
+ o, e := l.ObserveAdmissionAttempt(epoch, token)
+ observationOK(t, e)
+ if o.Admission.How != AdmissionRecoveredUnknown || !o.Admission.ParticipantTimesFilledByTerminal || !o.Attempt.Abandoned || o.Attempt.Run >= o.CurrentRun {
+ t.Fatalf("recovery: %+v", o)
+ }
+ if backwards && (!o.Admission.TransportDoneAt.Before(o.Admission.AdmittedAt) || !o.Admission.RecorderDoneAt.Before(o.Admission.AdmittedAt)) {
+ t.Fatal("fixture did not exercise backward recovery")
+ }
+ })
+ }
+}
+
+func TestAdmissionObservationBackwardClocks(t *testing.T) {
+ l, c, token, seq := observationSetup(t)
+ observationOK(t, c.Finish(seq, RequestOutcome{At: l.now().Add(-time.Hour), Answered: true, Status: 200, Incompletes: 1}))
+ observationOK(t, l.RecorderStarted(token, l.now().Add(time.Hour)))
+ observationOK(t, l.RecorderDone(token, l.now(), AdmissionFinished))
+ observationOK(t, l.TransportDone(token, l.now(), false))
+ o := observationRead(t, l, token)
+ if !o.Attempt.FinishedAt.Before(o.Admission.AdmittedAt) || !o.Admission.SettledAt.Before(o.Admission.RecorderStartedAt) {
+ t.Fatal("backwards facts lost")
+ }
+}
+
+func TestAdmissionObservationKnownAbsences(t *testing.T) {
+ for _, dimension := range []string{"incompletes", "lost"} {
+ t.Run(dimension, func(t *testing.T) {
+ var observations []AdmissionAttemptObservation
+ for i := 0; i < 2; i++ {
+ l, c, token, seq := observationSetup(t)
+ n := 0
+ if dimension == "incompletes" && i == 0 {
+ n = 2
+ }
+ observationOK(t, c.Finish(seq, RequestOutcome{At: l.now(), Answered: true, Status: 200, Incompletes: n}))
+ observationOK(t, l.RecorderStarted(token, l.now()))
+ observationOK(t, l.RecorderDone(token, l.now(), AdmissionFinished))
+ observationOK(t, l.TransportDone(token, l.now(), dimension == "lost" && i == 0))
+ o := observationRead(t, l, token)
+ o.Epoch = ""
+ o.Admission.Token = ""
+ o.Watermark = 0
+ observations = append(observations, o)
+ }
+ if !reflect.DeepEqual(observations[0], observations[1]) {
+ t.Fatalf("unavailable fact exposed: %+v %+v", observations[0], observations[1])
+ }
+ })
+ }
+}
+
+func TestAdmissionObservationAssociation(t *testing.T) {
+ l, _, token, first := observationSetup(t)
+ attr := tenantAttr
+ attr.Purpose = PurposeCheckpoint
+ c, e := l.Bind(attr)
+ observationOK(t, e)
+ other := mustAdmit(t, l, "fleet", attr.Account, attr.App, l.now())
+ second, e := c.StartAdmitted("UploadPart", l.now(), other)
+ observationOK(t, e)
+ observationOK(t, c.Finish(second, RequestOutcome{At: l.now(), Answered: true, Status: 503}))
+ a, b := observationRead(t, l, token), observationRead(t, l, other)
+ if a.Attempt.Seq != int64(first) || b.Attempt.Seq != int64(second) || a.Attempt.Operation != "PutObject" || b.Attempt.Operation != "UploadPart" || b.Attempt.Purpose != string(PurposeCheckpoint) || b.Attempt.Status != 503 || !b.Attempt.Answered {
+ t.Fatalf("association: %+v %+v", a, b)
+ }
+}
+
+func TestAdmissionObservationEpochAndMethods(t *testing.T) {
+ l, _, token, _ := observationSetup(t)
+ for _, epoch := range []string{"", "bad", strings.Repeat("a", 16)} {
+ if epoch == l.Epoch() {
+ t.Fatal("fixture epoch collision")
+ }
+ o, e := l.ObserveAdmissionAttempt(epoch, token)
+ if !errors.Is(e, ErrObservationEpoch) || !reflect.DeepEqual(o, AdmissionAttemptObservation{}) {
+ t.Fatalf("epoch: %+v %v", o, e)
+ }
+ }
+ for _, v := range []any{AdmissionAttemptObservation{}, AdmissionFacts{}, AttemptFacts{}, FleetFacts{}} {
+ typ := reflect.TypeOf(v)
+ if typ.NumMethod() != 0 || reflect.PointerTo(typ).NumMethod() != 0 {
+ t.Fatalf("exported method on %v", typ)
+ }
+ }
+}
+
+func observationDump(t *testing.T, l *DurabilityLedger) string {
+ t.Helper()
+ rows, e := l.db.Query(`SELECT name FROM sqlite_master WHERE type='table' ORDER BY name`)
+ observationOK(t, e)
+ var tables []string
+ for rows.Next() {
+ var s string
+ observationOK(t, rows.Scan(&s))
+ tables = append(tables, s)
+ }
+ observationOK(t, rows.Err())
+ observationOK(t, rows.Close())
+ var all []string
+ for _, table := range tables {
+ rows, e = l.db.Query(`SELECT * FROM "` + strings.ReplaceAll(table, `"`, `""`) + `"`)
+ observationOK(t, e)
+ cols, e := rows.Columns()
+ observationOK(t, e)
+ for rows.Next() {
+ v := make([]any, len(cols))
+ ptr := make([]any, len(cols))
+ for i := range v {
+ ptr[i] = &v[i]
+ }
+ observationOK(t, rows.Scan(ptr...))
+ all = append(all, fmt.Sprintf("%s:%#v", table, v))
+ }
+ observationOK(t, rows.Err())
+ observationOK(t, rows.Close())
+ }
+ sort.Strings(all)
+ return strings.Join(all, "\n")
+}
+
+func TestAdmissionObservationNoWritesOrMutex(t *testing.T) {
+ l, c, token, seq := observationSetup(t)
+ observationOK(t, c.Finish(seq, RequestOutcome{At: l.now()}))
+ l.mu.Lock()
+ stateBefore := l.state
+ l.mu.Unlock()
+ generationBefore := l.disc().gen()
+ l.latch.mu.Lock()
+ failureGenerationBefore := l.latch.generation
+ l.latch.mu.Unlock()
+ before := observationDump(t, l)
+ writer, err := l.db.Conn(context.Background())
+ observationOK(t, err)
+ defer writer.Close()
+ paused, resume := make(chan struct{}), make(chan struct{})
+ result := make(chan error, 1)
+ go func() {
+ _, e := l.observeAdmissionAttempt(l.Epoch(), token, func() { close(paused); <-resume })
+ result <- e
+ }()
+ select {
+ case <-paused:
+ case <-time.After(3 * time.Second):
+ writer.Close()
+ close(resume)
+ <-result
+ t.Fatal("reader did not reach seam")
+ }
+ l.mu.Lock()
+ close(resume)
+ var e error
+ select {
+ case e = <-result:
+ case <-time.After(3 * time.Second):
+ l.mu.Unlock()
+ writer.Close()
+ <-result
+ t.Fatal("queries acquired ledger mutex or writer connection")
+ }
+ l.mu.Unlock()
+ observationOK(t, writer.Close())
+ observationOK(t, e)
+ l.latch.mu.Lock()
+ failureGenerationAfter := l.latch.generation
+ l.latch.mu.Unlock()
+ if failureGenerationBefore != failureGenerationAfter {
+ t.Fatal("failure latch generation changed")
+ }
+ l.mu.Lock()
+ stateAfter := l.state
+ l.mu.Unlock()
+ if !reflect.DeepEqual(stateBefore, stateAfter) || generationBefore != l.disc().gen() {
+ t.Fatal("observation changed in-memory ledger state")
+ }
+ if after := observationDump(t, l); before != after {
+ t.Fatal("observation wrote database")
+ }
+ observationOK(t, l.Close(context.Background(), nil))
+ _, e = l.ObserveAdmissionAttempt(l.Epoch(), token)
+ if !errors.Is(e, errLedgerClosed) {
+ t.Fatalf("closed: %v", e)
+ }
+}
+
+func TestAdmissionObservationMalformed(t *testing.T) {
+ fixtures := map[string]string{
+ "late": `UPDATE attempts SET late=2`,
+ "abandoned_range": `UPDATE attempts SET abandoned=2`,
+ "seq_watermark": `UPDATE ledger SET next_seq=0`,
+ "ledger_box": `UPDATE ledger SET box='wrong'`,
+ "ledger_epoch": `UPDATE ledger SET epoch='wrong'`,
+ "how": `UPDATE admissions SET how='other',recorder_done_at=admitted_at`,
+ "unissued_unsettled": `DELETE FROM admission_attempts; UPDATE admissions SET how='unissued',recorder_done_at=admitted_at`,
+ "recovered_unsettled": `UPDATE admissions SET how='recovered-unknown',recorder_done_at=admitted_at`,
+ "done_without_how": `UPDATE admissions SET recorder_done_at=admitted_at`,
+ "how_without_done": `UPDATE admissions SET how='unrecorded'`,
+ "settled_without_transport": `UPDATE admissions SET how='unrecorded',recorder_done_at=admitted_at,settled_at=admitted_at`,
+ "settled_without_recorder": `UPDATE admissions SET transport_done_at=admitted_at,settled_at=admitted_at,how='unrecorded'`,
+ "settled_without_how": `UPDATE admissions SET transport_done_at=admitted_at,recorder_done_at=admitted_at,settled_at=admitted_at`,
+ "transport_backdated": `UPDATE admissions SET transport_done_at=admitted_at-1`,
+ "recorder_start_backdated": `UPDATE admissions SET recorder_started_at=admitted_at-1`,
+ "recorder_done_backdated": `UPDATE admissions SET recorder_done_at=admitted_at-1,how='unrecorded'`,
+ "settled_backdated": `UPDATE admissions SET how='recovered-unknown',transport_done_at=admitted_at-2,recorder_done_at=admitted_at-2,settled_at=admitted_at-1`,
+ "settled_before_transport": `UPDATE admissions SET transport_done_at=admitted_at+1,recorder_done_at=admitted_at,settled_at=admitted_at,how='unrecorded'`,
+ "settled_before_recorder": `UPDATE admissions SET transport_done_at=admitted_at,recorder_done_at=admitted_at+1,settled_at=admitted_at,how='unrecorded'`,
+ "missing_attempt": `DELETE FROM attempts`,
+ "missing_client": `DELETE FROM clients`,
+ "account": `UPDATE clients SET account='other'`,
+ "app": `UPDATE clients SET app='other'`,
+ "client_run": `UPDATE clients SET run=run+1`,
+ "future_run": `UPDATE attempts SET run=run+1; UPDATE clients SET run=run+1`,
+ "attribution": `UPDATE clients SET host='bad host'`,
+ "purpose": `UPDATE clients SET purpose='unknown'`,
+ "operation": `UPDATE attempts SET operation='bad operation'`,
+ "started_backdated": `UPDATE attempts SET started_at=started_at-1`,
+ "date": `UPDATE attempts SET date='2020-01-01'`,
+ "hour": `UPDATE attempts SET hour='02'`,
+ "abandoned_finished": `UPDATE attempts SET abandoned=1,finished_at=started_at,status=200,answered=1`,
+ "answered_invalid": `UPDATE attempts SET finished_at=started_at,status=200,answered=2`,
+ "answered_null": `UPDATE attempts SET finished_at=started_at,status=200`,
+ "status_null": `UPDATE attempts SET finished_at=started_at,answered=1`,
+ "status_low": `UPDATE attempts SET finished_at=started_at,status=99,answered=1`,
+ "status_high": `UPDATE attempts SET finished_at=started_at,status=600,answered=1`,
+ "unanswered_status": `UPDATE attempts SET finished_at=started_at,status=200,answered=0`,
+ "unfinished_status": `UPDATE attempts SET status=0`,
+ "unfinished_answered": `UPDATE attempts SET answered=0`,
+ "old_unfinished": `UPDATE ledger SET run=run+1`,
+ "unissued_associated": `UPDATE admissions SET how='unissued',transport_done_at=admitted_at,recorder_done_at=admitted_at,settled_at=admitted_at`,
+ // The unissued-started clause is also implied by rules 5 and 6.
+ "unissued_started": `DELETE FROM admission_attempts; UPDATE admissions SET how='unissued',recorder_started_at=admitted_at,transport_done_at=admitted_at,recorder_done_at=admitted_at,settled_at=admitted_at`,
+ "started_without_association": `DELETE FROM admission_attempts; UPDATE admissions SET recorder_started_at=admitted_at`,
+ "finished_missing": `DELETE FROM admission_attempts; UPDATE admissions SET how='finished',recorder_done_at=admitted_at`,
+ "finished_unfinished": `UPDATE admissions SET how='finished',recorder_done_at=admitted_at`,
+ "fleet_box": `UPDATE fleets SET box='wrong'`,
+ "fleet_owner": `UPDATE fleets SET owners='other/app'`,
+ }
+ for name, statement := range fixtures {
+ t.Run(name, func(t *testing.T) {
+ l, _, token, _ := observationSetup(t)
+ observationOK(t, l.OpenFleet("fleet", "box", 7, []string{tenantAttr.Account + "/" + tenantAttr.App}, "", l.now()))
+ if o := observationRead(t, l, token); o.Fleet == nil || o.Fleet.WitnessGeneration != 7 {
+ t.Fatal("fleet missing")
+ }
+ _, e := l.db.Exec(statement)
+ observationOK(t, e)
+ before := observationDump(t, l)
+ o, e := l.ObserveAdmissionAttempt(l.Epoch(), token)
+ if !errors.Is(e, ErrAdmissionObservationMalformed) || !reflect.DeepEqual(o, AdmissionAttemptObservation{}) {
+ t.Fatalf("malformed: %+v %v", o, e)
+ }
+ if observationDump(t, l) != before {
+ t.Fatal("refusal wrote database")
+ }
+ })
+ }
+}
+
+func TestAdmissionObservationAllowedFinishedAbandonedOutcome(t *testing.T) {
+ l, c, token, seq := observationSetup(t)
+ observationOK(t, c.Finish(seq, RequestOutcome{At: l.now(), Answered: false}))
+ observationOK(t, l.RecorderStarted(token, l.now()))
+ observationOK(t, l.RecorderDone(token, l.now(), AdmissionAbandoned))
+ o := observationRead(t, l, token)
+ if o.Attempt.FinishedAt.IsZero() || o.Attempt.Answered || o.Attempt.Status != 0 || o.Admission.How != AdmissionAbandoned {
+ t.Fatalf("allowed facts: %+v", o)
+ }
+}
+
+func TestAdmissionObservationRunFromSnapshot(t *testing.T) {
+ l, _, token, _ := observationSetup(t)
+ var run int64
+ observationOK(t, l.db.QueryRow(`SELECT run FROM ledger WHERE id=1`).Scan(&run))
+ l.mu.Lock()
+ saved := l.state
+ l.state.run++
+ l.mu.Unlock()
+ defer func() { l.mu.Lock(); l.state = saved; l.mu.Unlock() }()
+ if o := observationRead(t, l, token); o.CurrentRun != run {
+ t.Fatalf("current run = %d, persisted %d", o.CurrentRun, run)
+ }
+}
+
+func TestAdmissionObservationTokenBoundary(t *testing.T) {
+ l, _ := openAdmissionLedger(t)
+ for _, token := range []string{"", strings.Repeat("x", 129)} {
+ o, e := l.ObserveAdmissionAttempt(l.Epoch(), token)
+ if !errors.Is(e, ErrAdmissionObservationMalformed) || !reflect.DeepEqual(o, AdmissionAttemptObservation{}) {
+ t.Fatalf("token: %+v %v", o, e)
+ }
+ }
+ token := strings.Repeat("x", 128)
+ observationOK(t, l.Admit(token, "fleet", tenantAttr.Account, tenantAttr.App, l.now()))
+ if observationRead(t, l, token).Admission.Token != token {
+ t.Fatal("accepted token changed")
+ }
+}