diff --git a/lib/apps/fabro-server/src/server/handler/events.rs b/lib/apps/fabro-server/src/server/handler/events.rs index c032d57d8..8dfd3cd70 100644 --- a/lib/apps/fabro-server/src/server/handler/events.rs +++ b/lib/apps/fabro-server/src/server/handler/events.rs @@ -218,9 +218,9 @@ async fn append_run_event( if event.run_id != id { return ApiError::bad_request("Event run_id does not match path run ID.").into_response(); } - if let Some(denied) = denied_lifecycle_event_name(&event.body) { + if let Some(denied) = denied_dedicated_operation_event_name(&event.body) { return ApiError::bad_request(format!( - "{denied} is a lifecycle event; clients must call the corresponding operation endpoint instead of injecting it via append_run_event" + "{denied} must be performed through its dedicated operation endpoint instead of injecting it via append_run_event" )) .into_response(); } @@ -576,7 +576,7 @@ async fn attach_run_events( /// (e.g. "archive only from terminal") that a direct event append would /// bypass. Other run-lifecycle events flow through this endpoint legitimately: /// the worker subprocess emits state transitions during execution. -fn denied_lifecycle_event_name(body: &EventBody) -> Option<&str> { +fn denied_dedicated_operation_event_name(body: &EventBody) -> Option<&str> { match body { EventBody::RunArchived(_) | EventBody::RunUnarchived(_) @@ -585,7 +585,8 @@ fn denied_lifecycle_event_name(body: &EventBody) -> Option<&str> { | EventBody::RunPauseRequested(_) | EventBody::RunUnpauseRequested(_) | EventBody::PullRequestLinked(_) - | EventBody::PullRequestUnlinked(_) => Some(body.event_name()), + | EventBody::PullRequestUnlinked(_) + | EventBody::RunSessionCreated(_) => Some(body.event_name()), _ => None, } } diff --git a/lib/apps/fabro-server/src/server/handler/sessions.rs b/lib/apps/fabro-server/src/server/handler/sessions.rs index f5f1877b3..871b94114 100644 --- a/lib/apps/fabro-server/src/server/handler/sessions.rs +++ b/lib/apps/fabro-server/src/server/handler/sessions.rs @@ -184,14 +184,6 @@ async fn create_run_session( let session_id = SessionId::new(); let now = Utc::now(); - if let Err(err) = state - .store_ref() - .put_session_run_index(&session_id, &run_id) - .await - { - return store_error(&err).into_response(); - } - let event = match append_run_session_event( &run_store, run_id, @@ -1388,7 +1380,7 @@ async fn load_session( state: &AppState, session_id: SessionId, ) -> Result<(RunId, RunDatabase, ProjectedRunSession), Response> { - let run_id = match state.store_ref().get_session_run_id(&session_id).await { + let run_id = match state.store_ref().find_session_owner(&session_id).await { Ok(Some(run_id)) => run_id, Ok(None) => return Err(ApiError::not_found("Session not found.").into_response()), Err(err) => return Err(store_error(&err).into_response()), @@ -1408,7 +1400,7 @@ async fn load_session_read( state: &AppState, session_id: SessionId, ) -> Result<(RunId, ProjectedRunSession), Response> { - let run_id = match state.store_ref().get_session_run_id(&session_id).await { + let run_id = match state.store_ref().find_session_owner(&session_id).await { Ok(Some(run_id)) => run_id, Ok(None) => return Err(ApiError::not_found("Session not found.").into_response()), Err(err) => return Err(store_error(&err).into_response()), @@ -1428,7 +1420,7 @@ async fn load_session_run_reader( state: &AppState, session_id: SessionId, ) -> Result<(RunId, RunDatabase), Response> { - let run_id = match state.store_ref().get_session_run_id(&session_id).await { + let run_id = match state.store_ref().find_session_owner(&session_id).await { Ok(Some(run_id)) => run_id, Ok(None) => return Err(ApiError::not_found("Session not found.").into_response()), Err(err) => return Err(store_error(&err).into_response()), diff --git a/lib/apps/fabro-server/src/server/tests.rs b/lib/apps/fabro-server/src/server/tests.rs index 3ab74ee64..409800f22 100644 --- a/lib/apps/fabro-server/src/server/tests.rs +++ b/lib/apps/fabro-server/src/server/tests.rs @@ -12188,10 +12188,11 @@ async fn append_run_event_rejects_reserved_archive_event() { let response = app.oneshot(req).await.unwrap(); let body = response_json!(response, StatusCode::BAD_REQUEST).await; assert!( - body["errors"][0]["detail"] - .as_str() - .is_some_and(|message| message.contains("run.archived is a lifecycle event")), - "expected lifecycle rejection, got: {body}" + body["errors"][0]["detail"].as_str().is_some_and(|message| { + message + .contains("run.archived must be performed through its dedicated operation endpoint") + }), + "expected dedicated-operation rejection, got: {body}" ); } diff --git a/lib/apps/fabro-server/tests/it/api/sessions.rs b/lib/apps/fabro-server/tests/it/api/sessions.rs index c989aa4c3..9508e07fc 100644 --- a/lib/apps/fabro-server/tests/it/api/sessions.rs +++ b/lib/apps/fabro-server/tests/it/api/sessions.rs @@ -133,6 +133,65 @@ async fn run_bound_session_is_created_as_run_event_and_resolves_by_flat_id() { ); } +#[tokio::test] +async fn generic_session_creation_requires_the_dedicated_operation_without_advancing_history() { + let app = fabro_server::test_support::build_test_router(test_app_state()); + let run_id = create_run(&app).await; + let before_request = Request::builder() + .method("GET") + .uri(api(&format!("/runs/{run_id}/events"))) + .body(Body::empty()) + .expect("run-events request should build"); + let before = response_json( + app.clone().oneshot(before_request).await.unwrap(), + StatusCode::OK, + format!("GET /api/v1/runs/{run_id}/events before rejected append"), + ) + .await; + + let session_id = fabro_types::SessionId::new(); + let request = Request::builder() + .method("POST") + .uri(api(&format!("/runs/{run_id}/events"))) + .header("content-type", "application/json") + .body(Body::from( + serde_json::to_string(&serde_json::json!({ + "id": ulid::Ulid::new().to_string(), + "ts": "2026-08-31T12:00:00Z", + "run_id": run_id, + "event": "run.session.created", + "session_id": session_id, + "properties": { "title": "Injected session" }, + })) + .expect("session creation event should serialize"), + )) + .expect("append-event request should build"); + let rejected = response_json( + app.clone().oneshot(request).await.unwrap(), + StatusCode::BAD_REQUEST, + format!("POST /api/v1/runs/{run_id}/events with session creation"), + ) + .await; + let detail = rejected["errors"][0]["detail"] + .as_str() + .expect("error response should include detail"); + assert!(detail.contains("dedicated operation endpoint")); + assert!(detail.contains("run.session.created")); + + let after_request = Request::builder() + .method("GET") + .uri(api(&format!("/runs/{run_id}/events"))) + .body(Body::empty()) + .expect("run-events request should build"); + let after = response_json( + app.clone().oneshot(after_request).await.unwrap(), + StatusCode::OK, + format!("GET /api/v1/runs/{run_id}/events after rejected append"), + ) + .await; + assert_eq!(after, before); +} + #[tokio::test] async fn sessions_are_listed_only_under_their_owning_run() { let app = fabro_server::test_support::build_test_router(test_app_state()); diff --git a/lib/apps/fabro-server/tests/it/scenario/archive.rs b/lib/apps/fabro-server/tests/it/scenario/archive.rs index fb342650e..1f8f84d37 100644 --- a/lib/apps/fabro-server/tests/it/scenario/archive.rs +++ b/lib/apps/fabro-server/tests/it/scenario/archive.rs @@ -183,8 +183,8 @@ async fn appending_run_archived_event_directly_is_rejected() { .await; let detail = body["errors"][0]["detail"].as_str().unwrap_or_default(); assert!( - detail.contains("lifecycle event"), - "expected lifecycle rejection, got: {body}" + detail.contains("run.archived must be performed through its dedicated operation endpoint"), + "expected dedicated-operation rejection, got: {body}" ); } diff --git a/lib/components/fabro-store/src/keys.rs b/lib/components/fabro-store/src/keys.rs index 1f5d93a76..6e0feb7c1 100644 --- a/lib/components/fabro-store/src/keys.rs +++ b/lib/components/fabro-store/src/keys.rs @@ -2,7 +2,7 @@ use std::fmt::{self, Write}; #[cfg(test)] use std::ops::Range; -use fabro_types::{RunId, SessionId}; +use fabro_types::RunId; pub(crate) const MAX_EVENT_SEQ: u32 = 999_999; @@ -120,7 +120,8 @@ pub(crate) fn sessions_by_id_prefix() -> SlateKey { SlateKey::new("sessions").with("by-id").into_prefix() } -pub(crate) fn session_by_id_key(session_id: &SessionId) -> SlateKey { +#[cfg(test)] +pub(crate) fn session_by_id_key(session_id: &fabro_types::SessionId) -> SlateKey { SlateKey::new("sessions").with("by-id").with(session_id) } diff --git a/lib/components/fabro-store/src/legacy_run_history_import.rs b/lib/components/fabro-store/src/legacy_run_history_import.rs index 0851bb1e2..dfad82b55 100644 --- a/lib/components/fabro-store/src/legacy_run_history_import.rs +++ b/lib/components/fabro-store/src/legacy_run_history_import.rs @@ -1396,6 +1396,24 @@ mod tests { event_value(run_id, seq, "run.submitted", &serde_json::json!({})) } + fn session_created_value( + run_id: &RunId, + seq: u32, + session_id: &SessionId, + ) -> serde_json::Value { + let mut value = event_value( + run_id, + seq, + "run.session.created", + &serde_json::json!({ "title": "Imported session" }), + ); + value + .as_object_mut() + .unwrap() + .insert("session_id".to_string(), session_id.to_string().into()); + value + } + #[tokio::test] async fn legacy_source_identity_covers_exact_keys_and_json_bytes() -> TestResult<()> { let context = TestContext::new().await?; @@ -1534,17 +1552,20 @@ mod tests { } #[tokio::test] - async fn legacy_run_history_imports_exact_json_gaps_and_count_only_diagnostics() + async fn session_owner_imports_typed_event_and_counts_opaque_legacy_reverse_rows() -> TestResult<()> { let context = TestContext::new().await?; let first = run_id(1); let second = run_id(2); let empty_marker = run_id(3); let stale = run_id(4); + let session_id = SessionId::new(); let first_created = serde_json::to_string_pretty(&created_value(&first, "first"))?; + let first_session = serde_json::to_string(&session_created_value(&first, 3, &session_id))?; let first_submitted = serde_json::to_string(&submitted_value(&first, 4))?; let second_created = serde_json::to_string(&created_value(&second, "second"))?; context.put_event(&first, 1, 10, &first_created).await?; + context.put_event(&first, 3, 30, &first_session).await?; context.put_event(&first, 4, 40, &first_submitted).await?; context.put_event(&second, 1, 20, &second_created).await?; context.put_raw(keys::run_catalog_key(&first), b"").await?; @@ -1552,8 +1573,10 @@ mod tests { .put_raw(keys::run_catalog_key(&empty_marker), b"") .await?; context - .source - .put_session_run_index(&SessionId::new(), &first) + .put_raw( + keys::session_by_id_key(&session_id), + b"opaque legacy reverse row", + ) .await?; let stale_created = serde_json::to_string(&created_value(&stale, "stale"))?; @@ -1572,9 +1595,9 @@ mod tests { assert_eq!(report, LegacyRunHistoryImportReport { scanned_source_runs: 2, - scanned_source_events: 3, + scanned_source_events: 4, imported_runs: 2, - imported_events: 3, + imported_events: 4, discarded_projection_only_rows: 1, committed_run_transactions: 2, diagnostics: LegacyRunHistoryDiagnostics { @@ -1589,7 +1612,16 @@ mod tests { .bind(first.to_string()) .fetch_all(&context.sqlite) .await?; - assert_eq!(stored, vec![(1, first_created), (4, first_submitted)]); + assert_eq!(stored, vec![ + (1, first_created), + (3, first_session), + (4, first_submitted) + ]); + let summaries = RunSummaryStore::new(context.sqlite.clone()); + assert_eq!( + summaries.find_session_owner(&session_id).await?, + Some(first) + ); assert_eq!( sqlx::query_scalar::<_, i64>("SELECT source_last_seq FROM runs WHERE id = ?") .bind(first.to_string()) @@ -1603,7 +1635,7 @@ mod tests { .verify_legacy_run_history_in(&context.sqlite) .await?; assert_eq!(verification.target_runs, 2); - assert_eq!(verification.target_events, 3); + assert_eq!(verification.target_events, 4); Ok(()) } diff --git a/lib/components/fabro-store/src/run_summary_store.rs b/lib/components/fabro-store/src/run_summary_store.rs index d180213d3..b9008a384 100644 --- a/lib/components/fabro-store/src/run_summary_store.rs +++ b/lib/components/fabro-store/src/run_summary_store.rs @@ -435,6 +435,32 @@ ON CONFLICT(singleton) DO NOTHING .await } + pub(crate) async fn find_session_owner(&self, session_id: &SessionId) -> Result> { + let mut query = QueryBuilder::::new(SELECT_EVENT_COLUMNS); + query + .push(" WHERE session_id = ") + .push_bind(session_id.to_string()) + .push(" AND event_name = 'run.session.created'"); + let row = query.build().fetch_optional(&self.pool).await?; + let Some(row) = row else { + return Ok(None); + }; + + let stored_run_id: String = row.try_get("run_id")?; + let run_id = stored_run_id + .parse::() + .map_err(|_| Error::RunEventMismatch { + run_id: stored_run_id.clone(), + seq: 0, + field: "run_id", + })?; + // The WHERE clause pins the row's session_id and event_name columns + // to the requested values, and decoding verifies the envelope against + // every stored column, so a successful decode proves ownership. + decode_event_row(&row, &run_id, &stored_run_id)?; + Ok(Some(run_id)) + } + pub(crate) async fn delete_canonical(&self, run_id: &RunId, deleted_at_ms: i64) -> Result<()> { // This transaction reads the activation marker before it writes the // tombstone and run deletion. A deferred SQLite transaction can fail @@ -1536,6 +1562,17 @@ mod tests { ) } + fn session_created_payload(run_id: &RunId, session_id: &SessionId) -> EventPayload { + sql_event_payload( + run_id, + "run.session.created", + None, + None, + Some(session_id), + serde_json::json!({ "title": "Owned session" }), + ) + } + async fn seed_sql_event( store: &RunSummaryStore, run_id: &RunId, @@ -1807,6 +1844,122 @@ mod tests { assert_eq!(sequences, vec![1, 2, 3]); } + #[tokio::test] + async fn session_owner_lookup_resolves_only_typed_creation_events() { + let (_directory, store) = store().await; + let created_at = dt("2026-08-27T12:00:00Z"); + let id = run_id(created_at.timestamp_millis().cast_unsigned(), 31); + store + .upsert_projection(&entry(projection(id, "owner", created_at), 3)) + .await + .unwrap(); + + let owner_session = SessionId::new(); + seed_sql_event( + &store, + &id, + 2, + &session_created_payload(&id, &owner_session), + ) + .await; + assert_eq!( + store.find_session_owner(&owner_session).await.unwrap(), + Some(id) + ); + + let non_owner_session = SessionId::new(); + let non_creation = sql_event_payload( + &id, + "run.session.future", + None, + None, + Some(&non_owner_session), + serde_json::json!({ "kind": "future" }), + ); + seed_sql_event(&store, &id, 3, &non_creation).await; + assert_eq!( + store.find_session_owner(&non_owner_session).await.unwrap(), + None + ); + assert_eq!( + store.find_session_owner(&SessionId::new()).await.unwrap(), + None + ); + } + + #[tokio::test] + async fn session_owner_lookup_rejects_corrupt_selected_events() { + let (_directory, store) = store().await; + let created_at = dt("2026-08-27T12:00:00Z"); + let id = run_id(created_at.timestamp_millis().cast_unsigned(), 32); + let other_id = run_id(created_at.timestamp_millis().cast_unsigned() + 1, 33); + store + .upsert_projection(&entry(projection(id, "owner", created_at), 2)) + .await + .unwrap(); + let session_id = SessionId::new(); + let payload = session_created_payload(&id, &session_id); + seed_sql_event(&store, &id, 2, &payload).await; + + let mut wrong_event_name = payload.as_value().clone(); + wrong_event_name["event"] = "run.session.future".into(); + sqlx::query("UPDATE run_events SET event_json = ? WHERE run_id = ? AND seq = 2") + .bind(serde_json::to_string(&wrong_event_name).unwrap()) + .bind(id.to_string()) + .execute(&store.pool) + .await + .unwrap(); + assert!(matches!( + store.find_session_owner(&session_id).await.unwrap_err(), + Error::RunEventMismatch { + field: "event_name", + .. + } + )); + + seed_sql_event_restore(&store, &id, 2, &payload).await; + let mut wrong_run = payload.as_value().clone(); + wrong_run["run_id"] = other_id.to_string().into(); + sqlx::query("UPDATE run_events SET event_json = ? WHERE run_id = ? AND seq = 2") + .bind(serde_json::to_string(&wrong_run).unwrap()) + .bind(id.to_string()) + .execute(&store.pool) + .await + .unwrap(); + assert!(matches!( + store.find_session_owner(&session_id).await.unwrap_err(), + Error::RunEventMismatch { + field: "run_id", + .. + } + )); + + seed_sql_event_restore(&store, &id, 2, &payload).await; + let mut wrong_session = payload.as_value().clone(); + wrong_session["session_id"] = SessionId::new().to_string().into(); + sqlx::query("UPDATE run_events SET event_json = ? WHERE run_id = ? AND seq = 2") + .bind(serde_json::to_string(&wrong_session).unwrap()) + .bind(id.to_string()) + .execute(&store.pool) + .await + .unwrap(); + assert!(matches!( + store.find_session_owner(&session_id).await.unwrap_err(), + Error::RunEventMismatch { + field: "session_id", + .. + } + )); + + seed_sql_event_restore(&store, &id, 2, &payload).await; + sqlx::query("UPDATE run_events SET event_json = '{}' WHERE run_id = ? AND seq = 2") + .bind(id.to_string()) + .execute(&store.pool) + .await + .unwrap(); + assert!(store.find_session_owner(&session_id).await.is_err()); + } + #[tokio::test] async fn canonical_delete_waits_for_a_concurrent_writer() { let (_directory, store) = store().await; diff --git a/lib/components/fabro-store/src/slate/mod.rs b/lib/components/fabro-store/src/slate/mod.rs index 360e06562..9e92eb894 100644 --- a/lib/components/fabro-store/src/slate/mod.rs +++ b/lib/components/fabro-store/src/slate/mod.rs @@ -28,11 +28,6 @@ pub struct UnreadableRun { pub error: String, } -#[derive(Debug, Clone, serde::Deserialize, serde::Serialize)] -struct SessionRunIndexEntry { - run_id: RunId, -} - #[derive(Clone)] pub struct Database { object_store: Arc, @@ -374,27 +369,10 @@ impl Database { Ok(self.projection_cache.pending_pull_request_creations()) } - pub async fn put_session_run_index( - &self, - session_id: &SessionId, - run_id: &RunId, - ) -> Result<()> { - let db = self.open_db().await?; - db.put( - keys::session_by_id_key(session_id), - serde_json::to_vec(&SessionRunIndexEntry { run_id: *run_id })?, - ) - .await?; - Ok(()) - } - - pub async fn get_session_run_id(&self, session_id: &SessionId) -> Result> { - let db = self.open_db().await?; - if let Some(bytes) = db.get(keys::session_by_id_key(session_id)).await? { - let entry: SessionRunIndexEntry = serde_json::from_slice(&bytes)?; - return Ok(Some(entry.run_id)); - } - Ok(None) + /// Resolves the run that owns `session_id` from the canonical typed + /// creation event stored in SQLite. + pub async fn find_session_owner(&self, session_id: &SessionId) -> Result> { + self.run_summary_store.find_session_owner(session_id).await } pub(crate) fn remove_cached_run(&self, run_id: &RunId) { @@ -413,31 +391,6 @@ impl Database { .await?; active_runs.remove(run_id); self.remove_cached_run(run_id); - if let Err(err) = self.delete_session_indexes_for_run(run_id).await { - warn!( - run_id = %run_id, - error = %err, - "Failed to remove retired session reverse indexes after deleting run" - ); - } - Ok(()) - } - - async fn delete_session_indexes_for_run(&self, run_id: &RunId) -> Result<()> { - let db = self.open_db().await?; - let mut keys_to_delete = Vec::new(); - let mut iter = db.scan_prefix(keys::sessions_by_id_prefix()).await?; - while let Some(entry) = iter.next().await? { - let index: SessionRunIndexEntry = serde_json::from_slice(&entry.value)?; - if index.run_id == *run_id { - keys_to_delete.push(String::from_utf8(entry.key.to_vec()).map_err(|err| { - Error::Other(format!("stored key is not valid UTF-8: {err}")) - })?); - } - } - for key in keys_to_delete { - db.delete(key).await?; - } Ok(()) } @@ -695,6 +648,21 @@ mod tests { .unwrap() } + fn session_created_payload(label: &str, session_id: &SessionId) -> EventPayload { + EventPayload::new( + serde_json::json!({ + "id": format!("evt-{label}-session-created"), + "ts": "2026-03-27T12:00:05Z", + "run_id": test_run_id(label).to_string(), + "event": "run.session.created", + "session_id": session_id, + "properties": { "title": "Owned session" }, + }), + &test_run_id(label), + ) + .unwrap() + } + async fn append_created(run: &RunDatabase, label: &str, created_at: DateTime) { let run_spec = sample_run_spec(label); run.append_event(&event_payload( @@ -850,7 +818,7 @@ mod tests { } #[tokio::test] - async fn create_open_list_and_delete_full_lifecycle_in_shared_db() { + async fn create_open_list_and_delete_full_lifecycle_without_legacy_slate_writes() { let (object_store, store) = make_store(); let run_1 = store.create_run(&test_run_id("run-1")).await.unwrap(); let run_2 = store.create_run(&test_run_id("run-2")).await.unwrap(); @@ -883,7 +851,100 @@ mod tests { .unwrap(); assert_eq!(remaining.len(), 1); assert_eq!(remaining[0].id, test_run_id("run-2")); - assert!(!list_paths(object_store, "runs/").await.is_empty()); + assert!( + list_paths(object_store, "runs/").await.is_empty(), + "canonical run lifecycle must not open SlateDB solely for retired session indexes" + ); + } + + #[tokio::test] + async fn session_owner_claims_are_atomic_durable_and_ignore_legacy_reverse_rows() { + let directory = tempfile::tempdir().unwrap(); + let object_store: Arc = Arc::new(InMemory::new()); + let store = store_test_support::test_database_at( + Arc::clone(&object_store), + "session-owner", + Duration::from_millis(1), + None, + directory.path(), + ); + let first_id = test_run_id("run-1"); + let second_id = test_run_id("run-2"); + let first = store.create_run(&first_id).await.unwrap(); + let second = store.create_run(&second_id).await.unwrap(); + append_created(&first, "run-1", dt("2026-03-27T12:00:00Z")).await; + append_created(&second, "run-2", dt("2026-03-27T12:00:10Z")).await; + + let session_id = SessionId::new(); + assert_eq!( + first + .append_event(&session_created_payload("run-1", &session_id)) + .await + .unwrap(), + 2 + ); + assert_eq!( + store.find_session_owner(&session_id).await.unwrap(), + Some(first_id) + ); + + let legacy_key = keys::session_by_id_key(&session_id).as_ref().to_vec(); + let legacy = store.open_db().await.unwrap(); + assert!(legacy.get(&legacy_key).await.unwrap().is_none()); + legacy + .put( + &legacy_key, + serde_json::to_vec(&serde_json::json!({ "run_id": second_id })).unwrap(), + ) + .await + .unwrap(); + legacy.flush().await.unwrap(); + assert_eq!( + store.find_session_owner(&session_id).await.unwrap(), + Some(first_id), + "legacy reverse rows must not influence ownership" + ); + + assert!( + first + .append_event(&session_created_payload("run-1", &session_id)) + .await + .is_err() + ); + assert_eq!(first.last_event_seq().await.unwrap(), Some(2)); + assert!( + second + .append_event(&session_created_payload("run-2", &session_id)) + .await + .is_err() + ); + assert_eq!(second.last_event_seq().await.unwrap(), Some(1)); + assert_eq!( + store.find_session_owner(&session_id).await.unwrap(), + Some(first_id) + ); + + let reopened = store_test_support::test_database_at( + object_store, + "session-owner", + Duration::from_millis(1), + None, + directory.path(), + ); + assert_eq!( + reopened.find_session_owner(&session_id).await.unwrap(), + Some(first_id) + ); + + reopened.delete_run(&first_id).await.unwrap(); + assert_eq!( + reopened.find_session_owner(&session_id).await.unwrap(), + None + ); + assert!( + legacy.get(&legacy_key).await.unwrap().is_some(), + "legacy reverse rows remain diagnostic-only during the support window" + ); } #[tokio::test] diff --git a/lib/components/fabro-store/src/test_support/mod.rs b/lib/components/fabro-store/src/test_support/mod.rs index b651ca096..2ab338357 100644 --- a/lib/components/fabro-store/src/test_support/mod.rs +++ b/lib/components/fabro-store/src/test_support/mod.rs @@ -33,6 +33,7 @@ pub fn test_run_summary_store() -> Arc { fabro_db::RUNS_MIGRATION_SQL, fabro_db::RUN_EVENTS_MIGRATION_SQL, fabro_db::RUN_HISTORY_ACTIVATION_MIGRATION_SQL, + fabro_db::RUN_EVENT_SESSION_OWNER_MIGRATION_SQL, ]))) } @@ -106,6 +107,7 @@ pub fn test_run_summary_store_at(store_dir: &Path) -> Arc { fabro_db::RUNS_MIGRATION_SQL, fabro_db::RUN_EVENTS_MIGRATION_SQL, fabro_db::RUN_HISTORY_ACTIVATION_MIGRATION_SQL, + fabro_db::RUN_EVENT_SESSION_OWNER_MIGRATION_SQL, ], ))) } diff --git a/lib/foundation/fabro-db/migrations/2026083101_run_event_session_owner.sql b/lib/foundation/fabro-db/migrations/2026083101_run_event_session_owner.sql new file mode 100644 index 000000000..b15b237d3 --- /dev/null +++ b/lib/foundation/fabro-db/migrations/2026083101_run_event_session_owner.sql @@ -0,0 +1,4 @@ +CREATE UNIQUE INDEX run_events_by_session_owner +ON run_events(session_id) +WHERE session_id IS NOT NULL + AND event_name = 'run.session.created'; diff --git a/lib/foundation/fabro-db/src/lib.rs b/lib/foundation/fabro-db/src/lib.rs index 7d55de2ac..a921ec7eb 100644 --- a/lib/foundation/fabro-db/src/lib.rs +++ b/lib/foundation/fabro-db/src/lib.rs @@ -16,6 +16,8 @@ pub type DbPool = sqlx::SqlitePool; static MIGRATOR: Migrator = sqlx::migrate!("./migrations"); +const SESSION_OWNER_INDEX_MIGRATION_VERSION: i64 = 2_026_083_101; + /// The blob-table migration, exposed so fixtures in other crates can install /// the production blob schema without a filesystem path into this crate. pub const BLOBS_MIGRATION_SQL: &str = include_str!("../migrations/2026081301_blobs.sql"); @@ -28,6 +30,11 @@ pub const RUNS_MIGRATION_SQL: &str = include_str!("../migrations/2026071104_runs /// the production schema without a filesystem path into this crate. pub const RUN_EVENTS_MIGRATION_SQL: &str = include_str!("../migrations/2026082701_run_events.sql"); +/// The run-session owner index migration, exposed so fixtures in other crates +/// can install the production run-history indexes. +pub const RUN_EVENT_SESSION_OWNER_MIGRATION_SQL: &str = + include_str!("../migrations/2026083101_run_event_session_owner.sql"); + /// The temporary run-history activation migration, exposed so fixtures in /// other crates can install the production compatibility schema. pub const RUN_HISTORY_ACTIVATION_MIGRATION_SQL: &str = @@ -66,7 +73,11 @@ impl Database { } pub async fn migrate(&self) -> anyhow::Result<()> { - self.snapshot_before_new_migrations() + let applied = applied_migration_versions(&self.pool).await?; + self.preflight_session_owner_index(&applied) + .await + .context("checking session ownership before SQLite migrations")?; + self.snapshot_before_new_migrations(&applied) .await .context("snapshotting SQLite database before migrations")?; MIGRATOR @@ -75,6 +86,53 @@ impl Database { .context("running SQLite migrations") } + /// Refuse the unique owner index when old event history contains + /// collisions. The diagnostic is deliberately count-only because session + /// identifiers and event contents are not safe startup-log fields. + /// + /// Temporary compatibility guard: once every supported database has + /// applied the session-owner index migration the version check below + /// always short-circuits, and this preflight can be deleted along with + /// the run-history compatibility window. + async fn preflight_session_owner_index(&self, applied: &HashSet) -> anyhow::Result<()> { + if applied.contains(&SESSION_OWNER_INDEX_MIGRATION_VERSION) { + return Ok(()); + } + + let run_events_exists: bool = sqlx::query_scalar( + "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'run_events')", + ) + .fetch_one(&self.pool) + .await + .context("checking for the run event table")?; + if !run_events_exists { + return Ok(()); + } + + let collision_groups: i64 = sqlx::query_scalar( + r" +SELECT COUNT(*) +FROM ( + SELECT session_id + FROM run_events + WHERE session_id IS NOT NULL + AND event_name = 'run.session.created' + GROUP BY session_id + HAVING COUNT(*) > 1 +) +", + ) + .fetch_one(&self.pool) + .await + .context("counting duplicate session ownership groups")?; + if collision_groups > 0 { + anyhow::bail!( + "cannot create the unique session owner index: found {collision_groups} duplicate session ownership groups" + ); + } + Ok(()) + } + /// Copy the database aside before applying migrations it has not seen. /// /// A binary downgrade after new migrations have been applied fails sqlx's @@ -92,8 +150,7 @@ impl Database { /// from immediately before the most recent schema change. Failing to /// write the snapshot fails the migration: no rollback artifact, no /// schema change. - async fn snapshot_before_new_migrations(&self) -> anyhow::Result<()> { - let applied = applied_migration_versions(&self.pool).await?; + async fn snapshot_before_new_migrations(&self, applied: &HashSet) -> anyhow::Result<()> { let has_pending = MIGRATOR .iter() .any(|migration| !applied.contains(&migration.version)); diff --git a/lib/foundation/fabro-db/tests/sqlite.rs b/lib/foundation/fabro-db/tests/sqlite.rs index 33f0a697d..915343642 100644 --- a/lib/foundation/fabro-db/tests/sqlite.rs +++ b/lib/foundation/fabro-db/tests/sqlite.rs @@ -785,7 +785,7 @@ async fn runs_schema_creates_indexes_and_rejects_invalid_rows() -> anyhow::Resul } #[tokio::test] -async fn run_events_schema_has_final_shape_constraints_and_indexes() -> anyhow::Result<()> { +async fn session_owner_schema_has_final_shape_constraints_and_indexes() -> anyhow::Result<()> { let dir = tempfile::tempdir()?; let database = fabro_db::Database::connect(dir.path().join("fabro.sqlite3")).await?; database.migrate().await?; @@ -847,6 +847,7 @@ async fn run_events_schema_has_final_shape_constraints_and_indexes() -> anyhow:: }) .collect::>(); assert_eq!(named_indexes, vec![ + ("run_events_by_session_owner".to_string(), 1, 1), ( "run_events_by_pull_request_creation_request".to_string(), 0, @@ -858,6 +859,7 @@ async fn run_events_schema_has_final_shape_constraints_and_indexes() -> anyhow:: ]); assert!(indexes.iter().all(|index| { index.get::("unique") == 0 + || index.get::("name") == "run_events_by_session_owner" || index.get::("name") == "sqlite_autoindex_run_events_1" })); @@ -912,7 +914,8 @@ async fn run_events_schema_has_final_shape_constraints_and_indexes() -> anyhow:: } #[tokio::test] -async fn run_events_schema_query_plans_use_candidate_indexes() -> anyhow::Result<()> { +async fn run_events_schema_query_plans_use_candidate_indexes_including_session_owner() +-> anyhow::Result<()> { let dir = tempfile::tempdir()?; let database = fabro_db::Database::connect(dir.path().join("fabro.sqlite3")).await?; database.migrate().await?; @@ -938,6 +941,10 @@ async fn run_events_schema_query_plans_use_candidate_indexes() -> anyhow::Result "EXPLAIN QUERY PLAN SELECT * FROM run_events WHERE run_id = ? AND session_id = ? AND event_name GLOB 'run.session.*' ORDER BY seq ASC LIMIT ?", "run_events_by_session", ), + ( + "EXPLAIN QUERY PLAN SELECT run_id, seq, event_name, node_id, stage_id, session_id, event_json FROM run_events WHERE session_id = ? AND event_name = 'run.session.created'", + "run_events_by_session_owner", + ), ( "EXPLAIN QUERY PLAN SELECT * FROM run_events WHERE event_name = 'pull_request.creation_requested' ORDER BY run_id, seq", "run_events_by_pull_request_creation_request", @@ -989,6 +996,99 @@ async fn run_events_schema_query_plans_use_candidate_indexes() -> anyhow::Result Ok(()) } +#[tokio::test] +async fn session_owner_migration_preflight_is_count_only_retriable_and_idempotent() +-> anyhow::Result<()> { + let dir = tempfile::tempdir()?; + let database = fabro_db::Database::connect(dir.path().join("fabro.sqlite3")).await?; + database.migrate().await?; + + sqlx::query("DROP INDEX IF EXISTS run_events_by_session_owner") + .execute(database.pool()) + .await?; + sqlx::query("DELETE FROM _sqlx_migrations WHERE version = 2026083101") + .execute(database.pool()) + .await?; + + for run_id in ["first", "second", "third", "fourth"] { + insert_run_with_id(database.pool(), run_id, None).await?; + } + for (run_id, session_id) in [ + ("first", "collision-alpha"), + ("second", "collision-alpha"), + ("third", "collision-beta"), + ("fourth", "collision-beta"), + ] { + insert_session_creation_claim(database.pool(), run_id, session_id).await?; + } + + let error = database + .migrate() + .await + .expect_err("duplicate session owners must abort migration"); + let rendered = format!("{error:#}"); + assert!(rendered.contains("2 duplicate session ownership groups")); + assert!(!rendered.contains("collision-alpha")); + assert!(!rendered.contains("collision-beta")); + assert!(!rendered.contains("sensitive event contents")); + assert_eq!( + sqlx::query_scalar::<_, i64>( + "SELECT COUNT(*) FROM run_events WHERE event_name = 'run.session.created'" + ) + .fetch_one(database.pool()) + .await?, + 4 + ); + assert_eq!( + sqlx::query_scalar::<_, i64>( + "SELECT COUNT(*) FROM sqlite_master WHERE type = 'index' AND name = 'run_events_by_session_owner'" + ) + .fetch_one(database.pool()) + .await?, + 0 + ); + + sqlx::query("DELETE FROM run_events WHERE run_id IN ('second', 'fourth')") + .execute(database.pool()) + .await?; + database.migrate().await?; + database.migrate().await?; + assert_eq!( + sqlx::query_scalar::<_, i64>( + "SELECT COUNT(*) FROM sqlite_master WHERE type = 'index' AND name = 'run_events_by_session_owner'" + ) + .fetch_one(database.pool()) + .await?, + 1 + ); + Ok(()) +} + +async fn insert_session_creation_claim( + pool: &fabro_db::DbPool, + run_id: &str, + session_id: &str, +) -> Result<(), sqlx::Error> { + sqlx::query( + r" +INSERT INTO run_events (run_id, seq, event_name, session_id, event_json) +VALUES (?, 1, 'run.session.created', ?, json_object( + 'run_id', ?, + 'event', 'run.session.created', + 'session_id', ?, + 'properties', json_object('note', 'sensitive event contents') +)) +", + ) + .bind(run_id) + .bind(session_id) + .bind(run_id) + .bind(session_id) + .execute(pool) + .await?; + Ok(()) +} + async fn insert_run_event( pool: &fabro_db::DbPool, run_id: &str,