fix: readiness check, no-op graph publish, NUL-stranded runs, and an invalidated test

backend/fly.toml

@@ -46,10 +46,30 @@ swap_size_mb = 512
   size = 'shared-cpu-1x'
   memory = '1024mb'
 
+# Two checks, two questions. `health` asks "is the process alive?" and hits
+# /healthz, which returns a constant 200 and touches nothing. `ready` asks "can
+# this machine actually serve?" and hits /readyz, which pings the store pool.
+#
+# Only /healthz was wired up until 2026-09-03. That day argus-db filled its
+# volume and Postgres crash-looped for three days; /healthz kept returning
+# {"status":"ok"} the whole time, so Fly reported the app healthy while every
+# data request failed. A dead database must turn a check red.
+#
+# `ready` carries the longer grace_period because a booting machine opens the
+# pool after the listener, and the longer interval because each probe costs a
+# round trip to Postgres.
 [checks]
   [checks.health]
     port = 8080
     type = 'http'
     interval = '15s'
     timeout = '2s'
     path = '/healthz'
+
+  [checks.ready]
+    port = 8080
+    type = 'http'
+    interval = '30s'
+    timeout = '5s'
+    grace_period = '20s'
+    path = '/readyz'

backend/internal/graph/fullindex.go

@@ -498,6 +498,65 @@ func failGraphPublishLimit(ctx context.Context, st *store.Store, tx pgx.Tx, repo
 	return limitErr
 }
 
