From bd6c045c0b4c12cda813302c91f62a5c0b4070e3 Mon Sep 17 00:00:00 2001 From: mintaka Date: Thu, 27 Aug 2026 23:30:17 -0400 Subject: [PATCH 1/2] feat(store): watermark persistence + enabled-repo enumeration (RIG-2883 T4) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The board webhook lane's durable state, additive on the pre-live `0001_init.sql`. - `forge_repo_subscriptions` gains `swept_updated_at` (the per-repo updated-at watermark; NULL = never swept) and `list_etag` (the conditional-GET etag for the repo LIST walk), plus `LoadForgeRepoWatermark`/`StoreForgeRepoWatermark` (advance-after-sink; unknown coordinate reads zero/empty, store on a missing row is `ErrNotFound`), `ListEnabledForgeRepos` (ascending cross-coordinate enumeration for the reconciler), and `IsEnabledForgeRepo` (the point membership check the webhook arm gates on). - `issues` gains `forge_updated_at TIMESTAMPTZ`, INERT this slice — the OQ-6(a) recency-guard column. T4a threads its write path and makes the upsert conditional; the bare column is a no-op until then. The `forge_list_cursors` table and its store methods (`ForgeListCursor`/`UpsertForgeListCursorPage`/`PruneForgeListCursorPages`) stay: they retire atomically with their `serve.go` poll-driver consumer in T5, so this slice never breaks the tree build. pgtest covers the watermark round-trip, coordinate isolation, enabled-repo enumeration + point membership, and that the edited migration applies on a fresh database with the three new columns present. Spec-impact: none. Refs RIG-2883 Co-authored-by: Matt Wilkinson --- go/internal/store/forge_cursors.go | 133 ++++++++++++-- .../store/forge_cursors_pgtest_test.go | 163 +++++++++++++++++- go/internal/store/migrations/0001_init.sql | 10 +- 3 files changed, 291 insertions(+), 15 deletions(-) diff --git a/go/internal/store/forge_cursors.go b/go/internal/store/forge_cursors.go index c647904b7..bb814a44e 100644 --- a/go/internal/store/forge_cursors.go +++ b/go/internal/store/forge_cursors.go @@ -2,19 +2,23 @@ package store import ( "context" + "errors" "fmt" + "time" + + "github.com/jackc/pgx/v5" ) -// The forge poll driver's durable state (SEA-1810 T2, design -// docs/designs/product/compass-forge-poll-driver/design.md §T2): the repo-LIST -// per-page FETCH cursor (forge_list_cursors) and the board's per-REPO poll -// targets (forge_repo_subscriptions). The two DL-053 anticipatory tables -// (agent_forge_subscriptions, forge_artifact_cursors) are writer-less this -// slice and get their store surface with their writers. +// The board arm's durable state (RIG-2883): the per-REPO poll targets and their +// swept-updated-at watermark (forge_repo_subscriptions), plus the poll driver's +// per-page FETCH cursor (forge_list_cursors) — the latter retires atomically +// with its serve.go consumer in T5, so it survives this additive slice. The two +// DL-053 anticipatory tables (agent_forge_subscriptions, forge_artifact_cursors) +// are writer-less this slice and get their store surface with their writers. // ForgeListPageCursor is one durable page row of a repo's issue-LIST fetch // cursor (the DL-053 FETCH-cursor model at repo-LIST granularity). ETag "" -// means never fetched (an unconditional GET). +// means never fetched (an unconditional GET). Retires with the poll driver (T5). type ForgeListPageCursor struct { Provider ForgeProvider // GITHUB(1)/GITLAB(2)/FORGEJO(3)/LINEAR(4); never 0 Host string @@ -24,9 +28,9 @@ type ForgeListPageCursor struct { HasNext bool } -// ForgeRepoSubscription is one board poll target: a repo the poll driver walks +// ForgeRepoSubscription is one board poll target: a repo the board arm walks // (OQ-C's table model). Enabled=false soft-disables the target without deleting -// its cursor history. +// its watermark history. type ForgeRepoSubscription struct { Provider ForgeProvider Host string @@ -133,8 +137,65 @@ func (s *Store) PruneForgeListCursorPages(ctx context.Context, provider ForgePro return nil } +// LoadForgeRepoWatermark reads the repo's swept_updated_at watermark and its +// conditional-GET list_etag. A never-swept repo (swept_updated_at IS NULL) +// returns the zero time.Time; an unknown coordinate is not an error — it too +// returns the zero watermark and empty etag (the reconciler treats "no row" and +// "never swept" identically, walking from the beginning). Zero/empty coordinate +// fields -> ErrInvalidArgument. +func (s *Store) LoadForgeRepoWatermark(ctx context.Context, provider ForgeProvider, host, repo string) (time.Time, string, error) { + if err := validCoordinate(provider, host, repo); err != nil { + return time.Time{}, "", err + } + var swept *time.Time + var etag string + err := s.pool.QueryRow(ctx, + `SELECT swept_updated_at, list_etag + FROM forge_repo_subscriptions + WHERE forge_provider = $1 AND forge_host = $2 AND repo = $3`, + int32(provider), host, repo, + ).Scan(&swept, &etag) + if errors.Is(err, pgx.ErrNoRows) { + return time.Time{}, "", nil + } + if err != nil { + return time.Time{}, "", fmt.Errorf("store: load forge repo watermark: %w", err) + } + if swept == nil { + return time.Time{}, etag, nil + } + return *swept, etag, nil +} + +// StoreForgeRepoWatermark writes the repo's swept_updated_at watermark and +// list_etag, touching updated_at. An unknown coordinate -> ErrNotFound (the +// subscription must exist — the seed/upsert path owns row creation). Zero/empty +// coordinate fields -> ErrInvalidArgument. +func (s *Store) StoreForgeRepoWatermark(ctx context.Context, provider ForgeProvider, host, repo string, mark time.Time, etag string) error { + if err := validCoordinate(provider, host, repo); err != nil { + return err + } + var swept *time.Time + if !mark.IsZero() { + swept = &mark + } + tag, err := s.pool.Exec(ctx, + `UPDATE forge_repo_subscriptions + SET swept_updated_at = $4, list_etag = $5, updated_at = now() + WHERE forge_provider = $1 AND forge_host = $2 AND repo = $3`, + int32(provider), host, repo, swept, etag, + ) + if err != nil { + return fmt.Errorf("store: store forge repo watermark: %w", err) + } + if tag.RowsAffected() == 0 { + return fmt.Errorf("%w: forge repo subscription (%d, %q, %q)", ErrNotFound, provider, host, repo) + } + return nil +} + // EnsureForgeRepoSubscription inserts the target if absent; on conflict it DOES -// NOTHING — the T4 seed reconcile is a bootstrap-only insert and the table is +// NOTHING — the seed reconcile is a bootstrap-only insert and the table is // authoritative after the first insert (the seed never deletes, disables, or // re-enables an existing row). Zero/empty coordinate fields -> ErrInvalidArgument. func (s *Store) EnsureForgeRepoSubscription(ctx context.Context, sub ForgeRepoSubscription) error { @@ -152,9 +213,57 @@ func (s *Store) EnsureForgeRepoSubscription(ctx context.Context, sub ForgeRepoSu return nil } +// ListEnabledForgeRepos reads every enabled target's repo, ascending — the board +// reconciler's per-pass target enumeration across all coordinates. No rows is a +// nil slice, not an error. +func (s *Store) ListEnabledForgeRepos(ctx context.Context) ([]string, error) { + rows, err := s.pool.Query(ctx, + `SELECT repo + FROM forge_repo_subscriptions + WHERE enabled = TRUE + ORDER BY repo ASC`, + ) + if err != nil { + return nil, fmt.Errorf("store: list enabled forge repos: %w", err) + } + defer rows.Close() + + var out []string + for rows.Next() { + var repo string + if err := rows.Scan(&repo); err != nil { + return nil, fmt.Errorf("store: scan forge repo: %w", err) + } + out = append(out, repo) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("store: iterate forge repos: %w", err) + } + return out, nil +} + +// IsEnabledForgeRepo reports whether an enabled subscription exists for the repo +// (the point membership check the webhook arm gates on). An empty repo -> +// ErrInvalidArgument. +func (s *Store) IsEnabledForgeRepo(ctx context.Context, repo string) (bool, error) { + if repo == "" { + return false, fmt.Errorf("%w: repo is required", ErrInvalidArgument) + } + var exists bool + if err := s.pool.QueryRow(ctx, + `SELECT EXISTS ( + SELECT 1 FROM forge_repo_subscriptions + WHERE repo = $1 AND enabled = TRUE)`, + repo, + ).Scan(&exists); err != nil { + return false, fmt.Errorf("store: is enabled forge repo: %w", err) + } + return exists, nil +} + // ListEnabledForgeRepoSubscriptions reads the enabled targets for one (provider, -// host), ascending repo — the driver's per-pass target enumeration. No rows is a -// nil slice, not an error. Zero provider / empty host -> ErrInvalidArgument. +// host), ascending repo. No rows is a nil slice, not an error. Zero provider / +// empty host -> ErrInvalidArgument. func (s *Store) ListEnabledForgeRepoSubscriptions(ctx context.Context, provider ForgeProvider, host string) ([]ForgeRepoSubscription, error) { if provider == ForgeProviderUnspecified { return nil, fmt.Errorf("%w: forge provider is required", ErrInvalidArgument) diff --git a/go/internal/store/forge_cursors_pgtest_test.go b/go/internal/store/forge_cursors_pgtest_test.go index 7603ed98b..75b8de3c2 100644 --- a/go/internal/store/forge_cursors_pgtest_test.go +++ b/go/internal/store/forge_cursors_pgtest_test.go @@ -26,8 +26,9 @@ func TestMigration0016TablesExist(t *testing.T) { ctx := context.Background() s := newTestStore(t) - // forge_repo_subscriptions - mustExec(t, s, `INSERT INTO forge_repo_subscriptions (forge_provider, forge_host, repo) VALUES (1, 'github.com', 'a/b')`) + // forge_repo_subscriptions — including the RIG-2883 T4 columns + // (swept_updated_at watermark + list_etag), proving they applied. + mustExec(t, s, `INSERT INTO forge_repo_subscriptions (forge_provider, forge_host, repo, swept_updated_at, list_etag) VALUES (1, 'github.com', 'a/b', now(), '"e"')`) // forge_list_cursors mustExec(t, s, `INSERT INTO forge_list_cursors (forge_provider, forge_host, repo, page) VALUES (1, 'github.com', 'a/b', 1)`) // forge_artifact_cursors — both legal kind values (1=issue, 2=pull_request) @@ -35,6 +36,9 @@ func TestMigration0016TablesExist(t *testing.T) { // provider-domain 1..4 accept test. mustExec(t, s, `INSERT INTO forge_artifact_cursors (forge_provider, forge_host, repo, kind, number) VALUES (1, 'github.com', 'a/b', 1, 7)`) mustExec(t, s, `INSERT INTO forge_artifact_cursors (forge_provider, forge_host, repo, kind, number) VALUES (1, 'github.com', 'a/b', 2, 8)`) + // issues — the RIG-2883 T4 forge_updated_at column is present (INERT this + // slice; T4a threads its write path). Proven present by inserting it. + mustExec(t, s, `INSERT INTO issues (id, forge_provider, forge_host, repo, number, forge_updated_at) VALUES ('i-1', 1, 'github.com', 'a/b', 7, now())`) // agent_forge_subscriptions needs a real agent_account_id (FK); seed one. owner := mustUser(t, s, "forge-owner") agent := mustAgent(t, s, owner.ID, "forge-agent") @@ -373,6 +377,161 @@ func TestForgeCursorInvalidArgument(t *testing.T) { sentinelIs(t, s.SetForgeRepoSubscriptionEnabled(ctx, ForgeProviderGitHub, "h", "", true), ErrInvalidArgument, "set empty repo") } +// ── Test 8: forge_repo_subscriptions watermark round-trip (RIG-2883 T4) ─────── + +// TestForgeRepoWatermarkRoundTrip proves the swept_updated_at + list_etag +// watermark persists: a never-swept row reads zero/empty, an unknown coordinate +// also reads zero/empty (not an error — the reconciler treats "no row" and +// "never swept" identically), a store then reads back, and a store on an unknown +// coordinate is ErrNotFound (the subscription must exist first). +func TestForgeRepoWatermarkRoundTrip(t *testing.T) { + ctx := context.Background() + s := newTestStore(t) + + sub := ForgeRepoSubscription{Provider: ForgeProviderGitHub, Host: "github.com", Repo: "a/b", Enabled: true} + if err := s.EnsureForgeRepoSubscription(ctx, sub); err != nil { + t.Fatalf("ensure: %v", err) + } + + // Never-swept row: zero time, empty etag. + mark, etag, err := s.LoadForgeRepoWatermark(ctx, ForgeProviderGitHub, "github.com", "a/b") + if err != nil { + t.Fatalf("load never-swept: %v", err) + } + if !mark.IsZero() || etag != "" { + t.Fatalf("never-swept = (%v, %q), want (zero, \"\")", mark, etag) + } + + // Unknown coordinate (no row) also reads zero/empty, not an error. + mark, etag, err = s.LoadForgeRepoWatermark(ctx, ForgeProviderGitLab, "gitlab.com", "x/y") + if err != nil { + t.Fatalf("load unknown coordinate: %v", err) + } + if !mark.IsZero() || etag != "" { + t.Fatalf("unknown coordinate = (%v, %q), want (zero, \"\")", mark, etag) + } + + // Store then round-trip. + want := time.Date(2026, 8, 27, 12, 0, 0, 0, time.UTC) + if err := s.StoreForgeRepoWatermark(ctx, ForgeProviderGitHub, "github.com", "a/b", want, `"v1"`); err != nil { + t.Fatalf("store watermark: %v", err) + } + mark, etag, err = s.LoadForgeRepoWatermark(ctx, ForgeProviderGitHub, "github.com", "a/b") + if err != nil { + t.Fatalf("load after store: %v", err) + } + if !mark.Equal(want) { + t.Fatalf("watermark = %v, want %v", mark, want) + } + if etag != `"v1"` { + t.Fatalf("etag = %q, want %q", etag, `"v1"`) + } + + // Storing on an unknown coordinate -> ErrNotFound (the subscription must exist). + err = s.StoreForgeRepoWatermark(ctx, ForgeProviderGitHub, "github.com", "no/such", want, "") + sentinelIs(t, err, ErrNotFound, "store watermark on unknown coordinate") +} + +// ── Test 9: watermark coordinate isolation across (provider, host) ──────────── + +func TestForgeRepoWatermarkCoordinateIsolation(t *testing.T) { + ctx := context.Background() + s := newTestStore(t) + + const repo = "a/b" + subs := []struct { + provider ForgeProvider + host string + mark time.Time + etag string + }{ + {ForgeProviderGitHub, "github.com", time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC), `"gh"`}, + {ForgeProviderGitHub, "ghe.example.com", time.Date(2026, 2, 2, 0, 0, 0, 0, time.UTC), `"ghe"`}, + {ForgeProviderGitLab, "github.com", time.Date(2026, 3, 3, 0, 0, 0, 0, time.UTC), `"gl"`}, + } + for _, c := range subs { + if err := s.EnsureForgeRepoSubscription(ctx, ForgeRepoSubscription{Provider: c.provider, Host: c.host, Repo: repo, Enabled: true}); err != nil { + t.Fatalf("ensure (%d,%q): %v", c.provider, c.host, err) + } + if err := s.StoreForgeRepoWatermark(ctx, c.provider, c.host, repo, c.mark, c.etag); err != nil { + t.Fatalf("store (%d,%q): %v", c.provider, c.host, err) + } + } + // Each coordinate reads back exactly its own watermark and etag. + for _, c := range subs { + mark, etag, err := s.LoadForgeRepoWatermark(ctx, c.provider, c.host, repo) + if err != nil { + t.Fatalf("load (%d,%q): %v", c.provider, c.host, err) + } + if !mark.Equal(c.mark) || etag != c.etag { + t.Fatalf("coordinate (%d,%q) = (%v, %q), want (%v, %q)", c.provider, c.host, mark, etag, c.mark, c.etag) + } + } +} + +// ── Test 10: enabled-repo enumeration + point membership (RIG-2883 T4) ──────── + +func TestListAndIsEnabledForgeRepos(t *testing.T) { + ctx := context.Background() + s := newTestStore(t) + + // No rows -> nil, and the point check is false. + repos, err := s.ListEnabledForgeRepos(ctx) + if err != nil { + t.Fatalf("list enabled repos empty: %v", err) + } + if repos != nil { + t.Fatalf("empty = %v, want nil", repos) + } + ok, err := s.IsEnabledForgeRepo(ctx, "a/b") + if err != nil { + t.Fatalf("is-enabled empty: %v", err) + } + if ok { + t.Fatal("is-enabled = true on empty, want false") + } + + // Seed enabled repos out of lexical order across coordinates, plus one + // disabled — enumeration is ascending and excludes the disabled repo. + seed := []ForgeRepoSubscription{ + {Provider: ForgeProviderGitHub, Host: "github.com", Repo: "z/z", Enabled: true}, + {Provider: ForgeProviderGitLab, Host: "gitlab.com", Repo: "a/a", Enabled: true}, + {Provider: ForgeProviderGitHub, Host: "ghe.example.com", Repo: "m/m", Enabled: true}, + } + for _, sub := range seed { + if err := s.EnsureForgeRepoSubscription(ctx, sub); err != nil { + t.Fatalf("ensure %+v: %v", sub, err) + } + } + if err := s.EnsureForgeRepoSubscription(ctx, ForgeRepoSubscription{Provider: ForgeProviderGitHub, Host: "github.com", Repo: "d/d", Enabled: false}); err != nil { + t.Fatalf("ensure disabled: %v", err) + } + + repos, err = s.ListEnabledForgeRepos(ctx) + if err != nil { + t.Fatalf("list enabled repos: %v", err) + } + if len(repos) != 3 || repos[0] != "a/a" || repos[1] != "m/m" || repos[2] != "z/z" { + t.Fatalf("list enabled repos = %v, want [a/a m/m z/z] ascending", repos) + } + + // Point check: an enabled repo is true, the disabled repo is false. + ok, err = s.IsEnabledForgeRepo(ctx, "m/m") + if err != nil { + t.Fatalf("is-enabled m/m: %v", err) + } + if !ok { + t.Fatal("is-enabled m/m = false, want true") + } + ok, err = s.IsEnabledForgeRepo(ctx, "d/d") + if err != nil { + t.Fatalf("is-enabled d/d: %v", err) + } + if ok { + t.Fatal("is-enabled d/d (disabled) = true, want false") + } +} + // ── helpers ─────────────────────────────────────────────────────────────────── func mustExec(t *testing.T, s *Store, sql string) { diff --git a/go/internal/store/migrations/0001_init.sql b/go/internal/store/migrations/0001_init.sql index 5d57e5b6f..28969871d 100644 --- a/go/internal/store/migrations/0001_init.sql +++ b/go/internal/store/migrations/0001_init.sql @@ -586,6 +586,11 @@ CREATE TABLE issues ( labels TEXT[] NOT NULL DEFAULT '{}', agent_handle TEXT NOT NULL DEFAULT '', -- '' = non-Compass author + -- OQ-6(a) recency-guard column (RIG-2883 T4): the forge's last-updated + -- timestamp for the artifact. INERT until T4a threads the write path + -- (Issue.UpdatedAt reaches no writer today); the bare column is a no-op. + forge_updated_at TIMESTAMPTZ, + -- Compass machinery (server-owned; none on the forge). state defaults to -- BACKLOG; CHECK 1..8: a persisted issue is NEVER UNSPECIFIED(0). The -- machinery columns get their writers in later slices. @@ -618,6 +623,8 @@ CREATE TABLE forge_repo_subscriptions ( forge_host TEXT NOT NULL, repo TEXT NOT NULL, enabled BOOLEAN NOT NULL DEFAULT TRUE, + swept_updated_at TIMESTAMPTZ, -- last swept forge updated_at watermark; NULL = never swept + list_etag TEXT NOT NULL DEFAULT '', -- conditional-GET etag for the repo LIST walk created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), PRIMARY KEY (forge_provider, forge_host, repo) @@ -671,7 +678,8 @@ CREATE TABLE forge_artifact_cursors ( -- A durable conditional-GET cache; etag advances ONLY after every row of that -- page's content is durably sunk. has_next persists the Link-chain fact so a 304 -- can keep walking a multi-page repo. advanced_at records the last content --- advance (an etag-storing 200+sink), NOT the last poll. +-- advance (an etag-storing 200+sink), NOT the last poll. Retires with the poll +-- driver (RIG-2883 T5), atomically with its serve.go consumer. CREATE TABLE forge_list_cursors ( forge_provider SMALLINT NOT NULL CHECK (forge_provider IN (1, 2, 3, 4)), forge_host TEXT NOT NULL, From 3fde8b7f8fb7772bf7e761f2c0af9cf3e6e20a50 Mon Sep 17 00:00:00 2001 From: mintaka Date: Fri, 28 Aug 2026 00:53:28 -0400 Subject: [PATCH 2/2] docs(store): note the repo-only keying assumption on enabled-repo checks (RIG-2883 T4) ListEnabledForgeRepos / IsEnabledForgeRepo key on repo alone, matching the frozen repo-keyed ingest seam, while the watermark methods are coordinate-keyed (provider, host, repo). Unambiguous in a github.com-only deployment; documents the latent cross-coordinate ambiguity so a future multi-host enablement threads (provider, host) through the seam first. Refs RIG-2883 Co-authored-by: Matt Wilkinson --- go/internal/store/forge_cursors.go | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/go/internal/store/forge_cursors.go b/go/internal/store/forge_cursors.go index bb814a44e..6369e7144 100644 --- a/go/internal/store/forge_cursors.go +++ b/go/internal/store/forge_cursors.go @@ -216,6 +216,13 @@ func (s *Store) EnsureForgeRepoSubscription(ctx context.Context, sub ForgeRepoSu // ListEnabledForgeRepos reads every enabled target's repo, ascending — the board // reconciler's per-pass target enumeration across all coordinates. No rows is a // nil slice, not an error. +// +// Repo-only keyed (no provider/host), matching the frozen repo-keyed ingest +// seam. In a github.com-only deployment repo is unambiguous; if multi-host is +// ever enabled, two coordinates sharing a repo string (e.g. github.com and a GHE +// host both carrying "a/b") would collapse to one entry here and to an ambiguous +// watermark under the coordinate-keyed Load/Store methods — thread (provider, +// host) through this seam before enabling multi-host. func (s *Store) ListEnabledForgeRepos(ctx context.Context) ([]string, error) { rows, err := s.pool.Query(ctx, `SELECT repo @@ -244,7 +251,9 @@ func (s *Store) ListEnabledForgeRepos(ctx context.Context) ([]string, error) { // IsEnabledForgeRepo reports whether an enabled subscription exists for the repo // (the point membership check the webhook arm gates on). An empty repo -> -// ErrInvalidArgument. +// ErrInvalidArgument. Repo-only keyed like ListEnabledForgeRepos — returns true +// if ANY coordinate's subscription for the repo is enabled; unambiguous in a +// github.com-only deployment (see ListEnabledForgeRepos for the multi-host note). func (s *Store) IsEnabledForgeRepo(ctx context.Context, repo string) (bool, error) { if repo == "" { return false, fmt.Errorf("%w: repo is required", ErrInvalidArgument)