diff --git a/go/internal/ingest/notify_router.go b/go/internal/ingest/notify_router.go index 8aa58e85..fca1c048 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, ev, sub, priorRevision, revision) continue } n := r.notification(ev, sub.SubscriptionID, revision) @@ -373,6 +395,55 @@ 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, ev forge.ForgeEvent, 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, + "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, + "repo", ev.Repo, "number", ev.Number) + } +} + // 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 {