fix(query): close the memory-budget blind spot on the non-aggregate scan path

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,15 +155,37 @@ 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 {
             inner.allocated.fetch_add(bytes, Ordering::Relaxed);
         }
     }
 
+    /// 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();

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::<Binding>()` = 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<str>)`, `IriMatch { iri: Arc<str>, ledger_alias: Arc<str> }`,
+/// or a `Lit` with a `String` value) carries ~50-70 bytes of `Arc<str>` 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::<crate::binding::Binding>();
+/// 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::<crate::binding::Binding>());
 /// 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::<usize>().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
@@ -759,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 {
@@ -1581,3 +1638,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::<Binding>());
+        assert_eq!(
+            std::mem::size_of::<Binding>(),
+            88,
+            "binding.rs:14-17 documents size_of::<Binding>() == 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:?}")
+                }
+            }
+        }
+    }
+}

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 {