diff --git a/.doc/system-design/module-boundaries.yaml b/.doc/system-design/module-boundaries.yaml index 92b791ec..2793bef0 100644 --- a/.doc/system-design/module-boundaries.yaml +++ b/.doc/system-design/module-boundaries.yaml @@ -162,6 +162,9 @@ imports: artifactvault: - app/uow - fsutil + dataset: + - artifactvault + - app/uow table_owners: document_records: @@ -202,6 +205,16 @@ table_owners: - artifactvault artifact_backfill_exclusions: - artifactvault + dataset_registry: + - dataset + dataset_backfill_jobs: + - dataset + dataset_backfill_checkpoints: + - dataset + dataset_backfill_queue: + - dataset + dataset_destroy_requests: + - dataset path_owners: output.md: diff --git a/deploy/postgres/goose/00038_2026-09-15-dataset-lifecycle.sql b/deploy/postgres/goose/00038_2026-09-15-dataset-lifecycle.sql new file mode 100644 index 00000000..00f025d5 --- /dev/null +++ b/deploy/postgres/goose/00038_2026-09-15-dataset-lifecycle.sql @@ -0,0 +1,91 @@ +-- +goose Up +-- U9 dataset registry, checkpointed backfill queue, and destroy gates. +-- Do not modify historical Goose files. + +CREATE TABLE IF NOT EXISTS dataset_registry ( + dataset_id TEXT PRIMARY KEY, + class TEXT NOT NULL, + display_name TEXT NOT NULL, + storage_identity TEXT NOT NULL, + source_schema INT NOT NULL DEFAULT 0, + source_commit TEXT NOT NULL DEFAULT '', + status TEXT NOT NULL DEFAULT 'approved', + project_count INT NOT NULL DEFAULT 0, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + CONSTRAINT chk_dataset_class CHECK (class IN ( + 'pinned', 'demo', 'regression', 'migration_baseline', 'ephemeral', 'quarantine' + )), + CONSTRAINT chk_dataset_status CHECK (status IN ( + 'approved', 'imported', 'quarantine', 'retired' + )) +); + +CREATE TABLE IF NOT EXISTS dataset_backfill_jobs ( + job_id TEXT PRIMARY KEY, + status TEXT NOT NULL, + source_schema INT NOT NULL DEFAULT 0, + source_commit TEXT NOT NULL DEFAULT '', + storage_identity TEXT NOT NULL DEFAULT '', + paused_at TIMESTAMPTZ, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + CONSTRAINT chk_backfill_job_status CHECK (status IN ( + 'running', 'paused', 'completed', 'failed' + )) +); + +CREATE TABLE IF NOT EXISTS dataset_backfill_checkpoints ( + job_id TEXT NOT NULL REFERENCES dataset_backfill_jobs(job_id) ON DELETE CASCADE, + shard INT NOT NULL, + source_kind TEXT NOT NULL, + last_pk BIGINT NOT NULL DEFAULT 0, + scanned INT NOT NULL DEFAULT 0, + succeeded INT NOT NULL DEFAULT 0, + quarantined INT NOT NULL DEFAULT 0, + retry_watermark INT NOT NULL DEFAULT 0, + PRIMARY KEY (job_id, shard, source_kind), + CONSTRAINT chk_backfill_source_kind CHECK (source_kind IN ( + 'snapshot', 'agent_commit', 'approved', 'chat_turn' + )) +); + +CREATE TABLE IF NOT EXISTS dataset_backfill_queue ( + job_id TEXT NOT NULL REFERENCES dataset_backfill_jobs(job_id) ON DELETE CASCADE, + source_kind TEXT NOT NULL, + source_pk BIGINT NOT NULL, + shard INT NOT NULL DEFAULT 0, + status TEXT NOT NULL DEFAULT 'pending', + payload_sha256 CHAR(64) NOT NULL DEFAULT '', + logical_id TEXT NOT NULL DEFAULT '', + claimed_at TIMESTAMPTZ, + PRIMARY KEY (job_id, source_kind, source_pk), + CONSTRAINT chk_backfill_queue_status CHECK (status IN ( + 'pending', 'claimed', 'done', 'quarantine' + )) +); + +CREATE INDEX IF NOT EXISTS idx_dataset_backfill_queue_claim + ON dataset_backfill_queue (job_id, status, source_pk); + +CREATE TABLE IF NOT EXISTS dataset_destroy_requests ( + request_id TEXT PRIMARY KEY, + target_identity TEXT NOT NULL, + delete_manifest_sha CHAR(64) NOT NULL, + recoverable_location TEXT NOT NULL, + quarantine_until TIMESTAMPTZ NOT NULL, + approval_1_by TEXT NOT NULL DEFAULT '', + approval_2_by TEXT NOT NULL DEFAULT '', + coverage_ok BOOLEAN NOT NULL DEFAULT FALSE, + backup_ok BOOLEAN NOT NULL DEFAULT FALSE, + corpus_ok BOOLEAN NOT NULL DEFAULT FALSE, + status TEXT NOT NULL DEFAULT 'blocked', + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + CONSTRAINT chk_destroy_status CHECK (status IN ('blocked', 'approved', 'applied')) +); + +-- +goose Down +DROP TABLE IF EXISTS dataset_destroy_requests; +DROP TABLE IF EXISTS dataset_backfill_queue; +DROP TABLE IF EXISTS dataset_backfill_checkpoints; +DROP TABLE IF EXISTS dataset_backfill_jobs; +DROP TABLE IF EXISTS dataset_registry; diff --git a/internal/db/migrate_test.go b/internal/db/migrate_test.go index dfd3f360..e0922c81 100644 --- a/internal/db/migrate_test.go +++ b/internal/db/migrate_test.go @@ -81,6 +81,36 @@ func TestMigration032_AllowsMigrationRepairKinds(t *testing.T) { } } +func TestGooseU9_00038_HasUpAndDown(t *testing.T) { + files, err := ListMigrationFiles(ResolveMigrationsDir()) + if err != nil { + t.Fatal(err) + } + var found string + for _, name := range files { + if strings.HasPrefix(name, "00038_") && strings.HasSuffix(name, ".sql") { + found = name + break + } + } + if found == "" { + t.Fatal("expected goose 00038 for dataset lifecycle") + } + raw, err := os.ReadFile(filepath.Join(ResolveMigrationsDir(), found)) + if err != nil { + t.Fatal(err) + } + text := string(raw) + if !strings.Contains(text, "-- +goose Up") || !strings.Contains(text, "-- +goose Down") { + t.Fatal("00038 must have Up and Down") + } + for _, table := range []string{"dataset_registry", "dataset_backfill_jobs", "dataset_backfill_checkpoints", "dataset_backfill_queue", "dataset_destroy_requests"} { + if !strings.Contains(text, table) { + t.Fatalf("00038 must mention %s", table) + } + } +} + func TestGooseM2_00034to00036_DownThenUp(t *testing.T) { if testing.Short() { t.Skip("goose down is not part of -short") @@ -101,6 +131,7 @@ func TestGooseM2_00034to00036_DownThenUp(t *testing.T) { "00035_2026-09-14-durable-event-delivery.sql", "00036_2026-09-14-final-artifact-vault.sql", "00037_2026-09-15-task-run-manifests.sql", + "00038_2026-09-15-dataset-lifecycle.sql", } joined := strings.Join(files, "\n") for _, name := range want { @@ -136,7 +167,7 @@ func TestGooseM2_00034to00036_DownThenUp(t *testing.T) { t.Fatalf("up: %v", err) } ver, err := CurrentVersion(ctx, pool) - if err != nil || ver < 37 { + if err != nil || ver < 38 { t.Fatalf("version after up: %d %v", ver, err) } @@ -149,7 +180,7 @@ func TestGooseM2_00034to00036_DownThenUp(t *testing.T) { if err := goose.DownToContext(ctx, sqlDB, migDir, 33); err != nil { t.Fatalf("down to 33: %v", err) } - for _, table := range []string{"event_envelopes", "event_deliveries", "event_dead_letters", "final_artifacts", "task_run_manifests"} { + for _, table := range []string{"event_envelopes", "event_deliveries", "event_dead_letters", "final_artifacts", "task_run_manifests", "dataset_registry"} { var n int if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM information_schema.tables WHERE table_name=$1`, table).Scan(&n); err != nil { t.Fatal(err)