Skip to content

refactor!: consolidate migrations from 9 chronological files into 4 object-oriented files #76

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
deepjoy merged 1 commit into from
Mar 22, 2026
Merged

refactor!: consolidate migrations from 9 chronological files into 4 object-oriented files #76

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
115 changes: 53 additions & 62 deletions migrations/001_tasks.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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';
5 changes: 0 additions & 5 deletions migrations/002_add_label.sql
View file
Open in desktop

This file was deleted.

55 changes: 55 additions & 0 deletions migrations/002_task_history.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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;
16 changes: 0 additions & 16 deletions migrations/003_net_io_and_groups.sql
View file
Open in desktop

This file was deleted.

11 changes: 11 additions & 0 deletions migrations/003_task_deps.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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);
11 changes: 6 additions & 5 deletions migrations/007_tags.sql → migrations/004_task_tags.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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)
);

Expand Down
13 changes: 0 additions & 13 deletions migrations/004_ttl.sql
View file
Open in desktop

This file was deleted.

19 changes: 0 additions & 19 deletions migrations/005_scheduling.sql
View file
Open in desktop

This file was deleted.

22 changes: 0 additions & 22 deletions migrations/006_dependencies.sql
View file
Open in desktop

This file was deleted.

5 changes: 0 additions & 5 deletions migrations/008_retry_backoff.sql
View file
Open in desktop

This file was deleted.

3 changes: 0 additions & 3 deletions migrations/009_memo.sql
View file
Open in desktop

This file was deleted.

66 changes: 8 additions & 58 deletions src/store/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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::<Vec<_>>()
.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(())
}

Expand Down
Loading