From 5376bb379274504de75834a2b92303d1ccfccef0 Mon Sep 17 00:00:00 2001 From: mintaka Date: Thu, 1 Oct 2026 21:02:34 -0400 Subject: [PATCH] fix(delivery): wake owed agents when a Runner attaches (RIG-4166) A mention made after a restart, before the Runner re-enrolls, failed its wake with Unavailable and nothing retried it. When a Runner command stream attaches, the hub now notifies the delivery consumer, which wakes every agent with an owed mention. WakeAgent skips live agents and coalesces duplicates. Co-authored-by: Matt Wilkinson --- go/internal/delivery/consumer.go | 7 +++ go/internal/delivery/helpers_test.go | 10 ++++ go/internal/delivery/settle.go | 38 ++++++++++++ go/internal/runnerhub/hub.go | 18 ++++++ go/internal/store/db/delivery_cursors.sql.go | 26 +++++++++ go/internal/store/db/querier.go | 3 + go/internal/store/delivery_cursors.go | 13 +++++ .../store/queries/delivery_cursors.sql | 5 ++ go/server/offline_mention_e2e_pgtest_test.go | 58 ++++++++++++++++++- go/server/sinks.go | 1 + 10 files changed, 177 insertions(+), 2 deletions(-) diff --git a/go/internal/delivery/consumer.go b/go/internal/delivery/consumer.go index e299ac973..f3b595886 100644 --- a/go/internal/delivery/consumer.go +++ b/go/internal/delivery/consumer.go @@ -127,6 +127,9 @@ type DeliveryReads interface { //nolint:interfacebloat // one method per store r // CountOwedMentions returns the total owed_mention row count — the startup // visibility log (T2 observability). CountOwedMentions(ctx context.Context) (int, error) + // OwedMentionAccounts lists every agent still owed a mention, for the wake + // retried once a Runner can serve it. + OwedMentionAccounts(ctx context.Context) ([]store.AccountID, error) // MarkMentionsRouted stamps messageID's settle-edge mention pass complete // (mentions_routed_at = now, unix ms) — the recovery scan's mark after it // replays a message's mention pass, and the live path's mark (T3). Idempotent: @@ -224,6 +227,9 @@ type Consumer struct { // lostQueue buffers accounts whose session the Runner reported gone; the loop wakes // each, and the woken session's start sweep redelivers what is owed. lostQueue []store.AccountID + // owedRewake is set when a Runner command stream attaches: a wake that failed + // while none could serve it is retried for every agent still owed a mention. + owedRewake bool // notify wakes the loop when settleQueue OR startQueue grows. Buffered(1) // with a non-blocking send, so many edges between drains collapse to one // wakeup and the hook never blocks. @@ -352,6 +358,7 @@ func (c *Consumer) Run(ctx context.Context) error { c.drainSettles(ctx) c.drainStarts(ctx) c.drainLost(ctx) + c.drainOwedRewake(ctx) case event, ok := <-sub.Live: if !ok { if sub.Lagged() { diff --git a/go/internal/delivery/helpers_test.go b/go/internal/delivery/helpers_test.go index 8a428421f..eecfe629b 100644 --- a/go/internal/delivery/helpers_test.go +++ b/go/internal/delivery/helpers_test.go @@ -493,6 +493,16 @@ func (f *fakeReads) CountOwedMentions(_ context.Context) (int, error) { return n, nil } +func (f *fakeReads) OwedMentionAccounts(_ context.Context) ([]store.AccountID, error) { + f.mu.Lock() + defer f.mu.Unlock() + out := make([]store.AccountID, 0, len(f.owedMentions)) + for agent := range f.owedMentions { + out = append(out, agent) + } + return out, nil +} + // SweepChannels returns the seeded D1 disjunct channel set for agent — the pin // sweep's channel enumeration, mirroring the store read. func (f *fakeReads) SweepChannels(_ context.Context, agent store.AccountID) ([]store.ChannelID, error) { diff --git a/go/internal/delivery/settle.go b/go/internal/delivery/settle.go index 22a22d030..c304041b0 100644 --- a/go/internal/delivery/settle.go +++ b/go/internal/delivery/settle.go @@ -382,3 +382,41 @@ func (c *Consumer) drainLost(ctx context.Context) { c.wake(ctx, account) } } + +// OnRunnerReady is called when a Runner's command stream attaches. A mention made +// while no Runner could serve it left an owed row and a failed wake; the loop now +// wakes every owed agent. WakeAgent skips live agents and coalesces duplicates. +func (c *Consumer) OnRunnerReady() { + c.mu.Lock() + c.owedRewake = true + c.mu.Unlock() + select { + case c.notify <- struct{}{}: + default: + } +} + +// drainOwedRewake wakes every agent with an owed mention, once per OnRunnerReady. +func (c *Consumer) drainOwedRewake(ctx context.Context) { + c.mu.Lock() + run := c.owedRewake + c.owedRewake = false + c.mu.Unlock() + if !run { + return + } + accounts, err := c.st.OwedMentionAccounts(ctx) + if err != nil { + c.log.WarnContext(ctx, "delivery: list owed mention accounts on runner ready", "error", err) + return + } + // Off the loop: each wake is a Runner round trip, and a restart can owe many. + go func() { + for _, account := range accounts { + if ctx.Err() != nil { + return + } + c.wake(ctx, account) + } + }() +} diff --git a/go/internal/runnerhub/hub.go b/go/internal/runnerhub/hub.go index 6b602d6d8..8259aa68b 100644 --- a/go/internal/runnerhub/hub.go +++ b/go/internal/runnerhub/hub.go @@ -130,6 +130,12 @@ type SessionLostSink interface { OnSessionLost(sessionID string, account store.AccountID) } +// RunnerReadySink is told each time a Runner command stream attaches, so work that +// failed while no Runner could serve it is retried. Must return promptly. +type RunnerReadySink interface { + OnRunnerReady() +} + // SessionEndSink archives the transcript tail of a session the hub saw end without // Stop (a lost container or a re-enroll reap). ctx is scoped to the session tenant. // It may block on object-store I/O, so the hub calls it off its locks and loops. @@ -360,6 +366,7 @@ type Hub struct { // lost is notified after dropLostSession releases a dead session. Read under mu. lost SessionLostSink ended SessionEndSink + ready RunnerReadySink // presence is the RIG-1569 T8 presence projection's sink, notified at // deliverSession (lifecycle transition) and promoteSession (reconciliation). Nil // until SetPresenceSink; read under mu. Nil-safe (today's behavior). @@ -526,6 +533,13 @@ func (h *Hub) SetSessionReapSink(reap SessionReapSink) { h.reap = reap } +// SetRunnerReadySink wires the delivery consumer to retry owed wakes on attach. +func (h *Hub) SetRunnerReadySink(ready RunnerReadySink) { + h.mu.Lock() + defer h.mu.Unlock() + h.ready = ready +} + // SetSessionEndSink wires the transcript archive for sessions that end without Stop. func (h *Hub) SetSessionEndSink(end SessionEndSink) { h.mu.Lock() @@ -742,7 +756,11 @@ func (h *Hub) FrameDiagnostics() FrameDiagnostics { func (h *Hub) fireRunnerReady() { h.mu.Lock() hook := h.runnerReadyHook + ready := h.ready h.mu.Unlock() + if ready != nil { + ready.OnRunnerReady() + } if hook == nil { return } diff --git a/go/internal/store/db/delivery_cursors.sql.go b/go/internal/store/db/delivery_cursors.sql.go index c2d6c6559..0cc47e52c 100644 --- a/go/internal/store/db/delivery_cursors.sql.go +++ b/go/internal/store/db/delivery_cursors.sql.go @@ -122,6 +122,32 @@ func (q *Queries) MarkMentionsRouted(ctx context.Context, arg MarkMentionsRouted return err } +const owedMentionAccounts = `-- name: OwedMentionAccounts :many +SELECT DISTINCT agent_account_id FROM owed_mentions ORDER BY agent_account_id +` + +// The accounts still owed a mention: a wake that failed before any Runner could +// serve it is retried for these once one attaches. +func (q *Queries) OwedMentionAccounts(ctx context.Context) ([]string, error) { + rows, err := q.db.Query(ctx, owedMentionAccounts) + if err != nil { + return nil, err + } + defer rows.Close() + var items []string + for rows.Next() { + var agent_account_id string + if err := rows.Scan(&agent_account_id); err != nil { + return nil, err + } + items = append(items, agent_account_id) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const owedMentions = `-- name: OwedMentions :many SELECT m.id, m.topic_id, t.channel_id, m.author_account_id, COALESCE(ah.handle, '')::text AS author_handle, m.at_unix_ms, m.blocks FROM owed_mentions om diff --git a/go/internal/store/db/querier.go b/go/internal/store/db/querier.go index 56f0fb701..0ed8a52e8 100644 --- a/go/internal/store/db/querier.go +++ b/go/internal/store/db/querier.go @@ -433,6 +433,9 @@ type Querier interface { MessageInChannel(ctx context.Context, arg MessageInChannelParams) (int32, error) MessagesHeadSeq(ctx context.Context) (int64, error) MoveMessagesToTopic(ctx context.Context, arg MoveMessagesToTopicParams) error + // The accounts still owed a mention: a wake that failed before any Runner could + // serve it is retried for these once one attaches. + OwedMentionAccounts(ctx context.Context) ([]string, error) OwedMentions(ctx context.Context, agentAccountID string) ([]OwedMentionsRow, error) OwnerHasPresentAgent(ctx context.Context, arg OwnerHasPresentAgentParams) (bool, error) PinnedEntries(ctx context.Context, channelID string) ([]PinnedEntriesRow, error) diff --git a/go/internal/store/delivery_cursors.go b/go/internal/store/delivery_cursors.go index 581de6b3a..b868a2449 100644 --- a/go/internal/store/delivery_cursors.go +++ b/go/internal/store/delivery_cursors.go @@ -401,3 +401,16 @@ func (s *Store) InSweepSet(ctx context.Context, agent AccountID, channel Channel } return in, nil } + +// OwedMentionAccounts lists every agent with at least one owed mention. +func (s *Store) OwedMentionAccounts(ctx context.Context) ([]AccountID, error) { + rows, err := s.q.OwedMentionAccounts(ctx) + if err != nil { + return nil, fmt.Errorf("store: list owed mention accounts: %w", err) + } + out := make([]AccountID, len(rows)) + for i, r := range rows { + out[i] = AccountID(r) + } + return out, nil +} diff --git a/go/internal/store/queries/delivery_cursors.sql b/go/internal/store/queries/delivery_cursors.sql index 3adc6ac7f..f09f2bd6a 100644 --- a/go/internal/store/queries/delivery_cursors.sql +++ b/go/internal/store/queries/delivery_cursors.sql @@ -105,3 +105,8 @@ SELECT EXISTS( WHERE cm.account_id = $1 AND cm.channel_id = $2 AND (cm.subscribed OR cm.channel_id = aa.home_channel_id OR ch.mandatory_subscription)); + +-- The accounts still owed a mention: a wake that failed before any Runner could +-- serve it is retried for these once one attaches. +-- name: OwedMentionAccounts :many +SELECT DISTINCT agent_account_id FROM owed_mentions ORDER BY agent_account_id; diff --git a/go/server/offline_mention_e2e_pgtest_test.go b/go/server/offline_mention_e2e_pgtest_test.go index 1a62d7fa6..11f3a9b51 100644 --- a/go/server/offline_mention_e2e_pgtest_test.go +++ b/go/server/offline_mention_e2e_pgtest_test.go @@ -51,6 +51,15 @@ type mentionE2EWire struct { // also creates one shared OPEN channel the admin authors into, admin a member so // PostMessage's D9 membership gate admits the post. func newMentionE2EWire(t *testing.T) *mentionE2EWire { + t.Helper() + w := newMentionE2EWireNoRunner(t) + w.runner = attachFakeRunner(t, w.store, w.hub, false) + return w +} + +// newMentionE2EWireNoRunner is newMentionE2EWire before any Runner attaches: the +// gap after a Server restart, before the Runner re-enrolls. +func newMentionE2EWireNoRunner(t *testing.T) *mentionE2EWire { t.Helper() ctx := context.Background() // the test root context (rule://go-thread-context _test.go exemption) dsn := pgtest.RequireDSN(t) @@ -86,7 +95,6 @@ func newMentionE2EWire(t *testing.T) *mentionE2EWire { brd := board.NewProjection(bus) tail := newSessionTail() hub := newRunnerHub(st, brd, tail, commsSvc, slog.New(slog.DiscardHandler)) - runner := attachFakeRunner(t, st, hub, false) // The production delivery wire (sinks.go:142-155), assembled inline with the // REAL resume-based waker (newLifecycleService), not a fake. @@ -95,6 +103,7 @@ func newMentionE2EWire(t *testing.T) *mentionE2EWire { hub.SetSettleSink(c) hub.SetSessionStartSink(c) hub.SetSessionLostSink(c) + hub.SetRunnerReadySink(c) hub.SetDeliveryStore(st) runCtx, cancel := context.WithCancel(ctx) done := make(chan struct{}) @@ -123,7 +132,6 @@ func newMentionE2EWire(t *testing.T) *mentionE2EWire { hub: hub, comms: commsSvc, consumer: c, - runner: runner, dsn: dsn, adminID: admin.ID, channel: ch.ID, @@ -462,3 +470,49 @@ func TestDeliverToLostSessionWakesAndRedelivers(t *testing.T) { t.Fatalf("SessionForAccount after wake = (%q, %v), want (%q, true)", sess, ok, newSess) } } + +// TestMentionBeforeRunnerEnrollsWakesOnAttach: a mention made after a restart but +// before the Runner re-enrolls fails its wake (no Runner to serve it). Once a Runner +// command stream attaches, the owed agent is woken. +func TestMentionBeforeRunnerEnrollsWakesOnAttach(t *testing.T) { + w := newMentionE2EWireNoRunner(t) + agent := w.seedAgentMember(t, "gapwake", false) + w.post(t, "@gapwake are you there") + // Routing settles on the consumer loop, so the owed row lands asynchronously. + deadline := timeAfter() + for { + owed, err := w.store.OwedMentions(w.ctx, agent.ID) + if err != nil { + t.Fatalf("OwedMentions: %v", err) + } + if len(owed) == 1 { + break + } + select { + case <-deadline: + t.Fatalf("owed mention channels = %d, want 1 before any Runner attaches", len(owed)) + default: + } + runtime.Gosched() + } + + // The wake fires during attach, before the harness clears its probe bookkeeping, + // so assert the durable session the fresh start records, not the command log. + // Read under the system role, as the wake path does: a wake from the delivery + // loop records its session without a tenant. + w.runner = attachFakeRunner(t, w.store, w.hub, false) + deadline = timeAfter() + for { + if _, ok, err := w.store.LatestSessionForAccount(store.WithSystemRole(w.ctx), agent.ID); err != nil { + t.Fatalf("LatestSessionForAccount: %v", err) + } else if ok { + return + } + select { + case <-deadline: + t.Fatal("no session started for the owed agent after the Runner attached") + default: + } + runtime.Gosched() + } +} diff --git a/go/server/sinks.go b/go/server/sinks.go index a98d33b1d..75424c169 100644 --- a/go/server/sinks.go +++ b/go/server/sinks.go @@ -125,6 +125,7 @@ func startDeliveryConsumer(gctx context.Context, g *errgroup.Group, commsBus *ev hub.SetSessionStartSink(c) hub.SetSessionReapSink(c) hub.SetSessionLostSink(c) + hub.SetRunnerReadySink(c) hub.SetDeliveryStore(st) g.Go(func() error { return c.Run(gctx) }) }