diff --git a/Cargo.lock b/Cargo.lock index c2f741d..97f2c23 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -952,7 +952,10 @@ dependencies = [ name = "data-plane" version = "0.1.0" dependencies = [ + "datastream", "libc", + "serde", + "swactor", ] [[package]] diff --git a/crates/data-plane/Cargo.toml b/crates/data-plane/Cargo.toml index 672b50e..96fe77f 100644 --- a/crates/data-plane/Cargo.toml +++ b/crates/data-plane/Cargo.toml @@ -5,6 +5,9 @@ edition = "2024" publish = false [dependencies] +datastream = { path = "../datastream" } +serde = { version = "1", features = ["derive"] } +swactor = { path = "../.." } [target.'cfg(target_os = "linux")'.dependencies] libc = "0.2" diff --git a/crates/data-plane/DATA_PLANE_ACTOR_SPEC.md b/crates/data-plane/DATA_PLANE_ACTOR_SPEC.md new file mode 100644 index 0000000..495ba62 --- /dev/null +++ b/crates/data-plane/DATA_PLANE_ACTOR_SPEC.md @@ -0,0 +1,523 @@ +# Data Plane Actor Architecture Specification + +**Status:** implemented architecture contract for the `data-plane` crate. + +This document describes the behavioral boundary between reusable data-plane actors +and MVP-specific orchestration/runtime code. It is intentionally architectural: it +names responsibilities, actor roles, message families, and ownership boundaries +without prescribing file layout or migration steps. + +--- + +## 1. Purpose + +`data-plane` owns the behavior required to move model objects between stages and +between a node process and its local GPU worker process. + +The crate defines the actor protocol for: + +- provisioning logical data edges; +- distinguishing network/wire edges from node-local IPC rings; +- leasing, installing, readying, faulting, stopping, quiescing, and releasing + local rings; +- binding logical edges to transport endpoints and local worker rings; +- parsing, validating, sequencing, loading, producing, and reporting objects; +- gating readiness and object visibility on the correct lifecycle transitions; +- translating low-level arena, transport, and worker observations into coarse + data-plane outcomes. + +`mvp-system` uses `data-plane` as a reusable actor subsystem. It provides run +intent, concrete runtime actor addresses, and MVP-specific report sinks. It does +not own the fine-grained data-plane state machine. + +--- + +## 2. Core boundary + +Swactor actors carry control, lifecycle, and identity messages. Payload bytes do +not move through actor mailboxes. + +Payload bytes move through: + +- arena-backed shared-memory rings for node-local IPC; +- transport byte streams for node-to-node data edges; +- GPU-worker-owned device allocations for compute-ready objects. + +The data-plane actors decide when these byte paths are established, readable, +writable, faulted, stopped, and safe to release. Runtime adapter actors execute +concrete effects and report observations back. + +--- + +## 3. Edge and ring model + +### 3.1 Wire edge + +A wire edge is a logical run-plan connection between a producer endpoint and a +consumer endpoint. + +It carries: + +- run identity; +- edge identity; +- producer and consumer node identity; +- edge kind, such as token input, activation, or token output; +- object contract; +- transport contract; +- optional remote endpoint and remote actor identity. + +A wire edge answers: "which logical data stream connects these stage endpoints?" + +It does not answer: "which local arena offset or worker-process ring is being +used on this node?" + +### 3.2 Local IPC ring + +A local IPC ring is a node-local buffer used by a Rust node process and its owned +GPU worker process. + +It carries: + +- ring identity allocated by the local arena manager; +- arena layout and capacity; +- worker-process generation; +- local role port, such as input or output; +- direction relative to the worker process; +- object contract installed into the worker; +- quiescence and release state. + +A local IPC ring answers: "how does this node exchange bytes with its local +worker process for a specific edge endpoint?" + +It does not answer: "which remote node or distributed route owns the other side +of the logical edge?" + +### 3.3 Binding + +A wire edge endpoint may bind to zero or more local resources depending on its +role: + +```text +inbound wire edge endpoint + -> recv transport endpoint + -> local worker ingress ring + -> device object handle + +outbound wire edge endpoint + -> local worker egress ring + -> send transport endpoint + +local-only edge endpoint + -> local producer/consumer binding + -> optional worker ring +``` + +The binding is owned by data-plane state. MVP code may observe the binding only +through coarse reports such as edge ready, object loaded, object produced, edge +faulted, and edge stopped. + +--- + +## 4. Actor topology + +The target topology is actor-oriented. + +### 4.1 Data-plane node actor + +One data-plane node actor owns the data-plane state for one local node within one +active run. + +It owns: + +- local node identity; +- run-scoped edge table; +- mapping from wire edge endpoints to local rings; +- mapping from ring ids to edge endpoints; +- object sequence state; +- device-handle visibility state; +- data-plane child actor addresses; +- MVP report sink addresses. + +It receives provisioning intent from MVP code and observations from runtime +adapters. It emits actor messages to arena, worker, transport, and MVP report +sinks. + +### 4.2 Wire edge actor + +A wire edge actor owns the lifecycle of one logical edge endpoint on the local +node. + +It owns: + +- provisioning state; +- transport establishment state; +- send/receive pump readiness; +- stream faults; +- logical edge readiness; +- stop and fault propagation for that edge endpoint. + +It does not own worker-process state or arena layout details except through a +binding supplied by the data-plane node actor or ring actor. + +### 4.3 Local worker ring actor + +A local worker ring actor owns one local IPC ring lifecycle. + +It owns: + +- arena lease request and result; +- worker ring installation; +- worker ring readable/writable notifications; +- ring fault and quiescence observations; +- release proof collection; +- arena lease release. + +It does not own remote endpoint routing. It can be bound to a wire edge endpoint +by edge id, but the ring lifecycle remains local. + +### 4.4 GPU worker control adapter + +The GPU worker control adapter is the actor-facing boundary to the owned Python +worker process. + +It owns or fronts: + +- worker process generation; +- command serialization to the worker; +- stdout/stderr event parsing; +- device handle generation checks; +- worker stop/crash/restart observations. + +The data plane treats this as an actor endpoint. Worker-process JSON and Python +helper details are not exposed to MVP stage logic. + +### 4.5 Transport adapter actors + +Transport adapter actors own concrete wire byte movement. + +They own or front: + +- accepted edge streams; +- outbound edge streams; +- edge preamble validation; +- byte read/write readiness; +- transport-specific stream faults; +- pump stop observations. + +The data plane treats transport events as observations on a wire edge. Transport +actors do not decide stage readiness or object admission. + +### 4.6 MVP report sink + +The MVP report sink receives coarse data-plane outcomes and maps them to +MVP-specific control messages. + +Examples: + +- inbound edge ready; +- outbound edge ready; +- object loaded for stage execution; +- object produced for downstream transport; +- edge faulted; +- local edges stopped. + +The sink does not inspect ring cursors, arena leases, worker generations, or +transport pump internals. + +--- + +## 5. Actor API surface + +Concrete Rust names are schematic. The contract is the message shape and +ownership boundary. + +### 5.1 Provisioning input + +MVP sends provisioning intent to the data-plane node actor: + +```text +ProvisionDataPlaneRun { + run_id, + local_node_id, + arena_actor, + worker_actor, + transport_actor, + report_sink, +} + +ProvisionWireEdgeEndpoint { + run_id, + edge_id, + direction, + edge_kind, + local_role_port, + local_node_id, + peer_node_id, + peer_endpoint, + object_spec, + transport_spec, + local_ring_spec, +} +``` + +`direction` is relative to the local node's stage role: inbound means the local +stage consumes objects from the edge; outbound means the local stage produces +objects to the edge. + +Provisioning is declarative. MVP describes the intended edge endpoint and the +actors available to execute effects. It does not prescribe lease/install/driver +ordering. + +### 5.2 Runtime observations + +Runtime adapters report observations back to data-plane actors: + +```text +ArenaRingLeased +ArenaRingLeaseRejected +ArenaRingReleased +ArenaRingReleaseRejected + +WorkerReady +WorkerRingInstalled +WorkerRingFaulted +WorkerRingQuiesced +WorkerRingReadable +WorkerRingWritable +WorkerObjectLoaded +WorkerObjectProduced +WorkerObjectFailed +WorkerStopped +WorkerFaulted + +TransportEdgeReady +TransportBytesReceived +TransportBytesSent +TransportStreamClosed +TransportStreamFaulted +TransportPumpStopped +``` + +Observations are facts, not commands. The data plane decides the next state and +any follow-up messages. + +### 5.3 Data-plane effects + +The data plane sends effect requests to runtime adapter actors: + +```text +LeaseArenaRing +CancelArenaRingLease +ReleaseArenaRingLease + +InstallWorkerRing +UninstallWorkerRing +NotifyWorkerRingReadable +NotifyWorkerRingWritable +LoadObjectFromWorkerRing +ExecuteWorkerStep +ReleaseWorkerDeviceObject + +EstablishWireSend +EstablishWireRecv +WriteWireObject +StopWirePump +``` + +Effects are actor messages. The receiving adapter owns the concrete mechanism: +memfd/mmap, JSON stdin/stdout, process supervision, iroh streams, or test doubles. + +### 5.4 Data-plane reports + +The data plane reports only stable semantic outcomes to MVP: + +```text +InboundEdgeReady { edge_id } +OutboundEdgeReady { edge_id } +ObjectLoaded { edge_id, object_id, sequence, device_handle } +ObjectProduced { edge_id, object_id, sequence, extent } +EdgeFaulted { edge_id, reason } +EdgeStopped { edge_id } +LocalEdgesStopped { run_id } +WorkerDataPlaneFaulted { reason } +``` + +Reports are the only data-plane messages MVP stage/orchestrator actors should +need for normal stage progression. + +--- + +## 6. Behavior owned by data-plane + +### 6.1 Edge establishment + +For each provisioned edge endpoint, data-plane actors own the establishment +sequence. + +Inbound endpoint: + +```text +provision endpoint + -> lease local ingress ring + -> install ring into worker input port + -> establish receive transport if the edge is remote + -> report inbound edge ready +``` + +Outbound endpoint: + +```text +provision endpoint + -> lease local egress ring when worker output is required + -> install ring into worker output port + -> establish send transport if the edge is remote + -> report outbound edge ready +``` + +Readiness is reported only after every required local and wire resource for that +endpoint is ready. A local-only endpoint may omit transport establishment. A +wire-only endpoint may omit worker-ring establishment when it terminates outside +the local GPU worker. + +### 6.2 Object ingress + +For inbound data, data-plane actors own object admission. + +The data plane: + +- associates incoming bytes with the correct wire edge and stream; +- validates object framing and object spec constraints; +- preserves sequence ordering required by the edge contract; +- writes or exposes the object through the local ingress ring; +- asks the worker to load the object to device; +- waits for a valid worker object-loaded observation; +- reports object loaded to MVP only after the device handle is current and the + logical object is complete. + +MVP does not parse object headers, track ingress buffers, reload cursors, or gate +object-loaded visibility. + +### 6.3 Object egress + +For outbound data, data-plane actors own object production and forwarding. + +The data plane: + +- receives compute/output observations from the worker; +- binds produced objects to the correct outbound edge and sequence; +- validates object extent and object contract; +- publishes readable/writable state to the worker and transport actors; +- forwards complete object records on the wire when the edge is remote; +- reports object produced or step-visible outcomes to MVP at semantic + boundaries, not cursor boundaries. + +MVP does not decide when a local output ring is readable, when a transport stream +should consume it, or when a produced object is safe to expose downstream. + +### 6.4 Faults + +The data plane owns data movement fault classification and propagation. + +Fault sources include: + +- arena lease rejection or release rejection; +- worker ring installation failure; +- worker ring fault; +- malformed object framing; +- sequence violation; +- worker object load/produce failure; +- transport stream read/write/protocol failure; +- pump stop before quiescence; +- stale worker generation or device handle. + +A fault on one edge endpoint must not silently corrupt another endpoint. The data +plane maps local faults to edge-scoped or worker-scoped reports, starts the +required stop/quiescence path, and emits the appropriate MVP report. + +### 6.5 Stop, quiescence, and release + +The data plane owns teardown ordering for local resources. + +For a bound edge/ring pair, stop requires: + +```text +stop transport pump if present + -> uninstall or quiesce worker ring if installed + -> prove no local reader/writer still uses the ring + -> release arena lease + -> report edge stopped +``` + +Arena release must be gated by quiescence proof. MVP may request run or edge +stop, but it does not supply low-level release proof or decide when a ring is +safe to release. + +--- + +## 7. MVP-system utilization + +`mvp-system` remains responsible for MVP orchestration and stage semantics. + +It owns: + +- run planning and edge assignment; +- stage provisioning authority; +- membership/readiness gates outside the data plane; +- weight loading and role configuration intent; +- stage controller behavior; +- prompt injection and token consumption; +- mapping data-plane reports to MVP lifecycle messages; +- selecting concrete runtime adapters for arena, worker process, and transport. + +For data movement, MVP code acts as a client: + +1. Spawn or obtain actor addresses for the data-plane node actor and required + runtime adapters. +2. Send run and edge provisioning intent to the data-plane node actor. +3. Forward runtime observations from concrete adapter actors when those adapters + are MVP-owned. +4. Receive coarse data-plane reports. +5. Translate those reports into stage-controller or orchestrator messages. + +MVP must not rely on private edge states such as waiting-for-lease, +waiting-for-worker-ring, waiting-for-driver, pump-stopped, ring-quiesced, or +release-ready. Those are data-plane implementation states. + +--- + +## 8. Required invariants + +- `EdgeId` names a logical run-plan edge, not a local ring allocation. +- `RingId` names a local arena-backed IPC ring, not a distributed edge. +- A ring may be bound to an edge endpoint, but the identifiers are not + interchangeable. +- Actor messages carry control and identities, not tensor payload bytes. +- Object-loaded reports are emitted only for complete, validated objects with + current-generation device handles. +- Edge-ready reports are emitted only after required wire and local IPC resources + are ready. +- Arena lease release is gated by local quiescence proof. +- Worker process generation is part of every device-handle validity decision. +- Transport faults and worker faults are classified by data-plane before they + become MVP reports. +- MVP stage logic observes semantic outcomes, never ring cursor mechanics. + +--- + +## 9. Non-goals + +`data-plane` does not own: + +- global run planning; +- placement optimization; +- model layer assignment; +- weight download or weight loading semantics; +- prompt tokenization or output token policy; +- membership convergence; +- provider provisioning; +- concrete iroh endpoint construction; +- concrete Python helper implementation. + +The crate defines reusable actor protocols and data-movement behavior. Concrete +runtime adapters may live beside MVP code, inside reusable support crates, or in +tests, as long as they satisfy the actor contracts above. diff --git a/crates/data-plane/src/actor.rs b/crates/data-plane/src/actor.rs new file mode 100644 index 0000000..0ed7efa --- /dev/null +++ b/crates/data-plane/src/actor.rs @@ -0,0 +1,769 @@ +//! Swactor-facing data-plane actor API. +//! +//! The actor owns lifecycle decisions for logical wire edge endpoints and their +//! bound node-local worker rings. Runtime-specific actors execute the effect +//! messages and report observations back. + +use std::collections::{BTreeMap, BTreeSet}; + +use swactor::actor::{ActorAddress, ActorInterface}; +use swactor::runtime::Ctx; + +use crate::edge_lifecycle as lifecycle; +use crate::object_record; + +pub use lifecycle::{ + DType, EdgeFaultReason, EdgeId, LeaseRequestId, NodeId, ObjectKind, ObjectSpec, + QuiescenceProof, RingDirection, RingId, RingLayout, RingLeaseRejection, RingSpec, RunId, +}; + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct PortId(pub String); + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum EdgeKind { + TokenIn, + Activation, + TokenOut, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum EdgeEndpointDirection { + Inbound, + Outbound, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct PeerEndpoint(pub String); + +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum TransportBinding { + Remote { endpoint: Option }, + LocalOnly, +} + +impl TransportBinding { + fn requires_transport(&self) -> bool { + matches!(self, Self::Remote { .. }) + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum WorkerRingBinding { + Required, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct DataPlaneRunConfig { + pub run_id: RunId, + pub local_node_id: NodeId, + pub arena_actor: ActorAddress, + pub worker_actor: ActorAddress, + pub transport_actor: ActorAddress, + pub report_sink: ActorAddress, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct WireEdgeEndpoint { + pub run_id: RunId, + pub edge_id: EdgeId, + pub direction: EdgeEndpointDirection, + pub kind: EdgeKind, + pub local_node_id: NodeId, + pub peer_node_id: Option, + pub peer_endpoint: Option, + pub local_role_port: PortId, + pub object_spec: ObjectSpec, + pub ring_spec: RingSpec, + pub object_record_spec: object_record::ObjectSpec, + pub transport: TransportBinding, + pub worker_ring: WorkerRingBinding, +} + +impl WireEdgeEndpoint { + fn provision_event(&self) -> Option { + match self.direction { + EdgeEndpointDirection::Inbound => { + Some(lifecycle::EdgeEvent::ProvisionRx(lifecycle::ProvisionRx { + run_id: self.run_id, + edge_id: self.edge_id, + local_node_id: self.local_node_id, + object_spec: self.object_spec, + ring_spec: self.ring_spec, + })) + } + EdgeEndpointDirection::Outbound => { + Some(lifecycle::EdgeEvent::ProvisionTx(lifecycle::ProvisionTx { + run_id: self.run_id, + edge_id: self.edge_id, + local_node_id: self.local_node_id, + consumer_node_id: self.peer_node_id?, + object_spec: self.object_spec, + ring_spec: self.ring_spec, + })) + } + } + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum DataPlaneNodeMsg { + ConfigureRun(DataPlaneRunConfig), + ProvisionWireEdgeEndpoint(WireEdgeEndpoint), + Arena(ArenaObservation), + Worker(WorkerObservation), + Transport(TransportObservation), + StopEdge { edge_id: EdgeId }, + StopRun { run_id: RunId }, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum ArenaObservation { + RingLeased { + request_id: LeaseRequestId, + ring_id: RingId, + layout: RingLayout, + }, + RingLeaseRejected { + request_id: LeaseRequestId, + reason: RingLeaseRejection, + }, + RingReleased { + ring_id: RingId, + }, + RingReleaseRejected { + ring_id: RingId, + }, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum WorkerObservation { + WorkerReady, + RingInstalled { + edge_id: EdgeId, + ring_id: RingId, + }, + RingFaulted { + edge_id: EdgeId, + ring_id: RingId, + reason: lifecycle::RingFaultReason, + }, + RingQuiesced { + ring_id: RingId, + }, + QuiescenceProven { + ring_id: RingId, + }, + ObjectLoaded { + edge_id: EdgeId, + ring_id: RingId, + object_id: object_record::ObjectId, + sequence: u64, + extent: u64, + handle: DeviceHandle, + }, + ObjectProduced { + edge_id: EdgeId, + ring_id: RingId, + object_id: object_record::ObjectId, + sequence: u64, + extent: u64, + }, + ObjectFailed { + edge_id: EdgeId, + ring_id: RingId, + reason: object_record::ObjectFailureReason, + }, + WorkerFaulted, + WorkerStopped, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum TransportObservation { + EdgeReady { + edge_id: EdgeId, + }, + BytesAvailable { + edge_id: EdgeId, + stream_id: StreamId, + byte_count: u64, + }, + StreamClosed { + edge_id: EdgeId, + }, + StreamFaulted { + edge_id: EdgeId, + reason: lifecycle::StreamFaultReason, + }, + PumpStopped { + edge_id: EdgeId, + ring_id: RingId, + }, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum DataPlaneArenaMsg { + LeaseRing { + request_id: LeaseRequestId, + edge_id: EdgeId, + direction: RingDirection, + ring_spec: RingSpec, + }, + CancelQueuedLease { + request_id: LeaseRequestId, + edge_id: EdgeId, + }, + ReleaseRing { + ring_id: RingId, + proof: QuiescenceProof, + }, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum DataPlaneWorkerMsg { + InstallRing { + edge_id: EdgeId, + ring_id: RingId, + direction: RingDirection, + layout: RingLayout, + object_spec: ObjectSpec, + ring_spec: RingSpec, + role_port: PortId, + }, + UninstallRing { + edge_id: EdgeId, + ring_id: RingId, + }, + NotifyRingReadable { + ring_id: RingId, + }, + NotifyRingWritable { + ring_id: RingId, + }, + LoadObjectFromRing { + edge_id: EdgeId, + ring_id: RingId, + object_id: object_record::ObjectId, + sequence: u64, + extent: u64, + }, + ExecuteStep { + edge_id: EdgeId, + sequence: u64, + }, + ReleaseDeviceObject { + handle: DeviceHandle, + }, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum DataPlaneTransportMsg { + EstablishSend { + edge_id: EdgeId, + consumer_node_id: NodeId, + ring_id: RingId, + layout: RingLayout, + }, + EstablishRecv { + edge_id: EdgeId, + ring_id: RingId, + layout: RingLayout, + }, + WriteWireObject { + edge_id: EdgeId, + object_id: object_record::ObjectId, + sequence: u64, + extent: u64, + }, + StopWirePump { + edge_id: EdgeId, + ring_id: RingId, + }, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum DataPlaneReportMsg { + InboundEdgeReady { + edge_id: EdgeId, + }, + OutboundEdgeReady { + edge_id: EdgeId, + }, + ObjectLoaded { + edge_id: EdgeId, + object_id: object_record::ObjectId, + sequence: u64, + extent: u64, + handle: DeviceHandle, + }, + ObjectProduced { + edge_id: EdgeId, + object_id: object_record::ObjectId, + sequence: u64, + extent: u64, + }, + EdgeFaulted { + edge_id: EdgeId, + reason: EdgeFaultReason, + }, + EdgeStopped { + edge_id: EdgeId, + }, + LocalEdgesStopped { + run_id: RunId, + }, + WorkerDataPlaneFaulted { + reason: WorkerDataPlaneFaultReason, + }, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum WorkerDataPlaneFaultReason { + WorkerFaulted, + WorkerStopped, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct WorkerGeneration(pub u64); + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct DeviceHandle { + pub generation: WorkerGeneration, + pub id: u64, +} + +impl DeviceHandle { + pub const fn new(generation: WorkerGeneration, id: u64) -> Self { + Self { generation, id } + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct StreamId(pub u64); + +#[derive(Clone, Debug, PartialEq, Eq)] +struct EdgeBinding { + direction: EdgeEndpointDirection, + role_port_index: usize, + transport: TransportBinding, +} + +pub struct DataPlaneNodeActor { + run: Option, + lifecycle: lifecycle::EdgeEstablisher, + command_cursor: usize, + event_cursor: usize, + bindings: BTreeMap, + stopped_edges: BTreeSet, + role_ports: Vec, + rings: BTreeMap, + pending_reports: Vec, + stopping_run: Option, + local_edges_stopped_reported: bool, +} + +impl DataPlaneNodeActor { + pub fn new(local_node_id: NodeId) -> Self { + Self { + run: None, + lifecycle: lifecycle::EdgeEstablisher::new(local_node_id), + command_cursor: 0, + event_cursor: 0, + bindings: BTreeMap::new(), + stopped_edges: BTreeSet::new(), + role_ports: Vec::new(), + rings: BTreeMap::new(), + pending_reports: Vec::new(), + stopping_run: None, + local_edges_stopped_reported: false, + } + } + + fn configure(&mut self, config: DataPlaneRunConfig) { + self.lifecycle = lifecycle::EdgeEstablisher::new(config.local_node_id); + self.command_cursor = 0; + self.event_cursor = 0; + self.bindings.clear(); + self.role_ports.clear(); + self.stopped_edges.clear(); + self.rings.clear(); + self.pending_reports.clear(); + self.stopping_run = None; + self.local_edges_stopped_reported = false; + self.run = Some(config); + } + + fn observe(&mut self, msg: DataPlaneNodeMsg) { + match msg { + DataPlaneNodeMsg::ConfigureRun(config) => self.configure(config), + DataPlaneNodeMsg::ProvisionWireEdgeEndpoint(endpoint) => { + if let Some(event) = endpoint.provision_event() { + let role_port_index = self.role_ports.len(); + self.role_ports.push(endpoint.local_role_port.clone()); + self.bindings.insert( + endpoint.edge_id, + EdgeBinding { + direction: endpoint.direction, + role_port_index, + transport: endpoint.transport.clone(), + }, + ); + self.lifecycle.observe(event); + } + } + DataPlaneNodeMsg::Arena(observation) => match observation { + ArenaObservation::RingLeased { + request_id, + ring_id, + layout, + } => self.lifecycle.observe(lifecycle::EdgeEvent::RingLeased { + request_id, + ring_id, + layout, + }), + ArenaObservation::RingLeaseRejected { request_id, reason } => self + .lifecycle + .observe(lifecycle::EdgeEvent::RingLeaseRejected { request_id, reason }), + ArenaObservation::RingReleased { .. } + | ArenaObservation::RingReleaseRejected { .. } => {} + }, + DataPlaneNodeMsg::Worker(observation) => match observation { + WorkerObservation::WorkerReady => {} + WorkerObservation::RingInstalled { edge_id, ring_id } => { + self.rings.insert(ring_id, edge_id); + self.lifecycle + .observe(lifecycle::EdgeEvent::RingInstalled { edge_id, ring_id }); + } + WorkerObservation::RingFaulted { + edge_id, + ring_id, + reason, + } => self.lifecycle.observe(lifecycle::EdgeEvent::RingFault { + edge_id, + ring_id, + reason, + }), + WorkerObservation::RingQuiesced { ring_id } => self + .lifecycle + .observe(lifecycle::EdgeEvent::RingQuiesced { ring_id }), + WorkerObservation::QuiescenceProven { ring_id } => self + .lifecycle + .observe(lifecycle::EdgeEvent::QuiescenceProven { ring_id }), + WorkerObservation::ObjectLoaded { + edge_id, + object_id, + sequence, + extent, + handle, + .. + } => self.report_object_loaded(edge_id, object_id, sequence, extent, handle), + WorkerObservation::ObjectProduced { + edge_id, + object_id, + sequence, + extent, + .. + } => self.report_object_produced(edge_id, object_id, sequence, extent), + WorkerObservation::ObjectFailed { + edge_id, ring_id, .. + } => self.lifecycle.observe(lifecycle::EdgeEvent::RingFault { + edge_id, + ring_id, + reason: lifecycle::RingFaultReason::WorkerRejectedRing, + }), + WorkerObservation::WorkerFaulted => { + self.report_worker_fault(WorkerDataPlaneFaultReason::WorkerFaulted) + } + WorkerObservation::WorkerStopped => { + self.report_worker_fault(WorkerDataPlaneFaultReason::WorkerStopped) + } + }, + DataPlaneNodeMsg::Transport(observation) => match observation { + TransportObservation::EdgeReady { edge_id } => self + .lifecycle + .observe(lifecycle::EdgeEvent::DriverEdgeReady { edge_id }), + TransportObservation::BytesAvailable { .. } + | TransportObservation::StreamClosed { .. } => {} + TransportObservation::StreamFaulted { edge_id, reason } => self + .lifecycle + .observe(lifecycle::EdgeEvent::StreamFault { edge_id, reason }), + TransportObservation::PumpStopped { edge_id, ring_id } => self + .lifecycle + .observe(lifecycle::EdgeEvent::PumpStopped { edge_id, ring_id }), + }, + DataPlaneNodeMsg::StopEdge { edge_id } => self + .lifecycle + .observe(lifecycle::EdgeEvent::StopEdge { edge_id }), + DataPlaneNodeMsg::StopRun { run_id } => { + self.stopping_run = Some(run_id); + self.local_edges_stopped_reported = false; + let edge_ids = self.bindings.keys().copied().collect::>(); + for edge_id in edge_ids { + self.lifecycle + .observe(lifecycle::EdgeEvent::StopEdge { edge_id }); + } + self.maybe_report_local_edges_stopped(); + } + } + } + + fn ring_for_edge(&self, edge_id: EdgeId) -> Option { + self.rings + .iter() + .find_map(|(ring_id, seen_edge)| (*seen_edge == edge_id).then_some(*ring_id)) + } + + fn role_port_for(&self, edge_id: EdgeId) -> PortId { + self.bindings + .get(&edge_id) + .and_then(|binding| self.role_ports.get(binding.role_port_index)) + .cloned() + .unwrap_or_else(|| PortId(String::new())) + } + + fn report_object_loaded( + &mut self, + edge_id: EdgeId, + object_id: object_record::ObjectId, + sequence: u64, + extent: u64, + handle: DeviceHandle, + ) { + self.pending_reports.push(DataPlaneReportMsg::ObjectLoaded { + edge_id, + object_id, + sequence, + extent, + handle, + }); + } + + fn report_object_produced( + &mut self, + edge_id: EdgeId, + object_id: object_record::ObjectId, + sequence: u64, + extent: u64, + ) { + self.pending_reports + .push(DataPlaneReportMsg::ObjectProduced { + edge_id, + object_id, + sequence, + extent, + }); + } + + fn report_worker_fault(&mut self, reason: WorkerDataPlaneFaultReason) { + self.pending_reports + .push(DataPlaneReportMsg::WorkerDataPlaneFaulted { reason }); + } + + fn report_local_edges_stopped(&mut self, run_id: RunId) { + self.pending_reports + .push(DataPlaneReportMsg::LocalEdgesStopped { run_id }); + } + + fn maybe_report_local_edges_stopped(&mut self) { + let Some(run_id) = self.stopping_run else { + return; + }; + if self.local_edges_stopped_reported { + return; + } + if self + .bindings + .keys() + .all(|edge_id| self.stopped_edges.contains(edge_id)) + { + self.local_edges_stopped_reported = true; + self.report_local_edges_stopped(run_id); + } + } + + fn drain(&mut self, ctx: &Ctx) { + let Some(run) = self.run.clone() else { + return; + }; + + loop { + let mut progressed = false; + let mut immediate_ready = Vec::new(); + + while self.command_cursor < self.lifecycle.commands().len() { + let command = self.lifecycle.commands()[self.command_cursor].clone(); + self.command_cursor += 1; + progressed = true; + match command { + lifecycle::EdgeCommand::LeaseRing { + request_id, + edge_id, + direction, + ring_spec, + } => { + let _ = ctx.send( + run.arena_actor, + DataPlaneArenaMsg::LeaseRing { + request_id, + edge_id, + direction, + ring_spec, + }, + ); + } + lifecycle::EdgeCommand::CancelQueuedLease { + request_id, + edge_id, + } => { + let _ = ctx.send( + run.arena_actor, + DataPlaneArenaMsg::CancelQueuedLease { + request_id, + edge_id, + }, + ); + } + lifecycle::EdgeCommand::InstallWorkerRing { + edge_id, + ring_id, + direction, + layout, + object_spec, + ring_spec, + } => { + let _ = ctx.send( + run.worker_actor, + DataPlaneWorkerMsg::InstallRing { + edge_id, + ring_id, + direction, + layout, + object_spec, + ring_spec, + role_port: self.role_port_for(edge_id), + }, + ); + } + lifecycle::EdgeCommand::UninstallWorkerRing { edge_id, ring_id } => { + let _ = ctx.send( + run.worker_actor, + DataPlaneWorkerMsg::UninstallRing { edge_id, ring_id }, + ); + } + lifecycle::EdgeCommand::EstablishSend { + edge_id, + consumer_node_id, + layout, + } => { + let ring_id = self.ring_for_edge(edge_id).unwrap_or(RingId(0)); + if self + .bindings + .get(&edge_id) + .is_some_and(|binding| binding.transport.requires_transport()) + { + let _ = ctx.send( + run.transport_actor, + DataPlaneTransportMsg::EstablishSend { + edge_id, + consumer_node_id, + ring_id, + layout, + }, + ); + } else { + immediate_ready.push(edge_id); + } + } + lifecycle::EdgeCommand::EstablishRecv { edge_id, layout } => { + let ring_id = self.ring_for_edge(edge_id).unwrap_or(RingId(0)); + if self + .bindings + .get(&edge_id) + .is_some_and(|binding| binding.transport.requires_transport()) + { + let _ = ctx.send( + run.transport_actor, + DataPlaneTransportMsg::EstablishRecv { + edge_id, + ring_id, + layout, + }, + ); + } else { + immediate_ready.push(edge_id); + } + } + lifecycle::EdgeCommand::StopPump { edge_id, ring_id } => { + let _ = ctx.send( + run.transport_actor, + DataPlaneTransportMsg::StopWirePump { edge_id, ring_id }, + ); + } + lifecycle::EdgeCommand::ReleaseArenaLease { ring_id, proof } => { + let _ = ctx.send( + run.arena_actor, + DataPlaneArenaMsg::ReleaseRing { ring_id, proof }, + ); + } + } + } + + for edge_id in immediate_ready { + self.lifecycle + .observe(lifecycle::EdgeEvent::DriverEdgeReady { edge_id }); + } + + while self.event_cursor < self.lifecycle.events().len() { + let event = self.lifecycle.events()[self.event_cursor].clone(); + self.event_cursor += 1; + progressed = true; + match event { + lifecycle::EdgeLifecycleEvent::EdgeReady { edge_id, .. } => { + let report = + match self.bindings.get(&edge_id).map(|binding| binding.direction) { + Some(EdgeEndpointDirection::Inbound) => { + DataPlaneReportMsg::InboundEdgeReady { edge_id } + } + Some(EdgeEndpointDirection::Outbound) => { + DataPlaneReportMsg::OutboundEdgeReady { edge_id } + } + None => continue, + }; + let _ = ctx.send(run.report_sink, report); + } + lifecycle::EdgeLifecycleEvent::EdgeFaulted { edge_id, reason } => { + let _ = ctx.send( + run.report_sink, + DataPlaneReportMsg::EdgeFaulted { edge_id, reason }, + ); + } + lifecycle::EdgeLifecycleEvent::EdgeStopped { edge_id } => { + self.stopped_edges.insert(edge_id); + let _ = + ctx.send(run.report_sink, DataPlaneReportMsg::EdgeStopped { edge_id }); + self.maybe_report_local_edges_stopped(); + } + } + } + + for report in self.pending_reports.drain(..) { + progressed = true; + let _ = ctx.send(run.report_sink, report); + } + + if !progressed { + break; + } + } + } +} + +impl ActorInterface for DataPlaneNodeActor { + type Incoming = DataPlaneNodeMsg; + type Response = (); + + fn handle(&mut self, ctx: &Ctx, msg: Self::Incoming) { + self.observe(msg); + self.drain(ctx); + } +} diff --git a/crates/data-plane/src/arena.rs b/crates/data-plane/src/arena.rs index e610f30..ef9b572 100644 --- a/crates/data-plane/src/arena.rs +++ b/crates/data-plane/src/arena.rs @@ -1,7 +1,10 @@ //! Reusable arena-backed ring allocation contracts. use std::collections::{BTreeMap, VecDeque}; -use std::time::{SystemTime, UNIX_EPOCH}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use datastream::Record; +use serde::{Deserialize, Serialize}; #[derive(Clone, Debug, PartialEq, Eq)] pub struct ArenaSnapshot { @@ -17,6 +20,44 @@ pub struct ArenaSnapshot { pub release_failures_total: u64, } +pub const ARENA_SAMPLE_CHANNEL: &str = "mvp.arena"; +pub const ARENA_SAMPLE_INTERVAL: Duration = Duration::from_secs(1); + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct ArenaSample { + pub seq: u64, + pub sample_unix_ms: u64, + pub capacity_bytes: u64, + pub live_bytes: u64, + pub free_bytes: u64, + pub active_leases: u64, + pub pending_leases: u64, + pub largest_free_range_bytes: u64, + pub allocation_failures_total: u64, + pub release_failures_total: u64, +} + +impl Record for ArenaSample { + const CHANNEL: &'static str = ARENA_SAMPLE_CHANNEL; +} + +impl From for ArenaSample { + fn from(snapshot: ArenaSnapshot) -> Self { + Self { + seq: snapshot.seq, + sample_unix_ms: snapshot.sample_unix_ms, + capacity_bytes: snapshot.capacity_bytes, + live_bytes: snapshot.live_bytes, + free_bytes: snapshot.free_bytes, + active_leases: snapshot.active_leases, + pending_leases: snapshot.pending_leases, + largest_free_range_bytes: snapshot.largest_free_range_bytes, + allocation_failures_total: snapshot.allocation_failures_total, + release_failures_total: snapshot.release_failures_total, + } + } +} + #[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)] pub struct NodeId(pub u64); diff --git a/crates/mvp-system/src/node_data/edge_actor.rs b/crates/data-plane/src/edge_actor.rs similarity index 100% rename from crates/mvp-system/src/node_data/edge_actor.rs rename to crates/data-plane/src/edge_actor.rs diff --git a/crates/mvp-system/src/node/edge_lifecycle.rs b/crates/data-plane/src/edge_lifecycle.rs similarity index 100% rename from crates/mvp-system/src/node/edge_lifecycle.rs rename to crates/data-plane/src/edge_lifecycle.rs diff --git a/crates/mvp-system/src/worker/egress.rs b/crates/data-plane/src/egress.rs similarity index 99% rename from crates/mvp-system/src/worker/egress.rs rename to crates/data-plane/src/egress.rs index b03f757..8558d86 100644 --- a/crates/mvp-system/src/worker/egress.rs +++ b/crates/data-plane/src/egress.rs @@ -7,7 +7,7 @@ pub struct EdgeId(pub u64); #[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)] pub struct PortId(pub String); -pub use data_plane::object_record::{ +pub use crate::object_record::{ FLAG_BEGIN_SEQUENCE, FLAG_END_OF_SEQUENCE, HEADER_LEN, KNOWN_FLAGS_MASK, OBJECT_MAGIC, OBJECT_MAGIC_BYTES, OBJECT_VERSION, ObjectFlags, ObjectHeader, ObjectId, ObjectLayout, ObjectSpec, @@ -127,7 +127,6 @@ pub enum WakeHint { RingWritable { ring_id: RingId }, } -#[cfg(test)] #[derive(Clone, Debug, PartialEq, Eq)] struct PendingStep { step_id: StepId, @@ -135,7 +134,6 @@ struct PendingStep { role_state_updated: bool, } -#[cfg(test)] pub struct EgressProducerHarness { generation: WorkerGeneration, rings: std::collections::BTreeMap, @@ -149,7 +147,6 @@ pub struct EgressProducerHarness { events: Vec, } -#[cfg(test)] impl EgressProducerHarness { pub fn new(generation: WorkerGeneration) -> Self { Self { @@ -369,7 +366,6 @@ impl EgressProducerHarness { } } -#[cfg(test)] fn encode_header(output: OutputBinding) -> Vec { ObjectHeader { object_id: output.object_id, diff --git a/crates/mvp-system/src/node_data/ingress.rs b/crates/data-plane/src/ingress.rs similarity index 99% rename from crates/mvp-system/src/node_data/ingress.rs rename to crates/data-plane/src/ingress.rs index b445ddf..709cefa 100644 --- a/crates/mvp-system/src/node_data/ingress.rs +++ b/crates/data-plane/src/ingress.rs @@ -7,7 +7,7 @@ pub struct EdgeId(pub u64); #[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)] pub struct PortId(pub String); -pub use data_plane::object_record::{ +pub use crate::object_record::{ FLAG_BEGIN_SEQUENCE, FLAG_END_OF_SEQUENCE, HEADER_LEN, KNOWN_FLAGS_MASK, OBJECT_MAGIC, OBJECT_MAGIC_BYTES, OBJECT_VERSION, ObjectFailureReason, ObjectFlags, ObjectHeader, ObjectId, ObjectLayout, ObjectRecord, ObjectRecordBuilder, ObjectRecordRead, ObjectSpec, @@ -106,7 +106,6 @@ pub struct DeviceCopyLog { pub byte_count: u64, } -#[cfg(test)] #[derive(Clone, Debug, PartialEq, Eq)] struct PendingObject { record: ObjectRecord, @@ -114,7 +113,6 @@ struct PendingObject { handle: Option, } -#[cfg(test)] pub struct IngressParserHarness { generation: WorkerGeneration, install: Option, @@ -128,7 +126,6 @@ pub struct IngressParserHarness { events: Vec, } -#[cfg(test)] impl IngressParserHarness { pub fn new(generation: WorkerGeneration) -> Self { Self { @@ -300,7 +297,6 @@ impl IngressParserHarness { } } -#[cfg(test)] fn object_failure_metadata( bytes: &[u8], reason: ObjectFailureReason, diff --git a/crates/data-plane/src/lib.rs b/crates/data-plane/src/lib.rs index 7b76010..a7bf73a 100644 --- a/crates/data-plane/src/lib.rs +++ b/crates/data-plane/src/lib.rs @@ -1,5 +1,11 @@ -//! Reusable local data-plane contracts for arena-backed byte movement. +//! Reusable actor-oriented data-plane contracts for wire edges, local IPC rings, +//! arena-backed byte movement, and GPU worker object movement. +pub mod actor; pub mod arena; +pub mod edge_actor; +pub mod edge_lifecycle; +pub mod egress; +pub mod ingress; pub mod object_record; pub mod ring; diff --git a/crates/data-plane/tests/data_plane_actor_guarantees.rs b/crates/data-plane/tests/data_plane_actor_guarantees.rs new file mode 100644 index 0000000..db8ad93 --- /dev/null +++ b/crates/data-plane/tests/data_plane_actor_guarantees.rs @@ -0,0 +1,231 @@ +use data_plane::actor as dp; +use data_plane::object_record; +use swactor::config::RuntimeConfig; +use swactor::runtime::Runtime; + +fn object_spec() -> dp::ObjectSpec { + dp::ObjectSpec { + kind: dp::ObjectKind::Activation, + dtype: dp::DType::F16, + max_extent_bytes: 4096, + } +} + +fn record_spec() -> object_record::ObjectSpec { + object_record::ObjectSpec { + max_extent: 4096, + alignment: 4, + layout: object_record::ObjectLayout::Token, + } +} + +fn ring_spec() -> dp::RingSpec { + dp::RingSpec { + header_bytes: 128, + data_bytes: 4096, + alignment: 64, + } +} + +fn layout() -> dp::RingLayout { + dp::RingLayout { + start_offset: 0, + header_offset: 0, + data_offset: 128, + end_offset: 4224, + data_bytes: 4096, + alignment: 64, + } +} + +fn inbound_endpoint() -> dp::WireEdgeEndpoint { + dp::WireEdgeEndpoint { + run_id: dp::RunId(55), + edge_id: dp::EdgeId(7001), + direction: dp::EdgeEndpointDirection::Inbound, + kind: dp::EdgeKind::Activation, + local_node_id: dp::NodeId(10), + peer_node_id: Some(dp::NodeId(9)), + peer_endpoint: Some(dp::PeerEndpoint("node-9".to_owned())), + local_role_port: dp::PortId("input".to_owned()), + object_spec: object_spec(), + ring_spec: ring_spec(), + object_record_spec: record_spec(), + transport: dp::TransportBinding::Remote { + endpoint: Some(dp::PeerEndpoint("node-9".to_owned())), + }, + worker_ring: dp::WorkerRingBinding::Required, + } +} + +#[test] +fn inbound_wire_edge_establishes_through_arena_worker_transport_then_reports_ready() { + let runtime = Runtime::new(RuntimeConfig::default()); + let arena = runtime + .new_inbox::() + .expect("arena inbox"); + let worker = runtime + .new_inbox::() + .expect("worker inbox"); + let transport = runtime + .new_inbox::() + .expect("transport inbox"); + let reports = runtime + .new_inbox::() + .expect("report inbox"); + let actor = runtime + .spawn(dp::DataPlaneNodeActor::new(dp::NodeId(10))) + .expect("spawn data-plane actor"); + + runtime + .send_to( + actor, + dp::DataPlaneNodeMsg::ConfigureRun(dp::DataPlaneRunConfig { + run_id: dp::RunId(55), + local_node_id: dp::NodeId(10), + arena_actor: *arena.addr(), + worker_actor: *worker.addr(), + transport_actor: *transport.addr(), + report_sink: *reports.addr(), + }), + ) + .expect("configure run"); + runtime + .send_to( + actor, + dp::DataPlaneNodeMsg::ProvisionWireEdgeEndpoint(inbound_endpoint()), + ) + .expect("provision inbound"); + runtime.tick(); + + assert_eq!( + arena.try_recv(), + Some(dp::DataPlaneArenaMsg::LeaseRing { + request_id: dp::LeaseRequestId(1), + edge_id: dp::EdgeId(7001), + direction: dp::RingDirection::Ingress, + ring_spec: ring_spec(), + }) + ); + + runtime + .send_to( + actor, + dp::DataPlaneNodeMsg::Arena(dp::ArenaObservation::RingLeased { + request_id: dp::LeaseRequestId(1), + ring_id: dp::RingId(8001), + layout: layout(), + }), + ) + .expect("ring leased"); + runtime.tick(); + + assert_eq!( + worker.try_recv(), + Some(dp::DataPlaneWorkerMsg::InstallRing { + edge_id: dp::EdgeId(7001), + ring_id: dp::RingId(8001), + direction: dp::RingDirection::Ingress, + layout: layout(), + object_spec: object_spec(), + ring_spec: ring_spec(), + role_port: dp::PortId("input".to_owned()), + }) + ); + + runtime + .send_to( + actor, + dp::DataPlaneNodeMsg::Worker(dp::WorkerObservation::RingInstalled { + edge_id: dp::EdgeId(7001), + ring_id: dp::RingId(8001), + }), + ) + .expect("worker installed"); + runtime.tick(); + + assert_eq!( + transport.try_recv(), + Some(dp::DataPlaneTransportMsg::EstablishRecv { + edge_id: dp::EdgeId(7001), + ring_id: dp::RingId(8001), + layout: layout(), + }) + ); + + runtime + .send_to( + actor, + dp::DataPlaneNodeMsg::Transport(dp::TransportObservation::EdgeReady { + edge_id: dp::EdgeId(7001), + }), + ) + .expect("transport ready"); + runtime.tick(); + + assert_eq!( + reports.try_recv(), + Some(dp::DataPlaneReportMsg::InboundEdgeReady { + edge_id: dp::EdgeId(7001) + }) + ); +} + +#[test] +fn object_loaded_observation_is_reported_as_coarse_data_plane_outcome() { + let runtime = Runtime::new(RuntimeConfig::default()); + let arena = runtime + .new_inbox::() + .expect("arena inbox"); + let worker = runtime + .new_inbox::() + .expect("worker inbox"); + let transport = runtime + .new_inbox::() + .expect("transport inbox"); + let reports = runtime + .new_inbox::() + .expect("report inbox"); + let actor = runtime + .spawn(dp::DataPlaneNodeActor::new(dp::NodeId(10))) + .expect("spawn data-plane actor"); + + runtime + .send_to( + actor, + dp::DataPlaneNodeMsg::ConfigureRun(dp::DataPlaneRunConfig { + run_id: dp::RunId(55), + local_node_id: dp::NodeId(10), + arena_actor: *arena.addr(), + worker_actor: *worker.addr(), + transport_actor: *transport.addr(), + report_sink: *reports.addr(), + }), + ) + .expect("configure run"); + runtime + .send_to( + actor, + dp::DataPlaneNodeMsg::Worker(dp::WorkerObservation::ObjectLoaded { + edge_id: dp::EdgeId(7001), + ring_id: dp::RingId(8001), + object_id: object_record::ObjectId(9000), + sequence: 7, + extent: 16, + handle: dp::DeviceHandle::new(dp::WorkerGeneration(3), 42), + }), + ) + .expect("object loaded"); + runtime.tick(); + + assert_eq!( + reports.try_recv(), + Some(dp::DataPlaneReportMsg::ObjectLoaded { + edge_id: dp::EdgeId(7001), + object_id: object_record::ObjectId(9000), + sequence: 7, + extent: 16, + handle: dp::DeviceHandle::new(dp::WorkerGeneration(3), 42), + }) + ); +} diff --git a/crates/mvp-system/src/tests/edge_establisher_guarantees.rs b/crates/data-plane/tests/edge_lifecycle_guarantees.rs similarity index 98% rename from crates/mvp-system/src/tests/edge_establisher_guarantees.rs rename to crates/data-plane/tests/edge_lifecycle_guarantees.rs index 1f0e029..d674fad 100644 --- a/crates/mvp-system/src/tests/edge_establisher_guarantees.rs +++ b/crates/data-plane/tests/edge_lifecycle_guarantees.rs @@ -1,4 +1,4 @@ -//! Black-box contract tests for MVP edge establishment. +//! Black-box contract tests for data-plane edge establishment. //! //! These tests intentionally know only the public EdgeEstablisher surface: //! @@ -7,9 +7,9 @@ //! - arena, worker/token, driver, ready, fault, and release commands out //! //! They assert the guarantees in -//! `specs/mvp_system/edge_establisher_contract.md`. +//! the reusable data-plane edge lifecycle contract. -use mvp_system::node::edge_lifecycle as edge; +use data_plane::edge_lifecycle as edge; // A send provision carries the consumer node id because the driver must know // where to send. It deliberately carries no remote actor address. diff --git a/crates/mvp-system/src/tests/gpu_worker_egress_producer_guarantees.rs b/crates/data-plane/tests/egress_guarantees.rs similarity index 97% rename from crates/mvp-system/src/tests/gpu_worker_egress_producer_guarantees.rs rename to crates/data-plane/tests/egress_guarantees.rs index 7076fc6..f30f96f 100644 --- a/crates/mvp-system/src/tests/gpu_worker_egress_producer_guarantees.rs +++ b/crates/data-plane/tests/egress_guarantees.rs @@ -1,16 +1,15 @@ -//! Black-box contract tests for MVP GPU worker egress production. +//! Black-box contract tests for data-plane GPU worker egress production. //! -//! These tests intentionally know only the public worker egress surface: +//! These tests intentionally know only the public data-plane egress surface: //! //! - `InstallRing`, `ExecuteStep` output bindings, device-copy outcomes, //! backpressure, and shutdown events in //! - committed ring bytes, cursor publication, `ObjectProduced`, //! `StepCompleted`, and step failures out //! -//! They assert the guarantees in -//! `specs/mvp_system/gpu_worker_egress_producer_contract.md`. +//! They assert the reusable data-plane egress producer contract. -use mvp_system::node_data::egress; +use data_plane::egress; // A valid egress ring config supplies the edge object spec and current worker // generation. The producer remains free to choose copy scheduling internally. diff --git a/crates/mvp-system/src/lib.rs b/crates/mvp-system/src/lib.rs index dd7c7b1..2c6e8cb 100644 --- a/crates/mvp-system/src/lib.rs +++ b/crates/mvp-system/src/lib.rs @@ -5,7 +5,6 @@ extern crate self as mvp_system; pub mod chat; pub mod node; -pub mod node_data; pub mod observability; pub mod orchestration; pub mod prompt; diff --git a/crates/mvp-system/src/node/data_plane_bridge.rs b/crates/mvp-system/src/node/data_plane_bridge.rs new file mode 100644 index 0000000..bf9d9c4 --- /dev/null +++ b/crates/mvp-system/src/node/data_plane_bridge.rs @@ -0,0 +1,69 @@ +//! MVP adapter for coarse reusable data-plane actor reports. + +use data_plane::actor as dp; +use swactor::actor::{ActorAddress, ActorInterface}; +use swactor::runtime::Ctx; + +use crate::node::actor::NodeAgentMsg; + +pub struct MvpDataPlaneReportSinkActor { + node_agent: ActorAddress, +} + +impl MvpDataPlaneReportSinkActor { + pub fn new(node_agent: ActorAddress) -> Self { + Self { node_agent } + } +} + +impl ActorInterface for MvpDataPlaneReportSinkActor { + type Incoming = dp::DataPlaneReportMsg; + type Response = (); + + fn handle(&mut self, ctx: &Ctx, msg: Self::Incoming) { + match msg { + dp::DataPlaneReportMsg::InboundEdgeReady { edge_id } => { + let _ = ctx.send( + self.node_agent, + NodeAgentMsg::MarkInboundEdgeReady { edge_id: edge_id.0 }, + ); + } + dp::DataPlaneReportMsg::OutboundEdgeReady { edge_id } => { + let _ = ctx.send( + self.node_agent, + NodeAgentMsg::MarkOutboundEdgeReady { edge_id: edge_id.0 }, + ); + } + dp::DataPlaneReportMsg::ObjectLoaded { + edge_id, + object_id, + sequence, + handle, + .. + } => { + let _ = ctx.send( + self.node_agent, + NodeAgentMsg::ObjectLoaded { + edge_id: edge_id.0, + object_id: object_id.0, + sequence, + handle_generation: handle.generation.0, + handle_id: handle.id, + }, + ); + } + dp::DataPlaneReportMsg::ObjectProduced { .. } => {} + dp::DataPlaneReportMsg::EdgeFaulted { .. } + | dp::DataPlaneReportMsg::WorkerDataPlaneFaulted { .. } => { + let _ = ctx.send(self.node_agent, NodeAgentMsg::WorkerCrashed); + } + dp::DataPlaneReportMsg::EdgeStopped { .. } => {} + dp::DataPlaneReportMsg::LocalEdgesStopped { run_id } => { + let _ = ctx.send( + self.node_agent, + NodeAgentMsg::LocalEdgesStopped { run_id: run_id.0 }, + ); + } + } + } +} diff --git a/crates/mvp-system/src/node/mod.rs b/crates/mvp-system/src/node/mod.rs index 112c44c..4211389 100644 --- a/crates/mvp-system/src/node/mod.rs +++ b/crates/mvp-system/src/node/mod.rs @@ -5,5 +5,5 @@ pub mod actor; pub mod boot_lifecycle; -pub mod edge_lifecycle; +pub mod data_plane_bridge; pub mod worker_node_runtime; diff --git a/crates/mvp-system/src/node/worker_node_runtime.rs b/crates/mvp-system/src/node/worker_node_runtime.rs index 83b2e45..ec37e82 100644 --- a/crates/mvp-system/src/node/worker_node_runtime.rs +++ b/crates/mvp-system/src/node/worker_node_runtime.rs @@ -26,9 +26,6 @@ use crate::node::actor::{ NodeAgentActor, NodeAgentMsg, NodeAgentReport, StageCommandWire, StageInboundEdgeWire, StageObjectSpecWire, StageOutboundEdgeWire, StageRingSpecWire, }; -use crate::node::edge_lifecycle as edge; -use crate::node_data::arena; -use crate::node_data::ingress; use crate::observability::benchmark; use crate::orchestration::distribution_stack::DistributionRuntimeStack; use crate::orchestration::provider_adapters::relay::relay_runtime_config_from_env; @@ -41,6 +38,9 @@ use crate::transport::driver_pumps as driver_model; use crate::transport::endpoint_advertisement::{ EndpointAddrMask, MVP_IROH_ENDPOINT_ADDR_MASK_ENV, advertised_endpoint, }; +use data_plane::arena; +use data_plane::edge_lifecycle as edge; +use data_plane::ingress; use distribution::node::DistributedNodeConfig; use distribution::swim::telemetry::ObservedProbeEvent; use distribution::telemetry::{MembershipTransition, SwimProbeEvent}; @@ -737,7 +737,7 @@ fn spawn_arena_sampler( loop { interval.tick().await; - let sample = arena_manager.lock().sample(seq); + let sample: arena::ArenaSample = arena_manager.lock().sample(seq).into(); seq = seq.saturating_add(1); producer.submit_record(channel, &sample); } diff --git a/crates/mvp-system/src/node_data/arena.rs b/crates/mvp-system/src/node_data/arena.rs deleted file mode 100644 index 8feb92b..0000000 --- a/crates/mvp-system/src/node_data/arena.rs +++ /dev/null @@ -1,131 +0,0 @@ -//! MVP arena-manager adapter over reusable data-plane arena contracts. - -use std::time::Duration; - -use datastream::Record; -use serde::{Deserialize, Serialize}; - -pub use data_plane::arena::{ - ArenaCommand, ArenaConfig, ArenaEvent, ArenaFault, ArenaRequest, LayoutPointer, LeaseRequestId, - LeaseRing, NodeId, QuiescenceProof, RingId, RingLayout, RingLease, RingLeaseRejection, - RingReleaseRejection, RingSpec, -}; - -pub const ARENA_SAMPLE_CHANNEL: &str = "mvp.arena"; -pub const ARENA_SAMPLE_INTERVAL: Duration = Duration::from_secs(1); - -#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] -pub struct ArenaSample { - pub seq: u64, - pub sample_unix_ms: u64, - pub capacity_bytes: u64, - pub live_bytes: u64, - pub free_bytes: u64, - pub active_leases: u64, - pub pending_leases: u64, - pub largest_free_range_bytes: u64, - pub allocation_failures_total: u64, - pub release_failures_total: u64, -} - -impl Record for ArenaSample { - const CHANNEL: &'static str = ARENA_SAMPLE_CHANNEL; -} - -impl From for ArenaSample { - fn from(snapshot: data_plane::arena::ArenaSnapshot) -> Self { - Self { - seq: snapshot.seq, - sample_unix_ms: snapshot.sample_unix_ms, - capacity_bytes: snapshot.capacity_bytes, - live_bytes: snapshot.live_bytes, - free_bytes: snapshot.free_bytes, - active_leases: snapshot.active_leases, - pending_leases: snapshot.pending_leases, - largest_free_range_bytes: snapshot.largest_free_range_bytes, - allocation_failures_total: snapshot.allocation_failures_total, - release_failures_total: snapshot.release_failures_total, - } - } -} - -pub struct ArenaManager { - inner: data_plane::arena::ArenaManager, -} - -impl ArenaManager { - pub fn boot(config: ArenaConfig) -> Result { - data_plane::arena::ArenaManager::boot(config).map(|inner| Self { inner }) - } - - pub fn request(&mut self, request: ArenaRequest) -> Vec { - self.inner.request(request) - } - - pub fn live_leases(&self) -> &[RingLease] { - self.inner.live_leases() - } - - pub fn lookup_lease(&self, ring_id: RingId) -> Option<&RingLease> { - self.inner.lookup_lease(ring_id) - } - - pub fn sample(&self, seq: u64) -> ArenaSample { - self.inner.sample(seq).into() - } - - #[cfg(target_os = "linux")] - pub fn arena_fd(&self) -> std::os::fd::RawFd { - self.inner.arena_fd() - } - - #[cfg(target_os = "linux")] - pub fn write_arena(&self, offset: u64, bytes: &[u8]) -> Result<(), std::io::Error> { - self.inner.write_arena(offset, bytes) - } - - #[cfg(target_os = "linux")] - pub fn read_arena(&self, offset: u64, len: usize) -> Result, std::io::Error> { - self.inner.read_arena(offset, len) - } -} - -pub struct ArenaManagerHarness { - manager: ArenaManager, - events: Vec, - commands: Vec, -} - -impl ArenaManagerHarness { - pub fn boot(config: ArenaConfig) -> Result { - Ok(Self { - manager: ArenaManager::boot(config)?, - events: Vec::new(), - commands: Vec::new(), - }) - } - - pub fn request(&mut self, request: ArenaRequest) { - self.events.extend(self.manager.request(request)); - } - - pub fn events(&self) -> &[ArenaEvent] { - &self.events - } - - pub fn commands(&self) -> &[ArenaCommand] { - &self.commands - } - - pub fn live_leases(&self) -> &[RingLease] { - self.manager.live_leases() - } - - pub fn lookup_lease(&self, ring_id: RingId) -> Option<&RingLease> { - self.manager.lookup_lease(ring_id) - } - - pub fn sample(&self, seq: u64) -> ArenaSample { - self.manager.sample(seq) - } -} diff --git a/crates/mvp-system/src/node_data/mod.rs b/crates/mvp-system/src/node_data/mod.rs deleted file mode 100644 index 40b5c4c..0000000 --- a/crates/mvp-system/src/node_data/mod.rs +++ /dev/null @@ -1,25 +0,0 @@ -//! MVP node-local data-plane adapter public surface. -//! -//! Reusable arena, ring, and object-record contracts live in `data-plane`. -//! This module binds those contracts to MVP worker ingress, worker egress, -//! and edge actor behavior. - -pub mod arena; -pub mod edge_actor; -pub mod ingress; - -pub mod ring { - pub use data_plane::ring::*; -} - -pub mod object { - pub use data_plane::object_record::*; -} - -pub mod egress { - pub use crate::worker::egress::*; -} - -pub mod reusable { - pub use data_plane::{arena, object_record, ring}; -} diff --git a/crates/mvp-system/src/observability/telemetry.rs b/crates/mvp-system/src/observability/telemetry.rs index 60aa214..a6e218d 100644 --- a/crates/mvp-system/src/observability/telemetry.rs +++ b/crates/mvp-system/src/observability/telemetry.rs @@ -4,9 +4,9 @@ use datastream::hardware::net::HostNetSample; use datastream::{ChannelRegistry, Record}; use serde::{Deserialize, Serialize}; -use crate::node_data::arena::ArenaSample; use crate::observability::lifecycle as obs; use crate::orchestration::provisioning::{self, ProvisionLogStream}; +use data_plane::arena::ArenaSample; /// Structured MVP lifecycle facts: run, node, stage, edge, ring, object, step, and worker events. pub const MVP_LIFECYCLE: &str = "mvp.lifecycle"; diff --git a/crates/mvp-system/src/orchestration/app.rs b/crates/mvp-system/src/orchestration/app.rs index 89be1e5..563795f 100644 --- a/crates/mvp-system/src/orchestration/app.rs +++ b/crates/mvp-system/src/orchestration/app.rs @@ -22,7 +22,6 @@ use crate::orchestration::config::{DEFAULT_CONFIG_PATH, TomlConfigOverlay}; use crate::transport::codec_registry::register_mvp_actor_codecs; const PROVIDER_START_MAX_ATTEMPTS: usize = 4; -use crate::node_data::object as ingress; use crate::observability::telemetry::{ MVP_PROVISIONING_EVENTS, MvpProvisionEventRecord, MvpProvisionLogRecord, mvp_provision_log_channel, @@ -53,6 +52,7 @@ use crate::staging::gguf_shard::{StageShardPlan, plan_stage_shard}; use crate::transport::endpoint_advertisement::{ EndpointAddrMask, MVP_IROH_ENDPOINT_ADDR_MASK_ENV, advertised_endpoint, }; +use data_plane::object_record as ingress; use datastream::{ ChannelContent, ChannelId, ChannelRef, DatastreamEndpoint, DatastreamEvent, DatastreamProducer, DatastreamPublisherMsg, DatastreamSubscribe, Frame, Lifetime, NodeId, Record, StreamDescriptor, diff --git a/crates/mvp-system/src/tests/arena_manager_guarantees.rs b/crates/mvp-system/src/tests/arena_manager_guarantees.rs deleted file mode 100644 index 576e9d7..0000000 --- a/crates/mvp-system/src/tests/arena_manager_guarantees.rs +++ /dev/null @@ -1,416 +0,0 @@ -//! Black-box contract tests for MVP ArenaManager behavior. -//! -//! These tests intentionally know only the public arena-manager surface: -//! -//! - arena boot configuration -//! - lease, cancel, release, quiescence, and shutdown requests in -//! - lease events, rejected requests, released ranges, and faults out -//! -//! They assert the guarantees in -//! `specs/mvp_system/arena_manager_contract.md`. - -use mvp_system::node_data::arena; - -// A small deterministic arena makes overlap, alignment, queueing, and reuse -// proofs easy to inspect. The concrete mmap strategy remains outside the test. -fn arena_config() -> arena::ArenaConfig { - arena::ArenaConfig { - node_id: arena::NodeId(10), - reservation_ceiling: 4096, - base_alignment: 64, - } -} - -// Lease requests name size and alignment only. They do not request pointers or -// private allocator slots, which keeps layout authority inside ArenaManager. -fn lease_request(request_id: u64, bytes: u64, alignment: u64) -> arena::LeaseRing { - arena::LeaseRing { - request_id: arena::LeaseRequestId(request_id), - ring_spec: arena::RingSpec { - header_bytes: 128, - data_bytes: bytes, - alignment, - }, - } -} - -// The harness exposes public lease events and lease snapshots. Tests use those -// snapshots only after a RingLeased event, so private allocator state remains -// unobservable. -fn new_arena() -> arena::ArenaManagerHarness { - arena::ArenaManagerHarness::boot(arena_config()).expect("test arena must boot") -} - -// This helper proves two public layouts are disjoint using half-open ranges. -// It is more useful than comparing offsets directly because allocator choice is -// intentionally implementation-defined. -fn assert_non_overlapping(left: &arena::RingLayout, right: &arena::RingLayout) { - let left_range = left.start_offset..left.end_offset; - let right_range = right.start_offset..right.end_offset; - assert!( - left_range.end <= right_range.start || right_range.end <= left_range.start, - "live leases overlap: {left:?} and {right:?}" - ); -} - -// This proves sampling an unused arena reports only reserved capacity and no -// allocator activity. -#[test] -fn sample_reports_empty_arena_capacity_and_zero_activity() { - let harness = new_arena(); - - let sample = harness.sample(41); - - assert_eq!(sample.seq, 41); - assert!( - sample.sample_unix_ms > 0, - "sample timestamp must be a populated Unix epoch millisecond" - ); - assert_eq!(sample.capacity_bytes, arena_config().reservation_ceiling); - assert_eq!(sample.live_bytes, 0); - assert_eq!(sample.free_bytes, arena_config().reservation_ceiling); - assert_eq!(sample.active_leases, 0); - assert_eq!(sample.pending_leases, 0); - assert_eq!( - sample.largest_free_range_bytes, - arena_config().reservation_ceiling - ); - assert_eq!(sample.allocation_failures_total, 0); - assert_eq!(sample.release_failures_total, 0); -} - -// This proves a granted lease is reflected in live capacity accounting and the -// active lease count without relying on private allocator slots. -#[test] -fn sample_counts_live_bytes_and_active_leases_after_grant() { - let mut harness = new_arena(); - harness.request(arena::ArenaRequest::LeaseRing(lease_request(1, 512, 64))); - let lease = harness.live_leases()[0].clone(); - let lease_bytes = lease.layout.end_offset - lease.layout.start_offset; - - let sample = harness.sample(42); - - assert_eq!(sample.seq, 42); - assert_eq!(sample.capacity_bytes, arena_config().reservation_ceiling); - assert_eq!(sample.live_bytes, lease_bytes); - assert_eq!( - sample.free_bytes, - arena_config().reservation_ceiling - lease_bytes - ); - assert_eq!(sample.active_leases, 1); - assert_eq!(sample.pending_leases, 0); - assert_eq!(sample.allocation_failures_total, 0); - assert_eq!(sample.release_failures_total, 0); -} - -// This proves a proof-backed release returns the full range to the free pool and -// restores the largest allocatable span. -#[test] -fn sample_reports_full_free_space_after_release() { - let mut harness = new_arena(); - harness.request(arena::ArenaRequest::LeaseRing(lease_request(1, 512, 64))); - let ring_id = harness.live_leases()[0].ring_id; - - harness.request(arena::ArenaRequest::ReleaseRing { - ring_id, - proof: arena::QuiescenceProof::verified(), - }); - let sample = harness.sample(43); - - assert_eq!(sample.live_bytes, 0); - assert_eq!(sample.free_bytes, arena_config().reservation_ceiling); - assert_eq!(sample.active_leases, 0); - assert_eq!(sample.pending_leases, 0); - assert_eq!( - sample.largest_free_range_bytes, - arena_config().reservation_ceiling - ); -} - -// This proves queued leases are visible as pending work while existing live -// leases continue to own their bytes. -#[test] -fn sample_counts_queued_requests_as_pending_leases() { - let mut harness = new_arena(); - harness.request(arena::ArenaRequest::LeaseRing(lease_request(1, 3072, 64))); - let live = harness.live_leases()[0].clone(); - let live_bytes = live.layout.end_offset - live.layout.start_offset; - harness.request(arena::ArenaRequest::LeaseRing(lease_request(2, 1024, 64))); - - let sample = harness.sample(44); - - assert_eq!(sample.live_bytes, live_bytes); - assert_eq!( - sample.free_bytes, - arena_config().reservation_ceiling - live_bytes - ); - assert_eq!(sample.active_leases, 1); - assert_eq!(sample.pending_leases, 1); - assert_eq!(sample.allocation_failures_total, 0); - assert_eq!(sample.release_failures_total, 0); -} - -// This proves lease rejection increments the allocation failure counter without -// changing the arena's free capacity. -#[test] -fn sample_counts_rejected_leases_as_allocation_failures() { - let mut harness = new_arena(); - - harness.request(arena::ArenaRequest::LeaseRing(lease_request(1, 8192, 64))); - let sample = harness.sample(45); - - assert_eq!(sample.live_bytes, 0); - assert_eq!(sample.free_bytes, arena_config().reservation_ceiling); - assert_eq!(sample.active_leases, 0); - assert_eq!(sample.pending_leases, 0); - assert_eq!( - sample.largest_free_range_bytes, - arena_config().reservation_ceiling - ); - assert_eq!(sample.allocation_failures_total, 1); - assert_eq!(sample.release_failures_total, 0); -} - -// This proves release rejection increments the release failure counter and keeps -// the live lease accounted as active. -#[test] -fn sample_counts_rejected_releases_as_release_failures() { - let mut harness = new_arena(); - harness.request(arena::ArenaRequest::LeaseRing(lease_request(1, 512, 64))); - let lease = harness.live_leases()[0].clone(); - let lease_bytes = lease.layout.end_offset - lease.layout.start_offset; - - harness.request(arena::ArenaRequest::ReleaseRing { - ring_id: lease.ring_id, - proof: arena::QuiescenceProof::missing(), - }); - let sample = harness.sample(46); - - assert_eq!(sample.live_bytes, lease_bytes); - assert_eq!( - sample.free_bytes, - arena_config().reservation_ceiling - lease_bytes - ); - assert_eq!(sample.active_leases, 1); - assert_eq!(sample.pending_leases, 0); - assert_eq!(sample.allocation_failures_total, 0); - assert_eq!(sample.release_failures_total, 1); -} - -// This proves arena boot creates one stable sparse arena with one reservation -// ceiling, offset-only layouts, and typed boot failure. -#[test] -fn arena_boot_creates_stable_offset_only_layout_domain() { - // Boot a valid arena. - let mut harness = new_arena(); - - // Lease one ring so the public layout can be inspected. - harness.request(arena::ArenaRequest::LeaseRing(lease_request(1, 512, 64))); - let lease = harness - .events() - .iter() - .find_map(|event| match event { - arena::ArenaEvent::RingLeased { lease } => Some(lease), - _ => None, - }) - .expect("valid lease must be granted"); - - // Layout facts are arena offsets and stay under the reservation ceiling. - assert!(lease.layout.start_offset < arena_config().reservation_ceiling); - assert!(lease.layout.end_offset <= arena_config().reservation_ceiling); - assert!(matches!( - lease.layout.pointer, - arena::LayoutPointer::NoProcessPointer - )); - - // Boot failure is typed and emits no usable arena. - let failed = arena::ArenaManagerHarness::boot(arena::ArenaConfig { - reservation_ceiling: 0, - ..arena_config() - }); - assert!(matches!( - failed, - Err(arena::ArenaFault::InvalidReservationCeiling) - )); -} - -// This proves LeaseRing either leases, queues, or rejects, and that live RingId -// values are unique for the node lifetime. -#[test] -fn lease_requests_grant_queue_or_reject_with_unique_ring_ids() { - // Fill most of the arena with one live lease. - let mut harness = new_arena(); - harness.request(arena::ArenaRequest::LeaseRing(lease_request(1, 3072, 64))); - harness.request(arena::ArenaRequest::LeaseRing(lease_request(2, 1024, 64))); - - // A satisfiable request under pressure may queue instead of rejecting. - assert!(harness.events().iter().any(|event| { - matches!( - event, - arena::ArenaEvent::RingLeaseQueued { - request_id: arena::LeaseRequestId(2) - } - ) - })); - - // A request larger than the reservation ceiling must reject. - harness.request(arena::ArenaRequest::LeaseRing(lease_request(3, 8192, 64))); - assert!(harness.events().iter().any(|event| { - matches!( - event, - arena::ArenaEvent::RingLeaseRejected { - request_id: arena::LeaseRequestId(3), - reason: arena::RingLeaseRejection::CannotFitWithinCeiling, - } - ) - })); - - // Granted RingId values must be unique among all live leases. - let ids = harness - .live_leases() - .iter() - .map(|lease| lease.ring_id) - .collect::>(); - assert_eq!(ids.len(), harness.live_leases().len()); -} - -// This proves live leases are non-overlapping, in-bounds, aligned, and stable -// for the lease lifetime. -#[test] -fn live_layouts_are_non_overlapping_aligned_in_bounds_and_stable() { - // Lease two rings with explicit alignment requirements. - let mut harness = new_arena(); - harness.request(arena::ArenaRequest::LeaseRing(lease_request(1, 512, 64))); - harness.request(arena::ArenaRequest::LeaseRing(lease_request(2, 512, 128))); - - // Read the public lease snapshots. - let leases = harness.live_leases().to_vec(); - assert_eq!(leases.len(), 2); - - // Prove non-overlap and in-bounds without constraining allocator placement. - assert_non_overlapping(&leases[0].layout, &leases[1].layout); - for lease in &leases { - assert!(lease.layout.end_offset <= arena_config().reservation_ceiling); - assert_eq!(lease.layout.start_offset % lease.requested_alignment, 0); - } - - // Re-observing the same live lease must not change offsets. - let before = leases[0].layout.clone(); - let after = harness - .lookup_lease(leases[0].ring_id) - .expect("live lease must be lookupable") - .layout - .clone(); - assert_eq!(after, before); -} - -// This proves CancelLease removes queued work and suppresses later hot-path -// installation, including a fresh lease that races with cancellation. -#[test] -fn cancelled_queued_lease_never_installs_hot_path_state() { - // Fill the arena and queue a second satisfiable lease. - let mut harness = new_arena(); - harness.request(arena::ArenaRequest::LeaseRing(lease_request(1, 3072, 64))); - harness.request(arena::ArenaRequest::LeaseRing(lease_request(2, 1024, 64))); - - // Cancel the queued request before it is leased. - harness.request(arena::ArenaRequest::CancelLease { - request_id: arena::LeaseRequestId(2), - }); - - // Releasing pressure must not install worker or pump state for the canceled - // request. - let live = harness.live_leases()[0].ring_id; - harness.request(arena::ArenaRequest::ReleaseRing { - ring_id: live, - proof: arena::QuiescenceProof::verified(), - }); - assert!(!harness.commands().iter().any(|command| { - matches!( - command, - arena::ArenaCommand::InstallWorkerOrPumpState { - request_id: arena::LeaseRequestId(2), - .. - } - ) - })); - - // If a fresh lease was produced during the race, it must be released instead - // of becoming hot-path state. - assert!(harness.events().iter().any(|event| { - matches!( - event, - arena::ArenaEvent::CancelledFreshLeaseReleased { - request_id: arena::LeaseRequestId(2) - } - ) - })); -} - -// This proves ranges are released only with quiescence proof, are not reused -// while live work owns them, and may be reused after release. -#[test] -fn release_requires_quiescence_and_reuse_happens_only_after_release() { - // Lease one ring and record its range. - let mut harness = new_arena(); - harness.request(arena::ArenaRequest::LeaseRing(lease_request(1, 1024, 64))); - let first = harness.live_leases()[0].clone(); - - // Release without proof must reject and keep the range live. - harness.request(arena::ArenaRequest::ReleaseRing { - ring_id: first.ring_id, - proof: arena::QuiescenceProof::missing(), - }); - assert!(harness.lookup_lease(first.ring_id).is_some()); - - // A second lease while the first is live must not overlap the first range. - harness.request(arena::ArenaRequest::LeaseRing(lease_request(2, 1024, 64))); - let second = harness - .live_leases() - .iter() - .find(|lease| lease.ring_id != first.ring_id) - .expect("second live lease must exist") - .clone(); - assert_non_overlapping(&first.layout, &second.layout); - - // Verified quiescence allows release, after which reuse is legal. - harness.request(arena::ArenaRequest::ReleaseRing { - ring_id: first.ring_id, - proof: arena::QuiescenceProof::verified(), - }); - harness.request(arena::ArenaRequest::LeaseRing(lease_request(3, 1024, 64))); - assert!( - harness - .events() - .iter() - .any(|event| { matches!(event, arena::ArenaEvent::RingLeased { .. }) }) - ); -} - -// This proves shutdown rejects new leases, preserves live lease records, and -// does not release live ranges without quiescence proof. -#[test] -fn shutdown_rejects_new_leases_without_corrupting_live_records() { - // Create a live lease before shutdown. - let mut harness = new_arena(); - harness.request(arena::ArenaRequest::LeaseRing(lease_request(1, 512, 64))); - let live_before = harness.live_leases().to_vec(); - - // Shut the arena down. - harness.request(arena::ArenaRequest::Shutdown); - - // New leases are rejected after shutdown. - harness.request(arena::ArenaRequest::LeaseRing(lease_request(2, 512, 64))); - assert!(harness.events().iter().any(|event| { - matches!( - event, - arena::ArenaEvent::RingLeaseRejected { - request_id: arena::LeaseRequestId(2), - reason: arena::RingLeaseRejection::ArenaShuttingDown, - } - ) - })); - - // Existing live lease records remain intact until proof-backed release. - assert_eq!(harness.live_leases(), live_before.as_slice()); -} diff --git a/crates/mvp-system/src/tests/data_plane_bridge_guarantees.rs b/crates/mvp-system/src/tests/data_plane_bridge_guarantees.rs new file mode 100644 index 0000000..26d7763 --- /dev/null +++ b/crates/mvp-system/src/tests/data_plane_bridge_guarantees.rs @@ -0,0 +1,96 @@ +use data_plane::actor as dp; +use mvp_system::node::actor::NodeAgentMsg; +use mvp_system::node::data_plane_bridge::MvpDataPlaneReportSinkActor; +use swactor::config::RuntimeConfig; +use swactor::runtime::Runtime; + +fn spawn_bridge( + runtime: &Runtime, +) -> ( + swactor::actor::ActorAddress, + swactor::runtime::Inbox, +) { + let node_messages = runtime + .new_inbox::() + .expect("node-agent inbox"); + let bridge = runtime + .spawn(MvpDataPlaneReportSinkActor::new(*node_messages.addr())) + .expect("spawn bridge"); + (bridge, node_messages) +} + +#[test] +fn data_plane_reports_drive_mvp_stage_readiness_and_object_visibility() { + let runtime = Runtime::new(RuntimeConfig::default()); + let (bridge, node_messages) = spawn_bridge(&runtime); + + runtime + .send_to( + bridge, + dp::DataPlaneReportMsg::InboundEdgeReady { + edge_id: dp::EdgeId(7001), + }, + ) + .expect("send inbound ready"); + runtime + .send_to( + bridge, + dp::DataPlaneReportMsg::ObjectLoaded { + edge_id: dp::EdgeId(7001), + object_id: data_plane::object_record::ObjectId(9000), + sequence: 3, + extent: 8, + handle: dp::DeviceHandle::new(dp::WorkerGeneration(2), 44), + }, + ) + .expect("send object loaded"); + runtime.tick(); + + assert_eq!( + node_messages.try_recv(), + Some(NodeAgentMsg::MarkInboundEdgeReady { edge_id: 7001 }) + ); + assert_eq!( + node_messages.try_recv(), + Some(NodeAgentMsg::ObjectLoaded { + edge_id: 7001, + object_id: 9000, + sequence: 3, + handle_generation: 2, + handle_id: 44, + }) + ); +} + +#[test] +fn data_plane_fault_and_stop_reports_map_to_mvp_lifecycle_messages() { + let runtime = Runtime::new(RuntimeConfig::default()); + let (bridge, node_messages) = spawn_bridge(&runtime); + + runtime + .send_to( + bridge, + dp::DataPlaneReportMsg::EdgeFaulted { + edge_id: dp::EdgeId(7001), + reason: dp::EdgeFaultReason::StreamFault( + data_plane::edge_lifecycle::StreamFaultReason::ProtocolError, + ), + }, + ) + .expect("send edge fault"); + runtime + .send_to( + bridge, + dp::DataPlaneReportMsg::LocalEdgesStopped { + run_id: dp::RunId(55), + }, + ) + .expect("send stopped"); + runtime.tick(); + + assert_eq!(node_messages.try_recv(), Some(NodeAgentMsg::WorkerCrashed)); + assert_eq!( + node_messages.try_recv(), + Some(NodeAgentMsg::LocalEdgesStopped { run_id: 55 }) + ); +} diff --git a/crates/mvp-system/src/tests/gpu_worker_ingress_parser_guarantees.rs b/crates/mvp-system/src/tests/gpu_worker_ingress_parser_guarantees.rs index fe60f4f..c3b74f1 100644 --- a/crates/mvp-system/src/tests/gpu_worker_ingress_parser_guarantees.rs +++ b/crates/mvp-system/src/tests/gpu_worker_ingress_parser_guarantees.rs @@ -9,7 +9,7 @@ //! They assert the guarantees in //! `specs/mvp_system/gpu_worker_ingress_parser_contract.md`. -use mvp_system::node_data::ingress; +use data_plane::ingress; // A valid ingress ring config supplies the edge object spec and current worker // generation. The parser remains a black box behind ring helper operations. diff --git a/crates/mvp-system/src/tests/local_mock/environment.rs b/crates/mvp-system/src/tests/local_mock/environment.rs index add6759..2474474 100644 --- a/crates/mvp-system/src/tests/local_mock/environment.rs +++ b/crates/mvp-system/src/tests/local_mock/environment.rs @@ -1,6 +1,6 @@ use std::collections::{BTreeMap, BTreeSet}; -use mvp_system::node_data::edge_actor; +use data_plane::edge_actor; use mvp_system::observability::lifecycle as obs; use mvp_system::orchestration::engine_builder as engine; use mvp_system::orchestration::run_fsm as fsm; diff --git a/crates/mvp-system/src/tests/local_mock/mock_node.rs b/crates/mvp-system/src/tests/local_mock/mock_node.rs index 67c9903..6fb70f0 100644 --- a/crates/mvp-system/src/tests/local_mock/mock_node.rs +++ b/crates/mvp-system/src/tests/local_mock/mock_node.rs @@ -1,4 +1,4 @@ -use mvp_system::node_data::edge_actor; +use data_plane::edge_actor; use mvp_system::orchestration::run_plan as plan; use mvp_system::staging as stage; diff --git a/crates/mvp-system/src/tests/mod.rs b/crates/mvp-system/src/tests/mod.rs index 1e3d77f..e149422 100644 --- a/crates/mvp-system/src/tests/mod.rs +++ b/crates/mvp-system/src/tests/mod.rs @@ -1,11 +1,9 @@ -mod arena_manager_guarantees; mod bootstrap_datastream_guarantees; +mod data_plane_bridge_guarantees; mod device_bridge_guarantees; mod docker_cluster_provisioning_guarantees; -mod edge_establisher_guarantees; mod engine_builder_guarantees; mod gpu_worker_ctl_guarantees; -mod gpu_worker_egress_producer_guarantees; mod gpu_worker_ingress_parser_guarantees; mod gpu_worker_process_adapter_guarantees; mod local_mock; diff --git a/crates/mvp-system/src/tests/module_surface_guarantees.rs b/crates/mvp-system/src/tests/module_surface_guarantees.rs index 4bee400..ffd7844 100644 --- a/crates/mvp-system/src/tests/module_surface_guarantees.rs +++ b/crates/mvp-system/src/tests/module_surface_guarantees.rs @@ -3,9 +3,8 @@ //! These tests assert only the observable public-path contract: target modules //! expose existing behavior without introducing copied type definitions. -use mvp_system::{ - chat, node, node_data, observability, orchestration, prompt, staging, transport, worker, -}; +use data_plane::{arena as data_arena, edge_lifecycle as data_edge}; +use mvp_system::{chat, node, observability, orchestration, prompt, staging, transport, worker}; #[test] fn target_modules_offer_new_paths_to_existing_public_contracts() { @@ -15,14 +14,14 @@ fn target_modules_offer_new_paths_to_existing_public_contracts() { let boot_node_id: node::boot_lifecycle::NodeId = node::boot_lifecycle::NodeId(11); assert_eq!(boot_node_id, node::boot_lifecycle::NodeId(11)); - let arena_ring_id: node_data::arena::RingId = node_data::arena::RingId(3); - assert_eq!(arena_ring_id, node_data::arena::RingId(3)); + let arena_ring_id: data_arena::RingId = data_arena::RingId(3); + assert_eq!(arena_ring_id, data_arena::RingId(3)); let stage_edge_id: staging::EdgeId = staging::EdgeId(7001); assert_eq!(stage_edge_id, staging::EdgeId(7001)); - let transport_edge_id: node::edge_lifecycle::EdgeId = node::edge_lifecycle::EdgeId(7002); - assert_eq!(transport_edge_id, node::edge_lifecycle::EdgeId(7002)); + let transport_edge_id: data_edge::EdgeId = data_edge::EdgeId(7002); + assert_eq!(transport_edge_id, data_edge::EdgeId(7002)); let worker_generation: worker::WorkerGeneration = worker::WorkerGeneration(2); assert_eq!(worker_generation, worker::WorkerGeneration(2)); diff --git a/crates/mvp-system/src/tests/shared_ring_helper_abi_guarantees.rs b/crates/mvp-system/src/tests/shared_ring_helper_abi_guarantees.rs index a81ca11..ea5c292 100644 --- a/crates/mvp-system/src/tests/shared_ring_helper_abi_guarantees.rs +++ b/crates/mvp-system/src/tests/shared_ring_helper_abi_guarantees.rs @@ -9,7 +9,7 @@ //! They assert the guarantees in //! `specs/mvp_system/shared_ring_helper_abi_contract.md`. -use mvp_system::node_data::ring; +use data_plane::ring; // A small ring forces wraparound and full/empty transitions quickly. The helper // still owns the actual shared-memory atomics and process-local address math. diff --git a/crates/mvp-system/src/tests/tx_rx_edge_actor_guarantees.rs b/crates/mvp-system/src/tests/tx_rx_edge_actor_guarantees.rs index 1e4dbba..0c70b7e 100644 --- a/crates/mvp-system/src/tests/tx_rx_edge_actor_guarantees.rs +++ b/crates/mvp-system/src/tests/tx_rx_edge_actor_guarantees.rs @@ -8,7 +8,7 @@ //! They assert the guarantees in //! `specs/mvp_system/tx_rx_edge_actor_contract.md`. -use mvp_system::node_data::edge_actor; +use data_plane::edge_actor; // The edge id fixture gives both actors a shared identity while keeping Tx and // Rx lifecycle tests independent from driver and ring internals. diff --git a/crates/mvp-system/src/tests/worker_edge_adapter_guarantees.rs b/crates/mvp-system/src/tests/worker_edge_adapter_guarantees.rs index 0392fbf..d7bbabf 100644 --- a/crates/mvp-system/src/tests/worker_edge_adapter_guarantees.rs +++ b/crates/mvp-system/src/tests/worker_edge_adapter_guarantees.rs @@ -1,9 +1,9 @@ +use data_plane::object_record as ingress; use mvp_system::node::actor::{ NodeAgentActor, NodeAgentMsg, NodeAgentReport, StageCommandWire, StageEdgeKindWire, StageInboundEdgeWire, StageObjectSpecWire, StageOutboundEdgeWire, StageProvisionWire, StageRingSpecWire, }; -use mvp_system::node_data::object as ingress; use mvp_system::orchestration::run_plan::{self, GgufSource, TokenizerSource}; use mvp_system::staging as stage; use swactor::actor::ActorAddress; diff --git a/crates/mvp-system/src/worker/mod.rs b/crates/mvp-system/src/worker/mod.rs index 70731ef..15545f1 100644 --- a/crates/mvp-system/src/worker/mod.rs +++ b/crates/mvp-system/src/worker/mod.rs @@ -2,7 +2,6 @@ pub mod control; pub mod device_bridge; -pub mod egress; pub mod process_adapter; pub mod process {