+// stageGenerationIsNoOp reports whether publishing this generation would leave
+// code_nodes and code_edges exactly as they already are, making the
+// delete-and-reinsert swap pure churn. It also records this generation's
+// content hash, so the next publish has something to compare against.
+//
+// Two conditions must hold together:
+//
+//   - The staged payloads hash identical to the generation currently published
+//     for this repo. Same staged input, same derived graph.
+//   - Nothing has written to code_nodes or code_edges since that publish. The
+//     PR indexer mutates both between full publishes, and a swap removes the
+//     rows it added. Skipping while they exist would leave the live graph
+//     diverged from the generation that claims to describe it.
+//
+// An unset hash on either side reads as unknown and publishes. Every uncertain
+// case falls through to the swap, so this can only skip work it has proven is
+// redundant.
+func stageGenerationIsNoOp(ctx context.Context, tx pgx.Tx, repoID, generationID int64) (bool, error) {
+	// Hashing each row first keeps the aggregate to 32 bytes per file instead of
+	// concatenating every staged payload into one value.
+	var hash string
+	if err := tx.QueryRow(ctx, `
+		SELECT COALESCE(md5(string_agg(h, '' ORDER BY file_path)), '')
+		FROM (
+		  SELECT file_path,
+		         md5(file_path || symbols::text || edges::text || endpoints::text) AS h
+		  FROM graph_index_generation_files
+		  WHERE generation_id = $1 AND status = 'ready'
+		) staged`, generationID).Scan(&hash); err != nil {
+		return false, fmt.Errorf("publish graph generation: hash staged files: %w", err)
+	}
+	if _, err := tx.Exec(ctx, `UPDATE graph_index_generations SET content_hash = $2
+		WHERE id = $1`, generationID, hash); err != nil {
+		return false, fmt.Errorf("publish graph generation: record content hash: %w", err)
+	}
+	if hash == "" {
+		return false, nil
+	}
+
+	// A NULL updated_at counts as modified: it cannot be ordered against
+	// published_at, and guessing "untouched" would skip a swap that was needed.
+	var noOp bool
+	if err := tx.QueryRow(ctx, `
+		SELECT EXISTS (
+		  SELECT 1 FROM repos r
+		  JOIN graph_index_generations g ON g.id = r.graph_published_generation_id
+		  WHERE r.id = $1 AND g.content_hash = $2 AND g.published_at IS NOT NULL
+		    AND NOT EXISTS (
+		      SELECT 1 FROM code_nodes n WHERE n.repo_id = $1
+		        AND (n.updated_at IS NULL OR n.updated_at > g.published_at))
+		    AND NOT EXISTS (
+		      SELECT 1 FROM code_edges e WHERE e.repo_id = $1
+		        AND (e.updated_at IS NULL OR e.updated_at > g.published_at))
+		)`, repoID, hash).Scan(&noOp); err != nil {
+		return false, fmt.Errorf("publish graph generation: compare published hash: %w", err)
+	}
+	return noOp, nil
+}
+
 func publishGraphGeneration(ctx context.Context, st *store.Store, repoID, generationID int64) error {
 	tx, err := st.Pool.Begin(ctx)
 	if err != nil {
@@ -550,6 +609,14 @@ func publishGraphGeneration(ctx context.Context, st *store.Store, repoID, genera
 		return commitGraphPublishPreflightFailure(ctx, tx, repoID, generationID, limitErr)
 	}
 
+	noOp, err := stageGenerationIsNoOp(ctx, tx, repoID, generationID)
+	if err != nil {
+		return err
+	}
+	if noOp {
+		return finishGraphGenerationPublish(ctx, st, tx, repoID, generationID)
+	}
+
 	// This transaction is the visibility boundary. Any later limit or decoding
 	// error rolls the deletes and inserts back before the generation is failed.
 	if _, err := tx.Exec(ctx, `DELETE FROM code_nodes WHERE repo_id = $1`, repoID); err != nil {
@@ -791,6 +858,18 @@ func publishGraphGeneration(ctx context.Context, st *store.Store, repoID, genera
 		return err
 	}
 
+	return finishGraphGenerationPublish(ctx, st, tx, repoID, generationID)
+}
+
+// finishGraphGenerationPublish commits the bookkeeping half of a publish: mark
+// the generation published, advance the repo's graph pointers, drop the staging
+// payloads, commit, then relink cross-repo API endpoints.
+//
+// Both publish paths end here. The swap path runs it after rewriting
+// code_nodes; the no-op path runs it instead of rewriting anything. The repo
+// pointers must advance either way — they record which commit the graph now
+// reflects and clear the refresh request that triggered this generation.
+func finishGraphGenerationPublish(ctx context.Context, st *store.Store, tx pgx.Tx, repoID, generationID int64) error {
 	if _, err := tx.Exec(ctx, `UPDATE graph_index_generations
 		SET status = 'published', published_at = NOW(), updated_at = NOW()
 		WHERE id = $1 AND repo_id = $2`, generationID, repoID); err != nil {

backend/internal/graph/fullindex_pg_test.go

@@ -1635,3 +1635,88 @@ func TestListReposDueForGraphIndexSelectsLegacyRecentTimestampWithoutPublishedAu
 		t.Fatalf("legacy repo with recent timestamp and null generation authority not immediately due: %+v", targets)
 	}
 }
+
+// generationNodeIDs returns the repo's published node IDs in a stable order.
+// Identity matters here: the publish swap deletes and re-inserts, so surviving
+// IDs are proof the swap did not run.
+func generationNodeIDs(t *testing.T, ctx context.Context, pool *pgxpool.Pool, repoID int64) []int64 {
+	t.Helper()
+	rows, err := pool.Query(ctx, `SELECT id FROM code_nodes WHERE repo_id = $1 ORDER BY id`, repoID)
+	if err != nil {
+		t.Fatalf("read node ids: %v", err)
+	}
+	defer rows.Close()
+	var ids []int64
+	for rows.Next() {
+		var id int64
+		if err := rows.Scan(&id); err != nil {
+			t.Fatalf("scan node id: %v", err)
+		}
+		ids = append(ids, id)
+	}
+	if err := rows.Err(); err != nil {
+		t.Fatalf("iterate node ids: %v", err)
+	}
+	return ids
+}
+
+// A commit that touches no indexed symbol still resolves to a new head and
+// still builds a generation. Publishing it must advance the repo's bookkeeping
+// without rewriting code_nodes, because the swap would delete and re-insert
+// every row to arrive at the graph already there. That churn is what filled
+// the pggraph CDC log and the database volume on 2026-09-03.
+func TestPublishSkipsSwapWhenGenerationMatchesLiveGraph(t *testing.T) {
+	pool, ctx := generationTestPool(t)
+	st := store.NewWithDB(pool)
+	installationID := generationSeedInstallation(t, ctx, pool, "{}")
+	repoID := generationSeedRepo(t, ctx, pool, installationID, "generation/noop-publish")
+
+	gh := &fakeFullIndexGitHub{
+		sha:      "1111111111111111111111111111111111111111",
+		tree:     ghpkg.RepoTree{Files: []ghpkg.RepoTreeFile{{Path: "a.go", SHA: "sha-a.go"}, {Path: "b.go", SHA: "sha-b.go"}}},
+		contents: map[string]string{"a.go": "package p\nfunc Alpha() { Beta() }\n", "b.go": "package p\nfunc Beta() {}\n"},
+		fetchErr: map[string]error{},
+	}
+	first, err := IndexRepoBounded(ctx, st, gh, 1, "o", "r", "main", repoID, 10, 0)
+	if err != nil {
+		t.Fatalf("first index: %v", err)
+	}
+	if !first.Published {
+		t.Fatalf("first index did not publish: %+v", first)
+	}
+	before := generationNodeIDs(t, ctx, pool, repoID)
+	if len(before) == 0 {
+		t.Fatal("first publish wrote no nodes")
+	}
+
+	// New head, byte-identical sources: the generation hashes to what is live.
+	const secondSHA = "2222222222222222222222222222222222222222"
+	gh.sha = secondSHA
+	second, err := IndexRepoBounded(ctx, st, gh, 1, "o", "r", "main", repoID, 10, 0)
+	if err != nil {
+		t.Fatalf("second index: %v", err)
+	}
+	if !second.Published {
+		t.Fatalf("second index did not publish: %+v", second)
+	}
+
+	if after := generationNodeIDs(t, ctx, pool, repoID); !slices.Equal(before, after) {
+		t.Fatalf("identical generation rewrote code_nodes: %v -> %v", before, after)
+	}
+
+	// Bookkeeping still has to advance, or the repo re-indexes this head forever.
+	var publishedSHA string
+	var status string
+	if err := pool.QueryRow(ctx, `
+		SELECT r.graph_index_commit_sha, g.status
+		FROM repos r JOIN graph_index_generations g ON g.id = r.graph_published_generation_id
+		WHERE r.id = $1`, repoID).Scan(&publishedSHA, &status); err != nil {
+		t.Fatalf("read repo graph pointers: %v", err)
+	}
+	if publishedSHA != secondSHA {
+		t.Fatalf("graph_index_commit_sha = %q, want %q", publishedSHA, secondSHA)
+	}
+	if status != "published" {
+		t.Fatalf("published generation status = %q, want published", status)
+	}
+}

backend/internal/memory/mirror_worker_pg_test.go

@@ -122,9 +122,27 @@ func TestMirrorWorkerPreservesPipelinePatternProvenanceAndMetadata(t *testing.T)
 	}
 }
 
