perf(runtime): bound telemetry and hot-path work

Cap and fairly distribute live snapshots, serialize dashboard refreshes, reduce telemetry fanout copies, and use a faster codec registry map.

Cargo.lock

@@ -947,6 +947,7 @@ dependencies = [
 name = "data-plane"
 version = "0.1.0"
 dependencies = [
+ "criterion",
  "futures-lite",
  "libc",
  "parking_lot",
@@ -4348,6 +4349,7 @@ dependencies = [
  "ed25519-dalek 2.2.0",
  "proptest",
  "rand_core 0.6.4",
+ "rustc-hash",
  "serde",
  "serde_json",
  "swactor",

crates/bindings/python/src/job.rs

@@ -9,11 +9,13 @@ use std::time::Duration;
 use data_plane::blob::{Blob, BlobView, ContentDigest, WritableArenaView};
 use data_plane::bootstrap as dp_bootstrap;
 use data_plane::data_plane::{
-    BlobWriter, DataPlane, DataPlaneBootstrap, StreamReader, StreamWriter, parse_actor_address,
+    BlobWriter, DataPlane, DataPlaneBootstrap, Descriptor, DescriptorMapping, MapRequest,
+    MapTarget, Protection, Sharing, StreamReader, StreamWriter, parse_actor_address,
 };
 use data_plane::path::DataPath;
 use data_plane::protocol::{
-    BlobFailure, DataPlaneError, JobCapability, register_data_plane_codecs,
+    AccessMode, BlobAllocation, BlobFailure, DataPlaneError, DescriptorKind, Errno, JobCapability,
+    OpenOptions, register_data_plane_codecs,
 };
 use distribution::node::DistributedNodeConfig;
 use distribution::transport_bridge::{
@@ -23,7 +25,8 @@ use futures_lite::future;
 use iroh::{EndpointAddr, RelayMode};
 use iroh_driver::{IrohDriver, IrohDriverConfig};
 use parking_lot::Mutex as ParkingMutex;
-use pyo3::exceptions::{PyBufferError, PyPermissionError, PyRuntimeError};
+use pyo3::buffer::PyBuffer;
+use pyo3::exceptions::{PyBufferError, PyOSError, PyPermissionError, PyRuntimeError, PyValueError};
 use pyo3::ffi;
 use pyo3::prelude::*;
 use pyo3::types::{PyAny, PyBytes, PyModule};
@@ -36,6 +39,44 @@ use swactor_transport::{CodecRegistry, CodecRemoteSink, TransportRouter};
 const ROUTE_POLL: Duration = Duration::from_millis(5);
 const ROUTE_DEADLINE: Duration = Duration::from_secs(5);
 
+const O_RDONLY: i32 = 0;
+const O_WRONLY: i32 = 1;
+const O_RDWR: i32 = 2;
+const O_ACCMODE: i32 = 3;
+const O_CREAT: i32 = 0o100;
+const O_EXCL: i32 = 0o200;
+const O_TRUNC: i32 = 0o1000;
+const O_NONBLOCK: i32 = 0o4000;
+const KNOWN_OPEN_FLAGS: i32 = O_ACCMODE | O_CREAT | O_EXCL | O_TRUNC | O_NONBLOCK;
+
+fn python_open_options(flags: i32, length: Option<u64>) -> PyResult<OpenOptions> {
+    if flags & !KNOWN_OPEN_FLAGS != 0 {
+        return Err(PyValueError::new_err(format!(
+            "unsupported descriptor open flags: {:#x}",
+            flags & !KNOWN_OPEN_FLAGS
+        )));
+    }
+    let access = match flags & O_ACCMODE {
+        O_RDONLY => AccessMode::ReadOnly,
+        O_WRONLY => AccessMode::WriteOnly,
+        O_RDWR => AccessMode::ReadWrite,
+        _ => return Err(PyValueError::new_err("invalid descriptor access mode")),
+    };
+    let options = OpenOptions {
+        access,
+        create: flags & O_CREAT != 0,
+        exclusive: flags & O_EXCL != 0,
+        truncate: flags & O_TRUNC != 0,
+        nonblocking: flags & O_NONBLOCK != 0,
+        allocation: length.map(|length| BlobAllocation {
+            length,
+            digest: None,
+        }),
+    };
+    options.validate().map_err(raw_data_plane_error)?;
+    Ok(options)
+}
+
 pyo3::create_exception!(swactor, SwactorError, pyo3::exceptions::PyException);
 pyo3::create_exception!(swactor, BootstrapError, SwactorError);
 pyo3::create_exception!(swactor, DataPathError, SwactorError);
@@ -50,8 +91,8 @@ fn bootstrap_error(message: impl Into<String>) -> PyErr {
 fn data_plane_error(error: DataPlaneError) -> PyErr {
     match error {
         DataPlaneError::InvalidPath(reason) => PyErr::new::<DataPathError, _>(reason),
-        DataPlaneError::Unauthorized { path, operation } => {
-            PyPermissionError::new_err(format!("{operation:?} is not authorized for {path}"))
+        DataPlaneError::Unauthorized { path, access } => {
+            PyPermissionError::new_err(format!("{access:?} is not authorized for {path}"))
         }
         DataPlaneError::PathNotFound(path) => {
             PyErr::new::<DataPathError, _>(format!("data path not found: {path}"))
@@ -69,6 +110,29 @@ fn data_plane_error(error: DataPlaneError) -> PyErr {
     }
 }
 
+fn raw_data_plane_error(error: DataPlaneError) -> PyErr {
+    let code = match error.errno() {
+        Errno::Eacces => libc::EACCES,
+        Errno::Eagain => libc::EAGAIN,
+        Errno::Ebadf => libc::EBADF,
+        Errno::Ebusy => libc::EBUSY,
+        Errno::Ecanceled => libc::ECANCELED,
+        Errno::Econnreset => libc::ECONNRESET,
+        Errno::Eexist => libc::EEXIST,
+        Errno::Einval => libc::EINVAL,
+        Errno::Eio => libc::EIO,
+        Errno::Enodev => libc::ENODEV,
+        Errno::Enoent => libc::ENOENT,
+        Errno::Enomem => libc::ENOMEM,
+        Errno::Enospc => libc::ENOSPC,
+        Errno::Enotsup => libc::ENOTSUP,
+        Errno::Enxio => libc::ENXIO,
+        Errno::Epipe => libc::EPIPE,
+        Errno::Estale => libc::ESTALE,
+    };
+    PyOSError::new_err((code, error.to_string()))
+}
+
 fn bootstrap_env(name: &str) -> PyResult<String> {
     std::env::var(name)
         .map_err(|_| bootstrap_error(format!("bootstrap environment variable {name} is not set")))
@@ -239,6 +303,39 @@ pub struct PyDataPlane {
 
 #[pymethods]
 impl PyDataPlane {
+    #[pyo3(signature = (path, flags, *, length = None))]
+    fn open<'py>(
+        &self,
+        py: Python<'py>,
+        path: String,
+        flags: i32,
+        length: Option<u64>,
+    ) -> PyResult<Bound<'py, PyAny>> {
+        let path = DataPath::parse(path).map_err(|error| {
+            raw_data_plane_error(DataPlaneError::InvalidPath(error.to_string()))
+        })?;
+        let options = python_open_options(flags, length)?;
+        let data_plane = self.inner.clone();
+        pyo3_async_runtimes::tokio::future_into_py(py, async move {
+            let descriptor = data_plane
+                .open(&path, options)
+                .await
+                .map_err(raw_data_plane_error)?;
+            let kind = descriptor.kind();
+            let capabilities = descriptor.capabilities().bits();
+            Python::with_gil(|py| {
+                Py::new(
+                    py,
+                    PyDescriptor {
+                        descriptor: Arc::new(tokio::sync::Mutex::new(descriptor)),
+                        kind,
+                        capabilities,
+                    },
+                )
+            })
+        })
+    }
+
     fn read_blob<'py>(&self, py: Python<'py>, path: String) -> PyResult<Bound<'py, PyAny>> {
         let data_plane = self.inner.clone();
         pyo3_async_runtimes::tokio::future_into_py(py, async move {
@@ -294,6 +391,258 @@ impl PyDataPlane {
     }
 }
 
+fn byte_buffer(object: &Bound<'_, PyAny>, writable: bool) -> PyResult<PyBuffer<u8>> {
+    let buffer = PyBuffer::<u8>::get(object)?;
+    if writable && buffer.readonly() {
+        return Err(PyBufferError::new_err("buffer is read-only"));
+    }
+    if !buffer.is_c_contiguous() {
+        return Err(PyBufferError::new_err("buffer is not C-contiguous"));
+    }
+    Ok(buffer)
+}
+
+unsafe fn mutable_buffer_bytes<'a>(buffer: &'a PyBuffer<u8>) -> &'a mut [u8] {
+    // SAFETY: `byte_buffer` checked writability and contiguity, and the
+    // retained `PyBuffer` keeps the exporter and pointer valid.
+    unsafe { std::slice::from_raw_parts_mut(buffer.buf_ptr().cast(), buffer.len_bytes()) }
+}
+
+unsafe fn buffer_bytes<'a>(buffer: &'a PyBuffer<u8>) -> &'a [u8] {
+    // SAFETY: `byte_buffer` checked contiguity and retains the exporter.
+    unsafe { std::slice::from_raw_parts(buffer.buf_ptr().cast_const().cast(), buffer.len_bytes()) }
+}
+
+#[pyclass(name = "Descriptor")]
+pub struct PyDescriptor {
+    descriptor: Arc<tokio::sync::Mutex<Descriptor>>,
+    kind: DescriptorKind,
+    capabilities: u16,
+}
+
+#[pymethods]
+impl PyDescriptor {
+    #[getter]
+    fn kind(&self) -> &'static str {
+        match self.kind {
+            DescriptorKind::Blob => "blob",
+            DescriptorKind::Stream => "stream",
+        }
+    }
+
+    #[getter]
+    fn capabilities(&self) -> u16 {
+        self.capabilities
+    }
+
+    fn read<'py>(&self, py: Python<'py>, size: usize) -> PyResult<Bound<'py, PyAny>> {
+        let descriptor = self.descriptor.clone();
+        pyo3_async_runtimes::tokio::future_into_py(py, async move {
+            let mut bytes = vec![0_u8; size];
+            let count = descriptor
+                .lock()
+                .await
+                .read(&mut bytes)
+                .await
+                .map_err(raw_data_plane_error)?;
+            bytes.truncate(count);
+            Python::with_gil(|py| Ok(PyBytes::new(py, &bytes).unbind()))
+        })
+    }
+
+    fn readinto<'py>(
+        &self,
+        py: Python<'py>,
+        buffer: &Bound<'_, PyAny>,
+    ) -> PyResult<Bound<'py, PyAny>> {
+        let buffer = byte_buffer(buffer, true)?;
+        let descriptor = self.descriptor.clone();
+        pyo3_async_runtimes::tokio::future_into_py(py, async move {
+            // SAFETY: the owned `PyBuffer` remains alive through completion or
+            // cancellation and no pointer is retained by the Rust primitive.
+            let bytes = unsafe { mutable_buffer_bytes(&buffer) };
+            descriptor
+                .lock()
+                .await
+                .read(bytes)
+                .await
+                .map_err(raw_data_plane_error)
+        })
+    }
+
+    fn write<'py>(&self, py: Python<'py>, bytes: Vec<u8>) -> PyResult<Bound<'py, PyAny>> {
+        let descriptor = self.descriptor.clone();
+        pyo3_async_runtimes::tokio::future_into_py(py, async move {
+            descriptor
+                .lock()
+                .await
+                .write(&bytes)
+                .await
+                .map_err(raw_data_plane_error)
+        })
+    }
+
+    fn writefrom<'py>(
+        &self,
+        py: Python<'py>,
+        buffer: &Bound<'_, PyAny>,
+    ) -> PyResult<Bound<'py, PyAny>> {
+        let buffer = byte_buffer(buffer, false)?;
+        let descriptor = self.descriptor.clone();
+        pyo3_async_runtimes::tokio::future_into_py(py, async move {
+            // SAFETY: the owned `PyBuffer` retains a contiguous exporter until
+            // the descriptor primitive completes or is cancelled.
+            let bytes = unsafe { buffer_bytes(&buffer) };
+            descriptor
+                .lock()
+                .await
+                .write(bytes)
+                .await
+                .map_err(raw_data_plane_error)
+        })
+    }
+
+    #[pyo3(signature = (*, offset = 0, length, writable = false))]
+    fn map<'py>(
+        &self,
+        py: Python<'py>,
+        offset: u64,
+        length: u64,
+        writable: bool,
+    ) -> PyResult<Bound<'py, PyAny>> {
+        let descriptor = self.descriptor.clone();
+        pyo3_async_runtimes::tokio::future_into_py(py, async move {
+            let mapping = descriptor
+                .lock()
+                .await
+                .map(MapRequest {
+                    protection: if writable {
+                        Protection::ReadWrite
+                    } else {
+                        Protection::Read
+                    },
+                    sharing: Sharing::Shared,
+                    target: MapTarget::Host,
+                    offset,
+                    length,
+                })
+                .map_err(raw_data_plane_error)?;
+            Python::with_gil(|py| {
+                Py::new(
+                    py,
+                    PyDescriptorMapping {
+                        inner: Some(mapping),
+                        exports: 0,
+                    },
+                )
+            })
+        })
+    }
+
+    fn close<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
+        let descriptor = self.descriptor.clone();
+        pyo3_async_runtimes::tokio::future_into_py(py, async move {
+            descriptor
+                .lock()
+                .await
+                .close()
+                .await
+                .map_err(raw_data_plane_error)
+        })
+    }
+
+    fn abort<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
+        let descriptor = self.descriptor.clone();
+        pyo3_async_runtimes::tokio::future_into_py(py, async move {
+            descriptor
+                .lock()
+                .await
+                .abort()
+                .await
+                .map_err(raw_data_plane_error)
+        })
+    }
+}
+
+#[pyclass(name = "DescriptorMapping")]
+pub struct PyDescriptorMapping {
+    inner: Option<DescriptorMapping>,
+    exports: usize,
+}
+
+#[pymethods]
+impl PyDescriptorMapping {
+    fn __enter__(slf: PyRef<'_, Self>) -> PyResult<PyRef<'_, Self>> {
+        if slf.inner.is_none() {
+            return Err(PyBufferError::new_err("descriptor mapping is closed"));
+        }
+        Ok(slf)
+    }
+
+    fn __exit__(
+        &mut self,
+        _exception_type: &Bound<'_, PyAny>,
+        _exception: &Bound<'_, PyAny>,
+        _traceback: &Bound<'_, PyAny>,
+    ) -> PyResult<bool> {
+        self.close()?;
+        Ok(false)
+    }
+
+    fn close(&mut self) -> PyResult<()> {
+        if self.exports != 0 {
+            return Err(PyBufferError::new_err(
+                "cannot close a descriptor mapping with active buffer exports",
+            ));
+        }
+        self.inner.take();
+        Ok(())
+    }
+
+    unsafe fn __getbuffer__(
+        slf: Bound<'_, Self>,
+        view: *mut ffi::Py_buffer,
+        flags: c_int,
+    ) -> PyResult<()> {
+        let (pointer, length, readonly) = {
+            let mut borrowed = slf.borrow_mut();
+            let inner = borrowed
+                .inner
+                .as_mut()
+                .ok_or_else(|| PyBufferError::new_err("descriptor mapping is closed"))?;
+            match inner {
+                DescriptorMapping::ReadOnly(mapping) => {
+                    (mapping.as_ptr().cast_mut(), mapping.len(), true)
+                }
+                DescriptorMapping::WritableReadOnly(mapping) => {
+                    (mapping.as_ptr(), mapping.len(), true)
+                }
+                DescriptorMapping::Writable(mapping) => (mapping.as_ptr(), mapping.len(), false),
+            }
+        };
+        // SAFETY: the mapping owns the stable arena lease and the Python
+        // buffer retains `slf` until release.
+        unsafe {
+            fill_buffer(
+                view,
+                flags,
+                pointer,
+                length,
+                readonly,
+                slf.clone().into_any(),
+            )
+        }?;
+        slf.borrow_mut().exports += 1;
+        Ok(())
+    }
+
+    unsafe fn __releasebuffer__(&mut self, view: *mut ffi::Py_buffer) {
+        self.exports = self.exports.saturating_sub(1);
+        // SAFETY: `fill_buffer` allocated the format string for this export.
+        unsafe { release_buffer_format(view) };
+    }
+}
+
 #[pyclass(name = "Blob")]
 pub struct PyBlob {
     inner: Blob,
@@ -651,6 +1000,26 @@ impl PyStreamReader {
             Python::with_gil(|py| Ok(bytes.map(|bytes| PyBytes::new(py, &bytes).unbind())))
         })
     }
+
+    fn readinto<'py>(
+        &self,
+        py: Python<'py>,
+        buffer: &Bound<'_, PyAny>,
+    ) -> PyResult<Bound<'py, PyAny>> {
+        let buffer = byte_buffer(buffer, true)?;
+        let reader = self.reader.clone();
+        pyo3_async_runtimes::tokio::future_into_py(py, async move {
+            // SAFETY: the owned `PyBuffer` retains the writable exporter for
+            // the complete async operation and cancellation path.
+            let bytes = unsafe { mutable_buffer_bytes(&buffer) };
+            reader
+                .lock()
+                .await
+                .read_into(bytes)
+                .await
+                .map_err(data_plane_error)
+        })
+    }
 }
 
 #[pyclass(name = "StreamWriter")]
