1use 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 #[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 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 #[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 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 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 #[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 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 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 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 #[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 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 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}