Concept: Bounded Concurrency and Retrying the Retryable

Kind: Concept (read, do not code). This is where the coordinator stops being a library and starts being an engine.

Everything before this arc was a part waiting to be driven. control-core gave us the domain and the WorkerHandle seam; control-eval gave us the workload one worker runs; control-store gave us durable runs and jobs; control-service gave us the API that queues them. The scheduler is the piece that takes a queued run and executes it to completion — and doing that well is entirely about two questions:

1. How many jobs run at once? Not one at a time (we would waste the whole async foundation), and not all at once (that way lies exhaustion). Some bounded number.
2. What happens when a job fails? Some failures are worth retrying on another worker; some are not. The scheduler must tell them apart and act — which is exactly what the error taxonomy from Part II was built to let it do.

This chapter is about the two tools that answer those questions: futures::stream::buffer_unordered for bounded concurrency, and a small retry loop that walks the worker pool. Both are tiny. The reasons they are shaped the way they are are the whole lesson.

Why bounded, and not the two obvious extremes

Picture a run that fans out to 200 jobs — say 10 models × 20 epochs. You already know from the async arc that awaiting them one after another is the wrong move: each job spends almost all its wall-clock waiting on a model call, and doing them sequentially means the machine sits idle 199/200 of the time. Async exists precisely so those waits overlap.

So the naive fix is "launch all 200 at once" — futures::future::join_all(jobs) or a tokio::spawn per job. That overlaps every wait, and for a toy it looks great. In a real coordinator it is a foot-gun:

• You open 200 concurrent connections to the model API. Most providers rate-limit hard; you get a wall of 429s, and your "concurrency" turns into a wall of retries.
• You hold 200 jobs' worth of vignette batches and in-flight buffers in memory at once. For large runs that is real pressure, and it scales with the run size — the one thing you do not control.
• You have no back-pressure. If workers are slower than you launch, the pile of started-but-unfinished work grows without limit.

The honest answer is a ceiling: run at most N jobs concurrently, and as each one finishes, start the next. N waits overlap — you get most of the throughput of "all at once" — but you never have more than N in flight, so memory, connections, and rate-limit pressure are all bounded by a number you choose, independent of how big the run is. That is bounded concurrency, and buffer_unordered is the one-line way to get it.

buffer_unordered, and proof the ceiling holds

buffer_unordered(n) is an adapter on a stream of futures. You hand it a Stream whose items are Futures; it polls up to n of them at once, yields each result as it resolves, and starts another future each time a slot frees. You drive the whole thing by pulling the resulting stream with .next().await (or .collect()).

Here is the shape, with an instrument bolted on to prove the ceiling is real. Twelve jobs, each a 50ms stand-in for a model call; a buffer_unordered(4); and two atomics that track how many are in flight and the high-water mark. Predict the peak before you run it.

use futures::stream::{self, StreamExt};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::Duration;
use tokio::time::sleep;

#[tokio::main]
async fn main() {
    let inflight = Arc::new(AtomicUsize::new(0));
    let peak = Arc::new(AtomicUsize::new(0));

    // Twelve jobs, each a stand-in for a model call that mostly waits.
    let tasks = (0..12u32).map(|job| {
        let inflight = inflight.clone();
        let peak = peak.clone();
        async move {
            let now = inflight.fetch_add(1, Ordering::SeqCst) + 1;
            peak.fetch_max(now, Ordering::SeqCst);
            sleep(Duration::from_millis(50)).await; // waiting on the "network"
            inflight.fetch_sub(1, Ordering::SeqCst);
            job * 2
        }
    });

    // buffer_unordered(4): at most four futures in flight at once.
    let mut stream = stream::iter(tasks).buffer_unordered(4);
    let mut sum = 0u32;
    while let Some(doubled) = stream.next().await {
        sum += doubled;
    }
    println!("sum = {sum}, peak concurrency = {}", peak.load(Ordering::SeqCst));
}

sum = 132, peak concurrency = 4

