Bounded memory in action

The arbitrator built the shared, interior-mutable state. This one is about the decisions it makes with that state: when memory gets tight, does it spill a buffer, pause a source, or abort the run? And, the part that saves you from a wedged job, how does it refuse a budget it can never satisfy, before a single record moves? The answer threads back through two earlier ideas: a swappable strategy (where a trait object is finally the right call) and the fail-fast error taxonomy from Planning & Expressions. By the end you’ll have written the gate-and-policy machinery yourself and watched it pick a victim under pressure.

Prerequisites

• The memory arbitrator

→

Leads to

• FieldStr

What you’ll be able to do

• Explain the spill/pause/abort decision as a set of boolean gates the arbitrator exposes, not one verdict enum.
• Trace what should_spill checks (peak RSS or summed charged bytes against the soft limit) and what it triggers when it trips.
• Write a runtime-chosen ArbitrationPolicy behind a trait object and observe two policies pick different victims under the same pressure.
• Distinguish why operators are a closed enum while the memory policy is a trait object: closed-and-owned vs open-and-configured.
• Explain the unsatisfiable-budget rejection: why a limit below baseline RSS deadlocks a pausing policy, and why it must fail before the run starts.

New terms in this lesson (each is also expanded inline at the point you first need it):

• bounded memory: the run-stays-under-a-cap guarantee this whole machine exists to enforce.
• spill: move records to disk so they stop costing RAM (you met the mechanism in Node buffers & spill).
• soft / hard limit: the two thresholds (a soft cap that triggers spill/pause, a hard cap that triggers abort).
• streaming: the default flow where a record comes in and a record goes out, holding nothing.

Predict first

✋ Predict first

The arbitrator has to express three different responses to memory pressure: spill a buffer, pause a source, or abort the run. Predict the shape it uses. Is it one enum like `enum Decision { Spill, Pause, Abort }` returned from a single function, or something else? Write down your guess before revealing.

Decide on your answer before you open the reveal — committing a guess is what makes it stick.

Reveal the answer

It is not one verdict enum. The arbitrator exposes a few boolean gates that operators poll on their own, chiefly should_spill (soft limit) and should_abort (hard limit). The spill-vs-pause choice is made separately, inside the arbitration round, by a swappable policy.

So the three responses come from three different places:

• Spill / pause, decided per-operator inside one arbitration round, by the policy.
• Abort, a separate should_abort gate on the hard limit, checked by callers.

The trap is imagining a single decide() -> Decision chokepoint. There isn’t one, and the reason is concurrency: many operators across many threads poll the gates independently, and a policy (not a fixed enum) gets to choose which operator to act on, so the strategy can change with config without rewriting the gates. Hold that distinction; the rest of the lesson earns it.

The decision is a set of gates, not one verdict

There’s no single enum { Spill, Pause, Abort }. The arbitrator exposes a few boolean gates that operators poll. The central one is should_spill, checked at every bulk admission:

clinker-exec ·memory.rs ·should_spill fn @19acdcb4

pub fn should_spill(&self) -> bool {
    self.observe();                          // refresh peak RSS
    let soft = self.soft_limit();
    let tripped = self.peak_rss.load(Ordering::Relaxed) > soft
        || self.sum_consumer_usage() > soft; // RSS OR summed charged bytes
    if tripped {
        self.poll_arbitration();             // run ONE arbitration round
    }
    tripped
}

Two things worth noting. It trips on either real process RSS or the summed charged bytes, so it still works on a platform where RSS can’t be read (the charged-byte sum is the backstop). And when it trips, it runs one arbitration round (poll_arbitration) that actually does something about the pressure. Its sibling should_abort checks the hard limit and is the line that turns runaway memory into a fatal error. The soft and hard limits are the two limits the whole machine watches.

The policy is a trait object, and that’s correct here

Inside the arbitration round, which operator gets acted on is a strategy, and clinker makes it swappable through a trait:

clinker-exec ·memory.rs ·ArbitrationPolicy trait @19acdcb4

pub trait ArbitrationPolicy: Send + Sync {
    fn select_victim(&self, consumers: &[(ConsumerId, &dyn MemoryConsumer)],
                     pressure_bytes: u64) -> Option<ConsumerId>;
}
// concrete policies: LargestFirst, Priority, BackPressurePreferred, NoOpPolicy