@@ -1208,7 +1577,16 @@ pub fn register(module: &Bound<'_, PyModule>) -> PyResult<()> {
     module.add("BlobError", module.py().get_type::<BlobError>())?;
     module.add("SessionError", module.py().get_type::<SessionError>())?;
     module.add("StreamError", module.py().get_type::<StreamError>())?;
+    module.add("O_RDONLY", O_RDONLY)?;
+    module.add("O_WRONLY", O_WRONLY)?;
+    module.add("O_RDWR", O_RDWR)?;
+    module.add("O_CREAT", O_CREAT)?;
+    module.add("O_EXCL", O_EXCL)?;
+    module.add("O_TRUNC", O_TRUNC)?;
+    module.add("O_NONBLOCK", O_NONBLOCK)?;
     module.add_class::<PyDataPlane>()?;
+    module.add_class::<PyDescriptor>()?;
+    module.add_class::<PyDescriptorMapping>()?;
     module.add_class::<PyBlob>()?;
     module.add_class::<PyBlobView>()?;
     module.add_class::<PyStreamWriter>()?;

crates/bindings/python/tests/test_bootstrap.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import asyncio
 import ctypes
+import errno
 import json
 import os
 import struct
@@ -150,7 +151,8 @@ def test_run_attaches_before_main_and_maps_blob_buffer_directly(monkeypatch, hos
                 _PY_RELEASE_BUFFER(ctypes.byref(exported))
 
         public = {name for name in dir(ctx.data) if not name.startswith("_")}
-        assert {"read_blob", "write_blob", "read_stream", "write_stream"} <= public
+        assert {"open", "read_blob", "write_blob", "read_stream", "write_stream"} <= public
+        assert isinstance(swactor.O_RDONLY, int)
         assert not public & {
             "actor",
             "actor_id",
@@ -201,6 +203,54 @@ def test_write_blob_seals_cleanly_and_exception_aborts(monkeypatch, host):
 
     swactor.run(main)
 
+def test_raw_descriptor_blob_io_mapping_and_errno(monkeypatch, host):
+    install_host_env(monkeypatch, host)
+
+    async def main(ctx):
+        logical = "/runs/self/results/raw-python"
+        writer = await ctx.data.open(
+            logical,
+            swactor.O_WRONLY | swactor.O_CREAT | swactor.O_TRUNC,
+            length=8,
+        )
+        assert isinstance(writer, swactor.Descriptor)
+        assert writer.kind == "blob"
+        assert await writer.write(b"abc") == 3
+        assert await writer.writefrom(b"defgh") == 5
+        await writer.close()
+        with pytest.raises(OSError) as closed:
+            await writer.close()
+        assert closed.value.errno == errno.EBADF
+
+        reader = await ctx.data.open(logical, swactor.O_RDONLY)
+        destination = bytearray(b"\xa5" * 10)
+        assert await reader.readinto(memoryview(destination)[1:7]) == 6
+        assert destination == b"\xa5abcdef\xa5\xa5\xa5"
+        assert await reader.read(8) == b"gh"
+        assert await reader.read(8) == b""
+        with pytest.raises(OSError) as wrong_access:
+            await reader.write(b"x")
+        assert wrong_access.value.errno == errno.EBADF
+
+        mapping = await reader.map(length=8)
+        assert isinstance(mapping, swactor.DescriptorMapping)
+        exported = memoryview(mapping)
+        assert exported.readonly
+        await reader.close()
+        assert bytes(exported) == b"abcdefgh"
+        with pytest.raises(BufferError, match="active buffer exports"):
+            mapping.close()
+        exported.release()
+        mapping.close()
+
+        with pytest.raises(FileNotFoundError):
+            await ctx.data.open("/models/raw-missing", swactor.O_RDONLY)
+        with pytest.raises(OSError) as unsupported:
+            await ctx.data.open(logical, swactor.O_RDONLY | swactor.O_NONBLOCK)
+        assert unsupported.value.errno in {errno.ENOTSUP, errno.EOPNOTSUPP}
+
+    swactor.run(main)
+
 
 def test_missing_path_and_authorization_are_typed(monkeypatch, host):
     install_host_env(monkeypatch, host)
@@ -217,14 +267,16 @@ def test_missing_path_and_authorization_are_typed(monkeypatch, host):
 def test_native_stream_round_trip_has_ordered_eof(monkeypatch, host):
     install_host_env(monkeypatch, host)
     received = []
+    raw_received = []
 
     async def main(ctx):
         async def receive():
             reader = await ctx.data.read_stream(
                 "/runs/self/results/predictions"
             )
-            while (chunk := await reader.read()) is not None:
-                received.append(chunk)
+            buffer = bytearray(3)
+            while (count := await reader.readinto(buffer)) != 0:
+                received.append(bytes(buffer[:count]))
 
         receiver = asyncio.create_task(receive())
         async with ctx.data.write_stream(
@@ -234,8 +286,20 @@ def test_native_stream_round_trip_has_ordered_eof(monkeypatch, host):
             await stream.write(b"result")
         await receiver
 
+        raw_reader, raw_writer = await asyncio.gather(
+            ctx.data.open("/runs/self/results/predictions", swactor.O_RDONLY),
+            ctx.data.open("/runs/self/results/predictions", swactor.O_WRONLY),
+        )
+        assert await raw_writer.write(b"raw-stream") == len(b"raw-stream")
+        await raw_writer.close()
+        buffer = bytearray(4)
+        while (count := await raw_reader.readinto(buffer)) != 0:
+            raw_received.append(bytes(buffer[:count]))
+        await raw_reader.close()
+
     swactor.run(main)
     assert b"".join(received) == b"native-result"
+    assert b"".join(raw_received) == b"raw-stream"
     assert "SWACTOR_DATA_PLANE_OUTPUT" not in host.env()
 
 

crates/dashboard/README.md

@@ -20,7 +20,7 @@ is hard-capped at 50.
 - `GET /api/frames` — recent raw frame window
 - `GET /api/views` — registered view metadata
 - `GET /view/telemetry/live` — generic live explorer over retained and incoming telemetry frames
-- `GET /api/view/telemetry/live` — 2,000 frames per stream/channel, newest lifetime per node, all live streams plus 50 stale streams
+- `GET /api/view/telemetry/live` — bounded bootstrap snapshot: up to 500 recent frames and 256 KiB of raw payload, distributed across channels; retention remains 500 frames per stream/channel for live inspection
 - `GET /view/fleet` — fused control-plane page (machine + actors per node)
 - `GET /api/view/fleet` — live/stale pools with per-node machine and roster snapshot
 - `GET /api/view/fleet/detail?stream=<node#life>&actor=<addr>` — bounded per-actor dossier detail (diet, history, sampled receipts)

crates/dashboard/src/control_plane_page.html

@@ -173,6 +173,7 @@ let rosterSort = { key: 'address', direction: 'asc' };
 let hardwareSource = 'node';
 let lastSnapshot = null;
 let detailTimer = null;
+let refreshInFlight = false;
 // A1: last-rendered HTML per region. A poll that yields identical markup
 // must not swap innerHTML — the swap destroyed hover/selection/presses and
 // restarted animations every second even in a fully converged steady state.
@@ -256,13 +257,18 @@ function pushUrl() {
 }
 
 async function refresh() {
-  let data;
+  if (refreshInFlight) return;
+  refreshInFlight = true;
   try {
     const response = await fetch('/api/view/fleet');
-    data = await response.json();
-  } catch { return; }
-  lastSnapshot = data;
-  render();
+    const data = await response.json();
+    lastSnapshot = data;
+    render();
+  } catch {
+    // Keep the last good snapshot while the next scheduled refresh retries.
+  } finally {
+    refreshInFlight = false;
+  }
 }
 
 function render() {
@@ -681,15 +687,20 @@ function stopDetailPolling() {
 
 function startDetailPolling() {
   stopDetailPolling();
+  let pollInFlight = false;
   const poll = async () => {
-    if (!selectedStream || !selectedActor) return;
-    let detail;
+    if (!selectedStream || !selectedActor || pollInFlight) return;
+    pollInFlight = true;
     try {
       const response = await fetch(`/api/view/fleet/detail?stream=${encodeURIComponent(selectedStream)}&actor=${encodeURIComponent(selectedActor)}`);
       if (!response.ok) { stopDetailPolling(); return; }
-      detail = await response.json();
-    } catch { return; }
-    renderDossier(detail);
+      const detail = await response.json();
+      renderDossier(detail);
+    } catch {
+      // Keep the current dossier while the next scheduled refresh retries.
+    } finally {
+      pollInFlight = false;
+    }
   };
   poll();
   detailTimer = setInterval(poll, POLL_MS);

crates/dashboard/src/live_explorer.rs

@@ -9,7 +9,9 @@ use crate::FrameEvent;
 use crate::view::DashboardView;
 
 const LIVE_EXPLORER_HTML: &str = include_str!("live_explorer_page.html");
-const FRAME_HISTORY_CAP: usize = 2_000;
+const FRAME_HISTORY_CAP: usize = 500;
+const SNAPSHOT_FRAME_CAP: usize = 500;
+const SNAPSHOT_PAYLOAD_BYTE_CAP: usize = 256 * 1024;
 const LIVE_TTL: Duration = Duration::from_secs(8);
 const STALE_STREAM_CAP: usize = 50;
 
@@ -65,12 +67,39 @@ impl LiveTelemetryExplorer {
     fn snapshot_at(&self, now: Instant) -> Value {
         let mut state = self.state.write();
         prune_stale(&mut state.streams, now);
-        let mut frames = state
+
+        // Take recent frames round-robin across channels so a busy channel
+        // cannot crowd quiet channels out of the bounded bootstrap snapshot.
+        let mut channels = state
             .streams
             .values()
             .flat_map(|stream| stream.channels.values())
-            .flat_map(|frames| frames.iter())
+            .map(|frames| frames.iter().rev())
             .collect::<Vec<_>>();
+        let mut frames = Vec::with_capacity(SNAPSHOT_FRAME_CAP.min(channels.len()));
+        let mut payload_bytes = 0;
+        'snapshot: loop {
+            let mut found_frame = false;
+            for channel in &mut channels {
+                let Some(frame) = channel.next() else {
+                    continue;
+                };
+                found_frame = true;
+                if frame.payload.len() > SNAPSHOT_PAYLOAD_BYTE_CAP - payload_bytes {
+                    continue;
+                }
+                payload_bytes += frame.payload.len();
+                frames.push(frame.clone());
+                if frames.len() == SNAPSHOT_FRAME_CAP {
+                    break 'snapshot;
+                }
+            }
+            if !found_frame || payload_bytes == SNAPSHOT_PAYLOAD_BYTE_CAP {
+                break;
+            }
+        }
+        drop(state);
+
         frames.sort_by(|left, right| {
             left.stream
                 .node
@@ -138,6 +167,16 @@ mod tests {
         for position in 0..=FRAME_HISTORY_CAP as u64 {
             ingest(&view, &stream, "busy", position);
         }
+        {
+            let state = view.state.read();
+            let busy = &state.streams.values().next().expect("stream").channels["busy"];
+            assert_eq!(busy.len(), FRAME_HISTORY_CAP);
+            assert_eq!(busy.front().map(|frame| frame.position), Some(1));
+            assert_eq!(
+                busy.back().map(|frame| frame.position),
+                Some(FRAME_HISTORY_CAP as u64)
+            );
+        }
 
         let snapshot = view.snapshot_json();
         let frames = snapshot["frames"].as_array().expect("frames array");
@@ -151,11 +190,7 @@ mod tests {
             .collect::<Vec<_>>();
 
         assert_eq!(quiet.len(), 1, "a quiet channel remains discoverable");
-        assert_eq!(busy.len(), FRAME_HISTORY_CAP);
-        assert_eq!(
-            busy.first().and_then(|frame| frame["position"].as_u64()),
-            Some(1)
-        );
+        assert_eq!(busy.len(), SNAPSHOT_FRAME_CAP - quiet.len());
         assert_eq!(
             busy.last().and_then(|frame| frame["position"].as_u64()),
             Some(FRAME_HISTORY_CAP as u64)
@@ -201,6 +236,37 @@ mod tests {
         assert!(streams.contains("node-59"));
     }
 
+    #[test]
+    fn snapshot_payload_is_bounded() {
+        let view = LiveTelemetryExplorer::default();
+        let stream = StreamId::new(NodeId::new("node-a"), Lifetime(1));
+        let payload = vec![7; SNAPSHOT_PAYLOAD_BYTE_CAP / 2];
+        for (position, channel) in ["first", "second", "third"].into_iter().enumerate() {
+            let event = FrameEvent {
+                stream: crate::StreamEvent {
+                    node: stream.node.as_str().to_owned(),
+                    life: stream.life.0,
+                    origin: None,
+                    label: None,
+                },
+                channel: channel.to_owned(),
+                position: position as u64,
+                payload: payload.clone(),
+            };
+            view.ingest_at(&event, Instant::now());
+        }
+
+        let snapshot = view.snapshot_json();
+        let frames = snapshot["frames"].as_array().expect("frames array");
+        let payload_bytes = frames
+            .iter()
+            .map(|frame| frame["payload"].as_array().expect("payload").len())
+            .sum::<usize>();
+        assert!(frames.len() <= SNAPSHOT_FRAME_CAP);
+        assert!(payload_bytes <= SNAPSHOT_PAYLOAD_BYTE_CAP);
+        assert_eq!(frames.len(), 2, "the payload byte cap bounds the snapshot");
+    }
+
     fn test_frame(position: u64) -> Frame {
         Frame::new(
             ChannelId(position as u32 + 1),

crates/dashboard/src/live_explorer_page.html

@@ -147,6 +147,8 @@ RISK: terminal cosplay if decoration creeps past data; held by the palette law.
     seq: 0,
     paintQueued: false,
     paintHandle: 0,
+    reconciling: false,
+    reconcilePending: false,
   };
 
   const dirty = {
@@ -523,6 +525,11 @@ RISK: terminal cosplay if decoration creeps past data; held by the palette law.
       state.messageVersion++;
       updateMetrics();
     }
+    if (state.reconciling) {
+      state.reconcilePending = true;
+      return;
+    }
+    state.reconciling = true;
     const messageVersion = state.messageVersion;
     const startedAt = performance.now();
     try {
@@ -554,6 +561,12 @@ RISK: terminal cosplay if decoration creeps past data; held by the palette law.
         updateMetrics();
         renderDetail();
       }
+    } finally {
+      state.reconciling = false;
+      if (state.reconcilePending) {
+        state.reconcilePending = false;
+        reconcileSnapshot({ notice: 'Catching up after concurrent telemetry updates.' });
+      }
     }
   }
 

crates/data-plane/Cargo.toml

@@ -18,9 +18,15 @@ parking_lot = "0.12"
 
 [dev-dependencies]
 parking_lot = "0.12"
+criterion = { version = "0.5", default-features = false }
 futures-lite = "2"
 proptest = "1"
 tokio.workspace = true
+
+[[bench]]
+name = "descriptor_performance"
+harness = false
+
 [target.'cfg(target_os = "linux")'.dependencies]
 libc = "0.2"
 

crates/data-plane/benches/descriptor_performance.rs

@@ -0,0 +1,81 @@
+#![cfg(target_os = "linux")]
+
+use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main};
+use data_plane::arena::{ArenaConfig, ArenaManager, NodeId};
+use data_plane::byte_ring::{ByteRingSpec, Endpoint, RecordKind, Role, attach, install};
+
+const PAYLOAD_LEN: usize = 64 * 1024;
+
+fn endpoints() -> (ArenaManager, Endpoint, Endpoint) {
+    let mut arena = ArenaManager::boot(ArenaConfig {
+        node_id: NodeId(91),
+        reservation_ceiling: 2 << 20,
+        base_alignment: 64,
+    })
+    .expect("benchmark arena");
+    let handle = install(
+        &mut arena,
+        ByteRingSpec {
+            capacity: (PAYLOAD_LEN + 5) as u64,
+            generation: 1,
+            alignment: 64,
+            request_id: 1,
+        },
+    )
+    .expect("benchmark ring");
+    let producer = attach(&arena, handle, Role::Producer).expect("benchmark producer");
+    let consumer = attach(&arena, handle, Role::Consumer).expect("benchmark consumer");
+    (arena, producer, consumer)
+}
+
+fn descriptor_stream_throughput(criterion: &mut Criterion) {
+    let payload = vec![0x5a_u8; PAYLOAD_LEN];
+    let mut destination = vec![0_u8; PAYLOAD_LEN];
+    let (_arena, mut producer, mut consumer) = endpoints();
+    let mut group = criterion.benchmark_group("descriptor_stream_read");
+    group.throughput(Throughput::Bytes(PAYLOAD_LEN as u64));
+
+    group.bench_with_input(
+        BenchmarkId::new("partial_allocation_free", PAYLOAD_LEN),
+        &PAYLOAD_LEN,
+        |bencher, _| {
+            bencher.iter(|| {
+                producer
+                    .send_record(RecordKind::Data, &payload)
+                    .expect("publish payload");
+                let cursor = consumer
+                    .record_cursor()
+                    .expect("read cursor")
+                    .expect("committed record");
+                let count = consumer
+                    .copy_record_range(cursor, 0, &mut destination)
+                    .expect("copy payload");
+                consumer
+                    .release_record_cursor(cursor)
+                    .expect("release payload");
+                criterion::black_box(count)
+            });
+        },
+    );
+
+    group.bench_with_input(
+        BenchmarkId::new("legacy_allocating_record", PAYLOAD_LEN),
+        &PAYLOAD_LEN,
+        |bencher, _| {
+            bencher.iter(|| {
+                producer
+                    .send_record(RecordKind::Data, &payload)
+                    .expect("publish payload");
+                let (_, bytes) = consumer
+                    .recv_record()
+                    .expect("receive payload")
+                    .expect("committed record");
+                criterion::black_box(bytes)
+            });
+        },
+    );
+    group.finish();
+}
+
+criterion_group!(benches, descriptor_stream_throughput);
+criterion_main!(benches);

crates/data-plane/src/blob.rs

@@ -7,6 +7,8 @@ use std::ptr::NonNull;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
 
+use parking_lot::Mutex;
+
 use serde::{Deserialize, Serialize};
 use sha2::{Digest as _, Sha256};
 
@@ -251,27 +253,60 @@ impl Blob {
         self.guard.lease
     }
 
-    pub fn map(&self) -> Result<BlobView, BlobError> {
+    pub(crate) fn copy_at(&self, offset: u64, destination: &mut [u8]) -> Result<usize, BlobError> {
+        if offset >= self.metadata.length || destination.is_empty() {
+            return Ok(0);
+        }
+        let count = destination
+            .len()
+            .min(usize::try_from(self.metadata.length - offset).unwrap_or(usize::MAX));
+        let view = self.map_range(offset, count as u64)?;
+        destination[..count].copy_from_slice(view.as_ref());
+        Ok(count)
+    }
+
+    pub fn map_range(&self, offset: u64, length: u64) -> Result<BlobView, BlobError> {
+        let end = offset
+            .checked_add(length)
+            .ok_or(BlobError::RangeOutOfBounds {
+                offset,
+                length,
+                arena_size: self.metadata.length,
+            })?;
+        if end > self.metadata.length {
+            return Err(BlobError::RangeOutOfBounds {
+                offset,
+                length,
+                arena_size: self.metadata.length,
+            });
+        }
         validate_mapped_header(
             &self.guard.arena,
             self.guard.lease,
             &self.metadata,
             BlobSharedState::Sealed,
         )?;
-        let payload_offset = self.guard.lease.offset.checked_add(BLOB_HEADER_LEN).ok_or(
-            BlobError::RangeOutOfBounds {
-                offset: self.guard.lease.offset,
-                length: self.guard.lease.length,
+        let payload_offset = self
+            .guard
+            .lease
+            .offset
+            .checked_add(BLOB_HEADER_LEN)
+            .and_then(|start| start.checked_add(offset))
+            .ok_or(BlobError::RangeOutOfBounds {
+                offset,
+                length,
                 arena_size: self.guard.arena.len() as u64,
-            },
-        )?;
-        let range = mapped_range(&self.guard.arena, payload_offset, self.guard.lease.length)?;
+            })?;
+        let range = mapped_range(&self.guard.arena, payload_offset, length)?;
         Ok(BlobView {
             guard: self.guard.clone(),
             payload_offset: range.start,
             length: range.len(),
         })
     }
+    pub fn map(&self) -> Result<BlobView, BlobError> {
+        self.map_range(0, self.metadata.length)
+    }
 }
 
 pub struct BlobView {
@@ -315,12 +350,17 @@ impl Deref for BlobView {
     }
 }
 
+pub(crate) trait WritableViewObserver: Send + Sync + 'static {
+    fn view_released(&self);
+}
+
 pub(crate) struct WritableBlobLease {
     arena: Arc<MappedArena>,
     lease: BlobLease,
     metadata: BlobMetadata,
     active_view: AtomicBool,
     finished: AtomicBool,
+    view_observer: Mutex<Option<Arc<dyn WritableViewObserver>>>,
 }
 
 impl WritableBlobLease {
@@ -339,6 +379,7 @@ impl WritableBlobLease {
             metadata,
             active_view: AtomicBool::new(false),
             finished: AtomicBool::new(false),
+            view_observer: Mutex::new(None),
         }))
     }
 
@@ -350,23 +391,111 @@ impl WritableBlobLease {
         &self.metadata
     }
 
-    pub(crate) fn map(self: &Arc<Self>) -> Result<WritableArenaView, BlobError> {
+    pub(crate) fn set_view_observer(&self, observer: Arc<dyn WritableViewObserver>) {
+        *self.view_observer.lock() = Some(observer);
+    }
+
+    pub(crate) fn has_active_view(&self) -> bool {
+        self.active_view.load(Ordering::Acquire)
+    }
+
+    pub(crate) fn copy_at(&self, offset: u64, destination: &mut [u8]) -> Result<usize, BlobError> {
+        if self.finished.load(Ordering::Acquire) {
+            return Err(BlobError::AlreadyFinished);
+        }
+        if self.active_view.load(Ordering::Acquire) {
+            return Err(BlobError::ActiveWritableView);
+        }
+        if offset >= self.metadata.length || destination.is_empty() {
+            return Ok(0);
+        }
+        let count = destination
+            .len()
+            .min(usize::try_from(self.metadata.length - offset).unwrap_or(usize::MAX));
+        let payload_offset = self.lease.offset + BLOB_HEADER_LEN + offset;
+        let range = mapped_range(&self.arena, payload_offset, count as u64)?;
+        // SAFETY: the mapped range was bounds-checked and no mutable view is active.
+        let source = unsafe {
+            std::slice::from_raw_parts(
+                self.arena.ptr_at(range.start).as_ptr().cast_const(),
+                range.len(),
+            )
+        };
+        destination[..count].copy_from_slice(source);
+        Ok(count)
+    }
+
+    pub(crate) fn copy_from(&self, offset: u64, source: &[u8]) -> Result<usize, BlobError> {
+        if self.finished.load(Ordering::Acquire) {
+            return Err(BlobError::AlreadyFinished);
+        }
+        if self.active_view.load(Ordering::Acquire) {
+            return Err(BlobError::ActiveWritableView);
+        }
+        let length = source.len() as u64;
+        let end = offset
+            .checked_add(length)
+            .ok_or(BlobError::RangeOutOfBounds {
+                offset,
+                length,
+                arena_size: self.metadata.length,
+            })?;
+        if end > self.metadata.length {
+            return Err(BlobError::RangeOutOfBounds {
+                offset,
+                length,
+                arena_size: self.metadata.length,
+            });
+        }
+        if source.is_empty() {
+            return Ok(0);
+        }
+        let payload_offset = self.lease.offset + BLOB_HEADER_LEN + offset;
+        let range = mapped_range(&self.arena, payload_offset, length)?;
+        // SAFETY: the mapped range was bounds-checked and no other mutable view is active.
+        let destination = unsafe {
+            std::slice::from_raw_parts_mut(self.arena.ptr_at(range.start).as_ptr(), range.len())
+        };
+        destination.copy_from_slice(source);
+        Ok(source.len())
+    }
+
+    pub(crate) fn map_range(
+        self: &Arc<Self>,
+        offset: u64,
+        length: u64,
+    ) -> Result<WritableArenaView, BlobError> {
+        let end = offset
+            .checked_add(length)
+            .ok_or(BlobError::RangeOutOfBounds {
+                offset,
+                length,
+                arena_size: self.metadata.length,
+            })?;
+        if end > self.metadata.length {
+            return Err(BlobError::RangeOutOfBounds {
+                offset,
+                length,
+                arena_size: self.metadata.length,
+            });
+        }
         if self.finished.load(Ordering::Acquire) {
             return Err(BlobError::AlreadyFinished);
         }
         self.active_view
             .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
             .map_err(|_| BlobError::ActiveWritableView)?;
-        let payload_offset =
-            self.lease
-                .offset
-                .checked_add(BLOB_HEADER_LEN)
-                .ok_or(BlobError::RangeOutOfBounds {
-                    offset: self.lease.offset,
-                    length: self.lease.length,
-                    arena_size: self.arena.len() as u64,
-                })?;
-        let range = match mapped_range(&self.arena, payload_offset, self.lease.length) {
+        let payload_offset = self
+            .lease
+            .offset
+            .checked_add(BLOB_HEADER_LEN)
+            .and_then(|start| start.checked_add(offset))
+            .ok_or(BlobError::RangeOutOfBounds {
+                offset,
+                length,
+                arena_size: self.arena.len() as u64,
+            })?;
+        let range = match mapped_range(&self.arena, payload_offset, length) {
             Ok(range) => range,
             Err(error) => {
                 self.active_view.store(false, Ordering::Release);
@@ -379,6 +508,9 @@ impl WritableBlobLease {
             length: range.len(),
         })
     }
+    pub(crate) fn map(self: &Arc<Self>) -> Result<WritableArenaView, BlobError> {
+        self.map_range(0, self.metadata.length)
+    }
 
     pub(crate) fn seal(&self) -> Result<BlobMetadata, BlobError> {
         if self.active_view.load(Ordering::Acquire) {
@@ -483,6 +615,10 @@ impl DerefMut for WritableArenaView {
 impl Drop for WritableArenaView {
     fn drop(&mut self) {
         self.owner.active_view.store(false, Ordering::Release);
+        let observer = self.owner.view_observer.lock().clone();
+        if let Some(observer) = observer {
+            observer.view_released();
+        }
     }
 }
 

crates/data-plane/src/byte_ring.rs

@@ -69,6 +69,7 @@ pub const OFF_CAPACITY: u64 = 8;
 pub const OFF_GENERATION: u64 = 16;
 pub const OFF_COMMIT: u64 = 24;
 pub const OFF_CONSUME: u64 = 32;
+pub const OFF_TERMINAL: u64 = 40;
 
 const ZERO_CHUNK: usize = 4 * 1024;
 
@@ -128,6 +129,30 @@ pub struct RecordMeta {
     pub len: u64,
 }
 
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub struct RecordCursor {
+    kind: RecordKind,
+    payload_start: u64,
+    payload_len: u64,
+    record_start: u64,
+    record_len: u64,
+    generation: u64,
+}
+
+impl RecordCursor {
+    pub const fn kind(self) -> RecordKind {
+        self.kind
+    }
+
+    pub fn len(self) -> usize {
+        self.payload_len as usize
+    }
+
+    pub const fn is_empty(self) -> bool {
+        self.payload_len == 0
+    }
+}
+
 /// Borrowed committed record. Dropping the view releases the complete framed
 /// record back to the producer.
 pub struct PinnedRecord<'a> {
@@ -557,6 +582,15 @@ pub fn attach_mapped(
     Ok(endpoint)
 }
 
+pub fn mark_peer_terminated_mapped(
+    arena: &crate::mapped_arena::MappedArena,
+    handle: RingHandle,
+) -> Result<(), AttachError> {
+    let endpoint = attach_mapped(arena, handle, Role::Producer)?;
+    endpoint.atomic(OFF_TERMINAL).store(1, Ordering::Release);
+    Ok(())
+}
+
 fn lease_ring(
     arena: &mut ArenaManager,
     request_id: u64,
@@ -572,6 +606,7 @@ fn lease_ring(
         (1, Some(ArenaEvent::RingLeaseRejected { reason, .. })) => {
             Err(InstallError::LeaseRejected(reason))
         }
+
         (1, Some(ArenaEvent::RingLeaseQueued { .. })) => Err(InstallError::LeaseQueued),
         _ => Err(InstallError::UnexpectedLeaseOutcome),
     }
@@ -586,6 +621,13 @@ impl Endpoint {
         self.role
     }
 
+    pub fn peer_terminated(&self) -> Result<bool, FlowError> {
+        self.validate_fixed().map_err(FlowError::Corrupt)?;
+        self.validate_generation(self.info.generation)
+            .map_err(FlowError::Corrupt)?;
+        Ok(self.atomic(OFF_TERMINAL).load(Ordering::Acquire) != 0)
+    }
+
     /// Current published producer and consumer positions. This role-neutral
     /// observation is used only to wait for an already-committed clean close
     /// to enter the downstream bounded transport.
@@ -764,6 +806,25 @@ impl Endpoint {
         }
     }
 
+    fn copy_into_slice(&self, stream_pos: u64, destination: &mut [u8]) {
+        let capacity = self.info.capacity as usize;
+        let start = (stream_pos % self.info.capacity) as usize;
+        let first = destination.len().min(capacity - start);
+        // SAFETY: the caller validated the source range against committed bytes.
+        unsafe {
+            std::ptr::copy_nonoverlapping(
+                self.data_ptr().add(start),
+                destination.as_mut_ptr(),
+                first,
+            );
+            std::ptr::copy_nonoverlapping(
+                self.data_ptr(),
+                destination.as_mut_ptr().add(first),
+                destination.len() - first,
+            );
+        }
+    }
+
     fn copy_out(&self, stream_pos: u64, len: u64) -> Vec<u8> {
         let capacity = self.info.capacity as usize;
         let start = (stream_pos % self.info.capacity) as usize;
@@ -805,6 +866,14 @@ impl Endpoint {
         })
     }
 
+    pub fn writable_payload_capacity(&self) -> Result<u64, FlowError> {
+        self.check("writable_payload_capacity", Role::Producer)?;
+        self.validate_generation(self.info.generation)
+            .map_err(FlowError::Corrupt)?;
+        let used = self.commit_cursor() - self.consume_cursor();
+        Ok((self.info.capacity - used).saturating_sub(RECORD_HEADER_LEN))
+    }
+
     /// Copy bytes into the reserved (producer-side) span.
     pub fn write(&self, reservation: &Reservation, bytes: &[u8]) -> Result<(), FlowError> {
         self.check("write", Role::Producer)?;
@@ -996,6 +1065,97 @@ impl Endpoint {
         }))
     }
 
+    /// Acquire an owned cursor for the next complete record. The cursor does
+    /// not release capacity and remains valid only while that record is first.
+    pub fn record_cursor(&self) -> Result<Option<RecordCursor>, FlowError> {
+        self.check("record_cursor", Role::Consumer)?;
+        self.validate_generation(self.info.generation)
+            .map_err(FlowError::Corrupt)?;
+        let (commit, consume) = (self.commit_cursor(), self.consume_cursor());
+        let readable = commit - consume;
+        if readable < RECORD_HEADER_LEN {
+            return Ok(None);
+        }
+        let mut prefix = [0_u8; RECORD_HEADER_LEN as usize];
+        self.copy_into_slice(consume, &mut prefix);
+        let kind = RecordKind::from_byte(prefix[0])
+            .ok_or(FlowError::BadRecord(RecordError::InvalidKind(prefix[0])))?;
+        let payload_len = u64::from(u32::from_le_bytes(prefix[1..5].try_into().unwrap()));
+        let record_len = RECORD_HEADER_LEN + payload_len;
+        if record_len > self.info.capacity {
+            return Err(FlowError::BadRecord(RecordError::LengthExceedsCapacity {
+                len: payload_len,
+                capacity: self.info.capacity,
+            }));
+        }
+        if readable < record_len {
+            return Ok(None);
+        }
+        Ok(Some(RecordCursor {
+            kind,
+            payload_start: consume + RECORD_HEADER_LEN,
+            payload_len,
+            record_start: consume,
+            record_len,
+            generation: self.info.generation,
+        }))
+    }
+
+    pub fn copy_record_range(
+        &self,
+        cursor: RecordCursor,
+        offset: u64,
+        destination: &mut [u8],
+    ) -> Result<usize, FlowError> {
+        self.check("copy_record_range", Role::Consumer)?;
+        let generation = self.field_u64(OFF_GENERATION);
+        if generation != cursor.generation {
+            return Err(FlowError::StaleReservation {
+                reservation: cursor.generation,
+                ring: generation,
+            });
+        }
+        let consume = self.consume_cursor();
+        if consume != cursor.record_start {
+            return Err(FlowError::PinnedRecordMoved {
+                expected: cursor.record_start,
+                found: consume,
+            });
+        }
+        if offset > cursor.payload_len {
+            return Err(FlowError::BeyondCommitted {
+                requested: offset,
+                readable: cursor.payload_len,
+            });
+        }
+        let count = destination
+            .len()
+            .min(usize::try_from(cursor.payload_len - offset).unwrap_or(usize::MAX));
+        self.copy_into_slice(cursor.payload_start + offset, &mut destination[..count]);
+        Ok(count)
+    }
+
+    pub fn release_record_cursor(&mut self, cursor: RecordCursor) -> Result<(), FlowError> {
+        self.check("release_record_cursor", Role::Consumer)?;
+        let generation = self.field_u64(OFF_GENERATION);
+        if generation != cursor.generation {
+            return Err(FlowError::StaleReservation {
+                reservation: cursor.generation,
+                ring: generation,
+            });
+        }
+        let consume = self.consume_cursor();
+        if consume != cursor.record_start {
+            return Err(FlowError::PinnedRecordMoved {
+                expected: cursor.record_start,
+                found: consume,
+            });
+        }
+        self.atomic(OFF_CONSUME)
+            .store(cursor.record_start + cursor.record_len, Ordering::Release);
+        Ok(())
+    }
+
     /// Receive one complete record; `Ok(None)` when nothing (or only a
     /// torn, uncommitted prefix) is readable.
     pub fn recv_record(&mut self) -> Result<Option<(RecordKind, Vec<u8>)>, FlowError> {

crates/data-plane/src/namespace.rs

@@ -39,6 +39,12 @@ pub enum EntryKind {
     Stream,
 }
 
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct NamespaceNode {
+    pub kind: EntryKind,
+    pub revision: u64,
+}
+
 #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub enum StreamRole {
     Source,
@@ -64,6 +70,7 @@ pub struct StreamMatch {
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub enum NamespaceError {
     PathNotFound(DataPath),
+    PathExists(DataPath),
     WrongEntryType {
         path: DataPath,
         expected: EntryKind,
@@ -89,6 +96,7 @@ impl fmt::Display for NamespaceError {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
             Self::PathNotFound(path) => write!(f, "data path not found: {path}"),
+            Self::PathExists(path) => write!(f, "data path already exists: {path}"),
             Self::WrongEntryType {
                 path,
                 expected,
@@ -140,6 +148,7 @@ pub enum DataDirectoryIn {
         length: u64,
         recovery: SourceRecovery,
         operation_id: OperationId,
+        reservation: Option<OperationId>,
         reply_to: ActorAddress,
     },
     Resolve {
@@ -147,6 +156,23 @@ pub enum DataDirectoryIn {
         path: DataPath,
         reply_to: ActorAddress,
     },
+    Lookup {
+        request_id: DirectoryRequestId,
+        path: DataPath,
+        reply_to: ActorAddress,
+    },
+    ReserveBlob {
+        request_id: DirectoryRequestId,
+        path: DataPath,
+        operation_id: OperationId,
+        reply_to: ActorAddress,
+    },
+    ReleaseBlobReservation {
+        request_id: DirectoryRequestId,
+        path: DataPath,
+        operation_id: OperationId,
+        reply_to: ActorAddress,
+    },
     Unregister {
         request_id: DirectoryRequestId,
         path: DataPath,
@@ -160,6 +186,8 @@ pub enum DataDirectoryIn {
         descriptor: Vec<u8>,
         endpoint: ActorAddress,
         replace: bool,
+        ensure: bool,
+        expected_revision: Option<u64>,
         operation_id: OperationId,
         reply_to: ActorAddress,
     },
@@ -193,6 +221,21 @@ pub enum DataDirectoryOut {
         authority_epoch: u64,
         result: Result<BlobBinding, NamespaceError>,
     },
+    LookedUp {
+        request_id: DirectoryRequestId,
+        authority_epoch: u64,
+        result: Result<NamespaceNode, NamespaceError>,
+    },
+    BlobReserved {
+        request_id: DirectoryRequestId,
+        authority_epoch: u64,
+        result: Result<(), NamespaceError>,
+    },
+    BlobReservationReleased {
+        request_id: DirectoryRequestId,
+        authority_epoch: u64,
+        result: Result<(), NamespaceError>,
+    },
     Unregistered {
         request_id: DirectoryRequestId,
         authority_epoch: u64,
@@ -214,6 +257,9 @@ impl DataDirectoryOut {
         match self {
             Self::Registered { request_id, .. }
             | Self::Resolved { request_id, .. }
+            | Self::LookedUp { request_id, .. }
+            | Self::BlobReserved { request_id, .. }
+            | Self::BlobReservationReleased { request_id, .. }
             | Self::Unregistered { request_id, .. }
             | Self::StreamOpened { request_id, .. }
             | Self::StreamClosed { request_id, .. } => *request_id,
@@ -229,10 +275,22 @@ pub enum NamespaceRequest {
         length: u64,
         recovery: SourceRecovery,
         operation_id: OperationId,
+        reservation: Option<OperationId>,
     },
     Resolve {
         path: DataPath,
     },
+    Lookup {
+        path: DataPath,
+    },
+    ReserveBlob {
+        path: DataPath,
+        operation_id: OperationId,
+    },
+    ReleaseBlobReservation {
+        path: DataPath,
+        operation_id: OperationId,
+    },
     Unregister {
         path: DataPath,
         operation_id: OperationId,
@@ -243,6 +301,8 @@ pub enum NamespaceRequest {
         endpoint: ActorAddress,
         descriptor: Vec<u8>,
         replace: bool,
+        ensure: bool,
+        expected_revision: Option<u64>,
         operation_id: OperationId,
     },
     CloseStream {
@@ -259,6 +319,11 @@ pub enum NamespaceClientIn {
     Cancel {
         reply_to: ActorAddress,
     },
+    CancelBlobReservation {
+        path: DataPath,
+        operation_id: OperationId,
+        reply_to: ActorAddress,
+    },
     DirectoryReply(DataDirectoryOut),
     TransferFailed {
         destination: ActorAddress,
@@ -288,6 +353,7 @@ struct PendingStream {
     descriptor: Vec<u8>,
     reply_to: ActorAddress,
     revision: u64,
+    opened_from_revision: Option<u64>,
 }
 
 struct StreamOpenRequest {
@@ -296,6 +362,8 @@ struct StreamOpenRequest {
     role: StreamRole,
     endpoint: ActorAddress,
     replace: bool,
+    ensure: bool,
+    expected_revision: Option<u64>,
     descriptor: Vec<u8>,
     operation_id: OperationId,
     reply_to: ActorAddress,
@@ -316,6 +384,7 @@ pub struct DataDirectoryActor {
     store: NamespaceStore,
     sources: BTreeMap<DataPath, RuntimeSource>,
     streams: BTreeMap<DataPath, RuntimeStream>,
+    blob_reservations: BTreeMap<DataPath, OperationId>,
     authority_epoch: u64,
 }
 
@@ -339,6 +408,7 @@ impl DataDirectoryActor {
             store,
             sources,
             streams: BTreeMap::new(),
+            blob_reservations: BTreeMap::new(),
             authority_epoch,
         })
     }
@@ -375,6 +445,9 @@ impl DataDirectoryActor {
         operation_id: OperationId,
         retired: Option<ActorAddress>,
     ) -> Result<MutationReceipt, NamespaceError> {
+        if self.blob_reservations.contains_key(&path) {
+            return Err(NamespaceError::PathExists(path));
+        }
         let request = MutationRequest::BindStream { path: path.clone() };
         if let Some(replayed) = self.replay(operation_id, &request) {
             return replayed;
@@ -388,6 +461,7 @@ impl DataDirectoryActor {
         let mut next = self.store.snapshot().clone();
         next.next_revision = next_revision;
         next.bindings.remove(&path);
+        next.stream_nodes.insert(path.clone(), revision);
         if let Some(retired) = retired
             && !next.retirements.contains(&retired)
         {
@@ -440,10 +514,42 @@ impl DataDirectoryActor {
             role,
             endpoint,
             replace,
+            ensure,
+            expected_revision,
             descriptor,
             operation_id,
             reply_to,
         } = request;
+        if !ensure {
+            let snapshot = self.store.snapshot();
+            let Some(current_revision) = snapshot.stream_nodes.get(&path).copied() else {
+                let error = if snapshot.bindings.contains_key(&path) {
+                    NamespaceError::WrongEntryType {
+                        path,
+                        expected: EntryKind::Stream,
+                        found: EntryKind::Blob,
+                    }
+                } else {
+                    NamespaceError::PathNotFound(path)
+                };
+                self.send_stream_result(ctx, request_id, reply_to, Err(error));
+                return;
+            };
+            let pending_from_expected = matches!(
+                self.streams.get(&path),
+                Some(RuntimeStream::Pending(pending))
+                    if pending.opened_from_revision == expected_revision
+            );
+            if expected_revision != Some(current_revision) && !pending_from_expected {
+                self.send_stream_result(
+                    ctx,
+                    request_id,
+                    reply_to,
+                    Err(NamespaceError::PathReplaced(path)),
+                );
+                return;
+            }
+        }
         if let Some(RuntimeStream::Pending(pending)) = self.streams.get_mut(&path)
             && pending.operation_id == operation_id
         {
@@ -583,6 +689,7 @@ impl DataDirectoryActor {
                         request_id,
                         reply_to,
                         revision: receipt.revision,
+                        opened_from_revision: (!ensure).then_some(expected_revision).flatten(),
                     }),
                 );
             }
@@ -620,6 +727,41 @@ impl DataDirectoryActor {
         }
     }
 
+    fn reserve_blob(
+        &mut self,
+        path: DataPath,
+        operation_id: OperationId,
+    ) -> Result<(), NamespaceError> {
+        if self.store.snapshot().bindings.contains_key(&path)
+            || self.store.snapshot().stream_nodes.contains_key(&path)
+        {
+            return Err(NamespaceError::PathExists(path));
+        }
+        match self.blob_reservations.get(&path) {
+            Some(existing) if *existing == operation_id => Ok(()),
+            Some(_) => Err(NamespaceError::PathExists(path)),
+            None => {
+                self.blob_reservations.insert(path, operation_id);
+                Ok(())
+            }
+        }
+    }
+
+    fn release_blob_reservation(
+        &mut self,
+        path: &DataPath,
+        operation_id: OperationId,
+    ) -> Result<(), NamespaceError> {
+        match self.blob_reservations.get(path) {
+            Some(existing) if *existing == operation_id => {
+                self.blob_reservations.remove(path);
+                Ok(())
+            }
+            Some(_) => Err(NamespaceError::OperationConflict(operation_id)),
+            None => Ok(()),
+        }
+    }
+
     fn register(
         &mut self,
         path: DataPath,
@@ -627,6 +769,7 @@ impl DataDirectoryActor {
         length: u64,
         recovery: SourceRecovery,
         operation_id: OperationId,
+        reservation: Option<OperationId>,
         retired: Option<ActorAddress>,
     ) -> Result<MutationReceipt, NamespaceError> {
         let request = MutationRequest::Register {
@@ -637,6 +780,12 @@ impl DataDirectoryActor {
         if let Some(replayed) = self.replay(operation_id, &request) {
             return replayed;
         }
+        match self.blob_reservations.get(&path) {
+            Some(existing) if Some(*existing) == reservation => {}
+            Some(_) => return Err(NamespaceError::PathExists(path)),
+            None if reservation.is_some() => return Err(NamespaceError::PathReplaced(path)),
+            None => {}
+        }
         let revision = self.store.snapshot().next_revision;
         let next_revision = revision
             .checked_add(1)
@@ -645,6 +794,7 @@ impl DataDirectoryActor {
         let receipt = MutationReceipt { revision };
         let mut next = self.store.snapshot().clone();
         next.next_revision = next_revision;
+        next.stream_nodes.remove(&path);
         next.bindings.insert(
             path.clone(),
             PersistedBinding {
@@ -666,12 +816,15 @@ impl DataDirectoryActor {
             },
         );
         self.store.commit(next)?;
+        if reservation.is_some() {
+            self.blob_reservations.remove(&path);
+        }
         self.sources.insert(path, RuntimeSource::Available(source));
         Ok(receipt)
     }
 
     fn resolve(&self, path: &DataPath) -> Result<BlobBinding, NamespaceError> {
-        if self.streams.contains_key(path) {
+        if self.store.snapshot().stream_nodes.contains_key(path) {
             return Err(NamespaceError::WrongEntryType {
                 path: path.clone(),
                 expected: EntryKind::Blob,
@@ -699,17 +852,43 @@ impl DataDirectoryActor {
         }
     }
 
+    fn lookup(&self, path: &DataPath) -> Result<NamespaceNode, NamespaceError> {
+        if self.blob_reservations.contains_key(path) {
+            return Err(NamespaceError::PathExists(path.clone()));
+        }
+        let snapshot = self.store.snapshot();
+        match (snapshot.bindings.get(path), snapshot.stream_nodes.get(path)) {
+            (Some(binding), None) => Ok(NamespaceNode {
+                kind: EntryKind::Blob,
+                revision: binding.revision,
+            }),
+            (None, Some(revision)) => Ok(NamespaceNode {
+                kind: EntryKind::Stream,
+                revision: *revision,
+            }),
+            (None, None) => Err(NamespaceError::PathNotFound(path.clone())),
+            (Some(_), Some(_)) => Err(NamespaceError::Storage(format!(
+                "data path {path} is bound as both blob and stream"
+            ))),
+        }
+    }
+
     fn unregister(
         &mut self,
         path: DataPath,
         operation_id: OperationId,
         retired: Option<ActorAddress>,
     ) -> Result<MutationReceipt, NamespaceError> {
+        if self.blob_reservations.contains_key(&path) {
+            return Err(NamespaceError::PathExists(path));
+        }
         let request = MutationRequest::Unregister { path: path.clone() };
         if let Some(replayed) = self.replay(operation_id, &request) {
             return replayed;
         }
-        if !self.store.snapshot().bindings.contains_key(&path) {
+        if !self.store.snapshot().bindings.contains_key(&path)
+            && !self.store.snapshot().stream_nodes.contains_key(&path)
+        {
             let mut next = self.store.snapshot().clone();
             next.operations.insert(
                 operation_id,
@@ -732,6 +911,7 @@ impl DataDirectoryActor {
         let mut next = self.store.snapshot().clone();
         next.next_revision = next_revision;
         next.bindings.remove(&path);
+        next.stream_nodes.remove(&path);
         if let Some(retired) = retired
             && !next.retirements.contains(&retired)
         {
@@ -769,6 +949,7 @@ impl ActorInterface for DataDirectoryActor {
                 length,
                 recovery,
                 operation_id,
+                reservation,
                 reply_to,
             } => {
                 let logical = path.clone();
@@ -780,7 +961,15 @@ impl ActorInterface for DataDirectoryActor {
                         RuntimeSource::Available(actor) if *actor != source => Some(*actor),
                         RuntimeSource::Available(_) | RuntimeSource::Unavailable(_) => None,
                     });
-                let result = self.register(path, source, length, recovery, operation_id, retired);
+                let result = self.register(
+                    path,
+                    source,
+                    length,
+                    recovery,
+                    operation_id,
+                    reservation,
+                    retired,
+                );
                 if result.is_ok()
                     && let Some(retired) = retired
                 {
@@ -813,6 +1002,53 @@ impl ActorInterface for DataDirectoryActor {
                     }),
                 );
             }
+            DataDirectoryIn::Lookup {
+                request_id,
+                path,
+                reply_to,
+            } => {
+                let result = self.lookup(&path);
+                let _ = ctx.send(
+                    reply_to,
+                    NamespaceClientIn::DirectoryReply(DataDirectoryOut::LookedUp {
+                        request_id,
+                        authority_epoch: self.authority_epoch,
+                        result,
+                    }),
+                );
+            }
+            DataDirectoryIn::ReserveBlob {
+                request_id,
+                path,
+                operation_id,
+                reply_to,
+            } => {
+                let result = self.reserve_blob(path, operation_id);
+                let _ = ctx.send(
+                    reply_to,
+                    NamespaceClientIn::DirectoryReply(DataDirectoryOut::BlobReserved {
+                        request_id,
+                        authority_epoch: self.authority_epoch,
+                        result,
+                    }),
+                );
+            }
+            DataDirectoryIn::ReleaseBlobReservation {
+                request_id,
+                path,
+                operation_id,
+                reply_to,
+            } => {
+                let result = self.release_blob_reservation(&path, operation_id);
+                let _ = ctx.send(
+                    reply_to,
+                    NamespaceClientIn::DirectoryReply(DataDirectoryOut::BlobReservationReleased {
+                        request_id,
+                        authority_epoch: self.authority_epoch,
+                        result,
+                    }),
+                );
+            }
             DataDirectoryIn::Unregister {
                 request_id,
                 path,
@@ -864,6 +1100,8 @@ impl ActorInterface for DataDirectoryActor {
                 endpoint,
                 descriptor,
                 replace,
+                ensure,
+                expected_revision,
                 operation_id,
                 reply_to,
             } => self.open_stream(
@@ -874,6 +1112,8 @@ impl ActorInterface for DataDirectoryActor {
                     role,
                     endpoint,
                     replace,
+                    ensure,
+                    expected_revision,
                     descriptor,
                     operation_id,
                     reply_to,
@@ -948,6 +1188,7 @@ impl NamespaceClientActor {
                 length,
                 recovery,
                 operation_id,
+                reservation,
             } => DataDirectoryIn::Register {
                 request_id,
                 path: path.clone(),
@@ -955,6 +1196,7 @@ impl NamespaceClientActor {
                 length: *length,
                 recovery: recovery.clone(),
                 operation_id: *operation_id,
+                reservation: *reservation,
                 reply_to: ctx.self_addr(),
             },
             NamespaceRequest::Resolve { path } => DataDirectoryIn::Resolve {
@@ -962,6 +1204,25 @@ impl NamespaceClientActor {
                 path: path.clone(),
                 reply_to: ctx.self_addr(),
             },
+            NamespaceRequest::Lookup { path } => DataDirectoryIn::Lookup {
+                request_id,
+                path: path.clone(),
+                reply_to: ctx.self_addr(),
+            },
+            NamespaceRequest::ReserveBlob { path, operation_id } => DataDirectoryIn::ReserveBlob {
+                request_id,
+                path: path.clone(),
+                operation_id: *operation_id,
+                reply_to: ctx.self_addr(),
+            },
+            NamespaceRequest::ReleaseBlobReservation { path, operation_id } => {
+                DataDirectoryIn::ReleaseBlobReservation {
+                    request_id,
+                    path: path.clone(),
+                    operation_id: *operation_id,
+                    reply_to: ctx.self_addr(),
+                }
+            }
             NamespaceRequest::Unregister { path, operation_id } => DataDirectoryIn::Unregister {
                 request_id,
                 path: path.clone(),
@@ -974,6 +1235,8 @@ impl NamespaceClientActor {
                 endpoint,
                 descriptor,
                 replace,
+                ensure,
+                expected_revision,
                 operation_id,
             } => DataDirectoryIn::OpenStream {
                 request_id,
@@ -982,6 +1245,8 @@ impl NamespaceClientActor {
                 endpoint: *endpoint,
                 descriptor: descriptor.clone(),
                 replace: *replace,
+                ensure: *ensure,
+                expected_revision: *expected_revision,
                 operation_id: *operation_id,
                 reply_to: ctx.self_addr(),
             },
@@ -1058,6 +1323,25 @@ impl ActorInterface for NamespaceClientActor {
                 self.pending
                     .retain(|_, pending| pending.reply_to != reply_to);
             }
+            NamespaceClientIn::CancelBlobReservation {
+                path,
+                operation_id,
+                reply_to,
+            } => {
+                self.pending
+                    .retain(|_, pending| pending.reply_to != reply_to);
+                if let Some(directory) = self.discovery.current_directory() {
+                    let _ = ctx.send(
+                        directory,
+                        DataDirectoryIn::ReleaseBlobReservation {
+                            request_id: DirectoryRequestId(0),
+                            path,
+                            operation_id,
+                            reply_to: ctx.self_addr(),
+                        },
+                    );
+                }
+            }
             NamespaceClientIn::DirectoryReply(reply) => {
                 if let Some(pending) = self.pending.remove(&reply.request_id()) {
                     let _ = ctx.send(pending.reply_to, reply);
@@ -1155,6 +1439,7 @@ impl NamespaceClient {
                 length,
                 recovery,
                 operation_id,
+                reservation: None,
             })
             .await?
         {
@@ -1174,6 +1459,15 @@ impl NamespaceClient {
         }
     }
 
+    pub async fn lookup(&self, path: DataPath) -> Result<NamespaceNode, NamespaceError> {
+        match self.request(NamespaceRequest::Lookup { path }).await? {
+            DataDirectoryOut::LookedUp { result, .. } => result,
+            other => Err(NamespaceError::Protocol(format!(
+                "expected lookup reply, received {other:?}"
+            ))),
+        }
+    }
+
     pub async fn unregister(
         &self,
         path: DataPath,
@@ -1204,6 +1498,8 @@ impl NamespaceClient {
                 role,
                 endpoint,
                 replace,
+                ensure: true,
+                expected_revision: None,
                 descriptor: Vec::new(),
                 operation_id,
             })
@@ -1335,6 +1631,7 @@ impl DirectoryClient {
                     length,
                     recovery,
                     operation_id,
+                    reservation: None,
                     reply_to: *inbox.addr(),
                 },
             )
@@ -1370,6 +1667,29 @@ impl DirectoryClient {
         }
     }
 
+    pub async fn lookup(&self, path: DataPath) -> Result<NamespaceNode, NamespaceError> {
+        let inbox = self
+            .runtime
+            .new_inbox::<NamespaceClientIn>()
+            .map_err(|error| NamespaceError::DirectoryUnavailable(error.to_string()))?;
+        self.runtime
+            .send_to(
+                self.directory,
+                DataDirectoryIn::Lookup {
+                    request_id: self.request_id(),
+                    path,
+                    reply_to: *inbox.addr(),
+                },
+            )
+            .map_err(|error| NamespaceError::DirectoryUnavailable(error.to_string()))?;
+        match self.receive(&inbox).await? {
+            DataDirectoryOut::LookedUp { result, .. } => result,
+            other => Err(NamespaceError::Protocol(format!(
+                "expected lookup reply, received {other:?}"
+            ))),
+        }
+    }
+
     pub async fn unregister(
         &self,
         path: DataPath,
@@ -1426,6 +1746,8 @@ impl DirectoryClient {
                     role,
                     endpoint,
                     replace,
+                    ensure: true,
+                    expected_revision: None,
                     operation_id,
                     descriptor: Vec::new(),
                     reply_to: *inbox.addr(),

crates/data-plane/src/namespace_store.rs

@@ -108,6 +108,8 @@ pub struct NamespaceSnapshot {
     pub authority_epoch: u64,
     pub next_revision: u64,
     pub bindings: BTreeMap<DataPath, PersistedBinding>,
+    #[serde(default)]
+    pub stream_nodes: BTreeMap<DataPath, u64>,
     pub operations: BTreeMap<OperationId, PersistedOperation>,
     #[serde(default)]
     pub retirements: Vec<ActorAddress>,
@@ -121,6 +123,7 @@ impl Default for NamespaceSnapshot {
             next_revision: 1,
             bindings: BTreeMap::new(),
             operations: BTreeMap::new(),
+            stream_nodes: BTreeMap::new(),
             retirements: Vec::new(),
         }
     }
@@ -143,6 +146,10 @@ impl NamespaceSnapshot {
             .bindings
             .values()
             .any(|binding| binding.revision == 0 || binding.revision >= self.next_revision)
+            || self
+                .stream_nodes
+                .values()
+                .any(|revision| *revision == 0 || *revision >= self.next_revision)
         {
             return Err(NamespaceStoreError::Corrupt(
                 "binding revision is outside the committed revision range".to_owned(),

crates/data-plane/src/protocol.rs

@@ -6,10 +6,12 @@ use serde::{Deserialize, Serialize};
 use swactor::actor::ActorAddress;
 use swactor_transport::{CodecRegistry, JsonCodec, NetworkMessage};
 
-use crate::blob::{BlobError, BlobLease, BlobMetadata};
+use crate::blob::{BlobError, BlobLease, BlobMetadata, ContentDigest};
 use crate::byte_ring::{RingHandle, Role};
 use crate::ids::BlobLeaseId;
-use crate::namespace::{EntryKind, NamespaceError, StreamIncarnation, StreamMatch};
+use crate::namespace::{
+    EntryKind, NamespaceError, NamespaceNode, OperationId, StreamIncarnation, StreamMatch,
+};
 use crate::path::DataPath;
 use crate::stream_transport::{StreamPeerDescriptor, StreamTransportEvent};
 
@@ -45,11 +47,155 @@ impl JobCapability {
 }
 
 #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
-pub enum DataOperation {
-    ReadBlob,
-    WriteBlob,
-    ReadStream,
-    WriteStream,
+pub enum AccessMode {
+    ReadOnly,
+    WriteOnly,
+    ReadWrite,
+}
+
+impl AccessMode {
+    pub const fn can_read(self) -> bool {
+        matches!(self, Self::ReadOnly | Self::ReadWrite)
+    }
+
+    pub const fn can_write(self) -> bool {
+        matches!(self, Self::WriteOnly | Self::ReadWrite)
+    }
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct BlobAllocation {
+    pub length: u64,
+    pub digest: Option<ContentDigest>,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct OpenOptions {
+    pub access: AccessMode,
+    pub create: bool,
+    pub exclusive: bool,
+    pub truncate: bool,
+    pub nonblocking: bool,
+    pub allocation: Option<BlobAllocation>,
+}
+
+impl OpenOptions {
+    pub const fn read_only() -> Self {
+        Self {
+            access: AccessMode::ReadOnly,
+            create: false,
+            exclusive: false,
+            truncate: false,
+            nonblocking: false,
+            allocation: None,
+        }
+    }
+
+    pub fn staged_blob(length: u64) -> Self {
+        Self {
+            access: AccessMode::WriteOnly,
+            create: true,
+            exclusive: false,
+            truncate: true,
+            nonblocking: false,
+            allocation: Some(BlobAllocation {
+                length,
+                digest: None,
+            }),
+        }
+    }
+
+    pub fn validate(&self) -> Result<(), DataPlaneError> {
+        if self.nonblocking {
+            return Err(DataPlaneError::Unsupported(
+                "O_NONBLOCK is not supported".to_owned(),
+            ));
+        }
+        if self.exclusive && !self.create {
+            return Err(DataPlaneError::InvalidArgument(
+                "O_EXCL requires O_CREAT".to_owned(),
+            ));
+        }
+        if (self.create || self.truncate) && !self.access.can_write() {
+            return Err(DataPlaneError::InvalidArgument(
+                "creation and truncation require write access".to_owned(),
+            ));
+        }
+        if self.allocation.is_some() && !(self.access.can_write() && self.truncate) {
+            return Err(DataPlaneError::InvalidArgument(
+                "blob allocation requires a truncating writable open".to_owned(),
+            ));
+        }
+        Ok(())
+    }
+}
+
+impl Default for OpenOptions {
+    fn default() -> Self {
+        Self::read_only()
+    }
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub enum DescriptorKind {
+    Blob,
+    Stream,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct DescriptorCapabilities(u16);
+
+impl DescriptorCapabilities {
+    pub const READ: Self = Self(1 << 0);
+    pub const WRITE: Self = Self(1 << 1);
+    pub const MAP_HOST: Self = Self(1 << 2);
+    pub const MAP_DEVICE: Self = Self(1 << 3);
+    pub const SEEK: Self = Self(1 << 4);
+    pub const POLL: Self = Self(1 << 5);
+    pub const CONTROL: Self = Self(1 << 6);
+
+    pub const fn empty() -> Self {
+        Self(0)
+    }
+
+    pub const fn contains(self, capability: Self) -> bool {
+        self.0 & capability.0 == capability.0
+    }
+
+    pub const fn union(self, capability: Self) -> Self {
+        Self(self.0 | capability.0)
+    }
+
+    pub const fn bits(self) -> u16 {
+        self.0
+    }
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub enum Errno {
+    Eacces,
+    Eagain,
+    Ebadf,
+    Ebusy,
+    Ecanceled,
+    Econnreset,
+    Eexist,
+    Einval,
+    Eio,
+    Enodev,
+    Enoent,
+    Enomem,
+    Enospc,
+    Enotsup,
+    Enxio,
+    Epipe,
+    Estale,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub enum OpenPolicy {
+    Ordinary,
+    EnsureStream { replace: bool },
 }
 
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
@@ -98,19 +244,27 @@ impl From<BlobError> for BlobFailure {
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub enum DataPlaneError {
     InvalidPath(String),
+    InvalidArgument(String),
     InvalidCapability,
     Attachment(AttachmentFailure),
     SessionNotRunning,
     SessionFailed(String),
     Unauthorized {
         path: DataPath,
-        operation: DataOperation,
+        access: AccessMode,
     },
     PathNotFound(DataPath),
+    PathExists(DataPath),
     SourceFailure(String),
     ArenaExhausted,
     Blob(BlobFailure),
     OperationCancelled,
+    BadDescriptor,
+    Unsupported(String),
+    MappingUnsupported,
+    Busy(String),
+    Stale(String),
+    BrokenPipe,
     WrongEntryType {
         path: DataPath,
         expected: EntryKind,
@@ -126,18 +280,26 @@ impl fmt::Display for DataPlaneError {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
             Self::InvalidPath(reason) => write!(f, "invalid data path: {reason}"),
+            Self::InvalidArgument(reason) => write!(f, "invalid argument: {reason}"),
             Self::InvalidCapability => f.write_str("invalid job capability"),
             Self::Attachment(reason) => write!(f, "data-plane attachment failed: {reason:?}"),
             Self::SessionNotRunning => f.write_str("data-plane session is not running"),
             Self::SessionFailed(reason) => write!(f, "data-plane session failed: {reason}"),
-            Self::Unauthorized { path, operation } => {
-                write!(f, "{operation:?} is not authorized for {path}")
+            Self::Unauthorized { path, access } => {
+                write!(f, "{access:?} access is not authorized for {path}")
             }
             Self::PathNotFound(path) => write!(f, "data path not found: {path}"),
+            Self::PathExists(path) => write!(f, "data path already exists: {path}"),
             Self::SourceFailure(reason) => write!(f, "blob source failed: {reason}"),
             Self::ArenaExhausted => f.write_str("data-plane arena is exhausted"),
             Self::Blob(reason) => write!(f, "blob lease failure: {reason:?}"),
             Self::OperationCancelled => f.write_str("data-plane operation was cancelled"),
+            Self::BadDescriptor => f.write_str("bad descriptor"),
+            Self::Unsupported(reason) => write!(f, "operation is not supported: {reason}"),
+            Self::MappingUnsupported => f.write_str("object does not support mapping"),
+            Self::Busy(reason) => write!(f, "resource is busy: {reason}"),
+            Self::Stale(reason) => write!(f, "stale capability: {reason}"),
+            Self::BrokenPipe => f.write_str("stream peer is closed"),
             Self::WrongEntryType {
                 path,
                 expected,
@@ -154,6 +316,44 @@ impl fmt::Display for DataPlaneError {
     }
 }
 
+impl DataPlaneError {
+    pub const fn errno(&self) -> Errno {
+        match self {
+            Self::InvalidPath(_) | Self::InvalidArgument(_) | Self::InvalidCapability => {
+                Errno::Einval
+            }
+            Self::Attachment(_)
+            | Self::SessionNotRunning
+            | Self::SessionFailed(_)
+            | Self::SourceFailure(_)
+            | Self::StreamFault(_) => Errno::Eio,
+            Self::Unauthorized { .. } => Errno::Eacces,
+            Self::PathNotFound(_) => Errno::Enoent,
+            Self::PathExists(_) => Errno::Eexist,
+            Self::ArenaExhausted => Errno::Enospc,
+            Self::Blob(BlobFailure::Bounds | BlobFailure::Length { .. }) => Errno::Einval,
+            Self::Blob(BlobFailure::StaleGeneration { .. }) | Self::Stale(_) => Errno::Estale,
+            Self::Blob(BlobFailure::Access) | Self::BadDescriptor | Self::StreamClosed => {
+                Errno::Ebadf
+            }
+            Self::Blob(BlobFailure::ActiveWritableView) | Self::Busy(_) => Errno::Ebusy,
+            Self::Blob(
+                BlobFailure::Digest
+                | BlobFailure::State { .. }
+                | BlobFailure::InvalidLease
+                | BlobFailure::AlreadyFinished,
+            ) => Errno::Eio,
+            Self::OperationCancelled => Errno::Ecanceled,
+            Self::Unsupported(_) => Errno::Enotsup,
+            Self::MappingUnsupported => Errno::Enodev,
+            Self::BrokenPipe => Errno::Epipe,
+            Self::WrongEntryType { .. } => Errno::Enxio,
+            Self::PathReplaced(_) => Errno::Estale,
+            Self::PeerLost => Errno::Econnreset,
+        }
+    }
+}
+
 impl std::error::Error for DataPlaneError {}
 
 impl From<BlobError> for DataPlaneError {
@@ -170,33 +370,24 @@ pub enum HostSessionIn {
         job_capability: JobCapability,
         child_node: Option<[u8; 32]>,
     },
-    OpenReadBlob {
+    Open {
         path: DataPath,
+        options: OpenOptions,
+        policy: OpenPolicy,
         child_session: ActorAddress,
         operation: ActorAddress,
     },
-    CancelReadBlob {
+    OpenResolved {
         operation: ActorAddress,
+        result: Result<NamespaceNode, NamespaceError>,
     },
-    OpenWriteBlob {
+    BlobReserved {
+        operation: ActorAddress,
         path: DataPath,
-        length: u64,
-        child_session: ActorAddress,
-        operation: ActorAddress,
+        reservation: OperationId,
+        result: Result<(), NamespaceError>,
     },
-    OpenReadStream {
-        path: DataPath,
-        child_session: ActorAddress,
-        operation: ActorAddress,
-        replace: bool,
-    },
-    OpenWriteStream {
-        path: DataPath,
-        child_session: ActorAddress,
-        operation: ActorAddress,
-        replace: bool,
-    },
-    CancelStream {
+    CancelOpen {
         operation: ActorAddress,
     },
     StreamControl {
@@ -253,16 +444,13 @@ pub enum ChildSessionIn {
         error: DataPlaneError,
     },
     AttachmentDeadline,
-    ReadBlob {
+    Open {
         path: DataPath,
+        options: OpenOptions,
+        policy: OpenPolicy,
         reply_to: ActorAddress,
     },
-    CancelRead {
-        reply_to: ActorAddress,
-    },
-    OpenWriteBlob {
-        path: DataPath,
-        length: u64,
+    CancelOpen {
         reply_to: ActorAddress,
     },
     OpenReadStream {
@@ -270,11 +458,6 @@ pub enum ChildSessionIn {
         reply_to: ActorAddress,
         replace: bool,
     },
-    OpenWriteStream {
-        path: DataPath,
-        reply_to: ActorAddress,
-        replace: bool,
-    },
     CancelStream {
         reply_to: ActorAddress,
     },
@@ -356,6 +539,10 @@ pub enum HostStreamIn {
     PeerTerminated {
         incarnation: StreamIncarnation,
         error: DataPlaneError,
+        reply_to: Option<ActorAddress>,
+    },
+    PeerTerminationAck {
+        incarnation: StreamIncarnation,
     },
     ReleaseComplete(Result<(), DataPlaneError>),
 }

crates/data-plane/tests/actor_blob_guarantees.rs

@@ -1,357 +1,6 @@
 #![cfg(target_os = "linux")]
 
-use std::sync::Arc;
-use std::sync::atomic::{AtomicU64, Ordering};
-use std::time::Duration;
-
-use data_plane::arena::{ArenaConfig, ArenaManager, NodeId};
-use data_plane::blob::{
-    BLOB_HEADER_LEN, Blob, BlobError, BlobLease, BlobMetadata, BlobSharedState, LeaseReleaser,
-};
-use data_plane::bootstrap::{self, BootstrapSpec};
-use data_plane::data_plane::DataPlaneBootstrap;
-use data_plane::host::{HostDataPlaneConfig, HostDataPlaneSessionActor};
-use data_plane::path::{DataPath, JobContext};
-use data_plane::protocol::{DataPlaneError, JobCapability};
-use futures_lite::future::{self, FutureExt};
-use swactor::Error;
-use swactor::actor::ActorAddress;
-use swactor::config::RuntimeConfig;
-use swactor::runtime::{RemoteSink, Runtime, RuntimeParts};
-use swactor_engine::{Engine, TokioBackend, TokioConfig};
-
-const CAPABILITY: JobCapability = JobCapability::new([9; 32]);
-const ARENA_GENERATION: u64 = 17;
-const SESSION_GENERATION: u64 = 29;
-const WEIGHTS: &[u8] = b"0123456789abcdefghijklmn";
-static STREAM_TEST_LOCK: parking_lot::Mutex<()> = parking_lot::Mutex::new(());
-
-struct DirectRuntimeSink {
-    destination: Runtime,
-}
-
-impl RemoteSink for DirectRuntimeSink {
-    fn send(
-        &self,
-        address: ActorAddress,
-        message: Box<dyn std::any::Any + Send>,
-    ) -> Result<(), Error> {
-        self.destination.deliver_raw(address, message)
-    }
-}
-
-struct BlackHoleSink;
-
-impl RemoteSink for BlackHoleSink {
-    fn send(
-        &self,
-        _address: ActorAddress,
-        _message: Box<dyn std::any::Any + Send>,
-    ) -> Result<(), Error> {
-        Ok(())
-    }
-}
-
-struct Harness {
-    _host_engine: Engine,
-    _child_engine: Engine,
-    _temp: TempState,
-    bootstrap: DataPlaneBootstrap,
-}
-
-static NEXT_TEMP: AtomicU64 = AtomicU64::new(1);
-
-struct TempState {
-    root: std::path::PathBuf,
-}
-
-impl TempState {
-    fn new() -> Self {
-        let sequence = NEXT_TEMP.fetch_add(1, Ordering::Relaxed);
-        let root = std::env::temp_dir().join(format!(
-            "swactor-actor-blob-{}-{sequence}",
-            std::process::id()
-        ));
-        std::fs::create_dir_all(&root).unwrap();
-        Self { root }
-    }
-}
-
-impl Drop for TempState {
-    fn drop(&mut self) {
-        let _ = std::fs::remove_dir_all(&self.root);
-    }
-}
-
-fn path(value: &str) -> DataPath {
-    DataPath::parse(value).expect("test path")
-}
-
-fn runtime_parts() -> (RuntimeParts, Runtime) {
-    let parts = RuntimeParts::new(RuntimeConfig {
-        worker_count: 1,
-        ..RuntimeConfig::default()
-    });
-    let runtime = parts.runtime().clone();
-    (parts, runtime)
-}
-
-struct LoopbackSender {
-    runtime: Runtime,
-}
-
-impl data_plane::blob_transfer::BlobTransferSender for LoopbackSender {
-    fn start_file(
-        &self,
-        request: data_plane::blob_transfer::FileTransferRequest,
-    ) -> Result<(), String> {
-        use std::os::unix::fs::FileExt;
-        let mut bytes = vec![0_u8; request.length as usize];
-        request
-            .file
-            .read_exact_at(&mut bytes, request.offset)
-            .map_err(|error| error.to_string())?;
-        self.runtime
-            .send_to(
-                request.offer.destination,
-                data_plane::blob_transfer::BlobTransferEvent::Chunk {
-                    transfer_id: request.offer.transfer_id,
-                    bytes,
-                },
-            )
-            .map_err(|error| error.to_string())?;
-        self.runtime
-            .send_to(
-                request.offer.destination,
-                data_plane::blob_transfer::BlobTransferEvent::Finished {
-                    transfer_id: request.offer.transfer_id,
-                },
-            )
-            .map_err(|error| error.to_string())?;
-        request.completion.complete(Ok(()));
-        Ok(())
-    }
-}
-
-struct DirectReceiver;
-
-impl data_plane::blob_transfer::BlobTransferReceiver for DirectReceiver {
-    fn open(
-        &self,
-        destination: ActorAddress,
-        transfer_id: data_plane::blob_transfer::BlobTransferId,
-    ) -> Result<data_plane::blob_transfer::BlobTransferOffer, String> {
-        Ok(data_plane::blob_transfer::BlobTransferOffer {
-            transfer_id,
-            destination,
-            failure_proxy: None,
-            transport: Vec::new(),
-        })
-    }
-
-    fn cancel(&self, _offer: &data_plane::blob_transfer::BlobTransferOffer) {}
-}
-
-struct StaticDiscovery(ActorAddress);
-
-impl data_plane::namespace::NamespaceDiscovery for StaticDiscovery {
-    fn current_directory(&self) -> Option<ActorAddress> {
-        Some(self.0)
-    }
-}
-
-struct NoopSourceRegistrar;
-
-impl data_plane::source::BlobSourcePublisher for NoopSourceRegistrar {
-    fn publish_source(&self, _source: ActorAddress) -> Result<(), String> {
-        Ok(())
-    }
-}
-
-struct RejectingStreamTransport;
-
-impl data_plane::stream_transport::StreamTransport for RejectingStreamTransport {
-    fn descriptor(&self) -> Result<data_plane::stream_transport::StreamPeerDescriptor, String> {
-        Ok(data_plane::stream_transport::StreamPeerDescriptor(vec![1]))
-    }
-
-    fn install_source(
-        &self,
-        _request: data_plane::stream_transport::StreamSourceRequest,
-    ) -> Result<(), String> {
-        Err("injected source transport failure".to_owned())
-    }
-
-    fn install_sink(
-        &self,
-        _request: data_plane::stream_transport::StreamSinkRequest,
-    ) -> Result<(), String> {
-        Err("injected sink transport failure".to_owned())
-    }
-
-    fn source_progress(&self, _incarnation: data_plane::namespace::StreamIncarnation) {}
-
-    fn sink_progress(&self, _incarnation: data_plane::namespace::StreamIncarnation) {}
-
-    fn terminate(&self, _incarnation: data_plane::namespace::StreamIncarnation) {}
-}
-
-struct CollectBytes(Arc<parking_lot::Mutex<Vec<u8>>>);
-
-impl data_plane::data_plane::StreamConsumer for CollectBytes {
-    fn consume(&self, bytes: &[u8]) -> Result<(), String> {
-        self.0.lock().extend_from_slice(bytes);
-        Ok(())
-    }
-}
-
-fn harness(arena_bytes: u64) -> Harness {
-    harness_with_transport(
-        arena_bytes,
-        Arc::new(data_plane::stream_transport::LocalStreamTransport::new()),
-    )
-}
-
-fn harness_with_transport(
-    arena_bytes: u64,
-    stream_transport: Arc<dyn data_plane::stream_transport::StreamTransport>,
-) -> Harness {
-    let temp = TempState::new();
-    let mut arena = ArenaManager::boot(ArenaConfig {
-        node_id: NodeId(1),
-        reservation_ceiling: arena_bytes,
-        base_alignment: 64,
-    })
-    .expect("host arena");
-    let handoff = bootstrap::write_bootstrap(
-        &mut arena,
-        BootstrapSpec {
-            arena_generation: ARENA_GENERATION,
-            alignment: 64,
-        },
-    )
-    .expect("bootstrap");
-
-    let (host_parts, host_runtime) = runtime_parts();
-    let (child_parts, child_runtime) = runtime_parts();
-    host_runtime.set_remote_sink(Arc::new(DirectRuntimeSink {
-        destination: child_runtime.clone(),
-    }));
-    child_runtime.set_remote_sink(Arc::new(DirectRuntimeSink {
-        destination: host_runtime.clone(),
-    }));
-    let host_engine = Engine::new(
-        host_parts,
-        TokioBackend::new(TokioConfig {
-            worker_threads: 1,
-            ..TokioConfig::default()
-        })
-        .expect("host backend"),
-    )
-    .expect("host engine");
-    let child_engine = Engine::new(
-        child_parts,
-        TokioBackend::new(TokioConfig {
-            worker_threads: 1,
-            ..TokioConfig::default()
-        })
-        .expect("child backend"),
-    )
-    .expect("child engine");
-
-    let sender: Arc<dyn data_plane::blob_transfer::BlobTransferSender> = Arc::new(LoopbackSender {
-        runtime: host_runtime.clone(),
-    });
-    let directory_actor = data_plane::namespace::DataDirectoryActor::recover(
-        temp.root.join("namespace.json"),
-        |_record, _length| {
-            Err(data_plane::namespace::NamespaceError::SourceRecovery(
-                "unexpected recovery".to_owned(),
-            ))
-        },
-    )
-    .unwrap();
-    let directory = host_runtime.spawn(directory_actor).unwrap();
-    let directory_client =
-        data_plane::namespace::DirectoryClient::new(host_runtime.clone(), directory);
-    for (logical, name, bytes) in [
-        ("/models/tiny-linear/weights", "weights.bin", WEIGHTS),
-        ("/models/second", "second.bin", b"second-blob".as_slice()),
-    ] {
-        let file_path = temp.root.join(name);
-        std::fs::write(&file_path, bytes).unwrap();
-        let source = data_plane::source::FileBlobSourceActor::open(
-            host_runtime.clone(),
-            Arc::clone(&sender),
-            &file_path,
-        )
-        .unwrap();
-        let length = source.length();
-        let recovery = source.recovery();
-        let source = host_runtime.spawn(source).unwrap();
-        future::block_on(directory_client.register(
-            path(logical),
-            source,
-            length,
-            recovery,
-            data_plane::namespace::OperationId::from_u128(u128::from(length) + 1),
-        ))
-        .unwrap();
-    }
-    let proxy = host_runtime
-        .spawn(data_plane::namespace::NamespaceClientActor::new(
-            host_engine.handle(),
-            host_runtime.create_sender(),
-            Arc::new(StaticDiscovery(directory)),
-            Duration::from_millis(5),
-        ))
-        .unwrap();
-    let namespace = data_plane::namespace::NamespaceClient::new(host_runtime.clone(), proxy);
-    let host_session = host_runtime
-        .spawn(
-            HostDataPlaneSessionActor::new(HostDataPlaneConfig {
-                runtime: host_runtime.clone(),
-                arena,
-                arena_generation: ARENA_GENERATION,
-                session_generation: SESSION_GENERATION,
-                capability: CAPABILITY,
-                job_context: JobContext {
-                    run_id: "run-7".to_owned(),
-                    read_prefixes: vec![path("/models"), path("/runs/run-7/results")],
-                    write_prefixes: vec![path("/runs/run-7/results")],
-                },
-                namespace: Some(namespace),
-                transfer_receiver: Some(Arc::new(DirectReceiver)),
-                source_sender: Some(sender),
-                source_publisher: Some(Arc::new(NoopSourceRegistrar)),
-                route_registrar: None,
-                stream_transport: Some(stream_transport),
-            })
-            .expect("host session config"),
-        )
-        .expect("spawn host session");
-    let bootstrap = future::block_on(DataPlaneBootstrap::attach(
-        handoff.arena_fd,
-        child_runtime,
-        host_session,
-        CAPABILITY,
-    ))
-    .expect("routed attachment");
-
-    Harness {
-        _host_engine: host_engine,
-        _child_engine: child_engine,
-        _temp: temp,
-        bootstrap,
-    }
-}
-
-#[derive(Default)]
-struct NoopReleaser;
-
-impl LeaseReleaser for NoopReleaser {
-    fn release(&self, _lease: BlobLease) {}
-}
+include!("data_plane_test_support.inc");
 
 #[test]
 fn attachment_without_a_host_reply_fails_on_actor_deadline() {
@@ -874,3 +523,190 @@ fn actor_stream_consumer_registers_before_writer_and_collects_to_eof() {
     completion.wait().expect("collector completes");
     assert_eq!(&*observed.lock(), b"actor-consumer");
 }
+
+#[test]
+fn raw_blob_descriptor_enforces_offsets_rights_and_terminal_state() {
+    let harness = harness(2 << 20);
+    let data_plane = harness.bootstrap.data_plane.clone();
+    let logical = path("/runs/self/results/raw-blob");
+
+    future::block_on(async {
+        let mut writer = data_plane
+            .open(&logical, OpenOptions::staged_blob(8))
+            .await
+            .expect("open staged descriptor");
+        assert_eq!(writer.kind(), DescriptorKind::Blob);
+        assert!(
+            writer
+                .capabilities()
+                .contains(DescriptorCapabilities::WRITE)
+        );
+        assert_eq!(writer.write(b"abc").await.expect("first write"), 3);
+        assert_eq!(writer.write(b"defgh").await.expect("second write"), 5);
+        assert_eq!(writer.last_route(), Some(TransferRoute::Staged));
+        assert_eq!(
+            writer
+                .write(b"!")
+                .await
+                .expect_err("growth rejected")
+                .errno(),
+            Errno::Enotsup
+        );
+        writer.close().await.expect("publish");
+        assert_eq!(
+            writer.close().await.expect_err("double close").errno(),
+            Errno::Ebadf
+        );
+
+        let mut reader = data_plane
+            .open(&logical, OpenOptions::read_only())
+            .await
+            .expect("open published descriptor");
+        assert_eq!(reader.kind(), DescriptorKind::Blob);
+        let mut first = [0xa5; 5];
+        reader
+            .read_exact(&mut first[..3])
+            .await
+            .expect("exact prefix read");
+        assert_eq!(&first, b"abc\xa5\xa5");
+        let mut rest = [0_u8; 8];
+        assert_eq!(reader.read(&mut rest).await.expect("remaining read"), 5);
+        assert_eq!(reader.read(&mut rest).await.expect("eof"), 0);
+        assert_eq!(
+            reader.write(b"x").await.expect_err("wrong access").errno(),
+            Errno::Ebadf
+        );
+        assert_eq!(reader.read(&mut []).await.expect("zero length"), 0);
+        reader.close().await.expect("close reader");
+
+        let mut source = data_plane
+            .open(&logical, OpenOptions::read_only())
+            .await
+            .expect("open arena-region source");
+        let target_path = path("/runs/self/results/raw-region-target");
+        let mut target = data_plane
+            .open(&target_path, OpenOptions::staged_blob(8))
+            .await
+            .expect("open arena-region target");
+        let mut target_mapping = target
+            .map(MapRequest {
+                protection: Protection::ReadWrite,
+                sharing: Sharing::Shared,
+                target: MapTarget::Host,
+                offset: 0,
+                length: 8,
+            })
+            .expect("map arena target");
+        assert_eq!(target_mapping.route(), TransferRoute::Direct);
+        let count = source
+            .read_into(RegionSlice::arena(
+                target_mapping.as_mut().expect("writable arena region"),
+            ))
+            .await
+            .expect("read into arena region");
+        assert_eq!(count, 8);
+        assert_eq!(target_mapping.as_ref(), b"abcdefgh");
+        drop(target_mapping);
+        target.abort().await.expect("abort arena target");
+        source.close().await.expect("close arena source");
+        assert_eq!(
+            reader
+                .read(&mut rest)
+                .await
+                .expect_err("read after close")
+                .errno(),
+            Errno::Ebadf
+        );
+    });
+}
+
+#[test]
+fn raw_stream_descriptor_hides_record_boundaries_and_preserves_eof() {
+    let _stream_test = STREAM_TEST_LOCK.lock();
+    let harness = harness(2 << 20);
+    let data_plane = harness.bootstrap.data_plane.clone();
+    let logical = path("/runs/self/results/raw-stream");
+
+    future::block_on(async {
+        let typed_reader = data_plane.read_stream(&logical);
+        let typed_writer = data_plane.write_stream(&logical);
+        let (typed_reader, typed_writer) = future::zip(typed_reader, typed_writer).await;
+        let mut typed_reader = typed_reader.expect("seed stream reader");
+        let mut typed_writer = typed_writer.expect("seed stream writer");
+        typed_writer.close().await.expect("seed close");
+        assert_eq!(typed_reader.read().await.expect("seed eof"), None);
+
+        let raw_reader = data_plane.open(&logical, OpenOptions::read_only());
+        let raw_writer = data_plane.open(
+            &logical,
+            OpenOptions {
+                access: AccessMode::WriteOnly,
+                ..OpenOptions::default()
+            },
+        );
+        let (raw_reader, raw_writer) = future::zip(raw_reader, raw_writer).await;
+        let mut reader = raw_reader.expect("raw stream reader");
+        let mut writer = raw_writer.expect("raw stream writer");
+        assert_eq!(reader.kind(), DescriptorKind::Stream);
+        writer
+            .write_all(b"abcdefgh")
+            .await
+            .expect("stream write all");
+        writer.close().await.expect("writer close");
+
+        let mut chunk = [0_u8; 3];
+        assert_eq!(reader.read(&mut chunk).await.expect("chunk one"), 3);
+        assert_eq!(&chunk, b"abc");
+        assert_eq!(reader.read(&mut chunk).await.expect("chunk two"), 3);
+        assert_eq!(&chunk, b"def");
+        assert_eq!(reader.read(&mut chunk).await.expect("chunk three"), 2);
+        assert_eq!(&chunk[..2], b"gh");
+        assert_eq!(reader.read(&mut chunk).await.expect("stream eof"), 0);
+        assert_eq!(reader.read(&mut chunk).await.expect("sticky eof"), 0);
+        reader.close().await.expect("reader close");
+    });
+}
+
+#[test]
+fn raw_blob_mapping_is_bounded_and_can_outlive_descriptor_close() {
+    let harness = harness(2 << 20);
+    let data_plane = harness.bootstrap.data_plane.clone();
+    let logical = path("/runs/self/results/raw-mapped-blob");
+
+    future::block_on(async {
+        let mut writer = data_plane
+            .open(&logical, OpenOptions::staged_blob(6))
+            .await
+            .expect("open mapped writer");
+        let mut mapping = writer
+            .map(MapRequest {
+                protection: Protection::ReadWrite,
+                sharing: Sharing::Shared,
+                target: MapTarget::Host,
+                offset: 1,
+                length: 4,
+            })
+            .expect("bounded writable mapping");
+        assert_eq!(mapping.route(), TransferRoute::Direct);
+        mapping
+            .as_mut()
+            .expect("writable mapping")
+            .copy_from_slice(b"data");
+        writer.close().await.expect("deferred close intent");
+        assert_eq!(mapping.as_ref(), b"data");
+        drop(mapping);
+
+        let deadline = std::time::Instant::now() + Duration::from_secs(2);
+        let blob = loop {
+            match data_plane.read_blob(&logical).await {
+                Ok(blob) => break blob,
+                Err(DataPlaneError::PathNotFound(_)) if std::time::Instant::now() < deadline => {
+                    future::yield_now().await;
+                }
+                Err(error) => panic!("deferred publication failed: {error}"),
+            }
+        };
+        let view = blob.map().expect("published mapping");
+        assert_eq!(&view[..], b"\0data\0");
+    });
+}

crates/data-plane/tests/byte_ring_guarantees.rs

@@ -557,3 +557,73 @@ fn writable_record_is_invisible_until_commit() {
         .collect();
     assert_eq!(observed, (0..17).collect::<Vec<_>>());
 }
+
+#[test]
+fn partial_record_cursor_keeps_capacity_pinned_until_full_release() {
+    let (arena, handle) = installed(32, 1);
+    let mut producer = attach(&arena, handle, Role::Producer).expect("producer");
+    let mut consumer = attach(&arena, handle, Role::Consumer).expect("consumer");
+    producer
+        .send_record(RecordKind::Data, b"abcdefgh")
+        .expect("record");
+
+    let cursor = consumer
+        .record_cursor()
+        .expect("cursor")
+        .expect("record visible");
+    let mut first = [0xa5; 5];
+    assert_eq!(
+        consumer
+            .copy_record_range(cursor, 0, &mut first[..3])
+            .expect("partial prefix"),
+        3
+    );
+    assert_eq!(&first, b"abc\xa5\xa5");
+    assert!(matches!(
+        producer.reserve(20),
+        Err(FlowError::InsufficientSpace { .. })
+    ));
+
+    let mut rest = [0_u8; 8];
+    assert_eq!(
+        consumer
+            .copy_record_range(cursor, 3, &mut rest)
+            .expect("partial suffix"),
+        5
+    );
+    assert_eq!(&rest[..5], b"defgh");
+    consumer
+        .release_record_cursor(cursor)
+        .expect("release complete record");
+    assert!(producer.reserve(20).is_ok());
+}
+
+#[test]
+fn partial_record_cursor_hides_payload_wraparound() {
+    let (arena, handle) = installed(32, 1);
+    let mut producer = attach(&arena, handle, Role::Producer).expect("producer");
+    let mut consumer = attach(&arena, handle, Role::Consumer).expect("consumer");
+    producer
+        .send_record(RecordKind::Data, &[1; 18])
+        .expect("advance cursor");
+    consumer.recv_record().expect("consume advance");
+    producer
+        .send_record(RecordKind::Data, b"0123456789abcde")
+        .expect("wrapped record");
+
+    let cursor = consumer
+        .record_cursor()
+        .expect("cursor")
+        .expect("wrapped record visible");
+    let mut observed = [0_u8; 15];
+    assert_eq!(
+        consumer
+            .copy_record_range(cursor, 0, &mut observed)
+            .expect("copy wrapped payload"),
+        observed.len()
+    );
+    assert_eq!(&observed, b"0123456789abcde");
+    consumer
+        .release_record_cursor(cursor)
+        .expect("release wrapped record");
+}

crates/data-plane/tests/data_plane_test_support.inc

@@ -0,0 +1,364 @@
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::Duration;
+
+use data_plane::arena::{ArenaConfig, ArenaManager, NodeId};
+#[allow(unused_imports)]
+use data_plane::blob::{
+    BLOB_HEADER_LEN, Blob, BlobError, BlobLease, BlobMetadata, BlobSharedState, LeaseReleaser,
+};
+use data_plane::bootstrap::{self, BootstrapSpec};
+#[allow(unused_imports)]
+use data_plane::data_plane::{
+    DataPlaneBootstrap, MapRequest, MapTarget, Protection, RegionSlice, Sharing, TransferRoute,
+};
+use data_plane::host::{HostDataPlaneConfig, HostDataPlaneSessionActor};
+use data_plane::path::{DataPath, JobContext};
+#[allow(unused_imports)]
+use data_plane::protocol::{
+    AccessMode, DataPlaneError, DescriptorCapabilities, DescriptorKind, Errno, JobCapability,
+    OpenOptions,
+};
+#[allow(unused_imports)]
+use futures_lite::future::{self, FutureExt};
+use swactor::Error;
+use swactor::actor::ActorAddress;
+use swactor::config::RuntimeConfig;
+use swactor::runtime::{RemoteSink, Runtime, RuntimeParts};
+use swactor_engine::{Engine, TokioBackend, TokioConfig};
+
+const CAPABILITY: JobCapability = JobCapability::new([9; 32]);
+const ARENA_GENERATION: u64 = 17;
+const SESSION_GENERATION: u64 = 29;
+const WEIGHTS: &[u8] = b"0123456789abcdefghijklmn";
+static STREAM_TEST_LOCK: parking_lot::Mutex<()> = parking_lot::Mutex::new(());
+
+struct DirectRuntimeSink {
+    destination: Runtime,
+}
+
+impl RemoteSink for DirectRuntimeSink {
+    fn send(
+        &self,
+        address: ActorAddress,
+        message: Box<dyn std::any::Any + Send>,
+    ) -> Result<(), Error> {
+        self.destination.deliver_raw(address, message)
+    }
+}
+
+#[allow(dead_code)]
+struct BlackHoleSink;
+
+impl RemoteSink for BlackHoleSink {
+    fn send(
+        &self,
+        _address: ActorAddress,
+        _message: Box<dyn std::any::Any + Send>,
+    ) -> Result<(), Error> {
+        Ok(())
+    }
+}
+
+struct Harness {
+    _host_engine: Engine,
+    _child_engine: Engine,
+    _temp: TempState,
+    bootstrap: DataPlaneBootstrap,
+}
+
+static NEXT_TEMP: AtomicU64 = AtomicU64::new(1);
+
+struct TempState {
+    root: std::path::PathBuf,
+}
+
+impl TempState {
+    fn new() -> Self {
+        let sequence = NEXT_TEMP.fetch_add(1, Ordering::Relaxed);
+        let root = std::env::temp_dir().join(format!(
+            "swactor-actor-blob-{}-{sequence}",
+            std::process::id()
+        ));
+        std::fs::create_dir_all(&root).unwrap();
+        Self { root }
+    }
+}
+
+impl Drop for TempState {
+    fn drop(&mut self) {
+        let _ = std::fs::remove_dir_all(&self.root);
+    }
+}
+
+fn path(value: &str) -> DataPath {
+    DataPath::parse(value).expect("test path")
+}
+
+fn runtime_parts() -> (RuntimeParts, Runtime) {
+    let parts = RuntimeParts::new(RuntimeConfig {
+        worker_count: 1,
+        ..RuntimeConfig::default()
+    });
+    let runtime = parts.runtime().clone();
+    (parts, runtime)
+}
+
+struct LoopbackSender {
+    runtime: Runtime,
+}
+
+impl data_plane::blob_transfer::BlobTransferSender for LoopbackSender {
+    fn start_file(
+        &self,
+        request: data_plane::blob_transfer::FileTransferRequest,
+    ) -> Result<(), String> {
+        use std::os::unix::fs::FileExt;
+        let mut bytes = vec![0_u8; request.length as usize];
+        request
+            .file
+            .read_exact_at(&mut bytes, request.offset)
+            .map_err(|error| error.to_string())?;
+        self.runtime
+            .send_to(
+                request.offer.destination,
+                data_plane::blob_transfer::BlobTransferEvent::Chunk {
+                    transfer_id: request.offer.transfer_id,
+                    bytes,
+                },
+            )
+            .map_err(|error| error.to_string())?;
+        self.runtime
+            .send_to(
+                request.offer.destination,
+                data_plane::blob_transfer::BlobTransferEvent::Finished {
+                    transfer_id: request.offer.transfer_id,
+                },
+            )
+            .map_err(|error| error.to_string())?;
+        request.completion.complete(Ok(()));
+        Ok(())
+    }
+}
+
+struct DirectReceiver;
+
+impl data_plane::blob_transfer::BlobTransferReceiver for DirectReceiver {
+    fn open(
+        &self,
+        destination: ActorAddress,
+        transfer_id: data_plane::blob_transfer::BlobTransferId,
+    ) -> Result<data_plane::blob_transfer::BlobTransferOffer, String> {
+        Ok(data_plane::blob_transfer::BlobTransferOffer {
+            transfer_id,
+            destination,
+            failure_proxy: None,
+            transport: Vec::new(),
+        })
+    }
+
+    fn cancel(&self, _offer: &data_plane::blob_transfer::BlobTransferOffer) {}
+}
+
+struct StaticDiscovery(ActorAddress);
+
+impl data_plane::namespace::NamespaceDiscovery for StaticDiscovery {
+    fn current_directory(&self) -> Option<ActorAddress> {
+        Some(self.0)
+    }
+}
+
+struct NoopSourceRegistrar;
+
+impl data_plane::source::BlobSourcePublisher for NoopSourceRegistrar {
+    fn publish_source(&self, _source: ActorAddress) -> Result<(), String> {
+        Ok(())
+    }
+}
+
+struct RejectingStreamTransport;
+
+impl data_plane::stream_transport::StreamTransport for RejectingStreamTransport {
+    fn descriptor(&self) -> Result<data_plane::stream_transport::StreamPeerDescriptor, String> {
+        Ok(data_plane::stream_transport::StreamPeerDescriptor(vec![1]))
+    }
+
+    fn install_source(
+        &self,
+        _request: data_plane::stream_transport::StreamSourceRequest,
+    ) -> Result<(), String> {
+        Err("injected source transport failure".to_owned())
+    }
+
+    fn install_sink(
+        &self,
+        _request: data_plane::stream_transport::StreamSinkRequest,
+    ) -> Result<(), String> {
+        Err("injected sink transport failure".to_owned())
+    }
+
+    fn source_progress(&self, _incarnation: data_plane::namespace::StreamIncarnation) {}
+
+    fn sink_progress(&self, _incarnation: data_plane::namespace::StreamIncarnation) {}
+
+    fn terminate(&self, _incarnation: data_plane::namespace::StreamIncarnation) {}
+}
+
+#[allow(dead_code)]
+struct CollectBytes(Arc<parking_lot::Mutex<Vec<u8>>>);
+
+impl data_plane::data_plane::StreamConsumer for CollectBytes {
+    fn consume(&self, bytes: &[u8]) -> Result<(), String> {
+        self.0.lock().extend_from_slice(bytes);
+        Ok(())
+    }
+}
+
+fn harness(arena_bytes: u64) -> Harness {
+    harness_with_transport(
+        arena_bytes,
+        Arc::new(data_plane::stream_transport::LocalStreamTransport::new()),
+    )
+}
+
+fn harness_with_transport(
+    arena_bytes: u64,
+    stream_transport: Arc<dyn data_plane::stream_transport::StreamTransport>,
+) -> Harness {
+    let temp = TempState::new();
+    let mut arena = ArenaManager::boot(ArenaConfig {
+        node_id: NodeId(1),
+        reservation_ceiling: arena_bytes,
+        base_alignment: 64,
+    })
+    .expect("host arena");
+    let handoff = bootstrap::write_bootstrap(
+        &mut arena,
+        BootstrapSpec {
+            arena_generation: ARENA_GENERATION,
+            alignment: 64,
+        },
+    )
+    .expect("bootstrap");
+
+    let (host_parts, host_runtime) = runtime_parts();
+    let (child_parts, child_runtime) = runtime_parts();
+    host_runtime.set_remote_sink(Arc::new(DirectRuntimeSink {
+        destination: child_runtime.clone(),
+    }));
+    child_runtime.set_remote_sink(Arc::new(DirectRuntimeSink {
+        destination: host_runtime.clone(),
+    }));
+    let host_engine = Engine::new(
+        host_parts,
+        TokioBackend::new(TokioConfig {
+            worker_threads: 1,
+            ..TokioConfig::default()
+        })
+        .expect("host backend"),
+    )
+    .expect("host engine");
+    let child_engine = Engine::new(
+        child_parts,
+        TokioBackend::new(TokioConfig {
+            worker_threads: 1,
+            ..TokioConfig::default()
+        })
+        .expect("child backend"),
+    )
+    .expect("child engine");
+
+    let sender: Arc<dyn data_plane::blob_transfer::BlobTransferSender> = Arc::new(LoopbackSender {
+        runtime: host_runtime.clone(),
+    });
+    let directory_actor = data_plane::namespace::DataDirectoryActor::recover(
+        temp.root.join("namespace.json"),
+        |_record, _length| {
+            Err(data_plane::namespace::NamespaceError::SourceRecovery(
+                "unexpected recovery".to_owned(),
+            ))
+        },
+    )
+    .unwrap();
+    let directory = host_runtime.spawn(directory_actor).unwrap();
+    let directory_client =
+        data_plane::namespace::DirectoryClient::new(host_runtime.clone(), directory);
+    for (logical, name, bytes) in [
+        ("/models/tiny-linear/weights", "weights.bin", WEIGHTS),
+        ("/models/second", "second.bin", b"second-blob".as_slice()),
+    ] {
+        let file_path = temp.root.join(name);
+        std::fs::write(&file_path, bytes).unwrap();
+        let source = data_plane::source::FileBlobSourceActor::open(
+            host_runtime.clone(),
+            Arc::clone(&sender),
+            &file_path,
+        )
+        .unwrap();
+        let length = source.length();
+        let recovery = source.recovery();
+        let source = host_runtime.spawn(source).unwrap();
+        future::block_on(directory_client.register(
+            path(logical),
+            source,
+            length,
+            recovery,
+            data_plane::namespace::OperationId::from_u128(u128::from(length) + 1),
+        ))
+        .unwrap();
+    }
+    let proxy = host_runtime
+        .spawn(data_plane::namespace::NamespaceClientActor::new(
+            host_engine.handle(),
+            host_runtime.create_sender(),
+            Arc::new(StaticDiscovery(directory)),
+            Duration::from_millis(5),
+        ))
+        .unwrap();
+    let namespace = data_plane::namespace::NamespaceClient::new(host_runtime.clone(), proxy);
+    let host_session = host_runtime
+        .spawn(
+            HostDataPlaneSessionActor::new(HostDataPlaneConfig {
+                runtime: host_runtime.clone(),
+                arena,
+                arena_generation: ARENA_GENERATION,
+                session_generation: SESSION_GENERATION,
+                capability: CAPABILITY,
+                job_context: JobContext {
+                    run_id: "run-7".to_owned(),
+                    read_prefixes: vec![path("/models"), path("/runs/run-7/results")],
+                    write_prefixes: vec![path("/runs/run-7/results")],
+                },
+                namespace: Some(namespace),
+                transfer_receiver: Some(Arc::new(DirectReceiver)),
+                source_sender: Some(sender),
+                source_publisher: Some(Arc::new(NoopSourceRegistrar)),
+                route_registrar: None,
+                stream_transport: Some(stream_transport),
+            })
+            .expect("host session config"),
+        )
+        .expect("spawn host session");
+    let bootstrap = future::block_on(DataPlaneBootstrap::attach(
+        handoff.arena_fd,
+        child_runtime,
+        host_session,
+        CAPABILITY,
+    ))
+    .expect("routed attachment");
+
+    Harness {
+        _host_engine: host_engine,
+        _child_engine: child_engine,
+        _temp: temp,
+        bootstrap,
+    }
+}
+
+#[allow(dead_code)]
+#[derive(Default)]
+struct NoopReleaser;
+
+impl LeaseReleaser for NoopReleaser {
+    fn release(&self, _lease: BlobLease) {}
+}

crates/data-plane/tests/descriptor_guarantees.rs

@@ -0,0 +1,612 @@
+#![cfg(target_os = "linux")]
+
+include!("data_plane_test_support.inc");
+
+use std::alloc::{GlobalAlloc, Layout, System};
+use std::cell::Cell;
+
+use proptest::prelude::*;
+
+struct ThreadCountingAllocator;
+
+thread_local! {
+    static COUNT_ALLOCATIONS: Cell<bool> = const { Cell::new(false) };
+    static ALLOCATION_COUNT: Cell<usize> = const { Cell::new(0) };
+}
+
+#[global_allocator]
+static COUNTING_ALLOCATOR: ThreadCountingAllocator = ThreadCountingAllocator;
+
+unsafe impl GlobalAlloc for ThreadCountingAllocator {
+    unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
+        COUNT_ALLOCATIONS.with(|enabled| {
+            if enabled.get() {
+                ALLOCATION_COUNT.with(|count| count.set(count.get() + 1));
+            }
+        });
+        // SAFETY: this allocator delegates the unchanged layout to `System`.
+        unsafe { System.alloc(layout) }
+    }
+
+    unsafe fn dealloc(&self, pointer: *mut u8, layout: Layout) {
+        // SAFETY: `pointer` came from `System` with this layout.
+        unsafe { System.dealloc(pointer, layout) };
+    }
+
+    unsafe fn alloc_zeroed(&self, layout: Layout) -> *mut u8 {
+        COUNT_ALLOCATIONS.with(|enabled| {
+            if enabled.get() {
+                ALLOCATION_COUNT.with(|count| count.set(count.get() + 1));
+            }
+        });
+        // SAFETY: this allocator delegates the unchanged layout to `System`.
+        unsafe { System.alloc_zeroed(layout) }
+    }
+
+    unsafe fn realloc(&self, pointer: *mut u8, layout: Layout, size: usize) -> *mut u8 {
+        COUNT_ALLOCATIONS.with(|enabled| {
+            if enabled.get() {
+                ALLOCATION_COUNT.with(|count| count.set(count.get() + 1));
+            }
+        });
+        // SAFETY: `pointer` and `layout` came from `System`.
+        unsafe { System.realloc(pointer, layout, size) }
+    }
+}
+
+fn start_allocation_count() {
+    ALLOCATION_COUNT.with(|count| count.set(0));
+    COUNT_ALLOCATIONS.with(|enabled| enabled.set(true));
+}
+
+fn finish_allocation_count() -> usize {
+    COUNT_ALLOCATIONS.with(|enabled| enabled.set(false));
+    ALLOCATION_COUNT.with(Cell::get)
+}
+
+const REQUIRED_MUTATION_TARGETS: &[(&str, &str)] = &[
+    (
+        "omit_access_check",
+        "forbidden_actions_report_stable_errno_and_do_not_mutate_live_state",
+    ),
+    (
+        "advance_requested_count",
+        "generated_blob_sequences_preserve_exact_tagged_prefixes",
+    ),
+    (
+        "early_stream_eof",
+        "raw_stream_descriptor_hides_record_boundaries_and_preserves_eof",
+    ),
+    (
+        "release_partial_record",
+        "partial_record_cursor_keeps_capacity_pinned_until_full_release",
+    ),
+    (
+        "duplicate_stream_prefix",
+        "raw_stream_descriptor_hides_record_boundaries_and_preserves_eof",
+    ),
+    (
+        "cross_wire_open_grant",
+        "concurrent_descriptor_opens_keep_paths_and_payloads_correlated",
+    ),
+    (
+        "publish_aborted_blob",
+        "legal_blob_sequences_cover_boundaries_offsets_mapping_close_and_abort",
+    ),
+    (
+        "reclaim_live_mapping",
+        "raw_blob_mapping_is_bounded_and_can_outlive_descriptor_close",
+    ),
+    (
+        "writable_read_only_export",
+        "test_raw_descriptor_blob_io_mapping_and_errno",
+    ),
+    (
+        "retain_cancelled_waiter",
+        "cancellation_and_transport_faults_reclaim_waiters_and_preserve_unrelated_progress",
+    ),
+    (
+        "double_publication",
+        "raw_blob_descriptor_enforces_offsets_rights_and_terminal_state",
+    ),
+    (
+        "authorize_unresolved_alias",
+        "missing_and_unauthorized_paths_are_rejected",
+    ),
+    (
+        "silently_stage_direct_map",
+        "raw_blob_mapping_is_bounded_and_can_outlive_descriptor_close",
+    ),
+    (
+        "touch_outside_region",
+        "generated_blob_sequences_preserve_exact_tagged_prefixes",
+    ),
+    (
+        "leak_failed_open_lease",
+        "cancelled_write_open_releases_queued_grant",
+    ),
+];
+
+#[test]
+fn mutation_adequacy_targets_cover_every_required_bad_behavior() {
+    let mut identifiers: Vec<&str> = REQUIRED_MUTATION_TARGETS
+        .iter()
+        .map(|(identifier, _)| *identifier)
+        .collect();
+    assert_eq!(identifiers.len(), 15);
+    identifiers.sort_unstable();
+    identifiers.dedup();
+    assert_eq!(identifiers.len(), 15, "mutation identifiers must be unique");
+    assert!(
+        REQUIRED_MUTATION_TARGETS
+            .iter()
+            .all(|(_, oracle)| !oracle.is_empty())
+    );
+}
+
+#[test]
+fn legal_blob_sequences_cover_boundaries_offsets_mapping_close_and_abort() {
+    let harness = harness(2 << 20);
+    let data_plane = harness.bootstrap.data_plane.clone();
+
+    future::block_on(async {
+        for (case, length) in [0_usize, 1, 7, 64].into_iter().enumerate() {
+            let logical = path(&format!("/runs/self/results/legal-{case}"));
+            let payload: Vec<u8> = (0..length).map(|index| index as u8).collect();
+            let mut writer = data_plane
+                .open(&logical, OpenOptions::staged_blob(length as u64))
+                .await
+                .expect("legal staged open");
+            for chunk in payload.chunks(3) {
+                writer.write_all(chunk).await.expect("legal partial write");
+            }
+            writer.close().await.expect("legal publication");
+
+            let mut reader = data_plane
+                .open(&logical, OpenOptions::read_only())
+                .await
+                .expect("legal read open");
+            let mapping = reader
+                .map(MapRequest {
+                    protection: Protection::Read,
+                    sharing: Sharing::Shared,
+                    target: MapTarget::Host,
+                    offset: 0,
+                    length: length as u64,
+                })
+                .expect("legal read mapping");
+            assert_eq!(mapping.as_ref(), payload);
+            assert_eq!(mapping.route(), TransferRoute::Direct);
+
+            let mut observed = Vec::new();
+            let mut destination = [0xa5_u8; 11];
+            loop {
+                let count = reader
+                    .read(&mut destination[..(case + 1).min(11)])
+                    .await
+                    .expect("legal sequential read");
+                if count == 0 {
+                    break;
+                }
+                observed.extend_from_slice(&destination[..count]);
+            }
+            assert_eq!(observed, payload);
+            assert_eq!(reader.read(&mut destination).await.expect("sticky eof"), 0);
+            reader.close().await.expect("legal reader close");
+        }
+
+        let aborted = path("/runs/self/results/legal-abort");
+        let mut writer = data_plane
+            .open(&aborted, OpenOptions::staged_blob(4))
+            .await
+            .expect("abort candidate");
+        writer.write_all(b"nope").await.expect("staged bytes");
+        writer.abort().await.expect("explicit abort");
+        assert_eq!(
+            data_plane
+                .open(&aborted, OpenOptions::read_only())
+                .await
+                .expect_err("aborted blob is absent")
+                .errno(),
+            Errno::Enoent
+        );
+    });
+}
+
+proptest! {
+    #![proptest_config(ProptestConfig::with_cases(8))]
+
+    #[test]
+    fn generated_blob_sequences_preserve_exact_tagged_prefixes(
+        payload in prop::collection::vec(any::<u8>(), 0..96),
+        write_size in 1_usize..17,
+        read_size in 1_usize..17,
+    ) {
+        let harness = harness(2 << 20);
+        let data_plane = harness.bootstrap.data_plane.clone();
+        future::block_on(async {
+            let logical = path("/runs/self/results/property-sequence");
+            let mut writer = data_plane
+                .open(&logical, OpenOptions::staged_blob(payload.len() as u64))
+                .await
+                .expect("property writer");
+            for chunk in payload.chunks(write_size) {
+                writer.write_all(chunk).await.expect("property write");
+            }
+            writer.close().await.expect("property publish");
+
+            let mut reader = data_plane
+                .open(&logical, OpenOptions::read_only())
+                .await
+                .expect("property reader");
+            let mut destination = vec![0xa5_u8; read_size + 2];
+            let mut observed = Vec::new();
+            loop {
+                destination.fill(0xa5);
+                let count = reader
+                    .read(&mut destination[1..=read_size])
+                    .await
+                    .expect("property read");
+                assert_eq!(destination[0], 0xa5);
+                assert_eq!(destination[read_size + 1], 0xa5);
+                if count == 0 {
+                    break;
+                }
+                observed.extend_from_slice(&destination[1..1 + count]);
+            }
+            // Kills: advance offset by requested rather than completed bytes,
+            // touch destination outside the returned prefix, duplicate/drop data.
+            prop_assert_eq!(observed, payload);
+            Ok(())
+        })?;
+    }
+}
+
+#[test]
+fn forbidden_actions_report_stable_errno_and_do_not_mutate_live_state() {
+    let harness = harness(2 << 20);
+    let data_plane = harness.bootstrap.data_plane.clone();
+    let logical = path("/runs/self/results/negative");
+
+    future::block_on(async {
+        let mut writer = data_plane
+            .open(&logical, OpenOptions::staged_blob(4))
+            .await
+            .expect("negative fixture writer");
+        writer.write_all(b"safe").await.expect("fixture bytes");
+        assert_eq!(
+            writer
+                .read(&mut [0_u8; 1])
+                .await
+                .expect_err("wrong access")
+                .errno(),
+            Errno::Ebadf
+        );
+        writer.close().await.expect("fixture publication");
+
+        let invalid = OpenOptions {
+            exclusive: true,
+            ..OpenOptions::default()
+        };
+        assert_eq!(
+            data_plane
+                .open(&logical, invalid)
+                .await
+                .expect_err("invalid flags")
+                .errno(),
+            Errno::Einval
+        );
+        assert_eq!(
+            data_plane
+                .open(
+                    &logical,
+                    OpenOptions {
+                        nonblocking: true,
+                        ..OpenOptions::default()
+                    },
+                )
+                .await
+                .expect_err("unsupported nonblocking")
+                .errno(),
+            Errno::Enotsup
+        );
+        assert_eq!(
+            data_plane
+                .open(&path("/models/missing-raw"), OpenOptions::read_only())
+                .await
+                .expect_err("missing path")
+                .errno(),
+            Errno::Enoent
+        );
+        assert_eq!(
+            data_plane
+                .open(
+                    &path("/models/tiny-linear/weights"),
+                    OpenOptions::staged_blob(1),
+                )
+                .await
+                .expect_err("unauthorized write")
+                .errno(),
+            Errno::Eacces
+        );
+        assert_eq!(
+            data_plane
+                .open(
+                    &logical,
+                    OpenOptions {
+                        exclusive: true,
+                        ..OpenOptions::staged_blob(4)
+                    },
+                )
+                .await
+                .expect_err("exclusive existing")
+                .errno(),
+            Errno::Eexist
+        );
+
+        let mut reader = data_plane
+            .open(&logical, OpenOptions::read_only())
+            .await
+            .expect("negative fixture reader");
+        assert_eq!(
+            reader
+                .map(MapRequest {
+                    protection: Protection::Read,
+                    sharing: Sharing::Shared,
+                    target: MapTarget::Host,
+                    offset: 3,
+                    length: 2,
+                })
+                .expect_err("mapping overrun")
+                .errno(),
+            Errno::Einval
+        );
+        assert_eq!(
+            reader
+                .write(b"x")
+                .await
+                .expect_err("read-only write")
+                .errno(),
+            Errno::Ebadf
+        );
+        let mut bytes = [0_u8; 4];
+        reader
+            .read_exact(&mut bytes)
+            .await
+            .expect("state unchanged");
+        assert_eq!(&bytes, b"safe");
+        reader.close().await.expect("first close");
+        assert_eq!(
+            reader.close().await.expect_err("double close").errno(),
+            Errno::Ebadf
+        );
+        // Kills: omitted access checks, mutation after failed bounds checks,
+        // silent flag downgrade, and revival after close.
+    });
+}
+
+#[test]
+fn exclusive_create_is_atomic_and_abort_releases_its_reservation() {
+    let harness = harness(2 << 20);
+    let data_plane = harness.bootstrap.data_plane.clone();
+    let logical = path("/runs/self/results/exclusive-race");
+    let options = OpenOptions {
+        exclusive: true,
+        ..OpenOptions::staged_blob(4)
+    };
+
+    future::block_on(async {
+        let first = data_plane.open(&logical, options.clone());
+        let second = data_plane.open(&logical, options.clone());
+        let (first, second) = future::zip(first, second).await;
+        let (mut winner, loser) = match (first, second) {
+            (Ok(winner), Err(loser)) | (Err(loser), Ok(winner)) => (winner, loser),
+            (Ok(_), Ok(_)) => panic!("both exclusive creators succeeded"),
+            (Err(first), Err(second)) => {
+                panic!("both exclusive creators failed: {first}; {second}")
+            }
+        };
+        assert_eq!(loser.errno(), Errno::Eexist);
+        winner.abort().await.expect("abort exclusive winner");
+
+        let mut replacement = data_plane
+            .open(&logical, options)
+            .await
+            .expect("reservation released after abort");
+        replacement
+            .abort()
+            .await
+            .expect("abort replacement reservation");
+    });
+}
+
+#[test]
+fn raw_stream_abort_completes_before_peer_observes_broken_pipe() {
+    let _stream_test = STREAM_TEST_LOCK.lock();
+    let harness = harness(2 << 20);
+    let data_plane = harness.bootstrap.data_plane.clone();
+    let logical = path("/runs/self/results/raw-broken-pipe");
+
+    future::block_on(async {
+        let seed_reader = data_plane.read_stream(&logical);
+        let seed_writer = data_plane.write_stream(&logical);
+        let (seed_reader, seed_writer) = future::zip(seed_reader, seed_writer).await;
+        let mut seed_reader = seed_reader.expect("seed reader");
+        let mut seed_writer = seed_writer.expect("seed writer");
+        seed_writer.close().await.expect("seed writer close");
+        assert_eq!(seed_reader.read().await.expect("seed eof"), None);
+
+        let reader = data_plane.open(&logical, OpenOptions::read_only());
+        let writer = data_plane.open(
+            &logical,
+            OpenOptions {
+                access: AccessMode::WriteOnly,
+                ..OpenOptions::default()
+            },
+        );
+        let (reader, writer) = future::zip(reader, writer).await;
+        let mut reader = reader.expect("raw reader");
+        let mut writer = writer.expect("raw writer");
+        reader.abort().await.expect("reader abort completion");
+        assert_eq!(
+            writer
+                .write(b"late")
+                .await
+                .expect_err("write after reader abort")
+                .errno(),
+            Errno::Epipe
+        );
+    });
+}
+
+#[test]
+fn concurrent_descriptor_opens_keep_paths_and_payloads_correlated() {
+    let harness = harness(8 << 20);
+    let data_plane = harness.bootstrap.data_plane.clone();
+
+    std::thread::scope(|scope| {
+        for index in 0_u8..16 {
+            let data_plane = data_plane.clone();
+            scope.spawn(move || {
+                future::block_on(async move {
+                    let logical = path(&format!("/runs/self/results/concurrent-{index}"));
+                    let payload = [index; 32];
+                    let mut writer = data_plane
+                        .open(&logical, OpenOptions::staged_blob(payload.len() as u64))
+                        .await
+                        .expect("concurrent writer");
+                    writer
+                        .write_all(&payload)
+                        .await
+                        .expect("concurrent payload");
+                    writer.close().await.expect("concurrent publish");
+                });
+            });
+        }
+    });
+
+    std::thread::scope(|scope| {
+        for index in 0_u8..16 {
+            let data_plane = data_plane.clone();
+            scope.spawn(move || {
+                future::block_on(async move {
+                    let logical = path(&format!("/runs/self/results/concurrent-{index}"));
+                    let mut reader = data_plane
+                        .open(&logical, OpenOptions::read_only())
+                        .await
+                        .expect("concurrent reader");
+                    let mut payload = [0_u8; 32];
+                    reader
+                        .read_exact(&mut payload)
+                        .await
+                        .expect("correlated read");
+                    assert_eq!(payload, [index; 32]);
+                });
+            });
+        }
+    });
+    // Kills: accepting a grant for the wrong operation/path or crossing leases.
+}
+
+#[test]
+fn cancellation_and_transport_faults_reclaim_waiters_and_preserve_unrelated_progress() {
+    let _stream_test = STREAM_TEST_LOCK.lock();
+    let harness = harness(2 << 20);
+    let data_plane = harness.bootstrap.data_plane.clone();
+    let cancelled_path = path("/runs/self/results/cancelled-open");
+
+    future::block_on(async {
+        let mut cancelled = Box::pin(data_plane.read_stream(&cancelled_path));
+        assert!(future::poll_once(cancelled.as_mut()).await.is_none());
+        drop(cancelled);
+        std::thread::sleep(Duration::from_millis(10));
+
+        let reader = data_plane.read_stream(&cancelled_path);
+        let writer = data_plane.write_stream(&cancelled_path);
+        let (reader, writer) = future::zip(reader, writer).await;
+        let mut reader = reader.expect("replacement reader");
+        let mut writer = writer.expect("replacement writer");
+        writer.close().await.expect("replacement close");
+        assert_eq!(reader.read().await.expect("replacement eof"), None);
+
+        let mut unrelated = data_plane
+            .open(&path("/models/second"), OpenOptions::read_only())
+            .await
+            .expect("unrelated progress");
+        let mut bytes = [0_u8; 11];
+        unrelated
+            .read_exact(&mut bytes)
+            .await
+            .expect("unrelated read");
+        assert_eq!(&bytes, b"second-blob");
+    });
+
+    let fault_harness = harness_with_transport(2 << 20, Arc::new(RejectingStreamTransport));
+    let fault_plane = fault_harness.bootstrap.data_plane.clone();
+    let fault_path = path("/runs/self/results/transport-fault");
+    future::block_on(async {
+        let mut reader_open = Box::pin(fault_plane.read_stream(&fault_path));
+        assert!(future::poll_once(reader_open.as_mut()).await.is_none());
+        let writer_error = match fault_plane.write_stream(&fault_path).await {
+            Ok(_) => panic!("faulted writer opened"),
+            Err(error) => error,
+        };
+        let reader_error = match reader_open.await {
+            Ok(_) => panic!("faulted reader opened"),
+            Err(error) => error,
+        };
+        assert!(matches!(
+            reader_error,
+            DataPlaneError::PeerLost | DataPlaneError::StreamFault(_)
+        ));
+        assert!(matches!(
+            writer_error,
+            DataPlaneError::PeerLost | DataPlaneError::StreamFault(_)
+        ));
+
+        let mut unrelated = fault_plane
+            .open(&path("/models/second"), OpenOptions::read_only())
+            .await
+            .expect("fault isolation");
+        let mut bytes = [0_u8; 11];
+        unrelated
+            .read_exact(&mut bytes)
+            .await
+            .expect("fault-isolated read");
+        assert_eq!(&bytes, b"second-blob");
+    });
+    // Kills: retain a cancelled waiter, complete only one matched endpoint,
+    // and propagate one descriptor fault into unrelated operations.
+}
+
+#[test]
+fn steady_state_blob_primitives_allocate_no_heap_memory() {
+    let harness = harness(2 << 20);
+    let data_plane = harness.bootstrap.data_plane.clone();
+    future::block_on(async {
+        let logical = path("/runs/self/results/allocation-count");
+        let mut writer = data_plane
+            .open(&logical, OpenOptions::staged_blob(64))
+            .await
+            .expect("allocation writer");
+        let payload = [7_u8; 64];
+        start_allocation_count();
+        let result = writer.write(&payload).await;
+        let write_allocations = finish_allocation_count();
+        assert_eq!(result.expect("allocation-count write"), payload.len());
+        assert_eq!(write_allocations, 0, "blob write primitive allocated");
+        writer.close().await.expect("allocation publish");
+
+        let mut reader = data_plane
+            .open(&logical, OpenOptions::read_only())
+            .await
+            .expect("allocation reader");
+        let mut destination = [0_u8; 64];
+        start_allocation_count();
+        let result = reader.read(&mut destination).await;
+        let read_allocations = finish_allocation_count();
+        assert_eq!(result.expect("allocation-count read"), destination.len());
+        assert_eq!(read_allocations, 0, "blob read primitive allocated");
+        assert_eq!(destination, payload);
+    });
+}

crates/data-plane/tests/namespace_guarantees.proptest-regressions

@@ -0,0 +1,7 @@
+# Seeds for failure cases proptest has generated in the past. It is
+# automatically read and these particular cases re-run before any
+# novel cases are generated.
+#
+# It is recommended to check this file in to source control so that
+# everyone who runs the test benefits from these saved cases.
+cc ebf4fc53222ac2fcd156f9e3f34dcde38fa3a79d784320c929ba2e2d43e6f9de # shrinks to actions = [89, 2, 155]

crates/data-plane/tests/namespace_guarantees.rs

@@ -115,6 +115,13 @@ fn namespace_mutations_are_linearizable_and_durable() {
             .await
             .expect("register first source");
         assert_eq!(registered.revision, 1);
+        let node = directory
+            .client
+            .lookup(logical.clone())
+            .await
+            .expect("lookup first blob node");
+        assert_eq!(node.kind, EntryKind::Blob);
+        assert_eq!(node.revision, registered.revision);
 
         let selected_first = directory
             .client
@@ -186,6 +193,12 @@ fn stream_rendezvous_is_symmetric_and_incarnations_are_isolated() {
             OperationId::from_u128(10),
         ));
         assert!(future::poll_once(source_open.as_mut()).await.is_none());
+        let node = directory
+            .client
+            .lookup(logical.clone())
+            .await
+            .expect("lookup ensured stream node");
+        assert_eq!(node.kind, EntryKind::Stream);
 
         let sink_match = directory
             .client
@@ -592,7 +605,7 @@ struct ModelBinding {
 #[derive(Clone, Debug)]
 enum TypedModelEntry {
     Blob(ModelBinding),
-    Stream(data_plane::namespace::StreamMatch),
+    Stream(Option<data_plane::namespace::StreamMatch>),
 }
 
 proptest! {
@@ -714,15 +727,16 @@ proptest! {
                     next_operation += 1;
                     let source_match = future::block_on(source_open).expect("source match");
                     prop_assert_eq!(&source_match, &sink_match);
-                    model.insert(logical, TypedModelEntry::Stream(sink_match));
+                    model.insert(logical, TypedModelEntry::Stream(Some(sink_match)));
                 }
                 2 => {
-                    if let Some(TypedModelEntry::Stream(binding)) = model.get(&logical) {
+                    if let Some(TypedModelEntry::Stream(binding)) = model.get_mut(&logical)
+                        && let Some(binding) = binding.take()
+                    {
                         future::block_on(directory.client.close_stream(
                             logical.clone(),
                             binding.incarnation,
                         )).expect("close current stream");
-                        model.remove(&logical);
                     }
                 }
                 _ => {

crates/telemetry/src/endpoint.rs

@@ -112,7 +112,7 @@ impl TelemetrySubscription {
     }
 
     pub fn drain_available(&self) -> Vec<TelemetryEvent> {
-        let mut out = Vec::new();
+        let mut out = Vec::with_capacity(self.rx.len());
         while let Ok(event) = self.rx.try_recv() {
             out.push(event);
         }
@@ -137,6 +137,31 @@ struct FanoutReport {
     disconnected: bool,
 }
 
+fn publish_to_target(
+    target: FanoutTarget,
+    events: impl IntoIterator<Item = TelemetryEvent>,
+) -> (usize, Option<FanoutReport>) {
+    let mut delivered = 0;
+    let mut dropped = 0;
+    let mut disconnected = false;
+    for event in events {
+        match target.tx.try_send(event) {
+            Ok(()) => delivered += 1,
+            Err(TrySendError::Full(_)) => dropped += 1,
+            Err(TrySendError::Disconnected(_)) => {
+                dropped += 1;
+                disconnected = true;
+            }
+        }
+    }
+    let report = (dropped > 0 || disconnected).then_some(FanoutReport {
+        id: target.id,
+        dropped,
+        disconnected,
+    });
+    (delivered, report)
+}
+
 struct FanoutState {
     next_id: u64,
     subscribers: BTreeMap<SubscriptionId, SubscriberSlot>,
@@ -236,11 +261,12 @@ impl DeliveryFanout {
         if events.is_empty() {
             return EndpointTick::default();
         }
+        let drained = events.len();
 
         // Snapshot sender handles while holding the subscriber map lock, then
         // deliver outside the lock so large batches or slow subscribers do not
         // block subscribe/snapshot control-plane operations.
-        let (targets, subscribers) = {
+        let (mut targets, subscribers) = {
             let state = self.state.lock().expect("telemetry fanout poisoned");
             let subscribers = state.subscribers.len();
             let targets = state
@@ -256,7 +282,7 @@ impl DeliveryFanout {
 
         if targets.is_empty() {
             return EndpointTick {
-                drained: events.len(),
+                drained,
                 subscribers: 0,
                 ..EndpointTick::default()
             };
@@ -264,27 +290,19 @@ impl DeliveryFanout {
 
         let mut delivered = 0;
         let mut reports = Vec::new();
+        let last_target = targets.pop().expect("nonempty fanout targets");
         for target in targets {
-            let mut dropped = 0;
-            let mut disconnected = false;
-            for event in &events {
-                match target.tx.try_send(event.clone()) {
-                    Ok(()) => delivered += 1,
-                    Err(TrySendError::Full(_)) => dropped += 1,
-                    Err(TrySendError::Disconnected(_)) => {
-                        dropped += 1;
-                        disconnected = true;
-                    }
-                }
-            }
-            if dropped > 0 || disconnected {
-                reports.push(FanoutReport {
-                    id: target.id,
-                    dropped,
-                    disconnected,
-                });
+            let (target_delivered, report) = publish_to_target(target, events.iter().cloned());
+            delivered += target_delivered;
+            if let Some(report) = report {
+                reports.push(report);
             }
         }
+        let (target_delivered, report) = publish_to_target(last_target, events);
+        delivered += target_delivered;
+        if let Some(report) = report {
+            reports.push(report);
+        }
 
         let dropped_for_subscribers = reports.iter().map(|report| report.dropped).sum::<u64>();
         if !reports.is_empty() {
@@ -302,7 +320,7 @@ impl DeliveryFanout {
         }
 
         EndpointTick {
-            drained: events.len(),
+            drained,
             delivered,
             dropped_for_subscribers: usize::try_from(dropped_for_subscribers).unwrap_or(usize::MAX),
             subscribers,

crates/telemetry/src/store.rs

@@ -143,6 +143,9 @@ impl Store {
 
     /// The stored stream for a node's life, creating an empty one if needed.
     pub fn stream_mut(&mut self, id: &StreamId) -> &mut StoredStream {
+        if self.streams.contains_key(id) {
+            return self.streams.get_mut(id).expect("stream checked as present");
+        }
         self.streams.entry(id.clone()).or_default()
     }
 

crates/transport/Cargo.toml

@@ -14,6 +14,7 @@ ed25519-dalek = { version = "2", features = ["std", "rand_core", "serde"] }
 rand_core = { version = "0.6", features = ["getrandom"] }
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
+rustc-hash = "2"
 
 [dev-dependencies]
 proptest = "1"

crates/transport/src/codec.rs

@@ -5,6 +5,7 @@
 //! per-type encoders/decoders so a type-erased message can be put on the wire
 //! and a [`WireEnvelope`] can be turned back into a concrete message.
 
+use rustc_hash::FxHashMap;
 use std::any::{Any, TypeId};
 use std::collections::HashMap;
 use std::sync::Arc;
@@ -150,7 +151,7 @@ type DecodeFn = Box<dyn Fn(&[u8]) -> Result<Box<dyn Any + Send>, Error> + Send +
 /// [`register_decoder`](Self::register_decoder) (variant-multiplexing encode,
 /// fan-in decode), then shared read-only via `Arc`.
 pub struct CodecRegistry {
-    encoders: HashMap<TypeId, EncodeFn>,
+    encoders: FxHashMap<TypeId, EncodeFn>,
     decoders: HashMap<String, DecodeFn>,
 }
 
@@ -163,7 +164,7 @@ impl Default for CodecRegistry {
 impl CodecRegistry {
     pub fn new() -> Self {
         Self {
-            encoders: HashMap::new(),
+            encoders: FxHashMap::default(),
             decoders: HashMap::new(),
         }
     }