From 682747e9f73300379a41b4e3bbdd8219d97bb174 Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 6 Nov 2026 04:51:35 +0900 Subject: [PATCH] fix: remove aggregate history size limit --- crates/session-store/src/committed_replay.rs | 14 +-- .../src/public_material/tests.rs | 8 +- crates/worker-runtime/src/retention.rs | 97 +++++++++++++++++++ 3 files changed, 101 insertions(+), 18 deletions(-) diff --git a/crates/session-store/src/committed_replay.rs b/crates/session-store/src/committed_replay.rs index 051096cca..5e0760e49 100644 --- a/crates/session-store/src/committed_replay.rs +++ b/crates/session-store/src/committed_replay.rs @@ -53,8 +53,6 @@ pub struct PublicMaterialLimits { pub max_entries: usize, /// Allocation ceiling for one persisted JSONL record, not an entire run. pub max_record_bytes: u64, - /// Aggregate fixed storage extent, bounding replay metadata as well as body decoding. - pub max_storage_bytes: u64, } impl Default for PublicMaterialLimits { @@ -63,7 +61,6 @@ impl Default for PublicMaterialLimits { max_segments: 256, max_entries: 1_000_000, max_record_bytes: 64 * 1024 * 1024, - max_storage_bytes: 256 * 1024 * 1024, } } } @@ -376,7 +373,6 @@ impl CommittedPublicReplay { Layout::Fs(root) => root.clone(), }; let mut file_ends = HashMap::new(); - let mut fixed_bytes = 0u64; for entry in std::fs::read_dir(directory)? { let entry = entry?; let path = entry.path(); @@ -394,14 +390,8 @@ impl CommittedPublicReplay { return Err(PublicMaterialError::ResourceLimit); } let reader = RecordReader::open(&path, limits.max_record_bytes)?; - // This also bounds retained descriptor/identity allocations during replay, - // not just the final capture. Oversized histories fail explicitly. - fixed_bytes = fixed_bytes - .checked_add(reader.end) - .ok_or(PublicMaterialError::ResourceLimit)?; - if fixed_bytes > limits.max_storage_bytes { - return Err(PublicMaterialError::ResourceLimit); - } + // Freeze the extent, but do not cap the lifetime storage total. Records + // are decoded individually; segment and entry limits bound replay metadata. file_ends.insert(id, reader.end); } let mut chain = Vec::new(); diff --git a/crates/session-store/src/public_material/tests.rs b/crates/session-store/src/public_material/tests.rs index b7f4a8e7c..849d28599 100644 --- a/crates/session-store/src/public_material/tests.rs +++ b/crates/session-store/src/public_material/tests.rs @@ -1802,7 +1802,7 @@ fn recording_capture_fence_blocks_publication_and_value_survives_later_append_an } #[test] -fn recording_capture_rejects_foreign_guard_and_storage_or_body_allocation_excess() { +fn recording_capture_rejects_foreign_guard_and_record_or_body_allocation_excess() { use crate::capture::*; let temp = tempfile::tempdir().unwrap(); let root = temp.path().join("session"); @@ -1834,7 +1834,7 @@ fn recording_capture_rejects_foreign_guard_and_storage_or_body_allocation_excess )); let guard = retention.lock_live(session).unwrap(); let limits = PublicMaterialLimits { - max_storage_bytes: 1, + max_record_bytes: 1, ..Default::default() }; assert!(matches!( @@ -2457,10 +2457,6 @@ fn recording_catalog_errors_do_not_invoke_callback_or_relax_resource_and_identit Err(PublicMaterialError::Inconsistent) )); for limits in [ - PublicMaterialLimits { - max_storage_bytes: 1, - ..Default::default() - }, PublicMaterialLimits { max_record_bytes: 1, ..Default::default() diff --git a/crates/worker-runtime/src/retention.rs b/crates/worker-runtime/src/retention.rs index c11802fd4..3990ea30a 100644 --- a/crates/worker-runtime/src/retention.rs +++ b/crates/worker-runtime/src/retention.rs @@ -2069,6 +2069,103 @@ mod tests { compact } + #[test] + fn archive_removal_preserves_history_larger_than_256_mib_and_replays_receipt() { + use session_store::{ + LogEntry, LoggedContentPart, LoggedHistoryEntry, LoggedItem, LoggedRole, + LoggedSessionHistoryEntryId, LoggedSessionHistoryMetadata, LoggedSessionHistoryOrigin, + Store, + }; + + let temp = tempfile::tempdir().unwrap(); + let worker = WorkerId::now_v7(); + let worker_dir = temp.path().join("workers").join(worker.to_string()); + write_json( + &worker_dir.join("worker.json"), + &serde_json::json!({"workspace_id": "workspace-a"}), + ); + let session = session_store::new_session_id(); + let segment = session_store::new_segment_id(); + let store = session_store::WorkerSessionStore::new(worker_dir.join("session")).unwrap(); + store + .create_segment( + session, + segment, + &[LogEntry::AnnotatedSegmentStart { + ts: 1, + session_id: session, + system_prompt: None, + config: Default::default(), + history: Vec::new(), + forked_from: None, + compacted_from: None, + }], + ) + .unwrap(); + // Each record is small enough to decode independently. Only the lifetime + // total exceeds the former ceiling; no individual resource limit is relaxed. + let text = "x".repeat(1024 * 1024); + for index in 0..257 { + store + .append( + session, + segment, + &LogEntry::AnnotatedUserInput { + ts: index + 2, + segments: Vec::new(), + extensions: Vec::new(), + history: vec![LoggedHistoryEntry { + item: LoggedItem::Message { + role: LoggedRole::User, + content: vec![LoggedContentPart::Text { text: text.clone() }], + }, + metadata: LoggedSessionHistoryMetadata { + input_metadata: None, + entry_id: LoggedSessionHistoryEntryId(format!("input-{index}")), + origin: LoggedSessionHistoryOrigin::LocalHumanInput, + derivation: None, + }, + }], + }, + ) + .unwrap(); + } + let mut metadata = session_store::WorkerMetadata::new( + worker.to_string(), + Some(session_store::WorkerActiveSegmentRef::active_segment( + session, segment, + )), + ); + metadata.workspace_id = Some("workspace-a".into()); + write_json(&worker_dir.join("metadata.json"), &metadata); + let segment_path = format!("session/segments/{segment}.jsonl"); + let source_bytes = fs::metadata(worker_dir.join(&segment_path)).unwrap().len(); + assert!(source_bytes > 256 * 1024 * 1024); + + let provider = FsWorkerRetentionProvider::new(temp.path()); + let request = request(worker, SessionDisposition::Archive); + let result = provider.execute(&request).unwrap(); + assert!(result.source_removed); + assert!(!worker_dir.exists()); + let archive = provider.archive_dir("archive-a").unwrap(); + assert_eq!( + fs::metadata(archive.join(&segment_path)).unwrap().len(), + source_bytes + ); + let reader = session_store::public_material::SessionPublicMaterialReader::open_worker( + &archive.join("session"), + segment, + Default::default(), + ) + .unwrap(); + assert_eq!(reader.session_id(), session); + assert_eq!(reader.descriptors().len(), 257); + let retry = provider.execute(&request).unwrap(); + assert!(retry.source_removed); + assert_eq!(retry.archive, result.archive); + assert_eq!(retry.operation_id, result.operation_id); + } + #[test] fn session_migration_repairs_failed_archive_and_preserves_ingestion_receipt_on_retry() { let temp = tempfile::tempdir().unwrap();