Skip to main content

hawdb_executor/
memory_ledger.rs

1// Copyright 2026 Nowledge
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15//! Query-owned hierarchical memory accounting.
16
17use hawdb_core::{
18    HawDBError, ResourceBudgetError, ResourceBudgetKind, ResourceBudgetScope, Result,
19};
20use std::collections::BTreeMap;
21use std::num::NonZeroUsize;
22use std::sync::{Arc, Mutex, MutexGuard};
23
24#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
25pub enum QueryMemoryClass {
26    PipelineBatch,
27    BlockingState,
28    ExternalRead,
29    SpillStaging,
30    MorselOutput,
31    ResultMaterialization,
32}
33
34impl QueryMemoryClass {
35    pub const fn as_str(self) -> &'static str {
36        match self {
37            Self::PipelineBatch => "pipeline_batch",
38            Self::BlockingState => "blocking_state",
39            Self::ExternalRead => "external_read",
40            Self::SpillStaging => "spill_staging",
41            Self::MorselOutput => "morsel_output",
42            Self::ResultMaterialization => "result_materialization",
43        }
44    }
45}
46
47#[derive(Debug, Clone, PartialEq, Eq)]
48pub struct QueryMemoryClassSnapshot {
49    pub class: QueryMemoryClass,
50    pub used_bytes: usize,
51    pub peak_bytes: usize,
52}
53
54#[derive(Debug, Clone, PartialEq, Eq)]
55pub struct QueryMemoryLedgerSnapshot {
56    pub budget_bytes: usize,
57    pub used_bytes: usize,
58    pub peak_bytes: usize,
59    pub account_count: usize,
60    pub classes: Vec<QueryMemoryClassSnapshot>,
61}
62
63#[derive(Debug, Clone)]
64pub struct QueryMemoryLedger {
65    inner: Arc<QueryMemoryLedgerInner>,
66    file_descriptors: Option<hawdb_storage::file_descriptors::FileOpenContext>,
67}
68
69#[derive(Debug)]
70struct QueryMemoryLedgerInner {
71    budget_bytes: usize,
72    budget_option: &'static str,
73    state: Mutex<QueryMemoryLedgerState>,
74}
75
76#[derive(Debug, Default)]
77struct QueryMemoryLedgerState {
78    used_bytes: usize,
79    peak_bytes: usize,
80    next_account_id: u64,
81    accounts: BTreeMap<u64, QueryMemoryAccountState>,
82    classes: BTreeMap<QueryMemoryClass, QueryMemoryClassState>,
83}
84
85#[derive(Debug)]
86struct QueryMemoryAccountState {
87    class: QueryMemoryClass,
88    owner: Arc<str>,
89    budget_bytes: usize,
90    budget_option: &'static str,
91    used_bytes: usize,
92    peak_bytes: usize,
93}
94
95#[derive(Debug, Default)]
96struct QueryMemoryClassState {
97    used_bytes: usize,
98    peak_bytes: usize,
99}
100
101impl QueryMemoryAccountState {
102    fn budget_error(
103        &self,
104        scope: ResourceBudgetScope,
105        requested: usize,
106        used: usize,
107        budget: usize,
108        budget_option: &'static str,
109        message: String,
110    ) -> HawDBError {
111        HawDBError::from(ResourceBudgetError {
112            resource: ResourceBudgetKind::MemoryBytes,
113            scope,
114            owner: self.owner.to_string(),
115            memory_class: Some(self.class.as_str()),
116            requested: requested as u64,
117            used: used as u64,
118            budget: budget as u64,
119            budget_option,
120            message,
121        })
122    }
123
124    fn account_budget_error(&self, requested: usize, message: String) -> HawDBError {
125        self.budget_error(
126            ResourceBudgetScope::Account,
127            requested,
128            self.used_bytes,
129            self.budget_bytes,
130            self.budget_option,
131            message,
132        )
133    }
134}
135
136impl QueryMemoryLedger {
137    /// Creates a query root with the effective host or runtime-granted limit.
138    #[doc(hidden)]
139    pub fn for_query(
140        memory: &crate::ExecutionMemoryConfig,
141        task_context: Option<&hawdb_core::RuntimeTaskContext>,
142    ) -> Result<Self> {
143        Ok(Self::new_with_budget_option(
144            crate::memory::enforced_query_memory_budget(memory, task_context)?,
145            if task_context
146                .and_then(hawdb_core::RuntimeTaskContext::memory_reservation)
147                .is_some()
148            {
149                "RuntimeTaskContext.memory_reservation.memory_bytes"
150            } else {
151                "ExecutionMemoryConfig.query_memory_bytes"
152            },
153        ))
154    }
155
156    pub(crate) fn budget_bytes(&self) -> NonZeroUsize {
157        NonZeroUsize::new(self.inner.budget_bytes).unwrap()
158    }
159
160    pub(crate) fn source_account(
161        &self,
162        owner: &'static str,
163        source_budget: NonZeroUsize,
164        retained_budget: NonZeroUsize,
165    ) -> QueryMemoryAccount {
166        self.source_account_retaining(
167            owner,
168            source_budget,
169            self.account(QueryMemoryClass::BlockingState, owner, retained_budget),
170        )
171    }
172
173    pub(crate) fn source_account_retaining(
174        &self,
175        owner: &'static str,
176        source_budget: NonZeroUsize,
177        retained: QueryMemoryAccount,
178    ) -> QueryMemoryAccount {
179        self.account(QueryMemoryClass::ExternalRead, owner, source_budget)
180            .with_retained_state(retained)
181    }
182
183    pub fn new(budget_bytes: NonZeroUsize) -> Self {
184        Self::new_with_budget_option(budget_bytes, "QueryMemoryLedger::new(budget_bytes)")
185    }
186
187    /// Names the configuration option or runtime grant supplying this limit.
188    pub fn new_with_budget_option(budget_bytes: NonZeroUsize, budget_option: &'static str) -> Self {
189        Self {
190            inner: Arc::new(QueryMemoryLedgerInner {
191                budget_bytes: budget_bytes.get(),
192                budget_option,
193                state: Mutex::new(QueryMemoryLedgerState::default()),
194            }),
195            file_descriptors: None,
196        }
197    }
198
199    /// Bind query IO to its store while retaining the existing memory ledger.
200    #[doc(hidden)]
201    pub fn with_file_descriptors(
202        mut self,
203        context: Option<hawdb_storage::file_descriptors::FileOpenContext>,
204    ) -> Self {
205        self.file_descriptors = context.or(self.file_descriptors);
206        self
207    }
208
209    pub(crate) fn file_descriptor_context(
210        &self,
211    ) -> Option<&hawdb_storage::file_descriptors::FileOpenContext> {
212        self.file_descriptors.as_ref()
213    }
214
215    pub fn account(
216        &self,
217        class: QueryMemoryClass,
218        owner: impl Into<Arc<str>>,
219        budget_bytes: NonZeroUsize,
220    ) -> QueryMemoryAccount {
221        self.account_with_budget_option(
222            class,
223            owner,
224            budget_bytes,
225            "QueryMemoryLedger::account(budget_bytes)",
226        )
227    }
228
229    /// Creates an account with an explicit origin for its effective byte limit.
230    pub fn account_with_budget_option(
231        &self,
232        class: QueryMemoryClass,
233        owner: impl Into<Arc<str>>,
234        budget_bytes: NonZeroUsize,
235        budget_option: &'static str,
236    ) -> QueryMemoryAccount {
237        // Caller code must run before locking: conversion can panic or re-enter the ledger.
238        let owner = owner.into();
239        let mut state = lock_recover(&self.inner.state);
240        let account_id = state.next_account_id;
241        state.next_account_id = state.next_account_id.saturating_add(1);
242        state.accounts.insert(
243            account_id,
244            QueryMemoryAccountState {
245                class,
246                owner,
247                budget_bytes: budget_bytes.get(),
248                budget_option,
249                used_bytes: 0,
250                peak_bytes: 0,
251            },
252        );
253        QueryMemoryAccount {
254            ledger: self.clone(),
255            account_id,
256            backing: None,
257            retained_state: None,
258        }
259    }
260
261    pub fn snapshot(&self) -> QueryMemoryLedgerSnapshot {
262        let state = lock_recover(&self.inner.state);
263        QueryMemoryLedgerSnapshot {
264            budget_bytes: self.inner.budget_bytes,
265            used_bytes: state.used_bytes,
266            peak_bytes: state.peak_bytes,
267            account_count: state.accounts.len(),
268            classes: state
269                .classes
270                .iter()
271                .map(|(class, state)| QueryMemoryClassSnapshot {
272                    class: *class,
273                    used_bytes: state.used_bytes,
274                    peak_bytes: state.peak_bytes,
275                })
276                .collect(),
277        }
278    }
279
280    /// Test observation of the actual owner, rather than unrelated root charges.
281    #[cfg(test)]
282    pub(crate) fn test_owner_used_bytes(&self, class: QueryMemoryClass, owner: &str) -> usize {
283        lock_recover(&self.inner.state)
284            .accounts
285            .values()
286            .filter(|account| account.class == class && account.owner.as_ref() == owner)
287            .map(|account| account.used_bytes)
288            .sum()
289    }
290
291    fn reserve(&self, account_id: u64, bytes: usize) -> Result<()> {
292        if bytes == 0 {
293            return Ok(());
294        }
295        let mut state = lock_recover(&self.inner.state);
296        let account = state.accounts.get(&account_id).ok_or_else(|| {
297            HawDBError::Execution("query memory account is no longer registered".to_string())
298        })?;
299        let class = account.class;
300        let owner = &account.owner;
301        let account_budget = account.budget_bytes;
302        let account_next = account.used_bytes.checked_add(bytes).ok_or_else(|| {
303            account.account_budget_error(
304                bytes,
305                format!(
306                    "query memory account {owner} ({}) byte accounting overflow",
307                    class.as_str()
308                ),
309            )
310        })?;
311        if account_next > account_budget {
312            return Err(account.account_budget_error(bytes, format!(
313                "query memory account {owner} ({}) would use {account_next} bytes, exceeding its {account_budget}-byte budget",
314                class.as_str()
315            )));
316        }
317        let query_error = |message| {
318            account.budget_error(
319                ResourceBudgetScope::Query,
320                bytes,
321                state.used_bytes,
322                self.inner.budget_bytes,
323                self.inner.budget_option,
324                message,
325            )
326        };
327        let root_next = state.used_bytes.checked_add(bytes).ok_or_else(|| {
328            query_error(format!(
329                "query memory ledger byte accounting overflow while charging {owner} ({})",
330                class.as_str()
331            ))
332        })?;
333        if root_next > self.inner.budget_bytes {
334            return Err(query_error(format!(
335                "query memory ledger would use {root_next} bytes while charging {owner} ({}), exceeding query_memory_bytes {}",
336                class.as_str(),
337                self.inner.budget_bytes
338            )));
339        }
340
341        // Complete allocation before changing any of the hierarchical counters.
342        state.classes.entry(class).or_default();
343        let account = state
344            .accounts
345            .get_mut(&account_id)
346            .expect("validated query memory account remains registered");
347        account.used_bytes = account_next;
348        account.peak_bytes = account.peak_bytes.max(account_next);
349        state.used_bytes = root_next;
350        state.peak_bytes = state.peak_bytes.max(root_next);
351        let class_state = state
352            .classes
353            .get_mut(&class)
354            .expect("prepared query memory class remains registered");
355        class_state.used_bytes = class_state.used_bytes.saturating_add(bytes);
356        class_state.peak_bytes = class_state.peak_bytes.max(class_state.used_bytes);
357        Ok(())
358    }
359
360    fn can_reserve(&self, account_id: u64, bytes: usize) -> bool {
361        let state = lock_recover(&self.inner.state);
362        let Some(account) = state.accounts.get(&account_id) else {
363            return false;
364        };
365        let Some(account_next) = account.used_bytes.checked_add(bytes) else {
366            return false;
367        };
368        let Some(root_next) = state.used_bytes.checked_add(bytes) else {
369            return false;
370        };
371        account_next <= account.budget_bytes && root_next <= self.inner.budget_bytes
372    }
373
374    fn release(&self, account_id: u64, bytes: usize) {
375        if bytes == 0 {
376            return;
377        }
378        let mut state = lock_recover(&self.inner.state);
379        let Some(account) = state.accounts.get_mut(&account_id) else {
380            return;
381        };
382        let released = bytes.min(account.used_bytes);
383        let class = account.class;
384        account.used_bytes -= released;
385        state.used_bytes = state.used_bytes.saturating_sub(released);
386        if let Some(class_state) = state.classes.get_mut(&class) {
387            class_state.used_bytes = class_state.used_bytes.saturating_sub(released);
388        }
389    }
390
391    fn transfer(
392        &self,
393        source_account_id: u64,
394        source_bytes: usize,
395        target_account_id: u64,
396        target_bytes: usize,
397    ) -> Result<()> {
398        if source_account_id == target_account_id {
399            return Err(HawDBError::Execution(
400                "query memory transfer requires distinct accounts".to_string(),
401            ));
402        }
403        let mut state = lock_recover(&self.inner.state);
404        let (source_class, source_used) = state
405            .accounts
406            .get(&source_account_id)
407            .map(|account| (account.class, account.used_bytes))
408            .ok_or_else(|| {
409                HawDBError::Execution(
410                    "source query memory account is no longer registered".to_string(),
411                )
412            })?;
413        if source_bytes > source_used {
414            return Err(HawDBError::Execution(format!(
415                "query memory transfer tried to release {source_bytes} bytes from a source account using {source_used} bytes"
416            )));
417        }
418        let target_account = state.accounts.get(&target_account_id).ok_or_else(|| {
419            HawDBError::Execution("target query memory account is no longer registered".to_string())
420        })?;
421        let target_class = target_account.class;
422        let target_owner = &target_account.owner;
423        let target_budget = target_account.budget_bytes;
424        let target_used = target_account.used_bytes;
425        let target_next = target_used.checked_add(target_bytes).ok_or_else(|| {
426            target_account.account_budget_error(
427                target_bytes,
428                format!(
429                    "query memory account {target_owner} ({}) byte accounting overflow",
430                    target_class.as_str()
431                ),
432            )
433        })?;
434        if target_next > target_budget {
435            return Err(target_account.account_budget_error(target_bytes, format!(
436                "query memory account {target_owner} ({}) would use {target_next} bytes, exceeding its {target_budget}-byte budget",
437                target_class.as_str()
438            )));
439        }
440        let root_after_release = state.used_bytes.checked_sub(source_bytes).ok_or_else(|| {
441            HawDBError::Execution(
442                "query memory ledger underflow during ownership transfer".to_string(),
443            )
444        })?;
445        let query_error = |message| {
446            target_account.budget_error(
447                ResourceBudgetScope::Query,
448                target_bytes.saturating_sub(source_bytes),
449                state.used_bytes,
450                self.inner.budget_bytes,
451                self.inner.budget_option,
452                message,
453            )
454        };
455        let root_next = root_after_release.checked_add(target_bytes).ok_or_else(|| {
456            query_error(format!(
457                "query memory ledger byte accounting overflow while transferring ownership to {target_owner} ({})",
458                target_class.as_str()
459            ))
460        })?;
461        if root_next > self.inner.budget_bytes {
462            return Err(query_error(format!(
463                "query memory ledger would use {root_next} bytes while transferring ownership to {target_owner} ({}), exceeding query_memory_bytes {}",
464                target_class.as_str(),
465                self.inner.budget_bytes
466            )));
467        }
468
469        state.classes.entry(target_class).or_default();
470        let source = state
471            .accounts
472            .get_mut(&source_account_id)
473            .expect("validated source query memory account remains registered");
474        source.used_bytes -= source_bytes;
475        let target = state
476            .accounts
477            .get_mut(&target_account_id)
478            .expect("validated target query memory account remains registered");
479        target.used_bytes = target_next;
480        target.peak_bytes = target.peak_bytes.max(target_next);
481        state.used_bytes = root_next;
482        state.peak_bytes = state.peak_bytes.max(root_next);
483        if let Some(class_state) = state.classes.get_mut(&source_class) {
484            class_state.used_bytes = class_state.used_bytes.saturating_sub(source_bytes);
485        }
486        let class_state = state
487            .classes
488            .get_mut(&target_class)
489            .expect("prepared target query memory class remains registered");
490        class_state.used_bytes = class_state.used_bytes.saturating_add(target_bytes);
491        class_state.peak_bytes = class_state.peak_bytes.max(class_state.used_bytes);
492        Ok(())
493    }
494}
495
496#[derive(Debug, Clone)]
497pub struct QueryMemoryAccount {
498    ledger: QueryMemoryLedger,
499    account_id: u64,
500    // A child ledger owns an up-front reservation in its parent. Every account
501    // and outstanding lease keeps that reservation alive.
502    backing: Option<Arc<QueryMemoryLease>>,
503    retained_state: Option<Arc<QueryMemoryAccount>>,
504}
505
506impl QueryMemoryAccount {
507    pub(crate) fn query_budget_bytes(&self) -> NonZeroUsize {
508        self.ledger.budget_bytes()
509    }
510
511    pub(crate) fn with_retained_state(mut self, account: Self) -> Self {
512        self.retained_state = Some(Arc::new(account));
513        self
514    }
515
516    pub(crate) fn retained_state(&self) -> Option<&Self> {
517        self.retained_state.as_deref()
518    }
519
520    /// Reserve the entire child allowance before dispatch, without charging
521    /// individual child allocations to the query root a second time.
522    pub(crate) fn sub_account(&self, budget: NonZeroUsize) -> Result<Self> {
523        let backing = Arc::new(self.reserve(budget.get())?);
524        let (class, owner) = {
525            let state = lock_recover(&self.ledger.inner.state);
526            let account = &state.accounts[&self.account_id];
527            (account.class, Arc::clone(&account.owner))
528        };
529        let origin = "QueryMemoryAccount::sub_account(budget)";
530        let mut child = QueryMemoryLedger::new_with_budget_option(budget, origin)
531            .account_with_budget_option(class, owner, budget, origin);
532        child.backing = Some(backing);
533        Ok(child)
534    }
535
536    pub(crate) fn budget_bytes(&self) -> NonZeroUsize {
537        NonZeroUsize::new(
538            lock_recover(&self.ledger.inner.state).accounts[&self.account_id].budget_bytes,
539        )
540        .unwrap()
541    }
542
543    /// Remaining capacity in both this account and its shared query root.
544    /// This is a sizing observation; the following reservation still arbitrates
545    /// concurrent owners and can fail without advancing the producer.
546    #[doc(hidden)]
547    pub fn available_bytes(&self) -> usize {
548        let state = lock_recover(&self.ledger.inner.state);
549        let account = &state.accounts[&self.account_id];
550        account.budget_bytes.saturating_sub(account.used_bytes).min(
551            self.ledger
552                .inner
553                .budget_bytes
554                .saturating_sub(state.used_bytes),
555        )
556    }
557
558    pub(crate) fn peak_bytes(&self) -> usize {
559        lock_recover(&self.ledger.inner.state).accounts[&self.account_id].peak_bytes
560    }
561
562    pub(crate) fn query_available_bytes(&self) -> usize {
563        let state = lock_recover(&self.ledger.inner.state);
564        self.ledger
565            .inner
566            .budget_bytes
567            .saturating_sub(state.used_bytes)
568    }
569
570    pub(crate) fn sibling(
571        &self,
572        class: QueryMemoryClass,
573        owner: impl Into<Arc<str>>,
574        budget_bytes: NonZeroUsize,
575    ) -> Self {
576        self.sibling_with_budget_option(
577            class,
578            owner,
579            budget_bytes,
580            "QueryMemoryAccount::sibling(budget_bytes)",
581        )
582    }
583
584    pub(crate) fn sibling_with_budget_option(
585        &self,
586        class: QueryMemoryClass,
587        owner: impl Into<Arc<str>>,
588        budget_bytes: NonZeroUsize,
589        budget_option: &'static str,
590    ) -> Self {
591        let mut account =
592            self.ledger
593                .account_with_budget_option(class, owner, budget_bytes, budget_option);
594        account.backing = self.backing.clone();
595        account
596    }
597
598    pub(crate) fn can_reserve(&self, bytes: usize) -> bool {
599        self.ledger.can_reserve(self.account_id, bytes)
600    }
601
602    pub fn reserve(&self, bytes: usize) -> Result<QueryMemoryLease> {
603        self.ledger.reserve(self.account_id, bytes)?;
604        Ok(QueryMemoryLease {
605            account: self.clone(),
606            bytes,
607        })
608    }
609}
610
611#[derive(Debug)]
612pub struct QueryMemoryLease {
613    account: QueryMemoryAccount,
614    bytes: usize,
615}
616
617impl QueryMemoryLease {
618    pub(crate) fn operator_budget_error(
619        &self,
620        requested: usize,
621        used: usize,
622        budget: usize,
623        budget_option: &'static str,
624        message: String,
625    ) -> HawDBError {
626        let state = lock_recover(&self.account.ledger.inner.state);
627        state.accounts[&self.account.account_id].budget_error(
628            ResourceBudgetScope::Operator,
629            requested,
630            used,
631            budget,
632            budget_option,
633            message,
634        )
635    }
636
637    pub(crate) fn belongs_to(&self, account: &QueryMemoryAccount) -> bool {
638        self.account.account_id == account.account_id
639            && Arc::ptr_eq(&self.account.ledger.inner, &account.ledger.inner)
640    }
641
642    pub fn bytes(&self) -> usize {
643        self.bytes
644    }
645
646    /// Transfers part of this already admitted charge to an independent owner.
647    /// The ledger charge never disappears or grows during the handoff.
648    pub(crate) fn split_off(&mut self, bytes: usize) -> Result<Self> {
649        if bytes > self.bytes {
650            return Err(HawDBError::Execution(
651                "query memory lease split exceeds its admitted charge".into(),
652            ));
653        }
654        self.bytes -= bytes;
655        Ok(Self {
656            account: self.account.clone(),
657            bytes,
658        })
659    }
660
661    pub fn grow(&mut self, bytes: usize) -> Result<()> {
662        self.account
663            .ledger
664            .reserve(self.account.account_id, bytes)?;
665        self.bytes = self.bytes.saturating_add(bytes);
666        Ok(())
667    }
668
669    pub fn shrink(&mut self, bytes: usize) {
670        let released = bytes.min(self.bytes);
671        self.account
672            .ledger
673            .release(self.account.account_id, released);
674        self.bytes -= released;
675    }
676
677    pub fn reset(&mut self) {
678        self.shrink(self.bytes);
679    }
680
681    pub(crate) fn transfer_to(
682        &mut self,
683        source_bytes: usize,
684        target: &mut Self,
685        target_bytes: usize,
686    ) -> Result<()> {
687        if !Arc::ptr_eq(&self.account.ledger.inner, &target.account.ledger.inner) {
688            return Err(HawDBError::Execution(
689                "query memory transfer requires accounts from the same ledger".to_string(),
690            ));
691        }
692        if source_bytes > self.bytes {
693            return Err(HawDBError::Execution(format!(
694                "query memory transfer tried to release {source_bytes} bytes from a {}-byte lease",
695                self.bytes,
696            )));
697        }
698        self.account.ledger.transfer(
699            self.account.account_id,
700            source_bytes,
701            target.account.account_id,
702            target_bytes,
703        )?;
704        self.bytes -= source_bytes;
705        target.bytes = target.bytes.saturating_add(target_bytes);
706        Ok(())
707    }
708}
709
710impl Drop for QueryMemoryLease {
711    fn drop(&mut self) {
712        self.account
713            .ledger
714            .release(self.account.account_id, self.bytes);
715        self.bytes = 0;
716    }
717}
718
719fn lock_recover<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
720    // State never escapes the ledger. Checks and allocation precede counter updates,
721    // and no caller code runs under this lock. Recovering poisoning preserves that
722    // valid state; it is not a repair mechanism for arbitrary accounting corruption.
723    mutex
724        .lock()
725        .unwrap_or_else(|poisoned| poisoned.into_inner())
726}
727
728#[cfg(test)]
729#[path = "memory_ledger_hardening_tests.rs"]
730mod hardening_tests;
731
732#[cfg(test)]
733#[path = "memory_ledger_budget_tests.rs"]
734mod budget_tests;
735
736#[cfg(test)]
737mod tests {
738    use super::*;
739
740    #[test]
741    fn splitting_a_full_lease_preserves_charge_and_independent_drop() {
742        let budget = NonZeroUsize::new(8).unwrap();
743        let ledger = QueryMemoryLedger::new(budget);
744        let account = ledger.account(QueryMemoryClass::ResultMaterialization, "retained", budget);
745        let mut owner = account.reserve(8).unwrap();
746        assert!(owner.split_off(9).is_err());
747        assert_eq!(owner.bytes(), 8);
748        let view = owner.split_off(3).unwrap();
749        assert_eq!(owner.bytes(), 5);
750        assert_eq!(view.bytes(), 3);
751        assert_eq!(ledger.snapshot().used_bytes, 8);
752        assert_eq!(ledger.snapshot().peak_bytes, 8);
753        drop(owner);
754        assert_eq!(ledger.snapshot().used_bytes, 3);
755        drop(view);
756        assert_eq!(ledger.snapshot().used_bytes, 0);
757    }
758
759    #[test]
760    fn source_account_reuses_retained_cap_without_registering_an_unused_account() {
761        let ledger = QueryMemoryLedger::new(NonZeroUsize::new(1024).unwrap());
762        let retained = ledger.account(
763            QueryMemoryClass::BlockingState,
764            "shared state",
765            NonZeroUsize::new(512).unwrap(),
766        );
767        let source =
768            ledger.source_account_retaining("source", ledger.budget_bytes(), retained.clone());
769        assert_eq!(ledger.snapshot().account_count, 2);
770        let mut grant = crate::store::admit_graph_read(&source, None, 512).unwrap();
771        grant.retain_state().unwrap();
772        assert!(retained.reserve(1).is_err());
773        assert_eq!(ledger.snapshot().used_bytes, 512);
774        drop(grant);
775        let reused = retained.reserve(512).unwrap();
776        assert_eq!(ledger.snapshot().account_count, 2);
777        drop(reused);
778        assert_eq!(ledger.snapshot().used_bytes, 0);
779    }
780
781    #[test]
782    fn sub_accounts_reserve_parent_capacity_without_double_charging() {
783        let budget = NonZeroUsize::new(100).unwrap();
784        let ledger = QueryMemoryLedger::new(budget);
785        let parent = ledger.account(QueryMemoryClass::BlockingState, "merge", budget);
786        let retained = parent.reserve(10).unwrap();
787        let left = parent.sub_account(NonZeroUsize::new(60).unwrap()).unwrap();
788        let right = parent.sub_account(NonZeroUsize::new(30).unwrap()).unwrap();
789        assert_eq!(parent.available_bytes(), 0);
790        assert!(parent.sub_account(NonZeroUsize::MIN).is_err());
791        let mut left_lease = left.reserve(60).unwrap();
792        let right_lease = right.reserve(30).unwrap();
793        assert!(left_lease.grow(1).is_err());
794        assert_eq!(ledger.snapshot().used_bytes, 100);
795        drop((left, right, retained));
796        assert_eq!(ledger.snapshot().used_bytes, 90);
797        drop(left_lease);
798        assert_eq!(ledger.snapshot().used_bytes, 30);
799        drop(right_lease);
800        assert_eq!(ledger.snapshot().used_bytes, 0);
801        assert_eq!(ledger.snapshot().peak_bytes, 100);
802    }
803
804    #[test]
805    fn nested_accounts_and_siblings_keep_backing_until_last_owner_drops() {
806        let budget = NonZeroUsize::new(8).unwrap();
807        let ledger = QueryMemoryLedger::new(budget);
808        let parent = ledger.account(QueryMemoryClass::BlockingState, "merge", budget);
809        let child = parent.sub_account(budget).unwrap();
810        let sibling = child.sibling(QueryMemoryClass::SpillStaging, "staging", budget);
811        let nested = child.sub_account(NonZeroUsize::new(4).unwrap()).unwrap();
812        assert!(sibling.reserve(5).is_err());
813        let lease = nested.reserve(4).unwrap();
814        drop((parent, child, nested));
815        assert_eq!(ledger.snapshot().used_bytes, 8);
816        drop(lease);
817        assert_eq!(ledger.snapshot().used_bytes, 8);
818        let sibling_lease = sibling.reserve(8).unwrap();
819        drop(sibling);
820        assert_eq!(ledger.snapshot().used_bytes, 8);
821        drop(sibling_lease);
822        assert_eq!(ledger.snapshot().used_bytes, 0);
823    }
824
825    #[test]
826    fn failed_sub_account_admission_does_not_leak_root_capacity() {
827        let budget = NonZeroUsize::new(100).unwrap();
828        let ledger = QueryMemoryLedger::new(budget);
829        let parent = ledger.account(QueryMemoryClass::BlockingState, "merge", budget);
830        let other = ledger.account(QueryMemoryClass::PipelineBatch, "input", budget);
831        let retained = other.reserve(70).unwrap();
832        assert!(parent.sub_account(NonZeroUsize::new(31).unwrap()).is_err());
833        assert_eq!(ledger.snapshot().used_bytes, 70);
834        let child = parent.sub_account(NonZeroUsize::new(30).unwrap()).unwrap();
835        assert!(child.reserve(31).is_err());
836        drop((child, retained));
837        assert_eq!(ledger.snapshot().used_bytes, 0);
838    }
839
840    #[test]
841    fn sibling_accounts_share_one_root_budget() {
842        let ledger = QueryMemoryLedger::new(NonZeroUsize::new(10).unwrap());
843        let left = ledger.account(
844            QueryMemoryClass::BlockingState,
845            "left",
846            NonZeroUsize::new(10).unwrap(),
847        );
848        let right = ledger.account(
849            QueryMemoryClass::PipelineBatch,
850            "right",
851            NonZeroUsize::new(10).unwrap(),
852        );
853        let left_lease = left.reserve(6).unwrap();
854        let error = right.reserve(5).unwrap_err();
855
856        assert!(error.to_string().contains("query_memory_bytes 10"));
857        assert_eq!(ledger.snapshot().used_bytes, 6);
858        drop(left_lease);
859        assert_eq!(ledger.snapshot().used_bytes, 0);
860    }
861
862    #[test]
863    fn capacity_preflight_checks_account_and_root_without_charging() {
864        let ledger = QueryMemoryLedger::new(NonZeroUsize::new(10).unwrap());
865        let left = ledger.account(
866            QueryMemoryClass::BlockingState,
867            "left",
868            NonZeroUsize::new(8).unwrap(),
869        );
870        let right = ledger.account(
871            QueryMemoryClass::PipelineBatch,
872            "right",
873            NonZeroUsize::new(10).unwrap(),
874        );
875
876        assert!(left.can_reserve(8));
877        assert_eq!(ledger.snapshot().used_bytes, 0);
878        let left_lease = left.reserve(6).unwrap();
879        assert!(!left.can_reserve(3));
880        assert!(right.can_reserve(4));
881        assert!(!right.can_reserve(5));
882        assert_eq!(ledger.snapshot().used_bytes, 6);
883
884        drop(left_lease);
885        assert_eq!(ledger.snapshot().used_bytes, 0);
886    }
887
888    #[test]
889    fn derived_sibling_account_keeps_the_original_query_root() {
890        let budget = NonZeroUsize::new(8).unwrap();
891        let ledger = QueryMemoryLedger::new(budget);
892        let state = ledger.account(QueryMemoryClass::BlockingState, "merge", budget);
893        let output = state.sibling(QueryMemoryClass::PipelineBatch, "output", budget);
894        let state_lease = state.reserve(6).unwrap();
895        assert!(output
896            .reserve(3)
897            .unwrap_err()
898            .to_string()
899            .contains("query_memory_bytes 8"));
900        let output_lease = output.reserve(2).unwrap();
901        assert_eq!(ledger.snapshot().used_bytes, 8);
902        drop((state_lease, output_lease));
903        assert_eq!(ledger.snapshot().used_bytes, 0);
904    }
905
906    #[test]
907    fn ownership_transfer_does_not_require_double_root_capacity() {
908        let budget = NonZeroUsize::new(8).unwrap();
909        let ledger = QueryMemoryLedger::new(budget);
910        let source = ledger.account(QueryMemoryClass::BlockingState, "source", budget);
911        let target = ledger.account(QueryMemoryClass::PipelineBatch, "target", budget);
912        let mut source_lease = source.reserve(8).unwrap();
913        let mut target_lease = target.reserve(0).unwrap();
914
915        source_lease.transfer_to(8, &mut target_lease, 8).unwrap();
916
917        assert_eq!(source_lease.bytes(), 0);
918        assert_eq!(target_lease.bytes(), 8);
919        let snapshot = ledger.snapshot();
920        assert_eq!(snapshot.used_bytes, 8);
921        assert_eq!(snapshot.peak_bytes, 8);
922        drop((source_lease, target_lease));
923        assert_eq!(ledger.snapshot().used_bytes, 0);
924    }
925
926    #[test]
927    fn failed_ownership_transfer_preserves_source_charge() {
928        let budget = NonZeroUsize::new(8).unwrap();
929        let ledger = QueryMemoryLedger::new(budget);
930        let source = ledger.account(QueryMemoryClass::BlockingState, "source", budget);
931        let target = ledger.account(QueryMemoryClass::PipelineBatch, "target", budget);
932        let mut source_lease = source.reserve(8).unwrap();
933        let mut target_lease = target.reserve(0).unwrap();
934
935        let error = source_lease
936            .transfer_to(8, &mut target_lease, 9)
937            .unwrap_err();
938
939        assert!(error.to_string().contains("9 bytes"));
940        assert_eq!(source_lease.bytes(), 8);
941        assert_eq!(target_lease.bytes(), 0);
942        assert_eq!(ledger.snapshot().used_bytes, 8);
943    }
944
945    #[test]
946    fn account_budget_is_enforced_before_root_budget() {
947        let ledger = QueryMemoryLedger::new(NonZeroUsize::new(100).unwrap());
948        let account = ledger.account(
949            QueryMemoryClass::BlockingState,
950            "sort",
951            NonZeroUsize::new(4).unwrap(),
952        );
953
954        let error = account.reserve(5).unwrap_err();
955
956        assert!(error.to_string().contains("sort"));
957        assert!(error.to_string().contains("4-byte budget"));
958        assert_eq!(ledger.snapshot().used_bytes, 0);
959    }
960
961    #[test]
962    fn lease_growth_shrink_and_drop_update_hierarchy() {
963        let ledger = QueryMemoryLedger::new(NonZeroUsize::new(100).unwrap());
964        let account = ledger.account(
965            QueryMemoryClass::MorselOutput,
966            "reorder",
967            NonZeroUsize::new(80).unwrap(),
968        );
969        let mut lease = account.reserve(20).unwrap();
970        lease.grow(30).unwrap();
971        lease.shrink(10);
972        assert_eq!(lease.bytes(), 40);
973        assert_eq!(ledger.snapshot().used_bytes, 40);
974
975        drop(lease);
976        let snapshot = ledger.snapshot();
977        assert_eq!(snapshot.used_bytes, 0);
978        assert_eq!(snapshot.peak_bytes, 50);
979        assert_eq!(snapshot.classes[0].peak_bytes, 50);
980    }
981
982    #[test]
983    fn unwind_releases_query_memory_lease() {
984        let ledger = QueryMemoryLedger::new(NonZeroUsize::new(100).unwrap());
985        let account = ledger.account(
986            QueryMemoryClass::SpillStaging,
987            "spill",
988            NonZeroUsize::new(100).unwrap(),
989        );
990        let result = std::panic::catch_unwind(|| {
991            let _lease = account.reserve(80).unwrap();
992            panic!("injected panic");
993        });
994
995        assert!(result.is_err());
996        assert_eq!(ledger.snapshot().used_bytes, 0);
997    }
998}