db: fill manifest_key inside its migration, not at runtime

0033 added the column and left the fill to the Jetstream backfill. That is fine
for a running system and wrong for replay: a database several releases behind
runs every pending migration back-to-back at boot, long before any worker
starts. Anything built on top of manifest_key would, on that path, silently
operate on NULLs while working perfectly on a system that had been up for a
while. Establishing a migration's data precondition out-of-band means replay
cannot see it.

The runner now supports a Go step per migration version, running inside the same
transaction as that migration's SQL, after the DDL it depends on and before the
version is recorded. A version is never recorded without its Go half.

0033's step fills manifest_key for every row lacking one. It has to be Go: the
value is a truncated sha256 and SQLite has no hash builtin. It pages through the
table and writes one UPDATE ... CASE per 500 rows, because a statement per row
would be correct and unusably slow against a remote primary.

Verified by unregistering the hook: replay then leaves 3 of 3 seeded manifests
with NULL keys. With it, all three are filled, from a snapshot of the pre-0009
schema forward.

The upserts keep their "OR manifests.manifest_key IS NULL" clause. It is a
self-healing net for rows that somehow arrive without a key rather than the
mechanism anything depends on.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

pkg/appview/db/go_migrations.go

@@ -0,0 +1,129 @@
+package db
+
+import (
+	"database/sql"
+	"fmt"
+	"log/slog"
+	"strings"
+)
+
+// goMigrations holds migration steps that cannot be expressed in SQL, keyed by
+// the migration version they belong to.
+//
+// A hook runs inside the same transaction as its migration's SQL, immediately
+// after it and before the version is recorded, so a version is either fully
+// applied or not applied at all.
+//
+// This exists because a migration's data precondition has to be established
+// inside the migration sequence. Anything satisfied out-of-band (at runtime, by
+// a background worker, by an operator) is invisible to replay: a database
+// upgrading across several releases at once runs every pending migration
+// back-to-back at boot, long before any worker starts, so a later migration that
+// assumed the out-of-band step had happened would silently operate on unprepared
+// data.
+//
+// Keep hooks idempotent. A migration that fails partway is rolled back and
+// retried on the next boot, and a hook may also run against a database where
+// some or all of its work has already happened by other means.
+var goMigrations = map[int]func(*sql.Tx) error{
+	33: backfillManifestKeys,
+}
+
+// manifestKeyBackfillBatch is how many manifests get their key set per
+// statement. Each row contributes three placeholders, so this stays far below
+// the driver's parameter ceiling while keeping the number of round trips to a
+// remote primary low: one per 500 rows rather than one per row.
+const manifestKeyBackfillBatch = 500
+
+// backfillManifestKeys fills manifests.manifest_key for every row that lacks it.
+//
+// The value is a truncated sha256 of (did, repository, digest). SQLite has no
+// hash builtin and go-libsql cannot register one, which is why this is a Go hook
+// rather than an UPDATE in 0033's SQL.
+//
+// On a live database this is usually a no-op or close to it, because the
+// Jetstream backfill fills the column as it re-upserts each manifest. It matters
+// on the path that has no backfill to rely on: a database replaying migrations
+// across releases, where 0033 and anything built on top of it run seconds apart
+// with no worker having started. Without this, a later migration that moves
+// layers and manifest_references onto manifest_key would copy NULLs.
+func backfillManifestKeys(tx *sql.Tx) error {
+	type pending struct {
+		id  int64
+		key string
+	}
+
+	var filled int
+	for {
+		// Read a page, then close the cursor before writing. Updating a table
+		// while a SELECT over it is still open is asking for trouble.
+		rows, err := tx.Query(`
+			SELECT id, did, repository, digest
+			FROM manifests
+			WHERE manifest_key IS NULL
+			LIMIT ?
+		`, manifestKeyBackfillBatch)
+		if err != nil {
+			return fmt.Errorf("select manifests missing a key: %w", err)
+		}
+
+		batch := make([]pending, 0, manifestKeyBackfillBatch)
+		for rows.Next() {
+			var (
+				id                      int64
+				did, repository, digest string
+			)
+			if err := rows.Scan(&id, &did, &repository, &digest); err != nil {
+				rows.Close()
+				return fmt.Errorf("scan manifest: %w", err)
+			}
+			batch = append(batch, pending{id: id, key: ManifestKey(did, repository, digest)})
+		}
+		if err := rows.Err(); err != nil {
+			rows.Close()
+			return fmt.Errorf("iterate manifests: %w", err)
+		}
+		rows.Close()
+
+		if len(batch) == 0 {
+			break
+		}
+
+		// One statement per page: UPDATE ... SET manifest_key = CASE id WHEN ...
+		// END WHERE id IN (...). A statement per row would be correct too, and
+		// unusably slow against a remote primary.
+		var sb strings.Builder
+		args := make([]any, 0, len(batch)*3)
+		sb.WriteString("UPDATE manifests SET manifest_key = CASE id")
+		for _, p := range batch {
+			sb.WriteString(" WHEN ? THEN ?")
+			args = append(args, p.id, p.key)
+		}
+		sb.WriteString(" END WHERE id IN (")
+		for i, p := range batch {
+			if i > 0 {
+				sb.WriteByte(',')
+			}
+			sb.WriteByte('?')
+			args = append(args, p.id)
+		}
+		sb.WriteByte(')')
+
+		if _, err := tx.Exec(sb.String(), args...); err != nil {
+			return fmt.Errorf("set manifest keys: %w", err)
+		}
+		filled += len(batch)
+
+		// A short page means the table is exhausted. Checking this rather than
+		// looping again saves a query, and the loop still terminates either way
+		// because every row written stops matching the WHERE.
+		if len(batch) < manifestKeyBackfillBatch {
+			break
+		}
+	}
+
+	if filled > 0 {
+		slog.Info("Backfilled manifest_key during migration", "manifests", filled)
+	}
+	return nil
+}

