diff --git a/Cargo.lock b/Cargo.lock index f364649..24c47da 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3295,6 +3295,19 @@ dependencies = [ "unindent", ] +[[package]] +name = "pyo3-async-runtimes" +version = "0.23.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "977dc837525cfd22919ba6a831413854beb7c99a256c03bf8624ad707e45810e" +dependencies = [ + "futures", + "once_cell", + "pin-project-lite", + "pyo3", + "tokio", +] + [[package]] name = "pyo3-build-config" version = "0.23.5" @@ -3344,9 +3357,13 @@ dependencies = [ name = "python" version = "0.1.0" dependencies = [ + "data-plane", + "libc", "pyo3", + "pyo3-async-runtimes", "swactor", "swactor-engine", + "tokio", ] [[package]] diff --git a/DATA_PLANE_PATH_API_HANDOFF.md b/DATA_PLANE_PATH_API_HANDOFF.md new file mode 100644 index 0000000..1c4a8ac --- /dev/null +++ b/DATA_PLANE_PATH_API_HANDOFF.md @@ -0,0 +1,296 @@ +# Data-plane path API handoff + +## Purpose + +Pause-point for the first Myelin GPU job flow. We proved the current end-to-end path, then identified that its temporary Python/data-plane boundary is the wrong long-term abstraction. + +Resume by working backward from the approved namespace/path API and designing fundamental path resolution before changing the binding again. + +## What was accomplished + +### Working manual GPU flow + +The repository now has a manually exercised path through: + +1. Myelin orchestrator startup. +2. Local Docker GPU-node provisioning from the Provision panel. +3. Runtime-image selection during provisioning. +4. Fleet-panel GPU-job submission. +5. Model-weight transfer over the data-plane edge transport. +6. Tinygrad CUDA inference in the worker job image. +7. Result transfer back to the orchestrator and bare Fleet display. + +The observed deterministic result was: + +```text +CUDA -> [2.75, -8.75] +``` + +The flow was repeated after worker teardown/reprovisioning and produced the same result. + +### Runtime and deterministic fixture + +- `apps/myelin/node-image/Dockerfile.job-tinygrad` builds the small Tinygrad CUDA job image. +- The selected image is carried through the existing provisioning request/spec and shown by the existing UI. +- `apps/myelin/jobs/tiny_linear.weights` is a 24-byte, six-`f32` fixture: + +```text +matrix = [[1.5, -2.0], + [0.5, 4.0]] +bias = [0.25, -0.75] +input = [2.0, -1.0] +output = [2.75, -8.75] +``` + +- `apps/myelin/jobs/tiny_linear_inference.py` performs GPU-only inference and rejects non-CUDA Tinygrad selection. + +### Job/data-plane integration + +The current implementation connects job-runner lifecycle to real Iroh edge streams and the Myelin Fleet probe. Work completed during this pass included: + +- model ingress and result egress over `EDGE_ALPN` rather than actor-message payloads; +- reverse-route assignment acknowledgement so a fast job-exit event is not lost; +- edge-stream completion acknowledgement before sender teardown; +- explicit stream-fault propagation into Fleet job failure; +- result framing/parsing and deterministic Fleet result display; +- job-runner assignment/lifecycle coverage for the acknowledgement ordering. + +### Verification already performed + +At the completed checkpoint: + +- `swactor-job-runner`: 12 focused tests passed; +- `iroh-driver`: 14 tests passed; +- Python binding built successfully with stable-ABI forward compatibility enabled for the workstation Python version; +- Myelin test targets compiled; +- final Docker node and Tinygrad job images built; +- the UI-driven local Docker GPU scenario returned `[2.75, -8.75]`; +- temporary worker containers, orchestrators, state directories, and debug sessions were removed; built images were intentionally retained. + +## The approved Python API direction + +Python names the data it wants. Swactor does not declare application ports or application data types. + +```python +async def main(ctx: swactor.Context) -> None: + weights = await ctx.data.read_blob( + "/models/tiny-linear/weights", + ) + + async with ctx.data.write_stream( + "/runs/self/results/inference", + ) as results: + await results.write(b"opaque application bytes") + + +swactor.run(main) +``` + +The intended binding primitives are: + +```python +ctx.data.read_blob(path) # finite immutable bytes +ctx.data.write_blob(path) # finite output +ctx.data.read_stream(path) # incrementally readable bytes +ctx.data.write_stream(path) # incrementally writable bytes +``` + +Public concepts should remain limited to: + +```text +Context +DataPlane +Blob +ArenaView +StreamReader +StreamWriter +``` + +### Boundary + +Swactor may understand: + +- logical names and paths; +- blob versus stream; +- byte length and optional content digest; +- stream ordering and termination; +- arena leases, alignment, generations, cursors, and backpressure; +- authorization, placement, routing, and lifecycle. + +Swactor must not understand: + +- tensors; +- dtypes or shapes; +- byte order as application schema; +- JSON or JSON schemas; +- model formats; +- inference-result semantics. + +Therefore there should be no `swactor.Tensor`, `swactor.Json`, typed job-port manifest, or orchestrator-provided application port declaration. + +Application code interprets bytes. For the fixed model: + +```python +weights_blob = await ctx.data.read_blob("/models/tiny-linear/weights") + +expected_bytes = struct.calcsize("<6f") +if weights_blob.length != expected_bytes: + raise ValueError( + f"expected {expected_bytes} model bytes, " + f"received {weights_blob.length}" + ) + +with weights_blob.map() as mapped: + weights = struct.unpack_from("<6f", mapped) +``` + +Tinygrad owns tensor construction and the host-to-GPU upload. JSON encoding of results likewise remains Python logic: + +```python +encoded = json.dumps( + {"device": device, "output": output}, + separators=(",", ":"), + sort_keys=True, +).encode("utf-8") + +async with ctx.data.write_stream( + "/runs/self/results/inference" +) as results: + await results.write(encoded) +``` + +A stream is opaque ordered bytes. If the application wants records, JSON Lines, protobuf, or another framing scheme, the application supplies it. + +## Shared-memory direction + +Unix sockets are rejected for the Python/data-plane boundary. + +The data-plane already has the correct foundation: + +- `ArenaManager::arena_fd()` exposes Linux shared backing; +- arena leases are offset-based rather than process-pointer-based; +- ring identity, generation, reservation, commit, consume, and wake contracts already exist; +- `EdgeRuntime` already composes network edges, arena leases, and worker loading. + +The intended local flow is: + +```text +remote edge -> ingress arena ring -> Python mapped view +Python bytes -> egress arena ring -> remote edge +``` + +Python should inherit and map the arena backing. Async operations wait on shared ring state plus a notification primitive; payloads remain in shared memory. + +### Current implementation gaps + +These are implementation gaps, not reasons to add another transport: + +- `ArenaBacking` exposes `read_at`/`write_at`, but the binding does not yet expose an `mmap` view. +- `RingHelperHarness` still uses a process-local `Vec`; production ring headers and atomic cursors must live in arena memory. +- `LayoutPointer::NoProcessPointer` is not yet resolved into a mapped process-local view. +- Python process bootstrap does not yet receive the arena and wake descriptors plus its initial control-ring layouts. +- The Python binding currently contains Tokio `UnixStream` input/output classes. They are temporary and should be removed. +- `EmbeddedJobDataPlane` currently bridges through local Unix sockets. That is also temporary. +- The Tinygrad test bridge currently reads arena data as a file and uses line-delimited JSON control. The final binding should map the arena directly. +- The first honest GPU path will have one arena-to-CUDA upload performed by Tinygrad. Direct device-handle integration is separate from path resolution and is not required to design the namespace API. +## Bootstrap slice (implemented) + +The first unknown down the approved API chain — what `swactor.run(main)` +consumes before `main(ctx)` can exist — is now implemented and tested. + +Contract: + +- The host execs the job process with two inherited descriptors named by + `SWACTOR_ARENA_FD` and `SWACTOR_WAKE_FD`; nothing else crosses the + boundary at spawn. +- Arena offset 0 holds a fixed 48-byte little-endian bootstrap header + (`SWBS` magic, version, reserved-zero fields, arena size, control-ring + offset/capacity/generation). Canonical definition: + `crates/data-plane/src/bootstrap.rs` — header layout, shared parser, + `write_bootstrap` host writer (header region and control ring are + disjoint leases under the arena's own placement law; all writes complete + before the handoff is returned). +- The binding's `run()` maps the arena read-only, takes ground truth from + `fstat` (never the header's size claim), validates via the shared + parser, arms `FD_CLOEXEC` on the wake descriptor, closes the arena + descriptor after mapping, builds `Context`/`DataPlane`, and drives + `main` on asyncio. Fail-fast: any defect raises `swactor.BootstrapError` + (`swactor.SwactorError` subclass) before `main` is invoked. + +Removed with this slice: the binding's Tokio `UnixStream` +`DataPlaneInput`/`DataPlaneOutput` classes and the +`SWACTOR_DATA_PLANE_INPUT/OUTPUT` environment variables. + +Verification (all passing): + +- `data-plane`: 12 bootstrap guarantee tests (parser defect table, fuzzed + pages, writer round-trip, ring zero/stamp, env ABI, fd shape). +- Python binding: 29 tests (`crates/bindings/python/tests/`) covering the + defect table in-process and across real exec boundaries, wake opacity, + CLOEXEC grandchild isolation, exit-code contract, and surface leakage. +- `iroh-driver`, `myelin`, `swactor-job-runner` still compile; all + pre-existing suites pass. + +Deliberately unchanged: `EmbeddedJobDataPlane` still bridges weights/results +over local Unix sockets, and its `configure()` still emits the legacy socket + environment — the binding no longer reads it. That bridge is replaced when + the arena carries real path data (next slice: control-ring framing for + `read_blob`), so the proven GPU scenario is not left half-cut-over. + + +## Where design resumes: fundamental path resolution + +The next work starts with path semantics and resolution, not binding implementation. + +Two layers must remain distinct: + +```text +logical path + -> blob owner or matched stream participants + -> actor addresses + -> current nodes + -> live transport route + -> provisioned edge and arena rings +``` + +The candidate direction discussed, but not yet finalized, is: + +1. A data-directory authority resolves canonical paths. +2. Blob paths resolve to a current owner and pinned generation. +3. Stream paths rendezvous one egress participant with one ingress participant for v1. +4. Existing swactor actor-directory state resolves participant actors to their current nodes. +5. The Iroh driver resolves nodes to live direct/relay transport routes. +6. Edge provisioning allocates an `EdgeId`, installs ingress/egress arena rings, and only then opens the byte stream. +7. Python sees none of the actor, node, endpoint, edge, or ring identities. + +The concrete scenario to use while settling the design: + +```text +/models/tiny-linear/weights + blob read by the worker job + +/runs/self/results/inference + live stream written by the worker job + `self` scoped to the current run +``` + +Questions to settle first when work resumes: + +- Path grammar, normalization, and scoped aliases such as `self`. +- Who owns the root namespace and whether authority delegates by prefix. +- How a blob publisher binds, updates, and tombstones a path generation. +- How stream readers and writers register, wait, match, cancel, and close. +- How path capabilities are attached to the job context and enforced. +- Which actor owns edge-plan creation and the ordering between ring installation and transport connection. +- Failure behavior when a path is absent, an owner moves, a participant exits, or a generation changes during resolution. + +Do not redesign the UI, job specification, model format, or inference logic while resolving these fundamentals. Work backward from the approved Python API and preserve the already-proven manual GPU scenario as the integration target. + +## Explicitly rejected directions + +- Orchestration declaring named application ports for Python. +- Static job-port manifests. +- Actor IDs, edge IDs, peers, or socket paths exposed to Python. +- Per-port Unix sockets or a separate local control socket. +- Swactor-owned tensor, JSON, model, or inference schemas. +- Embedding application semantics into the data-plane. diff --git a/apps/myelin/jobs/tiny_linear_inference.py b/apps/myelin/jobs/tiny_linear_inference.py new file mode 100755 index 0000000..cbbb228 --- /dev/null +++ b/apps/myelin/jobs/tiny_linear_inference.py @@ -0,0 +1,44 @@ +#!/usr/bin/env python3 +"""Deterministic GPU-only linear inference over swactor data-plane paths.""" + +from __future__ import annotations + +import json +import struct + +import swactor +from tinygrad import Device, Tensor + +_WEIGHT_COUNT = 6 +_EXPECTED_WEIGHT_BYTES = _WEIGHT_COUNT * 4 +_INPUT = [2.0, -1.0] + + +async def main(ctx: swactor.Context) -> None: + weights_blob = await ctx.data.read_blob("/models/tiny-linear/weights") + if weights_blob.length != _EXPECTED_WEIGHT_BYTES: + raise ValueError( + f"expected {_EXPECTED_WEIGHT_BYTES} model bytes, " + f"received {weights_blob.length}" + ) + with weights_blob.map() as mapped: + weights = struct.unpack_from("<6f", mapped) + + matrix = Tensor(weights[:4]).reshape(2, 2) + bias = Tensor(weights[4:]) + output = (Tensor(_INPUT).reshape(1, 2) @ matrix + bias).realize() + device = Device.DEFAULT + if not device.startswith("CUDA"): + raise RuntimeError(f"GPU required; tinygrad selected {device}") + + payload = json.dumps( + {"device": device, "output": output.tolist()[0]}, + separators=(",", ":"), + sort_keys=True, + ).encode("utf-8") + async with ctx.data.write_stream("/runs/self/results/inference") as results: + await results.write(payload) + + +if __name__ == "__main__": + swactor.run(main) diff --git a/crates/bindings/python/Cargo.toml b/crates/bindings/python/Cargo.toml index bde3cc8..fc569ec 100644 --- a/crates/bindings/python/Cargo.toml +++ b/crates/bindings/python/Cargo.toml @@ -11,4 +11,8 @@ crate-type = ["cdylib"] [dependencies] swactor = { path = "../../.." } swactor-engine = { path = "../../engine", default-features = false } +data-plane = { path = "../../data-plane" } +libc = "0.2" pyo3 = { version = "0.23", features = ["extension-module"] } +pyo3-async-runtimes = { version = "0.23", features = ["tokio-runtime"] } +tokio.workspace = true diff --git a/crates/bindings/python/pyproject.toml b/crates/bindings/python/pyproject.toml index e2e5ac9..37fe2a7 100644 --- a/crates/bindings/python/pyproject.toml +++ b/crates/bindings/python/pyproject.toml @@ -8,7 +8,7 @@ version = "0.1.0" requires-python = ">=3.9" [dependency-groups] -dev = ["jupyter", "ipykernel"] +dev = ["jupyter", "ipykernel", "pytest"] [tool.uv] cache-keys = [ diff --git a/crates/bindings/python/src/lib.rs b/crates/bindings/python/src/lib.rs index 311cdf2..82d5fee 100644 --- a/crates/bindings/python/src/lib.rs +++ b/crates/bindings/python/src/lib.rs @@ -1,5 +1,8 @@ use std::any::Any; use std::cell::RefCell; +use std::fs::File; +use std::os::fd::{AsRawFd, FromRawFd, OwnedFd}; +use std::sync::Arc; use pyo3::prelude::*; use pyo3::types::PyModule; @@ -11,6 +14,8 @@ use ::swactor::config::RuntimeConfig; use ::swactor::runtime::{Inbox, Runtime, RuntimeParts}; use swactor_engine::{Engine, SteppingBackend}; +use data_plane::bootstrap as dp_bootstrap; + // ─── PyMsg newtype ─────────────────────────────────────────────────────────── /// Newtype around `PyObject` that implements `Clone + Send + Sync`. @@ -445,9 +450,188 @@ fn build_stats(runtime: &Runtime) -> PyRuntimeStats { } } +// ─── Job entrypoint: swactor.run ───────────────────────────────────────────── + +pyo3::create_exception!(swactor, SwactorError, pyo3::exceptions::PyException); +pyo3::create_exception!(swactor, BootstrapError, SwactorError); + +/// Read-only shared mapping of the inherited arena. +/// +/// The mapping is immutable for the process lifetime once created, which is +/// what makes the `Send + Sync` impls sound; `Drop` unmaps exactly once. +struct ArenaMap { + ptr: *mut u8, + len: usize, +} + +unsafe impl Send for ArenaMap {} +unsafe impl Sync for ArenaMap {} + +impl ArenaMap { + fn map(fd: std::os::fd::RawFd, len: usize) -> std::io::Result { + // SAFETY: mmap with a validated length and live descriptor; the + // mapping is checked against MAP_FAILED immediately. + let ptr = unsafe { + libc::mmap( + std::ptr::null_mut(), + len, + libc::PROT_READ, + libc::MAP_SHARED, + fd, + 0, + ) + }; + if ptr == libc::MAP_FAILED { + Err(std::io::Error::last_os_error()) + } else { + Ok(Self { + ptr: ptr.cast(), + len, + }) + } + } + + fn header(&self) -> &[u8] { + let len = self.len.min(dp_bootstrap::HEADER_LEN); + // SAFETY: `ptr..ptr+len` is inside the mapping by construction. + unsafe { std::slice::from_raw_parts(self.ptr, len) } + } +} + +impl Drop for ArenaMap { + fn drop(&mut self) { + // SAFETY: unmaps exactly the mapping created in `ArenaMap::map`. + unsafe { + libc::munmap(self.ptr.cast(), self.len); + } + } +} + +/// The job's data plane. Carries the resolved bootstrap state; the four path +/// primitives (`read_blob`, `write_blob`, `read_stream`, `write_stream`) +/// arrive with path resolution. +#[pyclass(name = "DataPlane")] +pub struct PyDataPlane { + // Held (never read) to keep the arena mapping alive for the process + // lifetime; dropping it unmaps. + #[allow(dead_code)] + arena: Arc, + // Consumed by the path-resolution primitives (read_blob & co., next + // slice); carried now so the bootstrap result lives with its owner. + #[allow(dead_code)] + resolved: dp_bootstrap::ResolvedBootstrap, +} + +/// The job context handed to `main` by [`run`]. +#[pyclass(name = "Context")] +pub struct PyContext { + data: Py, +} + +#[pymethods] +impl PyContext { + #[getter] + fn data(&self, py: Python<'_>) -> Py { + self.data.clone_ref(py) + } +} + +fn bootstrap_error(message: impl Into) -> PyErr { + PyErr::new::(message.into()) +} + +fn bootstrap_env_fd(name: &str) -> PyResult { + let value = std::env::var_os(name).ok_or_else(|| { + bootstrap_error(format!("bootstrap environment variable {name} is not set")) + })?; + let text = value.to_string_lossy(); + text.parse::() + .map_err(|_| bootstrap_error(format!("{name}={text:?} is not a descriptor number"))) +} + +/// Arm `FD_CLOEXEC` on the wake descriptor. Failing also proves the +/// descriptor exists at all. +fn arm_cloexec(fd: std::os::fd::RawFd) -> PyResult<()> { + // SAFETY: fcntl on a borrowed descriptor number. + let rc = unsafe { libc::fcntl(fd, libc::F_SETFD, libc::FD_CLOEXEC) }; + if rc < 0 { + Err(bootstrap_error(format!( + "wake descriptor {fd} is unusable: {}", + std::io::Error::last_os_error() + ))) + } else { + Ok(()) + } +} + +/// Job-process entrypoint: consume the bootstrap handoff, construct the +/// application [`Context`](crate::PyContext), and drive `main` to completion +/// on the ambient asyncio loop. +/// +/// Bootstrap is fail-fast: if the inherited descriptors or the arena header +/// are absent or malformed, `main` is never invoked and `BootstrapError` +/// propagates (the process should exit non-zero). +#[pyfunction] +fn run(py: Python<'_>, main: Bound<'_, PyAny>) -> PyResult<()> { + let arena_fd = bootstrap_env_fd(dp_bootstrap::ENV_ARENA_FD)?; + let wake_fd = bootstrap_env_fd(dp_bootstrap::ENV_WAKE_FD)?; + if arena_fd < 0 || wake_fd < 0 { + return Err(bootstrap_error( + "bootstrap descriptor numbers must be non-negative", + )); + } + + // Take ownership of both inherited descriptors for the process lifetime. + // SAFETY: the environment contract hands us sole ownership of these. + let arena_file = unsafe { File::from_raw_fd(arena_fd) }; + let wake_owned = unsafe { OwnedFd::from_raw_fd(wake_fd) }; + + // Ground truth for the header's arena-size claim is the descriptor's own + // length, never the header. + let backing_len = arena_file + .metadata() + .map_err(|error| bootstrap_error(format!("stat arena descriptor: {error}")))? + .len(); + if backing_len < dp_bootstrap::HEADER_LEN as u64 { + return Err(bootstrap_error(format!( + "arena backing is {backing_len} bytes, shorter than the {}-byte bootstrap header", + dp_bootstrap::HEADER_LEN + ))); + } + let map = ArenaMap::map(arena_fd, usize::try_from(backing_len).expect("usize arena")) + .map_err(|error| bootstrap_error(format!("map arena: {error}")))?; + + // The mapping keeps the arena alive; dropping the descriptor both stops + // grandchild leakage and makes a second `run` fail loudly. + drop(arena_file); + + // B5: the wake descriptor is the only bootstrap fd left open; arm + // CLOEXEC so nothing this process spawns inherits it. + arm_cloexec(wake_owned.as_raw_fd())?; + + let resolved = dp_bootstrap::parse_bootstrap(map.header(), backing_len) + .map_err(|error| bootstrap_error(error.to_string()))?; + + let data = Py::new( + py, + PyDataPlane { + arena: Arc::new(map), + resolved, + }, + )?; + let context = Py::new(py, PyContext { data })?; + let coroutine = main.call1((context,))?; + let asyncio = PyModule::import(py, "asyncio")?; + asyncio.call_method1("run", (coroutine,))?; + Ok(()) +} + // ─── Module registration ───────────────────────────────────────────────────── + fn register(m: &Bound<'_, PyModule>) -> PyResult<()> { + m.add("SwactorError", m.py().get_type::())?; + m.add("BootstrapError", m.py().get_type::())?; m.add_class::()?; m.add_class::()?; m.add_class::()?; @@ -455,7 +639,9 @@ fn register(m: &Bound<'_, PyModule>) -> PyResult<()> { m.add_class::()?; m.add_class::()?; m.add_class::()?; - m.add_class::()?; + m.add_class::()?; + m.add_class::()?; + m.add_function(wrap_pyfunction!(run, m)?)?; Ok(()) } diff --git a/crates/bindings/python/tests/probe.py b/crates/bindings/python/tests/probe.py new file mode 100755 index 0000000..b6cd882 --- /dev/null +++ b/crates/bindings/python/tests/probe.py @@ -0,0 +1,42 @@ +#!/usr/bin/env python3 +"""L3 probe for the bootstrap slice. + +Runs under ``swactor.run`` and reports facts observable without any +data-plane primitives: context shape, wake readability, and descriptor +hygiene across an exec boundary. +""" + +from __future__ import annotations + +import os +import select +import subprocess +import sys + +import swactor + +_CHILD_CHECK = ( + "import os, sys; " + "sys.exit(0 if sys.argv[1] not in os.listdir('/proc/self/fd') else 3)" +) + + +async def main(ctx: swactor.Context) -> None: + facts = [f"HAS_DATA={int(isinstance(ctx.data, swactor.DataPlane))}"] + + wake = int(os.environ["SWACTOR_WAKE_FD"]) + readable = bool(select.select([wake], [], [], 0)[0]) + facts.append(f"WAKE_READABLE={int(readable)}") + + # B5: a close_fds=False child must not see the CLOEXEC-armed wake fd. + child = subprocess.Popen( + [sys.executable, "-c", _CHILD_CHECK, str(wake)], + close_fds=False, + ) + child.wait() + facts.append(f"WAKE_LEAKED={int(child.returncode == 3)}") + print(" ".join(facts)) + + +if __name__ == "__main__": + swactor.run(main) diff --git a/crates/bindings/python/tests/test_bootstrap.py b/crates/bindings/python/tests/test_bootstrap.py new file mode 100644 index 0000000..0db2c9b --- /dev/null +++ b/crates/bindings/python/tests/test_bootstrap.py @@ -0,0 +1,460 @@ +"""Bootstrap-slice behavior guarantees for the ``swactor`` Python binding. + +Defends the agreed invariants: +- B1 deterministic handoff: a valid exec-time bootstrap reaches ``main`` +- B2 fail-fast purity: every bootstrap defect raises ``BootstrapError`` + before ``main`` is invoked +- B3 version gate rejects, never guesses +- B4 arbitrary header bytes never crash the process +- B5 grandchild isolation: post-bootstrap children inherit no bootstrap fds +- B6 wake opacity: bootstrap does not consume the wake descriptor +- B7 exit-code contract across a real process boundary +- B8 the two environment names are the whole handoff ABI +- B9 no identity leakage on the Python-visible surface +""" + +from __future__ import annotations + +import os +import random +import select +import struct +import subprocess +import sys +import tempfile +from pathlib import Path + +import pytest + +import swactor + +MAGIC = int.from_bytes(b"SWBS", "little") +ENV_ARENA = "SWACTOR_ARENA_FD" +ENV_WAKE = "SWACTOR_WAKE_FD" +VERSION = 1 +ARENA_SIZE = 1 << 20 + +MAIN_RAN = [] + + +# ─── fixtures ──────────────────────────────────────────────────────────────── + +def pack_header( + arena_size=ARENA_SIZE, + ring_offset=4096, + ring_capacity=8192, + ring_generation=1, + *, + magic=MAGIC, + version=VERSION, + reserved0=0, + reserved1=0, +): + return struct.pack( + " int: + if hasattr(os, "memfd_create"): + return os.memfd_create("test-arena", 0) + # Not every interpreter exposes memfd_create; an unlinked temp file is an + # equally anonymous regular-file inode for ftruncate/pwrite/mmap. + fd, path = tempfile.mkstemp(prefix="swactor-test-arena-") + os.unlink(path) + return fd + + +def make_arena(header: bytes, size: int = ARENA_SIZE) -> int: + fd = _create_backing() + os.ftruncate(fd, size) + os.pwrite(fd, header, 0) + return fd + + +def make_wake(signaled: bool) -> tuple[int, int]: + read_fd, write_fd = os.pipe() + if signaled: + os.write(write_fd, b"x") + # Keep the write end open: closing it would signal EOF and make the read + # end "readable" without a wake ever being delivered. + return read_fd, write_fd + + +@pytest.fixture +def valid_env(monkeypatch): + """A complete, valid bootstrap in the current process.""" + keep_open = [] + + def install(*, header=None, arena_size=ARENA_SIZE, signaled=True): + arena_fd = make_arena( + header if header is not None else pack_header(arena_size=arena_size), + arena_size, + ) + wake_fd, write_fd = make_wake(signaled) + keep_open.append(write_fd) + # Make the wake fd inheritable so only run()'s CLOEXEC can hide it + # from children (B5 evidence). + os.set_inheritable(wake_fd, True) + monkeypatch.setenv(ENV_ARENA, str(arena_fd)) + monkeypatch.setenv(ENV_WAKE, str(wake_fd)) + return arena_fd, wake_fd + + yield install + for write_fd in keep_open: + os.close(write_fd) + +# ─── B1 / B9: valid bootstrap reaches main with a clean surface ───────────── + +def test_run_constructs_context_and_data(valid_env): + valid_env() + seen = {} + + async def main(ctx): + seen["ctx"] = ctx + + swactor.run(main) + assert isinstance(seen["ctx"], swactor.Context) + assert isinstance(seen["ctx"].data, swactor.DataPlane) + # B9: no identities leak onto the surface. + assert {a for a in dir(seen["ctx"]) if not a.startswith("_")} == {"data"} + leaked = {"arena", "resolved", "control_ring", "offset", "generation"} + assert not leaked & {a for a in dir(seen["ctx"].data) if not a.startswith("_")} + +def test_run_passes_the_same_context_instance(valid_env): + valid_env() + seen = {} + + async def main(ctx): + seen["one"] = ctx + seen["data"] = ctx.data + + swactor.run(main) + assert seen["one"].data is seen["data"] + + +# ─── B5: fd hygiene ────────────────────────────────────────────────────────── + +def test_arena_fd_closed_after_run(valid_env): + arena_fd, _ = valid_env() + + async def main(ctx): + pass + + swactor.run(main) + with pytest.raises(OSError): + os.fstat(arena_fd) + + +# ─── B6: wake opacity ──────────────────────────────────────────────────────── + + +@pytest.mark.parametrize("signaled,expected", [(True, 1), (False, 0)]) +def test_bootstrap_does_not_consume_wake(valid_env, signaled, expected): + valid_env(signaled=signaled) + readable = {} + + async def main(ctx): + wake = int(os.environ[ENV_WAKE]) + readable["value"] = int(bool(select.select([wake], [], [], 0)[0])) + + swactor.run(main) + assert readable["value"] == expected + + + +def test_wake_fd_cloexec_armed_by_run(valid_env): + _, wake_fd = valid_env() + + async def main(ctx): + child = subprocess.Popen( + [ + sys.executable, + "-c", + "import os, sys; " + "sys.exit(0 if sys.argv[1] not in os.listdir('/proc/self/fd') else 3)", + str(wake_fd), + ], + close_fds=False, + ) + child.wait() + assert child.returncode != 3, "close_fds=False child saw the wake fd" + + swactor.run(main) + + +# ─── B2 / B3 / B4: defect table, in process ───────────────────────────────── + +def _expect_bootstrap_failure(monkeypatch, header=None, *, env=None, arena_size=ARENA_SIZE, match=""): + calls = [] + + async def main(ctx): # pragma: no cover - must never run + calls.append(ctx) + + if env is not None: + monkeypatch.setenv(ENV_ARENA, str(env[0])) + monkeypatch.setenv(ENV_WAKE, str(env[1])) + else: + arena_fd = make_arena(header if header is not None else pack_header(arena_size=arena_size), arena_size) + wake_fd, _keep_wake = make_wake(True) + monkeypatch.setenv(ENV_ARENA, str(arena_fd)) + monkeypatch.setenv(ENV_WAKE, str(wake_fd)) + + with pytest.raises(swactor.BootstrapError, match=match) as excinfo: + swactor.run(main) + assert isinstance(excinfo.value, swactor.SwactorError) + assert calls == [], "main must never run on a bootstrap defect" +def test_missing_env_rejected(monkeypatch): + monkeypatch.delenv(ENV_ARENA, raising=False) + monkeypatch.delenv(ENV_WAKE, raising=False) + calls = [] + + async def main(ctx): # pragma: no cover - must never run + calls.append(ctx) + + with pytest.raises(swactor.BootstrapError, match="is not set"): + swactor.run(main) + assert calls == [] + + +def test_non_numeric_fd_rejected(monkeypatch): + _expect_bootstrap_failure( + monkeypatch, env=("not-a-number", "also-not"), match="not a descriptor number" + ) + + + +def test_negative_fd_rejected(monkeypatch): + _expect_bootstrap_failure(monkeypatch, env=("-1", "-1"), match="non-negative") + + +def test_closed_fd_rejected(monkeypatch): + closed = os.pipe()[0] + os.close(closed) + _expect_bootstrap_failure(monkeypatch, env=(closed, 1), match="stat arena descriptor") + + +def test_short_backing_rejected(monkeypatch): + _expect_bootstrap_failure( + monkeypatch, + header=b"\0" * 16, + arena_size=16, + match="shorter than the 48-byte bootstrap header", + ) + + +@pytest.mark.parametrize( + "header,match", + [ + (pack_header(magic=MAGIC ^ 0xFF), "magic mismatch"), + (pack_header(version=0), "unsupported bootstrap version 0"), + (pack_header(version=2), "unsupported bootstrap version 2"), + (pack_header(reserved0=1), "reserved bytes at offset 6"), + (pack_header(reserved1=1), "reserved bytes at offset 40"), + ( + pack_header(arena_size=ARENA_SIZE, ring_offset=16), + "overlaps the bootstrap header", + ), + ( + pack_header(ring_offset=ARENA_SIZE, ring_capacity=1), + "exceeds arena size", + ), + ( + pack_header(ring_offset=2**64 - 8, ring_capacity=16), + "exceeds arena size", + ), + (pack_header(ring_capacity=0), "capacity is zero"), + (pack_header(ring_generation=0), "generation is zero"), + ], +) +def test_header_defects_rejected(monkeypatch, header, match): + _expect_bootstrap_failure(monkeypatch, header, match=match) + + +def test_lying_arena_size_rejected(monkeypatch): + _expect_bootstrap_failure( + monkeypatch, + pack_header(arena_size=ARENA_SIZE), + arena_size=ARENA_SIZE + 1, + match="disagrees with backing length", + ) + + +def test_arbitrary_header_bytes_never_crash(monkeypatch): + rng = random.Random(0x53574253) + for _ in range(128): + fields = { + "arena_size": rng.choice([rng.getrandbits(64), ARENA_SIZE, 0, 1]), + "ring_offset": rng.choice( + [rng.getrandbits(64), 16, 4096, ARENA_SIZE, 2**64 - 8] + ), + "ring_capacity": rng.choice([rng.getrandbits(64), 0, 1, 8192, 2**64 - 1]), + "ring_generation": rng.choice([rng.getrandbits(64), 0, 1, 7]), + } + header = pack_header(**fields) + ran = [] + + async def main(ctx): # pragma: no cover - only on accidental success + ran.append(ctx) + + arena_fd = make_arena(header) + wake_fd, _keep_wake = make_wake(False) + monkeypatch.setenv(ENV_ARENA, str(arena_fd)) + monkeypatch.setenv(ENV_WAKE, str(wake_fd)) + + end = fields["ring_offset"] + fields["ring_capacity"] + should_parse = ( + fields["arena_size"] == ARENA_SIZE + and fields["ring_capacity"] > 0 + and fields["ring_generation"] > 0 + and fields["ring_offset"] >= 48 + and end <= ARENA_SIZE + ) + if should_parse: + swactor.run(main) + assert ran, "parser accepted fields that satisfy no invariant" + else: + with pytest.raises(swactor.BootstrapError): + swactor.run(main) + assert not ran + + +# ─── B7 / B1 / B8: real process boundary ──────────────────────────────────── + +PROBE = Path(__file__).parent / "probe.py" + + +def _spawn(argv, *, arena_fd, wake_fd, env_extra=None): + env = {k: v for k, v in os.environ.items() if k not in (ENV_ARENA, ENV_WAKE)} + env[ENV_ARENA] = str(arena_fd) + env[ENV_WAKE] = str(wake_fd) + if env_extra: + env.update(env_extra) + for fd in (arena_fd, wake_fd): + os.set_inheritable(fd, True) + return subprocess.run( + argv, + env=env, + pass_fds=(arena_fd, wake_fd), + capture_output=True, + text=True, + timeout=60, + ) + + +def _parse_facts(stdout): + facts = dict( + part.split("=", 1) for part in stdout.strip().split() if "=" in part + ) + return facts + + +def test_probe_succeeds_across_exec_boundary(): + arena_fd = make_arena(pack_header()) + wake_fd, _keep_wake = make_wake(signaled=True) + result = _spawn([sys.executable, str(PROBE)], arena_fd=arena_fd, wake_fd=wake_fd) + assert result.returncode == 0, result.stderr + facts = _parse_facts(result.stdout) + assert facts["HAS_DATA"] == "1" + assert facts["WAKE_READABLE"] == "1", "B6: presignaled wake must survive exec" + assert facts["WAKE_LEAKED"] == "0", "B5: grandchild must not see the wake fd" + + +def test_probe_reports_unsignaled_wake_across_exec_boundary(): + arena_fd = make_arena(pack_header()) + wake_fd, _keep_wake = make_wake(signaled=False) + result = _spawn([sys.executable, str(PROBE)], arena_fd=arena_fd, wake_fd=wake_fd) + assert result.returncode == 0, result.stderr + assert _parse_facts(result.stdout)["WAKE_READABLE"] == "0" + + +_MAIN_RETURN_SCRIPT = ( + "import swactor\n" + "async def main(ctx):\n" + " print('MAIN_OK')\n" + "swactor.run(main)\n" +) +_MAIN_RAISE_SCRIPT = ( + "import swactor\n" + "async def main(ctx):\n" + " raise RuntimeError('boom')\n" + "swactor.run(main)\n" +) +_NEVER_RUN_SCRIPT = ( + "import swactor\n" + "async def main(ctx):\n" + " raise AssertionError('must not run')\n" + "swactor.run(main)\n" +) + + +def test_exit_zero_when_main_returns(): + arena_fd = make_arena(pack_header()) + wake_fd, _keep_wake = make_wake(True) + result = _spawn( + [sys.executable, "-c", _MAIN_RETURN_SCRIPT], + arena_fd=arena_fd, + wake_fd=wake_fd, + ) + assert result.returncode == 0, result.stderr + assert "MAIN_OK" in result.stdout + + +def test_exit_nonzero_when_main_raises_with_traceback(): + arena_fd = make_arena(pack_header()) + wake_fd, _keep_wake = make_wake(True) + result = _spawn( + [sys.executable, "-c", _MAIN_RAISE_SCRIPT], + arena_fd=arena_fd, + wake_fd=wake_fd, + ) + assert result.returncode != 0 + assert "boom" in result.stderr + + +def test_exit_nonzero_on_bootstrap_defect_without_entering_main(): + arena_fd = make_arena(pack_header(magic=MAGIC ^ 0xFF)) + wake_fd, _keep_wake = make_wake(True) + result = _spawn( + [sys.executable, "-c", _NEVER_RUN_SCRIPT], + arena_fd=arena_fd, + wake_fd=wake_fd, + ) + assert result.returncode != 0 + assert "BootstrapError" in result.stderr + assert "must not run" not in result.stderr + + +def test_handoff_uses_only_the_two_env_names(): + arena_fd = make_arena(pack_header()) + wake_fd, _keep_wake = make_wake(True) + result = _spawn( + [sys.executable, "-c", _MAIN_RETURN_SCRIPT], + arena_fd=arena_fd, + wake_fd=wake_fd, + ) + assert result.returncode == 0 + # B8 is structural: the binding reads exactly ENV_ARENA and ENV_WAKE. + # Prove it by renaming one and observing failure. + env = {k: v for k, v in os.environ.items() if k not in (ENV_ARENA, ENV_WAKE)} + env["SWACTOR_DATA_PLANE_INPUT"] = "ignored-legacy-name" + env[ENV_WAKE] = str(wake_fd) + for fd in (arena_fd, wake_fd): + os.set_inheritable(fd, True) + missing = subprocess.run( + [sys.executable, "-c", _NEVER_RUN_SCRIPT], + env=env, + pass_fds=(arena_fd, wake_fd), + capture_output=True, + text=True, + timeout=60, + ) + assert missing.returncode != 0 + assert ENV_ARENA in missing.stderr diff --git a/crates/bindings/python/uv.lock b/crates/bindings/python/uv.lock index 965a70b..2c5fab6 100644 --- a/crates/bindings/python/uv.lock +++ b/crates/bindings/python/uv.lock @@ -581,6 +581,32 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/fa/5e/f8e9a1d23b9c20a551a8a02ea3637b4642e22c2626e3a13a9a29cdea99eb/importlib_metadata-8.7.1-py3-none-any.whl", hash = "sha256:5a1f80bf1daa489495071efbb095d75a634cf28a8bc299581244063b53176151", size = 27865, upload-time = "2025-12-21T10:00:18.329Z" }, ] +[[package]] +name = "iniconfig" +version = "2.1.0" +source = { registry = "https://pypi.org/simple" } +resolution-markers = [ + "python_full_version < '3.10'", +] +sdist = { url = "https://files.pythonhosted.org/packages/f2/97/ebf4da567aa6827c909642694d71c9fcf53e5b504f2d96afea02718862f3/iniconfig-2.1.0.tar.gz", hash = "sha256:3abbd2e30b36733fee78f9c7f7308f2d0050e88f0087fd25c2645f63c773e1c7", size = 4793, upload-time = "2025-03-19T20:09:59.721Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/2c/e1/e6716421ea10d38022b952c159d5161ca1193197fb744506875fbb87ea7b/iniconfig-2.1.0-py3-none-any.whl", hash = "sha256:9deba5723312380e77435581c6bf4935c94cbfab9b1ed33ef8d238ea168eb760", size = 6050, upload-time = "2025-03-19T20:10:01.071Z" }, +] + +[[package]] +name = "iniconfig" +version = "2.3.0" +source = { registry = "https://pypi.org/simple" } +resolution-markers = [ + "python_full_version >= '3.14'", + "python_full_version >= '3.11' and python_full_version < '3.14'", + "python_full_version == '3.10.*'", +] +sdist = { url = "https://files.pythonhosted.org/packages/72/34/14ca021ce8e5dfedc35312d08ba8bf51fdd999c576889fc2c24cb97f4f10/iniconfig-2.3.0.tar.gz", hash = "sha256:c76315c77db068650d49c5b56314774a7804df16fee4402c1f19d6d15d8c4730", size = 20503, upload-time = "2025-10-18T21:55:43.219Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/cb/b1/3846dd7f199d53cb17f49cba7e651e9ce294d8497c8c150530ed11865bb8/iniconfig-2.3.0-py3-none-any.whl", hash = "sha256:f631c04d2c48c52b84d0d0549c99ff3859c98df65b3101406327ecc7d53fbf12", size = 7484, upload-time = "2025-10-18T21:55:41.639Z" }, +] + [[package]] name = "ipykernel" version = "6.31.0" @@ -1463,6 +1489,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/cb/28/3bfe2fa5a7b9c46fe7e13c97bda14c895fb10fa2ebf1d0abb90e0cea7ee1/platformdirs-4.5.1-py3-none-any.whl", hash = "sha256:d03afa3963c806a9bed9d5125c8f4cb2fdaf74a55ab60e5d59b3fde758104d31", size = 18731, upload-time = "2025-12-05T13:52:56.823Z" }, ] +[[package]] +name = "pluggy" +version = "1.6.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/f9/e2/3e91f31a7d2b083fe6ef3fa267035b518369d9511ffab804f839851d2779/pluggy-1.6.0.tar.gz", hash = "sha256:7dcc130b76258d33b90f61b658791dede3486c3e6bfb003ee5c9bfb396dd22f3", size = 69412, upload-time = "2025-05-15T12:30:07.975Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/54/20/4d324d65cc6d9205fabedc306948156824eb9f0ee1633355a8f7ec5c66bf/pluggy-1.6.0-py3-none-any.whl", hash = "sha256:e920276dd6813095e9377c0bc5566d94c932c33b27a3e3945d8389c374dd4746", size = 20538, upload-time = "2025-05-15T12:30:06.134Z" }, +] + [[package]] name = "prometheus-client" version = "0.24.1" @@ -1565,6 +1600,50 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/c7/21/705964c7812476f378728bdf590ca4b771ec72385c533964653c68e86bdc/pygments-2.19.2-py3-none-any.whl", hash = "sha256:86540386c03d588bb81d44bc3928634ff26449851e99741617ecb9037ee5ec0b", size = 1225217, upload-time = "2025-06-21T13:39:07.939Z" }, ] +[[package]] +name = "pytest" +version = "8.4.2" +source = { registry = "https://pypi.org/simple" } +resolution-markers = [ + "python_full_version < '3.10'", +] +dependencies = [ + { name = "colorama", marker = "python_full_version < '3.10' and sys_platform == 'win32'" }, + { name = "exceptiongroup", marker = "python_full_version < '3.10'" }, + { name = "iniconfig", version = "2.1.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.10'" }, + { name = "packaging", marker = "python_full_version < '3.10'" }, + { name = "pluggy", marker = "python_full_version < '3.10'" }, + { name = "pygments", marker = "python_full_version < '3.10'" }, + { name = "tomli", marker = "python_full_version < '3.10'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/a3/5c/00a0e072241553e1a7496d638deababa67c5058571567b92a7eaa258397c/pytest-8.4.2.tar.gz", hash = "sha256:86c0d0b93306b961d58d62a4db4879f27fe25513d4b969df351abdddb3c30e01", size = 1519618, upload-time = "2025-09-04T14:34:22.711Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/a8/a4/20da314d277121d6534b3a980b29035dcd51e6744bd79075a6ce8fa4eb8d/pytest-8.4.2-py3-none-any.whl", hash = "sha256:872f880de3fc3a5bdc88a11b39c9710c3497a547cfa9320bc3c5e62fbf272e79", size = 365750, upload-time = "2025-09-04T14:34:20.226Z" }, +] + +[[package]] +name = "pytest" +version = "9.1.1" +source = { registry = "https://pypi.org/simple" } +resolution-markers = [ + "python_full_version >= '3.14'", + "python_full_version >= '3.11' and python_full_version < '3.14'", + "python_full_version == '3.10.*'", +] +dependencies = [ + { name = "colorama", marker = "python_full_version >= '3.10' and sys_platform == 'win32'" }, + { name = "exceptiongroup", marker = "python_full_version == '3.10.*'" }, + { name = "iniconfig", version = "2.3.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.10'" }, + { name = "packaging", marker = "python_full_version >= '3.10'" }, + { name = "pluggy", marker = "python_full_version >= '3.10'" }, + { name = "pygments", marker = "python_full_version >= '3.10'" }, + { name = "tomli", marker = "python_full_version == '3.10.*'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/e4/47/b9efed96c114afcfa3c9d3fe98a76a1d14c74a9e266d397cf6eb64be5e01/pytest-9.1.1.tar.gz", hash = "sha256:1088fbde8f2b49d95a549a195707afa7a76a3ce9bcadc26b6d71f0ffda5fe313", size = 1636369, upload-time = "2026-06-19T10:58:32.857Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/24/25/1de2678b631f5a49215c6c96fff41ba892b0a34df68d6d80292b1b48aa7f/pytest-9.1.1-py3-none-any.whl", hash = "sha256:37a86b45efb9a47a61a36449063e8e18d0cab3161329fc099eb21783169c4f0c", size = 386536, upload-time = "2026-06-19T10:58:31.347Z" }, +] + [[package]] name = "python-dateutil" version = "2.9.0.post0" @@ -2234,6 +2313,8 @@ dev = [ { name = "ipykernel", version = "6.31.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.10'" }, { name = "ipykernel", version = "7.1.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.10'" }, { name = "jupyter" }, + { name = "pytest", version = "8.4.2", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.10'" }, + { name = "pytest", version = "9.1.1", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.10'" }, ] [package.metadata] @@ -2242,6 +2323,7 @@ dev = [ dev = [ { name = "ipykernel" }, { name = "jupyter" }, + { name = "pytest" }, ] [[package]] diff --git a/crates/data-plane/src/arena.rs b/crates/data-plane/src/arena.rs index dee1bf0..efd0f3a 100644 --- a/crates/data-plane/src/arena.rs +++ b/crates/data-plane/src/arena.rs @@ -289,6 +289,15 @@ impl ArenaManager { } } + /// True backing length of the arena in bytes. + /// + /// This is the ground truth the bootstrap header's `arena_size` must + /// match; it equals `config.reservation_ceiling` because the backing is + /// created with exactly that length. + pub fn arena_len(&self) -> u64 { + self.config.reservation_ceiling + } + #[cfg(target_os = "linux")] pub fn arena_fd(&self) -> std::os::fd::RawFd { self._backing.fd() diff --git a/crates/data-plane/src/bootstrap.rs b/crates/data-plane/src/bootstrap.rs new file mode 100644 index 0000000..8bfa19a --- /dev/null +++ b/crates/data-plane/src/bootstrap.rs @@ -0,0 +1,455 @@ +//! Exec-time bootstrap ABI shared by the job host and the Python binding. +//! +//! The host launches a job process with two inherited file descriptors named +//! by environment variables ([`ENV_ARENA_FD`], [`ENV_WAKE_FD`]) and, before +//! exec, writes a fixed-layout header at arena offset 0. The job-side entry +//! point maps the arena, validates the header against the descriptor's true +//! length (`fstat`, never the header's own claim), and only then constructs +//! the application context's data plane. +//! +//! All identities (actors, nodes, edges, rings) stay on the host side of this +//! boundary: the header names one control-ring region and nothing else. +//! +//! Layout (all integers little-endian, no implicit padding): +//! +//! | offset | size | field | +//! |--------|------|-------------------------------| +//! | 0 | 4 | magic (`SWBS`) | +//! | 4 | 2 | version | +//! | 6 | 2 | reserved (must be zero) | +//! | 8 | 8 | arena_size | +//! | 16 | 8 | control_ring_offset | +//! | 24 | 8 | control_ring_capacity | +//! | 32 | 8 | control_ring_generation | +//! | 40 | 8 | reserved (must be zero) | + +use std::fmt; + +/// `"SWBS"` read little-endian. +pub const BOOTSTRAP_MAGIC: u32 = u32::from_le_bytes(*b"SWBS"); +/// The only version this crate understands. +pub const BOOTSTRAP_VERSION: u16 = 1; + +/// Environment variable naming the inherited arena memfd. +pub const ENV_ARENA_FD: &str = "SWACTOR_ARENA_FD"; +/// Environment variable naming the inherited wake descriptor. +pub const ENV_WAKE_FD: &str = "SWACTOR_WAKE_FD"; + +/// Fixed byte length of the bootstrap header at arena offset 0. +pub const HEADER_LEN: usize = 48; +/// Arena offsets `[0, HEADER_END_OFFSET)` belong to the bootstrap header. +pub const HEADER_END_OFFSET: u64 = HEADER_LEN as u64; + +const RESERVED0_OFFSET: usize = 6; +const RESERVED1_OFFSET: usize = 40; + +fn read_u32(page: &[u8], offset: usize) -> u32 { + u32::from_le_bytes(page[offset..offset + 4].try_into().expect("u32 slice")) +} + +fn read_u16(page: &[u8], offset: usize) -> u16 { + u16::from_le_bytes(page[offset..offset + 2].try_into().expect("u16 slice")) +} + +fn read_u64(page: &[u8], offset: usize) -> u64 { + u64::from_le_bytes(page[offset..offset + 8].try_into().expect("u64 slice")) +} + +/// The control-ring region named by the bootstrap header. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct ControlRingLayout { + pub offset: u64, + pub capacity: u64, + pub generation: u64, +} + +/// The validated bootstrap header, as encoded at arena offset 0. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct BootstrapHeader { + pub arena_size: u64, + pub control_ring: ControlRingLayout, +} + +impl BootstrapHeader { + /// Encode into the fixed 48-byte little-endian layout. + pub fn encode(&self) -> [u8; HEADER_LEN] { + let mut page = [0_u8; HEADER_LEN]; + page[0..4].copy_from_slice(&BOOTSTRAP_MAGIC.to_le_bytes()); + page[4..6].copy_from_slice(&BOOTSTRAP_VERSION.to_le_bytes()); + page[8..16].copy_from_slice(&self.arena_size.to_le_bytes()); + page[16..24].copy_from_slice(&self.control_ring.offset.to_le_bytes()); + page[24..32].copy_from_slice(&self.control_ring.capacity.to_le_bytes()); + page[32..40].copy_from_slice(&self.control_ring.generation.to_le_bytes()); + // Reserved bytes stay zero. + page + } +} + +/// The result of a successful bootstrap parse. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct ResolvedBootstrap { + pub arena_size: u64, + pub control_ring: ControlRingLayout, +} + +/// Every way a bootstrap page can fail validation. +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum BootstrapError { + /// Fewer than [`HEADER_LEN`] bytes available at offset 0. + Truncated { available: usize }, + BadMagic { found: u32 }, + /// A reserved field is nonzero: the page is not a v1-shape header. + ReservedBytesNotZero { at: usize }, + UnsupportedVersion { found: u16, supported: u16 }, + /// The header's claimed size disagrees with the descriptor's true length. + ArenaSizeMismatch { header: u64, backing: u64 }, + /// The control ring overlaps the bootstrap header region. + RingOverlapsHeader { offset: u64 }, + /// The control ring (or its end offset) falls outside the arena, including + /// offset+capacity overflow. + RingOutOfBounds { + offset: u64, + capacity: u64, + arena_size: u64, + }, + ZeroRingCapacity, + ZeroRingGeneration, +} + +impl fmt::Display for BootstrapError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Truncated { available } => { + write!(f, "bootstrap header truncated: {available} of {HEADER_LEN} bytes") + } + Self::BadMagic { found } => { + write!(f, "bootstrap magic mismatch: found {found:#010x}") + } + Self::ReservedBytesNotZero { at } => { + write!(f, "bootstrap reserved bytes at offset {at} are nonzero") + } + Self::UnsupportedVersion { found, supported } => { + write!(f, "unsupported bootstrap version {found} (supported: {supported})") + } + Self::ArenaSizeMismatch { header, backing } => { + write!( + f, + "bootstrap arena size {header} disagrees with backing length {backing}" + ) + } + Self::RingOverlapsHeader { offset } => { + write!(f, "control ring offset {offset} overlaps the bootstrap header") + } + Self::RingOutOfBounds { + offset, + capacity, + arena_size, + } => { + write!( + f, + "control ring [{offset}, +{capacity}] exceeds arena size {arena_size}" + ) + } + Self::ZeroRingCapacity => write!(f, "control ring capacity is zero"), + Self::ZeroRingGeneration => write!(f, "control ring generation is zero"), + } + } +} + +impl std::error::Error for BootstrapError {} + +/// Validate a bootstrap page read from arena offset 0. +/// +/// `backing_len` is the true length of the arena descriptor (from `fstat` on +/// the job side, from the arena config on the host side) — never a value +/// taken from the page itself. +pub fn parse_bootstrap(page: &[u8], backing_len: u64) -> Result { + if page.len() < HEADER_LEN { + return Err(BootstrapError::Truncated { + available: page.len(), + }); + } + let magic = read_u32(page, 0); + if magic != BOOTSTRAP_MAGIC { + return Err(BootstrapError::BadMagic { found: magic }); + } + if read_u16(page, RESERVED0_OFFSET) != 0 { + return Err(BootstrapError::ReservedBytesNotZero { + at: RESERVED0_OFFSET, + }); + } + let version = read_u16(page, 4); + if version != BOOTSTRAP_VERSION { + return Err(BootstrapError::UnsupportedVersion { + found: version, + supported: BOOTSTRAP_VERSION, + }); + } + if page[RESERVED1_OFFSET..HEADER_LEN].iter().any(|&b| b != 0) { + return Err(BootstrapError::ReservedBytesNotZero { + at: RESERVED1_OFFSET, + }); + } + let arena_size = read_u64(page, 8); + if arena_size != backing_len { + return Err(BootstrapError::ArenaSizeMismatch { + header: arena_size, + backing: backing_len, + }); + } + let control_ring = ControlRingLayout { + offset: read_u64(page, 16), + capacity: read_u64(page, 24), + generation: read_u64(page, 32), + }; + if control_ring.capacity == 0 { + return Err(BootstrapError::ZeroRingCapacity); + } + if control_ring.generation == 0 { + return Err(BootstrapError::ZeroRingGeneration); + } + if control_ring.offset < HEADER_END_OFFSET { + return Err(BootstrapError::RingOverlapsHeader { + offset: control_ring.offset, + }); + } + match control_ring.offset.checked_add(control_ring.capacity) { + Some(end) if end <= arena_size => Ok(ResolvedBootstrap { + arena_size, + control_ring, + }), + _ => Err(BootstrapError::RingOutOfBounds { + offset: control_ring.offset, + capacity: control_ring.capacity, + arena_size, + }), + } +} + +// ─── Host-side writer ──────────────────────────────────────────────────────── + +#[cfg(target_os = "linux")] +mod write { + use std::collections::BTreeMap; + use std::os::fd::{AsRawFd, FromRawFd, OwnedFd}; + + use crate::arena::{ + ArenaEvent, ArenaManager, ArenaRequest, LeaseRing, LeaseRequestId, RingLease, + RingLeaseRejection, RingSpec, + }; + use crate::bootstrap::{ + BootstrapHeader, ControlRingLayout, HEADER_LEN as HEADER_REGION_BYTES, + HEADER_END_OFFSET, + }; + + /// First generation stamped into a freshly written control ring. + pub const CONTROL_RING_GENERATION: u64 = 1; + + /// Request id for the one-time bootstrap-header region lease. + const HEADER_REGION_REQUEST_ID: u64 = 1; + /// Request id for the control-ring lease. + const CONTROL_RING_REQUEST_ID: u64 = 2; + + const ZERO_CHUNK: usize = 4 * 1024; + + /// Caller-requested shape of the control ring. + #[derive(Clone, Copy, Debug)] + pub struct ControlRingSpec { + pub data_bytes: u64, + pub alignment: u64, + } + + /// Everything the spawner needs to exec a job process. + #[derive(Debug)] + pub struct JobHandoff { + /// Environment map naming the two inherited descriptors. + pub env: BTreeMap, + /// Inheritable (no `FD_CLOEXEC`) duplicate of the arena memfd. + pub arena_fd: OwnedFd, + /// Inheritable (no `FD_CLOEXEC`) duplicate of the wake eventfd. + pub wake_fd: OwnedFd, + /// Host-side `EFD_CLOEXEC` eventfd copy; write to it to wake the job. + pub host_wake_fd: OwnedFd, + /// The control-ring region recorded in the bootstrap header. + pub control_ring: ControlRingLayout, + } + + #[derive(Debug)] + pub enum BootstrapWriteError { + Io(std::io::Error), + InvalidControlRingSpec, + ControlRingLeaseRejected(RingLeaseRejection), + /// A fresh arena must place leases immediately; queueing is a bug. + ControlRingLeaseQueued, + UnexpectedLeaseOutcome, + FdSetup(std::io::Error), + } + + impl std::fmt::Display for BootstrapWriteError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Io(error) => write!(f, "arena write failed: {error}"), + Self::InvalidControlRingSpec => { + write!(f, "control ring spec needs nonzero data_bytes and alignment") + } + Self::ControlRingLeaseRejected(reason) => { + write!(f, "control ring lease rejected: {reason:?}") + } + Self::ControlRingLeaseQueued => { + write!(f, "control ring lease queued on a fresh arena") + } + Self::UnexpectedLeaseOutcome => { + write!(f, "arena produced an unexpected lease outcome") + } + Self::FdSetup(error) => write!(f, "bootstrap fd setup failed: {error}"), + } + } + } + + impl std::error::Error for BootstrapWriteError {} + + /// Reserve the bootstrap header region, lease and initialize the control + /// ring, write the header, and produce the job-process handoff. + /// + /// The header region (`0..HEADER_END_OFFSET`, rounded up by arena + /// alignment) and the control ring are disjoint leases from the arena's + /// own placement law, so no other lease can ever overlap them. All writes + /// complete before this returns: after [`JobHandoff`] exists, the child + /// may exec at any time. + pub fn write_bootstrap( + arena: &mut ArenaManager, + spec: ControlRingSpec, + ) -> Result { + use super::ControlRingLayout as Layout; + + if spec.data_bytes == 0 || spec.alignment == 0 { + return Err(BootstrapWriteError::InvalidControlRingSpec); + } + + // Reserve offset 0 for the header so no ring can ever be placed there. + let header_lease = lease( + arena, + HEADER_REGION_REQUEST_ID, + RingSpec { + header_bytes: HEADER_REGION_BYTES as u64, + data_bytes: 0, + alignment: spec.alignment, + }, + )?; + debug_assert!(header_lease.layout.start_offset < HEADER_END_OFFSET); + + let ring_lease = lease( + arena, + CONTROL_RING_REQUEST_ID, + RingSpec { + // The control ring's own header is a u64 generation slot for + // now; framing arrives with path resolution. + header_bytes: 8, + data_bytes: spec.data_bytes, + alignment: spec.alignment, + }, + )?; + + let ring = Layout { + offset: ring_lease.layout.start_offset, + capacity: ring_lease.layout.end_offset - ring_lease.layout.start_offset, + generation: CONTROL_RING_GENERATION, + }; + + zero_region(arena, ring_lease.layout.start_offset, ring_lease.layout.end_offset)?; + arena + .write_arena(ring_lease.layout.start_offset, &ring.generation.to_le_bytes()) + .map_err(BootstrapWriteError::Io)?; + + let header = BootstrapHeader { + arena_size: arena.arena_len(), + control_ring: ring, + }; + arena + .write_arena(header_lease.layout.start_offset, &header.encode()) + .map_err(BootstrapWriteError::Io)?; + + let host_wake_fd = create_wake_eventfd()?; + let wake_fd = dup_without_cloexec(host_wake_fd.as_raw_fd())?; + let arena_fd = dup_without_cloexec(arena.arena_fd())?; + + let env = BTreeMap::from([ + (super::ENV_ARENA_FD.to_owned(), arena_fd.as_raw_fd().to_string()), + (super::ENV_WAKE_FD.to_owned(), wake_fd.as_raw_fd().to_string()), + ]); + + Ok(JobHandoff { + env, + arena_fd, + wake_fd, + host_wake_fd, + control_ring: ring, + }) + } + + fn lease( + arena: &mut ArenaManager, + request_id: u64, + ring_spec: RingSpec, + ) -> Result { + let request = LeaseRing { + request_id: LeaseRequestId(request_id), + ring_spec, + }; + let mut events = arena.request(ArenaRequest::LeaseRing(request)); + match (events.len(), events.pop()) { + (1, Some(ArenaEvent::RingLeased { lease })) => Ok(lease), + (1, Some(ArenaEvent::RingLeaseRejected { reason, .. })) => { + Err(BootstrapWriteError::ControlRingLeaseRejected(reason)) + } + (1, Some(ArenaEvent::RingLeaseQueued { .. })) => { + Err(BootstrapWriteError::ControlRingLeaseQueued) + } + _ => Err(BootstrapWriteError::UnexpectedLeaseOutcome), + } + } + + fn zero_region( + arena: &ArenaManager, + start: u64, + end: u64, + ) -> Result<(), BootstrapWriteError> { + let zeros = [0_u8; ZERO_CHUNK]; + let mut offset = start; + while offset < end { + let take = ((end - offset) as usize).min(ZERO_CHUNK); + arena + .write_arena(offset, &zeros[..take]) + .map_err(BootstrapWriteError::Io)?; + offset += take as u64; + } + Ok(()) + } + + fn create_wake_eventfd() -> Result { + // SAFETY: eventfd returns a new owned fd or -1. + let fd = unsafe { libc::eventfd(0, libc::EFD_CLOEXEC) }; + if fd < 0 { + return Err(BootstrapWriteError::FdSetup(std::io::Error::last_os_error())); + } + // SAFETY: fd is a fresh, uniquely owned descriptor. + Ok(unsafe { OwnedFd::from_raw_fd(fd) }) + } + + /// `dup` clears `FD_CLOEXEC` on the copy, which is exactly the child's + /// view: inheritable at exec, and nothing else changes. + fn dup_without_cloexec(fd: std::os::fd::RawFd) -> Result { + // SAFETY: dup returns a new owned fd or -1. + let dup = unsafe { libc::dup(fd) }; + if dup < 0 { + return Err(BootstrapWriteError::FdSetup(std::io::Error::last_os_error())); + } + // SAFETY: dup is a fresh, uniquely owned descriptor. + Ok(unsafe { OwnedFd::from_raw_fd(dup) }) + } +} + +#[cfg(target_os = "linux")] +pub use write::{ + BootstrapWriteError, ControlRingSpec, JobHandoff, CONTROL_RING_GENERATION, write_bootstrap, +}; diff --git a/crates/data-plane/src/lib.rs b/crates/data-plane/src/lib.rs index 2825dcf..5748cb2 100644 --- a/crates/data-plane/src/lib.rs +++ b/crates/data-plane/src/lib.rs @@ -6,6 +6,7 @@ //! ([`edge_wire`]), the arena ([`arena`]), and an application worker port. pub mod arena; +pub mod bootstrap; pub mod edge_lifecycle; pub mod edge_runtime; pub mod edge_wire; diff --git a/crates/data-plane/tests/bootstrap_guarantees.rs b/crates/data-plane/tests/bootstrap_guarantees.rs new file mode 100644 index 0000000..8a3d217 --- /dev/null +++ b/crates/data-plane/tests/bootstrap_guarantees.rs @@ -0,0 +1,352 @@ +//! Bootstrap ABI guarantees (behavior invariants, not implementation shape). +//! +//! Defends: +//! - B3/B4: the parser rejects every malformed page with a typed error and +//! never trusts page claims over ground truth. +//! - B8: the writer emits exactly the two environment ABI names. +//! - B10: after `write_bootstrap` returns, the header is readable from the +//! arena and parseable, the ring region is zeroed with its generation +//! stamped, and header/ring leases are disjoint by arena placement law. +#![cfg(target_os = "linux")] + +use data_plane::arena::{ArenaConfig, ArenaManager, NodeId}; +use data_plane::bootstrap::{ + self, parse_bootstrap, BootstrapError, ControlRingLayout, BOOTSTRAP_MAGIC, BOOTSTRAP_VERSION, + ENV_ARENA_FD, ENV_WAKE_FD, HEADER_END_OFFSET, HEADER_LEN, +}; + +fn arena() -> ArenaManager { + ArenaManager::boot(ArenaConfig { + node_id: NodeId(10), + reservation_ceiling: 1 << 20, + base_alignment: 64, + }) + .expect("test arena must boot") +} + +fn spec() -> bootstrap::ControlRingSpec { + bootstrap::ControlRingSpec { + data_bytes: 4096, + alignment: 64, + } +} + +fn valid_header(arena_size: u64) -> data_plane::bootstrap::BootstrapHeader { + data_plane::bootstrap::BootstrapHeader { + arena_size, + control_ring: ControlRingLayout { + offset: 4096, + capacity: 8192, + generation: 1, + }, + } +} + +// ─── Parser ────────────────────────────────────────────────────────────────── + +#[test] +fn parser_round_trips_encoded_header() { + let page = valid_header(1 << 20).encode(); + let resolved = parse_bootstrap(&page, 1 << 20).expect("valid header must parse"); + assert_eq!(resolved.arena_size, 1 << 20); + assert_eq!(resolved.control_ring, valid_header(1 << 20).control_ring); +} + +#[test] +fn parser_rejects_short_pages() { + let error = parse_bootstrap(&valid_header(1 << 20).encode()[..HEADER_LEN - 1], 1 << 20) + .expect_err("truncated page must fail"); + assert_eq!( + error, + BootstrapError::Truncated { + available: HEADER_LEN - 1 + } + ); +} + +#[test] +fn parser_rejects_bad_magic() { + let mut page = valid_header(1 << 20).encode(); + page[0] ^= 0xFF; + let error = parse_bootstrap(&page, 1 << 20).expect_err("bad magic must fail"); + assert_eq!( + error, + BootstrapError::BadMagic { + found: BOOTSTRAP_MAGIC ^ 0xFF + } + ); +} + +#[test] +fn parser_rejects_unsupported_versions_without_guessing() { + for version in [0_u16, 2, 3, u16::MAX] { + let mut page = valid_header(1 << 20).encode(); + page[4..6].copy_from_slice(&version.to_le_bytes()); + let error = parse_bootstrap(&page, 1 << 20) + .expect_err("unsupported version must not parse"); + assert_eq!( + error, + BootstrapError::UnsupportedVersion { + found: version, + supported: BOOTSTRAP_VERSION + } + ); + } +} + +#[test] +fn parser_rejects_nonzero_reserved_bytes() { + for (at, byte) in [(6_usize, 7_u8), (40, 1), (47, 0xFF)] { + let mut page = valid_header(1 << 20).encode(); + page[at] = byte; + let error = parse_bootstrap(&page, 1 << 20) + .expect_err("nonzero reserved bytes must not parse"); + assert_eq!( + error, + BootstrapError::ReservedBytesNotZero { + at: if at < 8 { 6 } else { 40 } + } + ); + } +} + +#[test] +fn parser_rejects_lying_arena_size() { + let page = valid_header(1 << 20).encode(); + let error = + parse_bootstrap(&page, (1 << 20) + 1).expect_err("size disagreement must fail"); + assert_eq!( + error, + BootstrapError::ArenaSizeMismatch { + header: 1 << 20, + backing: (1 << 20) + 1 + } + ); +} + +#[test] +fn parser_rejects_ring_defects() { + let base = valid_header(1 << 20); + let cases: Vec<(ControlRingLayout, BootstrapError)> = vec![ + ( + ControlRingLayout { + offset: HEADER_END_OFFSET - 8, + ..base.control_ring + }, + BootstrapError::RingOverlapsHeader { + offset: HEADER_END_OFFSET - 8, + }, + ), + ( + ControlRingLayout { + offset: 1 << 20, + capacity: 1, + ..base.control_ring + }, + BootstrapError::RingOutOfBounds { + offset: 1 << 20, + capacity: 1, + arena_size: 1 << 20, + }, + ), + ( + ControlRingLayout { + capacity: 0, + ..base.control_ring + }, + BootstrapError::ZeroRingCapacity, + ), + ( + ControlRingLayout { + generation: 0, + ..base.control_ring + }, + BootstrapError::ZeroRingGeneration, + ), + ]; + for (ring, expected) in cases { + let page = data_plane::bootstrap::BootstrapHeader { + control_ring: ring, + ..base + } + .encode(); + let error = parse_bootstrap(&page, 1 << 20) + .expect_err("defective ring layout must not parse"); + assert_eq!(error, expected); + } +} + +#[test] +fn parser_survives_offset_capacity_wraparound() { + let page = data_plane::bootstrap::BootstrapHeader { + control_ring: ControlRingLayout { + offset: u64::MAX - 8, + capacity: 16, + generation: 1, + }, + arena_size: 1 << 20, + } + .encode(); + let error = parse_bootstrap(&page, 1 << 20).expect_err("wrapping ring must fail"); + assert_eq!( + error, + BootstrapError::RingOutOfBounds { + offset: u64::MAX - 8, + capacity: 16, + arena_size: 1 << 20 + } + ); +} + +/// Deterministic xorshift fuzz: arbitrary page bytes never panic, and a +/// successful parse implies every invariant (B4). +#[test] +fn parser_never_panics_on_arbitrary_pages() { + let mut state = 0x9E3779B97F4A7C15_u64; + let mut next = move || { + state ^= state << 13; + state ^= state >> 7; + state ^= state << 17; + state + }; + for _ in 0..512 { + let mut page = [0_u8; HEADER_LEN]; + for chunk in page.chunks_mut(8) { + chunk.copy_from_slice(&next().to_le_bytes()[..chunk.len()]); + } + match parse_bootstrap(&page, 1 << 20) { + Ok(resolved) => { + let ring = resolved.control_ring; + assert!(ring.capacity > 0); + assert!(ring.generation > 0); + assert!(ring.offset >= HEADER_END_OFFSET); + assert!(ring + .offset + .checked_add(ring.capacity) + .is_some_and(|end| end <= resolved.arena_size)); + } + Err(_) => {} + } + } +} + +// ─── Writer ────────────────────────────────────────────────────────────────── + +#[test] +fn writer_produces_parseable_header_and_disjoint_ring() { + let mut arena = arena(); + let handoff = bootstrap::write_bootstrap(&mut arena, spec()).expect("bootstrap must write"); + + let page = arena.read_arena(0, HEADER_LEN).expect("read header"); + assert_eq!(&page[0..4], b"SWBS", "magic bytes are frozen ABI"); + + let resolved = + parse_bootstrap(&page, arena.arena_len()).expect("written header must self-parse"); + assert_eq!(resolved.arena_size, arena.arena_len()); + assert_eq!(resolved.control_ring, handoff.control_ring); + assert!( + resolved.control_ring.offset >= HEADER_END_OFFSET, + "control ring must not overlap the header region" + ); + assert_eq!( + resolved.control_ring.generation, + bootstrap::CONTROL_RING_GENERATION + ); +} + +#[test] +fn writer_leaves_ring_zeroed_with_generation_stamped() { + let mut arena = arena(); + let handoff = bootstrap::write_bootstrap(&mut arena, spec()).expect("bootstrap must write"); + + let ring = handoff.control_ring; + let bytes = arena + .read_arena(ring.offset, ring.capacity as usize) + .expect("read ring region"); + assert_eq!( + &bytes[0..8], + &bootstrap::CONTROL_RING_GENERATION.to_le_bytes(), + "generation is stamped at ring start" + ); + assert!( + bytes[8..].iter().all(|&b| b == 0), + "rest of the fresh ring must be zero" + ); +} + +#[test] +fn writer_emits_exactly_the_environment_abi() { + use std::os::fd::AsRawFd; + let mut arena = arena(); + let handoff = bootstrap::write_bootstrap(&mut arena, spec()).expect("bootstrap must write"); + + let env = handoff.env; + assert_eq!( + env.keys().collect::>(), + vec![ENV_ARENA_FD, ENV_WAKE_FD], + "B8: the two names are the whole env contract" + ); + assert_eq!( + env[ENV_ARENA_FD], + handoff.arena_fd.as_raw_fd().to_string() + ); + assert_eq!(env[ENV_WAKE_FD], handoff.wake_fd.as_raw_fd().to_string()); +} + +#[test] +fn writer_handoff_fds_are_distinct_and_open() { + use std::fs::File; + use std::io::{Read, Write}; + use std::os::fd::AsRawFd; + + let mut arena = arena(); + let handoff = bootstrap::write_bootstrap(&mut arena, spec()).expect("bootstrap must write"); + + let arena_fd = handoff.arena_fd.as_raw_fd(); + let wake_fd = handoff.wake_fd.as_raw_fd(); + let host_wake_fd = handoff.host_wake_fd.as_raw_fd(); + assert_ne!(arena_fd, wake_fd); + assert_ne!(arena_fd, host_wake_fd); + assert_ne!(wake_fd, host_wake_fd); + + // The wake descriptors must behave like eventfds: write 1, read counter. + let mut wake = File::from(handoff.wake_fd); + wake.write_all(&1_u64.to_le_bytes()) + .expect("eventfd write"); + let mut counter = [0_u8; 8]; + wake.read_exact(&mut counter).expect("eventfd read"); + assert_eq!(counter, 1_u64.to_le_bytes()); +} + +#[test] +fn writer_rejects_invalid_specs_and_oversized_rings() { + let mut arena = arena(); + for bad in [ + bootstrap::ControlRingSpec { + data_bytes: 0, + alignment: 64, + }, + bootstrap::ControlRingSpec { + data_bytes: 4096, + alignment: 0, + }, + ] { + let error = bootstrap::write_bootstrap(&mut arena, bad) + .expect_err("invalid spec must fail before leasing"); + assert!(matches!( + error, + bootstrap::BootstrapWriteError::InvalidControlRingSpec + )); + } + + let oversized = bootstrap::ControlRingSpec { + data_bytes: 1 << 21, + alignment: 64, + }; + let error = bootstrap::write_bootstrap(&mut arena, oversized) + .expect_err("ring larger than the arena must fail"); + assert!(matches!( + error, + bootstrap::BootstrapWriteError::ControlRingLeaseRejected(_) + )); +}