From 70d153b36533f029baafe0e3e9a830232e4f8b69 Mon Sep 17 00:00:00 2001 From: Hare Date: Tue, 20 Oct 2026 17:39:38 +0900 Subject: [PATCH] fix: isolate unreadable archives from runtime startup --- .../src/legacy_session_log/tests/v5.rs | 8 +- .../session-store/src/worker_session_store.rs | 27 +++ crates/worker-runtime/src/retention.rs | 140 +++++++++---- .../src/retention/runtime_v5_tests.rs | 198 +++++++++++++++--- crates/workspace-server/src/server.rs | 52 +++++ .../workspace-schema-migrations.md | 4 +- 6 files changed, 357 insertions(+), 72 deletions(-) diff --git a/crates/session-store/src/legacy_session_log/tests/v5.rs b/crates/session-store/src/legacy_session_log/tests/v5.rs index 847b81bf0..fac87798b 100644 --- a/crates/session-store/src/legacy_session_log/tests/v5.rs +++ b/crates/session-store/src/legacy_session_log/tests/v5.rs @@ -211,7 +211,7 @@ fn v1_through_v4_worker_migration_reopen_and_mixed_file_retry() { migrate(&mut converted); let mut mixed = encode(&converted[..1]); mixed.extend_from_slice(source.split_once('\n').unwrap().1.as_bytes()); - fs::write(&path, mixed).unwrap(); + fs::write(&path, &mixed).unwrap(); let new_path = root .path() .join("segments") @@ -224,6 +224,11 @@ fn v1_through_v4_worker_migration_reopen_and_mixed_file_retry() { .join(format!("{}.jsonl", SegmentId::from_u128(3))); fs::write(&old_path, source).unwrap(); + let marker = fs::read(root.path().join("session.json")).unwrap(); + WorkerSessionStore::validate_migration(root.path()).unwrap(); + assert_eq!(fs::read(&path).unwrap(), mixed); + assert_eq!(fs::read(&old_path).unwrap(), source.as_bytes()); + assert_eq!(fs::read(root.path().join("session.json")).unwrap(), marker); let store = WorkerSessionStore::new(root.path()).unwrap(); assert_marker(root.path(), 6); let restored = store.read_all(session, segment).unwrap(); @@ -286,6 +291,7 @@ fn all_segments_preflight_keeps_old_bytes_on_decode_or_conversion_failure() { .join(format!("{}.jsonl", SegmentId::from_u128(2))); fs::write(&later, &bad).unwrap(); let marker = fs::read(root.path().join("session.json")).unwrap(); + assert!(WorkerSessionStore::validate_migration(root.path()).is_err()); assert!(WorkerSessionStore::new(root.path()).is_err()); assert_eq!(fs::read(&first).unwrap(), V4.as_bytes()); assert_eq!(fs::read(&later).unwrap(), bad.as_bytes()); diff --git a/crates/session-store/src/worker_session_store.rs b/crates/session-store/src/worker_session_store.rs index 9c6d8db00..a19f1b2db 100644 --- a/crates/session-store/src/worker_session_store.rs +++ b/crates/session-store/src/worker_session_store.rs @@ -259,6 +259,33 @@ impl WorkerSessionStore { }) } + /// Validate an existing Session's conversion without publishing any bytes. + /// Hosts can isolate unreadable archives before admitting a staged migration; + /// the versioned decoder and conversion checks remain owned by this store. + pub fn validate_migration(root: &Path) -> Result<(), StoreError> { + let manifest: SessionManifest = + serde_json::from_slice(&fs::read(root.join(SESSION_FILE))?)?; + match manifest.schema_version { + SESSION_SCHEMA_VERSION => validate_canonical_segment_logs(root), + LEGACY_SESSION_SCHEMA_VERSION + | PREVIOUS_SESSION_SCHEMA_VERSION + | ANNOTATED_SESSION_SCHEMA_VERSION + | COMPACTION_SESSION_SCHEMA_VERSION + | SYSTEM_INPUT_SESSION_SCHEMA_VERSION => migrate_segment_logs_to_current( + segment_log_paths(root)?, + manifest.session_id, + manifest.schema_version, + |_, _| Ok(()), + ), + version => Err(StoreError::Corrupt { + line: 0, + message: format!( + "unsupported Worker Session schema version {version}, expected {SESSION_SCHEMA_VERSION}" + ), + }), + } + } + /// Open an already-retained Session without creating directories or migrating /// persisted data. Observation paths must never mutate retained state. pub fn open_read_only(root: impl Into) -> Result { diff --git a/crates/worker-runtime/src/retention.rs b/crates/worker-runtime/src/retention.rs index 6e2496246..f283a03dd 100644 --- a/crates/worker-runtime/src/retention.rs +++ b/crates/worker-runtime/src/retention.rs @@ -641,6 +641,13 @@ impl FsWorkerRetentionProvider { let (manifest, session_root) = self.session_archive_identity(archive_id)?; let archive_dir = self.archive_dir(archive_id)?; let manifest = validate_existing_archive(&archive_dir, &manifest, false)?; + session_store::WorkerSessionStore::open_read_only(&session_root).map_err(|_| { + RuntimeError::StoreCorrupt { + operation: "read Worker Session archive", + path: session_root.clone(), + message: "archive Session is unavailable in the current schema".into(), + } + })?; Ok((manifest, session_root)) } @@ -653,13 +660,7 @@ impl FsWorkerRetentionProvider { &archive_dir.join("manifest.json"), "read Worker Session archive", )?; - if manifest.schema_version != ARCHIVE_SCHEMA_VERSION || manifest.archive_id != archive_id { - return Err(RuntimeError::StoreCorrupt { - operation: "read Worker Session archive", - path: archive_dir.join("manifest.json"), - message: "archive manifest identity mismatch".to_string(), - }); - } + validate_archive_identity(&archive_dir, &manifest)?; Ok((manifest, archive_dir.join("session"))) } @@ -1204,9 +1205,10 @@ fn archive_storage( } } -// Share discovery and erasure fencing between the read-only Runtime preflight -// and its private staging pass. Session Store remains the authority for schema -// admission and conversion; Runtime must not duplicate its version table. +// Share discovery and erasure fencing between Runtime preflight and staging. +// Unreadable archives are retained but excluded before any conversion writes; +// one historical archive must not block the live Runtime's startup. Session +// Store owns conversion admission, including the current resource limits. fn for_each_live_session( root: &Path, mut visit: impl FnMut(&Path, bool) -> Result<(), RuntimeError>, @@ -1232,39 +1234,80 @@ fn for_each_live_session( })?; let aggregate = entry.path(); let session = aggregate.join("session"); - if !session - .join("session.json") - .try_exists() - .map_err(|source| RuntimeError::StoreIo { - operation: "inspect Session migration", - path: session.clone(), - source, - })? - { - continue; - } - let identity: CanonicalSessionManifest = read_json( - &session.join("session.json"), - "read Session migration identity", - )?; - let session_id = identity.session_id.parse().map_err(|_| { - RuntimeError::InvalidRequest("Session migration identity is invalid".into()) - })?; + let is_archive = directory == root.join("archives/workers"); + let identity = (|| { + if !session + .join("session.json") + .try_exists() + .map_err(|source| RuntimeError::StoreIo { + operation: "inspect Session migration", + path: session.clone(), + source, + })? + { + return Ok(None); + } + let identity: CanonicalSessionManifest = read_json( + &session.join("session.json"), + "read Session migration identity", + )?; + identity.session_id.parse().map(Some).map_err(|_| { + RuntimeError::InvalidRequest("Session migration identity is invalid".into()) + }) + })(); + let session_id = match identity { + Ok(Some(session_id)) => session_id, + Ok(None) => continue, + Err(_) if is_archive => { + warn_unavailable_archive(&aggregate); + continue; + } + Err(error) => return Err(error), + }; match runtime_session_retention(root).ensure_live(session_id) { Ok(()) => {} - // Erasure fences remain authoritative even if physical removal - // was interrupted. Do not inspect or migrate fenced originals. + // Retention-ledger failures are not archive-local. Never bypass + // erasure authority while isolating an unreadable archive. Err(session_store::StoreError::SessionErased(_)) => continue, Err(error) => return Err(retention_error(error)), } - visit(&aggregate, directory == root.join("archives/workers"))?; + if is_archive && validate_archive_migration(&aggregate).is_err() { + warn_unavailable_archive(&aggregate); + continue; + } + // Publication failures still abort the outer staging transaction. + visit(&aggregate, is_archive)?; } } Ok(()) } -/// Read-only admission probe: a rejected Session needs the existing staged -/// migration attempt (which will also reject corrupt/unsupported inputs). +fn validate_archive_migration(aggregate: &Path) -> Result<(), RuntimeError> { + let session = aggregate.join("session"); + // Current archives need no conversion. Their payload integrity is checked + // on access, not by scanning every retained byte during Runtime startup. + if session_store::WorkerSessionStore::open_read_only(&session).is_ok() { + return Ok(()); + } + let manifest: WorkerSessionArchiveManifest = read_json( + &aggregate.join("manifest.json"), + "read archive before Session migration", + )?; + validate_existing_archive(aggregate, &manifest, false)?; + session_store::WorkerSessionStore::validate_migration(&session).map_err(retention_error) +} + +fn warn_unavailable_archive(aggregate: &Path) { + let archive_id = aggregate.file_name().unwrap_or_default().to_string_lossy(); + tracing::warn!( + archive_id = %bounded_diagnostic_id(&archive_id), + code = "session_archive_unavailable", + "Retained Session archive could not be admitted for migration; original bytes retained" + ); +} + +/// Read-only admission probe. Only convertible archives request a staged pass; +/// unreadable originals remain available for archive-local diagnostics/recovery. /// Never create a writable Session store or rewrite archive attestations here. pub(crate) fn sessions_require_migration(root: &Path) -> Result { let mut required = false; @@ -1281,6 +1324,11 @@ pub(crate) fn sessions_require_migration(root: &Path) -> Result Result<(), RuntimeError> { for_each_live_session(root, |aggregate, is_archive| { let session = aggregate.join("session"); + // Current Sessions need no writable admission, and their storage + // attestations must remain byte-for-byte unchanged. + if session_store::WorkerSessionStore::open_read_only(&session).is_ok() { + return Ok(()); + } let archive = if is_archive { let manifest: WorkerSessionArchiveManifest = read_json( &aggregate.join("manifest.json"), @@ -1290,11 +1338,6 @@ pub(crate) fn migrate_staged_sessions(root: &Path) -> Result<(), RuntimeError> { } else { None }; - // Current Sessions need no writable admission, and their storage - // attestations must remain byte-for-byte unchanged. - if session_store::WorkerSessionStore::open_read_only(&session).is_ok() { - return Ok(()); - } session_store::WorkerSessionStore::new(&session).map_err(|error| { RuntimeError::StoreCorrupt { operation: "migrate Session schema", @@ -1323,6 +1366,22 @@ pub(crate) fn migrate_staged_sessions(root: &Path) -> Result<(), RuntimeError> { }) } +fn validate_archive_identity( + archive_dir: &Path, + manifest: &WorkerSessionArchiveManifest, +) -> Result<(), RuntimeError> { + if manifest.schema_version != ARCHIVE_SCHEMA_VERSION + || archive_dir.file_name().and_then(|name| name.to_str()) != Some(&manifest.archive_id) + { + return Err(RuntimeError::StoreCorrupt { + operation: "read Worker Session archive", + path: archive_dir.join("manifest.json"), + message: "archive manifest identity mismatch".into(), + }); + } + Ok(()) +} + fn validate_existing_archive( archive_dir: &Path, expected: &WorkerSessionArchiveManifest, @@ -1332,6 +1391,7 @@ fn validate_existing_archive( &archive_dir.join("manifest.json"), "verify Worker Session archive", )?; + validate_archive_identity(archive_dir, &existing)?; let storage = archive_storage(archive_dir, &existing)?; let mut comparable_expected = expected.clone(); comparable_expected.archived_at_unix_seconds = existing.archived_at_unix_seconds; @@ -2040,7 +2100,7 @@ mod tests { } #[test] - fn session_migration_rejects_corrupt_archive_before_replacing_its_bytes() { + fn session_migration_preserves_corrupt_archive_without_blocking_other_sessions() { let temp = tempfile::tempdir().unwrap(); let worker = WorkerId::now_v7(); let compact = old_compacted_source(temp.path(), worker); @@ -2052,7 +2112,7 @@ mod tests { .join("session/segments") .join(format!("{compact}.jsonl")); fs::write(&path, b"corrupt\n").unwrap(); - assert!(migrate_staged_sessions(temp.path()).is_err()); + migrate_staged_sessions(temp.path()).unwrap(); assert_eq!(fs::read(&path).unwrap(), b"corrupt\n"); assert!(!archive.join(ARCHIVE_STORAGE_FILE).exists()); } diff --git a/crates/worker-runtime/src/retention/runtime_v5_tests.rs b/crates/worker-runtime/src/retention/runtime_v5_tests.rs index 5527dba73..d9a3bccdd 100644 --- a/crates/worker-runtime/src/retention/runtime_v5_tests.rs +++ b/crates/worker-runtime/src/retention/runtime_v5_tests.rs @@ -359,7 +359,7 @@ fn runtime_v11_and_current_route_v4_stopped_orphan_archive_through_staging() { } #[test] -fn runtime_v5_archive_failure_discards_staging_without_losing_originals_and_retry_converges() { +fn corrupt_archive_does_not_block_startup_or_healthy_session_migration_and_repair_converges() { for runtime_schema in [11, 12] { let temp = tempfile::tempdir().unwrap(); let root = temp.path().join("runtime"); @@ -373,18 +373,48 @@ fn runtime_v5_archive_failure_discards_staging_without_losing_originals_and_retr .join(format!("{archived_segment}.jsonl")); let original = fs::read(&path).unwrap(); fs::write(&path, b"corrupt archive\n").unwrap(); - let before = tree_bytes(&root); + let (healthy_worker, healthy_session, healthy_segment) = + stopped_worker(&root, runtime_schema); + let mut healthy_request = + super::tests::request(healthy_worker, SessionDisposition::Archive); + healthy_request.operation_id = "healthy-operation".into(); + healthy_request.archive_id = Some("healthy-archive".into()); + let healthy_manifest = commit_session_archive( + &provider, + &healthy_request, + &provider.worker_dir(healthy_worker), + ) + .unwrap(); + fs::remove_dir_all(provider.worker_dir(healthy_worker)).unwrap(); + let before = tree_bytes(&archive); let options = FsRuntimeStoreOptions::new(&root).with_runtime_id("runtime-a"); assert!( FsRuntimeStore::migration_plan(&options) .unwrap() .migration_required ); - // workers are visited before archives: earlier successful staged - // conversion must not leak into the original tree on this later error. - assert!(FsRuntimeStore::open_or_create(root.clone(), "runtime-a").is_err()); - assert_eq!(tree_bytes(&root), before); + drop(FsRuntimeStore::open_or_create(root.clone(), "runtime-a").unwrap()); + assert_eq!(tree_bytes(&archive), before); + assert!(provider.session_archive("archive-a").is_err()); + let (manifest, healthy_root) = provider.session_archive("healthy-archive").unwrap(); + assert_eq!(manifest, healthy_manifest); + assert_generic(&healthy_root, healthy_session, healthy_segment); assert!(!archive.join(ARCHIVE_STORAGE_FILE).exists()); + assert_generic( + &root + .join("workers") + .join(worker.to_string()) + .join("session"), + session, + segment, + ); + assert!( + !FsRuntimeStore::migration_plan(&options) + .unwrap() + .migration_required + ); + drop(FsRuntimeStore::open_or_create(root.clone(), "runtime-a").unwrap()); + assert_eq!(tree_bytes(&archive), before); fs::write(&path, original).unwrap(); FsRuntimeStore::migrate(&options).unwrap(); assert_generic( @@ -401,52 +431,162 @@ fn runtime_v5_archive_failure_discards_staging_without_losing_originals_and_retr } #[test] -fn runtime_v5_typed_legacy_payload_failure_keeps_valid_archive_evidence_on_retries() { - for (from, to) in [ - ( - "\"snapshot_sha256\":\"snapshot-digest\"", - "\"snapshot_sha256\":42", - ), - ("\"behavior_md\":\"exact behavior\"", "\"behavior_md\":[]"), +fn unmigratable_archive_preserves_evidence_across_startup_and_reopen() { + for (field, value) in [ + ("snapshot_sha256", json!(42)), + ("behavior_md", json!([])), + ("snapshot", json!({"large": "x".repeat(128 * 1024 + 1)})), + ("body", json!("x".repeat(256 * 1024 + 1))), ] { let temp = tempfile::tempdir().unwrap(); let root = temp.path().join("runtime"); - runtime_fixture(&root, 12); + let (live_worker, live_session, live_segment) = runtime_fixture(&root, 12); let (worker, _, segment) = stopped_worker(&root, 12); let provider = FsWorkerRetentionProvider::new(&root); let path = provider .worker_dir(worker) .join("session/segments") .join(format!("{segment}.jsonl")); - let original = fs::read_to_string(&path).unwrap(); - assert!(original.contains(from)); - fs::write(&path, original.replace(from, to)).unwrap(); + let source = fs::read_to_string(&path).unwrap(); + let mut replaced = false; + let mut altered = String::new(); + for line in source.lines() { + let mut entry: Value = serde_json::from_str(line).unwrap(); + if let Some(item) = entry.pointer_mut("/entry/item") + && item.get(field).is_some() + { + item[field] = value.clone(); + replaced = true; + } + altered.push_str(&serde_json::to_string(&entry).unwrap()); + altered.push('\n'); + } + assert!(replaced); + fs::write(&path, altered).unwrap(); let request = request(worker, SessionDisposition::Archive); let manifest = commit_session_archive(&provider, &request, &provider.worker_dir(worker)).unwrap(); fs::remove_dir_all(provider.worker_dir(worker)).unwrap(); let archive = provider.archive_dir("archive-a").unwrap(); - // This archive checksum is valid; failure must come from the legacy - // typed decoder, not from deliberately broken ingestion evidence. validate_existing_archive(&archive, &manifest, false).unwrap(); - let before = tree_bytes(&root); + let before = tree_bytes(&archive); let options = FsRuntimeStoreOptions::new(&root).with_runtime_id("runtime-a"); for _ in 0..2 { - let error = FsRuntimeStore::migrate(&options).unwrap_err(); - let RuntimeError::StoreCorrupt { - operation, path, .. - } = error - else { - panic!("unexpected failure: {error}"); + let runtime = crate::Runtime::with_fs_store(options.clone()).unwrap(); + assert_eq!(runtime.summary().unwrap().worker_count, 1); + runtime.bind_runtime_identity("runtime-a").unwrap(); + let search = runtime_api::SessionPublicSearchRequest { + workspace_id: "workspace-a".into(), + source: runtime_api::SessionPublicSource::Archived { + archive_id: "archive-a".into(), + }, + expected_session_id: manifest.source_session_id.clone(), + expected_generation: None, + query: None, + kind: None, + tool_name: None, + tool_part: runtime_api::SessionPublicToolPart::Both, + scan_cursor: None, + limit: 10, + max_scan_bytes: 1024 * 1024, + max_segments: 10, + max_entries: 100, }; - assert_eq!(operation, "migrate Session schema"); - assert!(path.starts_with(temp.path().join(".runtime.schema-v12-staging"))); - assert_eq!(tree_bytes(&root), before); + let availability = runtime + .session_public_search_scoped( + &crate::runtime::RuntimeWorkspaceScope::new("workspace-a", "owner-a"), + &search, + ) + .unwrap(); + assert!(matches!( + availability, + runtime_api::SessionPublicSearchAvailability::Unavailable { + reason: runtime_api::SessionPublicUnavailableReason::ArchiveIncomplete, + .. + } + )); + drop(runtime); + assert_eq!(tree_bytes(&archive), before); validate_existing_archive(&archive, &manifest, false).unwrap(); + assert!(matches!( + provider.session_archive("archive-a"), + Err(RuntimeError::StoreCorrupt { .. }) + )); + assert!( + !FsRuntimeStore::migration_plan(&options) + .unwrap() + .migration_required + ); + assert_generic( + &provider.worker_dir(live_worker).join("session"), + live_session, + live_segment, + ); } } } +#[test] +fn unreadable_archive_metadata_does_not_block_archive_only_runtime_startup() { + for (path, field, value) in [ + ("session/session.json", None, Value::Null), + ("manifest.json", None, Value::Null), + (ARCHIVE_STORAGE_FILE, None, Value::Null), + ("session/session.json", Some("schema_version"), json!(999)), + ("manifest.json", Some("schema_version"), json!(999)), + ( + "manifest.json", + Some("archive_id"), + json!("another-archive"), + ), + ] { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().join("runtime"); + let (worker, _, _) = runtime_fixture(&root, 12); + completed_v4_archive(&root, 12); + let provider = FsWorkerRetentionProvider::new(&root); + fs::remove_dir_all(provider.worker_dir(worker)).unwrap(); + let archive = provider.archive_dir("archive-a").unwrap(); + if let Some(field) = field { + let mut record: Value = read_json(&archive.join(path), "test").unwrap(); + record[field] = value; + write_json(&archive.join(path), &record); + } else { + fs::write(archive.join(path), b"{broken\n").unwrap(); + } + let before = tree_bytes(&archive); + let options = FsRuntimeStoreOptions::new(&root).with_runtime_id("runtime-a"); + for _ in 0..2 { + assert!( + !FsRuntimeStore::migration_plan(&options) + .unwrap() + .migration_required + ); + let runtime = crate::Runtime::with_fs_store(options.clone()).unwrap(); + assert_eq!(runtime.summary().unwrap().worker_count, 0); + drop(runtime); + assert!(provider.session_archive("archive-a").is_err()); + assert_eq!(tree_bytes(&archive), before); + } + } +} + +#[test] +fn archive_isolation_does_not_bypass_corrupt_erasure_authority() { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().join("runtime"); + runtime_fixture(&root, 12); + let (session, _, _, _) = completed_v4_archive(&root, 12); + let ledger = root.join("retention/session-sources"); + fs::create_dir_all(&ledger).unwrap(); + fs::write(ledger.join(format!("{session}.json")), b"{broken\n").unwrap(); + let before = tree_bytes(&root); + let options = FsRuntimeStoreOptions::new(&root).with_runtime_id("runtime-a"); + assert!(FsRuntimeStore::migration_plan(&options).is_err()); + assert!(crate::Runtime::with_fs_store(options).is_err()); + assert_eq!(tree_bytes(&root), before); +} + #[test] fn runtime_v5_probe_and_staging_skip_erased_originals() { let temp = tempfile::tempdir().unwrap(); diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 1c4bb8dcb..fe8b6ca7c 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -58627,6 +58627,58 @@ mod tests { .unwrap(); } + #[tokio::test] + async fn server_preload_keeps_workspace_available_with_unreadable_embedded_archive() { + // Exercise catalog preload through the production embedded Runtime/store; + // only the database and filesystem roots are isolated fixtures. + let temp = tempfile::tempdir().unwrap(); + let config = test_server_config(temp.path()); + let store = + Arc::new(SqliteWorkspaceStore::open_api_fixture(&config.database_path).unwrap()); + seed_test_registered_workspace(store.as_ref(), &config) + .await + .unwrap(); + let workspace = store + .get_workspace(TEST_WORKSPACE_ID) + .await + .unwrap() + .unwrap(); + let scoped = config + .for_catalog_workspace(&workspace, Vec::new()) + .unwrap(); + drop( + worker_runtime::Runtime::with_fs_store( + worker_runtime::fs_store::FsRuntimeStoreOptions::new( + &scoped.embedded_runtime_store_root, + ) + .with_runtime_id(EMBEDDED_WORKER_RUNTIME_ID), + ) + .unwrap(), + ); + let archive = scoped + .embedded_runtime_store_root + .join("archives/workers/unreadable-archive"); + fs::create_dir_all(archive.join("session")).unwrap(); + let marker = archive.join("session/session.json"); + fs::write(&marker, b"{broken archive identity\n").unwrap(); + let server = WorkspaceServerApi::new_for_test(config, store); + server.preload().await.unwrap(); + let api = server + .api_for_workspace(TEST_WORKSPACE_ID) + .await + .unwrap() + .unwrap(); + assert!( + api.runtime + .list_runtimes(100) + .items + .iter() + .any(|runtime| { runtime.runtime_id == EMBEDDED_WORKER_RUNTIME_ID }) + ); + assert_eq!(fs::read(marker).unwrap(), b"{broken archive identity\n"); + server.shutdown_feature_storage().await.unwrap(); + } + #[tokio::test] async fn api_startup_revalidates_workspace_schema_after_store_was_opened() { let temp = tempfile::tempdir().unwrap(); diff --git a/docs/development/workspace-schema-migrations.md b/docs/development/workspace-schema-migrations.md index fa91cfbf3..6159d1646 100644 --- a/docs/development/workspace-schema-migrations.md +++ b/docs/development/workspace-schema-migrations.md @@ -12,6 +12,6 @@ schema変更では、既存の順序付きmigrationへ新しい処理を追加 旧Host compactionのoriginはSession storeのversioned migrationで永続的に更新する。通常復元やpublic readerでは文字列markerによる読み替えを行わない。本文・logical entry ID・derivationを維持し、位置由来のRun boundary IDは書換え前のoffsetから保存する。StandaloneのSessionも同じ変換を使う。 -Runtimeの移行はowner lock下のstaging treeで停止中Worker・orphan・既存Session archiveを更新し、成功したtreeだけを公開する。全treeのコピーに必要な空き容量が必要。失敗時は元treeを維持し、切替中の中断は既存のstaging/backup回復経路で扱う。 +Runtimeの移行はowner lock下のstaging treeで停止中Worker・orphan・移行可能な既存Session archiveを更新し、成功したtreeだけを公開する。全treeのコピーに必要な空き容量が必要。書込み失敗時は元treeを維持し、切替中の中断は既存のstaging/backup回復経路で扱う。履歴archiveの読取り・検証・変換不能はServer全体の起動条件にせず、書込み前のpreflightでそのarchiveだけを除外する。除外した原本と取り込み証拠は変更せず、警告と利用時の利用不能応答で示す。Runtime・Workerの正本や削除権限の異常、移行結果の公開失敗まで握り潰さない。 -Archiveの`manifest.json`はBackend catalogと完了receiptにも保存された取り込み時の証拠なので変更しない。移行前のchecksumを検証したうえでpayloadを現行形式に置換し、元manifest全体と現行payloadのchecksum/bytes/file countを結ぶ`session-storage.json`を同じtree切替で公開する。旧payloadは読取り用に残さない。通常のarchive検証では現行payloadを検証し、APIとoperation replayには元の取り込みreceiptを返す。移行後の削除再試行では、現行原本とarchiveの現行storage summaryが一致することを要求する。 +Archiveの`manifest.json`はBackend catalogと完了receiptにも保存された取り込み時の証拠なので変更しない。移行前のchecksumを検証したうえでpayloadを現行形式に置換し、元manifest全体と現行payloadのchecksum/bytes/file countを結ぶ`session-storage.json`を同じtree切替で公開する。移行に成功した旧payloadは読取り用に残さず、移行不能で保持した旧payloadも現行readerでは受理しない。通常のarchive検証では現行payloadを検証し、APIとoperation replayには元の取り込みreceiptを返す。移行後の削除再試行では、現行原本とarchiveの現行storage summaryが一致することを要求する。