Cargo.lock

@@ -947,7 +947,6 @@ dependencies = [
 name = "data-plane"
 version = "0.1.0"
 dependencies = [
- "criterion",
  "futures-lite",
  "libc",
  "parking_lot",
@@ -4349,7 +4348,6 @@ 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,13 +9,11 @@ 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, Descriptor, DescriptorMapping, MapRequest,
+    BlobWriter, DataPlane, DataPlaneBootstrap, StreamReader, StreamWriter, parse_actor_address,
-    MapTarget, Protection, Sharing, StreamReader, StreamWriter, parse_actor_address,
 };
 use data_plane::path::DataPath;
 use data_plane::protocol::{
-    AccessMode, BlobAllocation, BlobFailure, DataPlaneError, DescriptorKind, Errno, JobCapability,
+    BlobFailure, DataPlaneError, JobCapability, register_data_plane_codecs,
-    OpenOptions, register_data_plane_codecs,
 };
 use distribution::node::DistributedNodeConfig;
 use distribution::transport_bridge::{
@@ -25,8 +23,7 @@ use futures_lite::future;
 use iroh::{EndpointAddr, RelayMode};
 use iroh_driver::{IrohDriver, IrohDriverConfig};
 use parking_lot::Mutex as ParkingMutex;
-use pyo3::buffer::PyBuffer;
+use pyo3::exceptions::{PyBufferError, PyPermissionError, PyRuntimeError};
-use pyo3::exceptions::{PyBufferError, PyOSError, PyPermissionError, PyRuntimeError, PyValueError};
 use pyo3::ffi;
 use pyo3::prelude::*;
 use pyo3::types::{PyAny, PyBytes, PyModule};
@@ -39,44 +36,6 @@ 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);
@@ -91,8 +50,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, access } => {
+        DataPlaneError::Unauthorized { path, operation } => {
-            PyPermissionError::new_err(format!("{access:?} is not authorized for {path}"))
+            PyPermissionError::new_err(format!("{operation:?} is not authorized for {path}"))
         }
         DataPlaneError::PathNotFound(path) => {
             PyErr::new::<DataPathError, _>(format!("data path not found: {path}"))
@@ -110,29 +69,6 @@ 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")))
@@ -303,39 +239,6 @@ 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 {
@@ -391,258 +294,6 @@ 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,
@@ -1000,26 +651,6 @@ 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")]
@@ -1577,16 +1208,7 @@ 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,7 +4,6 @@ from __future__ import annotations
 
 import asyncio
 import ctypes
-import errno
 import json
 import os
 import struct
@@ -151,8 +150,7 @@ 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 {"open", "read_blob", "write_blob", "read_stream", "write_stream"} <= public
+        assert {"read_blob", "write_blob", "read_stream", "write_stream"} <= public
-        assert isinstance(swactor.O_RDONLY, int)
         assert not public & {
             "actor",
             "actor_id",
@@ -203,54 +201,6 @@ 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)
@@ -267,16 +217,14 @@ 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"
             )
-            buffer = bytearray(3)
+            while (chunk := await reader.read()) is not None:
-            while (count := await reader.readinto(buffer)) != 0:
+                received.append(chunk)
-                received.append(bytes(buffer[:count]))
 
         receiver = asyncio.create_task(receive())
         async with ctx.data.write_stream(
@@ -286,20 +234,8 @@ 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` — 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 /api/view/telemetry/live` — 2,000 frames per stream/channel, newest lifetime per node, all live streams plus 50 stale streams
 - `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,7 +173,6 @@ 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.
@@ -257,18 +256,13 @@ function pushUrl() {
 }
 
 async function refresh() {
-  if (refreshInFlight) return;
+  let data;
-  refreshInFlight = true;
   try {
     const response = await fetch('/api/view/fleet');
-    const data = await response.json();
+    data = await response.json();
+  } catch { return; }
   lastSnapshot = data;
   render();
-  } catch {
-    // Keep the last good snapshot while the next scheduled refresh retries.
-  } finally {
-    refreshInFlight = false;
-  }
 }
 
 function render() {
@@ -687,20 +681,15 @@ function stopDetailPolling() {
 
 function startDetailPolling() {
   stopDetailPolling();
-  let pollInFlight = false;
   const poll = async () => {
-    if (!selectedStream || !selectedActor || pollInFlight) return;
+    if (!selectedStream || !selectedActor) return;
-    pollInFlight = true;
+    let detail;
     try {
       const response = await fetch(`/api/view/fleet/detail?stream=${encodeURIComponent(selectedStream)}&actor=${encodeURIComponent(selectedActor)}`);
       if (!response.ok) { stopDetailPolling(); return; }
-      const detail = await response.json();
+      detail = await response.json();
+    } catch { return; }
     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,9 +9,7 @@ use crate::FrameEvent;
 use crate::view::DashboardView;
 
 const LIVE_EXPLORER_HTML: &str = include_str!("live_explorer_page.html");
-const FRAME_HISTORY_CAP: usize = 500;
+const FRAME_HISTORY_CAP: usize = 2_000;
-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;
 
@@ -67,39 +65,12 @@ 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())
-            .map(|frames| frames.iter().rev())
+            .flat_map(|frames| frames.iter())
             .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
@@ -167,16 +138,6 @@ 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");
@@ -190,7 +151,11 @@ mod tests {
             .collect::<Vec<_>>();
 
         assert_eq!(quiet.len(), 1, "a quiet channel remains discoverable");
-        assert_eq!(busy.len(), SNAPSHOT_FRAME_CAP - quiet.len());
+        assert_eq!(busy.len(), FRAME_HISTORY_CAP);
+        assert_eq!(
+            busy.first().and_then(|frame| frame["position"].as_u64()),
+            Some(1)
+        );
         assert_eq!(
             busy.last().and_then(|frame| frame["position"].as_u64()),
             Some(FRAME_HISTORY_CAP as u64)
@@ -236,37 +201,6 @@ 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,8 +147,6 @@ 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 = {
@@ -525,11 +523,6 @@ 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 {
@@ -561,12 +554,6 @@ 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,15 +18,9 @@ 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

@@ -1,81 +0,0 @@
-#![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,8 +7,6 @@ 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};
 
@@ -253,60 +251,27 @@ impl Blob {
         self.guard.lease
     }
 
-    pub(crate) fn copy_at(&self, offset: u64, destination: &mut [u8]) -> Result<usize, BlobError> {
+    pub fn map(&self) -> Result<BlobView, 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
+        let payload_offset = self.guard.lease.offset.checked_add(BLOB_HEADER_LEN).ok_or(
-            .guard
+            BlobError::RangeOutOfBounds {
-            .lease
+                offset: self.guard.lease.offset,
-            .offset
+                length: self.guard.lease.length,
-            .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, length)?;
+        )?;
+        let range = mapped_range(&self.guard.arena, payload_offset, self.guard.lease.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 {
@@ -350,17 +315,12 @@ 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 {
@@ -379,7 +339,6 @@ impl WritableBlobLease {
             metadata,
             active_view: AtomicBool::new(false),
             finished: AtomicBool::new(false),
-            view_observer: Mutex::new(None),
         }))
     }
 
@@ -391,111 +350,23 @@ impl WritableBlobLease {
         &self.metadata
     }
 
-    pub(crate) fn set_view_observer(&self, observer: Arc<dyn WritableViewObserver>) {
+    pub(crate) fn map(self: &Arc<Self>) -> Result<WritableArenaView, BlobError> {
-        *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
+        let payload_offset =
-            .lease
+            self.lease
                 .offset
                 .checked_add(BLOB_HEADER_LEN)
-            .and_then(|start| start.checked_add(offset))
                 .ok_or(BlobError::RangeOutOfBounds {
-                offset,
+                    offset: self.lease.offset,
-                length,
+                    length: self.lease.length,
                     arena_size: self.arena.len() as u64,
                 })?;
-        let range = match mapped_range(&self.arena, payload_offset, length) {
+        let range = match mapped_range(&self.arena, payload_offset, self.lease.length) {
             Ok(range) => range,
             Err(error) => {
                 self.active_view.store(false, Ordering::Release);
@@ -508,9 +379,6 @@ 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) {
@@ -615,10 +483,6 @@ 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,7 +69,6 @@ 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;
 
@@ -129,30 +128,6 @@ 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> {
@@ -582,15 +557,6 @@ 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,
@@ -606,7 +572,6 @@ fn lease_ring(
         (1, Some(ArenaEvent::RingLeaseRejected { reason, .. })) => {
             Err(InstallError::LeaseRejected(reason))
         }
-
         (1, Some(ArenaEvent::RingLeaseQueued { .. })) => Err(InstallError::LeaseQueued),
         _ => Err(InstallError::UnexpectedLeaseOutcome),
     }
@@ -621,13 +586,6 @@ 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.
@@ -806,25 +764,6 @@ 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;
@@ -866,14 +805,6 @@ 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)?;
@@ -1065,97 +996,6 @@ 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,12 +39,6 @@ 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,
@@ -70,7 +64,6 @@ pub struct StreamMatch {
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub enum NamespaceError {
     PathNotFound(DataPath),
-    PathExists(DataPath),
     WrongEntryType {
         path: DataPath,
         expected: EntryKind,
@@ -96,7 +89,6 @@ 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,
@@ -148,7 +140,6 @@ pub enum DataDirectoryIn {
         length: u64,
         recovery: SourceRecovery,
         operation_id: OperationId,
-        reservation: Option<OperationId>,
         reply_to: ActorAddress,
     },
     Resolve {
@@ -156,23 +147,6 @@ 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,
@@ -186,8 +160,6 @@ pub enum DataDirectoryIn {
         descriptor: Vec<u8>,
         endpoint: ActorAddress,
         replace: bool,
-        ensure: bool,
-        expected_revision: Option<u64>,
         operation_id: OperationId,
         reply_to: ActorAddress,
     },
@@ -221,21 +193,6 @@ 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,
@@ -257,9 +214,6 @@ 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,
@@ -275,22 +229,10 @@ 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,
@@ -301,8 +243,6 @@ pub enum NamespaceRequest {
         endpoint: ActorAddress,
         descriptor: Vec<u8>,
         replace: bool,
-        ensure: bool,
-        expected_revision: Option<u64>,
         operation_id: OperationId,
     },
     CloseStream {
@@ -319,11 +259,6 @@ pub enum NamespaceClientIn {
     Cancel {
         reply_to: ActorAddress,
     },
-    CancelBlobReservation {
-        path: DataPath,
-        operation_id: OperationId,
-        reply_to: ActorAddress,
-    },
     DirectoryReply(DataDirectoryOut),
     TransferFailed {
         destination: ActorAddress,
@@ -353,7 +288,6 @@ struct PendingStream {
     descriptor: Vec<u8>,
     reply_to: ActorAddress,
     revision: u64,
-    opened_from_revision: Option<u64>,
 }
 
 struct StreamOpenRequest {
@@ -362,8 +296,6 @@ struct StreamOpenRequest {
     role: StreamRole,
     endpoint: ActorAddress,
     replace: bool,
-    ensure: bool,
-    expected_revision: Option<u64>,
     descriptor: Vec<u8>,
     operation_id: OperationId,
     reply_to: ActorAddress,
@@ -384,7 +316,6 @@ pub struct DataDirectoryActor {
     store: NamespaceStore,
     sources: BTreeMap<DataPath, RuntimeSource>,
     streams: BTreeMap<DataPath, RuntimeStream>,
-    blob_reservations: BTreeMap<DataPath, OperationId>,
     authority_epoch: u64,
 }
 
@@ -408,7 +339,6 @@ impl DataDirectoryActor {
             store,
             sources,
             streams: BTreeMap::new(),
-            blob_reservations: BTreeMap::new(),
             authority_epoch,
         })
     }
@@ -445,9 +375,6 @@ 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;
@@ -461,7 +388,6 @@ 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)
         {
@@ -514,42 +440,10 @@ 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
         {
@@ -689,7 +583,6 @@ impl DataDirectoryActor {
                         request_id,
                         reply_to,
                         revision: receipt.revision,
-                        opened_from_revision: (!ensure).then_some(expected_revision).flatten(),
                     }),
                 );
             }
@@ -727,41 +620,6 @@ 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,
@@ -769,7 +627,6 @@ impl DataDirectoryActor {
         length: u64,
         recovery: SourceRecovery,
         operation_id: OperationId,
-        reservation: Option<OperationId>,
         retired: Option<ActorAddress>,
     ) -> Result<MutationReceipt, NamespaceError> {
         let request = MutationRequest::Register {
@@ -780,12 +637,6 @@ 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)
@@ -794,7 +645,6 @@ 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 {
@@ -816,15 +666,12 @@ 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.store.snapshot().stream_nodes.contains_key(path) {
+        if self.streams.contains_key(path) {
             return Err(NamespaceError::WrongEntryType {
                 path: path.clone(),
                 expected: EntryKind::Blob,
@@ -852,43 +699,17 @@ 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,
@@ -911,7 +732,6 @@ 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)
         {
@@ -949,7 +769,6 @@ impl ActorInterface for DataDirectoryActor {
                 length,
                 recovery,
                 operation_id,
-                reservation,
                 reply_to,
             } => {
                 let logical = path.clone();
@@ -961,15 +780,7 @@ impl ActorInterface for DataDirectoryActor {
                         RuntimeSource::Available(actor) if *actor != source => Some(*actor),
                         RuntimeSource::Available(_) | RuntimeSource::Unavailable(_) => None,
                     });
-                let result = self.register(
+                let result = self.register(path, source, length, recovery, operation_id, retired);
-                    path,
-                    source,
-                    length,
-                    recovery,
-                    operation_id,
-                    reservation,
-                    retired,
-                );
                 if result.is_ok()
                     && let Some(retired) = retired
                 {
@@ -1002,53 +813,6 @@ 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,
@@ -1100,8 +864,6 @@ impl ActorInterface for DataDirectoryActor {
                 endpoint,
                 descriptor,
                 replace,
-                ensure,
-                expected_revision,
                 operation_id,
                 reply_to,
             } => self.open_stream(
@@ -1112,8 +874,6 @@ impl ActorInterface for DataDirectoryActor {
                     role,
                     endpoint,
                     replace,
-                    ensure,
-                    expected_revision,
                     descriptor,
                     operation_id,
                     reply_to,
@@ -1188,7 +948,6 @@ impl NamespaceClientActor {
                 length,
                 recovery,
                 operation_id,
-                reservation,
             } => DataDirectoryIn::Register {
                 request_id,
                 path: path.clone(),
@@ -1196,7 +955,6 @@ impl NamespaceClientActor {
                 length: *length,
                 recovery: recovery.clone(),
                 operation_id: *operation_id,
-                reservation: *reservation,
                 reply_to: ctx.self_addr(),
             },
             NamespaceRequest::Resolve { path } => DataDirectoryIn::Resolve {
@@ -1204,25 +962,6 @@ 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(),
@@ -1235,8 +974,6 @@ impl NamespaceClientActor {
                 endpoint,
                 descriptor,
                 replace,
-                ensure,
-                expected_revision,
                 operation_id,
             } => DataDirectoryIn::OpenStream {
                 request_id,
@@ -1245,8 +982,6 @@ impl NamespaceClientActor {
                 endpoint: *endpoint,
                 descriptor: descriptor.clone(),
                 replace: *replace,
-                ensure: *ensure,
-                expected_revision: *expected_revision,
                 operation_id: *operation_id,
                 reply_to: ctx.self_addr(),
             },
@@ -1323,25 +1058,6 @@ 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);
@@ -1439,7 +1155,6 @@ impl NamespaceClient {
                 length,
                 recovery,
                 operation_id,
-                reservation: None,
             })
             .await?
         {
@@ -1459,15 +1174,6 @@ 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,
@@ -1498,8 +1204,6 @@ impl NamespaceClient {
                 role,
                 endpoint,
                 replace,
-                ensure: true,
-                expected_revision: None,
                 descriptor: Vec::new(),
                 operation_id,
             })
@@ -1631,7 +1335,6 @@ impl DirectoryClient {
                     length,
                     recovery,
                     operation_id,
-                    reservation: None,
                     reply_to: *inbox.addr(),
                 },
             )
@@ -1667,29 +1370,6 @@ 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,
@@ -1746,8 +1426,6 @@ 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,8 +108,6 @@ 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>,
@@ -123,7 +121,6 @@ impl Default for NamespaceSnapshot {
             next_revision: 1,
             bindings: BTreeMap::new(),
             operations: BTreeMap::new(),
-            stream_nodes: BTreeMap::new(),
             retirements: Vec::new(),
         }
     }
@@ -146,10 +143,6 @@ 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,12 +6,10 @@ use serde::{Deserialize, Serialize};
 use swactor::actor::ActorAddress;
 use swactor_transport::{CodecRegistry, JsonCodec, NetworkMessage};
 
-use crate::blob::{BlobError, BlobLease, BlobMetadata, ContentDigest};
+use crate::blob::{BlobError, BlobLease, BlobMetadata};
 use crate::byte_ring::{RingHandle, Role};
 use crate::ids::BlobLeaseId;
-use crate::namespace::{
+use crate::namespace::{EntryKind, NamespaceError, StreamIncarnation, StreamMatch};
-    EntryKind, NamespaceError, NamespaceNode, OperationId, StreamIncarnation, StreamMatch,
-};
 use crate::path::DataPath;
 use crate::stream_transport::{StreamPeerDescriptor, StreamTransportEvent};
 
@@ -47,155 +45,11 @@ impl JobCapability {
 }
 
 #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
-pub enum AccessMode {
+pub enum DataOperation {
-    ReadOnly,
+    ReadBlob,
-    WriteOnly,
+    WriteBlob,
-    ReadWrite,
+    ReadStream,
-}
+    WriteStream,
-
-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)]
@@ -244,27 +98,19 @@ 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,
-        access: AccessMode,
+        operation: DataOperation,
     },
     PathNotFound(DataPath),
-    PathExists(DataPath),
     SourceFailure(String),
     ArenaExhausted,
     Blob(BlobFailure),
     OperationCancelled,
-    BadDescriptor,
-    Unsupported(String),
-    MappingUnsupported,
-    Busy(String),
-    Stale(String),
-    BrokenPipe,
     WrongEntryType {
         path: DataPath,
         expected: EntryKind,
@@ -280,26 +126,18 @@ 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, access } => {
+            Self::Unauthorized { path, operation } => {
-                write!(f, "{access:?} access is not authorized for {path}")
+                write!(f, "{operation:?} 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,
@@ -316,44 +154,6 @@ 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 {
@@ -370,24 +170,33 @@ pub enum HostSessionIn {
         job_capability: JobCapability,
         child_node: Option<[u8; 32]>,
     },
-    Open {
+    OpenReadBlob {
         path: DataPath,
-        options: OpenOptions,
-        policy: OpenPolicy,
         child_session: ActorAddress,
         operation: ActorAddress,
     },
-    OpenResolved {
+    CancelReadBlob {
         operation: ActorAddress,
-        result: Result<NamespaceNode, NamespaceError>,
     },
-    BlobReserved {
+    OpenWriteBlob {
-        operation: ActorAddress,
         path: DataPath,
-        reservation: OperationId,
+        length: u64,
-        result: Result<(), NamespaceError>,
+        child_session: ActorAddress,
+        operation: ActorAddress,
     },
-    CancelOpen {
+    OpenReadStream {
+        path: DataPath,
+        child_session: ActorAddress,
+        operation: ActorAddress,
+        replace: bool,
+    },
+    OpenWriteStream {
+        path: DataPath,
+        child_session: ActorAddress,
+        operation: ActorAddress,
+        replace: bool,
+    },
+    CancelStream {
         operation: ActorAddress,
     },
     StreamControl {
@@ -444,13 +253,16 @@ pub enum ChildSessionIn {
         error: DataPlaneError,
     },
     AttachmentDeadline,
-    Open {
+    ReadBlob {
         path: DataPath,
-        options: OpenOptions,
-        policy: OpenPolicy,
         reply_to: ActorAddress,
     },
-    CancelOpen {
+    CancelRead {
+        reply_to: ActorAddress,
+    },
+    OpenWriteBlob {
+        path: DataPath,
+        length: u64,
         reply_to: ActorAddress,
     },
     OpenReadStream {
@@ -458,6 +270,11 @@ pub enum ChildSessionIn {
         reply_to: ActorAddress,
         replace: bool,
     },
+    OpenWriteStream {
+        path: DataPath,
+        reply_to: ActorAddress,
+        replace: bool,
+    },
     CancelStream {
         reply_to: ActorAddress,
     },
@@ -539,10 +356,6 @@ 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,6 +1,357 @@
 #![cfg(target_os = "linux")]
 
-include!("data_plane_test_support.inc");
+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) {}
+}
 
 #[test]
 fn attachment_without_a_host_reply_fails_on_actor_deadline() {
@@ -523,190 +874,3 @@ 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,73 +557,3 @@ 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

@@ -1,364 +0,0 @@
-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

@@ -1,612 +0,0 @@
-#![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

@@ -1,7 +0,0 @@
-# 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,13 +115,6 @@ 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
@@ -193,12 +186,6 @@ 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
@@ -605,7 +592,7 @@ struct ModelBinding {
 #[derive(Clone, Debug)]
 enum TypedModelEntry {
     Blob(ModelBinding),
-    Stream(Option<data_plane::namespace::StreamMatch>),
+    Stream(data_plane::namespace::StreamMatch),
 }
 
 proptest! {
@@ -727,16 +714,15 @@ 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(Some(sink_match)));
+                    model.insert(logical, TypedModelEntry::Stream(sink_match));
                 }
                 2 => {
-                    if let Some(TypedModelEntry::Stream(binding)) = model.get_mut(&logical)
+                    if let Some(TypedModelEntry::Stream(binding)) = model.get(&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::with_capacity(self.rx.len());
+        let mut out = Vec::new();
         while let Ok(event) = self.rx.try_recv() {
             out.push(event);
         }
@@ -137,31 +137,6 @@ 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>,
@@ -261,12 +236,11 @@ 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 (mut targets, subscribers) = {
+        let (targets, subscribers) = {
             let state = self.state.lock().expect("telemetry fanout poisoned");
             let subscribers = state.subscribers.len();
             let targets = state
@@ -282,7 +256,7 @@ impl DeliveryFanout {
 
         if targets.is_empty() {
             return EndpointTick {
-                drained,
+                drained: events.len(),
                 subscribers: 0,
                 ..EndpointTick::default()
             };
@@ -290,18 +264,26 @@ 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 (target_delivered, report) = publish_to_target(target, events.iter().cloned());
+            let mut dropped = 0;
-            delivered += target_delivered;
+            let mut disconnected = false;
-            if let Some(report) = report {
+            for event in &events {
-                reports.push(report);
+                match target.tx.try_send(event.clone()) {
+                    Ok(()) => delivered += 1,
+                    Err(TrySendError::Full(_)) => dropped += 1,
+                    Err(TrySendError::Disconnected(_)) => {
+                        dropped += 1;
+                        disconnected = true;
                     }
                 }
-        let (target_delivered, report) = publish_to_target(last_target, events);
+            }
-        delivered += target_delivered;
+            if dropped > 0 || disconnected {
-        if let Some(report) = report {
+                reports.push(FanoutReport {
-            reports.push(report);
+                    id: target.id,
+                    dropped,
+                    disconnected,
+                });
+            }
         }
 
         let dropped_for_subscribers = reports.iter().map(|report| report.dropped).sum::<u64>();
@@ -320,7 +302,7 @@ impl DeliveryFanout {
         }
 
         EndpointTick {
-            drained,
+            drained: events.len(),
             delivered,
             dropped_for_subscribers: usize::try_from(dropped_for_subscribers).unwrap_or(usize::MAX),
             subscribers,

crates/telemetry/src/store.rs

@@ -143,9 +143,6 @@ 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,7 +14,6 @@ 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,7 +5,6 @@
 //! 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;
@@ -151,7 +150,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: FxHashMap<TypeId, EncodeFn>,
+    encoders: HashMap<TypeId, EncodeFn>,
     decoders: HashMap<String, DecodeFn>,
 }
 
@@ -164,7 +163,7 @@ impl Default for CodecRegistry {
 impl CodecRegistry {
     pub fn new() -> Self {
         Self {
-            encoders: FxHashMap::default(),
+            encoders: HashMap::new(),
             decoders: HashMap::new(),
         }
     }