diff --git a/pkg/appview/db/go_migrations.go b/pkg/appview/db/go_migrations.go new file mode 100644 index 0000000..8e0e233 --- /dev/null +++ b/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 +} diff --git a/pkg/appview/db/migration_replay_test.go b/pkg/appview/db/migration_replay_test.go new file mode 100644 index 0000000..09a770f --- /dev/null +++ b/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 +} diff --git a/pkg/appview/db/migrations/0033_add_manifest_key.yaml b/pkg/appview/db/migrations/0033_add_manifest_key.yaml index 2a12c07..0ea9abb 100644 --- a/pkg/appview/db/migrations/0033_add_manifest_key.yaml +++ b/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); diff --git a/pkg/appview/db/schema.go b/pkg/appview/db/schema.go index 417de94..affc908 100644 --- a/pkg/appview/db/schema.go +++ b/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)