From 9148ed3587c68df674966068029b64d6620d747b Mon Sep 17 00:00:00 2001 From: Andrew Johnson Date: Sat, 18 Jul 2026 18:10:47 -0400 Subject: [PATCH 1/3] =?UTF-8?q?fix(query):=20honest=20memory-budget=20sizi?= =?UTF-8?q?ng=20=E2=80=94=20size=5Fof-derived=20binding=20estimate=20+=20p?= =?UTF-8?q?er-query=20division?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit F-AUD-3 sites D and C (audit-2026-07/V2-membudget-verification.md §6, §3). D: BINDING_EST_BYTES was a hand-picked 64 — a 27% under-count of the true 88-byte size_of::() (binding.rs:14-17), so every accounted operator checkpointed late. Derive it from the type and add a compile-time `const _` guard that refuses any future re-pin below the stack size. It is still a floor (ignores the Arc IRI heap a wide crawl carries, ~2.2x) — documented on the constant. C: set_memory_limit had no production caller, so N concurrent queries each compared their own counter against the FULL process budget (two 5 GB queries both read "under 8 GB" while the node sits at 10 GB). Pin a per-query ceiling of budget / FLUREE_QUERY_BUDGET_SHARE_DIV at the runner attach point. Default div=1 pins nothing — byte-for-byte today's behavior; an embedder's explicit ceiling is never clobbered. The sound dynamic form (divide by ACTUAL live top-level concurrency) needs the server request boundary (query_control.rs) to avoid miscounting nested policy/reasoning/sub-queries, and is deferred there. Tests: binding_est_bytes_is_at_least_binding_stack_size, per_query_ceiling_divides_and_floors, shared_ceiling_trips_each_query_at_its_divided_budget. --- fluree-db-query/src/context.rs | 140 +++++++++++++++++++++++++- fluree-db-query/src/execute/runner.rs | 13 +++ 2 files changed, 149 insertions(+), 4 deletions(-) diff --git a/fluree-db-query/src/context.rs b/fluree-db-query/src/context.rs index c4fbf71a6b..46e751095d 100644 --- a/fluree-db-query/src/context.rs +++ b/fluree-db-query/src/context.rs @@ -49,13 +49,60 @@ pub fn query_memory_budget_bytes() -> usize { }) } -/// R3-B: conservative per-binding byte estimate for the approximate memory-budget -/// accounting (a `Binding` is a small enum, occasionally with a heap `String`; -/// over-counting is safe — a too-tight budget only aborts a query already near OOM). -pub const BINDING_EST_BYTES: usize = 64; +/// R3-B: per-binding byte estimate for the approximate memory-budget accounting. +/// Derived from the true stack size of a [`Binding`](crate::binding::Binding) +/// (`size_of::()` = 88 bytes, documented at `binding.rs:14-17`) rather +/// than a hand-picked number, so it can never silently under-count the stack +/// footprint the way the previous `64` did (a 27% under-count of the 88-byte enum). +/// +/// This still under-counts the HEAP a binding owns: an IRI-bearing row +/// (`Binding::Iri(Arc)`, `IriMatch { iri: Arc, ledger_alias: Arc }`, +/// or a `Lit` with a `String` value) carries ~50-70 bytes of `Arc` payload +/// the stack size does not see, so a wide R2RML/Iceberg crawl of IRI rows is still +/// counted at roughly 1/2.2 of its true resident bytes. Over-counting is safe (a +/// too-tight budget only aborts a query already near OOM); this constant deliberately +/// stays a floor. A heap-aware per-binding estimate is a documented follow-up. +pub const BINDING_EST_BYTES: usize = std::mem::size_of::(); +/// F-AUD-3 site D: compile-time guard. The estimate must never drop below the +/// true `Binding` stack size — refusing any future edit that pins a smaller magic +/// number (the very regression the `64` was). Trivially holds while it is DEFINED +/// as `size_of`, and bites the moment someone changes that. +const _: () = assert!(BINDING_EST_BYTES >= std::mem::size_of::()); /// R3-B: conservative per-group overhead estimate (key bindings + aggregate state). +/// A flat estimate — like [`BINDING_EST_BYTES`] it ignores per-group heap (e.g. a +/// `GROUP_CONCAT`/`Collect` accumulator), the same documented conservatism. pub const GROUP_EST_BYTES: usize = 128; +/// F-AUD-3 site C: divisor applied to the process memory budget to derive a +/// per-query ceiling, read from `FLUREE_QUERY_BUDGET_SHARE_DIV`. Default `1` (and +/// any unparseable/zero value floors to `1`), which makes the per-query ceiling +/// equal to the full budget — byte-for-byte today's behavior. Set it to the +/// deployment's expected max query concurrency (e.g. a Lambda's reserved +/// concurrency) so N concurrent queries share the budget instead of each +/// comparing its own counter against the FULL budget (the over-admission V2 §3 +/// describes: two queries each accounting 5 GB both read "under 8 GB" while the +/// node sits at 10 GB). See [`per_query_memory_ceiling`]. +/// +/// This is the sound, minimal static form. A dynamic divisor equal to the ACTUAL +/// live concurrency (so a lone query keeps the full budget) needs a process-wide +/// active-*top-level*-query count — which only the server request boundary +/// (`fluree-db-server/src/query_control.rs`) can measure without miscounting +/// nested policy/reasoning/sub-queries that re-enter the engine. Deferred there. +pub fn query_budget_share_div() -> usize { + std::env::var("FLUREE_QUERY_BUDGET_SHARE_DIV") + .ok() + .and_then(|v| v.trim().parse::().ok()) + .unwrap_or(1) + .max(1) +} + +/// F-AUD-3 site C: the per-query memory ceiling = `full_budget / share_div`, with +/// `share_div` floored at 1. `share_div == 1` returns the full budget unchanged. +/// Pure so the division is unit-testable without touching the environment. +pub fn per_query_memory_ceiling(full_budget: usize, share_div: usize) -> usize { + full_budget / share_div.max(1) +} + /// Best-effort container/system memory limit (cgroup v2 → cgroup v1 → /// `/proc/meminfo`). `None` where none is readable (e.g. macOS dev), where the /// caller uses the absolute fallback. A cgroup "unlimited" sentinel (non-numeric @@ -1581,3 +1628,88 @@ impl WellKnownDatatypes { false } } + +#[cfg(test)] +mod budget_tests { + use super::*; + use crate::binding::Binding; + use crate::error::QueryError; + use crate::var_registry::VarRegistry; + use fluree_db_core::{LedgerSnapshot, QueryCancellation}; + + /// F-AUD-3 site D: the per-binding estimate must never be below the true stack + /// size of a `Binding` (the 64→88 fix). Deriving it from `size_of` makes the + /// `>=` hold by construction; the equality canary documents the 88-byte size + /// (binding.rs:14-17) and fails loudly if the enum grows so the estimate is + /// re-examined for the heap it still omits. + #[test] + fn binding_est_bytes_is_at_least_binding_stack_size() { + // Derived from the type, so it equals the stack size and can never silently + // under-count it the way the previous hardcoded 64 did. The 88 canary + // documents binding.rs:14-17 and fails loudly if the enum grows (prompting a + // re-look at the heap the estimate still omits). The `>= size_of` invariant + // itself is a compile-time `const _` assertion next to the constant. + assert_eq!(BINDING_EST_BYTES, std::mem::size_of::()); + assert_eq!( + std::mem::size_of::(), + 88, + "binding.rs:14-17 documents size_of::() == 88" + ); + } + + /// F-AUD-3 site C: the ceiling divides the full budget and floors the divisor + /// at 1, so `div == 1` (the default) is byte-for-byte the full budget. + #[test] + fn per_query_ceiling_divides_and_floors() { + let full = 8usize << 30; // 8 GiB + assert_eq!( + per_query_memory_ceiling(full, 1), + full, + "div=1 → full budget" + ); + assert_eq!( + per_query_memory_ceiling(full, 4), + 2usize << 30, + "div=4 → quarter" + ); + assert_eq!(per_query_memory_ceiling(full, 0), full, "div=0 floors to 1"); + } + + /// F-AUD-3 site C: two concurrent queries pinned to a shared (divided) ceiling + /// each trip at their divided budget — neither can consume the full budget the + /// way today's undivided per-query counter allows (V2 §3 over-admission). + #[test] + fn shared_ceiling_trips_each_query_at_its_divided_budget() { + let full = 8usize << 30; + let ceiling = per_query_memory_ceiling(full, 2); // 4 GiB each + assert_eq!(ceiling, 4usize << 30); + + let snapshot = LedgerSnapshot::genesis("test/main"); + let vars = VarRegistry::new(); + // Two independent queries, each pinned to the divided ceiling as the runner + // attach point does under FLUREE_QUERY_BUDGET_SHARE_DIV=2. + for _ in 0..2 { + let cancel = QueryCancellation::new(); + cancel.set_memory_limit(ceiling); + let ctx = ExecutionContext::new(&snapshot, &vars).with_cancellation(cancel); + // Recording just over the DIVIDED ceiling trips, though it is well under + // the full 8 GiB budget an undivided query would compare against. + ctx.record_alloc(ceiling + 1); + match ctx.checkpoint() { + Err(QueryError::MemoryBudgetExceeded { + used_bytes, + budget_bytes, + }) => { + assert_eq!( + budget_bytes, ceiling, + "enforced ceiling is the divided budget" + ); + assert_eq!(used_bytes, ceiling + 1); + } + other => { + panic!("expected MemoryBudgetExceeded at the divided ceiling, got {other:?}") + } + } + } + } +} diff --git a/fluree-db-query/src/execute/runner.rs b/fluree-db-query/src/execute/runner.rs index a0e618dd75..4fa89f7691 100644 --- a/fluree-db-query/src/execute/runner.rs +++ b/fluree-db-query/src/execute/runner.rs @@ -826,6 +826,19 @@ async fn execute_prepared_into<'a, S: BatchSink>( ctx = ctx.with_tracker(tracker.clone()); } if let Some(cancellation) = config.cancellation { + // F-AUD-3 site C: give this query its share of the process memory budget + // instead of letting it (and every concurrent query) compare its own counter + // against the FULL budget. `FLUREE_QUERY_BUDGET_SHARE_DIV` (default 1) is the + // divisor; div==1 pins nothing, so the checkpoint falls back to the full + // process budget exactly as before. An explicit ceiling already pinned by the + // embedder wins (never clobbered). See `context::per_query_memory_ceiling`. + let div = crate::context::query_budget_share_div(); + if div > 1 && cancellation.memory_limit().is_none() { + let full = crate::context::query_memory_budget_bytes(); + if full != 0 { + cancellation.set_memory_limit(crate::context::per_query_memory_ceiling(full, div)); + } + } ctx = ctx.with_cancellation(cancellation); } if let Some(enforcer) = config.policy_enforcer { From d9c1f4c5f99c575a6f50a7d94d319ae75660dad2 Mon Sep 17 00:00:00 2001 From: Andrew Johnson Date: Sat, 18 Jul 2026 18:11:03 -0400 Subject: [PATCH 2/3] fix(query): account the R2RML scan/crawl path against the memory budget MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit F-AUD-3 sites A1 and A2 (audit-2026-07/V2-membudget-verification.md §1, §4). The non-aggregate scan path had zero record_alloc / checkpoint — six check_cancelled only — so a wide crawl was invisible to the R3-B memory budget and OOM'd instead of aborting typed. Specimen 071cd59f (a point-lookup crawl that hard-OOM'd at 10237 MB) lived exactly here. A1: record each materialized window (produced_rows * cols * BINDING_EST) in advance_one_window and upgrade the pull-loop poll from check_cancelled() to checkpoint(), so cumulative window bytes trip a typed MemoryBudgetExceeded (507) before the loop pulls another window. One window is bounded (~materialize_window_rows) so it cannot itself OOM. A2: the fact-as-parent build (build_parent_lookup) transiently materializes a full parent-sized map (tens of millions of entries) unbounded by the memo cap — the cap only refuses to RETAIN it after it is fully built. Thread ctx in, account each batch, and checkpoint inside the build loop so it aborts typed BEFORE the whole map is resident. Both gated by FLUREE_SCAN_MEM_ACCOUNTING (default on; off is a clean revert — the scan records nothing, so checkpoint degrades to a pure cancellation poll). The counter is query-lifetime cumulative (no decrement), conservative for a streaming scan, matching the existing fold/join accounting. Per-file buffer accounting (V2 site B) is excluded — it needs a decrement primitive the monotonic counter lacks. Hermetics: r3b_scan_window_budget_aborts_typed (the 071cd59f regression), r3b_parent_build_budget_aborts_typed. --- fluree-db-query/src/r2rml/operator.rs | 237 ++++++++++++++++++++++---- 1 file changed, 206 insertions(+), 31 deletions(-) diff --git a/fluree-db-query/src/r2rml/operator.rs b/fluree-db-query/src/r2rml/operator.rs index 6c0cc47583..53d5dd6b34 100644 --- a/fluree-db-query/src/r2rml/operator.rs +++ b/fluree-db-query/src/r2rml/operator.rs @@ -236,6 +236,34 @@ fn topk_pushdown_enabled() -> bool { *ENABLED.get_or_init(|| super::env_switch_enabled("FLUREE_R2RML_TOPK_PUSHDOWN")) } +/// F-AUD-3 kill switch: whether the non-aggregate scan/crawl path records its +/// materialized windows and fact-parent lookup builds against the query memory +/// budget. Default ON (`FLUREE_SCAN_MEM_ACCOUNTING`, family falsy spellings via +/// [`super::env_switch_enabled`]). ON makes a wide crawl trip a typed +/// `MemoryBudgetExceeded` (507) instead of OOMing — closing the blind spot the +/// audit's specimen 071cd59f (a point-lookup crawl that hard-OOM'd at 10 GB) fell +/// into, which the hash-join / group-aggregate `record_alloc`s never covered. +/// OFF is a clean revert to the prior behavior (scan path invisible to the budget; +/// the `checkpoint()` polls degrade to pure cancellation checks because the scan +/// records nothing). The counter is query-lifetime cumulative and never +/// decremented, so a genuinely-streaming large scan is accounted as if all its +/// windows were simultaneously resident — the intended conservative direction +/// (over-count only ever aborts a query already near the budget), matching the +/// existing fold/join accounting; a high-water refinement is deferred with the +/// excluded per-file buffer accounting (V2 site B). +fn scan_mem_accounting_enabled() -> bool { + static ENABLED: std::sync::OnceLock = std::sync::OnceLock::new(); + *ENABLED.get_or_init(|| super::env_switch_enabled("FLUREE_SCAN_MEM_ACCOUNTING")) +} + +/// F-AUD-3: conservative per-entry byte estimate for a `build_parent_lookup` +/// entry (a join key vec + a materialized subject `RdfTerm`, typically an IRI +/// string) used to account the fact-as-parent build against the budget. ~200 B is +/// the V2 estimate; like [`crate::context::BINDING_EST_BYTES`] it is a floor +/// (ignores per-entry heap beyond the term), safe because over-counting only trips +/// a build already near OOM. +const PARENT_ENTRY_EST_BYTES: usize = 200; + /// How a window of produced rows is combined with the buffered child rows. /// /// The join is *flipped* relative to a naive per-child probe: the (small, @@ -563,8 +591,7 @@ impl R2rmlScanOperator { /// — the production build calls THIS fn, so the test exercises the real /// admission, not a copy. fn ref_template_shortcut_enabled(trust_fk_refs: bool, pattern: &R2rmlPattern) -> bool { - let star_free = - pattern.star_bindings.is_empty() && pattern.star_constraints.is_empty(); + let star_free = pattern.star_bindings.is_empty() && pattern.star_constraints.is_empty(); trust_fk_refs && (star_free || Self::folded_wildcard_all_scalar(pattern)) && pattern.predicate_filter.is_none() @@ -833,17 +860,15 @@ impl R2rmlScanOperator { // WILDCARD emits per-(p,o) across co-subject maps, so a map lacking the // folded constraint predicate can still contribute rows — pruning it would // drop those rows for vertically-partitioned subjects (F10-class mappings). - let star_required_preds: Vec = if star_prune_on - && self.pattern.predicate_var.is_none() - && self.has_star_members() - { - self.pattern_predicates() - .iter() - .map(|s| (*s).to_string()) - .collect() - } else { - Vec::new() - }; + let star_required_preds: Vec = + if star_prune_on && self.pattern.predicate_var.is_none() && self.has_star_members() { + self.pattern_predicates() + .iter() + .map(|s| (*s).to_string()) + .collect() + } else { + Vec::new() + }; // PR-3 fix (b'): resolution-only class prune (template-disjoint; see // `R2rmlPattern::class_prune_hint`). Gated by the same switch as fix (a). let prune_class: Option = if star_prune_on { @@ -1286,6 +1311,7 @@ impl R2rmlScanOperator { let parent_batches = collect_stream(parent_stream, ctx).await?; let lookup = Arc::new(build_parent_lookup( + ctx, parent_tm, &parent_join_cols, parent_batches, @@ -1431,6 +1457,23 @@ impl R2rmlScanOperator { ctx, )?; + // F-AUD-3 site A1: account the materialized window against the query + // memory budget so a wide non-aggregate crawl (the previously-blind scan + // path) trips a typed `MemoryBudgetExceeded` instead of OOMing. The window + // is bounded (~`materialize_window_rows`) so one window cannot itself OOM; + // `checkpoint()` here aborts once the cumulative recorded bytes (this + // window plus every prior one and any upstream fold/join) cross the + // budget, before the loop pulls another. Cumulative, never decremented — + // see `scan_mem_accounting_enabled`. + if scan_mem_accounting_enabled() { + let window_est = produced + .len() + .saturating_mul(num_cols) + .saturating_mul(crate::context::BINDING_EST_BYTES); + ctx.record_alloc(window_est); + ctx.checkpoint()?; + } + if !produced.is_empty() { emit_produced_window( &self.out_pos, @@ -2496,6 +2539,7 @@ fn materialize_batch( /// /// HashMap mapping join key (as `Vec`) to parent subject `RdfTerm`. fn build_parent_lookup( + ctx: &ExecutionContext<'_>, parent_tm: &TriplesMap, parent_columns: &[String], batches: Vec, @@ -2503,6 +2547,20 @@ fn build_parent_lookup( let mut lookup = ParentLookup::new(); for batch in batches { + // F-AUD-3 site A2: the fact-as-parent hazard (V2) — a RefObjectMap whose + // parent is a FACT table transiently builds a full parent-sized map (tens of + // millions of entries) here, unbounded by the memo cap, which only refuses to + // RETAIN an over-window lookup AFTER it is fully built. Account each batch's + // worst-case contribution and checkpoint inside the build loop so a + // budget-exceeding build aborts typed (`MemoryBudgetExceeded`) BEFORE the + // whole map is resident, instead of OOMing. `num_rows` over-counts skipped + // (null-subject / null-key) rows — deliberately conservative. Cumulative like + // site A1; the map is genuinely retained while built, so cumulative == the + // resident footprint for this structure. + if scan_mem_accounting_enabled() { + ctx.record_alloc(batch.num_rows.saturating_mul(PARENT_ENTRY_EST_BYTES)); + ctx.checkpoint()?; + } for row_idx in 0..batch.num_rows { // Materialize parent subject let subject_term = @@ -2607,13 +2665,16 @@ impl Operator for R2rmlScanOperator { .collect(); loop { - // Cancellation checkpoint at the top of the internal loop: this loop - // can pull many windows / files / child batches before returning a + // Cancellation + memory checkpoint at the top of the internal loop: this + // loop can pull many windows / files / child batches before returning a // full output batch, so the runner's between-`next_batch` check would // otherwise never run for a whole-table scan. Covers the loop's - // non-advancing branches (overflow drain, child pull) that site 2 - // does not. - ctx.check_cancelled()?; + // non-advancing branches (overflow drain, child pull) that site 2 does + // not. Upgraded from `check_cancelled()` to `checkpoint()` (F-AUD-3): it + // also enforces the query memory budget against the window bytes recorded + // by site A1 (a no-op for the budget when scan accounting is off, since + // nothing on this path records then). + ctx.checkpoint()?; // 1. Drain overflow from a prior window before doing more work. while !self.pending.is_empty() && columns[0].len() < ctx.batch_size { let row = self.pending.pop_front().unwrap(); @@ -2763,6 +2824,120 @@ mod tests { )); } + /// F-AUD-3 site A1 — specimen 071cd59f regression. A wide non-aggregate crawl + /// (`?s ?p ?o`) materializes a window of bindings that the pre-fix scan path + /// never recorded against the memory budget, so a runaway crawl OOM'd instead of + /// aborting typed. With scan accounting on (the default), the materialized + /// window is recorded and a tiny 1-byte ceiling makes the window checkpoint + /// abort TYPED (`MemoryBudgetExceeded`, a 507) rather than completing/OOMing. + #[tokio::test] + async fn r3b_scan_window_budget_aborts_typed() { + use crate::r2rml::R2rmlTableProvider; + use crate::seed::EmptyOperator; + use crate::var_registry::VarRegistry; + use fluree_db_core::QueryCancellation; + use fluree_db_r2rml::mapping::{ + CompiledR2rmlMapping, ObjectMap, PredicateMap, PredicateObjectMap, TriplesMap, + }; + use fluree_db_tabular::{BatchSchema, FieldInfo, FieldType}; + + #[derive(Debug)] + struct StoreProvider; + #[async_trait::async_trait] + impl R2rmlTableProvider for StoreProvider { + async fn scan_table( + &self, + _graph_source_id: &str, + _table_name: &str, + _projection: &[String], + _filters: &[crate::r2rml::ScanFilter], + _topk: Option<&crate::r2rml::ScanTopK>, + _as_of_t: Option, + ) -> Result { + let schema = Arc::new(BatchSchema::new(vec![FieldInfo { + name: "STORE_KEY".to_string(), + field_type: FieldType::Int64, + nullable: true, + field_id: 1, + }])); + let batch = + ColumnBatch::new(schema, vec![Column::Int64(vec![Some(1), Some(2), Some(3)])]) + .unwrap(); + Ok(Box::pin(futures::stream::once(async move { Ok(batch) }))) + } + } + + let tm = TriplesMap::new("#Store", "DIM_STORE") + .with_subject_template("http://ex/store/{STORE_KEY}") + .with_class("http://ex/Store") + .with_predicate_object(PredicateObjectMap { + predicate_map: PredicateMap::constant("http://ex/storeKey"), + object_map: ObjectMap::column("STORE_KEY"), + }); + let mapping = Arc::new(CompiledR2rmlMapping::new(vec![tm])); + let snapshot = fluree_db_core::LedgerSnapshot::genesis("test/main"); + let vars = VarRegistry::new(); + let provider = StoreProvider; + + let cancel = QueryCancellation::new(); + cancel.set_memory_limit(1); // 1-byte ceiling → the first window's record crosses it. + let mut ctx = ExecutionContext::new(&snapshot, &vars).with_cancellation(cancel); + ctx.r2rml_table_provider = Some(&provider); + + // `?s ?p ?o` — the true-wildcard crawl the blind spot lived on. + let pattern = + R2rmlPattern::new("gs:main", VarId(0), Some(VarId(2))).with_predicate_var(VarId(1)); + let mut op = R2rmlScanOperator::new(Box::new(EmptyOperator::new()), pattern); + op.mapping = Some(Arc::clone(&mapping)); + + let mut progress = op + .build_progress(&ctx, Batch::single_empty()) + .await + .expect("build_progress") + .expect("wildcard crawl resolves the one TriplesMap"); + let num_cols = op.schema().len(); + let mut columns: Vec> = (0..num_cols).map(|_| Vec::new()).collect(); + let err = op + .advance_one_window(&ctx, &mut progress, num_cols, &mut columns) + .await + .expect_err("the materialized window must trip the 1-byte budget"); + assert!( + matches!(err, QueryError::MemoryBudgetExceeded { .. }), + "wide crawl window must abort typed, got {err:?}" + ); + } + + /// F-AUD-3 site A2 — the fact-as-parent build. `build_parent_lookup` transiently + /// materializes a full parent-sized map; with a RefObjectMap whose parent is a + /// FACT table this is tens of millions of entries, unbounded by the memo cap. + /// The per-batch accounting + checkpoint makes a budget-exceeding build abort + /// TYPED before the whole map is resident (a 1-byte ceiling trips on the first + /// batch here) instead of OOMing. + #[test] + fn r3b_parent_build_budget_aborts_typed() { + use crate::var_registry::VarRegistry; + use fluree_db_core::{LedgerSnapshot, QueryCancellation}; + use fluree_db_r2rml::mapping::TriplesMap; + + let snapshot = LedgerSnapshot::genesis("test/main"); + let vars = VarRegistry::new(); + let cancel = QueryCancellation::new(); + cancel.set_memory_limit(1); // 1-byte ceiling. + let ctx = ExecutionContext::new(&snapshot, &vars).with_cancellation(cancel); + + let parent_tm = TriplesMap::new("#Customer", "customers") + .with_subject_template("http://ex/customer/{ID}"); + // One batch of rows is enough — the per-batch record_alloc crosses the ceiling + // before any row is inserted, so the build aborts on the first batch. + let batch = single_col_batch("ID", vec![Some(1), Some(2), Some(3)]); + let err = build_parent_lookup(&ctx, &parent_tm, &["ID".to_string()], vec![batch]) + .expect_err("the fact-parent build must trip the 1-byte budget"); + assert!( + matches!(err, QueryError::MemoryBudgetExceeded { .. }), + "fact-parent build must abort typed, got {err:?}" + ); + } + /// PR-5 soundness: the top-k pushdown must be declined whenever the scan /// carries a residual filter the operator enforces after emitting rows — /// otherwise the heap is fed pre-filter values and could prune files whose @@ -3607,12 +3782,7 @@ mod tests { #[test] fn f16_folded_wildcard_all_scalar_gate() { use crate::r2rml::{ObjectConstant, ScanValue}; - let scalar = |p: &str| { - ( - p.to_string(), - ObjectConstant::Scalar(ScanValue::Int(7)), - ) - }; + let scalar = |p: &str| (p.to_string(), ObjectConstant::Scalar(ScanValue::Int(7))); let iri = |p: &str| { ( p.to_string(), @@ -3633,8 +3803,7 @@ mod tests { // Mixed scalar + IRI → disqualified (every constraint must be scalar). let mut mixed = pass.clone(); - mixed.star_constraints = - vec![scalar("http://ex/lineNumber"), iri("http://ex/order")]; + mixed.star_constraints = vec![scalar("http://ex/lineNumber"), iri("http://ex/order")]; assert!(!R2rmlScanOperator::folded_wildcard_all_scalar(&mixed)); // Fixed-predicate star member present → not the crawl fold. @@ -3667,8 +3836,8 @@ mod tests { use crate::r2rml::{ObjectConstant, ScanValue}; // The ref side: single-column templated FK — shortcut-eligible (the // deployed `edw:order` → FactOrder shape). - let parent = - TriplesMap::new("#Order", "FACT_ORDER").with_subject_template("http://ex/order/{ORDER_KEY}"); + let parent = TriplesMap::new("#Order", "FACT_ORDER") + .with_subject_template("http://ex/order/{ORDER_KEY}"); let rom = RefObjectMap::new("#Order", "ORDER_KEY", "ORDER_KEY"); assert!( build_ref_shortcut(&parent, &rom).is_some(), @@ -3683,7 +3852,9 @@ mod tests { "http://ex/lineNumber".to_string(), ObjectConstant::Scalar(ScanValue::Int(1)), )]; - assert!(R2rmlScanOperator::ref_template_shortcut_enabled(true, &folded)); + assert!(R2rmlScanOperator::ref_template_shortcut_enabled( + true, &folded + )); // Iri-folded → admission false (sound parent-scan path; catch #11). let mut iri_folded = folded.clone(); @@ -3705,7 +3876,9 @@ mod tests { // unchanged by the F-16 amendment). let plain = R2rmlPattern::new("gs:main", VarId(0), Some(VarId(2))).with_predicate_var(VarId(1)); - assert!(R2rmlScanOperator::ref_template_shortcut_enabled(true, &plain)); + assert!(R2rmlScanOperator::ref_template_shortcut_enabled( + true, &plain + )); // Fixed-predicate star + trust ON → admission false (star shapes keep // parent-scan + dangling-FK semantics; unchanged). @@ -3714,7 +3887,9 @@ mod tests { "http://ex/lineNumber".to_string(), ObjectConstant::Scalar(ScanValue::Int(1)), )]; - assert!(!R2rmlScanOperator::ref_template_shortcut_enabled(true, &star)); + assert!(!R2rmlScanOperator::ref_template_shortcut_enabled( + true, &star + )); } /// A templated (non-constant) predicate binds `?p` from the row when the From cfd773d75f77e6551d223ec158349c3086171c0c Mon Sep 17 00:00:00 2001 From: Andrew Johnson Date: Sat, 18 Jul 2026 23:59:24 -0400 Subject: [PATCH 3/3] fix(query): release scan-window budget charges on hand-off (q038 false-abort) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The live re-bless caught a regression from the A1 scan-window accounting: q038 (a 36M-row un-fused COUNT on the per-row materialize path) false-aborted typed at 38s ("Query memory budget exceeded: ~8.61 GB > 8.59 GB") while completing fine at 52.5s with FLUREE_SCAN_MEM_ACCOUNTING=off and bounded resident memory. Root cause is the documented cumulative-no-decrement edge (V2): ~70 sequentially-FREED 512K-row scan windows SUM past the budget even though only one window is ever resident — a false-positive typed abort on exactly the long-scan class the accounting protects. Fix — window-scoped release: - add QueryCancellation::release (fluree-db-core) + the ExecutionContext::release wrapper: a saturating decrement of the budget counter, valid ONLY for allocations with a provable drop point. Documented caller invariant: never release a persistent allocation (the guard would then under-count live memory). - pair each A1 window charge in advance_one_window with a release once the window is emitted/handed off (`produced` drops), so a streaming scan accounts only its resident window, not the all-time sum. Charge + checkpoint still happen BEFORE emit, so an oversized single window — or this window atop a retained A2 build or an upstream fold — still aborts typed. - A2 fact-parent build charge stays cumulative (that map genuinely persists); fold/join/fused accounting untouched (their buffers persist too). Regression test r3b_scan_windows_release_no_false_abort: 64 one-row windows under an 8000-byte ceiling COMPLETE (verified to fail pre-fix at window ~16 with MemoryBudgetExceeded 8448 > 8000). The single-window abort (071cd59f) and parent-build abort tests still pass. New core tests cover release saturating-sub + disabled-handle no-op. --- fluree-db-core/src/cancellation.rs | 63 +++++++++- fluree-db-query/src/context.rs | 10 ++ fluree-db-query/src/r2rml/operator.rs | 175 +++++++++++++++++++++++--- 3 files changed, 224 insertions(+), 24 deletions(-) diff --git a/fluree-db-core/src/cancellation.rs b/fluree-db-core/src/cancellation.rs index be4a68ee92..7d68be95c3 100644 --- a/fluree-db-core/src/cancellation.rs +++ b/fluree-db-core/src/cancellation.rs @@ -65,9 +65,12 @@ impl fmt::Display for QueryCancellationReason { struct QueryCancellationInner { reason: AtomicU8, /// Bytes of retained query memory recorded via [`QueryCancellation::record_alloc`]. - /// A monotonic, deliberately-conservative high-water accumulator: callers record - /// where a retained buffer grows, so the total tracks a query's live post-scan - /// memory across all its operators. + /// A deliberately-conservative accumulator: callers record where a retained buffer + /// grows, so the total tracks a query's live post-scan memory across all its + /// operators. Mostly grows (persistent join/aggregate builds, retained lookups); + /// an allocation with a PROVABLE drop point (a scan window handed off and dropped) + /// may be paired with [`QueryCancellation::release`], so this is a high-water of + /// *live* accounted memory, not a monotonic all-time sum. allocated: AtomicUsize, /// Optional per-query memory ceiling in bytes (`NO_MEMORY_LIMIT` = unset). Stored /// as an opaque number — this crate never compares or enforces it; the query @@ -152,8 +155,10 @@ impl QueryCancellation { /// Record `bytes` of retained query memory into the shared counter. Callers record /// at the points where a retained buffer (a hash-join build table, a GROUP BY map, /// a fused dim-map) grows, so the counter tracks the query's live post-scan memory. - /// Monotonic and intentionally conservative — over-counting can only trip the guard - /// on a query already near its ceiling. No-op on a disabled handle. + /// Intentionally conservative — over-counting can only trip the guard on a query + /// already near its ceiling. Persistent allocations are never released; an + /// allocation with a provable drop point is paired with [`release`](Self::release). + /// No-op on a disabled handle. #[inline] pub fn record_alloc(&self, bytes: usize) { if let Some(inner) = &self.inner { @@ -161,6 +166,26 @@ impl QueryCancellation { } } + /// Release `bytes` previously recorded via [`record_alloc`](Self::record_alloc), + /// for an allocation whose drop point the caller can prove — e.g. a scan window + /// that has been handed off downstream and is about to drop, so a streaming scan + /// of N sequentially-freed windows accounts only the resident one rather than + /// their all-time sum. Saturating, so a double- or over-release can never + /// underflow the counter. It is a CALLER INVARIANT that `release` is paired only + /// with a matching non-persistent `record_alloc`: persistent buffers (join / + /// aggregate builds, retained lookups) must never be released, or the guard would + /// under-count live memory. No-op on a disabled handle. + #[inline] + pub fn release(&self, bytes: usize) { + if let Some(inner) = &self.inner { + let _ = inner + .allocated + .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |cur| { + Some(cur.saturating_sub(bytes)) + }); + } + } + /// Bytes recorded via [`record_alloc`](Self::record_alloc) so far (0 on a disabled /// handle). #[inline] @@ -261,6 +286,34 @@ mod tests { assert_eq!(derived.allocated_bytes(), 150); } + #[test] + fn release_subtracts_and_saturates_at_zero() { + let cancellation = QueryCancellation::new(); + let derived = cancellation.clone(); + + cancellation.record_alloc(1000); + // A provable-drop allocation is released — the counter tracks LIVE memory, so + // charging then releasing nets zero (a streaming scan's per-window pattern). + cancellation.release(400); + assert_eq!(cancellation.allocated_bytes(), 600); + assert_eq!( + derived.allocated_bytes(), + 600, + "release is shared across clones" + ); + + // Over-release saturates at zero rather than underflowing. + derived.release(10_000); + assert_eq!(cancellation.allocated_bytes(), 0); + } + + #[test] + fn disabled_handle_release_is_a_noop() { + let cancellation = QueryCancellation::disabled(); + cancellation.release(1 << 20); + assert_eq!(cancellation.allocated_bytes(), 0); + } + #[test] fn memory_limit_defaults_unset_and_round_trips() { let cancellation = QueryCancellation::new(); diff --git a/fluree-db-query/src/context.rs b/fluree-db-query/src/context.rs index 46e751095d..98fd55b2de 100644 --- a/fluree-db-query/src/context.rs +++ b/fluree-db-query/src/context.rs @@ -806,6 +806,16 @@ impl<'a> ExecutionContext<'a> { self.cancellation.record_alloc(bytes); } + /// Release `bytes` previously recorded via [`record_alloc`](Self::record_alloc) + /// for an allocation with a provable drop point (e.g. a materialized scan window + /// that has been emitted and is about to drop). Saturating. Only valid for + /// non-persistent allocations — persistent join/aggregate/lookup buffers must + /// never be released. See [`QueryCancellation::release`]. + #[inline] + pub fn release(&self, bytes: usize) { + self.cancellation.release(bytes); + } + /// Retained query memory recorded so far via [`record_alloc`](Self::record_alloc). #[inline] pub fn mem_used(&self) -> usize { diff --git a/fluree-db-query/src/r2rml/operator.rs b/fluree-db-query/src/r2rml/operator.rs index 53d5dd6b34..377298b6ee 100644 --- a/fluree-db-query/src/r2rml/operator.rs +++ b/fluree-db-query/src/r2rml/operator.rs @@ -245,12 +245,13 @@ fn topk_pushdown_enabled() -> bool { /// into, which the hash-join / group-aggregate `record_alloc`s never covered. /// OFF is a clean revert to the prior behavior (scan path invisible to the budget; /// the `checkpoint()` polls degrade to pure cancellation checks because the scan -/// records nothing). The counter is query-lifetime cumulative and never -/// decremented, so a genuinely-streaming large scan is accounted as if all its -/// windows were simultaneously resident — the intended conservative direction -/// (over-count only ever aborts a query already near the budget), matching the -/// existing fold/join accounting; a high-water refinement is deferred with the -/// excluded per-file buffer accounting (V2 site B). +/// records nothing). Each materialized window is charged then RELEASED once emitted +/// (it has a provable drop point), so a long streaming scan accounts only its +/// resident window rather than the all-time sum of every freed window — that sum +/// otherwise false-aborts a bounded-memory long scan (q038). The retained parent-map +/// build (A2) and the per-file buffers (V2 site B, still excluded) are the remaining +/// non-released allocations; A2 genuinely persists so its cumulative charge is +/// correct, and site B needs its own release pairing. fn scan_mem_accounting_enabled() -> bool { static ENABLED: std::sync::OnceLock = std::sync::OnceLock::new(); *ENABLED.get_or_init(|| super::env_switch_enabled("FLUREE_SCAN_MEM_ACCOUNTING")) @@ -1459,20 +1460,26 @@ impl R2rmlScanOperator { // F-AUD-3 site A1: account the materialized window against the query // memory budget so a wide non-aggregate crawl (the previously-blind scan - // path) trips a typed `MemoryBudgetExceeded` instead of OOMing. The window - // is bounded (~`materialize_window_rows`) so one window cannot itself OOM; - // `checkpoint()` here aborts once the cumulative recorded bytes (this - // window plus every prior one and any upstream fold/join) cross the - // budget, before the loop pulls another. Cumulative, never decremented — - // see `scan_mem_accounting_enabled`. - if scan_mem_accounting_enabled() { - let window_est = produced + // path) trips a typed `MemoryBudgetExceeded` instead of OOMing. Charge the + // resident window and `checkpoint()` BEFORE doing more work, so an oversized + // window (or this window on top of a retained A2 fact-parent build / an + // upstream fold) aborts typed. The window is then RELEASED once emitted + // (below) — it has a provable drop point, so a streaming scan of N + // sequentially-freed windows accounts only the resident one instead of + // their all-time sum. Without the release, ~70 freed 512K-row windows of a + // long un-fused-COUNT scan (q038) SUM past the budget and false-abort a + // query whose per-window resident memory is bounded and fine. + let window_est = if scan_mem_accounting_enabled() { + let est = produced .len() .saturating_mul(num_cols) .saturating_mul(crate::context::BINDING_EST_BYTES); - ctx.record_alloc(window_est); + ctx.record_alloc(est); ctx.checkpoint()?; - } + est + } else { + 0 + }; if !produced.is_empty() { emit_produced_window( @@ -1488,6 +1495,16 @@ impl R2rmlScanOperator { ctx, )?; } + + // The window has been handed off (its rows copied into `columns` / the + // bounded `self.pending` overflow) and `produced` drops at the end of this + // iteration — release its charge so only the in-flight window is counted. + // The retained overflow is bounded (≤ one window, drained before the next + // pull) and intentionally left untracked (minimal per V2 site B). The A2 + // fact-parent build charge is NOT released — that map genuinely persists. + if window_est != 0 { + ctx.release(window_est); + } // Geometric window growth. A budgeted (LIMIT) scan starts with a small // window (~the remaining budget) so a selective query does not explode // a full window into bindings before the first output row. But when the @@ -2554,9 +2571,10 @@ fn build_parent_lookup( // worst-case contribution and checkpoint inside the build loop so a // budget-exceeding build aborts typed (`MemoryBudgetExceeded`) BEFORE the // whole map is resident, instead of OOMing. `num_rows` over-counts skipped - // (null-subject / null-key) rows — deliberately conservative. Cumulative like - // site A1; the map is genuinely retained while built, so cumulative == the - // resident footprint for this structure. + // (null-subject / null-key) rows — deliberately conservative. Unlike the A1 + // scan window (released on hand-off), this charge is NOT released: the lookup + // is genuinely retained for the whole join, so the cumulative charge correctly + // equals its resident footprint. if scan_mem_accounting_enabled() { ctx.record_alloc(batch.num_rows.saturating_mul(PARENT_ENTRY_EST_BYTES)); ctx.checkpoint()?; @@ -2907,6 +2925,125 @@ mod tests { ); } + /// F-AUD-3 site A1 — q038 regression (the false-abort the live re-bless caught). + /// A long non-aggregate scan streams many windows that are each materialized then + /// FREED; the per-window budget charge is released on hand-off, so the windows do + /// not SUM to a false over-budget. Here 64 one-row windows (charge ~528 B each) + /// run under an 8000 B ceiling: each resident window fits and the scan COMPLETES — + /// whereas the pre-fix cumulative counter crossed the ceiling within ~16 windows + /// and false-aborted a bounded-memory scan. The single-window abort test above + /// (an oversized window on a 1-byte ceiling) still passes: the checkpoint fires + /// while the window is charged, before it is released. + #[tokio::test] + async fn r3b_scan_windows_release_no_false_abort() { + use crate::r2rml::R2rmlTableProvider; + use crate::seed::EmptyOperator; + use crate::var_registry::VarRegistry; + use fluree_db_core::QueryCancellation; + use fluree_db_r2rml::mapping::{ + CompiledR2rmlMapping, ObjectMap, PredicateMap, PredicateObjectMap, TriplesMap, + }; + use fluree_db_tabular::{BatchSchema, FieldInfo, FieldType}; + + const N_WINDOWS: usize = 64; + + #[derive(Debug)] + struct ManyRowsProvider; + #[async_trait::async_trait] + impl R2rmlTableProvider for ManyRowsProvider { + async fn scan_table( + &self, + _graph_source_id: &str, + _table_name: &str, + _projection: &[String], + _filters: &[crate::r2rml::ScanFilter], + _topk: Option<&crate::r2rml::ScanTopK>, + _as_of_t: Option, + ) -> Result { + let schema = Arc::new(BatchSchema::new(vec![FieldInfo { + name: "STORE_KEY".to_string(), + field_type: FieldType::Int64, + nullable: true, + field_id: 1, + }])); + // N single-row batches → with window_rows pinned to 1, one window each. + let batches: Vec> = (0..N_WINDOWS as i64) + .map(|k| { + Ok(ColumnBatch::new( + Arc::clone(&schema), + vec![Column::Int64(vec![Some(k + 1)])], + ) + .unwrap()) + }) + .collect(); + Ok(Box::pin(futures::stream::iter(batches))) + } + } + + let tm = TriplesMap::new("#Store", "DIM_STORE") + .with_subject_template("http://ex/store/{STORE_KEY}") + .with_class("http://ex/Store") + .with_predicate_object(PredicateObjectMap { + predicate_map: PredicateMap::constant("http://ex/storeKey"), + object_map: ObjectMap::column("STORE_KEY"), + }); + let mapping = Arc::new(CompiledR2rmlMapping::new(vec![tm])); + let snapshot = fluree_db_core::LedgerSnapshot::genesis("test/main"); + let vars = VarRegistry::new(); + let provider = ManyRowsProvider; + + let cancel = QueryCancellation::new(); + // Between one window (~528 B) and the naive cumulative (64×528 ≈ 34 KB): the + // pre-fix counter crosses this within ~16 windows; released windows never do. + cancel.set_memory_limit(8000); + let mut ctx = ExecutionContext::new(&snapshot, &vars).with_cancellation(cancel); + ctx.r2rml_table_provider = Some(&provider); + + let pattern = + R2rmlPattern::new("gs:main", VarId(0), Some(VarId(2))).with_predicate_var(VarId(1)); + let mut op = R2rmlScanOperator::new(Box::new(EmptyOperator::new()), pattern); + op.mapping = Some(Arc::clone(&mapping)); + + let mut progress = op + .build_progress(&ctx, Batch::single_empty()) + .await + .expect("build_progress") + .expect("wildcard crawl resolves the one TriplesMap"); + let num_cols = op.schema().len(); + let mut columns: Vec> = (0..num_cols).map(|_| Vec::new()).collect(); + + let mut windows = 0usize; + loop { + progress.window_rows = 1; // pin one row per window (bypass geometric growth) — env-free. + let more = op + .advance_one_window(&ctx, &mut progress, num_cols, &mut columns) + .await + .expect("a bounded-resident streaming scan must NOT false-abort on the budget"); + // Simulate the consumer draining the emitted batch so nothing accumulates + // outside the released window charge. + for c in &mut columns { + c.clear(); + } + op.pending.clear(); + // The window charge is released on hand-off, so live accounted memory + // never approaches the naive cumulative — it stays within one window. + assert!( + ctx.mem_used() < 8000, + "released window charge must keep live memory under budget, got {}", + ctx.mem_used() + ); + if !more { + break; + } + windows += 1; + assert!(windows < 1000, "safety bound"); + } + assert!( + windows >= N_WINDOWS, + "the whole streaming scan must complete; got {windows} windows" + ); + } + /// F-AUD-3 site A2 — the fact-as-parent build. `build_parent_lookup` transiently /// materializes a full parent-sized map; with a RefObjectMap whose parent is a /// FACT table this is tens of millions of entries, unbounded by the memo cap.