-// Two relational pattern rows may intentionally converge on the same
-// deterministic memory custom ID. Removing either projection must not remove
-// the shared memory while the other projection remains authoritative.
+// Two co-existing pattern rows can still share one deterministic memory
+// identity, and deleting either must not tombstone the memory the other owns.
+//
+// This case survives migration 086 because that index is PARTIAL:
+//
+//	CREATE UNIQUE INDEX patterns_installation_memory_custom_uniq
+//	  ON patterns (installation_id, memory_custom_id)
+//	  WHERE memory_custom_id IS NOT NULL;
+//
+// Rows carrying only a legacy memory_doc_id leave memory_custom_id NULL, so the
+// index never applies. Both DeletePattern (firstNonEmpty(MemoryCustomID,
+// MemoryDocID)) and patternMirrorDeleteAuthorized (COALESCE over the same two
+// columns) resolve identity through memory_doc_id, so the duplicate-owner guard
+// is live for exactly these rows.
+//
+// The memory_custom_id spelling of this test is what 086 made unconstructible:
+// with that column set, a second create cannot produce a second row. How it
+// fails depends on CreatePattern's conflict handling -- here the raw insert
+// raises "duplicate key value violates unique constraint"; where CreatePattern
+// upserts, it silently returns the first row instead. Either way the old test
+// deleted the only owner and asserted a memory nothing owned would survive.
 func TestMirrorWorkerDeleteKeepsMemoryOwnedByDuplicatePattern(t *testing.T) {
 	pool, install := pgTestPool(t)
 	ctx := context.Background()
@@ -138,29 +156,33 @@ func TestMirrorWorkerDeleteKeepsMemoryOwnedByDuplicatePattern(t *testing.T) {
 	var repoID int64
 	if err := pool.QueryRow(ctx, `
 		INSERT INTO repos (installation_id, github_id, full_name)
-		VALUES ($1, (random() * 1000000000)::bigint, 'acme/mirror-owner-duplicate')
+		VALUES ($1, (random() * 1000000000)::bigint, 'acme/mirror-owner-legacy-dup')
 		RETURNING id`, install).Scan(&repoID); err != nil {
 		t.Fatalf("seed repo: %v", err)
 	}
 	source := "manual"
-	customID := PatternCustomID("", "mirror-owner-duplicate", source, "guard shared writes")
-	first, err := st.CreatePattern(ctx, install, &repoID, "guard shared writes", nil, nil, &source, nil, nil, stringPointer(customID), nil)
+	// Legacy identity: memory_doc_id set, memory_custom_id left NULL.
+	docID := fmt.Sprintf("sm_legacy_dup_%d", time.Now().UnixNano())
+	first, err := st.CreatePattern(ctx, install, &repoID, "guard shared writes", stringPointer(docID), nil, &source, nil, nil, nil, nil)
 	if err != nil {
-		t.Fatalf("create first pattern: %v", err)
+		t.Fatalf("create first legacy pattern: %v", err)
 	}
-	second, err := st.CreatePattern(ctx, install, &repoID, "guard shared writes", nil, nil, &source, nil, nil, stringPointer(customID), nil)
+	second, err := st.CreatePattern(ctx, install, &repoID, "guard shared writes", stringPointer(docID), nil, &source, nil, nil, nil, nil)
 	if err != nil {
-		t.Fatalf("create duplicate pattern: %v", err)
+		t.Fatalf("create duplicate legacy pattern: %v", err)
 	}
 	t.Cleanup(func() {
 		bg := context.Background()
 		_, _ = pool.Exec(bg, `DELETE FROM memory_mirror_outbox WHERE installation_id=$1 AND aggregate_type='pattern' AND aggregate_id=ANY($2::bigint[])`, install, []int64{first.ID, second.ID})
 		_, _ = pool.Exec(bg, `DELETE FROM patterns WHERE id=ANY($1::bigint[])`, []int64{first.ID, second.ID})
 		_, _ = pool.Exec(bg, `DELETE FROM repos WHERE id=$1`, repoID)
 	})
+	if first.ID == second.ID {
+		t.Fatalf("legacy duplicates collapsed to one row (id %d): the partial index now covers memory_doc_id, so this scenario is gone", first.ID)
+	}
 
-	if _, err := idx.IndexPattern(ctx, "mirror-owner-duplicate", PatternMemory{
-		Content: "guard shared writes", CustomID: customID, Source: source,
+	if _, err := idx.IndexPattern(ctx, "mirror-owner-legacy-dup", PatternMemory{
+		Content: "guard shared writes", CustomID: docID, Source: source,
 	}); err != nil {
 		t.Fatalf("seed memory: %v", err)
 	}
@@ -178,11 +200,84 @@ func TestMirrorWorkerDeleteKeepsMemoryOwnedByDuplicatePattern(t *testing.T) {
 			break
 		}
 	}
-	if row := readRow(t, pool, install, customID); row.deletedAt != nil {
+	if row := readRow(t, pool, install, docID); row.deletedAt != nil {
 		t.Fatal("deleting one duplicate pattern tombstoned memory still owned by the other pattern")
 	}
 }
 
+// The sequential counterpart: a live row owns the custom ID and a superseded
+// row's delete arrives afterwards. patternMirrorDeleteAuthorized must see the
+// live owner and skip the tombstone.
+func TestMirrorWorkerDeleteKeepsMemoryOwnedByLivePattern(t *testing.T) {
+	pool, install := pgTestPool(t)
+	ctx := context.Background()
+	lockMirrorOutboxPGTests(t, pool, ctx)
+	st := store.NewWithDB(pool)
+	if _, err := pool.Exec(ctx, `DELETE FROM memory_mirror_outbox`); err != nil {
+		t.Fatalf("clear mirror outbox: %v", err)
+	}
+	idx := NewPGIndexer(pool, nil, install, pgTestDims, slog.New(slog.DiscardHandler))
+
+	var repoID int64
+	if err := pool.QueryRow(ctx, `
+		INSERT INTO repos (installation_id, github_id, full_name)
+		VALUES ($1, (random() * 1000000000)::bigint, 'acme/mirror-owner-live')
+		RETURNING id`, install).Scan(&repoID); err != nil {
+		t.Fatalf("seed repo: %v", err)
+	}
+	source := "manual"
+	customID := PatternCustomID("", "mirror-owner-live", source, "guard shared writes")
+	live, err := st.CreatePattern(ctx, install, &repoID, "guard shared writes", nil, nil, &source, nil, nil, stringPointer(customID), nil)
+	if err != nil {
+		t.Fatalf("create live pattern: %v", err)
+	}
+	t.Cleanup(func() {
+		bg := context.Background()
+		_, _ = pool.Exec(bg, `DELETE FROM memory_mirror_outbox WHERE installation_id=$1 AND aggregate_type='pattern'`, install)
+		_, _ = pool.Exec(bg, `DELETE FROM patterns WHERE id=$1`, live.ID)
+		_, _ = pool.Exec(bg, `DELETE FROM repos WHERE id=$1`, repoID)
+	})
+
+	if _, err := idx.IndexPattern(ctx, "mirror-owner-live", PatternMemory{
+		Content: "guard shared writes", CustomID: customID, Source: source,
+	}); err != nil {
+		t.Fatalf("seed memory: %v", err)
+	}
+
+	// A superseded row's delete, arriving after the live row already owns the
+	// custom ID. It has to be enqueued directly: the delete must be the LAST
+	// event drained, and DeletePattern-then-CreatePattern would order a reviving
+	// upsert behind it, which masks a tombstone rather than preventing one.
+	stalePayload, err := NewDeleteMirrorPayload(customID)
+	if err != nil {
+		t.Fatalf("build delete payload: %v", err)
+	}
+	staleAggregateID := live.ID + 1_000_000
+	if err := st.WithMemoryMirrorTx(ctx, func(pgx.Tx) (store.MemoryMirrorEvent, error) {
+		return store.MemoryMirrorEvent{
+			InstallationID: install, AggregateType: store.MemoryMirrorPattern,
+			AggregateID: staleAggregateID, Operation: store.MemoryMirrorDelete,
+			Payload: stalePayload,
+		}, nil
+	}); err != nil {
+		t.Fatalf("enqueue stale delete: %v", err)
+	}
+
+	worker := NewMirrorWorker(st, func(context.Context, int64) MirrorIndexer { return idx }, slog.New(slog.DiscardHandler))
+	for {
+		processed, err := worker.RunOnce(ctx, 50)
+		if err != nil {
+			t.Fatalf("drain mirror events: %v", err)
+		}
+		if processed == 0 {
+			break
+		}
+	}
+	if row := readRow(t, pool, install, customID); row.deletedAt != nil {
+		t.Fatal("a superseded pattern's delete tombstoned memory the live pattern still owns")
+	}
+}
+
 func stringPointer(value string) *string { return &value }
 
 // An old aggregate's delete and a recreated aggregate's upsert share memory