pkg/appview/db/migration_replay_test.go

@@ -0,0 +1,205 @@
+package db
+
+import (
+	"database/sql"
+	"os"
+	"testing"
+	"time"
+)
+
+// TestMigrationReplayPopulatesManifestKeys covers the upgrade path that has no
+// running application behind it.
+//
+// A database several releases behind runs every pending migration back-to-back
+// at boot, before any worker starts. So a migration cannot depend on a data
+// precondition that something else establishes at runtime: from replay's point
+// of view that step never happened. manifests.manifest_key is filled by the
+// Jetstream backfill on a live system, which is exactly the kind of out-of-band
+// step that is invisible here, so 0033 carries a Go hook that fills it in
+// sequence instead.
+//
+// Without the hook, anything built on top of manifest_key (moving layers and
+// manifest_references onto it) would silently copy NULLs on this path while
+// working fine on a system that had been running.
+func TestMigrationReplayPopulatesManifestKeys(t *testing.T) {
+	database := baseSchemaDB(t)
+	defer database.Close()
+
+	// Seed rows in the pre-0009 shape, as a database at that vintage would have.
+	if _, err := database.Exec(
+		`INSERT INTO users (did, handle, pds_endpoint, last_seen) VALUES (?, ?, ?, ?)`,
+		"did:plc:replay", "replay.example.com", "https://pds.example.com", time.Now(),
+	); err != nil {
+		t.Fatalf("seed user: %v", err)
+	}
+
+	seeded := []struct{ repo, digest string }{
+		{"myapp", "sha256:aaa"},
+		{"myapp", "sha256:bbb"},
+		{"otherapp", "sha256:ccc"},
+	}
+	for _, s := range seeded {
+		if _, err := database.Exec(`
+			INSERT INTO manifests
+			(did, repository, digest, hold_endpoint, schema_version, media_type, artifact_type, created_at)
+			VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+		`, "did:plc:replay", s.repo, s.digest, "did:web:hold.example.com", 2,
+			"application/vnd.oci.image.manifest.v1+json", "container-image", time.Now()); err != nil {
+			t.Fatalf("seed manifest %s/%s: %v", s.repo, s.digest, err)
+		}
+	}
+
+	// Replay everything from that vintage forward, with no application running.
+	if err := runMigrations(database, false); err != nil {
+		t.Fatalf("replay migrations: %v", err)
+	}
+
+	remaining, total, err := ManifestKeyBackfillProgress(database)
+	if err != nil {
+		t.Fatalf("ManifestKeyBackfillProgress: %v", err)
+	}
+	if total != int64(len(seeded)) {
+		t.Fatalf("manifest count = %d after replay, want %d", total, len(seeded))
+	}
+	if remaining != 0 {
+		t.Errorf("%d of %d manifests have no key after replay; a later migration "+
+			"depending on manifest_key would copy NULLs", remaining, total)
+	}
+
+	for _, s := range seeded {
+		var key string
+		if err := database.QueryRow(
+			`SELECT manifest_key FROM manifests WHERE did = ? AND repository = ? AND digest = ?`,
+			"did:plc:replay", s.repo, s.digest,
+		).Scan(&key); err != nil {
+			t.Fatalf("read key for %s/%s: %v", s.repo, s.digest, err)
+		}
+		if want := ManifestKey("did:plc:replay", s.repo, s.digest); key != want {
+			t.Errorf("%s/%s key = %q, want %q", s.repo, s.digest, key, want)
+		}
+	}
+}
+
+// TestManifestKeyHookIsIdempotent: a migration that fails partway is rolled back
+// and retried on the next boot, and the hook also runs against databases where
+// the runtime backfill already did some of the work. Neither may break it.
+func TestManifestKeyHookIsIdempotent(t *testing.T) {
+	database := revTestDB(t)
+	did := manifestKeyTestUser(t, database)
+
+	for _, digest := range []string{"sha256:aaa", "sha256:bbb"} {
+		if _, err := InsertManifest(database, sampleManifest(did, "myapp", digest)); err != nil {
+			t.Fatalf("InsertManifest: %v", err)
+		}
+	}
+	// One row already filled, one not: the mixed state a partly-backfilled
+	// database is actually in.
+	if _, err := database.Exec(
+		`UPDATE manifests SET manifest_key = NULL WHERE digest = ?`, "sha256:aaa"); err != nil {
+		t.Fatalf("clear one key: %v", err)
+	}
+
+	for pass := range 2 {
+		tx, err := database.Begin()
+		if err != nil {
+			t.Fatalf("begin: %v", err)
+		}
+		if err := backfillManifestKeys(tx); err != nil {
+			tx.Rollback()
+			t.Fatalf("pass %d: %v", pass, err)
+		}
+		if err := tx.Commit(); err != nil {
+			t.Fatalf("commit pass %d: %v", pass, err)
+		}
+	}
+
+	remaining, _, err := ManifestKeyBackfillProgress(database)
+	if err != nil {
+		t.Fatalf("ManifestKeyBackfillProgress: %v", err)
+	}
+	if remaining != 0 {
+		t.Errorf("%d manifests still unfilled after two passes", remaining)
+	}
+}
+
+// TestManifestKeyHookSpansPages: the hook pages through the table, so the
+// interesting case is more rows than one page.
+func TestManifestKeyHookSpansPages(t *testing.T) {
+	database := revTestDB(t)
+	did := manifestKeyTestUser(t, database)
+
+	const total = manifestKeyBackfillBatch + 21
+	for i := range total {
+		m := sampleManifest(did, "myapp", digestFor(i))
+		if _, err := InsertManifest(database, m); err != nil {
+			t.Fatalf("InsertManifest %d: %v", i, err)
+		}
+	}
+	if _, err := database.Exec(`UPDATE manifests SET manifest_key = NULL`); err != nil {
+		t.Fatalf("clear keys: %v", err)
+	}
+
+	tx, err := database.Begin()
+	if err != nil {
+		t.Fatalf("begin: %v", err)
+	}
+	if err := backfillManifestKeys(tx); err != nil {
+		tx.Rollback()
+		t.Fatalf("backfillManifestKeys: %v", err)
+	}
+	if err := tx.Commit(); err != nil {
+		t.Fatalf("commit: %v", err)
+	}
+
+	remaining, count, err := ManifestKeyBackfillProgress(database)
+	if err != nil {
+		t.Fatalf("ManifestKeyBackfillProgress: %v", err)
+	}
+	if count != total {
+		t.Fatalf("manifest count = %d, want %d", count, total)
+	}
+	if remaining != 0 {
+		t.Errorf("%d of %d unfilled; the hook stopped before the last page", remaining, total)
+	}
+}
+
+func digestFor(i int) string {
+	const hexdigits = "0123456789abcdef"
+	out := []byte("sha256:0000000000000000")
+	for pos, n := len(out)-1, i; pos >= 7 && n > 0; pos, n = pos-1, n/16 {
+		out[pos] = hexdigits[n%16]
+	}
+	return string(out)
+}
+
+// baseSchemaDB returns a database at the pre-0009 vintage, with no migrations
+// applied beyond the ones that snapshot records.
+func baseSchemaDB(t *testing.T) *sql.DB {
+	t.Helper()
+
+	base, err := os.ReadFile("testdata/base_schema.sql")
+	if err != nil {
+		t.Fatalf("read testdata/base_schema.sql: %v", err)
+	}
+	database := revTestDBRaw(t)
+	for i, stmt := range splitSQLStatements(string(base)) {
+		if _, err := database.Exec(stmt); err != nil {
+			t.Fatalf("apply base_schema.sql statement %d: %v", i+1, err)
+		}
+	}
+	return database
+}
+
+// revTestDBRaw opens an empty file-backed database with no schema applied, so
+// the caller can build one from a snapshot.
+func revTestDBRaw(t *testing.T) *sql.DB {
+	t.Helper()
+	connector, err := openLibsqlLocalConnector("file:" + t.TempDir() + "/replay.db")
+	if err != nil {
+		t.Fatalf("open connector: %v", err)
+	}
+	database := sql.OpenDB(connector)
+	database.SetMaxOpenConns(1)
+	t.Cleanup(func() { database.Close() })
+	return database
+}

