vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(#2279): idempotent column-guard migrations, drop invalid ADD COLUMN IF NOT EXISTS

Browse source
This commit is contained in:
damocles 2026-07-10 16:51:50 +02:00 • committed by mara
commit dfbf198ef3
5 changed files with 188 additions and 121 deletions
Download patch file Download diff file Expand all files Collapse all files

37
hive-c0re/src/stores/approvals.rs
View file

@ -10,6 +10,8 @@ use hive_sh4re::wire_time::now_unix;
use hive_sh4re::{Approval, ApprovalKind, ApprovalStatus};
use rusqlite::{Connection, OptionalExtension, params};
use crate::db::Migration;
const SCHEMA: &str = r"
CREATE TABLE IF NOT EXISTS approvals (
id INTEGER PRIMARY KEY AUTOINCREMENT,
@ -25,20 +27,32 @@ CREATE INDEX IF NOT EXISTS idx_approvals_pending
";
/// Ordered schema migrations tracked in `schema_versions` (key `"approvals"`).
/// Legacy databases are detected via the `submitter` column — the last
/// column added before versioning was introduced — and fast-forwarded past
/// all known migrations. New columns go here as v5, v6, …
const MIGRATIONS: &[&str] = &[
/// Each migration declares the column it adds, so a legacy DB (fully or
/// partially migrated before versioning) converges by skipping the migrations
/// whose column already exists. New columns go here as v5, v6, …
const MIGRATIONS: &[Migration] = &[
// v1: `kind` (pre-Phase-8 dbs): legacy rows default to `apply_commit`.
"ALTER TABLE approvals ADD COLUMN IF NOT EXISTS \
Migration {
sql: "ALTER TABLE approvals ADD COLUMN \
kind TEXT NOT NULL DEFAULT 'apply_commit'",
adds_column: Some(("approvals", "kind")),
},
// v2: `description`: manager-supplied note on the dashboard card.
"ALTER TABLE approvals ADD COLUMN IF NOT EXISTS description TEXT",
Migration {
sql: "ALTER TABLE approvals ADD COLUMN description TEXT",
adds_column: Some(("approvals", "description")),
},
// v3: `fetched_sha`: canonical sha hive-c0re resolved at submit time.
"ALTER TABLE approvals ADD COLUMN IF NOT EXISTS fetched_sha TEXT",
Migration {
sql: "ALTER TABLE approvals ADD COLUMN fetched_sha TEXT",
adds_column: Some(("approvals", "fetched_sha")),
},
// v4: `submitter`: authenticated agent that submitted the approval.
// Legacy rows are NULL → callers fall back to the root agent.
"ALTER TABLE approvals ADD COLUMN IF NOT EXISTS submitter TEXT",
Migration {
sql: "ALTER TABLE approvals ADD COLUMN submitter TEXT",
adds_column: Some(("approvals", "submitter")),
},
];
pub struct Approvals {
@ -50,12 +64,7 @@ impl Approvals {
let conn = crate::db::open(path, "approvals")?;
conn.execute_batch(SCHEMA)
.context("apply approvals schema")?;
crate::db::apply_versioned_migrations(
&conn,
"approvals",
("approvals", "submitter"),
MIGRATIONS,
)?;
crate::db::apply_versioned_migrations(&conn, "approvals", MIGRATIONS)?;
Ok(Self {
conn: Mutex::new(conn),
})

60
hive-c0re/src/stores/broker.rs
View file

@ -10,6 +10,8 @@ use anyhow::{Context, Result};
use chrono::{DateTime, Utc};
use hive_sh4re::wire_time::now_unix;
use hive_sh4re::{InboxRow, Message};
use crate::db::Migration;
use rusqlite::{Connection, OptionalExtension, params};
use serde::Serialize;
use tokio::sync::broadcast;
@ -157,52 +159,60 @@ pub struct Broker {
}
/// Ordered schema migrations for the broker. Tracked in `schema_versions`
/// under key `"broker"`. Legacy databases (fully migrated via the old
/// per-column check approach) are detected via the `priority` column on
/// `messages` — the last column added before versioning.
const BROKER_MIGRATIONS: &[&str] = &[
/// under key `"broker"`. Each migration declares the column it adds, so a
/// legacy DB (fully or partially migrated via the old per-column approach)
/// is converged by skipping the migrations whose column already exists.
const BROKER_MIGRATIONS: &[Migration] = &[
// v1: acked_at on messages, with backfill so existing delivered rows
// are not phantom-requeued on the next open.
//
// IF NOT EXISTS: safe to run on a partially-migrated legacy DB where
// acked_at already exists but priority (the legacy marker) does not.
// The BEGIN/COMMIT block makes ALTER + UPDATE atomic — if the UPDATE
// fails the version counter is not bumped and the whole step retries.
"BEGIN;\
ALTER TABLE messages ADD COLUMN IF NOT EXISTS acked_at INTEGER;\
// are not phantom-requeued on the next open. The BEGIN/COMMIT block
// makes ALTER + UPDATE atomic — either both land or neither, so the
// guard column (acked_at) faithfully marks the whole step as done.
Migration {
sql: "BEGIN;\
ALTER TABLE messages ADD COLUMN acked_at INTEGER;\
UPDATE messages SET acked_at = delivered_at \
WHERE delivered_at IS NOT NULL AND acked_at IS NULL;\
COMMIT;",
adds_column: Some(("messages", "acked_at")),
},
// v2: in_reply_to for thread-parent tracking. NULL = root of a thread.
"ALTER TABLE messages ADD COLUMN IF NOT EXISTS in_reply_to INTEGER",
Migration {
sql: "ALTER TABLE messages ADD COLUMN in_reply_to INTEGER",
adds_column: Some(("messages", "in_reply_to")),
},
// v3: priority for operator-message fast-path. Rebuild the delivery
// index to include priority as a secondary sort key.
"BEGIN;\
ALTER TABLE messages ADD COLUMN IF NOT EXISTS \
// index to include priority as a secondary sort key. Atomic BEGIN/COMMIT
// so priority existing implies the index rebuild also committed.
Migration {
sql: "BEGIN;\
ALTER TABLE messages ADD COLUMN \
priority INTEGER NOT NULL DEFAULT 0;\
DROP INDEX IF EXISTS idx_messages_undelivered;\
CREATE INDEX IF NOT EXISTS idx_messages_undelivered \
ON messages (recipient, priority DESC, id) WHERE delivered_at IS NULL;\
COMMIT;",
adds_column: Some(("messages", "priority")),
},
// v4: attempt_count on reminders for the MAX_REMINDER_ATTEMPTS cap.
"ALTER TABLE reminders ADD COLUMN IF NOT EXISTS \
Migration {
sql: "ALTER TABLE reminders ADD COLUMN \
attempt_count INTEGER NOT NULL DEFAULT 0",
adds_column: Some(("reminders", "attempt_count")),
},
// v5: last_error on reminders — last delivery failure surfaced on the
// dashboard so a stuck reminder is visible without digging in logs.
"ALTER TABLE reminders ADD COLUMN IF NOT EXISTS last_error TEXT",
Migration {
sql: "ALTER TABLE reminders ADD COLUMN last_error TEXT",
adds_column: Some(("reminders", "last_error")),
},
];
impl Broker {
pub fn open(path: &Path) -> Result<Self> {
let conn = crate::db::open(path, "broker")?;
conn.execute_batch(SCHEMA).context("apply broker schema")?;
crate::db::apply_versioned_migrations(
&conn,
"broker",
("messages", "priority"),
BROKER_MIGRATIONS,
)
.context("broker migrations")?;
crate::db::apply_versioned_migrations(&conn, "broker", BROKER_MIGRATIONS)
.context("broker migrations")?;
let (events, _) = broadcast::channel(EVENT_CHANNEL);
Ok(Self {
conn: Mutex::new(conn),

171
hive-c0re/src/stores/db.rs
View file

@ -11,10 +11,12 @@
//! tracks the applied count in a `schema_versions` table (one row per
//! `subsystem` key). Only migrations past the stored version run.
//!
//! Legacy databases (written before versioning) are detected via
//! `legacy_column = (table, column)` — a column that exists only in
//! a fully-migrated legacy DB. Present → skip all migrations. Absent →
//! start from 0. See each store's `MIGRATIONS` constant for the list.
//! Legacy databases (written before versioning) need no special marker:
//! each column-adding migration carries the `(table, column)` it
//! introduces, so re-walking a pre-versioning DB from version 0 simply
//! skips the migrations whose column already exists — converging a fully-
//! or partially-migrated legacy DB without ever re-running an `ADD COLUMN`
//! against an existing column. See each store's `MIGRATIONS` constant.
use std::path::Path;
use std::time::Duration;
@ -33,36 +35,57 @@ CREATE TABLE IF NOT EXISTS schema_versions (
version INTEGER NOT NULL DEFAULT 0
)";
/// One numbered schema migration.
///
/// `sql` runs once, at the index whose value is the current stored version.
/// `adds_column`, when set, is the `(table, column)` this migration
/// introduces: if that column already exists — a pre-versioning DB migrated
/// by the old try-and-ignore path, or a partially-migrated one — `sql` is
/// skipped and only the version counter advances. This makes column-adding
/// migrations idempotent *without* `SQLite`'s unsupported `ALTER TABLE … ADD
/// COLUMN IF NOT EXISTS`, so re-walking a legacy DB from version 0 never
/// re-runs an `ADD COLUMN` against a column that is already there (which
/// would fail with "duplicate column name" and wedge startup permanently).
///
/// Multi-statement migrations wrap their `ALTER` + backfill / index rebuild
/// in an explicit `BEGIN; … COMMIT;` batch, so "the added column exists" is a
/// faithful proxy for "this whole migration committed".
pub struct Migration {
/// SQL for this version — a single statement or a `BEGIN; … COMMIT;` batch.
pub sql: &'static str,
/// `(table, column)` this migration adds, if any. Present → the migration
/// is skipped when the column already exists (idempotent replay).
pub adds_column: Option<(&'static str, &'static str)>,
}
/// Apply numbered schema migrations for `subsystem`, tracking progress in
/// the shared `schema_versions` table.
///
/// - `subsystem` — unique key for this store in `schema_versions`
/// (e.g. `"approvals"`, `"broker"`).
/// - `legacy_column` — `(table_name, column_name)` of a column that exists
/// in a fully-migrated pre-versioning database. When there is no
/// `schema_versions` row yet and this column is present, all `migrations`
/// are skipped (they were previously applied via the old try-and-ignore
/// approach). When the column is absent, we start at version 0.
/// - `migrations` — ordered list of SQL batches; each batch runs once, at
/// the index whose value is the current version.
/// - `migrations` — ordered list of [`Migration`]s; migration `i` runs when
/// the stored version is `i`. A pre-versioning DB (no `schema_versions`
/// row) starts at version 0 and re-walks every migration, but each
/// column-adding migration is skipped when its `adds_column` is already
/// present — so a fully- or partially-migrated legacy DB converges without
/// re-running an `ADD COLUMN` against an existing column.
///
/// # Errors
///
/// Returns an error if any migration statement fails or the version update
/// fails. A failed migration leaves the DB at the last successfully
/// committed version (each step is its own implicit transaction via
/// `execute_batch`).
/// fails. Each migration + its version bump is a single `execute_batch`
/// (multi-statement migrations wrap themselves in `BEGIN; … COMMIT;`), so a
/// failure leaves the DB at the last successfully committed version.
pub fn apply_versioned_migrations(
conn: &Connection,
subsystem: &str,
legacy_column: (&str, &str),
migrations: &[&str],
migrations: &[Migration],
) -> Result<()> {
// Ensure the version-tracking table exists (idempotent).
conn.execute_batch(SCHEMA_VERSIONS_DDL)
.context("create schema_versions table")?;
// Look up the stored version for this subsystem.
// Current version for this subsystem — 0 (and a fresh row) when absent.
let stored: Option<i64> = conn
.query_row(
"SELECT version FROM schema_versions WHERE store = ?1",
@ -71,43 +94,32 @@ pub fn apply_versioned_migrations(
)
.optional()
.context("read schema_versions")?;
let version = if let Some(v) = stored {
usize::try_from(v.max(0)).unwrap_or(0)
} else {
// No entry yet. Check the legacy-column to decide where to start.
// If it exists, all known migrations were already applied by the
// old try-and-ignore path — skip them. If it doesn't exist, start
// at 0 (fresh install or partial state).
let (legacy_table, legacy_col) = legacy_column;
// SAFETY: `legacy_table` is always a static table-name literal
// supplied by each store's `open` (e.g. `"messages"`). Dynamic
// or untrusted table names must not be passed here — if that
// ever changes, switch to `pragma_table_info(?1)` with two
// bound parameters.
let is_legacy: bool = conn
.prepare(&format!(
"SELECT 1 FROM pragma_table_info('{legacy_table}') WHERE name = ?1"
))
.context("prepare legacy-column probe")?
.exists([legacy_col])
.context("check legacy column")?;
let initial = if is_legacy { migrations.len() } else { 0 };
conn.execute(
"INSERT INTO schema_versions (store, version) VALUES (?1, ?2)",
params![subsystem, i64::try_from(initial).unwrap_or(0)],
"INSERT INTO schema_versions (store, version) VALUES (?1, 0)",
[subsystem],
)
.context("insert schema_versions row")?;
initial
0
};
if version >= migrations.len() {
return Ok(());
}
for (i, stmt) in migrations.iter().enumerate().skip(version) {
conn.execute_batch(stmt)
.with_context(|| format!("{subsystem} migration v{} failed: {stmt}", i + 1))?;
for (i, mig) in migrations.iter().enumerate().skip(version) {
// Skip a column-adding migration whose column is already present:
// a legacy DB migrated by the old path, or a resumed partial one.
let already_applied = match mig.adds_column {
Some((table, column)) => column_exists(conn, table, column)?,
None => false,
};
if !already_applied {
conn.execute_batch(mig.sql)
.with_context(|| format!("{subsystem} migration v{} failed: {}", i + 1, mig.sql))?;
}
let next = i64::try_from(i + 1).unwrap_or(i64::MAX);
conn.execute(
"UPDATE schema_versions SET version = ?1 WHERE store = ?2",
@ -118,6 +130,17 @@ pub fn apply_versioned_migrations(
Ok(())
}
/// Whether `column` exists on `table` in this connection's schema. Both
/// arguments are bound parameters — `pragma_table_info(?1)` accepts the table
/// name as a bound value — so this is not a SQL-injection surface even if a
/// caller ever passes a non-static name.
fn column_exists(conn: &Connection, table: &str, column: &str) -> Result<bool> {
conn.prepare("SELECT 1 FROM pragma_table_info(?1) WHERE name = ?2")
.context("prepare column probe")?
.exists(params![table, column])
.context("check column existence")
}
/// Open a connection to the sqlite file at `path`, creating the parent
/// directory if needed. `subsystem` labels error contexts (`"broker"`,
/// `"approvals"`, …).
@ -147,17 +170,26 @@ mod tests {
}
const TEST_SCHEMA: &str = "CREATE TABLE IF NOT EXISTS things (id INTEGER PRIMARY KEY)";
const TEST_MIGRATIONS: &[&str] = &[
"ALTER TABLE things ADD COLUMN IF NOT EXISTS alpha TEXT",
"ALTER TABLE things ADD COLUMN IF NOT EXISTS beta INTEGER NOT NULL DEFAULT 0",
"ALTER TABLE things ADD COLUMN IF NOT EXISTS gamma TEXT",
const TEST_MIGRATIONS: &[Migration] = &[
Migration {
sql: "ALTER TABLE things ADD COLUMN alpha TEXT",
adds_column: Some(("things", "alpha")),
},
Migration {
sql: "ALTER TABLE things ADD COLUMN beta INTEGER NOT NULL DEFAULT 0",
adds_column: Some(("things", "beta")),
},
Migration {
sql: "ALTER TABLE things ADD COLUMN gamma TEXT",
adds_column: Some(("things", "gamma")),
},
];
#[test]
fn fresh_install_runs_all_migrations() {
let conn = open_tmp();
conn.execute_batch(TEST_SCHEMA).unwrap();
apply_versioned_migrations(&conn, "things", ("things", "gamma"), TEST_MIGRATIONS).unwrap();
apply_versioned_migrations(&conn, "things", TEST_MIGRATIONS).unwrap();
// All three columns should exist.
for col in ["alpha", "beta", "gamma"] {
let exists: bool = conn
@ -190,7 +222,7 @@ mod tests {
alpha TEXT, beta INTEGER NOT NULL DEFAULT 0, gamma TEXT)",
)
.unwrap();
apply_versioned_migrations(&conn, "things", ("things", "gamma"), TEST_MIGRATIONS).unwrap();
apply_versioned_migrations(&conn, "things", TEST_MIGRATIONS).unwrap();
// Version should be set to migrations.len() (3), not 0.
let v: i64 = conn
.query_row(
@ -212,7 +244,7 @@ mod tests {
INSERT INTO schema_versions (store, version) VALUES ('things', 1)",
)
.unwrap();
apply_versioned_migrations(&conn, "things", ("things", "gamma"), TEST_MIGRATIONS).unwrap();
apply_versioned_migrations(&conn, "things", TEST_MIGRATIONS).unwrap();
// Only beta and gamma should have been added (alpha already existed).
for col in ["alpha", "beta", "gamma"] {
let exists: bool = conn
@ -241,9 +273,9 @@ mod tests {
fn already_at_latest_is_noop() {
let conn = open_tmp();
conn.execute_batch(TEST_SCHEMA).unwrap();
apply_versioned_migrations(&conn, "things", ("things", "gamma"), TEST_MIGRATIONS).unwrap();
apply_versioned_migrations(&conn, "things", TEST_MIGRATIONS).unwrap();
// Second call is a no-op.
apply_versioned_migrations(&conn, "things", ("things", "gamma"), TEST_MIGRATIONS).unwrap();
apply_versioned_migrations(&conn, "things", TEST_MIGRATIONS).unwrap();
let v: i64 = conn
.query_row(
"SELECT version FROM schema_versions WHERE store = 'things'",
@ -265,15 +297,19 @@ mod tests {
apply_versioned_migrations(
&conn,
"store_a",
("a", "col1"),
&["ALTER TABLE a ADD COLUMN col1 TEXT"],
&[Migration {
sql: "ALTER TABLE a ADD COLUMN col1 TEXT",
adds_column: Some(("a", "col1")),
}],
)
.unwrap();
apply_versioned_migrations(
&conn,
"store_b",
("b", "colx"),
&["ALTER TABLE b ADD COLUMN colx INTEGER NOT NULL DEFAULT 0"],
&[Migration {
sql: "ALTER TABLE b ADD COLUMN colx INTEGER NOT NULL DEFAULT 0",
adds_column: Some(("b", "colx")),
}],
)
.unwrap();
// Each store tracks independently.
@ -296,23 +332,20 @@ mod tests {
}
/// A partially-migrated legacy database: some columns from the old
/// try-and-ignore path exist, but the legacy marker column (used to
/// fast-forward fresh schema_versions entries to `len`) is absent.
/// With idempotent `ADD COLUMN IF NOT EXISTS` migrations this must
/// succeed — the already-present columns are skipped and only the
/// missing ones are added.
/// try-and-ignore path exist, with no `schema_versions` row yet. The
/// per-migration `adds_column` guard must skip the already-present
/// columns and add only the missing ones — no "duplicate column name".
#[test]
fn partial_legacy_completes_without_error() {
let conn = open_tmp();
// Simulate a DB that has `alpha` (v1) but not `beta` (v2) or
// `gamma` (v3). No schema_versions row exists yet, and `gamma`
// (the legacy marker) is absent → the function cannot fast-forward
// and must start from v0 with idempotent migrations.
// `gamma` (v3). No schema_versions row exists → start from v0 and
// guard-skip alpha while adding beta + gamma.
conn.execute_batch(
"CREATE TABLE IF NOT EXISTS things (id INTEGER PRIMARY KEY, alpha TEXT)",
)
.unwrap();
apply_versioned_migrations(&conn, "things", ("things", "gamma"), TEST_MIGRATIONS).unwrap();
apply_versioned_migrations(&conn, "things", TEST_MIGRATIONS).unwrap();
// All three columns must now exist.
for col in ["alpha", "beta", "gamma"] {
let exists: bool = conn
@ -322,7 +355,10 @@ mod tests {
.unwrap()
.exists([])
.unwrap();
assert!(exists, "column {col} missing after partial-legacy migration");
assert!(
exists,
"column {col} missing after partial-legacy migration"
);
}
// Version must be at the latest.
let v: i64 = conn
@ -332,6 +368,9 @@ mod tests {
|r| r.get(0),
)
.unwrap();
assert_eq!(v, 3, "must reach latest version after completing partial legacy");
assert_eq!(
v, 3,
"must reach latest version after completing partial legacy"
);
}
}

26
hive-c0re/src/stores/operator_questions.rs
View file

@ -19,6 +19,8 @@ use hive_sh4re::wire_time::now_unix;
use rusqlite::{Connection, OptionalExtension, params};
use serde::Serialize;
use crate::db::Migration;
const SCHEMA: &str = r"
CREATE TABLE IF NOT EXISTS operator_questions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
@ -37,16 +39,25 @@ CREATE INDEX IF NOT EXISTS idx_operator_questions_pending
/// `"operator_questions"`). Legacy databases are detected via the `target`
/// column — the last column added before versioning — and fast-forwarded
/// past all known migrations.
const MIGRATIONS: &[&str] = &[
const MIGRATIONS: &[Migration] = &[
// v1: `multi` — checkbox-style multi-option questions.
"ALTER TABLE operator_questions ADD COLUMN IF NOT EXISTS \
Migration {
sql: "ALTER TABLE operator_questions ADD COLUMN \
multi INTEGER NOT NULL DEFAULT 0",
adds_column: Some(("operator_questions", "multi")),
},
// v2: `deadline_at` — optional TTL after which the watchdog auto-resolves.
"ALTER TABLE operator_questions ADD COLUMN IF NOT EXISTS deadline_at INTEGER",
Migration {
sql: "ALTER TABLE operator_questions ADD COLUMN deadline_at INTEGER",
adds_column: Some(("operator_questions", "deadline_at")),
},
// v3: `target` — recipient of the question. NULL = operator (back-compat
// default); non-null = peer-to-peer question. Dashboard's `pending()`
// filters on `target IS NULL` so peer questions never leak to the operator.
"ALTER TABLE operator_questions ADD COLUMN IF NOT EXISTS target TEXT",
Migration {
sql: "ALTER TABLE operator_questions ADD COLUMN target TEXT",
adds_column: Some(("operator_questions", "target")),
},
];
#[derive(Debug, Clone, Serialize)]
@ -80,12 +91,7 @@ impl OperatorQuestions {
let conn = crate::db::open(path, "operator_questions")?;
conn.execute_batch(SCHEMA)
.context("apply operator_questions schema")?;
crate::db::apply_versioned_migrations(
&conn,
"operator_questions",
("operator_questions", "target"),
MIGRATIONS,
)?;
crate::db::apply_versioned_migrations(&conn, "operator_questions", MIGRATIONS)?;
Ok(Self {
conn: Mutex::new(conn),
})

15
hive-c0re/src/stores/scheduled_prompts.rs
View file

@ -20,6 +20,8 @@ use hive_sh4re::wire_time::now_unix;
use rusqlite::{Connection, OptionalExtension, params};
use serde::{Deserialize, Serialize};
use crate::db::Migration;
/// Typed error returned by [`ScheduledPrompts::pause`] and
/// [`ScheduledPrompts::resume`] when the target row does not exist or
/// is already cancelled. Handlers downcast on this type to emit 404
@ -186,17 +188,18 @@ impl ScheduledPrompts {
.context("enable foreign keys")?;
conn.execute_batch(SCHEMA)
.context("apply scheduled_prompts schema")?;
// Versioned migration: add paused_at_unix. Legacy databases are
// detected via the presence of this column (it was the only
// migration before versioning was introduced).
// Versioned migration: add paused_at_unix. The per-migration
// `adds_column` guard skips the ALTER on a legacy DB that already
// has the column (it was the only migration before versioning).
crate::db::apply_versioned_migrations(
&conn,
"scheduled_prompts",
("scheduled_prompts", "paused_at_unix"),
&[
// v1: add paused_at_unix for per-schedule pause support.
"ALTER TABLE scheduled_prompts \
ADD COLUMN IF NOT EXISTS paused_at_unix INTEGER",
Migration {
sql: "ALTER TABLE scheduled_prompts ADD COLUMN paused_at_unix INTEGER",
adds_column: Some(("scheduled_prompts", "paused_at_unix")),
},
],
)?;
// Migration: recreate the due-rows index to also exclude paused