feat(db): add goose 00038 dataset lifecycle tables

Registry、checkpoint queue 与 destroy 请求表属主 dataset；必须可 Down，不改历史 Goose。 Co-authored-by: Cursor <cursoragent@cursor.com>
2026-09-15 02:31:34 +00:00
parent e81e7a1bcc
commit 5ac14d649b
3 changed files with 137 additions and 2 deletions
@@ -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:
@@ -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;
@@ -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)