pkg/appview/db/migrations/0033_add_manifest_key.yaml

@@ -8,15 +8,23 @@ description: |
   data that is already complete and already proven unique, rather than doing the
   fill and three table rebuilds in one step.
 
-  Nullable, and there is no UPDATE here to fill it, because the value cannot be
-  computed in SQL: it is a truncated sha256 and SQLite has no hash builtin, nor
-  can go-libsql register one. The fill happens through the machinery that already
-  exists. The Jetstream backfill re-upserts every manifest across the protocol on
-  startup, and both upsert paths now write manifest_key and carry an extra
-  "OR manifests.manifest_key IS NULL" clause on their WHERE guard. Without that
-  clause the upserts would skip unchanged manifests and the column would never
-  fill; with it, the existing backfill populates the table as a side effect of a
-  run it was making anyway.
+  The value cannot be computed in SQL: it is a truncated sha256, SQLite has no
+  hash builtin, and go-libsql cannot register one. So this migration carries a Go
+  step (backfillManifestKeys, see go_migrations.go) that fills the column in the
+  same transaction, right after the ALTER TABLE that creates it.
+
+  It is filled here, in sequence, rather than left to the runtime, because a
+  database several releases behind runs every pending migration back-to-back at
+  boot with no worker running. Any later migration that depends on manifest_key
+  being populated would, on that path, silently operate on NULLs. Establishing
+  the precondition inside the sequence is the only way replay can see it.
+
+  The Jetstream backfill also fills the column as it re-upserts each manifest,
+  since both upsert paths write manifest_key and carry an extra
+  "OR manifests.manifest_key IS NULL" clause on their WHERE guard (without it
+  they would skip unchanged manifests). That is now a self-healing safety net for
+  rows that somehow arrive without a key, not the mechanism this migration
+  depends on.
 
   The index is UNIQUE even though the column is nullable. SQLite treats NULLs as
   distinct, so rows not yet filled do not collide with each other, while every
