diff --git a/Cargo.lock b/Cargo.lock index 7282d306d..81a5e9b70 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -10221,9 +10221,9 @@ dependencies = [ [[package]] name = "zstd-sys" -version = "2.0.16+zstd.1.5.7" +version = "2.1.2+zstd.1.5.7" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "91e19ebc2adc8f83e43039e79776e3fda8ca919132d68a1fed6a5faca2683748" +checksum = "481ada9c543b2323cbb7bbdbb5249177fbf0188ed959ed2c15e4b22421c84ed3" dependencies = [ "cc", "pkg-config", diff --git a/crates/omnigraph-gqt/src/dst_runner.rs b/crates/omnigraph-gqt/src/dst_runner.rs index c31d2dcc9..de6293cf9 100644 --- a/crates/omnigraph-gqt/src/dst_runner.rs +++ b/crates/omnigraph-gqt/src/dst_runner.rs @@ -410,7 +410,7 @@ struct WorkerReport { observations: Vec, evidence: Vec, /// `--measure` bytes and request logs per step; outside the replay - /// comparison (`comparable`). + /// comparison (`comparable`, `replay_difference`). #[serde(default, skip_serializing_if = "Vec::is_empty")] measurements: Vec, } @@ -418,11 +418,8 @@ struct WorkerReport { /// The report minus its measurements: Lance stamps manifests with wall-clock /// time, so byte counts may differ between two runs of one seed while every /// request count stays equal. -fn comparable(report: &WorkerReport) -> Result, String> { - if report.measurements.is_empty() { - return json(report); - } - json(&WorkerReport { +fn comparable(report: &WorkerReport) -> WorkerReport { + WorkerReport { code: report.code.clone(), phase: report.phase.clone(), input_digest: report.input_digest.clone(), @@ -430,7 +427,71 @@ fn comparable(report: &WorkerReport) -> Result, String> { observations: report.observations.clone(), evidence: report.evidence.clone(), measurements: Vec::new(), - }) + } +} + +/// The first compared field whose JSON text differs between two reports of one +/// seed, for the `replay_mismatch` line: both values around their first differing +/// char, for a list at the first differing index or as the two lengths. +fn replay_difference(first: &WorkerReport, second: &WorkerReport) -> Option { + const SHOWN_CHARS: usize = 240; + const LEAD_CHARS: usize = 40; + fn text(value: &T) -> String { + serde_json::to_string(value).unwrap_or_else(|error| error.to_string()) + } + fn window(text: &str, start: usize) -> String { + let chars: Vec = text.chars().collect(); + let mut end = (start + SHOWN_CHARS).min(chars.len()); + if end < chars.len() && chars[end - 1] == '\\' { + end += 1; + } + let shown: String = chars[start..end].iter().collect(); + let head = if start > 0 { "…" } else { "" }; + if end < chars.len() { + format!("{head}{shown}… ({} chars)", chars.len()) + } else { + format!("{head}{shown}") + } + } + fn side_by_side(first: &str, second: &str) -> String { + let shared = first + .chars() + .zip(second.chars()) + .take_while(|(a, b)| a == b) + .count(); + let start = shared.saturating_sub(LEAD_CHARS); + format!( + "run 0 {} | run 1 {}", + window(first, start), + window(second, start) + ) + } + fn scalar(field: &str, first: &T, second: &T) -> Option { + let (first, second) = (text(first), text(second)); + (first != second).then(|| format!("{field} differs: {}", side_by_side(&first, &second))) + } + fn list(field: &str, first: &[T], second: &[T]) -> Option { + let texts = first.iter().map(text).zip(second.iter().map(text)); + if let Some((index, (a, b))) = texts.enumerate().find(|(_, (a, b))| a != b) { + return Some(format!( + "{field}[{index}] differs: {}", + side_by_side(&a, &b) + )); + } + (first.len() != second.len()).then(|| { + format!( + "{field} has {} entries in run 0 and {} in run 1; the shorter is a prefix of the longer", + first.len(), + second.len() + ) + }) + } + scalar("code", &first.code, &second.code) + .or_else(|| scalar("phase", &first.phase, &second.phase)) + .or_else(|| scalar("input_digest", &first.input_digest, &second.input_digest)) + .or_else(|| scalar("result", &first.result, &second.result)) + .or_else(|| list("observations", &first.observations, &second.observations)) + .or_else(|| list("evidence", &first.evidence, &second.evidence)) } #[derive(Debug, Serialize, Deserialize)] @@ -1728,7 +1789,7 @@ fn run_invocation( env )); } - reports.push(comparable(report)?); + reports.push(comparable(report)); } Err(error) => { failures.push(error.clone()); @@ -1747,13 +1808,12 @@ fn run_invocation( }); } if reports.len() == 2 { - if reports[0] != reports[1] { - failures.push(format!( - "replay_mismatch: {} seed={seed:?}; both reports retained", + match replay_difference(&reports[0], &reports[1]) { + Some(difference) => failures.push(format!( + "replay_mismatch: {} seed={seed:?}; {difference}; both reports retained", env - )); - } else { - println!("DST environment={} seed={seed:?} replay matched", env); + )), + None => println!("DST environment={} seed={seed:?} replay matched", env), } } } @@ -2384,10 +2444,10 @@ fn replay_attempts( return Err(failures.join("\n")); } }; - if comparable(&report)? != comparable(prior)? { + if let Some(difference) = replay_difference(prior, &report) { failures.push(format!( - "replay_mismatch: {} seed={:?}", - attempt.environment, attempt.seed + "replay_mismatch: {} seed={:?}; {difference} (run 0 the recorded replay={} attempt, run 1 its rerun)", + attempt.environment, attempt.seed, attempt.replay )); } if let Err(error) = &report.result { @@ -2672,8 +2732,8 @@ mod action_tests { .unwrap() } - let expected = super::comparable(&report(&[0, 2, 1, 3], "alice").await).unwrap(); - let interleaved = super::comparable(&report(&[2, 0, 3, 1], "alice").await).unwrap(); + let expected = super::comparable(&report(&[0, 2, 1, 3], "alice").await); + let interleaved = super::comparable(&report(&[2, 0, 3, 1], "alice").await); assert!( expected == interleaved, "session completion order must not change replay evidence" @@ -2685,7 +2745,7 @@ mod action_tests { ] { assert_ne!( expected, - super::comparable(&report(order, row).await).unwrap(), + super::comparable(&report(order, row).await), "per-session order, completeness and rows remain replay evidence" ); } @@ -2715,11 +2775,11 @@ mod action_tests { log: Vec::new(), wall_ms: 5, }; - let expected = super::comparable(&report(&sessions, &block).await).unwrap(); + let expected = super::comparable(&report(&sessions, &block).await); block.unattributed = 3; block.wall_ms = 9; block.log.push(serde_json::json!({"wall_ms": 7})); - let measured = super::comparable(&report(&sessions, &block).await).unwrap(); + let measured = super::comparable(&report(&sessions, &block).await); assert!( expected == measured, "background I/O and timing measurements must not change replay equality" @@ -2728,20 +2788,235 @@ mod action_tests { block.stuck_at = Some(0); assert_ne!( expected, - super::comparable(&report(&sessions, &block).await).unwrap() + super::comparable(&report(&sessions, &block).await) ); block.stuck_at = None; block.failure = Some("session starved".into()); assert_ne!( expected, - super::comparable(&report(&sessions, &block).await).unwrap() + super::comparable(&report(&sessions, &block).await) ); block.failure = None; sessions[0]["outcome"] = "failed".into(); sessions[0]["message"] = "query failed".into(); assert_ne!( expected, - super::comparable(&report(&sessions, &block).await).unwrap() + super::comparable(&report(&sessions, &block).await) + ); + } + + fn compared( + result: Result<(), String>, + observations: &[&str], + evidence: &[serde_json::Value], + ) -> super::WorkerReport { + super::WorkerReport { + code: "ok".into(), + phase: "run".into(), + input_digest: "digest".into(), + result, + observations: observations.iter().map(|text| text.to_string()).collect(), + evidence: evidence.to_vec(), + measurements: Vec::new(), + } + } + + #[test] + fn replay_difference_names_the_first_differing_field_and_entry() { + let steps = [ + serde_json::json!({"step": 1}), + serde_json::json!({"step": 2}), + ]; + let base = compared(Ok(()), &["a", "b"], &steps); + assert_eq!(super::replay_difference(&base, &base), None); + + let mut measured = compared(Ok(()), &["a", "b"], &steps); + measured.measurements.push(serde_json::json!({"bytes": 7})); + assert_eq!(super::replay_difference(&base, &measured), None); + + let changed = [ + serde_json::json!({"step": 1}), + serde_json::json!({"step": 9}), + ]; + let failed = compared(Err("rows differ".into()), &["a", "b"], &changed); + assert_eq!( + super::replay_difference(&base, &failed).as_deref(), + Some(r#"result differs: run 0 {"Ok":null} | run 1 {"Err":"rows differ"}"#), + "a scalar field is named before a later list difference" + ); + + let entry = compared(Ok(()), &["a", "b"], &changed); + assert_eq!( + super::replay_difference(&base, &entry).as_deref(), + Some(r#"evidence[1] differs: run 0 {"step":2} | run 1 {"step":9}"#) + ); + + let longer = compared(Ok(()), &["a", "b", "c"], &steps); + assert_eq!( + super::replay_difference(&base, &longer).as_deref(), + Some( + "observations has 2 entries in run 0 and 3 in run 1; the shorter is a prefix of the longer" + ) + ); + assert_eq!( + super::replay_difference(&longer, &base).as_deref(), + Some( + "observations has 3 entries in run 0 and 2 in run 1; the shorter is a prefix of the longer" + ) + ); + let none = compared(Ok(()), &[], &[]); + assert_eq!( + super::replay_difference(&none, &compared(Ok(()), &[], &steps[..1])).as_deref(), + Some( + "evidence has 0 entries in run 0 and 1 in run 1; the shorter is a prefix of the longer" + ) + ); + assert_eq!( + super::replay_difference(&base, &compared(Ok(()), &["b", "a"], &steps)).as_deref(), + Some(r#"observations[0] differs: run 0 "a" | run 1 "b""#) + ); + + let mut other = compared(Err("e".into()), &["b"], &[]); + other.code = "worker_failed".into(); + other.phase = "setup".into(); + other.input_digest = "other".into(); + assert_eq!( + super::replay_difference(&base, &other).as_deref(), + Some(r#"code differs: run 0 "ok" | run 1 "worker_failed""#) + ); + other.code = "ok".into(); + assert_eq!( + super::replay_difference(&base, &other).as_deref(), + Some(r#"phase differs: run 0 "run" | run 1 "setup""#) + ); + other.phase = "run".into(); + assert_eq!( + super::replay_difference(&base, &other).as_deref(), + Some(r#"input_digest differs: run 0 "digest" | run 1 "other""#) + ); + } + + #[test] + fn replay_difference_agrees_with_the_serialized_report_comparison() { + let base = compared(Ok(()), &["a"], &[serde_json::json!({"k": "v", "x": 0.0})]); + let mut only_measured = + compared(Ok(()), &["a"], &[serde_json::json!({"k": "v", "x": 0.0})]); + only_measured + .measurements + .push(serde_json::json!({"bytes": 7})); + let negative_zero: serde_json::Value = + serde_json::from_str(r#"{"k":"v","x":-0.0}"#).unwrap(); + let reordered: serde_json::Value = serde_json::from_str(r#"{"x":0.0,"k":"v"}"#).unwrap(); + let variants = [ + compared(Ok(()), &["a"], &[serde_json::json!({"k": "v", "x": 0.0})]), + only_measured, + compared( + Err("e".into()), + &["a"], + &[serde_json::json!({"k": "v", "x": 0.0})], + ), + compared(Ok(()), &["b"], &[serde_json::json!({"k": "v", "x": 0.0})]), + compared(Ok(()), &["a"], &[]), + compared(Ok(()), &["a"], &[negative_zero]), + compared(Ok(()), &["a"], &[reordered]), + compared(Ok(()), &["a"], &[serde_json::json!({"k": "v", "x": 0})]), + ]; + let bytes = |report: &super::WorkerReport| super::json(&super::comparable(report)).unwrap(); + for variant in &variants { + let differs = bytes(&base) != bytes(variant); + assert_eq!( + super::replay_difference(&base, variant).is_some(), + differs, + "{variant:?}" + ); + assert_eq!( + super::replay_difference(variant, &base).is_some(), + differs, + "{variant:?}" + ); + } + assert_eq!( + super::replay_difference(&base, &variants[5]).as_deref(), + Some(r#"evidence[0] differs: run 0 {"k":"v","x":0.0} | run 1 {"k":"v","x":-0.0}"#) + ); + assert_eq!( + super::replay_difference(&base, &variants[6]).as_deref(), + Some(r#"evidence[0] differs: run 0 {"k":"v","x":0.0} | run 1 {"x":0.0,"k":"v"}"#) + ); + } + + #[test] + fn replay_difference_windows_each_rendered_value() { + let long = compared( + Ok(()), + &[], + &[serde_json::json!({"text": "é".repeat(1_000)})], + ); + let short = compared(Ok(()), &[], &[serde_json::json!({"text": "short"})]); + let line = super::replay_difference(&long, &short).unwrap(); + assert!( + line.starts_with(r#"evidence[0] differs: run 0 {"text":"éééé"#), + "{line}" + ); + assert!(line.contains("… (1011 chars) | run 1 "), "{line}"); + assert!(line.ends_with(r#"| run 1 {"text":"short"}"#), "{line}"); + assert!(line.chars().count() < 320, "{line}"); + + let whole = "x".repeat(238); + let line = super::replay_difference( + &compared(Ok(()), &[&whole], &[]), + &compared(Ok(()), &["y"], &[]), + ) + .unwrap(); + assert_eq!( + line, + format!(r#"observations[0] differs: run 0 "{whole}" | run 1 "y""#) + ); + let cut = "x".repeat(239); + let line = super::replay_difference( + &compared(Ok(()), &[&cut], &[]), + &compared(Ok(()), &["y"], &[]), + ) + .unwrap(); + assert_eq!( + line, + format!(r#"observations[0] differs: run 0 "{cut}… (241 chars) | run 1 "y""#) + ); + + let prefix = "a".repeat(300); + let line = super::replay_difference( + &compared( + Ok(()), + &[], + &[serde_json::json!({"rows": format!("{prefix}1")})], + ), + &compared( + Ok(()), + &[], + &[serde_json::json!({"rows": format!("{prefix}2")})], + ), + ) + .unwrap(); + assert_eq!( + line, + format!( + r#"evidence[0] differs: run 0 …{0}1"}} | run 1 …{0}2"}}"#, + "a".repeat(40) + ) + ); + + let escaped = format!("{}\"", "x".repeat(238)); + let line = super::replay_difference( + &compared(Ok(()), &[&escaped], &[]), + &compared(Ok(()), &["y"], &[]), + ) + .unwrap(); + assert_eq!( + line, + format!( + r#"observations[0] differs: run 0 "{}\"… (242 chars) | run 1 "y""#, + "x".repeat(238) + ) ); } diff --git a/crates/omnigraph/src/engine/operators/fixtures.rs b/crates/omnigraph/src/engine/operators/fixtures.rs index e28eacfa7..a742edcb6 100644 --- a/crates/omnigraph/src/engine/operators/fixtures.rs +++ b/crates/omnigraph/src/engine/operators/fixtures.rs @@ -48,14 +48,21 @@ pub(super) fn texts(names: [&str; 2], batches: &[&[(&str, &str)]]) -> Arc (Arc, Arc) { let pool: Arc = Arc::new(GreedyMemoryPool::new(limit)); - let resources = Arc::new(QueryResources::new(Arc::clone(&pool), limit as u64)); + (Arc::clone(&pool), context_over(pool, limit, size)) +} + +/// A task context over `pool`, whose query limit is `limit` bytes, `size` +/// rows a batch. +pub(super) fn context_over( + pool: Arc, + limit: usize, + size: usize, +) -> Arc { + let resources = Arc::new(QueryResources::new(pool, limit as u64)); let config = SessionConfig::new() .with_batch_size(size) .with_extension(resources); - ( - pool, - Arc::new(TaskContext::default().with_session_config(config)), - ) + Arc::new(TaskContext::default().with_session_config(config)) } /// The `(m.mid, p.pid)` pairs of each of `batches`. diff --git a/crates/omnigraph/src/engine/operators/validate.rs b/crates/omnigraph/src/engine/operators/validate.rs index fbac4be83..3e89eb78b 100644 --- a/crates/omnigraph/src/engine/operators/validate.rs +++ b/crates/omnigraph/src/engine/operators/validate.rs @@ -144,12 +144,9 @@ impl ExecutionPlan for ValidateExec { .properties() .output_partitioning() .partition_count(); - let inputs = (0..partitions) - .map(|partition| self.input.execute(partition, Arc::clone(&ctx))) - .collect::>>()?; let source = Arc::clone(&self.input); let schema = self.input.schema(); - let memory = work(ctx, "ValidateExec", &self.metrics)?; + let memory = work(Arc::clone(&ctx), "ValidateExec", &self.metrics)?; memory.metric("input_rows", 0); memory.metric(VALIDATED_ROWS, 0); let judge = Arc::clone(&self.judge); @@ -164,6 +161,9 @@ impl ExecutionPlan for ValidateExec { &self.metrics, move |memory| async move { let admitted = admit_sets(&memory, &judge, &plan).await?; + let inputs = (0..partitions) + .map(|partition| source.execute(partition, Arc::clone(&ctx))) + .collect::>>()?; let (publish, published) = watch::channel(None::>); let (drained, set_guards) = futures::join!( drain_and_sweep( @@ -333,8 +333,9 @@ async fn drain_and_sweep( Ok((batches, partials, facts)) } -/// The swept endpoint checks' key sets admitted on a worker before the drain -/// starts, so a refused set precedes every partition's charge. +/// The swept endpoint checks' key sets admitted on a worker before the source +/// partitions are executed, so every set's charge precedes every charge of the +/// parse and a refused set precedes them all. async fn admit_sets( memory: &Arc, judge: &Arc, @@ -728,7 +729,11 @@ fn string_payload_bytes(column: &ArrayRef) -> usize { #[cfg(test)] mod tests { use std::fmt::Write; + use std::sync::Mutex; + use datafusion::execution::memory_pool::{ + GreedyMemoryPool, MemoryConsumer, MemoryPool, MemoryReservation, + }; use omnigraph_compiler::catalog::build_catalog; use omnigraph_compiler::ir::lower_load; use omnigraph_compiler::query::ast::{LoadStmt, LoadStmtMode, LoadTarget}; @@ -1350,4 +1355,124 @@ mod tests { drop(retained); assert_eq!(pool.reserved(), before); } + + /// A pool recording the consumer and size of every charge in the order it + /// saw them, over a greedy pool of `limit` bytes. + #[derive(Debug)] + struct ChargeLedger { + inner: GreedyMemoryPool, + charges: Mutex>, + } + + impl ChargeLedger { + fn new(limit: usize) -> Self { + Self { + inner: GreedyMemoryPool::new(limit), + charges: Mutex::new(Vec::new()), + } + } + + fn record(&self, reservation: &MemoryReservation, additional: usize) { + self.charges + .lock() + .unwrap() + .push((reservation.consumer().name().to_owned(), additional)); + } + + fn charges(&self) -> Vec<(String, usize)> { + self.charges.lock().unwrap().clone() + } + } + + impl fmt::Display for ChargeLedger { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str("charge ledger") + } + } + + impl MemoryPool for ChargeLedger { + fn name(&self) -> &str { + "charge ledger" + } + + fn register(&self, consumer: &MemoryConsumer) { + self.inner.register(consumer); + } + + fn unregister(&self, consumer: &MemoryConsumer) { + self.inner.unregister(consumer); + } + + fn grow(&self, reservation: &MemoryReservation, additional: usize) { + self.record(reservation, additional); + self.inner.grow(reservation, additional); + } + + fn shrink(&self, reservation: &MemoryReservation, shrink: usize) { + self.inner.shrink(reservation, shrink); + } + + fn try_grow(&self, reservation: &MemoryReservation, additional: usize) -> DfResult<()> { + self.record(reservation, additional); + self.inner.try_grow(reservation, additional) + } + + fn reserved(&self) -> usize { + self.inner.reserved() + } + } + + /// Every key-set charge precedes the parse's first charge: the sets are + /// admitted before the source partitions are executed, a thread order + /// GQT cannot observe (as `partitions_admit_their_source_before_the_first_poll`). + #[tokio::test] + async fn the_sets_are_charged_before_the_source_partitions_are_executed() { + let dir = tempfile::tempdir().unwrap(); + let db = keyed_graph(&dir).await; + let catalog = db.catalog(); + let checks = swept_checks(&catalog); + let snapshot = db.snapshot().await; + let exec = Arc::new(CsvSourceExec::new( + knows_request(), + bytes("from,to\n1,2\n2,3\n3,1\n"), + Arc::clone(&catalog), + target_arrow_schema(&catalog, &LoadTarget::Edge("Knows".to_string())).unwrap(), + 2, + 64, + Arc::default(), + )); + let validate = ValidateExec::new( + Arc::clone(&exec) as Arc, + knows_judge(&db, &checks, &snapshot), + ); + let ledger = Arc::new(ChargeLedger::new(64 << 20)); + let ctx = fixtures::context_over( + Arc::clone(&ledger) as Arc, + 64 << 20, + CSV_READ_BATCH_ROWS, + ); + let judged = validate + .execute(0, ctx) + .unwrap() + .collect::>() + .await + .into_iter() + .collect::>>() + .unwrap(); + assert_eq!(judged.iter().map(RecordBatch::num_rows).sum::(), 3); + let charges = ledger.charges(); + let first = |consumer: &str| { + charges + .iter() + .position(|(name, _)| name == consumer) + .unwrap_or_else(|| panic!("no charge by {consumer}: {charges:?}")) + }; + let last_set = charges + .iter() + .rposition(|(name, _)| name == "load integer endpoint keys") + .unwrap_or_else(|| panic!("no key-set charge: {charges:?}")); + assert!(last_set < first("CsvSourceExec"), "{charges:?}"); + assert!(last_set < first("CSV source bytes"), "{charges:?}"); + assert!(last_set < first("CSV decoding"), "{charges:?}"); + } } diff --git a/deny.toml b/deny.toml index e309ac717..bbbcf8247 100644 --- a/deny.toml +++ b/deny.toml @@ -25,6 +25,24 @@ unused-ignored-advisory = "deny" ignore = [ { id = "RUSTSEC-2024-0436", reason = "paste is unmaintained; a proc-macro pulled by lance-bitpacking 13.0.0, cleared when Lance replaces it" }, { id = "RUSTSEC-2023-0071", reason = "rsa 0.9.10 Marvin timing side channel via reqsign (lance-io, opendal); RustSec lists no patched version" }, + # zstd 0.13 and its zstd-safe 7 are pinned by arrow-ipc 58.3.0 (`zstd = "0.13.0"`) + # and lance-encoding 13.0.0 (`zstd = "0.13"`); the patched line is zstd 0.14.3 with + # zstd-safe 8.2.0, taken when arrow and Lance move. Those two crates reach only the + # bulk API and `stream::copy_decode` (a `read::Decoder` on a fresh context); no + # workspace crate names zstd. Each row names the API the advisory is about. + { id = "RUSTSEC-2026-0391", reason = "zstd 0.13.3 `with_context` stream constructors on a borrowed context; arrow-ipc and lance-encoding build every context themselves (bulk, `copy_decode`); cleared by zstd 0.14.3" }, + { id = "RUSTSEC-2026-0395", reason = "zstd-safe 7.2.4 `CParameter::StableInBuffer`; never set by zstd 0.13.3 or its two dependents; cleared by zstd-safe 8.2.0" }, + { id = "RUSTSEC-2026-0396", reason = "zstd-safe 7.2.4 `Seekable` seek table; the seekable API is not used; cleared by zstd-safe 8.2.0" }, + { id = "RUSTSEC-2026-0397", reason = "zstd-safe 7.2.4 `OutBuffer::around_pos`; zstd 0.13.3 never calls it and no crate uses zstd-safe streaming directly (lance-encoding calls only `compress_bound`); cleared by zstd-safe 8.2.0" }, + { id = "RUSTSEC-2026-0398", reason = "zstd-safe 7.2.4 `Seekable::decompress`; the seekable API is not used; cleared by zstd-safe 8.2.0" }, + { id = "RUSTSEC-2026-0399", reason = "zstd-safe 7.2.4 `init_using_cdict` / `init_using_ddict`; no dictionary is used; cleared by zstd-safe 8.2.0" }, + { id = "RUSTSEC-2026-0400", reason = "zstd-safe 7.2.4 one-shot dictionary call on a context with an open stream; no dictionary is used and the bulk contexts never carry a stream; cleared by zstd-safe 8.2.0" }, + { id = "RUSTSEC-2026-0401", reason = "zstd-safe 7.2.4 `train_from_buffer`; no dictionary is trained; cleared by zstd-safe 8.2.0" }, + { id = "RUSTSEC-2026-0402", reason = "zstd-safe 7.2.4 `CCtx::try_clone` on a multithreaded context; neither `try_clone` nor `zstdmt` is used; cleared by zstd-safe 8.2.0" }, + { id = "RUSTSEC-2026-0403", reason = "zstd-safe 7.2.4 legacy frame through `stream::write::Decoder`; the tree decodes through `read::Decoder` and the bulk API, which the advisory lists as unaffected; cleared by zstd-safe 8.2.0" }, + { id = "RUSTSEC-2026-0404", reason = "zstd-safe 7.2.4 `DCtx::try_clone`; not used; cleared by zstd-safe 8.2.0" }, + { id = "RUSTSEC-2026-0405", reason = "zstd-safe 7.2.4 out-of-range `InBuffer::pos` in the seekable compress stream; the seekable API is not used; cleared by zstd-safe 8.2.0" }, + { id = "RUSTSEC-2026-0406", reason = "zstd-safe 7.2.4 changing a multithreaded context with an unfinished frame; `zstdmt` is not used; cleared by zstd-safe 8.2.0" }, ] [licenses]