The arbitrator holds its policy as Box<dyn ArbitrationPolicy> and the round asks it for a victim, then pauses or spills that operator:

let victim = self.policy.select_victim(&snapshot, pressure);
if let Some(id) = victim {
    if consumer.can_back_pressure() { consumer.pause(); }  // a source: park it
    else { consumer.try_spill(pressure); }                 // an operator: spill it
}

Stop and compare with Dispatching a node. There, operators were a closed enum with no dyn, because the set is engine-owned and fixed. Here the memory policy is a trait object, because it’s a strategy chosen at run time from config: Spill, Pause, or Both map to different policy implementations, and a deployment picks one. That’s exactly the open, runtime-chosen seam where dyn pays off (like the IO seam). Same engine, same author, opposite tools, chosen by whether the set is closed-and-owned or open-and-configured. Recognising which situation you’re in is the whole skill.

 memory pressure ──▶ should_spill() trips ──▶ poll_arbitration()
                                                   │
                                          policy.select_victim()
                                                   │
                                   ┌───────────────┴───────────────┐
                            can_back_pressure?                  otherwise
                                   │                                │
                              pause the source                  spill the operator

Reject the impossible budget up front

Here’s the failure the design most wants to prevent. Suppose a config sets a memory limit below the process’s baseline RSS, the memory the engine occupies before reading any data. Under a producer-pausing policy, the arbitrator would pause everything to get under budget and then never be able to resume: a deadlock. So clinker checks for it before the run starts and refuses:

clinker-exec ·memory.rs ·reject_unsatisfiable_budget fn @19acdcb4

pub fn reject_unsatisfiable_budget(limit: u64, knob: BackpressureKnob)
    -> Result<(), PipelineError> {
    if !knob.pauses_producers() { return Ok(()); }   // only pausing policies can deadlock
    let Some(baseline_rss) = rss_bytes() else { return Ok(()); };
    if limit < baseline_rss {
        return Err(PipelineError::UnsatisfiableMemoryBudget { limit, baseline_rss });
    }
    Ok(())
}

That PipelineError::UnsatisfiableMemoryBudget (error code E312) is one of the always-abort variants from Error handling: there’s no DLQ for “your budget is impossible,” it stops the run before any thread spawns. This is the bounded-memory contract’s front door: fail loudly at startup, never wedge mid-run. A guarantee is most valuable when it’s checked before you’ve spent an hour on the job.

Worked → completion → faded

You’ve read the three pieces: the gate, the policy seam, the startup check. Now build them, one new idea per rung, and observe the bounded-memory machine react under pressure.

Worked: the spill gate, fully annotated

Here is a self-contained arbitrator with the should_spill gate. It tracks summed charged bytes against a soft limit and counts how many arbitration rounds it runs. Run it and watch the gate trip only once charged bytes cross the limit.

rust // editable

reset ▶ run

// A miniature arbitrator: one soft limit, a running tally of charged bytes,
// and a should_spill() gate that trips on EITHER measure crossing the limit.
struct Arbitrator {
  soft_limit: u64,
  charged_bytes: u64,
  rss: u64,           // stand-in for "peak RSS"; in the engine this is observed live
  rounds_run: u32,    // how many arbitration rounds we triggered
}

impl Arbitrator {
  // Trips on RSS OR summed charged bytes — the OR is the backstop for platforms
  // where RSS can't be read. When it trips, run ONE arbitration round.
  fn should_spill(&mut self) -> bool {
      let tripped = self.rss > self.soft_limit || self.charged_bytes > self.soft_limit;
      if tripped {
          self.rounds_run += 1; // poll_arbitration() would act on the pressure here
      }
      tripped
  }
}

fn main() {
  let mut arb = Arbitrator { soft_limit: 10_000, charged_bytes: 0, rss: 4_000, rounds_run: 0 };

  for admit in [3_000u64, 4_000, 5_000] {
      arb.charged_bytes += admit;
      let spilling = arb.should_spill();
      println!("charged={:>6}  rss={:>5}  should_spill={spilling}", arb.charged_bytes, arb.rss);
  }
  println!("arbitration rounds run: {}", arb.rounds_run); // trips once charged > 10_000
}

