From b3407d6f71775d25f025677356da8bcbde9aab68 Mon Sep 17 00:00:00 2001 From: mintaka Date: Sat, 12 Sep 2026 18:16:56 -0400 Subject: [PATCH 1/2] feat(ingest): advance a suppressed subscriber's delivery cursor when caught up (RIG-3326) A suppressed notification is never acked, so delivered_revision stays behind the cursor revision and the reconcile sweep re-delivers the self-notification as a synthetic UPDATE on every sweep. The suppress path therefore advances the cursor itself - a narrow amendment to the rule that the router never advances delivered_revision, since a suppressed dispatch has no agent to ack it. The advance is conditional on the subscriber being caught up. Advancing unconditionally would erase the gap left by a real event that was dispatched but never delivered, and the agent would never learn of it - losing work-relevant signal, which is strictly worse than a redundant wake. Caught-up means delivered_revision equals the cursor revision as it stood BEFORE this event's upsert, with an empty string for a coordinate never observed, matching a fresh subscriber's default. A separate compare-and-set in SQL guards the write, so a concurrent route cannot erase a gap it did not observe. A lost CAS or a store fault is logged and swallowed: the sweep then synthesizes one UPDATE, which degrades open, never closed. Container-scope subscribers never advance. Their cursor lives on the number=0 row and the sweep compares it against the container cursor's revision, so writing an artifact revision there would poison the row and synthesize an UPDATE every sweep forever. The ack arm's existing unguarded query is untouched: an ack must land regardless of the stored value, so the CAS is a sibling query. Co-authored-by: Matt Wilkinson --- go/internal/ingest/notify_router.go | 75 +++++- go/internal/ingest/notify_router_test.go | 235 +++++++++++++++++- .../store/db/forge_subscriptions.sql.go | 32 +++ go/internal/store/db/querier.go | 7 + go/internal/store/forge_subscriptions.go | 30 +++ .../store/forge_subscriptions_pgtest_test.go | 101 ++++++++ .../store/queries/forge_subscriptions.sql | 11 + go/server/forge_notify_matrix_test.go | 7 + go/server/forge_notify_pgtest_test.go | 186 ++++++++++++++ go/server/serve.go | 8 + 10 files changed, 685 insertions(+), 7 deletions(-) diff --git a/go/internal/ingest/notify_router.go b/go/internal/ingest/notify_router.go index 8aa58e85..39f8e9b6 100644 --- a/go/internal/ingest/notify_router.go +++ b/go/internal/ingest/notify_router.go @@ -66,9 +66,12 @@ type NotifyTarget struct { // NotifyStore is the durable surface the router + reconciler (T5) share — the // server wiring adapts *store.Store and binds (provider, host), the -// forgePollStore pattern (serve.go:1082-1090). There is deliberately NO -// delivered-revision advance here: the advance rides the hub's -// ForgeNotificationAck arm in go/server (W3), never the router. +// forgePollStore pattern (serve.go:1082-1090). The delivered-revision advance +// normally rides the hub's ForgeNotificationAck arm in go/server (W3), never +// the router — with ONE narrow exception: a self-origin-SUPPRESSED dispatch has +// no agent to ack it, so the router advances the cursor itself via +// AdvanceDeliveredRevisionCAS, and ONLY when the subscriber was already caught +// up (the CAS predicate makes that concurrency-safe). // // RECONCILED INCONSISTENCY (surfaced): the frozen interface block // (design.md:815-825) lists exactly three methods, but the frozen Route @@ -94,6 +97,15 @@ type NotifyStore interface { // snapshot + revision), BEFORE notify (fetch-side truth advances // unconditionally, DL-053's split). UpsertArtifactCursor(ctx context.Context, cur ArtifactCursor) error + // AdvanceDeliveredRevisionCAS advances one subscriber's delivered_revision + // from prior to next as a compare-and-set — the write lands only when the + // stored value still equals prior, so a concurrent route cannot erase a gap + // it did not observe. Reports whether the row advanced (a lost CAS is + // advanced=false with a nil error, distinct from a store fault). The suppress + // path is the SOLE caller (amending W3); a caught-up self-suppressed + // subscriber advances so the reconcile sweep does not resurrect the + // suppressed self-notification as a synthetic UPDATE. + AdvanceDeliveredRevisionCAS(ctx context.Context, agentAccountID, subscriptionID, prior, next string) (bool, error) } // NotifyDispatcher is the notify seam: resolve account -> live session -> @@ -324,10 +336,20 @@ func (r *NotifyRouter) Route(ctx context.Context, ev forge.ForgeEvent) error { // event, so resolve it once here; the per-subscriber handle is resolved in // selfOrigin, memoized per route. A nil identity seam leaves actor a zero // Handle and every selfOrigin call false (suppression disabled). + // + // priorRevision is the PRE-upsert cursor revision (cur was loaded at step 1, + // before step 4 rewrote it): "" when the coordinate was never observed + // (cur == nil), which correctly matches a fresh subscriber's default + // delivered_revision. cur is a *ArtifactCursor, so the nil guard is required. + priorRevision := "" + if cur != nil { + priorRevision = cur.Revision + } actor := r.actorHandle(ctx, ev) subMemo := map[string]Handle{} for _, sub := range subs { if r.selfOrigin(ctx, actor, sub, subMemo) { + r.advanceOnSuppress(ctx, sub, priorRevision, revision) continue } n := r.notification(ev, sub.SubscriptionID, revision) @@ -373,6 +395,53 @@ func (r *NotifyRouter) SynthesizeUpdate(ctx context.Context, sub NotifySubscribe } } +// advanceOnSuppress advances a self-origin-suppressed subscriber's +// delivered_revision to the route's revision — the one router-side advance +// (amending W3), because a suppressed notification is never acked and the +// reconcile sweep would otherwise resurrect it as a synthetic UPDATE every +// sweep. It is CONDITIONAL and SCOPED: +// +// - Scope: ARTIFACT only. A CONTAINER-scope subscriber's delivered_revision +// lives on its number=0 row and the sweep compares it against the CONTAINER +// cursor's revision; writing this ARTIFACT revision there would poison that +// row (the container sweep would then synthesize an UPDATE every sweep). So +// a container-scope suppressed sub is skipped but NEVER advanced. +// - Caught up (Go gate): advance only when this subscriber was already caught +// up to the PRIOR cursor revision — sub.DeliveredRevision == priorRevision. +// A trailing subscriber has a genuinely-missed earlier event (E1); advancing +// would erase that gap and the sweep would never recover it, silently losing +// work-relevant signal. So a trailing sub is skipped, and the next sweep's +// synthetic UPDATE is the correct recovery for its missed event. +// - Concurrency (SQL CAS): the write itself is a compare-and-set keyed on +// prior = sub.DeliveredRevision, so a concurrent route that advanced the row +// between step 5 and here cannot be clobbered — the CAS then lands zero rows +// (advanced=false) and we degrade open. +// +// A lost CAS (advanced=false) or any store fault degrades OPEN — logged and +// swallowed, worst case one synthetic UPDATE on the next sweep — never a route +// failure. +func (r *NotifyRouter) advanceOnSuppress(ctx context.Context, sub NotifySubscriber, priorRevision, next string) { + if sub.Scope != compassv1internal.ForgeSubscriptionScope_FORGE_SUBSCRIPTION_SCOPE_ARTIFACT { + return + } + if sub.DeliveredRevision != priorRevision { + return // trailing: leave the gap for the sweep to recover. + } + advanced, err := r.store.AdvanceDeliveredRevisionCAS(ctx, sub.AgentAccountID, sub.SubscriptionID, sub.DeliveredRevision, next) + if err != nil { + r.log.WarnContext(ctx, "forge notify suppress-advance failed", + "subscription_id", sub.SubscriptionID, + "account", sub.AgentAccountID, + "error", err) + return + } + if !advanced { + r.log.WarnContext(ctx, "forge notify suppress-advance lost CAS", + "subscription_id", sub.SubscriptionID, + "account", sub.AgentAccountID) + } +} + // actorHandle resolves the event's owner-qualified ACTOR handle per the // suppress/keep matrix. A zero Handle means no actor evidence or an unresolvable // one — the caller fails open. A nil identity seam short-circuits every arm to diff --git a/go/internal/ingest/notify_router_test.go b/go/internal/ingest/notify_router_test.go index 3c643ad1..d683b378 100644 --- a/go/internal/ingest/notify_router_test.go +++ b/go/internal/ingest/notify_router_test.go @@ -30,10 +30,21 @@ type fakeNotifyStore struct { lastOpened bool lastProject string - // advanceCalls MUST stay zero: the router never advances delivered_revision - // (W3). No NotifyStore method advances it; this counter is the runtime guard - // that no future edit sneaks an advance onto the router's path. - advanceCalls int + // advanceCalls counts every AdvanceDeliveredRevisionCAS call — it MUST stay + // zero on the normal (non-suppressed) path (W3), and on the suppress path is + // nonzero only after the Go-side caught-up gate passes for an artifact-scope + // sub. advanceArgs records each call's args so a test can assert the prior the + // router passed. The fake models the SQL CAS: advanced == (arg.prior == + // casStoredPrior), so a test forces a lost CAS by setting casStoredPrior to a + // value the router won't pass. casErr forces a store fault. + advanceCalls int + advanceArgs []casCall + casStoredPrior string + casErr error +} + +type casCall struct { + agent, subscriptionID, prior, next string } func (f *fakeNotifyStore) LoadArtifactCursor(_ context.Context, _ string, _ compassv1internal.ForgeArtifactKind, _ uint64) (*ArtifactCursor, error) { @@ -79,6 +90,19 @@ func (f *fakeNotifyStore) UpsertArtifactCursor(_ context.Context, cur ArtifactCu return nil } +// AdvanceDeliveredRevisionCAS records the call and models the SQL CAS: it +// advances (returns true) iff the router-supplied prior equals casStoredPrior. +// casErr forces a store fault. The router only reaches here after its own +// Go-side caught-up gate, so a call at all means that gate passed. +func (f *fakeNotifyStore) AdvanceDeliveredRevisionCAS(_ context.Context, agent, subscriptionID, prior, next string) (bool, error) { + f.advanceCalls++ + f.advanceArgs = append(f.advanceArgs, casCall{agent: agent, subscriptionID: subscriptionID, prior: prior, next: next}) + if f.casErr != nil { + return false, f.casErr + } + return prior == f.casStoredPrior, nil +} + // fakeDispatcher records every notification per account. type fakeDispatcher struct { sent []*compassv1internal.ForgeNotification @@ -158,6 +182,7 @@ const ( chUpdate = compassv1internal.ForgeNotificationKind_FORGE_NOTIFICATION_KIND_UPDATE chReview = compassv1internal.ForgeNotificationKind_FORGE_NOTIFICATION_KIND_REVIEW + scopeArtifact = compassv1internal.ForgeSubscriptionScope_FORGE_SUBSCRIPTION_SCOPE_ARTIFACT scopeContainer = compassv1internal.ForgeSubscriptionScope_FORGE_SUBSCRIPTION_SCOPE_CONTAINER // selfAgent is the agent handle every suppression fixture uses; the owner @@ -1006,3 +1031,205 @@ func TestSelfOriginNilResolverDeliversEverything(t *testing.T) { t.Errorf("notifications = %d, want 1 (nil resolver disables suppression)", len(d.sent)) } } + +// ---- self-origin suppress-path delivery-cursor advance (T2) ---- +// +// The suppress path advances a self-suppressed subscriber's delivered_revision +// to the route revision — but ONLY for an artifact-scope sub that was already +// caught up to the PRIOR cursor revision, via a compare-and-set. selfCommentIDs +// resolves the self-comment fixture's actor to a matching subscriber, so every +// case below suppresses (0 dispatched); what varies is whether the advance fires. +func selfCommentIDs() *fakeIdentityResolver { + return &fakeIdentityResolver{accounts: map[string]Handle{"acct-self": {Owner: "own", Agent: "atlas"}}} +} + +// TestSuppressAdvancesCaughtUpArtifactScope: a suppressed ARTIFACT-scope +// subscriber that is caught up to the prior cursor revision has its +// delivered_revision advanced to the route revision. The cursor's prior revision +// equals the subscriber's DeliveredRevision, so the caught-up gate passes and the +// CAS lands. +func TestSuppressAdvancesCaughtUpArtifactScope(t *testing.T) { + const prior = "rev-prior" + st := &fakeNotifyStore{ + cursor: &ArtifactCursor{Repo: "o/r", Kind: kindIssue, Number: 7, Revision: prior}, + artifactSub: []NotifySubscriber{{SubscriptionID: "s", AgentAccountID: "acct-self", DeliveredRevision: prior, Scope: scopeArtifact}}, + casStoredPrior: prior, // the CAS matches the caught-up prior. + } + d := &fakeDispatcher{} + if err := newRouterWithIDs(t, st, d, selfCommentIDs()).Route(context.Background(), compassComment("own")); err != nil { + t.Fatalf("Route: %v", err) + } + if len(d.sent) != 0 { + t.Fatalf("notifications = %d, want 0 (self-origin suppressed)", len(d.sent)) + } + if st.advanceCalls != 1 { + t.Fatalf("advanceCalls = %d, want 1 (caught-up artifact-scope advances)", st.advanceCalls) + } + got := st.advanceArgs[0] + if got.prior != prior { + t.Errorf("CAS prior = %q, want %q (the subscriber's caught-up revision)", got.prior, prior) + } + want := SnapshotRevision(new(ApplyEvent(decodeSnapshot(&ArtifactCursor{Repo: "o/r", Kind: kindIssue, Number: 7, Revision: prior}), compassComment("own")))) + if got.next != want { + t.Errorf("CAS next = %q, want the route revision %q", got.next, want) + } + if got.subscriptionID != "s" || got.agent != "acct-self" { + t.Errorf("CAS target = %s/%s, want s/acct-self", got.subscriptionID, got.agent) + } +} + +// TestSuppressDoesNotAdvanceTrailingSubscriber is the E1-never-learned safety +// case — the most important test in the slice. A suppressed subscriber whose +// DeliveredRevision TRAILS the prior cursor revision (an earlier undelivered +// event) must NOT advance: advancing would erase the gap and the reconcile sweep +// would never recover the missed event. The fixture is set up so the advance +// WOULD fire if the caught-up gate were dropped — the CAS would match — so a +// zero advanceCalls proves the Go gate, not the CAS, held the line. +func TestSuppressDoesNotAdvanceTrailingSubscriber(t *testing.T) { + const ( + priorCursor = "rev-cursor" // where the shared cursor is + trailing = "rev-old" // the subscriber trails it: a missed E1 + ) + st := &fakeNotifyStore{ + cursor: &ArtifactCursor{Repo: "o/r", Kind: kindIssue, Number: 7, Revision: priorCursor}, + artifactSub: []NotifySubscriber{{SubscriptionID: "s", AgentAccountID: "acct-self", DeliveredRevision: trailing, Scope: scopeArtifact}}, + casStoredPrior: trailing, // the CAS WOULD match if the gate were dropped. + } + d := &fakeDispatcher{} + if err := newRouterWithIDs(t, st, d, selfCommentIDs()).Route(context.Background(), compassComment("own")); err != nil { + t.Fatalf("Route: %v", err) + } + if len(d.sent) != 0 { + t.Fatalf("notifications = %d, want 0 (still suppressed)", len(d.sent)) + } + if st.advanceCalls != 0 { + t.Fatalf("advanceCalls = %d, want 0 (a trailing subscriber must NOT advance — the E1 gap is left for the sweep)", st.advanceCalls) + } +} + +// TestSuppressAdvancesFreshSubscriberNilCursor proves the nil-cursor guard: when +// the coordinate was NEVER observed (cur == nil), the prior revision is "", which +// matches a fresh subscriber's empty DeliveredRevision — so it counts as caught +// up and advances. A missing nil guard is a nil-deref panic, so this asserts the +// advance behaviour, not merely the absence of a crash. +func TestSuppressAdvancesFreshSubscriberNilCursor(t *testing.T) { + st := &fakeNotifyStore{ + cursor: nil, // never observed. + artifactSub: []NotifySubscriber{{SubscriptionID: "s", AgentAccountID: "acct-self", DeliveredRevision: "", Scope: scopeArtifact}}, + casStoredPrior: "", // the CAS matches the empty caught-up prior. + } + d := &fakeDispatcher{} + if err := newRouterWithIDs(t, st, d, selfCommentIDs()).Route(context.Background(), compassComment("own")); err != nil { + t.Fatalf("Route: %v", err) + } + if st.advanceCalls != 1 { + t.Fatalf("advanceCalls = %d, want 1 (nil cursor -> prior \"\" == fresh sub's \"\", caught up)", st.advanceCalls) + } + if got := st.advanceArgs[0].prior; got != "" { + t.Errorf("CAS prior = %q, want \"\" (nil cursor)", got) + } +} + +// TestSuppressNeverAdvancesContainerScope is the scope carve-out: a suppressed +// CONTAINER-scope subscriber (the OPENED fan-out case) is skipped but NEVER +// advanced, even when caught up — an artifact revision in a container cursor +// poisons the container sweep. The subscriber is caught up and the CAS would +// match, so a zero advanceCalls proves the scope check, not the caught-up gate, +// held the line. +func TestSuppressNeverAdvancesContainerScope(t *testing.T) { + const prior = "rev-prior" + st := &fakeNotifyStore{ + cursor: &ArtifactCursor{Repo: "RIG", Kind: kindIssue, Number: 42, Revision: prior}, + openedSub: []NotifySubscriber{{SubscriptionID: "s", AgentAccountID: "acct-self", Project: "proj-A", DeliveredRevision: prior, Scope: scopeContainer}}, + casStoredPrior: prior, // the CAS WOULD match if the scope check were dropped. + } + d := &fakeDispatcher{} + ids := &fakeIdentityResolver{ + author: Handle{Owner: "own", Agent: "atlas"}, + accounts: map[string]Handle{"acct-self": {Owner: "own", Agent: "atlas"}}, + } + ev := forge.ForgeEvent{ + Provider: compassv1.ForgeProvider_FORGE_PROVIDER_LINEAR, Host: "linear.app", + Repo: "RIG", Kind: kindIssue, Number: 42, Project: "proj-A", URL: "u", Change: chOpened, + } + if err := newRouterWithIDs(t, st, d, ids).Route(context.Background(), ev); err != nil { + t.Fatalf("Route: %v", err) + } + if len(d.sent) != 0 { + t.Fatalf("notifications = %d, want 0 (self-opened suppressed)", len(d.sent)) + } + if st.advanceCalls != 0 { + t.Fatalf("advanceCalls = %d, want 0 (container-scope is skipped but NEVER advanced)", st.advanceCalls) + } +} + +// TestSuppressAdvanceLostCASRoutesCleanly: a lost CAS (advanced=false, no error) +// degrades open — Route still returns nil. The caught-up gate passes (so the CAS +// is attempted) but casStoredPrior differs from the subscriber's revision, so the +// fake reports no advance. +func TestSuppressAdvanceLostCASRoutesCleanly(t *testing.T) { + const prior = "rev-prior" + st := &fakeNotifyStore{ + cursor: &ArtifactCursor{Repo: "o/r", Kind: kindIssue, Number: 7, Revision: prior}, + artifactSub: []NotifySubscriber{{SubscriptionID: "s", AgentAccountID: "acct-self", DeliveredRevision: prior, Scope: scopeArtifact}}, + casStoredPrior: "someone-else-advanced", // a concurrent route moved the row first. + } + d := &fakeDispatcher{} + if err := newRouterWithIDs(t, st, d, selfCommentIDs()).Route(context.Background(), compassComment("own")); err != nil { + t.Fatalf("Route returned error on a lost CAS, want nil (degrade open): %v", err) + } + if st.advanceCalls != 1 { + t.Fatalf("advanceCalls = %d, want 1 (the CAS was attempted)", st.advanceCalls) + } +} + +// TestSuppressAdvanceFaultRoutesCleanly: a store fault from the CAS is logged and +// swallowed — Route still returns nil and the OTHER (delivered) subscriber is +// unaffected. A second, non-self subscriber is present and must still receive its +// notification. +func TestSuppressAdvanceFaultRoutesCleanly(t *testing.T) { + const prior = "rev-prior" + st := &fakeNotifyStore{ + cursor: &ArtifactCursor{Repo: "o/r", Kind: kindIssue, Number: 7, Revision: prior}, + artifactSub: []NotifySubscriber{ + {SubscriptionID: "self", AgentAccountID: "acct-self", DeliveredRevision: prior, Scope: scopeArtifact}, + {SubscriptionID: "other", AgentAccountID: "acct-other", DeliveredRevision: prior, Scope: scopeArtifact}, + }, + casErr: errors.New("store unreachable"), + } + d := &fakeDispatcher{} + // acct-other resolves to a DIFFERENT handle, so it is not suppressed. + ids := &fakeIdentityResolver{accounts: map[string]Handle{ + "acct-self": {Owner: "own", Agent: "atlas"}, + "acct-other": {Owner: "own", Agent: "borges"}, + }} + if err := newRouterWithIDs(t, st, d, ids).Route(context.Background(), compassComment("own")); err != nil { + t.Fatalf("Route returned error on a CAS fault, want nil (swallowed): %v", err) + } + if st.advanceCalls != 1 { + t.Fatalf("advanceCalls = %d, want 1 (the self sub attempted its advance)", st.advanceCalls) + } + if len(d.sent) != 1 || d.sent[0].GetSubscriptionId() != "other" { + t.Fatalf("dispatched = %v, want exactly the non-self 'other' subscriber", subIDs(d.sent)) + } +} + +// TestDeliveredPathNeverAdvances: a normal (non-suppressed) dispatch never calls +// the CAS — W3 still holds for the delivery path. The resolver matches no +// subscriber (nil ids via newRouter), so every sub delivers and none advances. +func TestDeliveredPathNeverAdvances(t *testing.T) { + st := &fakeNotifyStore{ + cursor: &ArtifactCursor{Repo: "o/r", Kind: kindIssue, Number: 7, Revision: "rev-prior"}, + artifactSub: []NotifySubscriber{{SubscriptionID: "s", AgentAccountID: "acct-self", DeliveredRevision: "rev-prior", Scope: scopeArtifact}}, + } + d := &fakeDispatcher{} + if err := newRouter(t, st, d, &fakeChecksRoller{}).Route(context.Background(), commentEvent("https://gh/c1")); err != nil { + t.Fatalf("Route: %v", err) + } + if len(d.sent) != 1 { + t.Fatalf("notifications = %d, want 1 (delivered path)", len(d.sent)) + } + if st.advanceCalls != 0 { + t.Fatalf("advanceCalls = %d, want 0 (W3: the delivery path never advances)", st.advanceCalls) + } +} diff --git a/go/internal/store/db/forge_subscriptions.sql.go b/go/internal/store/db/forge_subscriptions.sql.go index 82baad9b..39d1d8fe 100644 --- a/go/internal/store/db/forge_subscriptions.sql.go +++ b/go/internal/store/db/forge_subscriptions.sql.go @@ -31,6 +31,38 @@ func (q *Queries) AdvanceForgeDeliveredRevision(ctx context.Context, arg Advance return result.RowsAffected(), nil } +const advanceForgeDeliveredRevisionCAS = `-- name: AdvanceForgeDeliveredRevisionCAS :execrows +UPDATE agent_forge_subscriptions + SET delivered_revision = $3, delivered_at = now() + WHERE id = $2 AND agent_account_id = $1 AND delivered_revision = $4 +` + +type AdvanceForgeDeliveredRevisionCASParams struct { + AgentAccountID string + ID string + DeliveredRevision string + DeliveredRevision_2 string +} + +// Compare-and-set advance for the notify-router suppress path: the write lands +// only when delivered_revision still equals $4 (the prior value the router read), +// so a concurrent route cannot erase a delivery gap it did not observe. Scoped to +// the owning agent (id AND agent_account_id). Zero rows affected is a lost CAS +// (someone else advanced first), NOT an error — the wrapper reports it as +// advanced=false. +func (q *Queries) AdvanceForgeDeliveredRevisionCAS(ctx context.Context, arg AdvanceForgeDeliveredRevisionCASParams) (int64, error) { + result, err := q.db.Exec(ctx, advanceForgeDeliveredRevisionCAS, + arg.AgentAccountID, + arg.ID, + arg.DeliveredRevision, + arg.DeliveredRevision_2, + ) + if err != nil { + return 0, err + } + return result.RowsAffected(), nil +} + const countAgentForgeSubscriptionsForArtifact = `-- name: CountAgentForgeSubscriptionsForArtifact :one SELECT count(*) FROM agent_forge_subscriptions WHERE forge_provider = $1 AND forge_host = $2 AND repo = $3 AND kind = $4 AND number = $5 diff --git a/go/internal/store/db/querier.go b/go/internal/store/db/querier.go index aeac89a2..3c80e184 100644 --- a/go/internal/store/db/querier.go +++ b/go/internal/store/db/querier.go @@ -16,6 +16,13 @@ type Querier interface { ActivityFor(ctx context.Context, dollar_1 []string) ([]ActivityForRow, error) AdvanceDeliveryCursor(ctx context.Context, arg AdvanceDeliveryCursorParams) error AdvanceForgeDeliveredRevision(ctx context.Context, arg AdvanceForgeDeliveredRevisionParams) (int64, error) + // Compare-and-set advance for the notify-router suppress path: the write lands + // only when delivered_revision still equals $4 (the prior value the router read), + // so a concurrent route cannot erase a delivery gap it did not observe. Scoped to + // the owning agent (id AND agent_account_id). Zero rows affected is a lost CAS + // (someone else advanced first), NOT an error — the wrapper reports it as + // advanced=false. + AdvanceForgeDeliveredRevisionCAS(ctx context.Context, arg AdvanceForgeDeliveredRevisionCASParams) (int64, error) AgentForContainer(ctx context.Context, containerName string) (string, error) // Presence-component read queries (sqlc adoption T4, RIG-3034). These replace the // const-hoisted SQL in internal/store/presence_reads.go (it was never in the diff --git a/go/internal/store/forge_subscriptions.go b/go/internal/store/forge_subscriptions.go index 18e000ca..2f496cc2 100644 --- a/go/internal/store/forge_subscriptions.go +++ b/go/internal/store/forge_subscriptions.go @@ -488,3 +488,33 @@ func (s *Store) AdvanceForgeDeliveredRevision(ctx context.Context, agent Account } return nil } + +// AdvanceForgeDeliveredRevisionCAS advances one subscription's per-subscriber +// DELIVERY cursor from prior to next as a COMPARE-AND-SET: the write lands only +// when the stored delivered_revision still equals prior, so a concurrent route +// cannot erase a delivery gap it did not observe. Scoped to the owning agent (id +// AND agent_account_id). Reports whether the row advanced — zero rows affected +// (a lost CAS, or an unknown/foreign id) is (false, nil), NOT folded into +// ErrNotFound, so the router's suppress path distinguishes a lost CAS (degrade +// open, one synthetic UPDATE) from a real store fault (err != nil). This is the +// notify-router suppress path's writer (amending W3); the ack arm keeps the +// unguarded AdvanceForgeDeliveredRevision above. Empty agent / subscription id +// -> ErrInvalidArgument. +func (s *Store) AdvanceForgeDeliveredRevisionCAS(ctx context.Context, agent AccountID, subscriptionID, prior, next string) (bool, error) { + if agent == "" { + return false, fmt.Errorf("%w: agent account id is required", ErrInvalidArgument) + } + if subscriptionID == "" { + return false, fmt.Errorf("%w: subscription id is required", ErrInvalidArgument) + } + affected, err := s.q.AdvanceForgeDeliveredRevisionCAS(ctx, db.AdvanceForgeDeliveredRevisionCASParams{ + AgentAccountID: string(agent), + ID: subscriptionID, + DeliveredRevision: next, + DeliveredRevision_2: prior, + }) + if err != nil { + return false, fmt.Errorf("store: advance forge delivered revision cas: %w", err) + } + return affected > 0, nil +} diff --git a/go/internal/store/forge_subscriptions_pgtest_test.go b/go/internal/store/forge_subscriptions_pgtest_test.go index 195c79cd..3feb7b37 100644 --- a/go/internal/store/forge_subscriptions_pgtest_test.go +++ b/go/internal/store/forge_subscriptions_pgtest_test.go @@ -782,6 +782,107 @@ func TestAdvanceForgeDeliveredRevision(t *testing.T) { } } +// ── T2: AdvanceForgeDeliveredRevisionCAS ────────────────────────────────────── + +// TestAdvanceForgeDeliveredRevisionCAS: the compare-and-set advance lands only +// when the stored delivered_revision still equals prior. A matching prior +// advances (true); a stale prior is a lost CAS (false, no error, row untouched); +// an unknown/foreign id is (false, nil) — never folded into ErrNotFound so the +// router distinguishes a lost CAS from a store fault; empty args are +// ErrInvalidArgument. +func TestAdvanceForgeDeliveredRevisionCAS(t *testing.T) { + ctx := context.Background() + s := newTestStore(t) + owner, _ := seedAgent(t, s, "t2-cas-owner") + foreign, _ := seedAgent(t, s, "t2-cas-foreign") + + id, err := s.EnsureAgentForgeSubscription(ctx, AgentForgeSubscription{ + AgentAccountID: owner, Provider: ForgeProviderGitHub, Host: "github.com", + Repo: "a/b", Kind: ForgeArtifactKindIssue, Number: 1, Scope: ForgeSubscriptionScopeArtifact, + }) + if err != nil { + t.Fatalf("ensure: %v", err) + } + + readRevision := func() string { + t.Helper() + var got string + if err := s.pool.QueryRow(ctx, + `SELECT delivered_revision FROM agent_forge_subscriptions WHERE id = $1`, id, + ).Scan(&got); err != nil { + t.Fatalf("read delivered_revision: %v", err) + } + return got + } + + // Caught up: prior "" (the fresh default) matches, so the CAS advances. + advanced, err := s.AdvanceForgeDeliveredRevisionCAS(ctx, owner, id, "", "rev-1") + if err != nil { + t.Fatalf("cas caught-up: %v", err) + } + if !advanced { + t.Fatal("cas caught-up: advanced = false, want true") + } + if got := readRevision(); got != "rev-1" { + t.Fatalf("delivered_revision = %q, want rev-1", got) + } + + // Stale prior: the row now holds rev-1, so a CAS with prior "" is a lost CAS + // — no advance, no error, row untouched. + advanced, err = s.AdvanceForgeDeliveredRevisionCAS(ctx, owner, id, "", "rev-2") + if err != nil { + t.Fatalf("cas stale prior: %v", err) + } + if advanced { + t.Fatal("cas stale prior: advanced = true, want false (lost CAS)") + } + if got := readRevision(); got != "rev-1" { + t.Fatalf("delivered_revision after lost CAS = %q, want rev-1 (untouched)", got) + } + + // Matching prior again advances forward. + advanced, err = s.AdvanceForgeDeliveredRevisionCAS(ctx, owner, id, "rev-1", "rev-2") + if err != nil { + t.Fatalf("cas advance forward: %v", err) + } + if !advanced { + t.Fatal("cas advance forward: advanced = false, want true") + } + if got := readRevision(); got != "rev-2" { + t.Fatalf("delivered_revision = %q, want rev-2", got) + } + + // Foreign agent on a real id, even with the right prior: no advance, no + // error (the agent scoping fails the WHERE, so zero rows), row untouched. + advanced, err = s.AdvanceForgeDeliveredRevisionCAS(ctx, foreign, id, "rev-2", "rev-x") + if err != nil { + t.Fatalf("cas foreign agent: %v", err) + } + if advanced { + t.Fatal("cas foreign agent: advanced = true, want false") + } + if got := readRevision(); got != "rev-2" { + t.Fatalf("delivered_revision after foreign CAS = %q, want rev-2 (untouched)", got) + } + + // Unknown id: no advance, no error. + advanced, err = s.AdvanceForgeDeliveredRevisionCAS(ctx, owner, "no-such-id", "", "rev-x") + if err != nil { + t.Fatalf("cas unknown id: %v", err) + } + if advanced { + t.Fatal("cas unknown id: advanced = true, want false") + } + + // Empty agent / empty subscription id -> ErrInvalidArgument (early guards). + if _, err := s.AdvanceForgeDeliveredRevisionCAS(ctx, "", id, "", "rev-x"); !errors.Is(err, ErrInvalidArgument) { + t.Fatalf("cas empty agent: err = %v, want ErrInvalidArgument", err) + } + if _, err := s.AdvanceForgeDeliveredRevisionCAS(ctx, owner, "", "", "rev-x"); !errors.Is(err, ErrInvalidArgument) { + t.Fatalf("cas empty id: err = %v, want ErrInvalidArgument", err) + } +} + // ── T7a: LoadForgeArtifactCursor point-read ─────────────────────────────────── // TestLoadForgeArtifactCursor: a written cursor round-trips through the diff --git a/go/internal/store/queries/forge_subscriptions.sql b/go/internal/store/queries/forge_subscriptions.sql index 1ac4150c..493402ba 100644 --- a/go/internal/store/queries/forge_subscriptions.sql +++ b/go/internal/store/queries/forge_subscriptions.sql @@ -93,3 +93,14 @@ WHERE forge_provider = $1 AND forge_host = $2 AND repo = $3 AND kind = $4 AND nu UPDATE agent_forge_subscriptions SET delivered_revision = $3, delivered_at = now() WHERE id = $2 AND agent_account_id = $1; + +-- name: AdvanceForgeDeliveredRevisionCAS :execrows +-- Compare-and-set advance for the notify-router suppress path: the write lands +-- only when delivered_revision still equals $4 (the prior value the router read), +-- so a concurrent route cannot erase a delivery gap it did not observe. Scoped to +-- the owning agent (id AND agent_account_id). Zero rows affected is a lost CAS +-- (someone else advanced first), NOT an error — the wrapper reports it as +-- advanced=false. +UPDATE agent_forge_subscriptions + SET delivered_revision = $3, delivered_at = now() + WHERE id = $2 AND agent_account_id = $1 AND delivered_revision = $4; diff --git a/go/server/forge_notify_matrix_test.go b/go/server/forge_notify_matrix_test.go index d15bcf82..70fd2835 100644 --- a/go/server/forge_notify_matrix_test.go +++ b/go/server/forge_notify_matrix_test.go @@ -324,6 +324,13 @@ func (f *matrixNotifyStore) UpsertArtifactCursor(_ context.Context, cur ingest.A return nil } +// AdvanceDeliveredRevisionCAS satisfies the widened NotifyStore seam. The matrix +// suite never exercises self-origin suppression (nil resolver), so the router +// never calls it; a no-op that reports no advance is correct here. +func (f *matrixNotifyStore) AdvanceDeliveredRevisionCAS(context.Context, string, string, string, string) (bool, error) { + return false, nil +} + // matrixDispatcher records every notification dispatched, per account. type matrixDispatcher struct { sent []*compassv1internal.ForgeNotification diff --git a/go/server/forge_notify_pgtest_test.go b/go/server/forge_notify_pgtest_test.go index 49dc9a42..c3c4c356 100644 --- a/go/server/forge_notify_pgtest_test.go +++ b/go/server/forge_notify_pgtest_test.go @@ -113,6 +113,26 @@ func deliveredRevision(t *testing.T, st *store.Store, agent store.AccountID, sub return "" } +// deliveredRevisionAt is deliveredRevision's provider-bound sibling: it reads one +// subscription's DELIVERY cursor for an arbitrary (provider, host), so a Linear +// container-scope sub can be asserted the same way. +func deliveredRevisionAt(t *testing.T, st *store.Store, provider store.ForgeProvider, host string, subID string) string { + t.Helper() + targets, err := st.ListForgeNotifyTargets(context.Background(), provider, host) + if err != nil { + t.Fatalf("ListForgeNotifyTargets: %v", err) + } + for _, tg := range targets { + for _, s := range tg.Subscribers { + if s.SubscriptionID == subID { + return s.DeliveredRevision + } + } + } + t.Fatalf("subscription %q not found among notify targets", subID) + return "" +} + // commentEvent builds a GitHub issue-comment ForgeEvent at the coordinate — the // simplest non-CHECKS event, so the router never touches the checks roller. func notifyCommentEvent(repo string, number uint64, url string) forge.ForgeEvent { @@ -422,3 +442,169 @@ func TestLinearNotifyRoutedOpenedFansOutToProject(t *testing.T) { t.Fatal("fetch cursor did not advance after the Linear OPENED route") } } + +// --- test: suppress-path delivery-cursor advance over the real store adapter -- + +// scriptedIdentityResolver is the ingest.IdentityResolver fake for the T2 store +// pgtests: it resolves each seeded account id to its owner-qualified handle and a +// single author handle for OPENED. It lets a self-comment event's actor match the +// subscriber so the suppress path fires against the real store adapter. +type scriptedIdentityResolver struct { + accounts map[string]ingest.Handle + author ingest.Handle +} + +func (r *scriptedIdentityResolver) HandleForAccount(_ context.Context, accountID string) (ingest.Handle, error) { + return r.accounts[accountID], nil +} + +func (r *scriptedIdentityResolver) AuthorHandle(_ context.Context, _ string, _ compassv1internal.ForgeArtifactKind, _ uint64) (ingest.Handle, error) { + return r.author, nil +} + +// selfCommentEvent builds a GitHub issue-comment event whose Compass commenter is +// owner-qualified (owner/agent), so the router resolves an actor handle that can +// match a subscriber. +func selfCommentEvent(repo string, number uint64, url, owner, agent string) forge.ForgeEvent { + ev := notifyCommentEvent(repo, number, url) + ev.Comment.Agent = &compassv1.AgentAttribution{AgentHandle: agent, OwnerHandle: owner} + return ev +} + +// TestForgeNotifySuppressAdvancesCaughtUpCursor drives the assembled router over +// the REAL store adapter: a self-comment from the subscribing agent is suppressed +// (no dispatch) AND, because the subscriber is caught up to the prior cursor +// revision, its delivered_revision advances to the route revision through the +// CAS. A first delivered comment establishes the caught-up state (its ack is +// simulated by advancing delivered_revision to the cursor revision); the second, +// self-authored comment is the suppressed one. +func TestForgeNotifySuppressAdvancesCaughtUpCursor(t *testing.T) { + st := forgeTestStore(t) + ctx := context.Background() // test root + const ( + repo = "a/b" + number = uint64(42) + owner = "own" + agent = "atlas" + ) + agentID, subID := seedNotifySubscription(t, st, repo, store.ForgeArtifactKindIssue, number) + + notifyStore := &forgeNotifyStore{st: st, provider: store.ForgeProviderGitHub, host: forgeTestHost} + disp := &recordingDispatcher{} + forgeRef := &compassv1.ForgeRef{Provider: compassv1.ForgeProvider_FORGE_PROVIDER_GITHUB, Host: forgeTestHost} + ids := &scriptedIdentityResolver{accounts: map[string]ingest.Handle{string(agentID): {Owner: owner, Agent: agent}}} + router := ingest.NewNotifyRouter(notifyStore, disp, fixedChecksRoller{}, nil, ids, forgeRef, nil) + + // A human comment first: delivered, advancing the shared cursor. Simulate the + // agent's ack so its delivered_revision catches up to the cursor revision. + if err := router.Route(ctx, notifyCommentEvent(repo, number, "https://github.com/a/b/issues/42#c1")); err != nil { + t.Fatalf("Route (human comment): %v", err) + } + cur, err := st.LoadForgeArtifactCursor(ctx, store.ForgeProviderGitHub, forgeTestHost, repo, store.ForgeArtifactKindIssue, number) + if err != nil || cur == nil { + t.Fatalf("LoadForgeArtifactCursor: %v (cur=%v)", err, cur) + } + if err := st.AdvanceForgeDeliveredRevision(ctx, agentID, subID, cur.Revision); err != nil { + t.Fatalf("simulate ack: %v", err) + } + if got := deliveredRevision(t, st, agentID, subID); got != cur.Revision { + t.Fatalf("precondition: delivered_revision = %q, want the caught-up cursor revision %q", got, cur.Revision) + } + + // The self-comment: suppressed, and the caught-up subscriber advances. + if err := router.Route(ctx, selfCommentEvent(repo, number, "https://github.com/a/b/issues/42#c2", owner, agent)); err != nil { + t.Fatalf("Route (self comment): %v", err) + } + // No new dispatch for the self-comment (only the first human comment). + if len(disp.sent) != 1 { + t.Fatalf("dispatched notifications = %d, want 1 (the self-comment is suppressed)", len(disp.sent)) + } + after, err := st.LoadForgeArtifactCursor(ctx, store.ForgeProviderGitHub, forgeTestHost, repo, store.ForgeArtifactKindIssue, number) + if err != nil || after == nil { + t.Fatalf("LoadForgeArtifactCursor (post): %v (cur=%v)", err, after) + } + if got := deliveredRevision(t, st, agentID, subID); got != after.Revision { + t.Fatalf("delivered_revision = %q after suppress, want the advanced route revision %q", got, after.Revision) + } +} + +// TestForgeNotifySuppressDoesNotAdvanceTrailingCursor is the forward-masking +// guard over the real store: a self-comment is suppressed, but the subscriber +// TRAILS the prior cursor revision (an undelivered earlier event), so its +// delivered_revision does NOT advance — the CAS gate leaves the gap for the +// sweep. The subscriber never acked the first comment, so it stays trailing. +func TestForgeNotifySuppressDoesNotAdvanceTrailingCursor(t *testing.T) { + st := forgeTestStore(t) + ctx := context.Background() // test root + const ( + repo = "a/b" + number = uint64(43) + owner = "own" + agent = "atlas" + ) + agentID, subID := seedNotifySubscription(t, st, repo, store.ForgeArtifactKindIssue, number) + + notifyStore := &forgeNotifyStore{st: st, provider: store.ForgeProviderGitHub, host: forgeTestHost} + disp := &recordingDispatcher{} + forgeRef := &compassv1.ForgeRef{Provider: compassv1.ForgeProvider_FORGE_PROVIDER_GITHUB, Host: forgeTestHost} + ids := &scriptedIdentityResolver{accounts: map[string]ingest.Handle{string(agentID): {Owner: owner, Agent: agent}}} + router := ingest.NewNotifyRouter(notifyStore, disp, fixedChecksRoller{}, nil, ids, forgeRef, nil) + + // A human comment: delivered, cursor advances — but the agent NEVER acks, so + // delivered_revision stays "" while the cursor moved ahead (the trailing gap). + if err := router.Route(ctx, notifyCommentEvent(repo, number, "https://github.com/a/b/issues/43#c1")); err != nil { + t.Fatalf("Route (human comment): %v", err) + } + if got := deliveredRevision(t, st, agentID, subID); got != "" { + t.Fatalf("precondition: delivered_revision = %q, want empty (trailing, unacked)", got) + } + + // The self-comment: suppressed, but the trailing subscriber must NOT advance. + if err := router.Route(ctx, selfCommentEvent(repo, number, "https://github.com/a/b/issues/43#c2", owner, agent)); err != nil { + t.Fatalf("Route (self comment): %v", err) + } + if got := deliveredRevision(t, st, agentID, subID); got != "" { + t.Fatalf("delivered_revision = %q after suppress, want empty (trailing sub NOT advanced — the E1 gap survives for the sweep)", got) + } +} + +// TestForgeNotifySuppressNeverAdvancesContainerCursor is the scope carve-out over +// the real store: a self-authored OPENED to a CONTAINER-scope subscriber is +// suppressed but its container delivery cursor is NEVER advanced (an artifact +// revision in the container row would poison the container sweep). The container +// sub is caught up ("" == "") so only the scope check prevents the advance. +func TestForgeNotifySuppressNeverAdvancesContainerCursor(t *testing.T) { + st := forgeTestStore(t) + ctx := context.Background() // test root + const ( + repo = "RIG" + host = "linear.app" + number = uint64(77) + project = "proj-A" + owner = "own" + agent = "atlas" + url = "https://linear.app/rig/issue/RIG-77" + ) + agentID, subID := seedLinearContainerSub(t, st, "lin-self", repo, project) + + notifyStore := &forgeNotifyStore{st: st, provider: store.ForgeProviderLinear, host: host} + disp := &recordingDispatcher{} + forgeRef := &compassv1.ForgeRef{Provider: compassv1.ForgeProvider_FORGE_PROVIDER_LINEAR, Host: host} + ids := &scriptedIdentityResolver{ + accounts: map[string]ingest.Handle{string(agentID): {Owner: owner, Agent: agent}}, + author: ingest.Handle{Owner: owner, Agent: agent}, // the OPENED author IS the subscriber. + } + router := ingest.NewNotifyRouter(notifyStore, disp, fixedChecksRoller{}, nil, ids, forgeRef, nil) + + if err := router.Route(ctx, linearOpenedEvent(repo, number, project, url)); err != nil { + t.Fatalf("Route (self OPENED): %v", err) + } + // Suppressed: no dispatch. + if len(disp.sent) != 0 { + t.Fatalf("dispatched notifications = %d, want 0 (self-opened suppressed)", len(disp.sent)) + } + // The container subscriber's delivery cursor is untouched (never advanced). + if got := deliveredRevisionAt(t, st, store.ForgeProviderLinear, host, subID); got != "" { + t.Fatalf("container delivered_revision = %q after suppress, want empty (NEVER advanced — poisons the container sweep)", got) + } +} diff --git a/go/server/serve.go b/go/server/serve.go index 134d3a46..0516abef 100644 --- a/go/server/serve.go +++ b/go/server/serve.go @@ -1553,6 +1553,14 @@ func (a *forgeNotifyStore) UpsertArtifactCursor(ctx context.Context, cur ingest. }) } +// AdvanceDeliveredRevisionCAS backs the router's suppress-path cursor advance +// over the store's compare-and-set writer. A lost CAS (advanced=false, nil err) +// is passed through unchanged so the router degrades open rather than treating +// it as a fault. +func (a *forgeNotifyStore) AdvanceDeliveredRevisionCAS(ctx context.Context, agentAccountID, subscriptionID, prior, next string) (bool, error) { + return a.st.AdvanceForgeDeliveredRevisionCAS(ctx, store.AccountID(agentAccountID), subscriptionID, prior, next) +} + // toIngestSubscribers converts the store subscriber rows to the ingest mirror // (the no-store rule keeps the store type out of the ingest package). func toIngestSubscribers(subs []store.ForgeNotifySubscriber) []ingest.NotifySubscriber { From 7750efb382df88ece34058b83c3c94b63005a456 Mon Sep 17 00:00:00 2001 From: mintaka Date: Sat, 12 Sep 2026 18:57:28 -0400 Subject: [PATCH 2/2] fix(ingest): log the coordinate on a suppress-advance failure (RIG-3326) Both suppress-advance warn logs carried only the subscription and account, so a failure could not be tied to the artifact it happened on without cross-referencing the subscription row. The sibling dispatch-failed log already carries repo and number; match it. Co-authored-by: Matt Wilkinson --- go/internal/ingest/notify_router.go | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/go/internal/ingest/notify_router.go b/go/internal/ingest/notify_router.go index 39f8e9b6..fca1c048 100644 --- a/go/internal/ingest/notify_router.go +++ b/go/internal/ingest/notify_router.go @@ -349,7 +349,7 @@ func (r *NotifyRouter) Route(ctx context.Context, ev forge.ForgeEvent) error { subMemo := map[string]Handle{} for _, sub := range subs { if r.selfOrigin(ctx, actor, sub, subMemo) { - r.advanceOnSuppress(ctx, sub, priorRevision, revision) + r.advanceOnSuppress(ctx, ev, sub, priorRevision, revision) continue } n := r.notification(ev, sub.SubscriptionID, revision) @@ -420,7 +420,7 @@ func (r *NotifyRouter) SynthesizeUpdate(ctx context.Context, sub NotifySubscribe // A lost CAS (advanced=false) or any store fault degrades OPEN — logged and // swallowed, worst case one synthetic UPDATE on the next sweep — never a route // failure. -func (r *NotifyRouter) advanceOnSuppress(ctx context.Context, sub NotifySubscriber, priorRevision, next string) { +func (r *NotifyRouter) advanceOnSuppress(ctx context.Context, ev forge.ForgeEvent, sub NotifySubscriber, priorRevision, next string) { if sub.Scope != compassv1internal.ForgeSubscriptionScope_FORGE_SUBSCRIPTION_SCOPE_ARTIFACT { return } @@ -432,13 +432,15 @@ func (r *NotifyRouter) advanceOnSuppress(ctx context.Context, sub NotifySubscrib r.log.WarnContext(ctx, "forge notify suppress-advance failed", "subscription_id", sub.SubscriptionID, "account", sub.AgentAccountID, + "repo", ev.Repo, "number", ev.Number, "error", err) return } if !advanced { r.log.WarnContext(ctx, "forge notify suppress-advance lost CAS", "subscription_id", sub.SubscriptionID, - "account", sub.AgentAccountID) + "account", sub.AgentAccountID, + "repo", ev.Repo, "number", ev.Number) } }