The sum is a sanity check that every one of the twelve jobs ran and was collected — (0+1+…+11)·2 = 132. The number that matters is peak concurrency = 4: no matter that we defined twelve jobs, never more than four were in flight at any instant. Raise the 4 to 8 and the peak becomes 8; the run finishes faster but leans harder on whatever is downstream. That single argument is the knob — the whole point is that it is your number, not the run's.

error[E0599]: no method named `buffer_unordered` found for struct
              `futures::stream::Iter<I>` in the current scope
   |
 5 |     let mut stream = stream::iter(tasks).buffer_unordered(4);
   |                                          ^^^^^^^^^^^^^^^^
   |
   = help: items from traits can only be used if the trait is in scope
help: trait `StreamExt` which provides `buffer_unordered` is implemented but
      not in scope; perhaps you want to import it
   |
 1 + use futures::StreamExt;
   |
help: there is a method `buffered` with a similar name

The second question: which failures are worth retrying

Bounded concurrency decides how many jobs run. The other half of the scheduler's job is deciding what to do when one fails — and this is where the error taxonomy you built in Part II finally does the work it was designed for.

Recall the one distinction that taxonomy encodes: ControlError::is_retryable() is true for Worker and Protocol (a worker died, a connection dropped) and false for Invalid and NotFound (a malformed job, a missing run). The scheduler is the code that reads that bit and acts on it:

• A retryable failure means "this exact job, handed to a different worker, might well succeed." A worker crashed mid-job; the network hiccuped. Re-dispatch it.
• A terminal failure means "this job will fail identically no matter who runs it." The spec is malformed. Retrying only burns time and other workers' attention. Surface it.

The mechanism is a loop that, on a retryable error, moves to the next worker in the pool and tries again — up to a cap so a genuinely broken job cannot retry forever. Here is that loop as a runnable toy, with a Worker whose behaviour we can pin, a RetryPolicy with max_attempts, and the exact index arithmetic the build uses (workers[(seed + attempt) % len]):

