server/lib/src/pull_through.rs
//! **Pull-through on the durable queue, and who pays for the upstream bytes.**
//!
//! Two small things the proxy ([`crate::proxy::ProxyBackend`]) asks while it
//! walks its upstream chain, carried to it on the thread that runs the walk (the
//! proxy is a `RepositoryBackendTrait`, whose `fetch` has no room for a caller):
//!
//! * **who is being served** — [`charged_to`] names the consumer, and every
//! upstream hit the proxy takes is written to that consumer's
//! `Cost.upstream_fetch` in [`crate::admission::book`]. Before this, the only
//! count was `proxy_upstream_hits`, server-wide (`metering.rs`).
//! * **whether the walk may be deferred** — [`queued`] installs a
//! [`PullThrough`]: a primary miss then becomes ONE [`crate::jobs::Job::Fetch`]
//! (deduplicated on the coordinate) and the request is answered
//! [`FetchQueued`] → `503` + `Retry-After`, which pip, cargo and Maven 3.9.5 and later
//! retry (Maven before 3.9.5 does not retry a `503`; HOL-DOCFIND i).
//! The job walks the chain with backoff and writes the bytes through into the
//! cache; the client's retry is then a local hit.
//!
//! **Nothing is installed ⇒ nothing changes.** A walk with no context (every
//! existing caller, every proxy test, a box that does not configure the queue)
//! is the inline walk, byte for byte.
use crate::lock::LockOrRecover as _;
use std::cell::RefCell;
#[cfg(feature = "jobs")]
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
#[cfg(feature = "jobs")]
use std::time::Duration;
use traits::ArtifactId;
thread_local! {
static CHARGE: RefCell<Option<Charge>> = const { RefCell::new(None) };
}
#[cfg(feature = "jobs")]
thread_local! {
static QUEUED: RefCell<Option<QueueCtx>> = const { RefCell::new(None) };
}
#[derive(Clone)]
struct Charge {
consumer: String,
/// The admitted request's cost, when the walk runs under a door ticket:
/// the bytes then reach `Admit::record` (and the book) with the rest of it.
sink: Option<Arc<Mutex<crate::admission::CostTally>>>,
}
fn with_charge<R>(charge: Charge, f: impl FnOnce() -> R) -> R {
let prev = CHARGE.with(|c| c.replace(Some(charge)));
let out = f();
CHARGE.with(|c| *c.borrow_mut() = prev);
out
}
/// Run `f` with `consumer` named as the one the upstream bytes are charged to
/// (straight into the book — the job runner's case, which has no ticket).
/// Nests: the previous charge is restored after.
pub fn charged_to<R>(consumer: &str, f: impl FnOnce() -> R) -> R {
with_charge(
Charge {
consumer: consumer.to_string(),
sink: None,
},
f,
)
}
/// Run `f` charged to an admitted request's ticket.
pub fn charged_ticket<R>(ticket: &crate::admission::Ticket, f: impl FnOnce() -> R) -> R {
with_charge(
Charge {
consumer: ticket.consumer().to_string(),
sink: Some(ticket.sink()),
},
f,
)
}
/// The consumer the running walk is charged to, if one was named.
pub fn current_consumer() -> Option<String> {
CHARGE.with(|c| c.borrow().as_ref().map(|c| c.consumer.clone()))
}
/// **The proxy took `bytes` from an upstream** — charge them to the consumer
/// this walk runs for (nothing is charged when none was named: an operator's
/// CLI, a test).
pub fn note_upstream_fetch(bytes: u64) {
let Some(charge) = CHARGE.with(|c| c.borrow().clone()) else {
return;
};
match charge.sink {
Some(sink) => sink.lock_or_recover().upstream_fetch += bytes,
None => crate::admission::book().add_upstream_fetch(&charge.consumer, bytes),
}
}
/// A primary miss was handed to the queue. The door answers `503` +
/// `Retry-After: retry_after_secs`, never a hard error.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct FetchQueued {
pub job: String,
/// `requested` / `fetching` / `failed` (a retry is scheduled).
pub stage: String,
pub retry_after_secs: u64,
/// The last attempt's reason while a retry waits, else `""`.
pub reason: String,
}
impl std::fmt::Display for FetchQueued {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(
f,
"the artifact is being fetched from upstream (job {}, {}); retry in {} s",
self.job, self.stage, self.retry_after_secs
)?;
if !self.reason.is_empty() {
write!(f, " — last attempt: {}", self.reason)?;
}
Ok(())
}
}
impl std::error::Error for FetchQueued {}
/// The queued pull-through for this coordinate FAILED for good (every attempt
/// spent, or a failure retrying cannot fix). Answered once, then forgotten, so
/// the next request asks again.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct FetchFailed {
pub job: String,
pub reason: String,
}
impl std::fmt::Display for FetchFailed {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "pull-through job {} failed: {}", self.job, self.reason)
}
}
impl std::error::Error for FetchFailed {}
/// Read a queued answer back out of an error chain (the doors map it to 503).
pub fn queued_of(e: &anyhow::Error) -> Option<&FetchQueued> {
e.chain().find_map(|c| c.downcast_ref::<FetchQueued>())
}
/// Read a failed pull-through back out of an error chain (the doors map it to 502).
pub fn failed_of(e: &anyhow::Error) -> Option<&FetchFailed> {
e.chain().find_map(|c| c.downcast_ref::<FetchFailed>())
}
#[cfg(feature = "jobs")]
#[derive(Clone)]
struct QueueCtx {
pt: Arc<PullThrough>,
priority: u8,
}
/// Run `f` (a backend walk) with the queue installed: a primary miss in a
/// writable proxy becomes a [`crate::jobs::Job::Fetch`] at `priority`
/// (`jobs::PRIORITY_*`, from the consumer's admission class).
#[cfg(feature = "jobs")]
pub fn queued<R>(pt: &Arc<PullThrough>, priority: u8, f: impl FnOnce() -> R) -> R {
let prev = QUEUED.with(|q| {
q.replace(Some(QueueCtx {
pt: Arc::clone(pt),
priority,
}))
});
let out = f();
QUEUED.with(|q| *q.borrow_mut() = prev);
out
}
/// Whether a door installed the queue for the walk running on this thread —
/// the cheap question the proxy asks before it does any work for [`intercept`].
pub(crate) fn armed() -> bool {
#[cfg(feature = "jobs")]
{
QUEUED.with(|q| q.borrow().is_some())
}
#[cfg(not(feature = "jobs"))]
{
false
}
}
/// `None` ⇒ walk inline, as always. `Some(Ok(()))` ⇒ the queue fetched it just now:
/// re-read the primary. `Some(Err(FetchQueued | FetchFailed))` ⇒ answer that.
pub(crate) fn intercept(repo: &str, id: &ArtifactId) -> Option<anyhow::Result<()>> {
#[cfg(feature = "jobs")]
{
let ctx = QUEUED.with(|q| q.borrow().clone())?;
let who = current_consumer().unwrap_or_else(|| "anonymous".to_string());
ctx.pt.defer(repo, id, &who, ctx.priority)
}
#[cfg(not(feature = "jobs"))]
{
let _ = (repo, id);
None
}
}
/// **The queued pull-through**: the job queue, what this process remembers of
/// the fetches it deferred, and how long a request waits before it is told to
/// come back.
#[cfg(feature = "jobs")]
pub struct PullThrough {
queue: Arc<crate::jobs::JobQueue>,
/// `content_id → job id` for fetches this process deferred and whose
/// outcome a request has not yet been told.
/// Sharded (one lock per shard, picked by the coordinate's hash): a miss on
/// one artifact never waits on a miss on another.
memo: Box<[Mutex<HashMap<String, crate::jobs::JobId>>]>,
/// `Retry-After` while a fetch is requested or fetching.
retry_after_secs: u64,
/// How long the request thread waits for the job before it answers 503.
/// A small artifact is usually fetched inside it and the client never
/// sees the 503 at all; a big one is not held on the request thread.
wait: Duration,
}
#[cfg(feature = "jobs")]
impl PullThrough {
/// Bounded memo: one entry per coordinate deferred and not yet answered.
const MEMO_CAP: usize = 65_536;
pub fn new(queue: Arc<crate::jobs::JobQueue>, retry_after_secs: u64, wait: Duration) -> Self {
let shards = std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(4)
.next_power_of_two()
* 4;
Self {
queue,
memo: (0..shards).map(|_| Mutex::new(HashMap::new())).collect(),
retry_after_secs: retry_after_secs.max(1),
wait,
}
}
fn memo(&self, cid: &str) -> std::sync::MutexGuard<'_, HashMap<String, crate::jobs::JobId>> {
use std::hash::{Hash, Hasher};
let mut h = std::collections::hash_map::DefaultHasher::new();
cid.hash(&mut h);
self.memo[(h.finish() as usize) % self.memo.len()].lock_or_recover()
}
pub fn queue(&self) -> &Arc<crate::jobs::JobQueue> {
&self.queue
}
fn defer(
&self,
repo: &str,
id: &ArtifactId,
who: &str,
priority: u8,
) -> Option<anyhow::Result<()>> {
let job = crate::jobs::Job::Fetch {
repo: repo.to_string(),
artifact: crate::jobs::JobArtifact {
namespace: id.namespace.clone().unwrap_or_default(),
name: id.name.clone(),
version: id.version.clone(),
},
requested_by: who.to_string(),
};
let cid = job.content_id();
// A fetch this process deferred earlier and that is no longer open:
// done ⇒ the bytes should be local now; if the primary still missed
// (the cache-fill gate served-but-did-not-cache), walk inline rather
// than queue the same fetch forever. Failed ⇒ say so once.
if self.queue.open_job_for(&cid).is_none() {
let remembered = self.memo(&cid).remove(&cid);
if let Some(jid) = remembered {
match self.queue.status_of(&jid).as_deref() {
Some(s) if s == crate::jobs::status::DONE => return None,
Some(s)
if s == crate::jobs::status::FAILED
|| s == crate::jobs::status::INTERRUPTED =>
{
let reason = self
.queue
.get(&jid)
.ok()
.flatten()
.map(|v| v.reason)
.unwrap_or_default();
return Some(Err(anyhow::Error::new(FetchFailed { job: jid.0, reason })));
}
_ => {}
}
}
}
let jid = match self.queue.enqueue_with_priority(job, priority) {
Ok(j) => j,
Err(e) => {
// The ledger refused (closed for a growth, a full disk): the
// request is served the old way rather than not at all.
log::warn!("pull-through: could not queue {cid} ({e:#}); walking inline");
return None;
}
};
{
let mut m = self.memo(&cid);
if m.len() < Self::MEMO_CAP {
m.insert(cid.clone(), jid.clone());
}
}
// Only a `Full` consumer may wait (bounded) for its fetch on the request
// thread; every other class is answered 503 at once and holds nothing.
if !self.wait.is_zero() && priority == crate::jobs::PRIORITY_FULL {
if let Ok(Some(v)) = self.queue.await_completion(&jid, self.wait) {
self.memo(&cid).remove(&cid);
if v.status == crate::jobs::status::DONE {
return Some(Ok(())); // fetched inside the wait: serve it now
}
return Some(Err(anyhow::Error::new(FetchFailed {
job: jid.0,
reason: v.reason,
})));
}
}
Some(Err(anyhow::Error::new(self.queued_answer(&jid))))
}
fn queued_answer(&self, jid: &crate::jobs::JobId) -> FetchQueued {
let status = self.queue.status_of(jid).unwrap_or_default();
let mut retry = self.retry_after_secs;
let mut reason = String::new();
if status == crate::jobs::status::RETRY_WAIT {
if let Ok(Some(v)) = self.queue.get(jid) {
reason = v.reason;
if let Some(at) = v.next_attempt_micros {
let now = crate::audit::now_nanos() / 1_000;
retry = retry.max(((at - now).max(0) as u64).div_ceil(1_000_000));
}
}
}
FetchQueued {
job: jid.0.clone(),
stage: crate::jobs::stage_of(crate::jobs::kind::FETCH, &status).to_string(),
retry_after_secs: retry,
reason,
}
}
}
#[cfg(all(test, feature = "jobs"))]
mod tests {
use super::*;
use crate::jobs::{status, JobQueue, ServerJobRunner};
use std::sync::atomic::{AtomicUsize, Ordering};
use traits::{ArtifactFormat, RepositoryBackendTrait};
/// A writable in-memory cache — the proxy's primary.
#[derive(Default)]
struct Cache {
blobs: Mutex<HashMap<String, Vec<u8>>>,
}
fn key(id: &ArtifactId) -> String {
format!("{}@{}", id.name, id.version)
}
impl RepositoryBackendTrait for Cache {
fn caps(&self) -> traits::Caps {
traits::Caps::NONE
}
fn name(&self) -> &str {
"cache"
}
fn format(&self) -> ArtifactFormat {
ArtifactFormat::Raw
}
fn is_writable(&self) -> bool {
true
}
fn fetch(&self, id: &ArtifactId) -> anyhow::Result<Option<traits::Bytes>> {
Ok(self
.blobs
.lock()
.unwrap()
.get(&key(id))
.cloned()
.map(traits::Bytes::from))
}
fn put(&self, id: &ArtifactId, data: &[u8]) -> anyhow::Result<()> {
self.blobs.lock().unwrap().insert(key(id), data.to_vec());
Ok(())
}
fn handle_http2_request(
&self,
_: &str,
_: &str,
_: &[u8],
) -> anyhow::Result<(u16, Vec<(String, String)>, Vec<u8>)> {
Ok((404, vec![], vec![]))
}
}
/// An upstream that errors its first `fail_first` fetches, then serves
/// `body` for every coordinate; `delay` per fetch; counts every call.
struct Upstream {
calls: AtomicUsize,
fail_first: usize,
body: Vec<u8>,
delay: Duration,
}
impl RepositoryBackendTrait for Upstream {
fn caps(&self) -> traits::Caps {
traits::Caps::NONE
}
fn name(&self) -> &str {
"upstream"
}
fn format(&self) -> ArtifactFormat {
ArtifactFormat::Raw
}
fn is_writable(&self) -> bool {
false
}
fn fetch(&self, _id: &ArtifactId) -> anyhow::Result<Option<traits::Bytes>> {
let n = self.calls.fetch_add(1, Ordering::SeqCst);
std::thread::sleep(self.delay);
if n < self.fail_first {
anyhow::bail!("upstream refused the connection (attempt {})", n + 1);
}
Ok(Some(traits::Bytes::from(self.body.clone())))
}
fn put(&self, _: &ArtifactId, _: &[u8]) -> anyhow::Result<()> {
anyhow::bail!("read-only")
}
fn handle_http2_request(
&self,
_: &str,
_: &str,
_: &[u8],
) -> anyhow::Result<(u16, Vec<(String, String)>, Vec<u8>)> {
Ok((404, vec![], vec![]))
}
}
struct Rig {
cache: Arc<Cache>,
upstream: Arc<Upstream>,
proxy: Arc<dyn RepositoryBackendTrait>,
routes: Arc<crate::exposed::fast_routes::FastRoutes>,
}
fn rig(fail_first: usize, delay: Duration) -> Rig {
let cache = Arc::new(Cache::default());
let upstream = Arc::new(Upstream {
calls: AtomicUsize::new(0),
fail_first,
body: b"crate-bytes".to_vec(),
delay,
});
let proxy: Arc<dyn RepositoryBackendTrait> = Arc::new(crate::proxy::ProxyBackend::new(
cache.clone() as Arc<dyn RepositoryBackendTrait>,
vec![upstream.clone() as Arc<dyn RepositoryBackendTrait>],
));
let routes = Arc::new(crate::exposed::fast_routes::FastRoutes::new(vec![(
"cache".to_string(),
Arc::clone(&proxy),
)]));
Rig {
cache,
upstream,
proxy,
routes,
}
}
fn serde_id() -> ArtifactId {
ArtifactId {
namespace: None,
name: "serde".into(),
version: "1.0.0".into(),
}
}
fn queue_at(
dir: &std::path::Path,
routes: &Arc<crate::exposed::fast_routes::FastRoutes>,
workers: usize,
) -> Arc<JobQueue> {
let (q, _) = JobQueue::open(dir, workers).unwrap();
let q = Arc::new(q.with_retry_base(Duration::from_millis(20)));
q.start(Arc::new(ServerJobRunner::new(Arc::clone(routes), None)));
q
}
/// § Pull-through is COUNTED to the consumer that asked: the upstream bytes
/// land in its `upstream_fetch`, not in a server-wide counter only.
#[test]
fn a_pull_through_is_charged_to_the_consumer_that_asked() {
let r = rig(0, Duration::ZERO);
let who = "principal:charge-test-alice";
let got = charged_to(who, || r.proxy.fetch(&serde_id())).unwrap();
assert_eq!(got.as_deref(), Some(b"crate-bytes".as_ref()));
let (window, _) = crate::admission::book().usage_of(who);
assert_eq!(
window.upstream_fetch, 11,
"the 11 upstream bytes are alice's"
);
// A second read is a cache hit: nothing more is charged upstream.
charged_to(who, || r.proxy.fetch(&serde_id())).unwrap();
assert_eq!(crate::admission::book().usage_of(who).0.upstream_fetch, 11);
// Another consumer was charged nothing.
assert_eq!(
crate::admission::book()
.usage_of("principal:charge-test-bob")
.0
.upstream_fetch,
0
);
}
/// § Through a door ticket, the upstream bytes reach the ticket (and so
/// `Admit::record`) and are booked under `pull_through` for that consumer.
#[test]
fn a_ticket_walk_books_the_upstream_half_under_pull_through() {
let r = rig(0, Duration::ZERO);
let door = crate::admission::DoorAdmission::pass_through();
let who = crate::admission::Who::Principal("ticket-walk-carol".into());
let t = door
.enter(&who, crate::admission::WorkKind::Download, 0)
.unwrap();
let got = t.walk(|| r.proxy.fetch(&serde_id())).unwrap().unwrap();
t.add_net_out(got.len() as u64);
drop(t);
let v = crate::admission::book().view(&who.key()).unwrap();
assert_eq!(v.by_work["pull_through"].actual.upstream_fetch, 11);
assert_eq!(v.by_work["download"].actual.net_out, 11);
assert_eq!(
v.window.bytes(),
22,
"upstream + served, both the consumer's"
);
}
/// § A failed upstream fetch becomes a job that says `failed` with the
/// reason and the next attempt, is retried, and then lands `done` — and
/// the client's next request is a local hit.
#[test]
fn a_failed_upstream_fetch_is_retried_with_its_reason_then_done() {
let dir = tempfile::tempdir().unwrap();
let r = rig(2, Duration::ZERO);
let q = queue_at(dir.path(), &r.routes, 2);
let pt = Arc::new(PullThrough::new(Arc::clone(&q), 1, Duration::ZERO));
let first = queued(&pt, crate::jobs::PRIORITY_FULL, || {
r.proxy.fetch(&serde_id())
});
let e = first.expect_err("a miss is queued, not served");
let answer = queued_of(&e).expect("FetchQueued").clone();
assert!(answer.retry_after_secs >= 1);
// Watch it: failed (retry_wait, with reason + next attempt) at least once.
let id = crate::jobs::JobId(answer.job.clone());
let mut saw_retry = None;
let deadline = std::time::Instant::now() + Duration::from_secs(20);
loop {
let v = q.get(&id).unwrap().unwrap();
if v.status == status::RETRY_WAIT && saw_retry.is_none() {
saw_retry = Some(v.clone());
}
if v.status == status::DONE {
assert_eq!(v.attempts, 3, "two failures, then the third attempt landed");
break;
}
assert!(
std::time::Instant::now() < deadline,
"the job never finished: {v:?}"
);
std::thread::sleep(Duration::from_millis(5));
}
let w = saw_retry.expect("a failed attempt was visible as retry_wait");
assert_eq!(w.stage, "failed");
assert!(
w.reason.contains("refused the connection"),
"reason: {}",
w.reason
);
assert!(w.next_attempt_micros.is_some(), "the next attempt is named");
assert_eq!(r.upstream.calls.load(Ordering::SeqCst), 3);
// The client's retry: a local (primary) hit, no new job, no new upstream call.
let again = queued(&pt, crate::jobs::PRIORITY_FULL, || {
r.proxy.fetch(&serde_id())
})
.unwrap();
assert_eq!(again.as_deref(), Some(b"crate-bytes".as_ref()));
assert_eq!(r.upstream.calls.load(Ordering::SeqCst), 3);
assert!(
r.cache.fetch(&serde_id()).unwrap().is_some(),
"written through into the cache"
);
q.shutdown();
}
/// § Concurrent requests for ONE artifact are ONE job and ONE upstream fetch.
#[test]
fn concurrent_requests_for_one_artifact_are_one_job() {
let dir = tempfile::tempdir().unwrap();
let r = rig(0, Duration::from_millis(200));
let q = queue_at(dir.path(), &r.routes, 4);
let pt = Arc::new(PullThrough::new(Arc::clone(&q), 1, Duration::ZERO));
let ids: Vec<String> = std::thread::scope(|s| {
let hs: Vec<_> = (0..16)
.map(|_| {
let (pt, proxy) = (Arc::clone(&pt), Arc::clone(&r.proxy));
// A32-14, stays a task: scoped threads, joined when the scope ends.
s.spawn(move || {
let e =
queued(&pt, crate::jobs::PRIORITY_FULL, || proxy.fetch(&serde_id()))
.unwrap_err();
queued_of(&e).unwrap().job.clone()
})
})
.collect();
hs.into_iter().map(|h| h.join().unwrap()).collect()
});
assert!(
ids.iter().all(|i| i == &ids[0]),
"16 requests, one job: {ids:?}"
);
q.await_completion(&crate::jobs::JobId(ids[0].clone()), Duration::from_secs(10))
.unwrap()
.unwrap();
assert_eq!(
r.upstream.calls.load(Ordering::SeqCst),
1,
"one upstream fetch"
);
let fetches = q
.list(0)
.unwrap()
.into_iter()
.filter(|v| v.kind == crate::jobs::kind::FETCH)
.count();
assert_eq!(fetches, 1, "one row on the ledger");
q.shutdown();
}
/// § A queued fetch survives a restart: the next process resumes it.
#[test]
fn a_queued_fetch_resumes_after_a_restart() {
let dir = tempfile::tempdir().unwrap();
let r = rig(0, Duration::ZERO);
let job = {
// No runner: exactly a process that queued and died.
let (q, _) = JobQueue::open(dir.path(), 2).unwrap();
let pt = Arc::new(PullThrough::new(Arc::new(q), 1, Duration::ZERO));
let e = queued(&pt, crate::jobs::PRIORITY_REDUCED, || {
r.proxy.fetch(&serde_id())
})
.unwrap_err();
queued_of(&e).unwrap().job.clone()
};
let (q, summary) = JobQueue::open(dir.path(), 2).unwrap();
assert_eq!(summary.resumed, 1, "the fetch was resumed, not interrupted");
q.start(Arc::new(ServerJobRunner::new(Arc::clone(&r.routes), None)));
let v = q
.await_completion(&crate::jobs::JobId(job), Duration::from_secs(10))
.unwrap()
.unwrap();
assert_eq!(v.status, status::DONE);
assert_eq!(
v.priority,
crate::jobs::PRIORITY_REDUCED,
"it kept its class's place"
);
assert!(r.cache.fetch(&serde_id()).unwrap().is_some());
q.shutdown();
}
/// § A non-Full class never holds the connection: it is answered at once
/// even when a Full request would have waited.
#[test]
fn a_non_full_class_is_answered_at_once() {
let dir = tempfile::tempdir().unwrap();
let r = rig(0, Duration::from_millis(1500));
let q = queue_at(dir.path(), &r.routes, 2);
let pt = Arc::new(PullThrough::new(Arc::clone(&q), 1, Duration::from_secs(5)));
let t0 = std::time::Instant::now();
let e = queued(&pt, crate::jobs::PRIORITY_FLOOR, || {
r.proxy.fetch(&serde_id())
})
.unwrap_err();
assert!(queued_of(&e).is_some());
assert!(
t0.elapsed() < Duration::from_millis(500),
"Floor waited {:?}",
t0.elapsed()
);
// A Full request for it waits (bounded) and is served inside the wait.
let got = queued(&pt, crate::jobs::PRIORITY_FULL, || {
r.proxy.fetch(&serde_id())
})
.unwrap();
assert_eq!(got.as_deref(), Some(b"crate-bytes".as_ref()));
q.shutdown();
}
/// § No queue installed ⇒ the inline walk, exactly as before.
#[test]
fn without_the_queue_the_walk_is_inline() {
let r = rig(0, Duration::ZERO);
let got = r.proxy.fetch(&serde_id()).unwrap();
assert_eq!(got.as_deref(), Some(b"crate-bytes".as_ref()));
assert_eq!(r.upstream.calls.load(Ordering::SeqCst), 1);
}
}