// A miniature arbitrator: one soft limit, a running tally of charged bytes, // and a should_spill() gate that trips on EITHER measure crossing the limit. struct Arbitrator { soft_limit: u64, charged_bytes: u64, rss: u64, // stand-in for "peak RSS"; in the engine this is observed live rounds_run: u32, // how many arbitration rounds we triggered } impl Arbitrator { // Trips on RSS OR summed charged bytes — the OR is the backstop for platforms // where RSS can't be read. When it trips, run ONE arbitration round. fn should_spill(&mut self) -> bool { let tripped = self.rss > self.soft_limit || self.charged_bytes > self.soft_limit; if tripped { self.rounds_run += 1; // poll_arbitration() would act on the pressure here } tripped } } fn main() { let mut arb = Arbitrator { soft_limit: 10_000, charged_bytes: 0, rss: 4_000, rounds_run: 0 }; for admit in [3_000u64, 4_000, 5_000] { arb.charged_bytes += admit; let spilling = arb.should_spill(); println!("charged={:>6} rss={:>5} should_spill={spilling}", arb.charged_bytes, arb.rss); } println!("arbitration rounds run: {}", arb.rounds_run); // trips once charged > 10_000 }

The gate is the heartbeat of bounded memory: every bulk admission polls it, and only a trip pays for an arbitration round. Notice nothing here decides spill vs pause; that’s the policy’s job, which is the next rung.

Completion: select a victim under one policy

Below, the LargestFirst policy is written for you; the arbitrate function that drives it has one gap. An arbitration round asks the policy for a victim only when pressure is positive: when pressure_bytes == 0 there’s nothing to relieve, so it should return None without consulting the policy. Fill in that guard.

rust // editable

reset ▶ run

// (operator id, charged bytes) for each live consumer.
type Usage = (u32, u64);

trait Policy {
  fn select_victim(&self, usage: &[Usage]) -> Option<u32>;
}

struct LargestFirst;
impl Policy for LargestFirst {
  fn select_victim(&self, usage: &[Usage]) -> Option<u32> {
      usage.iter().max_by_key(|(_, bytes)| *bytes).map(|(id, _)| *id)
  }
}

// Run one arbitration round: no pressure means no victim.
fn arbitrate(policy: &dyn Policy, usage: &[Usage], pressure_bytes: u64) -> Option<u32> {
  // TODO: if there is no pressure, return None before consulting the policy.
  policy.select_victim(usage)
}

fn main() {
  let usage = [(0u32, 2_000), (1, 9_000), (2, 5_000)];
  println!("under pressure: {:?}", arbitrate(&LargestFirst, &usage, 6_000)); // Some(1)
  println!("no pressure:    {:?}", arbitrate(&LargestFirst, &usage, 0));     // expect None
}

// (operator id, charged bytes) for each live consumer. type Usage = (u32, u64); trait Policy { fn select_victim(&self, usage: &[Usage]) -> Option<u32>; } struct LargestFirst; impl Policy for LargestFirst { fn select_victim(&self, usage: &[Usage]) -> Option<u32> { usage.iter().max_by_key(|(_, bytes)| *bytes).map(|(id, _)| *id) } } // Run one arbitration round: no pressure means no victim. fn arbitrate(policy: &dyn Policy, usage: &[Usage], pressure_bytes: u64) -> Option<u32> { // TODO: if there is no pressure, return None before consulting the policy. policy.select_victim(usage) } fn main() { let usage = [(0u32, 2_000), (1, 9_000), (2, 5_000)]; println!("under pressure: {:?}", arbitrate(&LargestFirst, &usage, 6_000)); // Some(1) println!("no pressure: {:?}", arbitrate(&LargestFirst, &usage, 0)); // expect None }

💡 Hint 1

An arbitration round only runs when should_spill tripped, i.e. when there’s real pressure. With pressure_bytes == 0 there’s nothing to act on, so guard it with an early return None; before the policy call.

Show solution

fn arbitrate(policy: &dyn Policy, usage: &[Usage], pressure_bytes: u64) -> Option<u32> {
    if pressure_bytes == 0 {
        return None; // nothing to relieve — don't pick a victim
    }
    policy.select_victim(usage)
}

