From 525e6dfc94f83ad60b953814adf2fab1e41a5d64 Mon Sep 17 00:00:00 2001
From: Jon-Carlos Rivera <jon.carlos.rivera@gmail.com>
Date: Mon, 19 Oct 2026 08:32:00 -0700
Subject: [PATCH 1/7] lanes: an input clock is cut no further than the cap, and
 a lane that runs a tick at a time takes its ticks in turns

An audio clock ticks every `window` samples, so foundmeat's listen_audio,
with a window of one sample, was cut 1024 ticks by every packet of 1024
samples and queued them all at once: the 1023 tasks against a cap of 10
in their trace. Every tick cost a turn of the scheduler and a wake of
every waiting thread, about 48 us of CPU, so the lane needed 2.3 cores to
keep up with 48000 ticks a second and fell behind when the box was busy.
Its reader carries the programme's own sound, so the programme slowed with
it.

The cut now stops at the cap and holds the next settled tick back, the
lane counts as full to its readers while it does, and the rest is cut as
the queue drains. A lane that runs one tick at a time takes the ticks
queued behind the one it starts, and those its clock settled past the cap,
up to 1024, in one turn. The progress marks of a turn leave once, and a
lane with a tick held back is picked first. listen_audio alone now runs
at 4.0 times real time with 0.32 cores of CPU, where it ran at 0.84
times with 3.7 cores.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
---
 sidecar/NODE-CLI.md              |  21 ++
 sidecar/ffrwd-wasm/src/lanes.rs  | 624 +++++++++++++++++++++++++++----
 sidecar/ffrwd-wasm/tests/live.rs |  44 +++
 3 files changed, 614 insertions(+), 75 deletions(-)

diff --git a/sidecar/NODE-CLI.md b/sidecar/NODE-CLI.md
index 342aaa0..98b1055 100644
--- a/sidecar/NODE-CLI.md
+++ b/sidecar/NODE-CLI.md
@@ -369,6 +369,27 @@ tick that held them.
   the clock reads: an `arrival` input's pts may sit on any origin (a feed
   at the epoch while the clock counts from 0), so they never hold the
   clock.
+- **Queues.** A node's ticks wait in a queue of at most `2 * jobs + 2`
+  until a worker runs them, and a reader waits while the queue of
+  anything reading its stream is full. Every clock is cut no further than
+  that. One arrival can settle many ticks: an audio clock ticks every
+  `window` samples, so a node with a window of one sample on a clock
+  bound to sound is cut 1024 ticks by a packet of 1024 samples, 48000 a
+  second. Those past the bound stay with the clock and are cut as the
+  queue drains, and while any are left the node counts as full to its
+  readers. A self-clocked node is handed one arrival an input a call, and
+  the arrivals waiting on its input count against the same bound, so a
+  decoder ahead of a slower encoder holds its reader back rather than
+  queue every picture it made. A node that runs one tick at a time takes
+  the ticks queued behind the one it starts, and those its clock has
+  settled past the bound, up to 1024, and runs them in order in one turn,
+  and the progress they send on each port leaves once, as the last of
+  them: a clock of small ticks costs a node call a tick, not a turn of
+  the scheduler a tick. A node with ticks held back goes first. A node
+  slower than its clock holds back the reader of its clock's input, and
+  with it every stream that input carries (two `asplit` copies of one
+  sound, one for a listener and one for the programme, ride one input),
+  so it can slow the whole programme.
 - **Hold inputs.** A hold input bound to a stream holds the tick until its
   source has arrived past it (its picture, or with none its sound, past
   `lead` beyond the tick), or ended, for no longer than its
diff --git a/sidecar/ffrwd-wasm/src/lanes.rs b/sidecar/ffrwd-wasm/src/lanes.rs
index ce5d851..edbbdfb 100644
--- a/sidecar/ffrwd-wasm/src/lanes.rs
+++ b/sidecar/ffrwd-wasm/src/lanes.rs
@@ -167,6 +167,16 @@ enum LaneIntake {
     },
 }
 