@@ -25,9 +33,8 @@ description: |
   key, the insert fails loudly here rather than silently attaching one manifest's
   layers to another after the swap.
 
-  AppView logs the count of unfilled rows at startup. When it reaches zero the
-  follow-up migration can make manifest_key NOT NULL, move layers and
-  manifest_references onto it, and drop id.
+  AppView logs the count of unfilled rows at startup, which should be zero
+  immediately after this migration.
 query: |
   ALTER TABLE manifests ADD COLUMN manifest_key TEXT;
   CREATE UNIQUE INDEX IF NOT EXISTS idx_manifests_manifest_key ON manifests(manifest_key);

pkg/appview/db/schema.go

@@ -256,6 +256,16 @@ func runMigrations(db *sql.DB, freshDB bool) error {
 			}
 		}
 
+		// Some migrations need a step SQL cannot express. It runs in this same
+		// transaction, after the DDL it depends on and before the version is
+		// recorded, so the version is never recorded without its Go half.
+		if hook, ok := goMigrations[m.Version]; ok {
+			slog.Info("Running Go migration step", "version", m.Version, "name", m.Name)
+			if err := hook(tx); err != nil {
+				return fmt.Errorf("failed to apply migration %d (%s) Go step: %w", m.Version, m.Name, err)
+			}
+		}
+
 		// Record migration
 		if _, err := tx.Exec("INSERT INTO schema_migrations (version) VALUES (?)", m.Version); err != nil {
 			return fmt.Errorf("failed to record migration %d: %w", m.Version, err)
@@ -303,9 +313,12 @@ func loadMigrations() ([]Migration, error) {
 		m.Version = version
 		m.Name = name
 
-		// Validate migration
+		// Validate migration. A migration may be pure SQL, SQL plus a Go step,
+		// or Go only; what it may not be is empty.
 		if m.Query == "" {
-			return nil, fmt.Errorf("missing migration 'query' in %s", file)
+			if _, hasHook := goMigrations[version]; !hasHook {
+				return nil, fmt.Errorf("missing migration 'query' in %s", file)
+			}
 		}
 
 		migrations = append(migrations, m)