From 8c86adaab2dd15467baff9380627a6b6bd7b7a25 Mon Sep 17 00:00:00 2001 From: Jorge Manrubia Date: Thu, 17 Sep 2026 06:47:06 +0200 Subject: [PATCH 1/3] Hand intake's ledger to admission: revision, verdict columns, the queue handoff Admission was built against its own Records and Ledger interfaces; intake's ledger had no revision to guard a verdict with and nowhere to keep one. The ledger now implements both: migration 4 adds the revision and the verdict columns, a verdict commits through the ledger's one state-changing write in a transaction that decides admitted or queued, and RunAdmission takes from the queue intake offers to. --- internal/connector/admission/commit.go | 2 +- internal/connector/admission/doc.go | 15 +- internal/connector/admission/event.go | 2 +- internal/connector/admission/run.go | 8 +- internal/connector/intake.go | 6 +- internal/connector/ledger.go | 38 ++ internal/connector/ledger_admission.go | 253 ++++++++ internal/connector/ledger_admission_test.go | 633 ++++++++++++++++++++ internal/connector/ledger_events.go | 195 +++++- 9 files changed, 1108 insertions(+), 44 deletions(-) create mode 100644 internal/connector/ledger_admission.go create mode 100644 internal/connector/ledger_admission_test.go diff --git a/internal/connector/admission/commit.go b/internal/connector/admission/commit.go index f8d356055..13a3cbce0 100644 --- a/internal/connector/admission/commit.go +++ b/internal/connector/admission/commit.go @@ -13,7 +13,7 @@ import ( var ErrAlreadyDecided = errors.New("admission: event already decided") // Ledger is the durable half of a verdict. Intake's SQLite ledger implements -// it once basecamp-cli PR 729 merges; until then only tests do. +// it (internal/connector, Ledger.Admission). type Ledger interface { // Commit writes v onto its event's record in one transaction and returns // the state it wrote. Within that transaction it must: diff --git a/internal/connector/admission/doc.go b/internal/connector/admission/doc.go index 1167b0fb0..0e6626566 100644 --- a/internal/connector/admission/doc.go +++ b/internal/connector/admission/doc.go @@ -73,13 +73,14 @@ // // # The seam with intake // -// Intake (basecamp-cli PR 729, internal/connector) hands admission an event id -// through its Queue; the record behind that id is the ledger row, whose -// pointer fields are exactly Event's. Run is written against two small -// interfaces, IDSource (satisfied by that Queue's Take) and Records (a ledger -// read returning a seen record as an Event), plus Ledger, whose Commit -// contract is invariant 5. The adapter from intake's ledger to those -// interfaces lands once 729 merges; this package imports nothing from it. +// Intake (internal/connector) hands admission an event id through its Queue; +// the record behind that id is the ledger row, whose pointer fields are +// exactly Event's. Run is written against two small interfaces, IDSource +// (satisfied by that Queue's Take) and Records (a ledger read returning a seen +// record as an Event), plus Ledger, whose Commit contract is invariant 5. +// Intake's ledger implements Records and Ledger (connector.Ledger.Admission), +// and connector.RunAdmission wires Run onto the queue; this package imports +// nothing from that one. // // # What is not here // diff --git a/internal/connector/admission/event.go b/internal/connector/admission/event.go index 1a19431ff..bcd23a0f6 100644 --- a/internal/connector/admission/event.go +++ b/internal/connector/admission/event.go @@ -10,7 +10,7 @@ import ( const ActorTypeAgent = "agent" // Event is the pointer admission decides on: the fields intake stores for a -// seen record (internal/connector Record in basecamp-cli PR 729), and nothing +// seen record (internal/connector's Record), and nothing // else. It carries no title, no content, no URL and no names; whatever // admission needs beyond ids it reads. type Event struct { diff --git a/internal/connector/admission/run.go b/internal/connector/admission/run.go index 421d94f02..37fcf079b 100644 --- a/internal/connector/admission/run.go +++ b/internal/connector/admission/run.go @@ -15,8 +15,8 @@ import ( // DefaultWorkers is the admission fetcher pool size. const DefaultWorkers = 4 -// IDSource hands over the next event id to admit. Intake's Queue (basecamp-cli -// PR 729, internal/connector) satisfies it. +// IDSource hands over the next event id to admit. Intake's Queue +// (internal/connector) satisfies it. type IDSource interface { Take(ctx context.Context) (int64, error) } @@ -27,8 +27,8 @@ type IDSource interface { // false when the id is unknown or the record is past deciding, so it is // skipped rather than decided twice. // -// This is the seam where intake hands admission an event: the adapter over -// intake's Ledger.Get lands once PR 729 merges. +// This is the seam where intake hands admission an event; intake's ledger +// implements it (internal/connector, Ledger.Admission). type Records interface { LoadUndecided(ctx context.Context, id int64) (ev Event, ok bool, err error) } diff --git a/internal/connector/intake.go b/internal/connector/intake.go index e47d2e5b3..21708f154 100644 --- a/internal/connector/intake.go +++ b/internal/connector/intake.go @@ -855,9 +855,9 @@ func (in *Intake) sweepStranded(ctx context.Context) { // so that a crash can never lose the pointer — but that ordering means a // failure after the commit leaves an event the ledger's own dedupe will // suppress on every retry. Nothing may be left in that state, so a start -// offers every record nothing has judged yet. Offering one twice costs -// nothing: the queue carries ids, and admission moves a record out of seen -// before it acts. +// offers every record nothing has judged yet. Offering one twice costs at +// most a second decision, never a second verdict: the queue carries ids, and +// admission's commit applies only at the revision its decision loaded. // // The pointer line is NOT written again here. A record that got no line when // it was recorded — the write that failed — never gets one, and the work still diff --git a/internal/connector/ledger.go b/internal/connector/ledger.go index 0a9589bd3..814dc83c3 100644 --- a/internal/connector/ledger.go +++ b/internal/connector/ledger.go @@ -281,6 +281,44 @@ WHEN OLD.state IN ('completed', 'discarded') AND NEW.state <> OLD.state BEGIN SELECT RAISE(ABORT, 'a terminal record cannot change state'); END; +`, + // Migration 4. What admission decides is written onto the record it + // decided, in the transaction that decides it. + // + // revision is the guard on that write: a decision carries the revision + // the record was loaded at, and applies only while the record is still + // there, so one event gets one verdict however many fetches decide it and + // an older decision never overwrites a newer one. Every state change bumps + // it, not only admission's. + // + // decided_at is when the latest verdict was written, blocked_at when the + // record entered its current run of blocked verdicts (the retry window + // counts from it), and retry_at a throttled verdict's server deadline. + // The rest is the verdict itself — what dispatch starts a task from, and + // what a blocked(no_route) record's holding reply needs. snapshot is the + // recording's content as admission read it, and only an admitted record + // carries one; nothing else in the ledger is content. + // + // Nothing has shipped that wrote a version 3 ledger, but a migration is + // how the schema changes regardless: the next time something has, this is + // the path that has to work. + ` +ALTER TABLE events ADD COLUMN revision INTEGER NOT NULL DEFAULT 0; +ALTER TABLE events ADD COLUMN decided_at TEXT; +ALTER TABLE events ADD COLUMN blocked_at TEXT; +ALTER TABLE events ADD COLUMN retry_at TEXT; +ALTER TABLE events ADD COLUMN trigger_name TEXT NOT NULL DEFAULT ''; +ALTER TABLE events ADD COLUMN acknowledge INTEGER NOT NULL DEFAULT 0; +ALTER TABLE events ADD COLUMN conversation_key TEXT NOT NULL DEFAULT ''; +ALTER TABLE events ADD COLUMN reply_kind TEXT NOT NULL DEFAULT ''; +ALTER TABLE events ADD COLUMN reply_recording_id INTEGER NOT NULL DEFAULT 0; +ALTER TABLE events ADD COLUMN routed INTEGER NOT NULL DEFAULT 0; +ALTER TABLE events ADD COLUMN route TEXT NOT NULL DEFAULT ''; +ALTER TABLE events ADD COLUMN class TEXT NOT NULL DEFAULT ''; +ALTER TABLE events ADD COLUMN recording_url TEXT NOT NULL DEFAULT ''; +ALTER TABLE events ADD COLUMN requester_id INTEGER NOT NULL DEFAULT 0; +ALTER TABLE events ADD COLUMN snapshot BLOB; +CREATE INDEX events_conversation ON events (conversation_key, state) WHERE conversation_key <> ''; `, } diff --git a/internal/connector/ledger_admission.go b/internal/connector/ledger_admission.go new file mode 100644 index 000000000..3bacf61a0 --- /dev/null +++ b/internal/connector/ledger_admission.go @@ -0,0 +1,253 @@ +package connector + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "fmt" + "log/slog" + + "github.com/basecamp/basecamp-cli/internal/connector/admission" + "github.com/basecamp/basecamp-cli/internal/connector/ndjson" +) + +// Admission is the ledger as admission sees it: the Records it loads events +// from and the Ledger its verdicts are committed to. +// +// Admission is its own package and imports nothing from this one. This is the +// whole of the join: the pointer intake recorded goes out as an admission +// Event with the revision it was loaded at, and a verdict comes back through +// the ledger's one state-changing write, so the lifecycle that keeps a +// finished record finished is the same lifecycle a verdict meets. +type Admission struct { + ledger *Ledger +} + +var ( + _ admission.Records = Admission{} + _ admission.Ledger = Admission{} +) + +// Admission returns the ledger's admission seam. +func (l *Ledger) Admission() Admission { return Admission{ledger: l} } + +// undecided is where a record may be for admission to decide it: seen, which +// nothing has judged yet, and blocked, which recovery and redispatch decide +// again. Everything past these was decided by admission or moved on by +// dispatch, and is never decided a second time. +var undecided = []RecordState{StateSeen, StateBlocked} + +// LoadUndecided loads a seen or blocked record as the event admission decides, +// with the revision it was loaded at. An unknown id, or a record past +// deciding, is not ok and is skipped. +func (a Admission) LoadUndecided(ctx context.Context, id int64) (admission.Event, bool, error) { + record, ok, err := a.ledger.Get(ctx, id) + if err != nil || !ok { + return admission.Event{}, false, err + } + if record.State != StateSeen && record.State != StateBlocked { + return admission.Event{}, false, nil + } + return admission.Event{ + ID: record.ID, + EventType: record.EventType, + BucketID: record.BucketID, + RecordingID: record.RecordingID, + CreatorID: record.CreatorID, + PerformedByID: record.PerformedByID, + ActorType: record.ActorType, + Details: record.Details, + Revision: record.Revision, + SeenAt: record.SeenAt, + }, true, nil +} + +// Commit writes a verdict onto its record in one transaction, and returns the +// state it wrote. It holds admission.Ledger's contract: +// +// - the write applies only while the record is still at the verdict's +// revision and still undecided, and it bumps the revision; otherwise it +// is admission.ErrAlreadyDecided; +// - an admitted verdict is written as queued when its conversation is +// live, decided inside the transaction: another record on the key is +// admitted and not yet dispatched (it becomes the task), dispatched (the +// task is running), or queued behind one of those; +// - content is written only with an admitted verdict; +// - a throttled verdict keeps its RetryAt. +func (a Admission) Commit(ctx context.Context, v admission.Verdict) (admission.State, error) { + state, err := verdictState(v) + if err != nil { + return "", err + } + var snapshot []byte + if v.Snapshot != nil { + if snapshot, err = json.Marshal(v.Snapshot); err != nil { + return "", fmt.Errorf("connector: commit verdict on %d: %w", v.EventID, err) + } + } + + var written RecordState + err = retryBusy(func() error { + var err error + written, err = a.commit(ctx, v, state, snapshot) + return err + }) + if err != nil { + return "", err + } + return admission.State(written), nil +} + +func (a Admission) commit(ctx context.Context, v admission.Verdict, state RecordState, snapshot []byte) (RecordState, error) { + l := a.ledger + tx, err := l.db.BeginTx(ctx, nil) + if err != nil { + return "", fmt.Errorf("connector: begin verdict on %d: %w", v.EventID, err) + } + defer func() { _ = tx.Rollback() }() + + if state == StateAdmitted { + // The record being decided is seen or blocked, so it never counts + // itself as the conversation's task. + var live bool + if err := tx.QueryRowContext(ctx, ` +SELECT EXISTS ( + SELECT 1 FROM events + WHERE conversation_key = ? AND state IN (?, ?, ?) +)`, v.ConversationKey, string(StateAdmitted), string(StateQueued), string(StateDispatched)).Scan(&live); err != nil { + return "", fmt.Errorf("connector: read conversation of %d: %w", v.EventID, err) + } + if live { + state = StateQueued + } + } + + now := l.timestamp() + var retryAt any + if state == StateBlocked && !v.RetryAt.IsZero() { + retryAt = stamp(v.RetryAt) + } + var blockedAt assignment + if state == StateBlocked { + // The retry window counts from the first of a run of blocked + // verdicts, so a record re-blocked every ten minutes still ages out + // of it. Read as the row was before this write. + blockedAt = assignment{column: "blocked_at", expr: "CASE WHEN state = 'blocked' AND blocked_at IS NOT NULL THEN blocked_at ELSE ? END", value: now} + } else { + blockedAt = assignment{column: "blocked_at", value: nil} + } + reply := admission.ReplyDestination{} + if v.Reply != nil { + reply = *v.Reply + } + revision := v.Revision + moved, err := l.move(ctx, tx, transition{ + id: v.EventID, + state: state, + reason: string(v.Reason), + from: undecided, + revision: &revision, + set: []assignment{ + {column: "decided_at", value: now}, + blockedAt, + {column: "retry_at", value: retryAt}, + {column: "trigger_name", value: string(v.Trigger)}, + {column: "acknowledge", value: v.Acknowledge}, + {column: "conversation_key", value: v.ConversationKey}, + {column: "reply_kind", value: string(reply.Kind)}, + {column: "reply_recording_id", value: reply.RecordingID}, + {column: "routed", value: v.Routed}, + {column: "route", value: v.Route}, + {column: "class", value: v.Class}, + {column: "recording_url", value: v.RecordingURL}, + {column: "requester_id", value: v.RequesterID}, + {column: "snapshot", value: snapshot}, + }, + }) + if err != nil { + return "", err + } + if !moved { + return "", explainVerdictRefusal(ctx, tx, v) + } + if err := tx.Commit(); err != nil { + return "", fmt.Errorf("connector: commit verdict on %d: %w", v.EventID, err) + } + return state, nil +} + +// verdictState checks a verdict is one the ledger can write and maps it to +// the ledger's state. Queued is the ledger's decision, never a verdict's. +func verdictState(v admission.Verdict) (RecordState, error) { + refuse := func(why string) (RecordState, error) { + return "", fmt.Errorf("connector: verdict on %d refused: %s", v.EventID, why) + } + switch v.State { + case admission.StateAdmitted: + if v.Snapshot == nil { + return refuse("an admitted verdict carries the recording it admitted") + } + if v.ConversationKey == "" { + return refuse("an admitted verdict needs a conversation key, or it can never queue") + } + return StateAdmitted, nil + case admission.StateBlocked, admission.StateDiscarded: + if v.Snapshot != nil { + return refuse("only an admitted verdict carries content") + } + return RecordState(v.State), nil + default: + return refuse(fmt.Sprintf("%q is not a state admission decides", v.State)) + } +} + +// explainVerdictRefusal says why a verdict changed nothing, reading the row in +// the verdict's own transaction so the answer is the state that refused it. +func explainVerdictRefusal(ctx context.Context, tx dbtx, v admission.Verdict) error { + var ( + state string + revision int64 + ) + switch err := tx.QueryRowContext(ctx, `SELECT state, revision FROM events WHERE id = ?`, v.EventID).Scan(&state, &revision); { + case errors.Is(err, sql.ErrNoRows): + return fmt.Errorf("connector: verdict on %d: %w", v.EventID, ErrNoSuchRecord) + case err != nil: + return fmt.Errorf("connector: verdict on %d: %w", v.EventID, err) + } + // Moved on since the decision loaded it, by another decision or by + // dispatch: the newer state stands. + return fmt.Errorf("connector: verdict on %d loaded at revision %d, record is %s at revision %d: %w", + v.EventID, v.Revision, state, revision, admission.ErrAlreadyDecided) +} + +// AdmissionOptions wires admission onto intake's queue and ledger. +type AdmissionOptions struct { + Ledger *Ledger + Queue *Queue + Admitter *admission.Admitter + // Workers is the fetcher pool size; admission's default when zero. + Workers int + // Lines is the writer intake's pointer lines go through. Verdict lines + // share it, so the two never tear each other on one stdout. + Lines *ndjson.Writer + Logger *slog.Logger +} + +// RunAdmission takes the ids intake hands over and decides each against the +// ledger, until ctx ends. It returns what admission.Run returns. +func RunAdmission(ctx context.Context, opts AdmissionOptions) error { + if opts.Ledger == nil || opts.Queue == nil { + return errors.New("connector: admission needs the ledger and the queue intake writes to") + } + records := opts.Ledger.Admission() + return admission.Run(ctx, admission.RunOptions{ + Source: opts.Queue, + Records: records, + Admitter: opts.Admitter, + Committer: admission.NewCommitter(records), + Workers: opts.Workers, + LineWriter: opts.Lines, + Logger: opts.Logger, + }) +} diff --git a/internal/connector/ledger_admission_test.go b/internal/connector/ledger_admission_test.go new file mode 100644 index 000000000..1786c84dd --- /dev/null +++ b/internal/connector/ledger_admission_test.go @@ -0,0 +1,633 @@ +package connector + +import ( + "bytes" + "context" + "encoding/base64" + "encoding/json" + "strconv" + "strings" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/basecamp/basecamp-sdk/go/pkg/basecamp" + + "github.com/basecamp/basecamp-cli/internal/connector/admission" + "github.com/basecamp/basecamp-cli/internal/connector/ndjson" +) + +// The join between intake's ledger and admission. Admission's own tests hold +// its contract against a fake; these hold the real ledger to the same one. + +const ( + adapterAgentID int64 = 52007412 + adapterOperatorID int64 = 26909558 + adapterBucketID int64 = 48699913 +) + +func seenRecord(t *testing.T, ledger *Ledger, id int64) Record { + t.Helper() + ctx := context.Background() + _, err := ledger.RecordSeen(ctx, testEvent(id), LanePoll) + require.NoError(t, err) + record, ok, err := ledger.Get(ctx, id) + require.NoError(t, err) + require.True(t, ok) + return record +} + +func getRecord(t *testing.T, ledger *Ledger, id int64) Record { + t.Helper() + record, ok, err := ledger.Get(context.Background(), id) + require.NoError(t, err) + require.True(t, ok) + return record +} + +func admittedVerdict(id, revision int64, key string) admission.Verdict { + return admission.Verdict{ + EventID: id, + EventType: "comment.created", + BucketID: adapterBucketID, + RecordingID: 10304028972, + Revision: revision, + RequesterID: adapterOperatorID, + State: admission.StateAdmitted, + Trigger: admission.TriggerMentioned, + Acknowledge: true, + ConversationKey: key, + Reply: &admission.ReplyDestination{Kind: admission.ReplyComment, RecordingID: 10304028989}, + Routed: true, + Route: "/work/connector", + Class: "internal", + RecordingURL: "https://app.basecamp.com/2914079/buckets/48699913/recordings/10304028972", + Snapshot: &admission.Snapshot{ + Type: "Comment", + Title: "A comment", + AppURL: "https://app.basecamp.com/2914079/buckets/48699913/recordings/10304028972", + Content: "
please look
", + UpdatedAt: time.Date(2026, 9, 16, 10, 0, 0, 0, time.UTC), + }, + } +} + +func blockedVerdict(id, revision int64, reason admission.Reason) admission.Verdict { + return admission.Verdict{ + EventID: id, EventType: "comment.created", BucketID: adapterBucketID, + RecordingID: 10304028972, Revision: revision, RequesterID: adapterOperatorID, + State: admission.StateBlocked, Reason: reason, + } +} + +func TestAdmissionLoadsAnUndecidedRecordAtItsRevision(t *testing.T) { + ledger := newTestLedger(t) + ctx := context.Background() + seen := seenRecord(t, ledger, 1) + store := ledger.Admission() + + ev, ok, err := store.LoadUndecided(ctx, 1) + require.NoError(t, err) + require.True(t, ok) + assert.Equal(t, admission.Event{ + ID: 1, EventType: "comment.created", BucketID: adapterBucketID, RecordingID: 10304028972, + CreatorID: adapterOperatorID, Revision: 0, SeenAt: seen.SeenAt, + }, ev) + assert.False(t, ev.SeenAt.IsZero(), "admission dates membership refusals from SeenAt") + + // A blocked record is decided again, at the revision its verdict left. + _, err = store.Commit(ctx, blockedVerdict(1, 0, admission.ReasonReadFailed)) + require.NoError(t, err) + ev, ok, err = store.LoadUndecided(ctx, 1) + require.NoError(t, err) + require.True(t, ok) + assert.Equal(t, int64(1), ev.Revision) + + _, ok, err = store.LoadUndecided(ctx, 404) + require.NoError(t, err) + assert.False(t, ok, "an unknown id is skipped") +} + +func TestAdmissionSkipsARecordPastDeciding(t *testing.T) { + for _, state := range []RecordState{StateAdmitted, StateQueued, StateDispatched, StateCompleted, StateDiscarded} { + t.Run(string(state), func(t *testing.T) { + ledger := newTestLedger(t) + ctx := context.Background() + seenRecord(t, ledger, 1) + switch state { + case StateDiscarded: + require.NoError(t, ledger.SetState(ctx, 1, StateDiscarded, "untrusted_author")) + case StateCompleted: + require.NoError(t, reachTerminal(t, ledger, 7, StateCompleted)) + require.NoError(t, ledger.SetState(ctx, 1, StateAdmitted, "")) + require.NoError(t, ledger.SetState(ctx, 1, StateDispatched, "")) + require.NoError(t, ledger.SetState(ctx, 1, StateCompleted, "")) + case StateDispatched: + require.NoError(t, ledger.SetState(ctx, 1, StateAdmitted, "")) + require.NoError(t, ledger.SetState(ctx, 1, StateDispatched, "")) + default: + require.NoError(t, ledger.SetState(ctx, 1, state, "")) + } + + _, ok, err := ledger.Admission().LoadUndecided(ctx, 1) + require.NoError(t, err) + assert.False(t, ok) + }) + } +} + +func TestAdmissionCommitWritesTheVerdictOntoTheRecord(t *testing.T) { + ledger := newTestLedger(t) + ctx := context.Background() + at := time.Date(2026, 9, 17, 9, 0, 0, 0, time.UTC) + ledger.now = func() time.Time { return at } + seenRecord(t, ledger, 1) + v := admittedVerdict(1, 0, "recording:10304028989") + + written, err := ledger.Admission().Commit(ctx, v) + require.NoError(t, err) + assert.Equal(t, admission.StateAdmitted, written) + + record := getRecord(t, ledger, 1) + assert.Equal(t, StateAdmitted, record.State) + assert.Empty(t, record.Reason) + assert.Equal(t, int64(1), record.Revision) + d := record.Decision + require.NotNil(t, d.DecidedAt) + assert.True(t, at.Equal(*d.DecidedAt)) + assert.Nil(t, d.BlockedAt) + assert.Nil(t, d.RetryAt) + assert.Equal(t, "mentioned", d.Trigger) + assert.True(t, d.Acknowledge) + assert.Equal(t, "recording:10304028989", d.ConversationKey) + assert.Equal(t, "comment", d.ReplyKind) + assert.Equal(t, int64(10304028989), d.ReplyRecordingID) + assert.True(t, d.Routed) + assert.Equal(t, "/work/connector", d.Route) + assert.Equal(t, "internal", d.Class) + assert.Equal(t, v.RecordingURL, d.RecordingURL) + assert.Equal(t, adapterOperatorID, d.RequesterID) + + var snapshot admission.Snapshot + require.NoError(t, json.Unmarshal(d.Snapshot, &snapshot)) + assert.Equal(t, *v.Snapshot, snapshot) +} + +// Invariant 5 of admission: one verdict per event. The same decision arriving +// twice — two fetchers, a restart — is written once. +func TestAdmissionCommitIsOneVerdictPerEvent(t *testing.T) { + ledger := newTestLedger(t) + ctx := context.Background() + seenRecord(t, ledger, 1) + store := ledger.Admission() + + _, err := store.Commit(ctx, admittedVerdict(1, 0, "recording:9")) + require.NoError(t, err) + _, err = store.Commit(ctx, admittedVerdict(1, 0, "recording:9")) + + require.ErrorIs(t, err, admission.ErrAlreadyDecided) + record := getRecord(t, ledger, 1) + assert.Equal(t, StateAdmitted, record.State, "not re-queued behind itself") + assert.Equal(t, int64(1), record.Revision) +} + +func TestAdmissionAnOlderDecisionNeverOverwritesANewer(t *testing.T) { + ledger := newTestLedger(t) + ctx := context.Background() + seenRecord(t, ledger, 1) + store := ledger.Admission() + + // Two decisions loaded the record at revision 0; the blocked one lands + // first. + _, err := store.Commit(ctx, blockedVerdict(1, 0, admission.ReasonReadFailed)) + require.NoError(t, err) + + _, err = store.Commit(ctx, admittedVerdict(1, 0, "recording:9")) + require.ErrorIs(t, err, admission.ErrAlreadyDecided) + record := getRecord(t, ledger, 1) + assert.Equal(t, StateBlocked, record.State, "a blocked verdict stands against an older decision too") + assert.Equal(t, "read_failed", record.Reason) + assert.Empty(t, record.Decision.Snapshot) + + // A decision loaded after it applies. + written, err := store.Commit(ctx, admittedVerdict(1, 1, "recording:9")) + require.NoError(t, err) + assert.Equal(t, admission.StateAdmitted, written) + assert.Empty(t, getRecord(t, ledger, 1).Reason) +} + +// Every state change bumps the revision, not only admission's: a decision +// loaded before anything else moved the record is stale. +func TestAdmissionAnyMoveStalesALoadedDecision(t *testing.T) { + ledger := newTestLedger(t) + ctx := context.Background() + seenRecord(t, ledger, 1) + store := ledger.Admission() + ev, ok, err := store.LoadUndecided(ctx, 1) + require.NoError(t, err) + require.True(t, ok) + + require.NoError(t, ledger.SetState(ctx, 1, StateBlocked, "no_route")) + + _, err = store.Commit(ctx, admittedVerdict(1, ev.Revision, "recording:9")) + require.ErrorIs(t, err, admission.ErrAlreadyDecided) + assert.Equal(t, StateBlocked, getRecord(t, ledger, 1).State) +} + +// A record admission already decided, or dispatch has moved on, is never +// decided again — even by a verdict carrying its current revision. +func TestAdmissionNeverDecidesARecordPastDeciding(t *testing.T) { + cases := map[string]func(*testing.T, *Ledger){ + "admitted": func(t *testing.T, l *Ledger) { + require.NoError(t, l.SetState(context.Background(), 1, StateAdmitted, "")) + }, + "dispatched": func(t *testing.T, l *Ledger) { + require.NoError(t, l.SetState(context.Background(), 1, StateAdmitted, "")) + require.NoError(t, l.SetState(context.Background(), 1, StateDispatched, "")) + }, + "discarded": func(t *testing.T, l *Ledger) { + require.NoError(t, l.SetState(context.Background(), 1, StateDiscarded, "by_operator")) + }, + } + for name, reach := range cases { + t.Run(name, func(t *testing.T) { + ledger := newTestLedger(t) + ctx := context.Background() + seenRecord(t, ledger, 1) + reach(t, ledger) + before := getRecord(t, ledger, 1) + + for _, v := range []admission.Verdict{ + admittedVerdict(1, before.Revision, "recording:9"), + blockedVerdict(1, before.Revision, admission.ReasonReadFailed), + } { + _, err := ledger.Admission().Commit(ctx, v) + require.ErrorIs(t, err, admission.ErrAlreadyDecided) + } + after := getRecord(t, ledger, 1) + assert.Equal(t, before.State, after.State) + assert.Equal(t, before.Revision, after.Revision) + }) + } +} + +func TestAdmissionCommitOnAMissingRecord(t *testing.T) { + _, err := newTestLedger(t).Admission().Commit(context.Background(), blockedVerdict(404, 0, admission.ReasonReadFailed)) + assert.ErrorIs(t, err, ErrNoSuchRecord) + assert.NotErrorIs(t, err, admission.ErrAlreadyDecided) +} + +// Admitted or queued is the ledger's decision, taken against the conversation +// as it stands in the verdict's own transaction. +func TestAdmissionQueuesBehindALiveConversation(t *testing.T) { + const key = "recording:10304028989" + cases := []struct { + name string + other func(*testing.T, *Ledger) + want admission.State + }{ + {"nothing else on the key", nil, admission.StateAdmitted}, + {"an admitted record becomes the task", func(t *testing.T, l *Ledger) { + _, err := l.Admission().Commit(context.Background(), admittedVerdict(2, 0, key)) + require.NoError(t, err) + }, admission.StateQueued}, + {"a dispatched record is a running task", func(t *testing.T, l *Ledger) { + _, err := l.Admission().Commit(context.Background(), admittedVerdict(2, 0, key)) + require.NoError(t, err) + require.NoError(t, l.SetState(context.Background(), 2, StateDispatched, "")) + }, admission.StateQueued}, + {"a queued record waits on a live task", func(t *testing.T, l *Ledger) { + _, err := l.Admission().Commit(context.Background(), admittedVerdict(2, 0, key)) + require.NoError(t, err) + _, err = l.Admission().Commit(context.Background(), admittedVerdict(3, 0, key)) + require.NoError(t, err) + require.NoError(t, l.SetState(context.Background(), 2, StateDispatched, "")) + require.NoError(t, l.SetState(context.Background(), 2, StateCompleted, "")) + }, admission.StateQueued}, + {"a completed task is not live", func(t *testing.T, l *Ledger) { + _, err := l.Admission().Commit(context.Background(), admittedVerdict(2, 0, key)) + require.NoError(t, err) + require.NoError(t, l.SetState(context.Background(), 2, StateDispatched, "")) + require.NoError(t, l.SetState(context.Background(), 2, StateCompleted, "")) + }, admission.StateAdmitted}, + {"a blocked record is not a task", func(t *testing.T, l *Ledger) { + v := admittedVerdict(2, 0, key) + v.State, v.Reason, v.Snapshot = admission.StateBlocked, admission.ReasonNoRoute, nil + _, err := l.Admission().Commit(context.Background(), v) + require.NoError(t, err) + }, admission.StateAdmitted}, + {"a live task on another conversation", func(t *testing.T, l *Ledger) { + _, err := l.Admission().Commit(context.Background(), admittedVerdict(2, 0, "recording:1")) + require.NoError(t, err) + }, admission.StateAdmitted}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + ledger := newTestLedger(t) + seenRecord(t, ledger, 1) + seenRecord(t, ledger, 2) + seenRecord(t, ledger, 3) + if tc.other != nil { + tc.other(t, ledger) + } + + written, err := ledger.Admission().Commit(context.Background(), admittedVerdict(1, 0, key)) + require.NoError(t, err) + assert.Equal(t, tc.want, written) + assert.Equal(t, RecordState(tc.want), getRecord(t, ledger, 1).State) + + // Through the committer too, which reports what the ledger wrote. + }) + } +} + +// Only an admitted verdict carries content, and a verdict the ledger refuses +// writes nothing at all. +func TestAdmissionRefusesAMalformedVerdictWithoutWriting(t *testing.T) { + blockedWithContent := blockedVerdict(1, 0, admission.ReasonNoRoute) + blockedWithContent.Snapshot = admittedVerdict(1, 0, "k").Snapshot + discardedWithContent := blockedWithContent + discardedWithContent.State, discardedWithContent.Reason = admission.StateDiscarded, admission.ReasonStale + admittedWithout := admittedVerdict(1, 0, "recording:9") + admittedWithout.Snapshot = nil + admittedNoKey := admittedVerdict(1, 0, "") + queued := admittedVerdict(1, 0, "recording:9") + queued.State, queued.Snapshot = admission.StateQueued, nil + blockedNoReason := blockedVerdict(1, 0, "") + + for name, v := range map[string]admission.Verdict{ + "blocked with content": blockedWithContent, + "discarded with content": discardedWithContent, + "admitted without content": admittedWithout, + "admitted without a key": admittedNoKey, + "queued is not a verdict": queued, + "blocked without a reason": blockedNoReason, + } { + t.Run(name, func(t *testing.T) { + ledger := newTestLedger(t) + seenRecord(t, ledger, 1) + + _, err := ledger.Admission().Commit(context.Background(), v) + + require.Error(t, err) + assert.NotErrorIs(t, err, admission.ErrAlreadyDecided) + record := getRecord(t, ledger, 1) + assert.Equal(t, StateSeen, record.State) + assert.Equal(t, int64(0), record.Revision) + assert.Nil(t, record.Decision.DecidedAt) + assert.Empty(t, record.Decision.Snapshot) + }) + } +} + +// A throttled verdict keeps the server's deadline, and the retry window counts +// from the first of a run of blocked verdicts. +func TestAdmissionKeepsTheBlockedScheduleInputs(t *testing.T) { + ledger := newTestLedger(t) + ctx := context.Background() + store := ledger.Admission() + t0 := time.Date(2026, 9, 17, 9, 0, 0, 0, time.UTC) + ledger.now = func() time.Time { return t0 } + seenRecord(t, ledger, 1) + + throttled := blockedVerdict(1, 0, admission.ReasonThrottled) + throttled.RetryAt = t0.Add(7 * time.Minute) + _, err := store.Commit(ctx, throttled) + require.NoError(t, err) + d := getRecord(t, ledger, 1).Decision + require.NotNil(t, d.RetryAt) + assert.True(t, throttled.RetryAt.Equal(*d.RetryAt)) + require.NotNil(t, d.BlockedAt) + assert.True(t, t0.Equal(*d.BlockedAt)) + + // Re-run ten minutes later and blocked again, on a read this time. + t1 := t0.Add(10 * time.Minute) + ledger.now = func() time.Time { return t1 } + _, err = store.Commit(ctx, blockedVerdict(1, 1, admission.ReasonReadFailed)) + require.NoError(t, err) + d = getRecord(t, ledger, 1).Decision + assert.Nil(t, d.RetryAt, "only a throttled verdict has a deadline") + require.NotNil(t, d.BlockedAt) + assert.True(t, t0.Equal(*d.BlockedAt), "the window still counts from the first block") + require.NotNil(t, d.DecidedAt) + assert.True(t, t1.Equal(*d.DecidedAt), "the last attempt is the latest verdict") + + // NextBlockedRetry reads what the ledger kept. + next, ok := admission.NextBlockedRetry(admission.Reason(getRecord(t, ledger, 1).Reason), *d.BlockedAt, *d.DecidedAt, time.Time{}) + require.True(t, ok) + assert.True(t, t1.Add(admission.BlockedRetryInterval).Equal(next)) + + // Admitted at last: no longer blocked, and a deadline is a blocked + // record's alone. + admitted := admittedVerdict(1, 2, "recording:9") + admitted.RetryAt = t1.Add(time.Hour) + _, err = store.Commit(ctx, admitted) + require.NoError(t, err) + d = getRecord(t, ledger, 1).Decision + assert.Nil(t, d.BlockedAt) + assert.Nil(t, d.RetryAt) +} + +// Two fetchers deciding one event at once: exactly one verdict is written. +func TestAdmissionConcurrentDecisionsWriteOne(t *testing.T) { + ledger := newTestLedger(t) + seenRecord(t, ledger, 1) + store := ledger.Admission() + + const racers = 8 + results := make([]error, racers) + var wg sync.WaitGroup + for i := range racers { + wg.Go(func() { + v := admittedVerdict(1, 0, "recording:9") + if i%2 == 1 { + v = blockedVerdict(1, 0, admission.ReasonReadFailed) + } + _, results[i] = store.Commit(context.Background(), v) + }) + } + wg.Wait() + + applied := 0 + for _, err := range results { + if err == nil { + applied++ + continue + } + require.ErrorIs(t, err, admission.ErrAlreadyDecided) + } + assert.Equal(t, 1, applied) + assert.Equal(t, int64(1), getRecord(t, ledger, 1).Revision) +} + +// Retention takes the verdict's content with the pointer's. +func TestDropContentTakesTheVerdictToo(t *testing.T) { + ledger := newTestLedger(t) + ctx := context.Background() + at := time.Date(2026, 9, 17, 9, 0, 0, 0, time.UTC) + ledger.now = func() time.Time { return at } + seenRecord(t, ledger, 1) + _, err := ledger.Admission().Commit(ctx, admittedVerdict(1, 0, "recording:9")) + require.NoError(t, err) + require.NoError(t, ledger.SetState(ctx, 1, StateDispatched, "")) + require.NoError(t, ledger.SetState(ctx, 1, StateCompleted, "")) + + dropped, err := ledger.DropContent(ctx, at.Add(time.Hour), at.Add(time.Hour)) + require.NoError(t, err) + require.Equal(t, 1, dropped) + + record := getRecord(t, ledger, 1) + assert.Equal(t, StateCompleted, record.State) + assert.Equal(t, Decision{DecidedAt: record.Decision.DecidedAt}, record.Decision, + "only the tombstone's timestamps survive") +} + +// Migration 4 carries a ledger written before it: its records load at +// revision 0 and are decided like any other. +func TestMigrationFourCarriesAnEarlierLedger(t *testing.T) { + path := t.TempDir() + "/state/connector.db" + all := migrations + migrations = all[:3] + old, err := OpenLedger(path) + migrations = all + require.NoError(t, err) + _, err = old.RecordSeen(context.Background(), testEvent(1), LanePoll) + require.NoError(t, err) + require.NoError(t, old.Close()) + + ledger, err := OpenLedger(path) + require.NoError(t, err) + t.Cleanup(func() { _ = ledger.Close() }) + version, err := ledger.SchemaVersion(context.Background()) + require.NoError(t, err) + assert.Equal(t, 4, version) + + ev, ok, err := ledger.Admission().LoadUndecided(context.Background(), 1) + require.NoError(t, err) + require.True(t, ok) + assert.Equal(t, int64(0), ev.Revision) + _, err = ledger.Admission().Commit(context.Background(), admittedVerdict(1, ev.Revision, "recording:9")) + require.NoError(t, err) + assert.Equal(t, StateAdmitted, getRecord(t, ledger, 1).State) +} + +// The handoff end to end: intake records and offers, admission takes from the +// same queue, decides against the real ledger, and reports on the shared +// writer without content. +func TestRunAdmissionDecidesWhatIntakeHandsOver(t *testing.T) { + ledger := newTestLedger(t) + queue, err := NewQueue(10, 100) + require.NoError(t, err) + var out bytes.Buffer + lines := ndjson.NewWriter(&out) + + const secret = "the instruction itself" + reads := &adapterReads{summaries: map[int64]*basecamp.RecordingSummary{}} + for _, id := range []int64{501, 502} { + reads.summaries[id] = &basecamp.RecordingSummary{ + ID: id, Status: "active", Type: "Todo", Title: "A to-do", + AppURL: "https://app.basecamp.com/2914079/buckets/48699913/todos/" + strconv.FormatInt(id, 10), + Bucket: &basecamp.Bucket{ID: adapterBucketID}, + Creator: &basecamp.Person{ID: adapterOperatorID}, + Content: mentionMarkup(adapterAgentID) + secret, + MentionedPersonIDs: []int64{adapterAgentID}, + UpdatedAt: time.Date(2026, 9, 17, 9, 0, 0, 0, time.UTC), + } + } + admitter, err := admission.NewAdmitter(admission.Policy{ + AgentID: adapterAgentID, + Trust: admission.Trust{Mode: admission.TrustOperator, OperatorID: adapterOperatorID}, + Projects: map[int64]admission.Route{adapterBucketID: {Path: "/work/connector", Class: "internal"}}, + }, admission.Reads{Summaries: reads, Subscriptions: reads, Assignments: reads}) + require.NoError(t, err) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan error, 1) + go func() { + done <- RunAdmission(ctx, AdmissionOptions{Ledger: ledger, Queue: queue, Admitter: admitter, Workers: 2, Lines: lines}) + }() + + // Two to-dos, then a third event on the first one's conversation: a + // to-do's conversation is itself, so the third queues behind the first. + for i, recording := range []int64{501, 502, 501} { + ev := testEvent(int64(i + 1)) + ev.EventType, ev.Kind, ev.RecordingID = "todo.created", "todo_created", recording + _, err := ledger.RecordSeen(ctx, ev, LanePoll) + require.NoError(t, err) + require.NoError(t, queue.Offer(ctx, ev.ID)) + require.Eventually(t, func() bool { + return getRecord(t, ledger, ev.ID).State != StateSeen + }, 5*time.Second, 10*time.Millisecond) + } + cancel() + require.NoError(t, <-done) + + assert.Equal(t, StateAdmitted, getRecord(t, ledger, 1).State) + assert.Equal(t, StateAdmitted, getRecord(t, ledger, 2).State) + third := getRecord(t, ledger, 3) + assert.Equal(t, StateQueued, third.State) + assert.Contains(t, string(third.Decision.Snapshot), secret, "the queued follow-up keeps its instruction") + + got := strings.Split(strings.TrimSpace(out.String()), "\n") + require.Len(t, got, 3) + assert.NotContains(t, out.String(), secret, "no content on the wire") + var last admission.Line + require.NoError(t, json.Unmarshal([]byte(got[2]), &last)) + assert.Equal(t, int64(3), last.EventID) + assert.Equal(t, admission.StateQueued, last.State, "the line reports what the ledger wrote") +} + +func TestRunAdmissionNeedsTheLedgerAndQueue(t *testing.T) { + admitter, err := admission.NewAdmitter(admission.Policy{ + AgentID: adapterAgentID, + Trust: admission.Trust{Mode: admission.TrustOperator, OperatorID: adapterOperatorID}, + }, admission.Reads{Summaries: &adapterReads{}, Subscriptions: &adapterReads{}, Assignments: &adapterReads{}}) + require.NoError(t, err) + queue, err := NewQueue(1, 1) + require.NoError(t, err) + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + + for name, opts := range map[string]AdmissionOptions{ + "no ledger": {Queue: queue, Admitter: admitter}, + "no queue": {Ledger: newTestLedger(t), Admitter: admitter}, + } { + t.Run(name, func(t *testing.T) { + err := RunAdmission(ctx, opts) + require.Error(t, err) + assert.NotErrorIs(t, err, context.DeadlineExceeded, "refused at once, not run until the deadline") + }) + } +} + +// adapterReads answers admission's reads for the handoff test. +type adapterReads struct { + summaries map[int64]*basecamp.RecordingSummary +} + +func (r *adapterReads) Summarize(_ context.Context, ref basecamp.RecordingRef) (*basecamp.RecordingSummary, error) { + s, ok := r.summaries[ref.RecordingID] + if !ok { + return nil, &basecamp.Error{Code: basecamp.CodeNotFound, Message: "not found"} + } + copied := *s + return &copied, nil +} + +func (r *adapterReads) Subscribed(context.Context, int64) (bool, error) { return false, nil } + +func (r *adapterReads) AddedPersonIDs(context.Context, int64, int64) ([]int64, bool, error) { + return nil, false, nil +} + +// mentionMarkup is a mention attachment naming id only inside its sgid, the +// way Basecamp renders one. +func mentionMarkup(id int64) string { + payload := `{"_rails":{"data":"gid://bc3/Person/` + strconv.FormatInt(id, 10) + `","pur":"attachable"}}` + sgid := base64.RawURLEncoding.EncodeToString([]byte(payload)) + return `` +} diff --git a/internal/connector/ledger_events.go b/internal/connector/ledger_events.go index 1867f7213..de2571d92 100644 --- a/internal/connector/ledger_events.go +++ b/internal/connector/ledger_events.go @@ -33,6 +33,40 @@ type Record struct { SeenAt time.Time UpdatedAt time.Time ContentDropped bool + + // Revision counts the writes that changed the record's state since + // intake recorded it. A decision applies only at the revision it loaded. + Revision int64 + // Decision is admission's latest verdict on the record; its zero value + // until one is written. + Decision Decision +} + +// Decision is what admission wrote onto a record with its verdict. The pointer +// fields stay intake's; these are admission's, and dispatch starts a task from +// them. +type Decision struct { + // DecidedAt is when the latest verdict was written. + DecidedAt *time.Time + // BlockedAt is when the record entered its current run of blocked + // verdicts; nil unless it is blocked. + BlockedAt *time.Time + // RetryAt is a throttled verdict's server deadline; nil otherwise. + RetryAt *time.Time + + Trigger string + Acknowledge bool + ConversationKey string + ReplyKind string + ReplyRecordingID int64 + Routed bool + Route string + Class string + RecordingURL string + RequesterID int64 + // Snapshot is the recording's content as admission read it, JSON, and is + // set on an admitted or queued record only. + Snapshot json.RawMessage } // EffectivePerformer is the id the performers/exclude_performers filters @@ -211,47 +245,121 @@ var ErrNotATransition = errors.New("not a transition the ledger's lifecycle allo // two statements is a race, and this is the guarantee that a finished record // stays finished. func (l *Ledger) SetState(ctx context.Context, id int64, state RecordState, reason string) error { - switch state { + moved, err := l.move(ctx, l.db, transition{id: id, state: state, reason: reason}) + if err != nil { + return err + } + if !moved { + return l.explainRefusal(ctx, id, state) + } + return nil +} + +// dbtx is what a transition runs against: the ledger's handle, or a +// transaction a caller already holds so the move commits with its other +// writes. +type dbtx interface { + ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error) + QueryRowContext(ctx context.Context, query string, args ...any) *sql.Row +} + +// transition is one state change and what is written with it. +type transition struct { + id int64 + state RecordState + reason string + // from narrows the states the record may be leaving. Empty means every + // state the lifecycle lets reach state; a narrowing can only remove edges, + // never add one. + from []RecordState + // revision, when set, applies the move only while the record is still at + // that revision. + revision *int64 + // set is further columns written in the same statement. + set []assignment +} + +// assignment is one further column a transition writes. column and expr are +// this package's own constants, never input; value is bound to the one "?" in +// expr, or to the column directly when expr is empty. +type assignment struct { + column string + expr string + value any +} + +// move is the ledger's one state-changing write. Every change of state goes +// through it, so every change meets the lifecycle, bumps the revision, and +// keeps the retention clock where a repeat must leave it. It reports whether +// the record moved; a refusal is the caller's to explain. +// +// revision is bumped on every applied write, a repeat included. It is what a +// decision loaded earlier is compared against, and any write since that load +// is a reason the decision may no longer hold. +func (l *Ledger) move(ctx context.Context, db dbtx, t transition) (bool, error) { + switch t.state { case StateBlocked, StateDiscarded: - if reason == "" { - return fmt.Errorf("connector: set state of %d: a %s record needs a reason", id, state) + if t.reason == "" { + return false, fmt.Errorf("connector: set state of %d: a %s record needs a reason", t.id, t.state) } case StateSeen, StateAdmitted, StateQueued, StateDispatched, StateCompleted: - if reason != "" { - return fmt.Errorf("connector: set state of %d: a %s record takes no reason", id, state) + if t.reason != "" { + return false, fmt.Errorf("connector: set state of %d: a %s record takes no reason", t.id, t.state) } default: // A state outside the lifecycle is a row no recovery scan looks for. - return fmt.Errorf("connector: set state of %d: %q is not a ledger state", id, state) + return false, fmt.Errorf("connector: set state of %d: %q is not a ledger state", t.id, t.state) } - froms := enterableFrom(state) - args := []any{string(state), reason, string(state), l.timestamp(), id} - for _, from := range froms { - args = append(args, from) + froms := enterableFrom(t.state) + if len(t.from) > 0 { + froms = slices.DeleteFunc(froms, func(from string) bool { + return !slices.Contains(t.from, RecordState(from)) + }) + if len(froms) == 0 { + return false, nil + } } - // The only thing concatenated is a list of "?" as long as the lifecycle's - // own edge list. Every value is bound. - //nolint:gosec // G202: placeholders, not values - // + // updated_at is left alone when the state does not change. It is the // retention clock DropContent reads, and a repeated write of the state a // record already has would silently restart the window on a finished - // record. - query := `UPDATE events SET state = ?, reason = ?, - updated_at = CASE WHEN state = ? THEN updated_at ELSE ? END -WHERE id = ? AND state IN (` + strings.TrimSuffix(strings.Repeat("?, ", len(froms)), ", ") + `)` - res, err := l.db.ExecContext(ctx, query, args...) + // record. Every right-hand side reads the row as it was before this + // statement, which is SQLite's rule for UPDATE. + var query strings.Builder + query.WriteString(`UPDATE events SET state = ?, reason = ?, revision = revision + 1, + updated_at = CASE WHEN state = ? THEN updated_at ELSE ? END`) + args := []any{string(t.state), t.reason, string(t.state), l.timestamp()} + for _, a := range t.set { + expr := a.expr + if expr == "" { + expr = "?" + } + query.WriteString(", " + a.column + " = " + expr) + args = append(args, a.value) + } + query.WriteString(" WHERE id = ?") + args = append(args, t.id) + if t.revision != nil { + query.WriteString(" AND revision = ?") + args = append(args, *t.revision) + } + query.WriteString(" AND state IN (" + strings.TrimSuffix(strings.Repeat("?, ", len(froms)), ", ") + ")") + for _, from := range froms { + args = append(args, from) + } + + // Concatenated are column names and expressions this package declares, + // and a list of "?" as long as the lifecycle's own edge list. Every value + // is bound. + res, err := db.ExecContext(ctx, query.String(), args...) //nolint:gosec // G202: constants and placeholders, not values if err != nil { - return fmt.Errorf("connector: set state of %d: %w", id, err) + return false, fmt.Errorf("connector: set state of %d: %w", t.id, err) } affected, err := res.RowsAffected() if err != nil { - return fmt.Errorf("connector: set state of %d: %w", id, err) + return false, fmt.Errorf("connector: set state of %d: %w", t.id, err) } - if affected == 0 { - return l.explainRefusal(ctx, id, state) - } - return nil + return affected > 0, nil } // explainRefusal says why an update changed nothing: there is no such record, @@ -286,7 +394,10 @@ func (l *Ledger) DropContent(ctx context.Context, discardedBefore, completedBefo UPDATE events SET details = NULL, event_type = '', kind = '', action = '', bucket_id = 0, creator_id = 0, performed_by_id = NULL, recording_id = 0, actor_type = '', - visible_to_clients = NULL, content_dropped = 1, updated_at = updated_at + visible_to_clients = NULL, content_dropped = 1, updated_at = updated_at, + snapshot = NULL, trigger_name = '', acknowledge = 0, conversation_key = '', + reply_kind = '', reply_recording_id = 0, routed = 0, route = '', class = '', + recording_url = '', requester_id = 0, blocked_at = NULL, retry_at = NULL WHERE content_dropped = 0 AND ((state = ? AND updated_at < ?) OR (state = ? AND updated_at < ?))`, string(StateDiscarded), stamp(discardedBefore), @@ -304,7 +415,10 @@ WHERE content_dropped = 0 const selectRecords = ` SELECT id, state, reason, lane, event_type, kind, action, bucket_id, creator_id, performed_by_id, recording_id, details, actor_type, visible_to_clients, - created_at, seen_at, updated_at, content_dropped + created_at, seen_at, updated_at, content_dropped, revision, decided_at, + blocked_at, retry_at, trigger_name, acknowledge, conversation_key, + reply_kind, reply_recording_id, routed, route, class, recording_url, + requester_id, snapshot FROM events` func scanRecords(rows *sql.Rows) ([]Record, error) { @@ -320,11 +434,19 @@ func scanRecords(rows *sql.Rows) ([]Record, error) { contentDropped int performedBy sql.NullInt64 visibleToClients sql.NullBool + decidedAt, blockedAt sql.NullString + retryAt sql.NullString + acknowledge, routed int + snapshot []byte + d = &r.Decision ) if err := rows.Scan(&r.ID, &state, &r.Reason, &lane, &r.EventType, &r.Kind, &r.Action, &r.BucketID, &r.CreatorID, &performedBy, &r.RecordingID, &details, &r.ActorType, &visibleToClients, &createdAt, &seenAt, - &updatedAt, &contentDropped); err != nil { + &updatedAt, &contentDropped, &r.Revision, &decidedAt, &blockedAt, + &retryAt, &d.Trigger, &acknowledge, &d.ConversationKey, &d.ReplyKind, + &d.ReplyRecordingID, &routed, &d.Route, &d.Class, &d.RecordingURL, + &d.RequesterID, &snapshot); err != nil { return nil, fmt.Errorf("connector: scan event record: %w", err) } r.State = RecordState(state) @@ -351,6 +473,23 @@ func scanRecords(rows *sql.Rows) ([]Record, error) { return nil, err } r.ContentDropped = contentDropped != 0 + d.Acknowledge, d.Routed = acknowledge != 0, routed != 0 + if len(snapshot) > 0 { + d.Snapshot = json.RawMessage(snapshot) + } + for _, stamped := range []struct { + raw sql.NullString + to **time.Time + }{{decidedAt, &d.DecidedAt}, {blockedAt, &d.BlockedAt}, {retryAt, &d.RetryAt}} { + if !stamped.raw.Valid { + continue + } + at, err := parseStamp(stamped.raw.String) + if err != nil { + return nil, err + } + *stamped.to = &at + } records = append(records, r) } return records, rows.Err() From 9ddc344fdb9042e513534cc2d49508b92ac44e4f Mon Sep 17 00:00:00 2001 From: Jorge Manrubia Date: Thu, 17 Sep 2026 06:54:56 +0200 Subject: [PATCH 2/3] Keep the blocked schedule in every move; match liveness to the contract The ledger's one state write now owns blocked_at, retry_at and dropping the snapshot on a move to blocked or discarded, so SetState cannot leave them stale. A conversation is live when a record on it is admitted or dispatched, as admission's contract says; a queued record alone would stall its key. The conversation index is unconditional so the liveness read can use it. The handoff test asserts verdict lines by event id, not by arrival order. --- internal/connector/ledger.go | 9 +- internal/connector/ledger_admission.go | 41 +++---- internal/connector/ledger_admission_test.go | 122 ++++++++++++++++++-- internal/connector/ledger_events.go | 57 +++++---- 4 files changed, 174 insertions(+), 55 deletions(-) diff --git a/internal/connector/ledger.go b/internal/connector/ledger.go index 814dc83c3..90a4fdfe6 100644 --- a/internal/connector/ledger.go +++ b/internal/connector/ledger.go @@ -296,8 +296,11 @@ END; // counts from it), and retry_at a throttled verdict's server deadline. // The rest is the verdict itself — what dispatch starts a task from, and // what a blocked(no_route) record's holding reply needs. snapshot is the - // recording's content as admission read it, and only an admitted record - // carries one; nothing else in the ledger is content. + // recording's content as admission read it. Only an admitted verdict + // writes one, whether the ledger writes it as admitted or as queued; a + // record keeps it through dispatch and completion until retention drops + // it, and loses it on any move to blocked or discarded. Nothing else in + // the ledger is content. // // Nothing has shipped that wrote a version 3 ledger, but a migration is // how the schema changes regardless: the next time something has, this is @@ -318,7 +321,7 @@ ALTER TABLE events ADD COLUMN class TEXT NOT NULL DEFAULT ''; ALTER TABLE events ADD COLUMN recording_url TEXT NOT NULL DEFAULT ''; ALTER TABLE events ADD COLUMN requester_id INTEGER NOT NULL DEFAULT 0; ALTER TABLE events ADD COLUMN snapshot BLOB; -CREATE INDEX events_conversation ON events (conversation_key, state) WHERE conversation_key <> ''; +CREATE INDEX events_conversation ON events (conversation_key, state); `, } diff --git a/internal/connector/ledger_admission.go b/internal/connector/ledger_admission.go index 3bacf61a0..d46aad1d2 100644 --- a/internal/connector/ledger_admission.go +++ b/internal/connector/ledger_admission.go @@ -7,6 +7,7 @@ import ( "errors" "fmt" "log/slog" + "time" "github.com/basecamp/basecamp-cli/internal/connector/admission" "github.com/basecamp/basecamp-cli/internal/connector/ndjson" @@ -71,8 +72,11 @@ func (a Admission) LoadUndecided(ctx context.Context, id int64) (admission.Event // is admission.ErrAlreadyDecided; // - an admitted verdict is written as queued when its conversation is // live, decided inside the transaction: another record on the key is -// admitted and not yet dispatched (it becomes the task), dispatched (the -// task is running), or queued behind one of those; +// admitted and not yet dispatched (it becomes the task) or dispatched +// (the task is running). A queued record alone is not a live +// conversation: queued records join their task before it closes, and +// counting one left behind would queue every later event on the key +// with nothing to start a task; // - content is written only with an admitted verdict; // - a throttled verdict keeps its RetryAt. func (a Admission) Commit(ctx context.Context, v admission.Verdict) (admission.State, error) { @@ -111,11 +115,7 @@ func (a Admission) commit(ctx context.Context, v admission.Verdict, state Record // The record being decided is seen or blocked, so it never counts // itself as the conversation's task. var live bool - if err := tx.QueryRowContext(ctx, ` -SELECT EXISTS ( - SELECT 1 FROM events - WHERE conversation_key = ? AND state IN (?, ?, ?) -)`, v.ConversationKey, string(StateAdmitted), string(StateQueued), string(StateDispatched)).Scan(&live); err != nil { + if err := tx.QueryRowContext(ctx, liveConversation, v.ConversationKey, string(StateAdmitted), string(StateDispatched)).Scan(&live); err != nil { return "", fmt.Errorf("connector: read conversation of %d: %w", v.EventID, err) } if live { @@ -123,19 +123,9 @@ SELECT EXISTS ( } } - now := l.timestamp() - var retryAt any - if state == StateBlocked && !v.RetryAt.IsZero() { - retryAt = stamp(v.RetryAt) - } - var blockedAt assignment + var retryAt time.Time if state == StateBlocked { - // The retry window counts from the first of a run of blocked - // verdicts, so a record re-blocked every ten minutes still ages out - // of it. Read as the row was before this write. - blockedAt = assignment{column: "blocked_at", expr: "CASE WHEN state = 'blocked' AND blocked_at IS NOT NULL THEN blocked_at ELSE ? END", value: now} - } else { - blockedAt = assignment{column: "blocked_at", value: nil} + retryAt = v.RetryAt } reply := admission.ReplyDestination{} if v.Reply != nil { @@ -148,10 +138,9 @@ SELECT EXISTS ( reason: string(v.Reason), from: undecided, revision: &revision, + retryAt: retryAt, set: []assignment{ - {column: "decided_at", value: now}, - blockedAt, - {column: "retry_at", value: retryAt}, + {column: "decided_at", value: l.timestamp()}, {column: "trigger_name", value: string(v.Trigger)}, {column: "acknowledge", value: v.Acknowledge}, {column: "conversation_key", value: v.ConversationKey}, @@ -177,6 +166,14 @@ SELECT EXISTS ( return state, nil } +// liveConversation asks whether a conversation has a task running or a record +// about to become one. It is read through events_conversation. +const liveConversation = ` +SELECT EXISTS ( + SELECT 1 FROM events + WHERE conversation_key = ? AND state IN (?, ?) +)` + // verdictState checks a verdict is one the ledger can write and maps it to // the ledger's state. Queued is the ledger's decision, never a verdict's. func verdictState(v admission.Verdict) (RecordState, error) { diff --git a/internal/connector/ledger_admission_test.go b/internal/connector/ledger_admission_test.go index 1786c84dd..4942de254 100644 --- a/internal/connector/ledger_admission_test.go +++ b/internal/connector/ledger_admission_test.go @@ -299,14 +299,15 @@ func TestAdmissionQueuesBehindALiveConversation(t *testing.T) { require.NoError(t, err) require.NoError(t, l.SetState(context.Background(), 2, StateDispatched, "")) }, admission.StateQueued}, - {"a queued record waits on a live task", func(t *testing.T, l *Ledger) { + {"a queued record alone is not a task", func(t *testing.T, l *Ledger) { _, err := l.Admission().Commit(context.Background(), admittedVerdict(2, 0, key)) require.NoError(t, err) _, err = l.Admission().Commit(context.Background(), admittedVerdict(3, 0, key)) require.NoError(t, err) require.NoError(t, l.SetState(context.Background(), 2, StateDispatched, "")) require.NoError(t, l.SetState(context.Background(), 2, StateCompleted, "")) - }, admission.StateQueued}, + require.Equal(t, StateQueued, getRecord(t, l, 3).State) + }, admission.StateAdmitted}, {"a completed task is not live", func(t *testing.T, l *Ledger) { _, err := l.Admission().Commit(context.Background(), admittedVerdict(2, 0, key)) require.NoError(t, err) @@ -338,8 +339,6 @@ func TestAdmissionQueuesBehindALiveConversation(t *testing.T) { require.NoError(t, err) assert.Equal(t, tc.want, written) assert.Equal(t, RecordState(tc.want), getRecord(t, ledger, 1).State) - - // Through the committer too, which reports what the ledger wrote. }) } } @@ -431,6 +430,99 @@ func TestAdmissionKeepsTheBlockedScheduleInputs(t *testing.T) { assert.Nil(t, d.RetryAt) } +// The blocked schedule's inputs belong to every state write, not only to a +// verdict: a record moved out of blocked by anything forgets when it was +// blocked, one moved back in starts a new window, and a record parked on +// blocked or discarded carries no content. +func TestEveryMoveKeepsTheBlockedScheduleInputs(t *testing.T) { + ledger := newTestLedger(t) + ctx := context.Background() + store := ledger.Admission() + t0 := time.Date(2026, 9, 17, 9, 0, 0, 0, time.UTC) + ledger.now = func() time.Time { return t0 } + seenRecord(t, ledger, 1) + throttled := blockedVerdict(1, 0, admission.ReasonThrottled) + throttled.RetryAt = t0.Add(time.Hour) + _, err := store.Commit(ctx, throttled) + require.NoError(t, err) + + // Out of blocked by a plain state write: no window, no deadline. + t1 := t0.Add(72 * time.Hour) + ledger.now = func() time.Time { return t1 } + require.NoError(t, ledger.SetState(ctx, 1, StateAdmitted, "")) + d := getRecord(t, ledger, 1).Decision + assert.Nil(t, d.BlockedAt, "a record that left blocked is not blocked") + assert.Nil(t, d.RetryAt) + + // Back into blocked after a dispatch: a new window from now, and the + // verdict that follows keeps it. + require.NoError(t, ledger.SetState(ctx, 1, StateDispatched, "")) + require.NoError(t, ledger.SetState(ctx, 1, StateBlocked, "read_failed")) + record := getRecord(t, ledger, 1) + require.NotNil(t, record.Decision.BlockedAt) + assert.True(t, t1.Equal(*record.Decision.BlockedAt)) + t2 := t1.Add(10 * time.Minute) + ledger.now = func() time.Time { return t2 } + _, err = store.Commit(ctx, blockedVerdict(1, record.Revision, admission.ReasonReadFailed)) + require.NoError(t, err) + d = getRecord(t, ledger, 1).Decision + require.NotNil(t, d.BlockedAt) + _, ok := admission.NextBlockedRetry(admission.ReasonReadFailed, *d.BlockedAt, *d.DecidedAt, time.Time{}) + assert.True(t, ok, "a record blocked again after a redispatch still gets its timed retries") + + // A deadline is a blocked record's alone. + _, err = ledger.move(ctx, ledger.db, transition{id: 1, state: StateBlocked, reason: "read_failed", retryAt: t2}) + require.NoError(t, err) + _, err = ledger.move(ctx, ledger.db, transition{id: 1, state: StateAdmitted, retryAt: t2}) + require.Error(t, err) +} + +func TestAMoveToBlockedOrDiscardedDropsTheSnapshot(t *testing.T) { + for _, target := range []RecordState{StateBlocked, StateDiscarded} { + t.Run(string(target), func(t *testing.T) { + ledger := newTestLedger(t) + ctx := context.Background() + seenRecord(t, ledger, 1) + _, err := ledger.Admission().Commit(ctx, admittedVerdict(1, 0, "recording:9")) + require.NoError(t, err) + require.NotEmpty(t, getRecord(t, ledger, 1).Decision.Snapshot) + + require.NoError(t, ledger.SetState(ctx, 1, target, "by_operator")) + + assert.Empty(t, getRecord(t, ledger, 1).Decision.Snapshot) + }) + } + + t.Run("dispatch keeps it", func(t *testing.T) { + ledger := newTestLedger(t) + ctx := context.Background() + seenRecord(t, ledger, 1) + _, err := ledger.Admission().Commit(ctx, admittedVerdict(1, 0, "recording:9")) + require.NoError(t, err) + require.NoError(t, ledger.SetState(ctx, 1, StateDispatched, "")) + require.NoError(t, ledger.SetState(ctx, 1, StateCompleted, "")) + assert.NotEmpty(t, getRecord(t, ledger, 1).Decision.Snapshot) + }) +} + +// The conversation read uses the conversation index rather than scanning +// every record in a live state. +func TestTheConversationReadUsesItsIndex(t *testing.T) { + ledger := newTestLedger(t) + var plan strings.Builder + rows, err := ledger.db.QueryContext(context.Background(), "EXPLAIN QUERY PLAN "+liveConversation, "recording:9", "admitted", "dispatched") + require.NoError(t, err) + defer rows.Close() + for rows.Next() { + var id, parent, unused int + var detail string + require.NoError(t, rows.Scan(&id, &parent, &unused, &detail)) + plan.WriteString(detail + "\n") + } + require.NoError(t, rows.Err()) + assert.Contains(t, plan.String(), "events_conversation") +} + // Two fetchers deciding one event at once: exactly one verdict is written. func TestAdmissionConcurrentDecisionsWriteOne(t *testing.T) { ledger := newTestLedger(t) @@ -572,13 +664,23 @@ func TestRunAdmissionDecidesWhatIntakeHandsOver(t *testing.T) { assert.Equal(t, StateQueued, third.State) assert.Contains(t, string(third.Decision.Snapshot), secret, "the queued follow-up keeps its instruction") - got := strings.Split(strings.TrimSpace(out.String()), "\n") - require.Len(t, got, 3) assert.NotContains(t, out.String(), secret, "no content on the wire") - var last admission.Line - require.NoError(t, json.Unmarshal([]byte(got[2]), &last)) - assert.Equal(t, int64(3), last.EventID) - assert.Equal(t, admission.StateQueued, last.State, "the line reports what the ledger wrote") + // Two workers: a verdict is visible in the ledger before its line is + // written, so lines arrive in no promised order. What is promised is one + // whole line per verdict, reporting the state the ledger wrote. + states := map[int64]admission.State{} + for _, raw := range strings.Split(strings.TrimSpace(out.String()), "\n") { + var line admission.Line + require.NoError(t, json.Unmarshal([]byte(raw), &line), "every line is whole JSON: %q", raw) + _, dup := states[line.EventID] + require.False(t, dup, "one line per verdict, event %d twice", line.EventID) + states[line.EventID] = line.State + } + assert.Equal(t, map[int64]admission.State{ + 1: admission.StateAdmitted, + 2: admission.StateAdmitted, + 3: admission.StateQueued, + }, states, "the line reports what the ledger wrote") } func TestRunAdmissionNeedsTheLedgerAndQueue(t *testing.T) { diff --git a/internal/connector/ledger_events.go b/internal/connector/ledger_events.go index de2571d92..47d3c41dd 100644 --- a/internal/connector/ledger_events.go +++ b/internal/connector/ledger_events.go @@ -34,8 +34,9 @@ type Record struct { UpdatedAt time.Time ContentDropped bool - // Revision counts the writes that changed the record's state since - // intake recorded it. A decision applies only at the revision it loaded. + // Revision counts every state write applied to the record since intake + // recorded it, a repeat of the state it already has included. A decision + // applies only at the revision it loaded. Revision int64 // Decision is admission's latest verdict on the record; its zero value // until one is written. @@ -64,8 +65,10 @@ type Decision struct { Class string RecordingURL string RequesterID int64 - // Snapshot is the recording's content as admission read it, JSON, and is - // set on an admitted or queued record only. + // Snapshot is the recording's content as admission read it, JSON. An + // admitted verdict writes it, as admitted or queued; it stays through + // dispatch and completion until retention drops it, and any move to + // blocked or discarded clears it. Snapshot json.RawMessage } @@ -275,16 +278,17 @@ type transition struct { // revision, when set, applies the move only while the record is still at // that revision. revision *int64 + // retryAt is a blocked record's not-before deadline, stored as retry_at; + // the zero time stores none. Only a move to blocked may carry one. + retryAt time.Time // set is further columns written in the same statement. set []assignment } -// assignment is one further column a transition writes. column and expr are -// this package's own constants, never input; value is bound to the one "?" in -// expr, or to the column directly when expr is empty. +// assignment is one further column a transition writes. column is this +// package's own constant, never input; value is bound. type assignment struct { column string - expr string value any } @@ -310,14 +314,14 @@ func (l *Ledger) move(ctx context.Context, db dbtx, t transition) (bool, error) // A state outside the lifecycle is a row no recovery scan looks for. return false, fmt.Errorf("connector: set state of %d: %q is not a ledger state", t.id, t.state) } + if !t.retryAt.IsZero() && t.state != StateBlocked { + return false, fmt.Errorf("connector: set state of %d: only a blocked record has a retry deadline", t.id) + } froms := enterableFrom(t.state) if len(t.from) > 0 { froms = slices.DeleteFunc(froms, func(from string) bool { return !slices.Contains(t.from, RecordState(from)) }) - if len(froms) == 0 { - return false, nil - } } // updated_at is left alone when the state does not change. It is the @@ -325,16 +329,29 @@ func (l *Ledger) move(ctx context.Context, db dbtx, t transition) (bool, error) // record already has would silently restart the window on a finished // record. Every right-hand side reads the row as it was before this // statement, which is SQLite's rule for UPDATE. + // + // The blocked schedule's inputs are the move's too, so no path into or out + // of blocked can leave them stale. blocked_at is when the record entered + // its current run of blocked states — cleared on leaving, so it is set + // exactly on entering and kept across a repeat — because the retry + // window counts from it. + // retry_at holds only for the move that set it. And a record moving to + // blocked or discarded loses its snapshot: only a record on its way to a + // worker carries content. + now := l.timestamp() + var retryAt any + if !t.retryAt.IsZero() { + retryAt = stamp(t.retryAt) + } var query strings.Builder query.WriteString(`UPDATE events SET state = ?, reason = ?, revision = revision + 1, - updated_at = CASE WHEN state = ? THEN updated_at ELSE ? END`) - args := []any{string(t.state), t.reason, string(t.state), l.timestamp()} + updated_at = CASE WHEN state = ? THEN updated_at ELSE ? END, + blocked_at = CASE WHEN ? <> 'blocked' THEN NULL ELSE COALESCE(blocked_at, ?) END, + retry_at = ?, + snapshot = CASE WHEN ? IN ('blocked', 'discarded') THEN NULL ELSE snapshot END`) + args := []any{string(t.state), t.reason, string(t.state), now, string(t.state), now, retryAt, string(t.state)} for _, a := range t.set { - expr := a.expr - if expr == "" { - expr = "?" - } - query.WriteString(", " + a.column + " = " + expr) + query.WriteString(", " + a.column + " = ?") args = append(args, a.value) } query.WriteString(" WHERE id = ?") @@ -348,7 +365,7 @@ func (l *Ledger) move(ctx context.Context, db dbtx, t transition) (bool, error) args = append(args, from) } - // Concatenated are column names and expressions this package declares, + // Concatenated are column names this package declares, // and a list of "?" as long as the lifecycle's own edge list. Every value // is bound. res, err := db.ExecContext(ctx, query.String(), args...) //nolint:gosec // G202: constants and placeholders, not values @@ -397,7 +414,7 @@ SET details = NULL, event_type = '', kind = '', action = '', bucket_id = 0, visible_to_clients = NULL, content_dropped = 1, updated_at = updated_at, snapshot = NULL, trigger_name = '', acknowledge = 0, conversation_key = '', reply_kind = '', reply_recording_id = 0, routed = 0, route = '', class = '', - recording_url = '', requester_id = 0, blocked_at = NULL, retry_at = NULL + recording_url = '', requester_id = 0 WHERE content_dropped = 0 AND ((state = ? AND updated_at < ?) OR (state = ? AND updated_at < ?))`, string(StateDiscarded), stamp(discardedBefore), From 9e06548225af39f3413bc44e8a69b656d6c4e2df Mon Sep 17 00:00:00 2001 From: Jorge Manrubia Date: Thu, 17 Sep 2026 06:59:28 +0200 Subject: [PATCH 3/3] Pin that liveness decides only the incoming record A queued record alone does not make a conversation live, per admission's Ledger contract, and deciding a later event neither moves nor re-decides it. The lifecycle comment names that definition and says a blocked record returns to work with content only through a new verdict. --- internal/connector/ledger_admission_test.go | 7 +++++++ internal/connector/ledger_events.go | 10 ++++++++-- 2 files changed, 15 insertions(+), 2 deletions(-) diff --git a/internal/connector/ledger_admission_test.go b/internal/connector/ledger_admission_test.go index 4942de254..fb6ca95b9 100644 --- a/internal/connector/ledger_admission_test.go +++ b/internal/connector/ledger_admission_test.go @@ -339,6 +339,13 @@ func TestAdmissionQueuesBehindALiveConversation(t *testing.T) { require.NoError(t, err) assert.Equal(t, tc.want, written) assert.Equal(t, RecordState(tc.want), getRecord(t, ledger, 1).State) + if tc.name == "a queued record alone is not a task" { + // Liveness decides only the incoming record. The queued one + // is not moved, expired or re-decided by it. + third := getRecord(t, ledger, 3) + assert.Equal(t, StateQueued, third.State) + assert.Equal(t, int64(1), third.Revision) + } }) } } diff --git a/internal/connector/ledger_events.go b/internal/connector/ledger_events.go index 47d3c41dd..54d6f3281 100644 --- a/internal/connector/ledger_events.go +++ b/internal/connector/ledger_events.go @@ -205,8 +205,14 @@ func (l *Ledger) CountInState(ctx context.Context, state RecordState) (int, erro // edges back into the working states. Completed and discarded have none. var lifecycle = map[RecordState][]RecordState{ // seen to queued is one edge, not two: admission commits an admitted - // verdict AS queued when the conversation is already live, so the record - // never passes through admitted at all. + // verdict AS queued when the conversation is already live — another + // record on it admitted or dispatched, admission.Ledger's definition — so + // the record never passes through admitted at all. Deciding that moves + // only the incoming record; a queued record leaves queued by dispatch. + // + // A blocked record's edges back into the working states are for the + // lifecycle's own bookkeeping. It returns to work with content only + // through a new verdict, because a move to blocked drops the snapshot. StateSeen: {StateAdmitted, StateQueued, StateBlocked, StateDiscarded}, StateAdmitted: {StateQueued, StateDispatched, StateBlocked, StateDiscarded}, StateQueued: {StateDispatched, StateBlocked, StateDiscarded},