With the guard, the no-pressure call returns None without naming a victim, matching the engine, where a round only runs because should_spill already tripped. The policy is consulted only when there’s pressure to relieve.

Faded: a second policy, and observe the swap

Your turn with much less scaffolding. Add a LowestIdFirst policy that selects the operator with the smallest id (a crude stand-in for “the highest-priority operator gets protected last”). Then box each policy as Box<dyn Policy> and run the same pressure through both: the value of a runtime-chosen strategy is that the victim changes with no other code touched.

rust // editable

reset ▶ run

type Usage = (u32, u64);

trait Policy {
  fn select_victim(&self, usage: &[Usage]) -> Option<u32>;
}

struct LargestFirst;
impl Policy for LargestFirst {
  fn select_victim(&self, usage: &[Usage]) -> Option<u32> {
      usage.iter().max_by_key(|(_, bytes)| *bytes).map(|(id, _)| *id)
  }
}

// TODO 1: define LowestIdFirst and impl Policy so select_victim returns the
//         operator with the SMALLEST id.

fn arbitrate(policy: &dyn Policy, usage: &[Usage]) -> Option<u32> {
  policy.select_victim(usage)
}

fn main() {
  let usage = [(0u32, 2_000), (1, 9_000), (2, 5_000)]; // (op_id, charged bytes)

  // TODO 2: box LargestFirst, then LowestIdFirst, and arbitrate the same usage
  //         through each. Config picks the policy at run time — hence `dyn`.
  // Expected: largest-first victim is op 1 (9000); lowest-id victim is op 0.
}

type Usage = (u32, u64); trait Policy { fn select_victim(&self, usage: &[Usage]) -> Option<u32>; } struct LargestFirst; impl Policy for LargestFirst { fn select_victim(&self, usage: &[Usage]) -> Option<u32> { usage.iter().max_by_key(|(_, bytes)| *bytes).map(|(id, _)| *id) } } // TODO 1: define LowestIdFirst and impl Policy so select_victim returns the // operator with the SMALLEST id. fn arbitrate(policy: &dyn Policy, usage: &[Usage]) -> Option<u32> { policy.select_victim(usage) } fn main() { let usage = [(0u32, 2_000), (1, 9_000), (2, 5_000)]; // (op_id, charged bytes) // TODO 2: box LargestFirst, then LowestIdFirst, and arbitrate the same usage // through each. Config picks the policy at run time — hence `dyn`. // Expected: largest-first victim is op 1 (9000); lowest-id victim is op 0. }

Show solution

struct LowestIdFirst;
impl Policy for LowestIdFirst {
    fn select_victim(&self, usage: &[Usage]) -> Option<u32> {
        usage.iter().min_by_key(|(id, _)| *id).map(|(id, _)| *id)
    }
}

fn main() {
    let usage = [(0u32, 2_000), (1, 9_000), (2, 5_000)];

    let policy: Box<dyn Policy> = Box::new(LargestFirst);
    println!("largest-first victim: {:?}", arbitrate(&*policy, &usage)); // Some(1)

    let policy: Box<dyn Policy> = Box::new(LowestIdFirst);
    println!("lowest-id victim:     {:?}", arbitrate(&*policy, &usage)); // Some(0)
}

Swap the boxed policy and the victim changes with no other code touched; that’s the value of a runtime-chosen strategy behind a trait. The real ArbitrationPolicy is this exact shape, just with LargestFirst / Priority / BackPressurePreferred and real operators, and the round then pauses or spills the chosen victim depending on whether it can back-pressure.

Solid on the spill/pause/abort decision and the policy seam? Skip ahead →

Why-bridge: a guarantee is only as good as its front door

Why split the work into polled gates plus a swappable policy plus a startup check, instead of one tidy decide() function? Because each piece answers a different pressure that the others can’t:

• Gates (should_spill / should_abort) are polled independently by many operators across many threads; there’s no single chokepoint to serialize on, which matters when the whole point is to not stall the pipeline.
• The policy is the one part a deployment legitimately wants to vary, so it’s the one part behind dyn: config chooses LargestFirst vs Priority without the engine naming each inline.
• The startup check turns the worst failure mode (a silent mid-run deadlock under a pausing policy) into a loud, immediate UnsatisfiableMemoryBudget abort. The bounded-memory contract is worth most when it’s verified before you’ve spent an hour on the job.

