Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

fix: migration error

Browse Source
This commit is contained in:
Hare
2026-08-23 02:10:03 +09:00
parent 24291a4545
commit b12e0785e2
1 changed files with 274 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/workspace-server/src/store.rs
+274 -7
View File
@@ -1220,6 +1220,15 @@ impl SqliteWorkspaceStore {
"materialize {assignment_worker_tombstone_repairs} legacy Ticket assignment Worker tombstone(s)"
));
}
apply_migrations_through(&candidate, 42)?;
if current_schema_version < 43 {
let stale_current = repairable_legacy_current_ticket_assignment_count(&candidate)?;
if stale_current > 0 {
repairs.push(format!(
"clear {stale_current} tombstoned legacy current Ticket assignment pointer(s)"
));
}
}
apply_migrations_through(&candidate, i64::MAX)?;
ticket::migrate_sqlite_ticket_schema(&candidate)?;
merge_request::migrate(&candidate).map_err(|error| Error::Store(error.to_string()))?;
@@ -7517,10 +7526,20 @@ fn create_workspace_memory_settings_authority(conn: &Connection) -> Result<()> {
Ok(())
}
fn generalize_ticket_role_assignments(conn: &Connection) -> Result<()> {
let invalid_current: i64 = conn.query_row(
fn repairable_legacy_current_ticket_assignment_count(conn: &Connection) -> Result<i64> {
let (stale_current, unclassified_current): (i64, i64) = conn.query_row(
r#"
SELECT COUNT(*)
SELECT
COALESCE(SUM(CASE
WHEN ticket.ticket_id IS NULL OR worker.worker_id IS NULL THEN 1
ELSE 0
END), 0),
COALESCE(SUM(CASE
WHEN (ticket.ticket_id IS NULL AND ticket_tombstone.ticket_id IS NULL)
OR (worker.worker_id IS NULL AND worker_tombstone.worker_id IS NULL)
THEN 1
ELSE 0
END), 0)
FROM ticket_current_worker_assignments AS current
LEFT JOIN typed_tickets AS ticket
ON ticket.workspace_id = current.workspace_id
@@ -7529,16 +7548,51 @@ fn generalize_ticket_role_assignments(conn: &Connection) -> Result<()> {
ON worker.workspace_id = current.workspace_id
AND worker.runtime_id = current.runtime_id
AND worker.worker_id = current.worker_id
WHERE ticket.ticket_id IS NULL OR worker.worker_id IS NULL
LEFT JOIN ticket_assignment_ticket_tombstones AS ticket_tombstone
ON ticket_tombstone.workspace_id = current.workspace_id
AND ticket_tombstone.ticket_id = current.ticket_id
LEFT JOIN ticket_assignment_worker_tombstones AS worker_tombstone
ON worker_tombstone.workspace_id = current.workspace_id
AND worker_tombstone.runtime_id = current.runtime_id
AND worker_tombstone.worker_id = current.worker_id
"#,
[],
|row| row.get(0),
|row| Ok((row.get(0)?, row.get(1)?)),
)?;
if invalid_current != 0 {
if unclassified_current != 0 {
return Err(Error::Store(format!(
"migration 43 cannot classify {invalid_current} legacy current Ticket assignment(s) as valid Coder Worker principals"
"migration 43 cannot repair {unclassified_current} legacy current Ticket assignment(s) whose missing Ticket or Worker has no matching tombstone"
)));
}
Ok(stale_current)
}
fn generalize_ticket_role_assignments(conn: &Connection) -> Result<()> {
let stale_current = repairable_legacy_current_ticket_assignment_count(conn)?;
if stale_current > 0 {
let cleared = conn.execute(
r#"
DELETE FROM ticket_current_worker_assignments AS current
WHERE NOT EXISTS (
SELECT 1 FROM typed_tickets AS ticket
WHERE ticket.workspace_id = current.workspace_id
AND ticket.ticket_id = current.ticket_id
)
OR NOT EXISTS (
SELECT 1 FROM worker_registry AS worker
WHERE worker.workspace_id = current.workspace_id
AND worker.runtime_id = current.runtime_id
AND worker.worker_id = current.worker_id
)
"#,
[],
)?;
if i64::try_from(cleared).ok() != Some(stale_current) {
return Err(Error::Store(format!(
"migration 43 classified {stale_current} stale current Ticket assignment(s) but cleared {cleared}"
)));
}
}
conn.execute_batch(
r#"
@@ -7574,6 +7628,7 @@ fn generalize_ticket_role_assignments(conn: &Connection) -> Result<()> {
assigned_by TEXT NOT NULL,
assigned_at TEXT NOT NULL,
PRIMARY KEY (workspace_id, assignment_id),
UNIQUE (workspace_id, ticket_id, assignment_id),
UNIQUE (workspace_id, ticket_id, role, assignment_id),
UNIQUE (workspace_id, ticket_id, role, assignment_id, principal_kind, principal_id, runtime_id, worker_id),
FOREIGN KEY (workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE,
@@ -8621,6 +8676,58 @@ pub(crate) fn apply_migrations_through(conn: &Connection, through_version: i64)
continue;
}
if migration.version == 43 {
// Other in-database authorities may reference the assignment history table. Keep
// those foreign keys on the canonical table name while rebuilding its role-aware
// schema, then validate the complete database before restoring FK enforcement.
conn.execute_batch("PRAGMA foreign_keys = OFF; PRAGMA legacy_alter_table = ON;")?;
let result = (|| -> Result<()> {
let tx = conn.unchecked_transaction()?;
(migration.apply)(&tx)?;
let dangling_reference: Option<(String, String)> = tx
.query_row(
"SELECT name, sql FROM sqlite_schema \
WHERE sql LIKE '%ticket_worker_assignments_v43%' LIMIT 1",
[],
|row| Ok((row.get(0)?, row.get(1)?)),
)
.optional()?;
if let Some((object, sql)) = dangling_reference {
return Err(Error::Store(format!(
"migration 43 left a temporary Ticket assignment reference in `{object}`: {sql}"
)));
}
let foreign_key_failures: i64 = tx
.query_row("SELECT COUNT(*) FROM pragma_foreign_key_check", [], |row| {
row.get(0)
})
.map_err(|error| {
Error::Store(format!(
"migration 43 could not evaluate foreign keys: {error}"
))
})?;
if foreign_key_failures != 0 {
return Err(Error::Store(format!(
"migration 43 found {foreign_key_failures} foreign key violation(s)"
)));
}
tx.execute(
"INSERT INTO __yoi_schema_migrations (version, name) VALUES (?1, ?2)",
params![migration.version, migration.name],
)?;
tx.commit()?;
Ok(())
})();
conn.execute_batch("PRAGMA legacy_alter_table = OFF; PRAGMA foreign_keys = ON;")
.map_err(|error| {
Error::Store(format!(
"migration 43 could not restore FK enforcement: {error}"
))
})?;
result?;
continue;
}
let tx = conn.unchecked_transaction()?;
if migration.version == 37 {
crate::retention::repair_worker_diagnostics_archive_table(&tx)?;
@@ -10493,6 +10600,166 @@ INSERT INTO worker_registry (
);
}
#[test]
fn schema_v43_clears_tombstoned_legacy_current_assignment_pointers() {
let dir = tempfile::tempdir().unwrap();
let db_path = dir.path().join("legacy-v42.db");
let conn = Connection::open(&db_path).unwrap();
configure_sqlite(&conn).unwrap();
apply_migrations_through(&conn, 42).unwrap();
conn.execute_batch(
r#"
INSERT INTO workspaces (
workspace_id, display_name, state, created_at, updated_at
) VALUES ('workspace-legacy', 'Legacy', 'active', '2026-09-01', '2026-09-01');
INSERT INTO typed_tickets (
workspace_id, ticket_id, slug, title, status, kind, priority, body,
workflow_state, workflow_state_explicit
) VALUES
('workspace-legacy', 'ticket-deleted-worker', 'deleted-worker',
'Deleted Worker', 'open', 'task', 'normal', '', 'in_progress', 1),
('workspace-legacy', 'ticket-deleted-ticket', 'deleted-ticket',
'Deleted Ticket', 'open', 'task', 'normal', '', 'in_progress', 1);
INSERT INTO worker_registry (
workspace_id, runtime_id, worker_id, display_name, retention_state,
created_at, updated_at
) VALUES
('workspace-legacy', 'runtime-legacy',
'00000000-0000-7000-8000-000000000001', 'Deleted Worker', 'normal',
'2026-09-01', '2026-09-01'),
('workspace-legacy', 'runtime-legacy',
'00000000-0000-7000-8000-000000000002', 'Retained Worker', 'normal',
'2026-09-01', '2026-09-01');
INSERT INTO ticket_worker_assignments (
workspace_id, ticket_id, assignment_id, runtime_id, worker_id,
assigned_by, assigned_at
) VALUES
('workspace-legacy', 'ticket-deleted-worker', 'assignment-deleted-worker',
'runtime-legacy', '00000000-0000-7000-8000-000000000001',
'legacy', '2026-09-01'),
('workspace-legacy', 'ticket-deleted-ticket', 'assignment-deleted-ticket',
'runtime-legacy', '00000000-0000-7000-8000-000000000002',
'legacy', '2026-09-01');
INSERT INTO ticket_current_worker_assignments (
workspace_id, ticket_id, assignment_id, runtime_id, worker_id, updated_at
) VALUES
('workspace-legacy', 'ticket-deleted-worker', 'assignment-deleted-worker',
'runtime-legacy', '00000000-0000-7000-8000-000000000001', '2026-09-01'),
('workspace-legacy', 'ticket-deleted-ticket', 'assignment-deleted-ticket',
'runtime-legacy', '00000000-0000-7000-8000-000000000002', '2026-09-01');
INSERT INTO ticket_worker_assignment_events (
workspace_id, ticket_id, event_id, action, assignment_id, actor, created_at
) VALUES
('workspace-legacy', 'ticket-deleted-worker', 'event-deleted-worker',
'assigned', 'assignment-deleted-worker', 'legacy', '2026-09-01'),
('workspace-legacy', 'ticket-deleted-ticket', 'event-deleted-ticket',
'assigned', 'assignment-deleted-ticket', 'legacy', '2026-09-01');
CREATE TABLE legacy_assignment_consumer (
workspace_id TEXT NOT NULL,
ticket_id TEXT NOT NULL,
assignment_id TEXT NOT NULL,
FOREIGN KEY (workspace_id, ticket_id, assignment_id)
REFERENCES ticket_worker_assignments (
workspace_id, ticket_id, assignment_id
)
);
INSERT INTO legacy_assignment_consumer
VALUES ('workspace-legacy', 'ticket-deleted-worker', 'assignment-deleted-worker');
DELETE FROM worker_registry
WHERE workspace_id = 'workspace-legacy'
AND worker_id = '00000000-0000-7000-8000-000000000001';
DELETE FROM typed_tickets
WHERE workspace_id = 'workspace-legacy'
AND ticket_id = 'ticket-deleted-ticket';
"#,
)
.unwrap();
assert_eq!(
conn.query_row(
"SELECT COUNT(*) FROM ticket_current_worker_assignments",
[],
|row| row.get::<_, i64>(0),
)
.unwrap(),
2
);
drop(conn);
let plan = SqliteWorkspaceStore::migration_plan(&db_path).unwrap();
assert!(plan.repairs.iter().any(|repair| {
repair == "clear 2 tombstoned legacy current Ticket assignment pointer(s)"
}));
let migrated = SqliteWorkspaceStore::open(&db_path).unwrap();
migrated
.with_conn(|conn| {
assert_eq!(current_schema_version(conn)?, 43);
assert_eq!(
conn.query_row("PRAGMA foreign_keys", [], |row| row.get::<_, i64>(0))?,
1,
"migration must restore foreign key enforcement"
);
assert_eq!(
conn.query_row(
"SELECT COUNT(*) FROM ticket_current_worker_assignments",
[],
|row| row.get::<_, i64>(0),
)?,
0
);
assert_eq!(
conn.query_row(
"SELECT COUNT(*) FROM ticket_worker_assignments WHERE role = 'coder'",
[],
|row| row.get::<_, i64>(0),
)?,
2,
"migration must retain assignment history"
);
assert_eq!(
conn.query_row(
"SELECT COUNT(*) FROM ticket_worker_assignment_events WHERE role = 'coder'",
[],
|row| row.get::<_, i64>(0),
)?,
2,
"migration must retain assignment events"
);
assert_eq!(
conn.query_row(
"SELECT COUNT(*) FROM ticket_assignment_worker_tombstones",
[],
|row| row.get::<_, i64>(0),
)?,
1
);
assert_eq!(
conn.query_row(
"SELECT COUNT(*) FROM ticket_assignment_ticket_tombstones",
[],
|row| row.get::<_, i64>(0),
)?,
1
);
assert_eq!(
conn.query_row(
"SELECT COUNT(*) FROM legacy_assignment_consumer",
[],
|row| row.get::<_, i64>(0),
)?,
1,
"migration must preserve external assignment history references"
);
let foreign_key_failures: i64 =
conn.query_row("SELECT COUNT(*) FROM pragma_foreign_key_check", [], |row| {
row.get(0)
})?;
assert_eq!(foreign_key_failures, 0);
Ok(())
})
.unwrap();
}
#[tokio::test]
async fn role_assignment_routes_and_manual_start_are_state_fenced_atomically() {
let dir = tempfile::tempdir().unwrap();