#[derive(Debug)]
enum DispatchError {
    Retryable(&'static str),
    Terminal(&'static str),
}
impl DispatchError {
    fn is_retryable(&self) -> bool {
        matches!(self, DispatchError::Retryable(_))
    }
    fn reason(&self) -> &str {
        match self {
            DispatchError::Retryable(m) | DispatchError::Terminal(m) => m,
        }
    }
}

enum Behavior {
    Ok,
    Retryable,
    Terminal,
}

struct Worker {
    id: &'static str,
    behavior: Behavior,
}
impl Worker {
    async fn dispatch(&self, job: u32) -> Result<String, DispatchError> {
        match self.behavior {
            Behavior::Ok => Ok(format!("{} handled job {job}", self.id)),
            Behavior::Retryable => Err(DispatchError::Retryable("connection reset")),
            Behavior::Terminal => Err(DispatchError::Terminal("malformed job spec")),
        }
    }
}

struct RetryPolicy {
    max_attempts: u32,
}

async fn run_job_with_retry(
    workers: &[Worker],
    policy: &RetryPolicy,
    seed: usize,
    job: u32,
) -> Result<String, DispatchError> {
    let mut attempt = 0u32;
    loop {
        let worker = &workers[(seed + attempt as usize) % workers.len()];
        println!("  attempt {attempt} -> worker {}", worker.id);
        match worker.dispatch(job).await {
            Ok(out) => return Ok(out),
            Err(e) if e.is_retryable() && attempt + 1 < policy.max_attempts => {
                println!("    retryable ({}) — moving to the next worker", e.reason());
                attempt += 1;
            }
            Err(e) => return Err(e),
        }
    }
}

#[tokio::main]
async fn main() {
    let policy = RetryPolicy { max_attempts: 3 };

    println!("scenario 1: flaky worker, then a good one");
    let workers = vec![
        Worker { id: "w0", behavior: Behavior::Retryable },
        Worker { id: "w1", behavior: Behavior::Ok },
    ];
    println!("  => {:?}\n", run_job_with_retry(&workers, &policy, 0, 42).await);

    println!("scenario 2: a terminal failure is not retried");
    let workers = vec![
        Worker { id: "w0", behavior: Behavior::Terminal },
        Worker { id: "w1", behavior: Behavior::Ok },
    ];
    println!("  => {:?}\n", run_job_with_retry(&workers, &policy, 0, 42).await);

    println!("scenario 3: every worker retryable — attempts exhaust");
    let workers = vec![
        Worker { id: "w0", behavior: Behavior::Retryable },
        Worker { id: "w1", behavior: Behavior::Retryable },
    ];
    println!("  => {:?}", run_job_with_retry(&workers, &policy, 0, 42).await);
}

Predict all three results before you run — which worker each attempt lands on, and whether it ends Ok or Err.

scenario 1: flaky worker, then a good one
  attempt 0 -> worker w0
    retryable (connection reset) — moving to the next worker
  attempt 1 -> worker w1
  => Ok("w1 handled job 42")

scenario 2: a terminal failure is not retried
  attempt 0 -> worker w0
  => Err(Terminal("malformed job spec"))

scenario 3: every worker retryable — attempts exhaust
  attempt 0 -> worker w0
    retryable (connection reset) — moving to the next worker
  attempt 1 -> worker w1
    retryable (connection reset) — moving to the next worker
  attempt 2 -> worker w0
  => Err(Retryable("connection reset"))

Read all three against the theme:

• Scenario 1 is the payoff. w0 fails retryably, the loop advances attempt and computes (0+1) % 2 = 1 → w1, which succeeds. A flaky worker cost one attempt, not the job. This is "lose nothing under worker failure" in miniature.
• Scenario 2 is the guard rail. w0 fails terminally, and the Err(e) => arm returns immediately — w1 is never tried. A malformed job does not get to march through every worker in the pool, one after another, failing identically each time.
• Scenario 3 is the cap earning its keep. Both workers are flaky, so no attempt ever succeeds; max_attempts = 3 stops the loop after three tries (indices 0,1,0 — the modulo wraps back to w0) and surfaces the last error instead of spinning forever.

One-for-one: the toy ↔ the real thing

Everything here maps straight onto the Scheduler you build next, piece for piece:

• buffer_unordered(4) over a stream of job-futures ↔ Scheduler::process_run — it builds one future per job and drives them with stream::iter(futures).buffer_unordered(self.concurrency), concurrency defaulting to 4.
• run_job_with_retry(workers, policy, seed, job) ↔ the real function of the same name — same workers[(seed + attempt) % workers.len()] walk, same RetryPolicy { max_attempts: 3 }, same "retryable-and-under-cap" guard.
• the toy Worker with an inherent async fn dispatch ↔ Arc<dyn WorkerHandle> — in the build the workers are trait objects behind the seam from Part I, so the same retry loop drives a LocalWorker today and a RemoteWorker in Part VIII without changing a line.
• DispatchError::is_retryable() ↔ ControlError::is_retryable() — the real taxonomy, doing the exact job it was designed for: telling the scheduler which failures to re-dispatch.
• the toy's per-job closure returning job * 2 ↔ the real closure that runs the retry loop, then appends the outcome's records to the log and records the outcome in the store — same bounded-concurrency skeleton, real side effects inside.

Hold the last point against the course's spine. The scheduler holds Arc<dyn WorkerHandle> and never learns local from remote; run_job_with_retry walks that pool on retryable failures. When Part VIII drops RemoteWorker into the pool, this loop is what transparently starts re-dispatching failed jobs across machines — bounded concurrency and retry-the-retryable, unchanged, now distributed.

Questions to lock

1. Why is bounded concurrency the right choice over both "one at a time" and "all at once"? What does the bound protect, and why does the right bound depend on downstream limits rather than on the run's size?
2. What does buffer_unordered(n) guarantee about how many futures are in flight, and why does it not require its futures to be 'static the way tokio::spawn would?
3. In run_job_with_retry, why are both clauses of the retry condition load-bearing? Describe the failure mode you get by dropping each one.
4. A Worker failure and a Protocol failure are both retryable; an Invalid failure is terminal. Trace what the scheduler does with each, and why "retry on the next worker" is the correct response to the retryable ones.

Next: we build it — plan_jobs, run_job_with_retry, Scheduler::process_run/tick, the LocalWorker behind the seam, and the WorkerSource that hands the scheduler its pool.