src/support/history/tests.rs
Rama main · réplica de lectura del repositorio original.
use super::{
History, MAX_BYTES, MAX_CYCLES, MAX_GROUPS, Message, QUEUE, Recorder, add, apply, read, save,
};
use serde_json::json;
use std::sync::{
Arc, Mutex,
atomic::{AtomicU64, Ordering},
mpsc,
};
use tracing_subscriber::Layer;
use tracing_subscriber::layer::SubscriberExt;
#[test]
fn history_roundtrip_groups_observations_and_bounds_restarts() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("support.json");
let mut h = History {
schema_version: 1,
..History::default()
};
for n in 0..(MAX_CYCLES + 4) {
apply(
&mut h,
Message::Begin(json!({"threads": 2, "username": "private-username"})),
);
for _ in 0..3 {
add(&mut h, "exact_lookup_rejection_v1", json!({"rejected_requests": n, "expected_owner": "absent", "stage": "master_reference_scope", "reason": "owner_unqualified"}).as_object().unwrap().clone());
}
apply(
&mut h,
Message::Complete(
json!({"bytes_downloaded": 604, "disk_bytes_written": 17}),
"partial_failure",
),
);
}
save(&path, &mut h).unwrap();
let (reopened, status) = read(&path);
assert_eq!(status, "available");
assert_eq!(reopened.cycles.len(), MAX_CYCLES);
assert_eq!(reopened.cycles_evicted, 4);
assert_eq!(
reopened.cycles.last().unwrap().diagnostics[0].observations,
3
);
assert_eq!(
reopened.cycles.last().unwrap().stats["disk_bytes_written"],
17
);
assert!(std::fs::metadata(&path).unwrap().len() <= MAX_BYTES);
assert!(
!serde_json::to_string(&reopened)
.unwrap()
.contains("private-username")
);
}
#[test]
fn interrupted_staging_keeps_previous_snapshot_and_is_bounded_on_restart() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("support.json");
let mut h = History {
schema_version: 1,
..History::default()
};
apply(&mut h, Message::Begin(json!({"threads": 2})));
save(&path, &mut h).unwrap();
std::fs::write(path.with_extension("support-tmp"), br#"{"half-written":1"#).unwrap();
let (mut reopened, status) = read(&path);
assert_eq!(status, "available");
assert!(reopened.cycles[0].completed_at.is_none());
apply(&mut reopened, Message::Begin(json!({"threads": 4})));
save(&path, &mut reopened).unwrap();
assert_eq!(read(&path).0.cycles.len(), 2);
assert!(!path.with_extension("support-tmp").exists());
}
#[test]
fn groups_and_queue_have_explicit_omission_counts() {
let mut h = History {
schema_version: 1,
..History::default()
};
apply(&mut h, Message::Begin(json!({})));
for n in 0..(MAX_GROUPS + 5) {
add(
&mut h,
"metadata_capture_ambiguity_counts_v1",
json!({"matching_children": n}).as_object().unwrap().clone(),
);
}
assert_eq!(h.cycles[0].diagnostics.len(), MAX_GROUPS);
assert_eq!(h.groups_omitted, 5);
let (sender, _receiver) = mpsc::sync_channel(QUEUE);
let recorder = Recorder {
sender,
generation: Arc::new(AtomicU64::new(0)),
dropped: Arc::new(AtomicU64::new(0)),
aliases: Arc::new(Mutex::new(std::collections::HashMap::new())),
alias_namespace: uuid::Uuid::new_v4(),
};
for _ in 0..(QUEUE + 5) {
recorder.send(Message::Observe(
"sparse_identity_state_failed",
Default::default(),
));
}
assert_eq!(recorder.dropped.load(Ordering::Relaxed), 5);
}
#[test]
fn diagnostic_layer_runs_below_log_filter_and_ignores_private_fields() {
let (sender, receiver) = mpsc::sync_channel(QUEUE);
let recorder = Recorder {
sender,
generation: Arc::new(AtomicU64::new(0)),
dropped: Arc::new(AtomicU64::new(0)),
aliases: Arc::new(Mutex::new(std::collections::HashMap::new())),
alias_namespace: uuid::Uuid::new_v4(),
};
let subscriber = tracing_subscriber::registry().with(recorder).with(
tracing_subscriber::fmt::layer()
.with_writer(std::io::sink)
.with_filter(tracing_subscriber::filter::LevelFilter::ERROR),
);
tracing::subscriber::with_default(subscriber, || {
tracing::debug!(
diagnostic = "exact_lookup_rejection_v1",
target = "paired",
stage = "master_reference_scope",
reason = "owner_unqualified",
expected_owner = "absent",
lookup_zone = "primary",
rejected_requests = 604u64,
asset_id = "private-id",
path = "/private/path",
error = "private-password",
token = "private-token",
"private free text"
);
tracing::error!(
diagnostic = "private-new-diagnostic",
error = "private-body",
"private message"
);
});
let Message::Tagged(_, message) = receiver.try_recv().unwrap() else {
panic!("expected attribution envelope")
};
let Message::Event(kind, fields) = *message else {
panic!("expected diagnostic")
};
assert_eq!(kind, "exact_lookup_rejection_v1");
assert_eq!(fields["rejected_requests"], 604);
assert_eq!(fields.len(), 6);
assert!(!serde_json::to_string(&fields).unwrap().contains("private"));
assert!(receiver.try_recv().is_err());
}
#[test]
fn untrusted_saved_strings_and_unknown_schema_never_escape() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("support.json");
let mut h = History {
schema_version: 1,
..History::default()
};
apply(
&mut h,
Message::Begin(json!({"username": "private-id", "resolution": "https://private-url"})),
);
let cycle = h.cycles.last_mut().unwrap();
cycle.build_version = "0.private-token".into();
cycle.outcome = "private-error".into();
cycle.stats = json!({"inventory_drop_library": "private-album", "bytes_downloaded": 1});
add(&mut h, "exact_lookup_rejection_v1", json!({"reason": "private-error", "target": "private-id", "rejected_requests": 4, "path": "/private/path"}).as_object().unwrap().clone());
std::fs::write(&path, serde_json::to_vec(&h).unwrap()).unwrap();
let (clean, status) = read(&path);
assert_eq!(status, "available");
assert!(!serde_json::to_string(&clean).unwrap().contains("private"));
std::fs::write(&path, b"{\"schema_version\":999,\"cycles\":[],\"cycles_evicted\":0,\"queue_dropped\":0,\"groups_omitted\":0,\"previous_history_unavailable\":false}").unwrap();
assert_eq!(read(&path).1, "unsupported_schema");
std::fs::write(&path, b"{incomplete").unwrap();
assert_eq!(read(&path).1, "invalid_or_interrupted");
std::fs::write(&path, vec![b'x'; usize::try_from(MAX_BYTES + 1).unwrap()]).unwrap();
assert_eq!(read(&path).1, "truncated_or_unreadable");
}
#[tokio::test]
async fn normal_worker_flushes_restart_history() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("support.json");
let (recorder, guard) = super::start(path.clone(), json!({"threads": 2})).unwrap();
recorder
.sender
.send(Message::Event(
"sparse_deletion_validation_failed".into(),
Default::default(),
))
.unwrap();
guard.finish().await;
let (recorder, guard) = super::start(path.clone(), json!({"threads": 4})).unwrap();
recorder
.sender
.send(Message::Complete(json!({"failed": 1}), "partial_failure"))
.unwrap();
guard.finish().await;
let (history, status) = read(&path);
assert_eq!(status, "available");
assert_eq!(history.cycles.len(), 2);
assert_eq!(history.cycles[1].stats["failed"], 1);
assert_eq!(
history.cycles[0].diagnostics[0].kind,
"sparse_deletion_validation_failed"
);
}
#[test]
fn serialized_byte_limit_rotates_complete_records() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("support.json");
let mut history = History {
schema_version: 1,
..History::default()
};
for _ in 0..MAX_CYCLES {
apply(&mut history, Message::Begin(json!({})));
// Fill fixed scalar fields with long numeric encodings, not private text.
for n in 0..MAX_GROUPS {
let fields = json!({"pages":n,"records":u64::MAX,"transferred_bytes":u64::MAX,"retained_bytes":u64::MAX,
"page_budget":u64::MAX,"record_budget":u64::MAX,"page_byte_budget":u64::MAX,"retained_byte_budget":u64::MAX,
"elapsed_secs":1.7976931348623157e308,"reason":"invalid_inventory_evidence","phase":"family_scan","subreason":"child_reference_scope_mismatch",
"eof_observed":true,"family_context":"unrelated","scope_mismatch_component":"owner","child_soft_deleted":false,
"expected_owner":"private_default","expected_owner_type":"string","supplied_owner":"other_string","supplied_owner_type":"string"});
add(
&mut history,
"legacy_inventory_failure_v2",
fields.as_object().unwrap().clone(),
);
}
}
assert!(serde_json::to_vec(&history).unwrap().len() as u64 > MAX_BYTES);
save(&path, &mut history).unwrap();
assert!(std::fs::metadata(&path).unwrap().len() <= MAX_BYTES);
assert!(history.cycles_evicted > 0);
let (reopened, status) = read(&path);
assert_eq!(status, "available");
assert_eq!(reopened.cycles_evicted, history.cycles_evicted);
}
#[cfg(unix)]
#[test]
fn staging_symlink_cannot_modify_another_file() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("support.json");
let target = dir.path().join("media");
std::fs::write(&target, b"private media").unwrap();
std::os::unix::fs::symlink(&target, path.with_extension("support-tmp")).unwrap();
let mut history = History {
schema_version: 1,
..History::default()
};
apply(&mut history, Message::Begin(json!({})));
assert!(save(&path, &mut history).is_err());
assert_eq!(std::fs::read(target).unwrap(), b"private media");
}
#[tokio::test]
async fn existing_writer_lock_preserves_prior_history() {
use fs4::fs_std::FileExt;
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("support.json");
let mut history = History {
schema_version: 1,
..History::default()
};
apply(&mut history, Message::Begin(json!({})));
save(&path, &mut history).unwrap();
let before = std::fs::read(&path).unwrap();
let lock = super::private_options()
.create(true)
.truncate(false)
.open(path.with_extension("support-lock"))
.unwrap();
lock.lock_exclusive().unwrap();
let (_, guard) = super::start(path.clone(), json!({"threads":42})).unwrap();
guard.finish().await;
assert_eq!(std::fs::read(path).unwrap(), before);
}
#[test]
fn restarting_does_not_complete_an_interrupted_prior_startup() {
let mut history = History {
schema_version: 1,
..History::default()
};
apply(&mut history, Message::Start(json!({})));
add(
&mut history,
"support_startup_v1",
json!({"phase":"starting"}).as_object().unwrap().clone(),
);
apply(&mut history, Message::Start(json!({})));
assert!(history.cycles[0].completed_at.is_none());
assert_eq!(history.cycles[0].outcome, "running");
add(
&mut history,
"support_startup_v1",
json!({"phase":"starting"}).as_object().unwrap().clone(),
);
apply(&mut history, Message::Begin(json!({})));
assert!(history.cycles[0].completed_at.is_none());
assert_eq!(history.cycles[1].outcome, "success");
assert!(history.cycles[1].completed_at.is_some());
}
#[test]
fn dropped_cycle_handoff_cannot_overwrite_prior_completed_evidence() {
let (sender, receiver) = mpsc::sync_channel(1);
let recorder = Recorder {
sender,
generation: Arc::new(AtomicU64::new(0)),
dropped: Arc::new(AtomicU64::new(0)),
aliases: Arc::new(Mutex::new(Default::default())),
alias_namespace: uuid::Uuid::new_v4(),
};
let mut history = History {
schema_version: 1,
..History::default()
};
recorder.send(Message::Begin(json!({"threads":2})));
apply(&mut history, receiver.recv().unwrap());
recorder.send(Message::Complete(json!({"downloaded":17}), "success"));
apply(&mut history, receiver.recv().unwrap());
let prior = serde_json::to_value(&history.cycles[0]).unwrap();
// A stalled writer has a full observation queue at the next cycle boundary.
recorder.send(Message::Observe(
"sparse_identity_state_failed",
Default::default(),
));
recorder.send(Message::Begin(json!({"threads":4})));
assert_eq!(recorder.dropped.load(Ordering::Relaxed), 1);
apply(&mut history, receiver.recv().unwrap());
recorder.send(Message::Complete(
json!({"downloaded":604}),
"partial_failure",
));
apply(&mut history, receiver.recv().unwrap());
assert_eq!(
serde_json::to_value(&history.cycles[0]).unwrap()["stats"],
prior["stats"]
);
assert_eq!(
history.cycles[0].completed_at,
prior["completed_at"].as_str().map(str::to_owned)
);
assert_eq!(history.cycles[0].outcome, "success");
assert_eq!(history.cycles.len(), 1);
assert_eq!(history.queue_dropped, 1);
// A later accepted handoff resumes accurate collection with a separate record.
recorder.send(Message::Begin(json!({"threads":8})));
apply(&mut history, receiver.recv().unwrap());
recorder.send(Message::Complete(json!({"downloaded":2}), "success"));
apply(&mut history, receiver.recv().unwrap());
assert_eq!(history.cycles.len(), 2);
assert_eq!(history.cycles[1].stats["downloaded"], 2);
}
#[cfg(unix)]
#[test]
fn failed_history_rename_never_uses_an_unowned_copy_fallback() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("support.json");
std::fs::create_dir(&path).unwrap();
let media = dir.path().join("media.jpg");
std::fs::write(&media, b"private media").unwrap();
let sibling = path.with_extension(format!("json.kei-xdev-tmp-{}", std::process::id()));
std::os::unix::fs::symlink(&media, &sibling).unwrap();
let mut history = History {
schema_version: 1,
..History::default()
};
apply(&mut history, Message::Start(json!({})));
assert!(save(&path, &mut history).is_err());
assert_eq!(std::fs::read(media).unwrap(), b"private media");
assert!(
std::fs::symlink_metadata(sibling)
.unwrap()
.file_type()
.is_symlink()
);
assert!(path.is_dir());
}
#[test]
fn late_failures_and_summary_survive_sampled_details_and_report_counts() {
let mut h = History {
schema_version: 1,
..History::default()
};
apply(&mut h, Message::Begin(json!({})));
for n in 0..200 {
let fields = json!({"phase":"scan","reason":"candidate","scan_index":n});
for _ in 0..2 {
add(
&mut h,
"support_import_v1",
fields.as_object().unwrap().clone(),
);
}
}
add(
&mut h,
"support_import_v1",
json!({"phase":"adoption","reason":"strict_refusal"})
.as_object()
.unwrap()
.clone(),
);
add(
&mut h,
"support_import_v1",
json!({"phase":"complete","total":200,"strict_refused":1})
.as_object()
.unwrap()
.clone(),
);
let c = &h.cycles[0];
assert_eq!(c.diagnostics.len(), super::MAX_DETAIL_GROUPS + 2);
assert_eq!(c.observations_total, Some(402));
assert_eq!(c.observations_shown, Some(130));
assert_eq!(c.observations_omitted, Some(272));
assert!(
c.diagnostics
.iter()
.any(|d| d.fields.get("strict_refused") == Some(&json!(1)))
);
let tmp = tempfile::tempdir().unwrap();
let path = tmp.path().join("support.json");
save(&path, &mut h).unwrap();
let (clean, status) = read(&path);
assert_eq!(status, "available");
assert_eq!(clean.cycles[0].observations_omitted, Some(272));
// Exhaust failure groups too: late summary displaces an earlier failure.
for n in 0..MAX_GROUPS {
add(
&mut h,
"exact_lookup_rejection_v1",
json!({"rejected_requests":n}).as_object().unwrap().clone(),
);
}
add(
&mut h,
"support_import_v1",
json!({"phase":"complete","total":201,"strict_refused":2})
.as_object()
.unwrap()
.clone(),
);
let c = &h.cycles[0];
assert_eq!(c.diagnostics.len(), MAX_GROUPS);
assert_eq!(
c.observations_total,
c.observations_shown
.zip(c.observations_omitted)
.map(|(a, b)| a + b)
);
assert!(
c.diagnostics
.iter()
.any(|d| d.fields.get("total") == Some(&json!(201)))
);
}
#[test]
fn older_history_observation_accounting_is_explicitly_unavailable() {
let mut h = History {
schema_version: 1,
..History::default()
};
apply(&mut h, Message::Begin(json!({})));
let mut saved = serde_json::to_value(&h).unwrap();
let c = saved["cycles"][0].as_object_mut().unwrap();
for field in [
"observations_total",
"observations_shown",
"observations_omitted",
] {
c.remove(field);
}
let tmp = tempfile::tempdir().unwrap();
let path = tmp.path().join("support.json");
std::fs::write(&path, serde_json::to_vec(&saved).unwrap()).unwrap();
let (read, status) = read(&path);
assert_eq!(status, "available");
assert!(read.cycles[0].observations_total.is_none());
assert!(read.cycles[0].observations_shown.is_none());
assert!(read.cycles[0].observations_omitted.is_none());
}
#[test]
fn issue956_pair_reference_reasons_survive_history_reopen_without_private_fields() {
let directory = tempfile::tempdir().unwrap();
let path = directory.path().join("history.json");
let mut history = History {
schema_version: 1,
..History::default()
};
apply(&mut history, Message::Begin(json!({})));
let reasons = [
"master_reference_missing",
"master_reference_malformed",
"child_master_mismatch",
"record_provider_error",
];
for reason in reasons {
add(
&mut history,
"exact_lookup_rejection_v1",
json!({
"reason":reason,"target":"paired","stage":"pairing","rejected_requests":1,
"recordName":"PRIVATE_RECORD_CANARY"
})
.as_object()
.unwrap()
.clone(),
);
}
save(&path, &mut history).unwrap();
let (reopened, status) = read(&path);
assert_eq!(status, "available");
let saved: Vec<_> = reopened.cycles[0]
.diagnostics
.iter()
.map(|observation| observation.fields["reason"].as_str().unwrap())
.collect();
assert_eq!(saved, reasons);
assert!(
!serde_json::to_string(&reopened)
.unwrap()
.contains("PRIVATE_RECORD_CANARY")
);
}
#[test]
fn missing_reference_history_survives_restart_and_revalidates_export() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("support.json");
let mut history = History {
schema_version: 1,
..History::default()
};
apply(&mut history, Message::Begin(json!({})));
let fields = json!({"record_history":"replacement", "prior_reference":"unavailable",
"sparse_marker":"not_recorded", "linkage":"not_recorded", "lookup_outcome":"scope_unavailable", "lookup_family":"not_established", "lookup_scope":"child_unavailable", "lookup_shape":"provider_error"});
add(
&mut history,
"support_missing_reference_v1",
fields.as_object().unwrap().clone(),
);
apply(
&mut history,
Message::Complete(json!({}), "partial_failure"),
);
save(&path, &mut history).unwrap();
let (reopened, status) = read(&path);
assert_eq!(status, "available");
let diagnostic = &reopened.cycles[0].diagnostics[0];
assert_eq!(diagnostic.kind, "support_missing_reference_v1");
assert_eq!(serde_json::Value::Object(diagnostic.fields.clone()), fields);
let mut poisoned: serde_json::Value =
serde_json::from_slice(&std::fs::read(&path).unwrap()).unwrap();
for key in ["lookup_family", "lookup_scope", "lookup_shape"] {
poisoned["cycles"][0]["diagnostics"][0]["fields"][key] = json!("private-provider-id");
}
std::fs::write(&path, serde_json::to_vec(&poisoned).unwrap()).unwrap();
assert!(
!serde_json::to_string(&read(&path).0)
.unwrap()
.contains("private-provider-id")
);
}
#[test]
fn legacy_receipt_prerequisites_roundtrip_only_fixed_labels() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("support.json");
let mut history = History {
schema_version: 1,
..History::default()
};
apply(&mut history, Message::Begin(json!({})));
let labels = [
"child_not_ready",
"rendition_missing",
"path_receipt_missing",
"receipt_binding_mismatch",
"source_receipt_missing",
"source_receipt_invalid",
"receipt_state_changed",
"diagnostic_unavailable",
];
for label in labels {
add(&mut history, "legacy_receipt_prerequisite_v1", json!({"prerequisite":label,"id":"private-id","path":"/private/path","checksum":"private-checksum","error":"private-error"}).as_object().unwrap().clone());
}
add(
&mut history,
"legacy_receipt_prerequisite_v1",
json!({"prerequisite":"private-new-label"})
.as_object()
.unwrap()
.clone(),
);
save(&path, &mut history).unwrap();
let (reopened, status) = read(&path);
assert_eq!(status, "available");
let observed: Vec<_> = reopened.cycles[0]
.diagnostics
.iter()
.filter_map(|d| {
d.fields
.get("prerequisite")
.and_then(serde_json::Value::as_str)
})
.collect();
assert_eq!(observed, labels);
assert!(
!serde_json::to_string(&reopened)
.unwrap()
.contains("private")
);
}
#[test]
fn issue956_lookup_record_statuses_survive_reopen_and_reject_untrusted_values() {
let directory = tempfile::tempdir().unwrap();
let path = directory.path().join("history.json");
let mut history = History {
schema_version: 1,
..History::default()
};
apply(&mut history, Message::Begin(json!({})));
for (master, child) in [("not_found", "returned"), ("returned", "access_denied")] {
add(
&mut history,
"exact_lookup_rejection_v1",
json!({
"target":"paired","stage":"record_error","reason":"record_provider_error",
"master_status":master,"child_status":child,"rejected_requests":3,
"serverErrorCode":"PRIVATE_ERROR_CANARY","recordName":"PRIVATE_RECORD_CANARY"
})
.as_object()
.unwrap()
.clone(),
);
}
save(&path, &mut history).unwrap();
let (reopened, status) = read(&path);
assert_eq!(status, "available");
let events = &reopened.cycles[0].diagnostics;
assert_eq!(events.len(), 2);
assert_eq!(events[0].fields["master_status"], "not_found");
assert_eq!(events[0].fields["child_status"], "returned");
assert_eq!(events[1].fields["master_status"], "returned");
assert_eq!(events[1].fields["child_status"], "access_denied");
assert!(
!serde_json::to_string(&reopened)
.unwrap()
.contains("PRIVATE_")
);
let mut untrusted = serde_json::to_value(&reopened).unwrap();
untrusted["cycles"][0]["diagnostics"][0]["fields"]["master_status"] =
json!("PRIVATE_STATUS_CANARY");
untrusted["cycles"][0]["diagnostics"][0]["fields"]["child_status"] = json!(401);
std::fs::write(&path, serde_json::to_vec(&untrusted).unwrap()).unwrap();
let (clean, status) = read(&path);
assert_eq!(status, "available");
assert!(
!clean.cycles[0].diagnostics[0]
.fields
.contains_key("master_status")
);
assert!(
!clean.cycles[0].diagnostics[0]
.fields
.contains_key("child_status")
);
assert_eq!(
clean.cycles[0].diagnostics[1].fields["child_status"],
"access_denied"
);
assert!(!serde_json::to_string(&clean).unwrap().contains("PRIVATE_"));
}
#[test]
fn issue956_lookup_statuses_cross_the_production_tracing_recorder_and_reopen() {
let statuses = [
"not_requested",
"omitted",
"returned",
"tombstone",
"not_found",
"access_denied",
"authentication_failed",
"zone_not_found",
"retry_later",
"throttled",
"server_error",
"other_error",
"malformed_error",
];
let _parallel_dispatch = tracing::Dispatch::new(tracing::subscriber::NoSubscriber::default());
let (recorder, drain) = super::test_layer();
let subscriber = tracing_subscriber::registry().with(recorder);
tracing::subscriber::with_default(subscriber, || {
for status in statuses {
tracing::info!(
diagnostic = "exact_lookup_rejection_v1",
target = "paired",
master_status = status,
child_status = status,
rejected_requests = 1u64,
record_name = "PRIVATE_RECORD_CANARY",
error = "PRIVATE_ERROR_CANARY",
"PRIVATE_MESSAGE_CANARY"
);
}
tracing::info!(
diagnostic = "exact_lookup_rejection_v1",
master_status = "PRIVATE_STATUS_CANARY",
child_status = 401u64,
rejected_requests = 2u64,
"untrusted categories"
);
});
let events = drain();
assert_eq!(events.len(), statuses.len() + 1);
for (event, status) in events.iter().zip(statuses) {
assert_eq!(event["fields"]["master_status"], status);
assert_eq!(event["fields"]["child_status"], status);
}
let last = events.last().unwrap()["fields"].as_object().unwrap();
assert_eq!(last.len(), 1);
assert_eq!(last["rejected_requests"], 2);
assert!(!serde_json::to_string(&events).unwrap().contains("PRIVATE_"));
let directory = tempfile::tempdir().unwrap();
let path = directory.path().join("history.json");
let mut history = History {
schema_version: 1,
..History::default()
};
apply(&mut history, Message::Begin(json!({})));
for event in events {
add(
&mut history,
event["kind"].as_str().unwrap(),
event["fields"].as_object().unwrap().clone(),
);
}
save(&path, &mut history).unwrap();
let (reopened, status) = read(&path);
assert_eq!(status, "available");
for (event, status) in reopened.cycles[0].diagnostics.iter().zip(statuses) {
assert_eq!(event.fields["master_status"], status);
assert_eq!(event.fields["child_status"], status);
}
}
#[test]
fn issue956_child_reference_status_crosses_tracing_and_reopen_without_private_values() {
use tracing_subscriber::layer::SubscriberExt;
let labels = [
"not_requested",
"omitted",
"unavailable",
"invalid_child",
"missing",
"malformed",
"sparse",
"matches_requested",
"different",
];
let _parallel_dispatch = tracing::Dispatch::new(tracing::subscriber::NoSubscriber::default());
let (recorder, drain) = super::test_layer();
tracing::subscriber::with_default(tracing_subscriber::registry().with(recorder), || {
for label in labels {
tracing::info!(
diagnostic = "exact_lookup_rejection_v1",
child_reference_status = label,
record_name = "PRIVATE_RECORD_CANARY",
"PRIVATE_MESSAGE_CANARY"
);
}
});
let events = drain();
assert_eq!(events.len(), labels.len());
let directory = tempfile::tempdir().unwrap();
let path = directory.path().join("history.json");
let mut history = History {
schema_version: 1,
..History::default()
};
apply(&mut history, Message::Begin(json!({})));
for (event, label) in events.iter().zip(labels) {
assert_eq!(event["fields"]["child_reference_status"], label);
add(
&mut history,
"exact_lookup_rejection_v1",
event["fields"].as_object().unwrap().clone(),
);
}
save(&path, &mut history).unwrap();
let (reopened, status) = read(&path);
assert_eq!(status, "available");
assert_eq!(reopened.cycles[0].diagnostics.len(), labels.len());
for (event, label) in reopened.cycles[0].diagnostics.iter().zip(labels) {
assert_eq!(event.fields["child_reference_status"], label);
}
assert!(
!serde_json::to_string(&reopened)
.unwrap()
.contains("PRIVATE_")
);
let mut untrusted = serde_json::to_value(&reopened).unwrap();
for (index, invalid) in [json!("PRIVATE_REFERENCE_CANARY"), json!(42)]
.into_iter()
.enumerate()
{
untrusted["cycles"][0]["diagnostics"][index]["fields"]["child_reference_status"] = invalid;
}
std::fs::write(&path, serde_json::to_vec(&untrusted).unwrap()).unwrap();
let (clean, status) = read(&path);
assert_eq!(status, "available");
assert!(
clean.cycles[0].diagnostics[..2]
.iter()
.all(|event| !event.fields.contains_key("child_reference_status"))
);
assert_eq!(
clean.cycles[0].diagnostics[2].fields["child_reference_status"],
labels[2]
);
assert!(!serde_json::to_string(&clean).unwrap().contains("PRIVATE_"));
}
#[test]
fn storage_reuse_counters_saturate_and_preserve_last_stop() {
let mut reuse = super::StorageReuse::default();
reuse.inbox.new = u64::MAX;
reuse.inbox.exact = u64::MAX;
reuse.inbox.equivalent = u64::MAX;
reuse.inbox.capacity_rejected = u64::MAX;
reuse.inbox.single_item_limit = u64::MAX;
reuse.record(
super::StorageComponent::Inbox,
super::StorageOutcome::New,
Some((7, 11)),
);
reuse.record(
super::StorageComponent::Inbox,
super::StorageOutcome::Exact,
None,
);
reuse.record(
super::StorageComponent::Inbox,
super::StorageOutcome::Equivalent,
None,
);
reuse.record(
super::StorageComponent::Inbox,
super::StorageOutcome::CapacityRejected,
Some((8, 11)),
);
assert!(reuse.inbox.observed);
assert!(!reuse.catalogue.observed);
assert_eq!(reuse.inbox.new, u64::MAX);
assert_eq!(reuse.inbox.exact, u64::MAX);
assert_eq!(reuse.inbox.equivalent, u64::MAX);
assert_eq!(reuse.inbox.capacity_rejected, u64::MAX);
assert_eq!(reuse.inbox.last_reason, super::ReuseReason::Capacity);
assert_eq!(reuse.inbox.last_observed_used, Some(8));
assert_eq!(reuse.inbox.last_observed_limit, Some(11));
reuse.record(
super::StorageComponent::Selection,
super::StorageOutcome::SingleItemLimit,
None,
);
assert_eq!(reuse.selection.single_item_limit, 1);
assert_eq!(
reuse.selection.last_reason,
super::ReuseReason::SingleItemLimit
);
assert_eq!(reuse.selection.last_observed_used, None);
reuse.record(
super::StorageComponent::Selection,
super::StorageOutcome::Exact,
Some((3, 9)),
);
assert_eq!(
reuse.selection.last_reason,
super::ReuseReason::SingleItemLimit
);
assert_eq!(reuse.selection.last_observed_used, Some(3));
assert_eq!(reuse.selection.last_observed_limit, Some(9));
}
#[test]
fn migration_evidence_replaces_counts_and_budgets_with_the_last_result() {
let mut history = History {
schema_version: 1,
..History::default()
};
apply(&mut history, Message::Begin(json!({})));
super::apply_migration(
&mut history,
super::MigrationEvent::Committed(super::CompactionSummary {
examined_raw_pages: 10,
logical_charge_before: Some(100),
logical_charge_after: Some(40),
..Default::default()
}),
);
super::apply_migration(
&mut history,
super::MigrationEvent::Committed(super::CompactionSummary {
examined_raw_pages: 2,
logical_charge_before: Some(9),
logical_charge_after: Some(4),
..Default::default()
}),
);
let evidence = &history.cycles[0].transport_compaction;
assert_eq!(evidence.committed, 2);
assert_eq!(evidence.examined_raw_pages, 2);
assert_eq!(evidence.logical_charge_before, Some(9));
assert_eq!(evidence.logical_charge_after, Some(4));
assert_eq!(evidence.committed_reclaim, Some(5));
super::apply_migration(
&mut history,
super::MigrationEvent::Failed(super::MigrationFailure::MigrationFailed),
);
let evidence = &history.cycles[0].transport_compaction;
assert_eq!(evidence.failed, 1);
assert_eq!(evidence.reason, super::CompactionReason::MigrationFailed);
assert_eq!(evidence.examined_raw_pages, 0);
assert_eq!(evidence.logical_charge_before, None);
assert_eq!(evidence.logical_charge_after, None);
assert_eq!(evidence.committed_reclaim, Some(0));
assert_eq!(evidence.committed, 2);
}
#[test]
fn older_storage_and_migration_fields_default_without_rejecting_history() {
let value = json!({
"schema_version": 1,
"cycles_evicted": 0,
"queue_dropped": 0,
"groups_omitted": 0,
"previous_history_unavailable": false,
"cycles": [{
"id": uuid::Uuid::new_v4().to_string(),
"started_at": "2026-10-22T00:00:00Z",
"completed_at": null,
"outcome": "running",
"build_version": "0.25.1",
"configuration": {},
"stats": {},
"diagnostics": [],
"storage_reuse": {
"inbox": {"new":1,"exact":0,"equivalent":0,"capacity_rejected":0},
"selection": {"new":0,"exact":0,"equivalent":0,"capacity_rejected":0},
"work": {"new":0,"exact":0,"equivalent":0,"capacity_rejected":0}
},
"transport_compaction": {
"committed": 1,
"failed": 0,
"reason": "committed",
"examined_raw_pages": 2
}
}]
});
let history: History = serde_json::from_value(value).unwrap();
assert_eq!(history.cycles[0].storage_reuse.inbox.new, 1);
assert_eq!(history.cycles[0].storage_reuse.catalogue.new, 0);
assert_eq!(history.cycles[0].storage_reuse.inbox.single_item_limit, 0);
assert_eq!(
history.cycles[0].transport_compaction.logical_charge_before,
None
);
}
#[test]
fn unknown_fixed_enum_labels_do_not_escape_from_stored_history() {
let directory = tempfile::tempdir().unwrap();
let path = directory.path().join("support.json");
let cycle = json!({
"id": uuid::Uuid::new_v4().to_string(),
"started_at": "2026-10-22T00:00:00Z",
"completed_at": null,
"outcome": "running",
"build_version": "0.25.1",
"configuration": {},
"stats": {},
"diagnostics": []
});
let mut value = json!({
"schema_version": 1,
"cycles_evicted": 0,
"queue_dropped": 0,
"groups_omitted": 0,
"previous_history_unavailable": false,
"cycles": [cycle]
});
value["cycles"][0]["storage_reuse"] = json!({"inbox":{"last_reason":"PRIVATE_ENUM_CANARY"}});
std::fs::write(&path, serde_json::to_vec(&value).unwrap()).unwrap();
let (clean, status) = read(&path);
assert_eq!(status, "invalid_or_interrupted");
assert!(
!serde_json::to_string(&clean)
.unwrap()
.contains("PRIVATE_ENUM_CANARY")
);
value["cycles"][0]["storage_reuse"] = json!({});
value["cycles"][0]["transport_compaction"] = json!({"reason":"PRIVATE_ENUM_CANARY"});
std::fs::write(&path, serde_json::to_vec(&value).unwrap()).unwrap();
let (clean, status) = read(&path);
assert_eq!(status, "invalid_or_interrupted");
assert!(
!serde_json::to_string(&clean)
.unwrap()
.contains("PRIVATE_ENUM_CANARY")
);
}