From 5a70aaea5f7462069c824ed9f39bd97ab85ffc4e Mon Sep 17 00:00:00 2001 From: DJ Majumdar Date: Sun, 22 Mar 2026 11:48:15 -0700 Subject: [PATCH] refactor!: consolidate 9 chronological migrations into 4 object-oriented files MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Collapse all ALTER TABLE additions (label, net IO, groups, TTL, scheduling, dependencies, retries, memo) into the CREATE TABLE definitions, eliminating every ALTER statement. New layout: 001_tasks.sql – tasks table + 6 indexes 002_task_history.sql – task_history table + 5 indexes 003_task_deps.sql – task_deps table + 1 index 004_task_tags.sql – task_tags & task_history_tags + 2 indexes BREAKING CHANGE: existing databases must be recreated (pre-1.0). --- migrations/001_tasks.sql | 115 ++++++++---------- migrations/002_add_label.sql | 5 - migrations/002_task_history.sql | 55 +++++++++ migrations/003_net_io_and_groups.sql | 16 --- migrations/003_task_deps.sql | 11 ++ .../{007_tags.sql => 004_task_tags.sql} | 11 +- migrations/004_ttl.sql | 13 -- migrations/005_scheduling.sql | 19 --- migrations/006_dependencies.sql | 22 ---- migrations/008_retry_backoff.sql | 5 - migrations/009_memo.sql | 3 - src/store/mod.rs | 66 ++-------- 12 files changed, 133 insertions(+), 208 deletions(-) delete mode 100644 migrations/002_add_label.sql create mode 100644 migrations/002_task_history.sql delete mode 100644 migrations/003_net_io_and_groups.sql create mode 100644 migrations/003_task_deps.sql rename migrations/{007_tags.sql => 004_task_tags.sql} (59%) delete mode 100644 migrations/004_ttl.sql delete mode 100644 migrations/005_scheduling.sql delete mode 100644 migrations/006_dependencies.sql delete mode 100644 migrations/008_retry_backoff.sql delete mode 100644 migrations/009_memo.sql diff --git a/migrations/001_tasks.sql b/migrations/001_tasks.sql index a1c50dc..c6d66f2 100644 --- a/migrations/001_tasks.sql +++ b/migrations/001_tasks.sql @@ -1,78 +1,69 @@ --- Active queue: pending, running, and paused tasks. +-- Active queue: pending, running, paused, and blocked tasks. -- The UNIQUE(key) constraint provides key-based deduplication — -- submitting a task with an existing key is a no-op (INSERT OR IGNORE). -- When a duplicate is submitted while the existing task is running or paused, -- the requeue flag is set so the task re-runs after the current execution. CREATE TABLE IF NOT EXISTS tasks ( - id INTEGER PRIMARY KEY, - task_type TEXT NOT NULL, - key TEXT NOT NULL, - priority INTEGER NOT NULL, - status TEXT NOT NULL DEFAULT 'pending', - payload BLOB, - expected_read_bytes INTEGER NOT NULL DEFAULT 0, - expected_write_bytes INTEGER NOT NULL DEFAULT 0, - retry_count INTEGER NOT NULL DEFAULT 0, - last_error TEXT, - created_at INTEGER NOT NULL, - started_at INTEGER, - requeue INTEGER NOT NULL DEFAULT 0, - requeue_priority INTEGER, - parent_id INTEGER, - fail_fast INTEGER NOT NULL DEFAULT 1, + id INTEGER PRIMARY KEY, + task_type TEXT NOT NULL, + key TEXT NOT NULL, + label TEXT NOT NULL DEFAULT '', + priority INTEGER NOT NULL, + status TEXT NOT NULL DEFAULT 'pending', + payload BLOB, + memo BLOB DEFAULT NULL, + expected_read_bytes INTEGER NOT NULL DEFAULT 0, + expected_write_bytes INTEGER NOT NULL DEFAULT 0, + expected_net_rx_bytes INTEGER NOT NULL DEFAULT 0, + expected_net_tx_bytes INTEGER NOT NULL DEFAULT 0, + retry_count INTEGER NOT NULL DEFAULT 0, + max_retries INTEGER, + last_error TEXT, + created_at INTEGER NOT NULL, + started_at INTEGER, + run_after INTEGER, + requeue INTEGER NOT NULL DEFAULT 0, + requeue_priority INTEGER, + parent_id INTEGER, + fail_fast INTEGER NOT NULL DEFAULT 1, + group_key TEXT, + on_dep_failure TEXT NOT NULL DEFAULT 'cancel', + ttl_seconds INTEGER, + ttl_from TEXT NOT NULL DEFAULT 'submission', + expires_at INTEGER, + recurring_interval_secs INTEGER, + recurring_max_executions INTEGER, + recurring_execution_count INTEGER NOT NULL DEFAULT 0, + recurring_paused INTEGER NOT NULL DEFAULT 0, UNIQUE(key) ); --- Index for the scheduler hot path: pop highest-priority pending task. +-- Scheduler hot path: pop highest-priority pending task. CREATE INDEX IF NOT EXISTS idx_tasks_pending ON tasks (status, priority ASC, id ASC) WHERE status = 'pending'; --- Completed and failed task history for queries and IO learning. -CREATE TABLE IF NOT EXISTS task_history ( - id INTEGER PRIMARY KEY, - task_type TEXT NOT NULL, - key TEXT NOT NULL, - priority INTEGER NOT NULL, - status TEXT NOT NULL, - payload BLOB, - expected_read_bytes INTEGER NOT NULL DEFAULT 0, - expected_write_bytes INTEGER NOT NULL DEFAULT 0, - actual_read_bytes INTEGER, - actual_write_bytes INTEGER, - retry_count INTEGER NOT NULL DEFAULT 0, - last_error TEXT, - created_at INTEGER NOT NULL, - started_at INTEGER, - completed_at INTEGER NOT NULL, - duration_ms INTEGER, - parent_id INTEGER, - fail_fast INTEGER NOT NULL DEFAULT 1 -); - --- Index for IO learning: recent completions by task type. -CREATE INDEX IF NOT EXISTS idx_history_type_completed - ON task_history (task_type, completed_at DESC) - WHERE status = 'completed'; - --- Index for task lookup by key (used by task dedup and status checks). -CREATE INDEX IF NOT EXISTS idx_history_key - ON task_history (key, completed_at DESC); - --- Index for paginating and pruning history by completion time. -CREATE INDEX IF NOT EXISTS idx_history_completed - ON task_history (completed_at DESC); - --- Index for filtering history by status (e.g. listing failed tasks). -CREATE INDEX IF NOT EXISTS idx_history_status - ON task_history (status, completed_at DESC); - --- Index for looking up children of a parent task (active queue). +-- Lookup children of a parent task. CREATE INDEX IF NOT EXISTS idx_tasks_parent ON tasks (parent_id) WHERE parent_id IS NOT NULL; --- Index for looking up children of a parent task (history). -CREATE INDEX IF NOT EXISTS idx_task_history_parent - ON task_history (parent_id) - WHERE parent_id IS NOT NULL; +-- Group concurrency checks (running tasks per group). +CREATE INDEX IF NOT EXISTS idx_tasks_group_running + ON tasks (group_key, status) + WHERE group_key IS NOT NULL AND status = 'running'; + +-- TTL expiry lookups for pending/paused/blocked tasks. +CREATE INDEX IF NOT EXISTS idx_tasks_expires + ON tasks (expires_at ASC) + WHERE expires_at IS NOT NULL AND status IN ('pending', 'paused', 'blocked'); + +-- Pending tasks with a future run_after (time-gating). +CREATE INDEX IF NOT EXISTS idx_tasks_run_after + ON tasks (run_after ASC) + WHERE run_after IS NOT NULL AND status = 'pending'; + +-- Blocked tasks (waiting on dependencies). +CREATE INDEX IF NOT EXISTS idx_tasks_blocked + ON tasks (status) + WHERE status = 'blocked'; diff --git a/migrations/002_add_label.sql b/migrations/002_add_label.sql deleted file mode 100644 index 060e7cb..0000000 --- a/migrations/002_add_label.sql +++ /dev/null @@ -1,5 +0,0 @@ --- Add a human-readable label column to both tables. --- The label stores the original dedup_key (or task_type if no explicit key), --- while the `key` column retains the SHA-256 hash for dedup. -ALTER TABLE tasks ADD COLUMN label TEXT NOT NULL DEFAULT ''; -ALTER TABLE task_history ADD COLUMN label TEXT NOT NULL DEFAULT ''; diff --git a/migrations/002_task_history.sql b/migrations/002_task_history.sql new file mode 100644 index 0000000..27eada3 --- /dev/null +++ b/migrations/002_task_history.sql @@ -0,0 +1,55 @@ +-- Completed and failed task history for queries and IO learning. +CREATE TABLE IF NOT EXISTS task_history ( + id INTEGER PRIMARY KEY, + task_type TEXT NOT NULL, + key TEXT NOT NULL, + label TEXT NOT NULL DEFAULT '', + priority INTEGER NOT NULL, + status TEXT NOT NULL, + payload BLOB, + memo BLOB DEFAULT NULL, + expected_read_bytes INTEGER NOT NULL DEFAULT 0, + expected_write_bytes INTEGER NOT NULL DEFAULT 0, + expected_net_rx_bytes INTEGER NOT NULL DEFAULT 0, + expected_net_tx_bytes INTEGER NOT NULL DEFAULT 0, + actual_read_bytes INTEGER, + actual_write_bytes INTEGER, + actual_net_rx_bytes INTEGER, + actual_net_tx_bytes INTEGER, + retry_count INTEGER NOT NULL DEFAULT 0, + max_retries INTEGER, + last_error TEXT, + created_at INTEGER NOT NULL, + started_at INTEGER, + completed_at INTEGER NOT NULL, + duration_ms INTEGER, + run_after INTEGER, + parent_id INTEGER, + fail_fast INTEGER NOT NULL DEFAULT 1, + group_key TEXT, + ttl_seconds INTEGER, + ttl_from TEXT NOT NULL DEFAULT 'submission', + expires_at INTEGER +); + +-- IO learning: recent completions by task type. +CREATE INDEX IF NOT EXISTS idx_history_type_completed + ON task_history (task_type, completed_at DESC) + WHERE status = 'completed'; + +-- Task lookup by key (dedup and status checks). +CREATE INDEX IF NOT EXISTS idx_history_key + ON task_history (key, completed_at DESC); + +-- Paginating and pruning history by completion time. +CREATE INDEX IF NOT EXISTS idx_history_completed + ON task_history (completed_at DESC); + +-- Filtering history by status (e.g. listing failed tasks). +CREATE INDEX IF NOT EXISTS idx_history_status + ON task_history (status, completed_at DESC); + +-- Lookup children of a parent task. +CREATE INDEX IF NOT EXISTS idx_task_history_parent + ON task_history (parent_id) + WHERE parent_id IS NOT NULL; diff --git a/migrations/003_net_io_and_groups.sql b/migrations/003_net_io_and_groups.sql deleted file mode 100644 index f3eb2a0..0000000 --- a/migrations/003_net_io_and_groups.sql +++ /dev/null @@ -1,16 +0,0 @@ --- Network IO columns for tasks table. -ALTER TABLE tasks ADD COLUMN expected_net_rx_bytes INTEGER NOT NULL DEFAULT 0; -ALTER TABLE tasks ADD COLUMN expected_net_tx_bytes INTEGER NOT NULL DEFAULT 0; -ALTER TABLE tasks ADD COLUMN group_key TEXT; - --- Network IO columns for task_history table. -ALTER TABLE task_history ADD COLUMN expected_net_rx_bytes INTEGER NOT NULL DEFAULT 0; -ALTER TABLE task_history ADD COLUMN expected_net_tx_bytes INTEGER NOT NULL DEFAULT 0; -ALTER TABLE task_history ADD COLUMN actual_net_rx_bytes INTEGER; -ALTER TABLE task_history ADD COLUMN actual_net_tx_bytes INTEGER; -ALTER TABLE task_history ADD COLUMN group_key TEXT; - --- Index for group concurrency checks (running tasks per group). -CREATE INDEX IF NOT EXISTS idx_tasks_group_running - ON tasks (group_key, status) - WHERE group_key IS NOT NULL AND status = 'running'; diff --git a/migrations/003_task_deps.sql b/migrations/003_task_deps.sql new file mode 100644 index 0000000..a178cda --- /dev/null +++ b/migrations/003_task_deps.sql @@ -0,0 +1,11 @@ +-- Dependency edges: (task_id=A, depends_on_id=B) means +-- "A cannot start until B completes." +CREATE TABLE IF NOT EXISTS task_deps ( + task_id INTEGER NOT NULL, + depends_on_id INTEGER NOT NULL, + PRIMARY KEY (task_id, depends_on_id) +); + +-- "Who depends on me?" query (used when a task completes). +CREATE INDEX IF NOT EXISTS idx_task_deps_depends_on + ON task_deps (depends_on_id); diff --git a/migrations/007_tags.sql b/migrations/004_task_tags.sql similarity index 59% rename from migrations/007_tags.sql rename to migrations/004_task_tags.sql index dc3ae29..6c3f34c 100644 --- a/migrations/007_tags.sql +++ b/migrations/004_task_tags.sql @@ -1,17 +1,18 @@ --- Task metadata tags (key-value pairs). +-- Task metadata tags (key-value pairs) for the active queue. CREATE TABLE IF NOT EXISTS task_tags ( task_id INTEGER NOT NULL, - key TEXT NOT NULL, - value TEXT NOT NULL, + key TEXT NOT NULL, + value TEXT NOT NULL, PRIMARY KEY (task_id, key) ); CREATE INDEX IF NOT EXISTS idx_task_tags_kv ON task_tags(key, value); +-- Task metadata tags (key-value pairs) for completed task history. CREATE TABLE IF NOT EXISTS task_history_tags ( history_rowid INTEGER NOT NULL, - key TEXT NOT NULL, - value TEXT NOT NULL, + key TEXT NOT NULL, + value TEXT NOT NULL, PRIMARY KEY (history_rowid, key) ); diff --git a/migrations/004_ttl.sql b/migrations/004_ttl.sql deleted file mode 100644 index 18e881f..0000000 --- a/migrations/004_ttl.sql +++ /dev/null @@ -1,13 +0,0 @@ --- Migration 004: Task TTL / automatic expiry --- Adds TTL columns to both tasks and task_history tables. - -ALTER TABLE tasks ADD COLUMN ttl_seconds INTEGER; -ALTER TABLE tasks ADD COLUMN ttl_from TEXT NOT NULL DEFAULT 'submission'; -ALTER TABLE tasks ADD COLUMN expires_at INTEGER; - -ALTER TABLE task_history ADD COLUMN ttl_seconds INTEGER; -ALTER TABLE task_history ADD COLUMN ttl_from TEXT NOT NULL DEFAULT 'submission'; -ALTER TABLE task_history ADD COLUMN expires_at INTEGER; - -CREATE INDEX IF NOT EXISTS idx_tasks_expires ON tasks (expires_at ASC) - WHERE expires_at IS NOT NULL AND status IN ('pending', 'paused', 'blocked'); diff --git a/migrations/005_scheduling.sql b/migrations/005_scheduling.sql deleted file mode 100644 index 0904b8c..0000000 --- a/migrations/005_scheduling.sql +++ /dev/null @@ -1,19 +0,0 @@ --- Delayed dispatch: task is pending but not eligible until this timestamp. --- NULL = immediately eligible (current behavior, backwards compatible). -ALTER TABLE tasks ADD COLUMN run_after INTEGER; - --- Recurring task template fields. --- Only set on the "template" row that spawns recurring instances. -ALTER TABLE tasks ADD COLUMN recurring_interval_secs INTEGER; -ALTER TABLE tasks ADD COLUMN recurring_max_executions INTEGER; -ALTER TABLE tasks ADD COLUMN recurring_execution_count INTEGER NOT NULL DEFAULT 0; -ALTER TABLE tasks ADD COLUMN recurring_paused INTEGER NOT NULL DEFAULT 0; - --- History: preserve the scheduling metadata for diagnostics. -ALTER TABLE task_history ADD COLUMN run_after INTEGER; - --- Partial index: only pending tasks with a future run_after need time-gating. --- The scheduler's peek query uses this to skip not-yet-ready tasks. -CREATE INDEX IF NOT EXISTS idx_tasks_run_after - ON tasks (run_after ASC) - WHERE run_after IS NOT NULL AND status = 'pending'; diff --git a/migrations/006_dependencies.sql b/migrations/006_dependencies.sql deleted file mode 100644 index 2077f5b..0000000 --- a/migrations/006_dependencies.sql +++ /dev/null @@ -1,22 +0,0 @@ --- Junction table for task dependency edges. --- A row (task_id=A, depends_on_id=B) means "A cannot start until B completes." -CREATE TABLE IF NOT EXISTS task_deps ( - task_id INTEGER NOT NULL, - depends_on_id INTEGER NOT NULL, - PRIMARY KEY (task_id, depends_on_id) -); - --- Index for the "who depends on me?" query (used when a task completes). -CREATE INDEX IF NOT EXISTS idx_task_deps_depends_on - ON task_deps (depends_on_id); - --- New status value: 'blocked' is now valid for tasks.status. --- No schema change needed — status is TEXT, not an enum. --- Add partial index for blocked tasks. -CREATE INDEX IF NOT EXISTS idx_tasks_blocked - ON tasks (status) - WHERE status = 'blocked'; - --- Column to store the dependency failure policy for blocked tasks. --- Values: 'cancel' (default), 'fail', 'ignore'. -ALTER TABLE tasks ADD COLUMN on_dep_failure TEXT NOT NULL DEFAULT 'cancel'; diff --git a/migrations/008_retry_backoff.sql b/migrations/008_retry_backoff.sql deleted file mode 100644 index c8a6a9e..0000000 --- a/migrations/008_retry_backoff.sql +++ /dev/null @@ -1,5 +0,0 @@ --- Per-task max_retries (resolved from per-type policy or global default at submit time). --- NULL means "use global default" — for backward compatibility with tasks --- created before this migration. -ALTER TABLE tasks ADD COLUMN max_retries INTEGER; -ALTER TABLE task_history ADD COLUMN max_retries INTEGER; diff --git a/migrations/009_memo.sql b/migrations/009_memo.sql deleted file mode 100644 index 8db0b72..0000000 --- a/migrations/009_memo.sql +++ /dev/null @@ -1,3 +0,0 @@ --- Execute-to-finalize memo: typed state persisted between phases. -ALTER TABLE tasks ADD COLUMN memo BLOB DEFAULT NULL; -ALTER TABLE task_history ADD COLUMN memo BLOB DEFAULT NULL; diff --git a/src/store/mod.rs b/src/store/mod.rs index 78912da..59a2590 100644 --- a/src/store/mod.rs +++ b/src/store/mod.rs @@ -237,70 +237,20 @@ impl TaskStore { /// Run the migration SQL. /// - /// Migrations are idempotent: `CREATE TABLE IF NOT EXISTS` for the - /// initial schema and `ALTER TABLE ADD COLUMN` for incremental changes - /// (SQLite returns "duplicate column name" if the column already exists, - /// which we silently ignore). + /// Migrations are idempotent (`CREATE TABLE/INDEX IF NOT EXISTS`). async fn migrate(&self) -> Result<(), StoreError> { sqlx::raw_sql(include_str!("../../migrations/001_tasks.sql")) .execute(&self.pool) .await?; - Self::run_alter_migration( - &self.pool, - include_str!("../../migrations/002_add_label.sql"), - ) - .await?; - Self::run_alter_migration( - &self.pool, - include_str!("../../migrations/003_net_io_and_groups.sql"), - ) - .await?; - Self::run_alter_migration(&self.pool, include_str!("../../migrations/004_ttl.sql")).await?; - Self::run_alter_migration( - &self.pool, - include_str!("../../migrations/005_scheduling.sql"), - ) - .await?; - Self::run_alter_migration( - &self.pool, - include_str!("../../migrations/006_dependencies.sql"), - ) - .await?; - Self::run_alter_migration(&self.pool, include_str!("../../migrations/007_tags.sql")) + sqlx::raw_sql(include_str!("../../migrations/002_task_history.sql")) + .execute(&self.pool) .await?; - Self::run_alter_migration( - &self.pool, - include_str!("../../migrations/008_retry_backoff.sql"), - ) - .await?; - Self::run_alter_migration(&self.pool, include_str!("../../migrations/009_memo.sql")) + sqlx::raw_sql(include_str!("../../migrations/003_task_deps.sql")) + .execute(&self.pool) + .await?; + sqlx::raw_sql(include_str!("../../migrations/004_task_tags.sql")) + .execute(&self.pool) .await?; - Ok(()) - } - - /// Run a migration containing `ALTER TABLE` statements, ignoring - /// "duplicate column name" errors (column already exists from a - /// previous run). - async fn run_alter_migration(pool: &SqlitePool, sql: &str) -> Result<(), StoreError> { - for statement in sql.split(';') { - // Strip comment-only lines but keep SQL after comments. - let meaningful: String = statement - .lines() - .filter(|line| !line.trim_start().starts_with("--")) - .collect::>() - .join("\n"); - let trimmed = meaningful.trim(); - if trimmed.is_empty() { - continue; - } - match sqlx::raw_sql(trimmed).execute(pool).await { - Ok(_) => {} - Err(e) if e.to_string().contains("duplicate column name") => { - continue; - } - Err(e) => return Err(e.into()), - } - } Ok(()) }