fix: isolate unreadable archives from runtime startup
This commit is contained in:
@@ -211,7 +211,7 @@ fn v1_through_v4_worker_migration_reopen_and_mixed_file_retry() {
|
|||||||
migrate(&mut converted);
|
migrate(&mut converted);
|
||||||
let mut mixed = encode(&converted[..1]);
|
let mut mixed = encode(&converted[..1]);
|
||||||
mixed.extend_from_slice(source.split_once('\n').unwrap().1.as_bytes());
|
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
|
let new_path = root
|
||||||
.path()
|
.path()
|
||||||
.join("segments")
|
.join("segments")
|
||||||
@@ -224,6 +224,11 @@ fn v1_through_v4_worker_migration_reopen_and_mixed_file_retry() {
|
|||||||
.join(format!("{}.jsonl", SegmentId::from_u128(3)));
|
.join(format!("{}.jsonl", SegmentId::from_u128(3)));
|
||||||
fs::write(&old_path, source).unwrap();
|
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();
|
let store = WorkerSessionStore::new(root.path()).unwrap();
|
||||||
assert_marker(root.path(), 6);
|
assert_marker(root.path(), 6);
|
||||||
let restored = store.read_all(session, segment).unwrap();
|
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)));
|
.join(format!("{}.jsonl", SegmentId::from_u128(2)));
|
||||||
fs::write(&later, &bad).unwrap();
|
fs::write(&later, &bad).unwrap();
|
||||||
let marker = fs::read(root.path().join("session.json")).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!(WorkerSessionStore::new(root.path()).is_err());
|
||||||
assert_eq!(fs::read(&first).unwrap(), V4.as_bytes());
|
assert_eq!(fs::read(&first).unwrap(), V4.as_bytes());
|
||||||
assert_eq!(fs::read(&later).unwrap(), bad.as_bytes());
|
assert_eq!(fs::read(&later).unwrap(), bad.as_bytes());
|
||||||
|
|||||||
@@ -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
|
/// Open an already-retained Session without creating directories or migrating
|
||||||
/// persisted data. Observation paths must never mutate retained state.
|
/// persisted data. Observation paths must never mutate retained state.
|
||||||
pub fn open_read_only(root: impl Into<PathBuf>) -> Result<Self, StoreError> {
|
pub fn open_read_only(root: impl Into<PathBuf>) -> Result<Self, StoreError> {
|
||||||
|
|||||||
@@ -641,6 +641,13 @@ impl FsWorkerRetentionProvider {
|
|||||||
let (manifest, session_root) = self.session_archive_identity(archive_id)?;
|
let (manifest, session_root) = self.session_archive_identity(archive_id)?;
|
||||||
let archive_dir = self.archive_dir(archive_id)?;
|
let archive_dir = self.archive_dir(archive_id)?;
|
||||||
let manifest = validate_existing_archive(&archive_dir, &manifest, false)?;
|
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))
|
Ok((manifest, session_root))
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -653,13 +660,7 @@ impl FsWorkerRetentionProvider {
|
|||||||
&archive_dir.join("manifest.json"),
|
&archive_dir.join("manifest.json"),
|
||||||
"read Worker Session archive",
|
"read Worker Session archive",
|
||||||
)?;
|
)?;
|
||||||
if manifest.schema_version != ARCHIVE_SCHEMA_VERSION || manifest.archive_id != archive_id {
|
validate_archive_identity(&archive_dir, &manifest)?;
|
||||||
return Err(RuntimeError::StoreCorrupt {
|
|
||||||
operation: "read Worker Session archive",
|
|
||||||
path: archive_dir.join("manifest.json"),
|
|
||||||
message: "archive manifest identity mismatch".to_string(),
|
|
||||||
});
|
|
||||||
}
|
|
||||||
Ok((manifest, archive_dir.join("session")))
|
Ok((manifest, archive_dir.join("session")))
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1204,9 +1205,10 @@ fn archive_storage(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Share discovery and erasure fencing between the read-only Runtime preflight
|
// Share discovery and erasure fencing between Runtime preflight and staging.
|
||||||
// and its private staging pass. Session Store remains the authority for schema
|
// Unreadable archives are retained but excluded before any conversion writes;
|
||||||
// admission and conversion; Runtime must not duplicate its version table.
|
// 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(
|
fn for_each_live_session(
|
||||||
root: &Path,
|
root: &Path,
|
||||||
mut visit: impl FnMut(&Path, bool) -> Result<(), RuntimeError>,
|
mut visit: impl FnMut(&Path, bool) -> Result<(), RuntimeError>,
|
||||||
@@ -1232,39 +1234,80 @@ fn for_each_live_session(
|
|||||||
})?;
|
})?;
|
||||||
let aggregate = entry.path();
|
let aggregate = entry.path();
|
||||||
let session = aggregate.join("session");
|
let session = aggregate.join("session");
|
||||||
if !session
|
let is_archive = directory == root.join("archives/workers");
|
||||||
.join("session.json")
|
let identity = (|| {
|
||||||
.try_exists()
|
if !session
|
||||||
.map_err(|source| RuntimeError::StoreIo {
|
.join("session.json")
|
||||||
operation: "inspect Session migration",
|
.try_exists()
|
||||||
path: session.clone(),
|
.map_err(|source| RuntimeError::StoreIo {
|
||||||
source,
|
operation: "inspect Session migration",
|
||||||
})?
|
path: session.clone(),
|
||||||
{
|
source,
|
||||||
continue;
|
})?
|
||||||
}
|
{
|
||||||
let identity: CanonicalSessionManifest = read_json(
|
return Ok(None);
|
||||||
&session.join("session.json"),
|
}
|
||||||
"read Session migration identity",
|
let identity: CanonicalSessionManifest = read_json(
|
||||||
)?;
|
&session.join("session.json"),
|
||||||
let session_id = identity.session_id.parse().map_err(|_| {
|
"read Session migration identity",
|
||||||
RuntimeError::InvalidRequest("Session migration identity is invalid".into())
|
)?;
|
||||||
})?;
|
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) {
|
match runtime_session_retention(root).ensure_live(session_id) {
|
||||||
Ok(()) => {}
|
Ok(()) => {}
|
||||||
// Erasure fences remain authoritative even if physical removal
|
// Retention-ledger failures are not archive-local. Never bypass
|
||||||
// was interrupted. Do not inspect or migrate fenced originals.
|
// erasure authority while isolating an unreadable archive.
|
||||||
Err(session_store::StoreError::SessionErased(_)) => continue,
|
Err(session_store::StoreError::SessionErased(_)) => continue,
|
||||||
Err(error) => return Err(retention_error(error)),
|
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(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Read-only admission probe: a rejected Session needs the existing staged
|
fn validate_archive_migration(aggregate: &Path) -> Result<(), RuntimeError> {
|
||||||
/// migration attempt (which will also reject corrupt/unsupported inputs).
|
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.
|
/// Never create a writable Session store or rewrite archive attestations here.
|
||||||
pub(crate) fn sessions_require_migration(root: &Path) -> Result<bool, RuntimeError> {
|
pub(crate) fn sessions_require_migration(root: &Path) -> Result<bool, RuntimeError> {
|
||||||
let mut required = false;
|
let mut required = false;
|
||||||
@@ -1281,6 +1324,11 @@ pub(crate) fn sessions_require_migration(root: &Path) -> Result<bool, RuntimeErr
|
|||||||
pub(crate) fn migrate_staged_sessions(root: &Path) -> Result<(), RuntimeError> {
|
pub(crate) fn migrate_staged_sessions(root: &Path) -> Result<(), RuntimeError> {
|
||||||
for_each_live_session(root, |aggregate, is_archive| {
|
for_each_live_session(root, |aggregate, is_archive| {
|
||||||
let session = aggregate.join("session");
|
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 archive = if is_archive {
|
||||||
let manifest: WorkerSessionArchiveManifest = read_json(
|
let manifest: WorkerSessionArchiveManifest = read_json(
|
||||||
&aggregate.join("manifest.json"),
|
&aggregate.join("manifest.json"),
|
||||||
@@ -1290,11 +1338,6 @@ pub(crate) fn migrate_staged_sessions(root: &Path) -> Result<(), RuntimeError> {
|
|||||||
} else {
|
} else {
|
||||||
None
|
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| {
|
session_store::WorkerSessionStore::new(&session).map_err(|error| {
|
||||||
RuntimeError::StoreCorrupt {
|
RuntimeError::StoreCorrupt {
|
||||||
operation: "migrate Session schema",
|
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(
|
fn validate_existing_archive(
|
||||||
archive_dir: &Path,
|
archive_dir: &Path,
|
||||||
expected: &WorkerSessionArchiveManifest,
|
expected: &WorkerSessionArchiveManifest,
|
||||||
@@ -1332,6 +1391,7 @@ fn validate_existing_archive(
|
|||||||
&archive_dir.join("manifest.json"),
|
&archive_dir.join("manifest.json"),
|
||||||
"verify Worker Session archive",
|
"verify Worker Session archive",
|
||||||
)?;
|
)?;
|
||||||
|
validate_archive_identity(archive_dir, &existing)?;
|
||||||
let storage = archive_storage(archive_dir, &existing)?;
|
let storage = archive_storage(archive_dir, &existing)?;
|
||||||
let mut comparable_expected = expected.clone();
|
let mut comparable_expected = expected.clone();
|
||||||
comparable_expected.archived_at_unix_seconds = existing.archived_at_unix_seconds;
|
comparable_expected.archived_at_unix_seconds = existing.archived_at_unix_seconds;
|
||||||
@@ -2040,7 +2100,7 @@ mod tests {
|
|||||||
}
|
}
|
||||||
|
|
||||||
#[test]
|
#[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 temp = tempfile::tempdir().unwrap();
|
||||||
let worker = WorkerId::now_v7();
|
let worker = WorkerId::now_v7();
|
||||||
let compact = old_compacted_source(temp.path(), worker);
|
let compact = old_compacted_source(temp.path(), worker);
|
||||||
@@ -2052,7 +2112,7 @@ mod tests {
|
|||||||
.join("session/segments")
|
.join("session/segments")
|
||||||
.join(format!("{compact}.jsonl"));
|
.join(format!("{compact}.jsonl"));
|
||||||
fs::write(&path, b"corrupt\n").unwrap();
|
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_eq!(fs::read(&path).unwrap(), b"corrupt\n");
|
||||||
assert!(!archive.join(ARCHIVE_STORAGE_FILE).exists());
|
assert!(!archive.join(ARCHIVE_STORAGE_FILE).exists());
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -359,7 +359,7 @@ fn runtime_v11_and_current_route_v4_stopped_orphan_archive_through_staging() {
|
|||||||
}
|
}
|
||||||
|
|
||||||
#[test]
|
#[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] {
|
for runtime_schema in [11, 12] {
|
||||||
let temp = tempfile::tempdir().unwrap();
|
let temp = tempfile::tempdir().unwrap();
|
||||||
let root = temp.path().join("runtime");
|
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"));
|
.join(format!("{archived_segment}.jsonl"));
|
||||||
let original = fs::read(&path).unwrap();
|
let original = fs::read(&path).unwrap();
|
||||||
fs::write(&path, b"corrupt archive\n").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");
|
let options = FsRuntimeStoreOptions::new(&root).with_runtime_id("runtime-a");
|
||||||
assert!(
|
assert!(
|
||||||
FsRuntimeStore::migration_plan(&options)
|
FsRuntimeStore::migration_plan(&options)
|
||||||
.unwrap()
|
.unwrap()
|
||||||
.migration_required
|
.migration_required
|
||||||
);
|
);
|
||||||
// workers are visited before archives: earlier successful staged
|
drop(FsRuntimeStore::open_or_create(root.clone(), "runtime-a").unwrap());
|
||||||
// conversion must not leak into the original tree on this later error.
|
assert_eq!(tree_bytes(&archive), before);
|
||||||
assert!(FsRuntimeStore::open_or_create(root.clone(), "runtime-a").is_err());
|
assert!(provider.session_archive("archive-a").is_err());
|
||||||
assert_eq!(tree_bytes(&root), before);
|
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!(!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();
|
fs::write(&path, original).unwrap();
|
||||||
FsRuntimeStore::migrate(&options).unwrap();
|
FsRuntimeStore::migrate(&options).unwrap();
|
||||||
assert_generic(
|
assert_generic(
|
||||||
@@ -401,52 +431,162 @@ fn runtime_v5_archive_failure_discards_staging_without_losing_originals_and_retr
|
|||||||
}
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn runtime_v5_typed_legacy_payload_failure_keeps_valid_archive_evidence_on_retries() {
|
fn unmigratable_archive_preserves_evidence_across_startup_and_reopen() {
|
||||||
for (from, to) in [
|
for (field, value) in [
|
||||||
(
|
("snapshot_sha256", json!(42)),
|
||||||
"\"snapshot_sha256\":\"snapshot-digest\"",
|
("behavior_md", json!([])),
|
||||||
"\"snapshot_sha256\":42",
|
("snapshot", json!({"large": "x".repeat(128 * 1024 + 1)})),
|
||||||
),
|
("body", json!("x".repeat(256 * 1024 + 1))),
|
||||||
("\"behavior_md\":\"exact behavior\"", "\"behavior_md\":[]"),
|
|
||||||
] {
|
] {
|
||||||
let temp = tempfile::tempdir().unwrap();
|
let temp = tempfile::tempdir().unwrap();
|
||||||
let root = temp.path().join("runtime");
|
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 (worker, _, segment) = stopped_worker(&root, 12);
|
||||||
let provider = FsWorkerRetentionProvider::new(&root);
|
let provider = FsWorkerRetentionProvider::new(&root);
|
||||||
let path = provider
|
let path = provider
|
||||||
.worker_dir(worker)
|
.worker_dir(worker)
|
||||||
.join("session/segments")
|
.join("session/segments")
|
||||||
.join(format!("{segment}.jsonl"));
|
.join(format!("{segment}.jsonl"));
|
||||||
let original = fs::read_to_string(&path).unwrap();
|
let source = fs::read_to_string(&path).unwrap();
|
||||||
assert!(original.contains(from));
|
let mut replaced = false;
|
||||||
fs::write(&path, original.replace(from, to)).unwrap();
|
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 request = request(worker, SessionDisposition::Archive);
|
||||||
let manifest =
|
let manifest =
|
||||||
commit_session_archive(&provider, &request, &provider.worker_dir(worker)).unwrap();
|
commit_session_archive(&provider, &request, &provider.worker_dir(worker)).unwrap();
|
||||||
fs::remove_dir_all(provider.worker_dir(worker)).unwrap();
|
fs::remove_dir_all(provider.worker_dir(worker)).unwrap();
|
||||||
let archive = provider.archive_dir("archive-a").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();
|
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");
|
let options = FsRuntimeStoreOptions::new(&root).with_runtime_id("runtime-a");
|
||||||
for _ in 0..2 {
|
for _ in 0..2 {
|
||||||
let error = FsRuntimeStore::migrate(&options).unwrap_err();
|
let runtime = crate::Runtime::with_fs_store(options.clone()).unwrap();
|
||||||
let RuntimeError::StoreCorrupt {
|
assert_eq!(runtime.summary().unwrap().worker_count, 1);
|
||||||
operation, path, ..
|
runtime.bind_runtime_identity("runtime-a").unwrap();
|
||||||
} = error
|
let search = runtime_api::SessionPublicSearchRequest {
|
||||||
else {
|
workspace_id: "workspace-a".into(),
|
||||||
panic!("unexpected failure: {error}");
|
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");
|
let availability = runtime
|
||||||
assert!(path.starts_with(temp.path().join(".runtime.schema-v12-staging")));
|
.session_public_search_scoped(
|
||||||
assert_eq!(tree_bytes(&root), before);
|
&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();
|
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]
|
#[test]
|
||||||
fn runtime_v5_probe_and_staging_skip_erased_originals() {
|
fn runtime_v5_probe_and_staging_skip_erased_originals() {
|
||||||
let temp = tempfile::tempdir().unwrap();
|
let temp = tempfile::tempdir().unwrap();
|
||||||
|
|||||||
@@ -58627,6 +58627,58 @@ mod tests {
|
|||||||
.unwrap();
|
.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]
|
#[tokio::test]
|
||||||
async fn api_startup_revalidates_workspace_schema_after_store_was_opened() {
|
async fn api_startup_revalidates_workspace_schema_after_store_was_opened() {
|
||||||
let temp = tempfile::tempdir().unwrap();
|
let temp = tempfile::tempdir().unwrap();
|
||||||
|
|||||||
@@ -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も同じ変換を使う。
|
旧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が一致することを要求する。
|
||||||
|
|||||||
Reference in New Issue
Block a user