Retrieval checkpoint

✓ Checkpoint — bounded memory

💡 Hint 1

Read should_spill: it trips on RSS OR summed charged bytes, then calls poll_arbitration. Then read reject_unsatisfiable_budget: under which policy does an impossible limit deadlock, and what error does it raise?

What the tour establishes

should_spill trips when peak RSS or summed charged bytes exceed the soft limit, then runs one arbitration round; poll_arbitration asks the Box<dyn ArbitrationPolicy> for a victim and pauses (a source) or spills (an operator) it. reject_unsatisfiable_budget refuses a limit below baseline RSS under a pausing policy, raising the always-abort PipelineError::UnsatisfiableMemoryBudget (E312) before any thread spawns. Closed enum for operators, trait object for the policy: the tool follows the shape of the choice.

// quick check

The arbitrator needs to express three responses: spill, pause, abort. How does it actually represent the decision?

• A single decide() function returns one enum { Spill, Pause, Abort } that callers act on
• Boolean gates (should_spill on the soft limit, should_abort on the hard limit) that operators poll independently, plus a swappable policy that chooses spill-vs-pause per victim inside an arbitration round
• Each operator stores its own copy of the budget and decides alone with no shared state

// quick check

Why is a memory limit below baseline RSS rejected at startup under a producer-pausing policy?

• Because small limits are always invalid
• Because the arbitrator would pause producers trying to get under budget but could never reach it (baseline RSS already exceeds the limit), wedging the run, so it fails fast with an always-abort error instead
• Because RSS can't be measured at startup

// quick check

An engineer claims: 'Operators are a closed enum, so for consistency ArbitrationPolicy should be one too; a Box<dyn> here is just inconsistency.' What is the misconception to correct?

• Correct: both are sets of variants, so both should be closed enums
• Closed-vs-open is a design decision, not a style nit: the operator set is engine-owned and fixed (a closed enum gives compile-time exhaustiveness), while the arbitration policy is selected at run time from config (open and configured), exactly the seam where a trait object is the right tool. They differ on purpose.
• Correct in reverse: operators should become a trait object to match the policy

// quick check

Spaced recall from Node buffers & spill: when the arbitration policy selects an operator and the round calls try_spill on it, what does spilling do to that operator's NodeBuffer, and what state can result?

• It deletes the buffered records to free memory immediately
• It pushes the buffer's records out of RAM to disk so they stop counting against the budget; a buffer that later receives new records becomes Mixed (an in-memory tail on top of on-disk chunks), and the records drain back in order
• It converts the records into an error and routes them to the DLQ

You should be able to:

• You can explain the spill/pause/abort decision as gates plus a policy, not one verdict enum
• You can explain what should_spill checks (RSS OR charged bytes) and that it runs an arbitration round when tripped
• You wrote two ArbitrationPolicy implementations and saw a boxed swap change the victim with no other code touched
• You can explain why ArbitrationPolicy is a trait object while operators are a closed enum
• You can explain the unsatisfiable-budget rejection and why it must happen before the run

Verify in the checkout:

grep -n 'fn should_spill' crates/clinker-exec/src/pipeline/memory.rs
grep -n 'trait ArbitrationPolicy' crates/clinker-exec/src/pipeline/memory.rs
grep -n 'fn reject_unsatisfiable_budget' crates/clinker-exec/src/pipeline/memory.rs
grep -n 'UnsatisfiableMemoryBudget' crates/clinker-plan/src/error.rs

Connections

You’ve now seen the whole bounded-memory machine: buffers that spill, an arbitrator that decides through polled gates, a policy that picks the victim, and a startup check that refuses an impossible budget. The next lesson, unsafe & FieldStr, drops to the lowest level the data layer reaches: the unsafe code behind the string type every record field uses, and the invariants that keep it sound.

Prerequisites

• The memory arbitrator

→

Leads to

• FieldStr

Go deeper on the Rust (optional, one-directional, the policy seam from first principles):

• Want trait objects and dyn dispatch drilled from first principles, no engine framing? Read the docs →

Glossary terms used: bounded memory, spill, soft / hard limit, streaming.

Mark this lesson complete