fix: remove aggregate history size limit

This commit is contained in:
2026-11-06 04:51:35 +09:00
parent 00f327da4b
commit 682747e9f7
3 changed files with 101 additions and 18 deletions
+2 -12
View File
@@ -53,8 +53,6 @@ pub struct PublicMaterialLimits {
pub max_entries: usize, pub max_entries: usize,
/// Allocation ceiling for one persisted JSONL record, not an entire run. /// Allocation ceiling for one persisted JSONL record, not an entire run.
pub max_record_bytes: u64, 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 { impl Default for PublicMaterialLimits {
@@ -63,7 +61,6 @@ impl Default for PublicMaterialLimits {
max_segments: 256, max_segments: 256,
max_entries: 1_000_000, max_entries: 1_000_000,
max_record_bytes: 64 * 1024 * 1024, max_record_bytes: 64 * 1024 * 1024,
max_storage_bytes: 256 * 1024 * 1024,
} }
} }
} }
@@ -376,7 +373,6 @@ impl CommittedPublicReplay {
Layout::Fs(root) => root.clone(), Layout::Fs(root) => root.clone(),
}; };
let mut file_ends = HashMap::new(); let mut file_ends = HashMap::new();
let mut fixed_bytes = 0u64;
for entry in std::fs::read_dir(directory)? { for entry in std::fs::read_dir(directory)? {
let entry = entry?; let entry = entry?;
let path = entry.path(); let path = entry.path();
@@ -394,14 +390,8 @@ impl CommittedPublicReplay {
return Err(PublicMaterialError::ResourceLimit); return Err(PublicMaterialError::ResourceLimit);
} }
let reader = RecordReader::open(&path, limits.max_record_bytes)?; let reader = RecordReader::open(&path, limits.max_record_bytes)?;
// This also bounds retained descriptor/identity allocations during replay, // Freeze the extent, but do not cap the lifetime storage total. Records
// not just the final capture. Oversized histories fail explicitly. // are decoded individually; segment and entry limits bound replay metadata.
fixed_bytes = fixed_bytes
.checked_add(reader.end)
.ok_or(PublicMaterialError::ResourceLimit)?;
if fixed_bytes > limits.max_storage_bytes {
return Err(PublicMaterialError::ResourceLimit);
}
file_ends.insert(id, reader.end); file_ends.insert(id, reader.end);
} }
let mut chain = Vec::new(); let mut chain = Vec::new();
@@ -1802,7 +1802,7 @@ fn recording_capture_fence_blocks_publication_and_value_survives_later_append_an
} }
#[test] #[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::*; use crate::capture::*;
let temp = tempfile::tempdir().unwrap(); let temp = tempfile::tempdir().unwrap();
let root = temp.path().join("session"); 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 guard = retention.lock_live(session).unwrap();
let limits = PublicMaterialLimits { let limits = PublicMaterialLimits {
max_storage_bytes: 1, max_record_bytes: 1,
..Default::default() ..Default::default()
}; };
assert!(matches!( assert!(matches!(
@@ -2457,10 +2457,6 @@ fn recording_catalog_errors_do_not_invoke_callback_or_relax_resource_and_identit
Err(PublicMaterialError::Inconsistent) Err(PublicMaterialError::Inconsistent)
)); ));
for limits in [ for limits in [
PublicMaterialLimits {
max_storage_bytes: 1,
..Default::default()
},
PublicMaterialLimits { PublicMaterialLimits {
max_record_bytes: 1, max_record_bytes: 1,
..Default::default() ..Default::default()
+97
View File
@@ -2069,6 +2069,103 @@ mod tests {
compact 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] #[test]
fn session_migration_repairs_failed_archive_and_preserves_ingestion_receipt_on_retry() { fn session_migration_repairs_failed_archive_and_preserves_ingestion_receipt_on_retry() {
let temp = tempfile::tempdir().unwrap(); let temp = tempfile::tempdir().unwrap();