+/// What a worker takes from a lane in one turn: the tasks it runs in
+/// order, the instance that runs them, and that instance or the means to
+/// open it.
+struct Turn {
+    tasks: Vec<Task>,
+    instance: usize,
+    runner: Option<Box<dyn Node>>,
+    opener: Option<Opener>,
+}
+
 struct Task {
     ordinal: u64,
     tick: Tick,
@@ -229,6 +239,10 @@ struct Lane {
     started: Option<Instant>,
     /// When a rate clock whose inputs all arrive unpaired last cut a tick.
     turned: Option<Instant>,
+    /// The next tick of an input clock whose cut stopped at the cap, and
+    /// where its interval ends: while one is held back the lane counts as
+    /// full to everything it reads.
+    withheld: Option<(Tick, Option<i64>)>,
     unbuffered: bool,
 }
 
@@ -390,7 +404,8 @@ impl State {
     /// taking one arrival an input, so what waits for it is the arrivals
     /// queued on its input, which count against the same bound as a task
     /// queue: without that, a decoder ahead of a slower encoder would queue
-    /// every picture it made.
+    /// every picture it made. An input clock is cut no further than the
+    /// cap, and while its cut stopped there the lane counts as full.
     fn room(&self, stream: u32, consumers: &[Consumer]) -> bool {
         consumers.iter().all(|c| match c {
             Consumer::Lane(j) => {
@@ -401,7 +416,8 @@ impl State {
                     }
                     _ => 0,
                 };
-                lane.stopped || (lane.queue.len() < self.cap && queued < self.cap)
+                lane.stopped
+                    || (lane.queue.len() < self.cap && lane.withheld.is_none() && queued < self.cap)
             }
             Consumer::Writer(w, _) => self.writers[*w].len() < self.cap,
         })
@@ -648,14 +664,36 @@ impl State {
                 let LaneIntake::Assembled(a) = &mut lane.intake else {
                     unreachable!("an input clock is assembled")
                 };
-                let mut cut = Vec::new();
-                while let Some(tick) = a.next_clocked()? {
-                    let last = tick.last;
-                    cut.push((tick, a.tick_end()));
+                // One arrival can settle many ticks (an audio clock cuts
+                // `window` samples a tick, so a packet of 1024 makes 1024
+                // ticks of one), so the cut stops at the cap as every other
+                // clock's does, holding the next tick back if one is
+                // settled, and the rest is cut as the queue drains.
+                let mut cut: Vec<(Tick, Option<i64>)> = Vec::new();
+                let mut last = false;
+                while lane.queue.len() + cut.len() < cap {
+                    let next = match lane.withheld.take() {
+                        Some(held) => held,
+                        None => match a.next_clocked()? {
+                            Some(tick) => {
+                                let end = a.tick_end();
+                                (tick, end)
+                            }
+                            None => break,
+                        },
+                    };
+                    last = next.0.last;
+                    cut.push(next);
                     if last {
                         break;
                     }
                 }
+                if !last && lane.withheld.is_none() && lane.queue.len() + cut.len() >= cap {
+                    if let Some(tick) = a.next_clocked()? {
+                        let end = a.tick_end();
+                        lane.withheld = Some((tick, end));
+                    }
+                }
                 for (tick, end) in cut {
                     Self::push_task(lane, tick, end);
                 }
@@ -844,31 +882,44 @@ impl State {
         }
     }
 
-    /// A rate clock's ticks are cut as its queue has room, and a
-    /// self-clocked node's as it returns, so both are cut here, where a
-    /// worker looks for work, as well as on arrivals.
+    /// A rate clock's ticks are cut as its queue has room, a self-clocked
+    /// node's as it returns, and an input clock's that the cap withheld as
+    /// its queue drains, so all are cut here, where a worker looks for
+    /// work, as well as on arrivals. A lane holding a tick back holds back
+    /// the readers of what it reads, so it goes first; then the lane with
+    /// the longest queue.
     fn pick(&mut self) -> Result<Option<usize>> {
         for j in 0..self.lanes.len() {
-            if matches!(
-                self.lanes[j].clock,
-                ClockKind::Rate { .. } | ClockKind::SelfClocked
-            ) {
+            let lane = &self.lanes[j];
+            if matches!(lane.clock, ClockKind::Rate { .. } | ClockKind::SelfClocked)
+                || (lane.withheld.is_some() && lane.queue.len() < self.cap)
+            {
                 self.cut(j)?;
             }
         }
         Ok((0..self.lanes.len())
             .filter(|i| self.dispatchable(*i))
-            .max_by_key(|i| self.lanes[*i].queue.len()))
+            .max_by_key(|i| {
+                (
+                    self.lanes[*i].withheld.is_some(),
+                    self.lanes[*i].queue.len(),
+                )
+            }))
     }
 
     /// Takes lane `i`'s front task with an instance, or the means to open
-    /// one, and hands a state input its earlier rows.
-    fn dispatch(&mut self, i: usize) -> (Task, usize, Option<Box<dyn Node>>, Option<Opener>) {
+    /// one, and hands a state input its earlier rows. A lane that runs one
+    /// tick at a time takes the tasks queued behind the front one too, and
+    /// the ticks its clock has settled past the cap, up to [`TURN`] and
+    /// short of its last, and runs them in order in one turn: a clock of
+    /// small ticks (an audio clock cut a sample a tick, 48000 a second)
+    /// would otherwise wait for a worker, and pay a turn of the scheduler
+    /// and a wake of every thread waiting on it, for each.
+    fn dispatch(&mut self, i: usize) -> Result<Turn> {
         let lane = &mut self.lanes[i];
-        let mut task = lane.queue.pop_front().expect("picked lanes have a task");
-        task.tick.ordinal = task.ordinal;
+        let front = lane.queue.pop_front().expect("picked lanes have a task");
         lane.in_flight += 1;
-        let wanted = match (task.tick.last, lane.finisher) {
+        let wanted = match (front.tick.last, lane.finisher) {
             (true, Some(finisher)) => lane.idle.iter().position(|(i, _)| *i == finisher),
             _ => lane.idle.len().checked_sub(1),
         };
@@ -880,55 +931,89 @@ impl State {
                 (instance, None, lane.opener.clone())
             }
         };
-        if !lane.state.is_empty() {
-            if let LaneIntake::Assembled(a) = &mut lane.intake {
-                for stream in &mut task.tick.streams {
-                    if lane.state.contains(&stream.id) {
-                        stream.earlier_rows = a.earlier.owed(instance, stream.id, task.ordinal);
-                    }
+        let mut tasks = vec![front];
+        if lane.width == 1 && !lane.unbuffered && !tasks[0].tick.last {
+            loop {
+                let lane = &mut self.lanes[i];
+                while tasks.len() < TURN && lane.queue.front().is_some_and(|t| !t.tick.last) {
+                    tasks.extend(lane.queue.pop_front());
                 }
-                a.earlier.processed(instance, task.ordinal);
-                if lane.created >= lane.width {
-                    a.earlier.forget_told(lane.created);
+                if tasks.len() >= TURN || !lane.queue.is_empty() || lane.withheld.is_none() {
+                    break;
                 }
+                self.cut(i)?;
             }
         }
-        if let LaneIntake::Assembled(a) = &mut lane.intake {
-            for stream in &mut task.tick.streams {
-                stream.ended_feeds = a.ended.owed(instance, stream.id, task.ordinal);
+        let lane = &mut self.lanes[i];
+        for task in &mut tasks {
+            task.tick.ordinal = task.ordinal;
+            if !lane.state.is_empty() {
+                if let LaneIntake::Assembled(a) = &mut lane.intake {
+                    for stream in &mut task.tick.streams {
+                        if lane.state.contains(&stream.id) {
+                            stream.earlier_rows = a.earlier.owed(instance, stream.id, task.ordinal);
+                        }
+                    }
+                    a.earlier.processed(instance, task.ordinal);
+                    if lane.created >= lane.width {
+                        a.earlier.forget_told(lane.created);
+                    }
+                }
             }
-            a.ended.processed(instance, task.ordinal);
-            if lane.created >= lane.width {
-                a.ended.forget_told(lane.created);
+            if let LaneIntake::Assembled(a) = &mut lane.intake {
+                for stream in &mut task.tick.streams {
+                    stream.ended_feeds = a.ended.owed(instance, stream.id, task.ordinal);
+                }
+                a.ended.processed(instance, task.ordinal);
+                if lane.created >= lane.width {
+                    a.ended.forget_told(lane.created);
+                }
             }
         }
-        (task, instance, runner, opener)
+        Ok(Turn {
+            tasks,
+            instance,
+            runner,
+            opener,
+        })
     }
 
+    /// The results of one turn of lane `i`, by ordinal, in the order they
+    /// ran.
     fn complete(
         &mut self,
         i: usize,
-        ordinal: u64,
+        results: Vec<(u64, Result<Done>)>,
         instance: usize,
         runner: Option<Box<dyn Node>>,
-        done: Result<Done>,
     ) {
         let lane = &mut self.lanes[i];
         lane.in_flight -= 1;
-        let last = done.as_ref().is_ok_and(|d| d.last);
+        let last = results
+            .iter()
+            .any(|(_, done)| done.as_ref().is_ok_and(|d| d.last));
         if let Some(runner) = runner {
             if !last {
                 lane.idle.push((instance, runner));
             }
         }
-        let flushed = match done {
-            Ok(done) => {
-                if ordinal >= lane.next_flush {
-                    lane.done.insert(ordinal, done);
+        let mut failed = None;
+        for (ordinal, done) in results {
+            match done {
+                Ok(done) => {
+                    if ordinal >= lane.next_flush {
+                        lane.done.insert(ordinal, done);
+                    }
+                }
+                Err(e) => {
+                    failed = Some(e.context(format!("in {}", lane.name)));
+                    break;
                 }
-                self.flush(i)
             }
-            Err(e) => Err(e.context(format!("in {}", self.lanes[i].name))),
+        }
+        let flushed = match failed {
+            Some(e) => Err(e),
+            None => self.flush(i),
         };
         if let Err(e) = flushed {
             if self.error.is_none() {
@@ -941,6 +1026,7 @@ impl State {
     /// port's readers, the node's progress down every port, and on the last
     /// tick the end of every port.
     fn flush(&mut self, i: usize) -> Result<()> {
+        let mut leaving: Vec<(u32, Delivery)> = Vec::new();
         loop {
             let lane = &mut self.lanes[i];
             let Some(done) = lane.done.remove(&lane.next_flush) else {
@@ -1098,6 +1184,7 @@ impl State {
                 lane.stopped = true;
                 lane.finisher = Some(instance);
                 lane.queue.clear();
+                lane.withheld = None;
                 lane.done.clear();
                 lane.next_flush = lane.next_ordinal;
                 let streams = lane
@@ -1123,17 +1210,18 @@ impl State {
                 });
                 lane.next_ordinal += 1;
             }
-            for (stream, delivery) in deliveries {
-                match delivery {
-                    Delivery::Item(item) => self.arrive(stream, item)?,
-                    Delivery::Progress(at) => self.progress(stream, at)?,
-                    Delivery::End => self.end(stream)?,
-                    Delivery::Trailing(rows) => self.trailing(stream, &rows),
-                    Delivery::Batch => {
-                        for consumer in self.consumers(stream) {
-                            if let Consumer::Writer(w, s) = consumer {
-                                self.writers[w].push(s, Out::Batch);
-                            }
+            leaving.extend(deliveries);
+        }
+        for (stream, delivery) in superseded_progress_dropped(leaving) {
+            match delivery {
+                Delivery::Item(item) => self.arrive(stream, item)?,
+                Delivery::Progress(at) => self.progress(stream, at)?,
+                Delivery::End => self.end(stream)?,
+                Delivery::Trailing(rows) => self.trailing(stream, &rows),
+                Delivery::Batch => {
+                    for consumer in self.consumers(stream) {
+                        if let Consumer::Writer(w, s) = consumer {
+                            self.writers[w].push(s, Out::Batch);
                         }
                     }
                 }
@@ -1162,6 +1250,26 @@ enum Delivery {
     Batch,
 }
 
+/// What several ticks of one lane hand on, less each progress mark that a
+/// later one on the same stream says again with nothing between them on
+/// that stream: a turn of many small ticks moves its readers' progress once.
+fn superseded_progress_dropped(deliveries: Vec<(u32, Delivery)>) -> Vec<(u32, Delivery)> {
+    let mut marked: HashSet<u32> = HashSet::new();
+    let mut kept: Vec<(u32, Delivery)> = deliveries
+        .into_iter()
+        .rev()
+        .filter(|(stream, delivery)| match delivery {
+            Delivery::Progress(_) => marked.insert(*stream),
+            _ => {
+                marked.remove(stream);
+                true
+            }
+        })
+        .collect();
+    kept.reverse();
+    kept
+}
+
 fn out_of(item: &Item) -> Out {
     match item {
         Item::Frame(f) => Out::Frame {
@@ -1180,6 +1288,10 @@ fn out_of(item: &Item) -> Out {
 /// core from the ffmpeg processes and encoders beside it.
 pub const DEFAULT_WORKERS: usize = 4;
 
+/// The most ticks one turn of a lane that runs a tick at a time takes: a
+/// packet of sound's worth on a clock cut a sample a tick.
+const TURN: usize = 1024;
+
 /// How many worker threads a run gets: `-jobs` when it was given, else
 /// `DEFAULT_WORKERS`, never more than the machine's effective core count.
 /// `-jobs 1` is the serial escape hatch.
@@ -1294,6 +1406,7 @@ impl Scheduler {
                 finisher: None,
                 started: None,
                 turned: None,
+                withheld: None,
                 unbuffered: spec.unbuffered,
             });
         }
@@ -1360,7 +1473,12 @@ impl Scheduler {
                     .map(|c| match c {
                         Consumer::Lane(j) => {
                             let lane = &state.lanes[*j];
-                            format!("{} queue {}", lane.name, lane.queue.len())
+                            let withheld = if lane.withheld.is_some() {
+                                " and a tick withheld"
+                            } else {
+                                ""
+                            };
+                            format!("{} queue {}{withheld}", lane.name, lane.queue.len())
                         }
                         Consumer::Writer(w, _) => format!("writer {w} {}", state.writers[*w].len()),
                     })
@@ -1542,7 +1660,19 @@ fn worker(shared: &Shared) {
             };
             continue;
         };
-        let (task, instance, runner, opener) = state.dispatch(i);
+        let Turn {
+            tasks,
+            instance,
+            runner,
+            opener,
+        } = match state.dispatch(i) {
+            Ok(taken) => taken,
+            Err(e) => {
+                state.error = Some(e);
+                shared.notify();
+                return;
+            }
+        };
         drop(state);
         shared.notify();
 
@@ -1553,27 +1683,40 @@ fn worker(shared: &Shared) {
                 None => Err(anyhow!("a lane ran out of instances")),
             },
         };
-        let Task { ordinal, tick, end } = task;
-        let (pts, last) = (tick.pts, tick.last);
-        let (runner, done) = match opened {
+        let mut results = Vec::with_capacity(tasks.len());
+        let runner = match opened {
             Ok(mut runner) => {
-                let started = Instant::now();
-                let result = runner.process(tick);
-                PROFILE.process.add(0, started);
-                let result = result.map(|emitted| Done {
-                    emitted,
-                    pts,
-                    end,
-                    last,
-                    instance,
-                });
-                (Some(runner), result)
+                for Task { ordinal, tick, end } in tasks {
+                    let (pts, last) = (tick.pts, tick.last);
+                    let started = Instant::now();
+                    let result = runner.process(tick);
+                    PROFILE.process.add(0, started);
+                    // A tick the node says it finished on, or failed on, is
+                    // the last of the turn: what was taken after it is not
+                    // run, as a lane that stops drops its queue.
+                    let stop = result.as_ref().map_or(true, |emitted| emitted.finished);
+                    let result = result.map(|emitted| Done {
+                        emitted,
+                        pts,
+                        end,
+                        last,
+                        instance,
+                    });
+                    results.push((ordinal, result));
+                    if stop {
+                        break;
+                    }
+                }
+                Some(runner)
+            }
+            Err(e) => {
+                results.push((tasks[0].ordinal, Err(e)));
+                None
             }
-            Err(e) => (None, Err(e)),
         };
 
         state = shared.lock();
-        state.complete(i, ordinal, instance, runner, done);
+        state.complete(i, results, instance, runner);
         shared.notify();
     }
 }
@@ -1592,8 +1735,8 @@ mod tests {
         assert_eq!(worker_count(Some(1)), 1);
         assert_eq!(worker_count(Some(usize::MAX)), cores);
     }
-    use ffrwd_wasm_runtime::runtime::{StreamInfo, VideoFormat};
-    use std::sync::mpsc;
+    use ffrwd_wasm_runtime::runtime::{AudioFormat, StreamInfo, VideoFormat};
+    use std::sync::{mpsc, OnceLock};
 
     /// What a publisher's calls brought: frames per call, and whether each
     /// was the last.
@@ -1847,4 +1990,335 @@ mod tests {
         writer.join();
         let _ = std::fs::remove_dir_all(&dir);
     }
+
+    /// Opens when told to, for a node to wait on in its first call.
+    #[derive(Default)]
+    struct Gate {
+        open: Mutex<bool>,
+        opened: Condvar,
+    }
+
+    impl Gate {
+        fn wait(&self) {
+            let mut open = self.open.lock().unwrap();
+            while !*open {
+                open = self.opened.wait(open).unwrap();
+            }
+        }
+
+        fn open(&self) {
+            *self.open.lock().unwrap() = true;
+            self.opened.notify_all();
+        }
+    }
+
+    /// A node that notes the pts of each tick and, at each call, how long
+    /// its lane's queue is; it waits on `gate` in its first call.
+    struct Noting {
+        shape: NodeShape,
+        seen: Arc<Mutex<Vec<i64>>>,
+        queues: Arc<Mutex<Vec<usize>>>,
+        gate: Arc<Gate>,
+        shared: Arc<OnceLock<Arc<Shared>>>,
+    }
+
+    impl Node for Noting {
+        fn name(&self) -> &str {
+            "noting"
+        }
+
+        fn shape(&self) -> &NodeShape {
+            &self.shape
+        }
+
+        fn set_params(&mut self, _params: &str) -> Result<()> {
+            Ok(())
+        }
+
+        fn process(&mut self, tick: Tick) -> Result<Emitted> {
+            if let Some(shared) = self.shared.get() {
+                let queue = shared.lock().lanes[0].queue.len();
+                self.queues.lock().unwrap().push(queue);
+            }
+            let first = {
+                let mut seen = self.seen.lock().unwrap();
+                if !tick.last {
+                    seen.push(tick.pts);
+                }
+                seen.len() == 1
+            };
+            if first {
+                self.gate.wait();
+            }
+            Ok(Emitted::default())
+        }
+    }
+
+    /// One lane running `node`, clocked by its one input `port` on stream
+    /// 0, and the scheduler running it on one worker (a cap of 4).
+    fn one_lane(shape: NodeShape, bound: BoundStream, node: Noting) -> Scheduler {
+        let assembler = Assembler::new(&shape, &[bound], "noting", &[]).expect("an assembler");
+        let plan = Plan {
+            lanes: vec![LaneSpec {
+                gpu: HashMap::new(),
+                name: "noting".to_string(),
+                shape,
+                intake: Intake::Assembled(Box::new(assembler)),
+                bound: vec![0],
+                state: Vec::new(),
+                tick_base: TimeBase {
+                    num: 1,
+                    den: 48_000,
+                },
+                runners: vec![Box::new(node)],
+                opener: None,
+                ports: Vec::new(),
+                rows: None,
+                unbuffered: false,
+            }],
+            consumers: HashMap::from([(0, vec![Consumer::Lane(0)])]),
+            writers: Vec::new(),
+            mirrors: HashMap::new(),
+        };
+        Scheduler::start(plan, 1, &WakeSlot::new()).expect("the lanes")
+    }
+
+    fn wait_until(what: &str, done: impl Fn() -> bool) {
+        let until = Instant::now() + Duration::from_secs(5);
+        while !done() {
+            assert!(Instant::now() < until, "{what}");
+            thread::sleep(Duration::from_millis(5));
+        }
+    }
+
+    fn input_port(name: &str, kind: PortKind, pairing: Pairing) -> InputPort {
+        InputPort {
+            name: name.to_string(),
+            kind,
+            required: true,
+            many: false,
+            pairing,
+            rows: RowsUse::Ignore,
+            window: 1,
+            stride: 1,
+            accepts: Accepts::default(),
+            schema: None,
+        }
+    }
+
+    /// A node clocked by its sound a sample a tick (a listener written for
+    /// pictures, bound to sound) is cut 1024 ticks by one packet of 1024
+    /// samples. The cut stops at the cap, the reader waits while ticks are
+    /// withheld, every tick still runs once and in order, and the ticks
+    /// queued behind the first are taken in one turn.
+    #[test]
+    fn an_audio_clock_a_sample_a_tick_queues_no_more_than_the_cap_and_runs_every_tick() {
+        let shape = NodeShape {
+            inputs: vec![input_port("a", PortKind::Audio, Pairing::Lockstep)],
+            outputs: Vec::new(),
+            clock: Clock::Input("a".to_string()),
+            pure: false,
+            one_to_one: false,
+            bounded: true,
+            relation: Vec::new(),
+        };
+        let samples = TimeBase {
+            num: 1,
+            den: 48_000,
+        };
+        let bound = BoundStream {
+            port: "a".to_string(),
+            id: 0,
+            info: StreamInfo::default(),
+            time_base: samples,
+            format: StreamFormat::Audio(AudioFormat {
+                sample_rate: 48_000,
+                channels: 1,
+                sample_fmt: "f32",
+                channel_layout: None,
+            }),
+            rendition: Default::default(),
+            row: None,
+            decode_delay: 0,
+            latency: None,
+            hint: Default::default(),
+        };
+        let seen = Arc::new(Mutex::new(Vec::new()));
+        let queues = Arc::new(Mutex::new(Vec::new()));
+        let gate = Arc::new(Gate::default());
+        let shared = Arc::new(OnceLock::new());
+        let node = Noting {
+            shape: shape.clone(),
+            seen: Arc::clone(&seen),
+            queues: Arc::clone(&queues),
+            gate: Arc::clone(&gate),
+            shared: Arc::clone(&shared),
+        };
+        let scheduler = one_lane(shape, bound, node);
+        let _ = shared.set(Arc::clone(&scheduler.shared));
+        let packet = |pts: i64| {
+            Item::Frame(TickFrame {
+                pts,
+                duration: Some(1024),
+                data: Arc::new(vec![0u8; 1024 * 4]).into(),
+                rows: Vec::new(),
+            })
+        };
+        assert!(scheduler.arrive(0, packet(0)));
+        wait_until("the first tick ran", || !seen.lock().unwrap().is_empty());
+        // The first packet's ticks went in one turn; the second's wait.
+        assert!(scheduler.arrive(0, packet(1024)));
+        {
+            let state = scheduler.shared.lock();
+            let lane = &state.lanes[0];
+            assert_eq!(
+                lane.queue.len(),
+                state.cap,
+                "the queue holds the cap's worth of the second packet's ticks"
+            );
+            assert!(
+                lane.withheld.is_some(),
+                "the rest of the packet's ticks wait"
+            );
+            assert!(
+                !state.room(0, &state.consumers(0)),
+                "the reader waits while ticks are withheld"
+            );
+        }
+        gate.open();
+        wait_until("every sample's tick ran", || {
+            seen.lock().unwrap().len() == 2048
+        });
+        assert!(scheduler.end(0));
+        let (done, finished) = mpsc::channel();
+        let shared_queues = Arc::clone(&queues);
+        thread::spawn(move || done.send(scheduler.finish()));
+        finished
+            .recv_timeout(Duration::from_secs(5))
+            .expect("the lane ended with its input")
+            .expect("the run");
+        assert_eq!(*seen.lock().unwrap(), (0..2048).collect::<Vec<i64>>());
+        let queues = shared_queues.lock().unwrap();
+        assert!(
+            queues[1..1024].iter().all(|q| *q == 4),
+            "the first packet's ticks ran in one turn, the second's queued behind it: {:?}",
+            &queues[..12]
+        );
+        assert!(
+            queues[1024..2048].iter().all(|q| *q == 0),
+            "the second packet's ticks ran in one turn: {:?}",
+            &queues[1024..1036]
+        );
+    }
+
+    /// A self-clocked reader (a decoder) slower than what reaches it holds
+    /// its writer back once the cap's worth of arrivals waits on its input,
+    /// whatever else the scheduler bounds.
+    #[test]
+    fn a_self_clocked_reader_holds_its_writer_at_the_cap() {
+        let shape = NodeShape {
+            inputs: vec![input_port("v", PortKind::Video, Pairing::Arrival)],
+            outputs: Vec::new(),
+            clock: Clock::SelfClocked,
+            pure: false,
+            one_to_one: false,
+            bounded: true,
+            relation: Vec::new(),
+        };
+        let bound = BoundStream {
+            port: "v".to_string(),
+            id: 0,
+            info: StreamInfo::default(),
+            time_base: TimeBase { num: 1, den: 30 },
+            format: StreamFormat::Video(VideoFormat {
+                width: 2,
+                height: 2,
+                pix_fmt: "rgba",
+                frame_len: 16,
+                color: None,
+            }),
+            rendition: Default::default(),
+            row: None,
+            decode_delay: 0,
+            latency: None,
+            hint: Default::default(),
+        };
+        let seen = Arc::new(Mutex::new(Vec::new()));
+        let gate = Arc::new(Gate::default());
+        let node = Noting {
+            shape: shape.clone(),
+            seen: Arc::clone(&seen),
+            queues: Arc::default(),
+            gate: Arc::clone(&gate),
+            shared: Arc::default(),
+        };
+        let scheduler = Arc::new(one_lane(shape, bound, node));
+        let delivered = Arc::new(Mutex::new(0usize));
+        let writer = {
+            let scheduler = Arc::clone(&scheduler);
+            let delivered = Arc::clone(&delivered);
+            thread::spawn(move || {
+                for k in 0..40 {
+                    let frame = TickFrame {
+                        pts: k,
+                        duration: None,
+                        data: Arc::new(vec![0; 16]).into(),
+                        rows: Vec::new(),
+                    };
+                    if !scheduler.arrive(0, Item::Frame(frame)) {
+                        return;
+                    }
+                    *delivered.lock().unwrap() += 1;
+                }
+                scheduler.end(0);
+            })
+        };
+        wait_until("the first call started", || {
+            !seen.lock().unwrap().is_empty()
+        });
+        thread::sleep(Duration::from_millis(200));
+        let held = *delivered.lock().unwrap();
+        assert!(
+            held <= 4 + 1,
+            "the writer waits once the cap's worth is queued ({held} delivered)"
+        );
+        gate.open();
+        writer.join().expect("the writer");
+        scheduler.finish().expect("the run");
+        assert_eq!(*delivered.lock().unwrap(), 40);
+    }
+
+    #[test]
+    fn a_progress_mark_a_later_one_says_again_is_dropped() {
+        let marks = |d: &[(u32, Delivery)]| -> Vec<String> {
+            d.iter()
+                .map(|(s, d)| match d {
+                    Delivery::Progress(at) => format!("{s}:p{at}"),
+                    Delivery::End => format!("{s}:end"),
+                    _ => format!("{s}:other"),
+                })
+                .collect()
+        };
+        let message = || {
+            Delivery::Item(Item::Message(Message {
+                pts: 0,
+                data: Vec::new(),
+            }))
+        };
+        let kept = superseded_progress_dropped(vec![
+            (1, Delivery::Progress(1)),
+            (1, Delivery::Progress(2)),
+            (2, Delivery::Progress(2)),
+            (1, message()),
+            (1, Delivery::Progress(3)),
+            (2, Delivery::Progress(3)),
+            (1, Delivery::Progress(4)),
+            (1, Delivery::End),
+        ]);
+        assert_eq!(
+            marks(&kept),
+            ["1:p2", "1:other", "2:p3", "1:p4", "1:end"],
+            "each stream's marks up to the next thing on it, or the end, are the last of them"
+        );
+    }
 }
diff --git a/sidecar/ffrwd-wasm/tests/live.rs b/sidecar/ffrwd-wasm/tests/live.rs
index 3577330..f557798 100644
--- a/sidecar/ffrwd-wasm/tests/live.rs
+++ b/sidecar/ffrwd-wasm/tests/live.rs
@@ -1588,6 +1588,50 @@ fn a_programme_without_a_feeder_passes_through_untouched() {
     assert_eq!(out.clock.len(), 3);
 }
 
+/// A node clocked by the programme's sound a sample a tick (48000 ticks a
+/// second, 1024 of them settled by each packet), on the same input as the
+/// programme's picture, as foundmeat's listener on the programme's sound
+/// is: every one of its ticks runs, and the programme leaves in its own
+/// time, the reader of the input not held back by the node's ticks.
+#[test]
+fn a_node_clocked_a_sample_a_tick_on_the_programmes_sound_keeps_the_programme_in_real_time() {
+    let _serial = serial();
+    let program = Program::new(60, 3.0);
+    let chain = "[v=0:v]inset=port={port}:lead=0.5[v=o];[a=0:a]shape_window=window=1[cues=c]";
+    let (mut run, port) = start(
+        "sample-clock",
+        &program,
+        "4",
+        &["inset", "shape_window"],
+        chain,
+    );
+    wait_for_port(&mut run, port);
+    let started = Instant::now();
+    let out = finish(run);
+    let took = started.elapsed().as_secs_f64();
+    assert_eq!(out.frames.len() as i64, program.frames(), "{}", out.stderr);
+    // The programme's sound is written in whole packets of 1024.
+    let samples = (program.seconds * RATE as f64) as usize / 1024 * 1024;
+    assert_eq!(
+        out.clock.len(),
+        samples,
+        "a tick for every sample of the programme's sound"
+    );
+    assert!(
+        out.clock
+            .iter()
+            .enumerate()
+            .all(|(k, cue)| cue["text"] == format!("window {k}")),
+        "the ticks ran in order"
+    );
+    assert!(
+        took < program.seconds + 1.5,
+        "the programme's {} s left in {took:.2} s\n{}",
+        program.seconds,
+        out.stderr
+    );
+}
+
 #[test]
 fn recipe_149_runs_with_a_feed_and_without_one_at_every_jobs() {
     let _serial = serial();

From d52c26fc61a603fd6b6710f38c2359e10bdb4f63 Mon Sep 17 00:00:00 2001
From: Jon-Carlos Rivera <jon.carlos.rivera@gmail.com>
Date: Mon, 19 Oct 2026 09:01:48 -0700
Subject: [PATCH 2/7] lanes: the reader of an audio clock with ticks held back
 waits for the packets behind it, not the ticks

A packet of 1024 samples settles 1024 ticks of a clock cut a sample a
tick, so a lane holding ticks back did so after every packet, and the
reader that carries the programme's sound beside its copy waited every
21 ms for the lane to be handed the rest. While ticks are held back, the
packets waiting on such a clock count against the cap, and its reader
waits once the lane is that many packets behind.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
---
 sidecar/NODE-CLI.md             |  3 +-
 sidecar/ffrwd-wasm/src/lanes.rs | 50 +++++++++++++++++++++++++++------
 sidecar/ffrwd-wasm/src/tick.rs  | 23 ++++++++++++++-
 3 files changed, 65 insertions(+), 11 deletions(-)

diff --git a/sidecar/NODE-CLI.md b/sidecar/NODE-CLI.md
index 98b1055..ba55ab9 100644
--- a/sidecar/NODE-CLI.md
+++ b/sidecar/NODE-CLI.md
@@ -377,7 +377,8 @@ tick that held them.
   bound to sound is cut 1024 ticks by a packet of 1024 samples, 48000 a
   second. Those past the bound stay with the clock and are cut as the
   queue drains, and while any are left the node counts as full to its
-  readers. A self-clocked node is handed one arrival an input a call, and
+  readers, except to the reader of that audio clock, which waits only
+  once the bound's worth of packets waits on the clock. A self-clocked node is handed one arrival an input a call, and
   the arrivals waiting on its input count against the same bound, so a
   decoder ahead of a slower encoder holds its reader back rather than
   queue every picture it made. A node that runs one tick at a time takes
diff --git a/sidecar/ffrwd-wasm/src/lanes.rs b/sidecar/ffrwd-wasm/src/lanes.rs
index edbbdfb..19807d2 100644
--- a/sidecar/ffrwd-wasm/src/lanes.rs
+++ b/sidecar/ffrwd-wasm/src/lanes.rs
@@ -405,7 +405,12 @@ impl State {
     /// queued on its input, which count against the same bound as a task
     /// queue: without that, a decoder ahead of a slower encoder would queue
     /// every picture it made. An input clock is cut no further than the
-    /// cap, and while its cut stopped there the lane counts as full.
+    /// cap, and while its cut stopped there the lane counts as full, but
+    /// to the reader of its audio clock: one packet of sound settles as
+    /// many ticks as it has samples over the stride, so while ticks are
+    /// held back what waits there is counted in packets, against the same
+    /// bound, and that reader is held back only once the lane is that many
+    /// packets behind.
     fn room(&self, stream: u32, consumers: &[Consumer]) -> bool {
         consumers.iter().all(|c| match c {
             Consumer::Lane(j) => {
@@ -416,8 +421,21 @@ impl State {
                     }
                     _ => 0,
                 };
+                let sound = match &lane.intake {
+                    LaneIntake::Assembled(a) if lane.clock == ClockKind::Input => {
+                        a.audio_backlog(stream)
+                    }
+                    _ => None,
+                };
                 lane.stopped
-                    || (lane.queue.len() < self.cap && lane.withheld.is_none() && queued < self.cap)
+                    || match sound {
+                        Some(packets) if lane.withheld.is_some() => packets < self.cap,
+                        _ => {
+                            lane.queue.len() < self.cap
+                                && lane.withheld.is_none()
+                                && queued < self.cap
+                        }
+                    }
             }
             Consumer::Writer(w, _) => self.writers[*w].len() < self.cap,
         })
@@ -2166,7 +2184,8 @@ mod tests {
         };
         assert!(scheduler.arrive(0, packet(0)));
         wait_until("the first tick ran", || !seen.lock().unwrap().is_empty());
-        // The first packet's ticks went in one turn; the second's wait.
+        // The first packet's ticks went in one turn; the second's wait, and
+        // the reader goes on until the cap's worth of packets waits.
         assert!(scheduler.arrive(0, packet(1024)));
         {
             let state = scheduler.shared.lock();
@@ -2180,14 +2199,28 @@ mod tests {
                 lane.withheld.is_some(),
                 "the rest of the packet's ticks wait"
             );
+            assert!(
+                state.room(0, &state.consumers(0)),
+                "one packet behind, the reader goes on"
+            );
+        }
+        for k in 2..5 {
+            assert!(scheduler.arrive(0, packet(k * 1024)));
+        }
+        {
+            let state = scheduler.shared.lock();
+            assert!(
+                state.lanes[0].queue.len() <= state.cap,
+                "the queue holds no more than the cap"
+            );
             assert!(
                 !state.room(0, &state.consumers(0)),
-                "the reader waits while ticks are withheld"
+                "the reader waits once the cap's worth of packets waits"
             );
         }
         gate.open();
         wait_until("every sample's tick ran", || {
-            seen.lock().unwrap().len() == 2048
+            seen.lock().unwrap().len() == 5 * 1024
         });
         assert!(scheduler.end(0));
         let (done, finished) = mpsc::channel();
@@ -2197,7 +2230,7 @@ mod tests {
             .recv_timeout(Duration::from_secs(5))
             .expect("the lane ended with its input")
             .expect("the run");
-        assert_eq!(*seen.lock().unwrap(), (0..2048).collect::<Vec<i64>>());
+        assert_eq!(*seen.lock().unwrap(), (0..5 * 1024).collect::<Vec<i64>>());
         let queues = shared_queues.lock().unwrap();
         assert!(
             queues[1..1024].iter().all(|q| *q == 4),
@@ -2205,9 +2238,8 @@ mod tests {
             &queues[..12]
         );
         assert!(
-            queues[1024..2048].iter().all(|q| *q == 0),
-            "the second packet's ticks ran in one turn: {:?}",
-            &queues[1024..1036]
+            queues.iter().all(|q| *q <= 4),
+            "no call saw more than the cap"
         );
     }
 
diff --git a/sidecar/ffrwd-wasm/src/tick.rs b/sidecar/ffrwd-wasm/src/tick.rs
index 2e11b36..eaf9e30 100644
--- a/sidecar/ffrwd-wasm/src/tick.rs
+++ b/sidecar/ffrwd-wasm/src/tick.rs
@@ -571,6 +571,9 @@ pub struct Assembler {
     /// how many, and their bytes, which a timing clock carries none of.
     audio_buffer: Vec<u8>,
     audio_buffered: usize,
+    /// How many samples the packet merged last brought, for how many
+    /// packets the samples not yet consumed are.
+    audio_packet: usize,
     audio_pts: Option<i64>,
     /// Whether the last tick has been made.
     done: bool,
@@ -893,6 +896,7 @@ impl Assembler {
             packets_time: None,
             audio_buffer: Vec::new(),
             audio_buffered: 0,
+            audio_packet: 0,
             audio_pts: None,
             done: false,
             made_end: None,
@@ -1007,6 +1011,21 @@ impl Assembler {
         Some(self.inputs.iter().find(|i| i.id == id)?.queue.len())
     }
 
+    /// How many packets wait on an audio clock bound to stream `id`, those
+    /// not yet looked at and those whose samples no tick has taken yet;
+    /// None where `id` is not an audio clock.
+    pub fn audio_backlog(&self, id: u32) -> Option<usize> {
+        let ClockState::Input(clock) = self.clock else {
+            return None;
+        };
+        let input = &self.inputs[clock];
+        if input.id != id || input.kind != PortKind::Audio {
+            return None;
+        }
+        let buffered = self.audio_buffered.div_ceil(self.audio_packet.max(1));
+        Some(input.queue.len() + buffered)
+    }
+
     /// How many more frames the hold input bound to stream `id` takes before
     /// its buffer is full, whoever feeds it; None where `id` is not held.
     pub fn held_room(&self, id: u32) -> Option<usize> {
@@ -1354,7 +1373,9 @@ impl Assembler {
                     if self.audio_buffered == 0 {
                         self.audio_pts = Some(frame.pts);
                     }
-                    self.audio_buffered += sample_count(frame, audio.sample_len());
+                    let samples = sample_count(frame, audio.sample_len());
+                    self.audio_buffered += samples;
+                    self.audio_packet = samples;
                     self.audio_buffer.extend_from_slice(frame.data.host());
                     input.queue.pop_front();
                 }

From fb1f933bc6dd8b4292f2db8e580696ee66092da1 Mon Sep 17 00:00:00 2001
From: Jon-Carlos Rivera <jon.carlos.rivera@gmail.com>
Date: Mon, 19 Oct 2026 09:01:49 -0700
Subject: [PATCH 3/7] plans: the 1023-task queue and the 0.32.0 pacing were
 foundmeat's listener on a clock cut a sample a tick

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
---
 plans/gpu-resident-frames.md | 34 ++++++++++++++++++++++------------
 1 file changed, 22 insertions(+), 12 deletions(-)

diff --git a/plans/gpu-resident-frames.md b/plans/gpu-resident-frames.md
index 03bf066..07e5e29 100644
--- a/plans/gpu-resident-frames.md
+++ b/plans/gpu-resident-frames.md
@@ -2924,18 +2924,28 @@ probe. The real findings from that work:
 
 ### What is left
 
-- A lane's task queue can grow past the scheduler's cap. On foundmeat's
-  libx264 run, a node of the foundmeat package in compose's sidecar reached
-  1023 tasks against a cap of 10, while the reader of the stream it reads
-  waited for room. How it got past the cap is not known yet. The cap
-  (`lane.queue.len() < cap`) is checked only where a producer waits for
-  room, before a reader delivers and before a lane starts a task, so tasks
-  queued any other way would not be counted. The check predates the
-  branch's change to the bound, since a6a649c counts only the arrivals
-  queued at a self-clocked reader. While such a queue stays above the cap,
-  every reader of the streams that lane reads is held back with it. Next:
-  find which path queued the tasks, with the `room:` trace and the lane's
-  intake, and bound it.
+- Done (2026-10-19, branch `fix/lane-queue-cap`): the 1023-task queue and
+  0.32.0 falling short of real time. The lane was `listen_audio` in
+  compose's sidecar. Its clock is the programme's sound, an `asplit` copy
+  that rides one input with compose's own `a`, and the module declares
+  a window of one sample, so each 1024-sample packet settled 1024 ticks.
+  An input clock's cut queued every settled tick at once (the cap was
+  checked only where a reader delivers), and every tick cost a turn of the
+  scheduler. On top of that, 3dad921's pace check walked the whole quarter
+  second of ticks (12000 of them) on every tick whenever the clock ran at
+  real time. The lane fell behind and held back the reader that carries
+  compose's sound, so the programme ran at 50 fps: 8.3 media seconds per
+  10 s of wall on 0.32.0 with the main feed alone, against 10.15 on
+  0.31.0. Bisected in a container on the box. a6a649c and 912bf25
+  reverted: 8.34. bcce48e: 10.15. 8f6b9d9: 8.34. 0.32.0 with the pace
+  check off: 10.15. Fixed: the pace keeps running extremes and is asked
+  only where a port-fed hold uses it; the cut stops at the cap and holds
+  the next tick back; a one-at-a-time lane takes up to 1024 settled ticks
+  in a turn; a turn's progress marks leave once; and an audio clock's
+  reader is held back by packets waiting, not by ticks. With two cams the
+  query keeps real time (10.15 against 8.02 on 0.32.0). Left for
+  foundmeat: `listen_audio` could declare a window of a packet or more,
+  since it reads only the times.
 - The host waits for the decoder's write before each copy is submitted,
   and the decode thread waits with it. A wait on the queue would need a
   submit with wait semaphores, which wgpu-core does not have; a raw submit

From 028792ac61161d251baf90cf5a303a4c039bed05 Mon Sep 17 00:00:00 2001
From: Jon-Carlos Rivera <jon.carlos.rivera@gmail.com>
Date: Mon, 19 Oct 2026 09:45:50 -0700
Subject: [PATCH 4/7] lanes: a node that finishes or fails in the middle of a
 turn, tested

The ticks of the turn after the one it finished on are skipped, as a
lane that stops drops its queue, a last tick queued behind them goes
with them, and the last call is made by the instance that finished.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
---
 sidecar/ffrwd-wasm/src/lanes.rs | 189 +++++++++++++++++++++++++++++++-
 1 file changed, 188 insertions(+), 1 deletion(-)

diff --git a/sidecar/ffrwd-wasm/src/lanes.rs b/sidecar/ffrwd-wasm/src/lanes.rs
index 19807d2..2d98e79 100644
--- a/sidecar/ffrwd-wasm/src/lanes.rs
+++ b/sidecar/ffrwd-wasm/src/lanes.rs
@@ -1271,6 +1271,8 @@ enum Delivery {
 /// What several ticks of one lane hand on, less each progress mark that a
 /// later one on the same stream says again with nothing between them on
 /// that stream: a turn of many small ticks moves its readers' progress once.
+/// A mark is kept when it is the last on its stream before the stream's
+/// next item, end or rows, or the last on its stream at all.
 fn superseded_progress_dropped(deliveries: Vec<(u32, Delivery)>) -> Vec<(u32, Delivery)> {
     let mut marked: HashSet<u32> = HashSet::new();
     let mut kept: Vec<(u32, Delivery)> = deliveries
@@ -1711,7 +1713,10 @@ fn worker(shared: &Shared) {
                     PROFILE.process.add(0, started);
                     // A tick the node says it finished on, or failed on, is
                     // the last of the turn: what was taken after it is not
-                    // run, as a lane that stops drops its queue.
+                    // run, as a lane that stops drops its queue. Its flush
+                    // skips their ordinals, drops what is queued (a last
+                    // tick behind them too) and queues the last call for
+                    // the instance that finished; a failure stops the run.
                     let stop = result.as_ref().map_or(true, |emitted| emitted.finished);
                     let result = result.map(|emitted| Done {
                         emitted,
@@ -2243,6 +2248,188 @@ mod tests {
         );
     }
 
+    /// A node that waits on `gate` in its first call and finishes, or
+    /// fails, on its third; it notes each call's pts and whether it was
+    /// the last.
+    struct Stopping {
+        shape: NodeShape,
+        calls: Arc<Mutex<Vec<(i64, bool)>>>,
+        gate: Arc<Gate>,
+        fail: bool,
+    }
+
+    impl Node for Stopping {
+        fn name(&self) -> &str {
+            "stopping"
+        }
+
+        fn shape(&self) -> &NodeShape {
+            &self.shape
+        }
+
+        fn set_params(&mut self, _params: &str) -> Result<()> {
+            Ok(())
+        }
+
+        fn process(&mut self, tick: Tick) -> Result<Emitted> {
+            let count = {
+                let mut calls = self.calls.lock().unwrap();
+                calls.push((tick.pts, tick.last));
+                calls.len()
+            };
+            if count == 1 {
+                self.gate.wait();
+            }
+            if count == 3 && !tick.last {
+                if self.fail {
+                    bail!("the third call fails");
+                }
+                return Ok(Emitted {
+                    finished: true,
+                    ..Emitted::default()
+                });
+            }
+            Ok(Emitted::default())
+        }
+    }
+
+    /// What a run of [`a_turn_cut_short`] came to: the node's calls, the
+    /// run's end, and the lane's next ordinal, its front task's and whether
+    /// that is the last, while the first call waited.
+    struct CutShort {
+        calls: Vec<(i64, bool)>,
+        result: Result<()>,
+        queued: Vec<Option<u64>>,
+    }
+
+    /// One lane of a node clocked by its sound a sample a tick, its first
+    /// turn holding six ticks and the last tick queued behind them, the node
+    /// finishing (or failing) on the third: the run ends, and nothing waits.
+    fn a_turn_cut_short(fail: bool) -> CutShort {
+        let shape = NodeShape {
+            inputs: vec![input_port("a", PortKind::Audio, Pairing::Lockstep)],
+            outputs: Vec::new(),
+            clock: Clock::Input("a".to_string()),
+            pure: false,
+            one_to_one: false,
+            bounded: true,
+            relation: Vec::new(),
+        };
+        let bound = BoundStream {
+            port: "a".to_string(),
+            id: 0,
+            info: StreamInfo::default(),
+            time_base: TimeBase {
+                num: 1,
+                den: 48_000,
+            },
+            format: StreamFormat::Audio(AudioFormat {
+                sample_rate: 48_000,
+                channels: 1,
+                sample_fmt: "f32",
+                channel_layout: None,
+            }),
+            rendition: Default::default(),
+            row: None,
+            decode_delay: 0,
+            latency: None,
+            hint: Default::default(),
+        };
+        let assembler = Assembler::new(&shape, &[bound], "stopping", &[]).expect("an assembler");
+        let calls = Arc::new(Mutex::new(Vec::new()));
+        let gate = Arc::new(Gate::default());
+        let node = Stopping {
+            shape: shape.clone(),
+            calls: Arc::clone(&calls),
+            gate: Arc::clone(&gate),
+            fail,
+        };
+        let plan = Plan {
+            lanes: vec![LaneSpec {
+                gpu: HashMap::new(),
+                name: "stopping".to_string(),
+                shape,
+                intake: Intake::Assembled(Box::new(assembler)),
+                bound: vec![0],
+                state: Vec::new(),
+                tick_base: TimeBase {
+                    num: 1,
+                    den: 48_000,
+                },
+                runners: vec![Box::new(node)],
+                opener: None,
+                ports: Vec::new(),
+                rows: None,
+                unbuffered: false,
+            }],
+            consumers: HashMap::from([(0, vec![Consumer::Lane(0)])]),
+            writers: Vec::new(),
+            mirrors: HashMap::new(),
+        };
+        let scheduler = Scheduler::start(plan, 1, &WakeSlot::new()).expect("the lanes");
+        let packet = TickFrame {
+            pts: 0,
+            duration: Some(6),
+            data: Arc::new(vec![0u8; 6 * 4]).into(),
+            rows: Vec::new(),
+        };
+        assert!(scheduler.arrive(0, Item::Frame(packet)));
+        wait_until("the first tick ran", || !calls.lock().unwrap().is_empty());
+        assert!(scheduler.end(0));
+        let queued = {
+            let state = scheduler.shared.lock();
+            let lane = &state.lanes[0];
+            vec![
+                Some(lane.next_ordinal),
+                lane.queue.front().map(|t| t.ordinal),
+                Some(u64::from(lane.queue.front().is_some_and(|t| t.tick.last))),
+            ]
+        };
+        gate.open();
+        let (done, finished) = mpsc::channel();
+        thread::spawn(move || done.send(scheduler.finish()));
+        let result = finished
+            .recv_timeout(Duration::from_secs(5))
+            .expect("the run ended, nothing waiting");
+        let calls = calls.lock().unwrap().clone();
+        CutShort {
+            calls,
+            result,
+            queued,
+        }
+    }
+
+    #[test]
+    fn a_node_that_finishes_in_a_turn_skips_the_rest_of_it_and_takes_its_last_call() {
+        let CutShort {
+            calls,
+            result,
+            queued,
+        } = a_turn_cut_short(false);
+        assert_eq!(
+            queued,
+            [Some(7), Some(6), Some(1)],
+            "six ticks in the turn, the last queued behind them"
+        );
+        result.expect("the run");
+        assert_eq!(
+            calls,
+            [(0, false), (1, false), (2, false), (3, true)],
+            "the ticks after the one it finished on are skipped, and the last call follows"
+        );
+    }
+
+    #[test]
+    fn a_node_that_fails_in_a_turn_stops_the_run() {
+        let CutShort { calls, result, .. } = a_turn_cut_short(true);
+        let error = result.expect_err("the run failed");
+        assert!(
+            format!("{error:#}").contains("the third call fails"),
+            "{error:#}"
+        );
+        assert_eq!(calls, [(0, false), (1, false), (2, false)]);
+    }
+
     /// A self-clocked reader (a decoder) slower than what reaches it holds
     /// its writer back once the cap's worth of arrivals waits on its input,
     /// whatever else the scheduler bounds.

From 618fff56efb99eec478667253f0784cd4694130e Mon Sep 17 00:00:00 2001
From: Jon-Carlos Rivera <jon.carlos.rivera@gmail.com>
Date: Mon, 19 Oct 2026 11:36:51 -0700
Subject: [PATCH 5/7] plans: the lane branch waits until its change is wanted;
 the pace fix shipped in 0.32.1

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
---
 plans/gpu-resident-frames.md | 9 +++++++--
 1 file changed, 7 insertions(+), 2 deletions(-)

diff --git a/plans/gpu-resident-frames.md b/plans/gpu-resident-frames.md
index 07e5e29..6e92857 100644
--- a/plans/gpu-resident-frames.md
+++ b/plans/gpu-resident-frames.md
@@ -2924,8 +2924,13 @@ probe. The real findings from that work:
 
 ### What is left
 
-- Done (2026-10-19, branch `fix/lane-queue-cap`): the 1023-task queue and
-  0.32.0 falling short of real time. The lane was `listen_audio` in
+- The 1023-task queue and 0.32.0 falling short of real time (2026-10-19).
+  The pace fix shipped in 0.32.1. The rest is on branch
+  `fix/lane-queue-cap`, unmerged until the lane change is wanted for
+  something: the 1023 queue did not keep 0.31.0 from real time, and the
+  hangs seen while testing the branch were the scheduler's waker being set
+  after its workers started, on main too (`fix/waker-before-workers`).
+  The lane was `listen_audio` in
   compose's sidecar. Its clock is the programme's sound, an `asplit` copy
   that rides one input with compose's own `a`, and the module declares
   a window of one sample, so each 1024-sample packet settled 1024 ticks.

From 7eaa59c48a20884f93a6bcde5d34642894c072f2 Mon Sep 17 00:00:00 2001
From: Jon-Carlos Rivera <jon.carlos.rivera@gmail.com>
Date: Mon, 19 Oct 2026 13:07:35 -0700
Subject: [PATCH 6/7] gpu: a frame's planes the host made are kept for the next
 frame of the same size and format

Every conversion on the device and every upload made new textures and
dropped them with the frame, an allocation and a free in the driver at each
frame. The textures the host makes are now kept, up to four free of each
size and format, and handed to the next plane; a module's texture a frame
owns is dropped as before.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
(cherry picked from commit f3c5b0e1acc78f64455d1e83df4f595ccbd5daf5)
---
 sidecar/runtime/src/gpu/convert.rs |  4 +-
 sidecar/runtime/src/gpu/frames.rs  | 76 +++++++++++++++++++++++++-----
 2 files changed, 67 insertions(+), 13 deletions(-)

diff --git a/sidecar/runtime/src/gpu/convert.rs b/sidecar/runtime/src/gpu/convert.rs
index 6604fe3..0108781 100644
--- a/sidecar/runtime/src/gpu/convert.rs
+++ b/sidecar/runtime/src/gpu/convert.rs
@@ -43,7 +43,7 @@ use wgpu_core::id::{BindGroupId, BindGroupLayoutId, BufferId, TextureViewId};
 use wgpu_core::id::{PipelineLayoutId, RenderPipelineId, ShaderModuleId};
 use wgpu_types::TextureFormat;
 
-use super::frames::{create_plane, Plane};
+use super::frames::{create_plane, release_plane, Plane};
 use crate::profile::PROFILE;
 
 const SHADER: &str = r#"
@@ -834,7 +834,7 @@ fn render(
         Ok(()) => Ok(planes),
         Err(e) => {
             for plane in &planes {
-                global.texture_drop(plane.texture);
+                release_plane(global, plane.texture);
             }
             Err(e)
         }
diff --git a/sidecar/runtime/src/gpu/frames.rs b/sidecar/runtime/src/gpu/frames.rs
index e9a4c8b..8b44a20 100644
--- a/sidecar/runtime/src/gpu/frames.rs
+++ b/sidecar/runtime/src/gpu/frames.rs
@@ -208,7 +208,7 @@ impl GpuFrame {
                 Ok(plane) => plane,
                 Err(e) => {
                     for made in &planes {
-                        global.texture_drop(made.texture);
+                        release_plane(&global, made.texture);
                     }
                     return Err(e);
                 }
@@ -233,7 +233,7 @@ impl GpuFrame {
             planes.push(plane);
             if let Err(e) = written {
                 for made in &planes {
-                    global.texture_drop(made.texture);
+                    release_plane(&global, made.texture);
                 }
                 bail!("uploading a {w}x{h} {format:?} plane: {e}");
             }
@@ -505,6 +505,54 @@ fn read_plane(
     result
 }
 
+/// The size and format of a texture [`create_plane`] made.
+type PlaneKey = (u32, u32, TextureFormat);
+
+/// The textures [`create_plane`] made, and those of them no frame holds,
+/// kept for the next plane of the same size and format. Making and
+/// destroying a texture costs the driver an allocation each time, at every
+/// frame of a conversion or an upload. A texture used again is written
+/// after every use queued before it, since wgpu orders the uses of a
+/// texture on its queue. Only textures made here are kept, so a module's
+/// texture a frame owns is never handed out as a plane.
+#[derive(Default)]
+struct Recycled {
+    made: std::collections::HashMap<TextureId, PlaneKey>,
+    free: std::collections::HashMap<PlaneKey, Vec<TextureId>>,
+}
+
+/// How many free textures of one size and format are kept.
+const KEPT: usize = 4;
+
+fn recycled() -> std::sync::MutexGuard<'static, Recycled> {
+    static RECYCLED: OnceLock<Mutex<Recycled>> = OnceLock::new();
+    RECYCLED
+        .get_or_init(Mutex::default)
+        .lock()
+        .unwrap_or_else(|e| e.into_inner())
+}
+
+/// Lets go of `texture`, a plane a frame owned: kept for a later plane
+/// where [`create_plane`] made it and fewer than [`KEPT`] of its kind are
+/// free, dropped otherwise.
+pub(super) fn release_plane(global: &Global, texture: TextureId) {
+    {
+        let mut recycled = recycled();
+        if let Some(key) = recycled.made.get(&texture).copied() {
+            let free = recycled.free.entry(key).or_default();
+            if free.len() < KEPT {
+                free.push(texture);
+                return;
+            }
+            recycled.made.remove(&texture);
+        }
+    }
+    global.texture_drop(texture);
+}
+
+/// A texture for one plane of a frame: a free one of the same size and
+/// format where one is kept ([`release_plane`]), else a new one. Every
+/// caller writes each texel before anything reads it.
 pub(super) fn create_plane(
     global: &Global,
     device: &SharedDevice,
@@ -513,6 +561,18 @@ pub(super) fn create_plane(
     format: TextureFormat,
     bytes_per_pixel: u32,
 ) -> Result<Plane> {
+    let key = (width, height, format);
+    let plane = |texture| Plane {
+        texture,
+        aspect: TextureAspect::All,
+        width,
+        height,
+        format,
+        bytes_per_pixel,
+    };
+    if let Some(texture) = recycled().free.get_mut(&key).and_then(Vec::pop) {
+        return Ok(plane(texture));
+    }
     let desc = wgpu_types::TextureDescriptor {
         label: Some(Cow::Borrowed("ffrwd frame")),
         size: Extent3d {
@@ -532,14 +592,8 @@ pub(super) fn create_plane(
         global.texture_drop(texture);
         bail!("a {width}x{height} {format:?} texture: {e}");
     }
-    Ok(Plane {
-        texture,
-        aspect: TextureAspect::All,
-        width,
-        height,
-        format,
-        bytes_per_pixel,
-    })
+    recycled().made.insert(texture, key);
+    Ok(plane(texture))
 }
 
 /// Copies one plane into `buffer`, waits for it, and appends its rows,
@@ -630,7 +684,7 @@ fn copy_out(
 impl Drop for GpuFrame {
     fn drop(&mut self) {
         for texture in &self.owned {
-            self.global.texture_drop(*texture);
+            release_plane(&self.global, *texture);
         }
         #[cfg(any(windows, target_os = "macos", target_os = "linux"))]
         if let Some(wrapped) = self

From 1a4453932fce6b7162d156e2217c5a43ebc74178 Mon Sep 17 00:00:00 2001
From: Jon-Carlos Rivera <jon.carlos.rivera@gmail.com>
Date: Mon, 19 Oct 2026 13:51:24 -0700
Subject: [PATCH 7/7] plans: where the Linux route's CPU went, and the two
 device-side copies left unmerged

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
---
 plans/gpu-resident-frames.md | 30 ++++++++++++++++++++++++------
 1 file changed, 24 insertions(+), 6 deletions(-)

diff --git a/plans/gpu-resident-frames.md b/plans/gpu-resident-frames.md
index 6e92857..2d380cb 100644
--- a/plans/gpu-resident-frames.md
+++ b/plans/gpu-resident-frames.md
@@ -2925,12 +2925,12 @@ probe. The real findings from that work:
 ### What is left
 
 - The 1023-task queue and 0.32.0 falling short of real time (2026-10-19).
-  The pace fix shipped in 0.32.1. The rest is on branch
-  `fix/lane-queue-cap`, unmerged until the lane change is wanted for
-  something: the 1023 queue did not keep 0.31.0 from real time, and the
-  hangs seen while testing the branch were the scheduler's waker being set
-  after its workers started, on main too (`fix/waker-before-workers`).
-  The lane was `listen_audio` in
+  The pace fix shipped in 0.32.1; the waker fix (the scheduler's waker
+  set after its workers started, on main too, the cause of the hangs
+  seen while testing) shipped in 0.32.2; the lane change below, held
+  back at first since the queue did not keep 0.31.0 from real time,
+  went in once the CPU profile showed it was the idle broadcast's
+  cost. The lane was `listen_audio` in
   compose's sidecar. Its clock is the programme's sound, an `asplit` copy
   that rides one input with compose's own `a`, and the module declares
   a window of one sample, so each 1024-sample packet settled 1024 ticks.
@@ -2951,6 +2951,24 @@ probe. The real findings from that work:
   query keeps real time (10.15 against 8.02 on 0.32.0). Left for
   foundmeat: `listen_audio` could declare a window of a packet or more,
   since it reads only the times.
+- The Linux route's CPU was measured on 2026-10-20 (perf, strace, gdb on
+  the box): 1.34 cores idle on 0.32.1, 61% of it the kernel, the four
+  workers making about 30,000 context switches a second each because a
+  node clocked a sample a tick (foundmeat's listener on the programme's
+  sound) was cut one task a sample and every dispatch and completion
+  woke every worker. The host's waits on the device cost under 1%, and
+  CUDA's waits already block (libav sets `CU_CTX_SCHED_BLOCKING_SYNC`
+  itself). The lane change (the cut stops at the cap, ticks taken in
+  turns) brought it to 0.99 and keeping the host's plane textures to
+  0.92, the Windows figure; the node's own window of a packet rather
+  than a sample (foundmeat 0.3.0) brought their idle broadcast to 0.46
+  on 0.32.2 alone. Two further commits, the decoder's copy and the
+  encoder's feed submitted on the device after the picture's semaphore
+  with no host wait (branch `fix/linux-route-cpu`, 91c0689 and
+  6a87172), measured within noise and rest on wgpu-core's submit
+  holding its device fence lock for writing, which is not documented;
+  they stay unmerged until a wgpu release documents that or the saving
+  is worth the dependence.
 - The host waits for the decoder's write before each copy is submitted,
   and the decode thread waits with it. A wait on the queue would need a
   submit with wait semaphores, which wgpu-core does not have; a raw submit