diff --git a/Cargo.lock b/Cargo.lock index 506601b..631ec03 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2522,6 +2522,7 @@ dependencies = [ "phoxal", "phoxal-api", "serde_json", + "tokio", ] [[package]] diff --git a/docs/CONVENTIONS.md b/docs/CONVENTIONS.md index 0b4f788..d847f6f 100644 --- a/docs/CONVENTIONS.md +++ b/docs/CONVENTIONS.md @@ -72,11 +72,24 @@ the only source of keys, and the wire body never appears in the key ## Logical time -- Participants track time, watchdogs, and staleness with **logical time** - only - never wall time directly. +- Robot services, drivers, and simulators track state-transition time, + watchdogs, and synchronous-input staleness with **logical time** - never wall + time directly. The runner owns one `ClockSource` and stamps every `StepContext` and every - `produced_at_ns` from it, so all participants share one time domain + `produced_at_ns` from it, so participants in the robot clock share one domain ([`phoxal/src/participant/clock.rs`](../phoxal/src/participant/clock.rs)). +- Tools are outside that clock. Their process launch contract has no `--clock` + flag or `PHOXAL_CLOCK` binding, and the normal embedding API accepts no clock + argument. They run from external events and host-monotonic timers in every + mode. Tool envelope metadata uses `phoxal::raw::host_time()`; the runner never + gives tools `StepContext` or enrolls them in logical scheduling. Official tool + sources are checked against simulation-clock imports; privileged user-authored + raw-bus tools must uphold the same rule and never decide freshness from robot + logical time. +- A logical-time consumer of asynchronous external input owns retention and + freshness. Keep only the latest bounded value, record its consumer-local + monotonic arrival instant, and sample that value at the logical step. A + logical pause must not accumulate a replay backlog. - `LogicalTime` is `{ epoch, time_ns }`. Within an epoch `time_ns` strictly increases; an epoch bump signals a reset. `RealClock` reads the host-wide UNIX-epoch domain (so cross-process staleness @@ -104,7 +117,8 @@ the only source of keys, and the wire body never appears in the key `#[phoxal::driver]` for per-component-instance participants that can call `ctx.component()`; `#[phoxal::tool]` for host-side utilities that inspect `ctx.robot()` (`Api` - defaults to `()` - tools stay raw-bus only); + defaults to `()` - tools stay raw-bus only and host/event driven, with no + logical clock accessor or scheduled step); `#[phoxal::simulator]` for simulation-only participants. Each kind embeds its own JSON metadata (id, kind, contract surface) as a static in a dedicated linker section on the compiled binary diff --git a/docs/GETTING_STARTED.md b/docs/GETTING_STARTED.md index 9b75cd5..29b2e04 100644 --- a/docs/GETTING_STARTED.md +++ b/docs/GETTING_STARTED.md @@ -23,6 +23,7 @@ examples/hello-rover/ robot.yaml # the manifest robot.dev.yaml # dev-only overlay: local path pins structure.urdf # the robot's own URDF (chassis + mounts) + worlds/default.wbt # minimal Webots world components/wheel_drive/ # one robot-local component definition component.yaml structure.urdf @@ -33,7 +34,7 @@ Both wheels are instances of the same `wheel_drive` component type. Motion is commanded through the official manual or navigation candidate contracts; the example intentionally carries no always-moving cruise service. It is deliberately small: no sensors beyond what the kinematic model needs, -no simulation world, and no hardware driver. +one minimal simulation world, and no hardware driver. Use it as the shape to copy from, not as a real robot. ## `robot.yaml` anatomy @@ -288,10 +289,15 @@ phoxal-cli validate --report --allow-user-service-drift # lower-level structur `check` resolves every official participant (services, tools, and any catalog-sourced component drivers) plus every user service and prints `ok: N participants validated` once the whole graph is coherent. -`hello-rover` has no Webots world checked in (its component uses primitive -URDF geometry, not meshes, so there is nothing to render yet); a project that -wants `phoxal-cli simulate ` needs a `worlds/.wbt` file, the -same way [`robot-v1`](https://github.com/phoxal/robot-v1) does. +`hello-rover` includes a minimal Webots world: + +```sh +phoxal-cli simulation run default --env dev +``` + +The CLI stages the authored world and injects the generated rover and Webots +controllers. Keep additional example worlds equally small; do not use a private +robot as the framework harness. ## Where to go from here diff --git a/examples/hello-rover/worlds/default.wbt b/examples/hello-rover/worlds/default.wbt new file mode 100644 index 0000000..fb68643 --- /dev/null +++ b/examples/hello-rover/worlds/default.wbt @@ -0,0 +1,21 @@ +#VRML_SIM R2025a utf8 + +EXTERNPROTO "https://raw.githubusercontent.com/cyberbotics/webots/R2025a/projects/objects/floors/protos/RectangleArena.proto" +EXTERNPROTO "https://raw.githubusercontent.com/cyberbotics/webots/R2025a/projects/objects/backgrounds/protos/TexturedBackground.proto" +EXTERNPROTO "https://raw.githubusercontent.com/cyberbotics/webots/R2025a/projects/objects/backgrounds/protos/TexturedBackgroundLight.proto" + +WorldInfo { + basicTimeStep 16 +} +Viewpoint { + orientation -0.36 -0.22 0.91 5.05 + position -2.5 3.5 2.2 +} +TexturedBackground { +} +TexturedBackgroundLight { +} +RectangleArena { + floorSize 4 4 + wallHeight 1 +} diff --git a/phoxal-macros/src/authoring.rs b/phoxal-macros/src/authoring.rs index 8afde06..053180d 100644 --- a/phoxal-macros/src/authoring.rs +++ b/phoxal-macros/src/authoring.rs @@ -702,6 +702,17 @@ impl ParticipantKind { } } + fn launch_policy(self, phoxal: &TokenStream) -> TokenStream { + match self { + ParticipantKind::Tool => { + quote!(#phoxal::participant::launch::ToolParticipantLaunch) + } + ParticipantKind::Service | ParticipantKind::Driver | ParticipantKind::Simulator => { + quote!(#phoxal::participant::launch::ClockedParticipantLaunch) + } + } + } + /// Default `Api` type when `api = …` is not given: tools stay raw-bus /// only (decided 2026-07-09 - no typed tool `Api` until a real need /// appears), every other kind defaults to the local `Api` struct. @@ -810,6 +821,7 @@ pub fn expand_participant( let phoxal = phoxal(); let artifact_kind = kind.artifact_kind(); let participant_class = kind.participant_class(); + let launch_policy = kind.launch_policy(&phoxal); let marker = kind.marker_impl(&phoxal, struct_name); let metadata_const_ident = Ident::new( &format!( @@ -841,6 +853,7 @@ pub fn expand_participant( const KIND: &'static str = #artifact_kind; const PARTICIPANT_CLASS: &'static str = #participant_class; const ID: &'static str = #id; + type LaunchPolicy = #launch_policy; type Config = #config_ty; type Api = #api_ty; } diff --git a/phoxal/README.md b/phoxal/README.md index 2cf3c54..dc890f1 100644 --- a/phoxal/README.md +++ b/phoxal/README.md @@ -80,6 +80,18 @@ Key rules the example shows: The runner also owns the rest of the lifecycle: `#[step(hz = ...)]` is the scheduled control loop, `#[server]` / `#[server_snapshot]` serve queries, and `#[shutdown]` runs graceful park/stop/flush before the bus closes. +`#[phoxal::tool]` is intentionally outside the robot clock. Tools use managed +host/event loops, host-monotonic timers for cadence and freshness, and +`phoxal::raw::host_time()` only when a generic bus envelope timestamp is +required. The macro rejects `#[step]`, the setup context exposes no clock, and +the tool process launch has no clock option. The normal embedding API likewise +accepts no clock argument; deterministic clock injection is restricted to typed +graph participants. Official tool sources are additionally checked against +importing or subscribing to simulation-clock surfaces; user-authored tools must +preserve the same boundary when using the privileged raw bus. A service that +consumes asynchronous tool input keeps a bounded latest value with +consumer-local monotonic arrival time and samples it at its own logical step. + ## Modules A participant depends on two crates: the `phoxal` engine (the modules below) and the `phoxal-api` contract tree. diff --git a/phoxal/src/lib.rs b/phoxal/src/lib.rs index 63037d1..8e97991 100644 --- a/phoxal/src/lib.rs +++ b/phoxal/src/lib.rs @@ -193,8 +193,21 @@ pub mod bus { /// graph checker still includes their contracts, but never lets their raw /// access satisfy checked topology. pub mod raw { - pub use crate::participant::runner::run_with_bus; + pub use crate::participant::runner::{run_with_bus, run_with_bus_clock}; pub use phoxal_bus::*; + + /// Current host wall time for privileged tools and bridges that need an + /// envelope timestamp without joining a robot's logical clock. + /// + /// This is metadata time only. Scheduling periodic work should use a host + /// monotonic timer such as [`std::time::Instant`] or + /// [`tokio::time::interval`]. + pub fn host_time() -> LogicalTime { + let elapsed = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default(); + LogicalTime::new(0, u64::try_from(elapsed.as_nanos()).unwrap_or(u64::MAX)) + } } /// The framework result type (`anyhow`-backed). Authoring code uses bare diff --git a/phoxal/src/participant/api.rs b/phoxal/src/participant/api.rs index 7d8d481..da1f48c 100644 --- a/phoxal/src/participant/api.rs +++ b/phoxal/src/participant/api.rs @@ -336,6 +336,10 @@ pub trait Participant: Sized + Send + 'static { const PARTICIPANT_CLASS: &'static str; /// The participant id (`id = "…"`, default kebab of the type name). const ID: &'static str; + /// The process launch contract. Tools use a clockless policy; checked graph + /// participants use the configurable robot-clock policy. + #[doc(hidden)] + type LaunchPolicy: crate::participant::launch::ParticipantLaunchPolicy; /// The participant's typed config (`robot.yaml` input). type Config: ParticipantConfig; /// The participant's bus-facing contract surface (`()` for a raw-bus @@ -660,23 +664,18 @@ impl SetupContextSimulatorExt for SetupContext } /// Tool-only `SetupContext` accessor (`R: Participant + IsTool`). Tools stay -/// raw-bus only (decided 2026-07-09), so this is their sole IO seam. +/// raw-bus only (decided 2026-07-09), so this is their sole IO seam. Tools do +/// not receive the runner clock: host timers and [`crate::raw::host_time`] keep +/// them outside robot logical/simulation time. pub trait SetupContextToolExt { /// Clone the runner-owned raw bus for privileged tool internals. The bus is /// already open from the launch contract, so a tool does not reparse launch /// env or open an unrelated session. fn raw_bus(&self) -> Bus; - - /// Clone the logical clock owned by the participant runner. - fn clock(&self) -> std::sync::Arc; } impl SetupContextToolExt for SetupContext { fn raw_bus(&self) -> Bus { self.bus().clone() } - - fn clock(&self) -> std::sync::Arc { - self.clock_source() - } } diff --git a/phoxal/src/participant/context.rs b/phoxal/src/participant/context.rs index 8fdb92b..0e74d48 100644 --- a/phoxal/src/participant/context.rs +++ b/phoxal/src/participant/context.rs @@ -8,7 +8,6 @@ use std::time::Duration; use crate::bus::{LogicalTime, OwnerCap}; use crate::model::v0::Robot; -use crate::participant::clock::ClockSource; use crate::participant::managed::{ManagedTaskPolicy, ManagedTasks}; use phoxal_bus::Bus; @@ -18,7 +17,6 @@ use phoxal_bus::Bus; /// enforces. pub struct SetupContext { bus: Bus, - clock: Arc, owner_cap: OwnerCap, robot: Option>, robot_root: Option, @@ -34,7 +32,6 @@ pub struct SetupContext { impl SetupContext { pub(crate) fn new( bus: Bus, - clock: Arc, owner_cap: OwnerCap, robot: Option>, robot_root: Option, @@ -42,7 +39,6 @@ impl SetupContext { ) -> Self { SetupContext { bus, - clock, owner_cap, robot, robot_root, @@ -114,12 +110,6 @@ impl SetupContext { &self.bus } - /// Clone the runner-owned logical clock so privileged raw-bus tools can - /// stamp messages in the active real or simulation time domain. - pub(crate) fn clock_source(&self) -> Arc { - Arc::clone(&self.clock) - } - /// The runner-minted owner capability (plan #00 L2). In-crate accessor the /// NEW model's `SetupContextApiExt::owner_capability` (`participant::api`) /// reads the raw field through. diff --git a/phoxal/src/participant/launch.rs b/phoxal/src/participant/launch.rs index 2d26ca5..4594115 100644 --- a/phoxal/src/participant/launch.rs +++ b/phoxal/src/participant/launch.rs @@ -1,9 +1,11 @@ //! `ParticipantLaunch` - the clap/env process launch contract. //! -//! Every participant binary accepts the same documented `--flag` set with -//! matching `PHOXAL_*` env fallbacks. Supervisors and systemd units use env, -//! while humans can use flags for bench runs. Flags win over env through clap's -//! native precedence, and `--help` is the user-facing contract documentation. +//! Participant binaries share one common `--flag` set with matching `PHOXAL_*` +//! env fallbacks. Clocked graph participants additionally accept `--clock` / +//! `PHOXAL_CLOCK`; tools do not expose either input. Supervisors and systemd +//! units use env, while humans can use flags for bench runs. Flags win over env +//! through clap's native precedence, and `--help` is the user-facing contract +//! documentation. use std::path::PathBuf; @@ -30,7 +32,7 @@ pub mod env { pub const CONNECT: &str = "PHOXAL_CONNECT"; /// Inline JSON participant config block. pub const CONFIG: &str = "PHOXAL_CONFIG"; - /// Clock mode: real or simulation. + /// Clock mode for services, drivers, and simulators: real or simulation. pub const CLOCK: &str = "PHOXAL_CLOCK"; /// All env names in contract order. @@ -58,7 +60,7 @@ pub struct ParticipantLaunch { /// The resolved bus profile. #[serde(default)] pub bus: BusProfile, - /// The clock mode. + /// The robot clock mode. Tool launch policies never read this field. #[serde(default)] pub clock: ClockMode, /// The participant's typed config block (`services..config`), if any. @@ -101,16 +103,6 @@ impl ParticipantLaunch { } } - pub(crate) fn from_cli( - default_participant_id: &'static str, - default_robot_id: &'static str, - ) -> crate::Result { - let matches = - LaunchCli::command_for(default_participant_id, default_robot_id).get_matches(); - let cli = LaunchCli::from_arg_matches(&matches)?; - cli.into_launch(default_participant_id, default_robot_id) - } - /// Set the robot model root for this launch. pub fn with_robot_root(mut self, root: impl Into) -> Self { self.robot_root = Some(root.into()); @@ -124,14 +116,9 @@ impl ParticipantLaunch { } } -/// The clap-derived launch contract for participant binaries. -#[derive(Debug, clap::Parser)] -#[command( - name = "phoxal-participant", - about = "Run a Phoxal participant.", - long_about = None -)] -struct LaunchCli { +/// The clap-derived launch fields shared by every participant binary. +#[derive(Debug, clap::Args)] +struct CommonLaunchCli { /// Bus-unique participant id. Defaults to the compiled participant artifact id. #[arg( long, @@ -190,8 +177,20 @@ struct LaunchCli { value_name = "JSON" )] config: Option, +} + +/// Launch contract for services, drivers, and simulators. +#[derive(Debug, clap::Parser)] +#[command( + name = "phoxal-participant", + about = "Run a Phoxal participant.", + long_about = None +)] +struct ClockedLaunchCli { + #[command(flatten)] + common: CommonLaunchCli, - /// Clock mode for the runner. + /// Clock mode for robot-state execution. #[arg( long, env = env::CLOCK, @@ -202,18 +201,20 @@ struct LaunchCli { clock: ClockMode, } -impl LaunchCli { - fn command_for( - default_participant_id: &'static str, - default_robot_id: &'static str, - ) -> clap::Command { - Self::command() - .mut_arg("participant_id", |arg| { - arg.default_value(default_participant_id) - }) - .mut_arg("robot_id", |arg| arg.default_value(default_robot_id)) - } +/// Launch contract for host/event-driven tools. It intentionally has no clock +/// flag or environment binding. +#[derive(Debug, clap::Parser)] +#[command( + name = "phoxal-tool", + about = "Run a Phoxal tool.", + long_about = None +)] +struct ToolLaunchCli { + #[command(flatten)] + common: CommonLaunchCli, +} +impl CommonLaunchCli { fn into_launch( self, default_participant_id: &'static str, @@ -241,11 +242,77 @@ impl LaunchCli { .context("PHOXAL_CONFIG must be valid JSON for the participant config")?, ); } - launch.clock = self.clock; Ok(launch) } } +fn command_for( + default_participant_id: &'static str, + default_robot_id: &'static str, +) -> clap::Command { + C::command() + .mut_arg("participant_id", |arg| { + arg.default_value(default_participant_id) + }) + .mut_arg("robot_id", |arg| arg.default_value(default_robot_id)) +} + +/// Type-level launch contract emitted by the participant macros. +#[doc(hidden)] +pub trait ParticipantLaunchPolicy: Send + Sync + 'static { + fn from_cli( + default_participant_id: &'static str, + default_robot_id: &'static str, + ) -> crate::Result; + + fn clock_mode(launch: &ParticipantLaunch) -> ClockMode; +} + +/// Launch policy for services, drivers, and simulators. +#[doc(hidden)] +pub struct ClockedParticipantLaunch; + +impl ParticipantLaunchPolicy for ClockedParticipantLaunch { + fn from_cli( + default_participant_id: &'static str, + default_robot_id: &'static str, + ) -> crate::Result { + let matches = + command_for::(default_participant_id, default_robot_id).get_matches(); + let cli = ClockedLaunchCli::from_arg_matches(&matches)?; + let mut launch = cli + .common + .into_launch(default_participant_id, default_robot_id)?; + launch.clock = cli.clock; + Ok(launch) + } + + fn clock_mode(launch: &ParticipantLaunch) -> ClockMode { + launch.clock + } +} + +/// Clockless launch policy for tools. +#[doc(hidden)] +pub struct ToolParticipantLaunch; + +impl ParticipantLaunchPolicy for ToolParticipantLaunch { + fn from_cli( + default_participant_id: &'static str, + default_robot_id: &'static str, + ) -> crate::Result { + let matches = + command_for::(default_participant_id, default_robot_id).get_matches(); + let cli = ToolLaunchCli::from_arg_matches(&matches)?; + cli.common + .into_launch(default_participant_id, default_robot_id) + } + + fn clock_mode(_launch: &ParticipantLaunch) -> ClockMode { + ClockMode::Real + } +} + fn nonempty_or(value: Option, default: impl FnOnce() -> String) -> String { value .filter(|value| !value.is_empty()) @@ -299,19 +366,29 @@ mod tests { } } - fn parse_from(args: &[&str]) -> crate::Result { - let matches = LaunchCli::command_for("default-id", "robot") + fn parse_clocked_from(args: &[&str]) -> crate::Result { + let matches = command_for::("default-id", "robot") + .try_get_matches_from(args) + .map_err(anyhow::Error::from)?; + let cli = ClockedLaunchCli::from_arg_matches(&matches).map_err(anyhow::Error::from)?; + let mut launch = cli.common.into_launch("default-id", "robot")?; + launch.clock = cli.clock; + Ok(launch) + } + + fn parse_tool_from(args: &[&str]) -> crate::Result { + let matches = command_for::("default-id", "robot") .try_get_matches_from(args) .map_err(anyhow::Error::from)?; - let cli = LaunchCli::from_arg_matches(&matches).map_err(anyhow::Error::from)?; - cli.into_launch("default-id", "robot") + let cli = ToolLaunchCli::from_arg_matches(&matches).map_err(anyhow::Error::from)?; + cli.common.into_launch("default-id", "robot") } #[test] #[serial] fn cli_with_nothing_set_matches_local_defaults() { clear_env(); - let launch = parse_from(&["participant-bin"]).unwrap(); + let launch = parse_clocked_from(&["participant-bin"]).unwrap(); assert_eq!(launch.participant_id, "default-id"); assert_eq!(launch.robot_id, "robot"); assert_eq!(launch.namespace, "dev"); @@ -336,7 +413,7 @@ mod tests { std::env::set_var(env::CONFIG, r#"{"rate_hz":10}"#); std::env::set_var(env::CLOCK, "simulation"); } - let launch = parse_from(&["participant-bin"]).unwrap(); + let launch = parse_clocked_from(&["participant-bin"]).unwrap(); assert_eq!(launch.participant_id, "tof-3"); assert_eq!(launch.robot_id, "robot-a"); assert_eq!(launch.namespace, "lab"); @@ -373,7 +450,7 @@ mod tests { std::env::set_var(env::CLOCK, "simulation"); } - let launch = parse_from(&[ + let launch = parse_clocked_from(&[ "participant-bin", "--participant-id", "flag-participant", @@ -414,12 +491,12 @@ mod tests { clear_env(); // SAFETY: serialized test; see clear_env. unsafe { std::env::set_var(env::CONFIG, "not json") }; - assert!(parse_from(&["participant-bin"]).is_err()); + assert!(parse_clocked_from(&["participant-bin"]).is_err()); unsafe { std::env::remove_var(env::CONFIG); std::env::set_var(env::CLOCK, "wallclock"); } - let err = LaunchCli::command_for("default-id", "robot") + let err = command_for::("default-id", "robot") .try_get_matches_from(["participant-bin"]) .unwrap_err(); assert_eq!(err.kind(), ErrorKind::InvalidValue); @@ -434,7 +511,7 @@ mod tests { unsafe { std::env::set_var(env::CONFIG, r#"{"secret":"do-not-print"}"#) }; let mut help = Vec::new(); - LaunchCli::command_for("default-id", "robot") + command_for::("default-id", "robot") .write_long_help(&mut help) .unwrap(); let help = String::from_utf8(help).unwrap(); @@ -455,4 +532,45 @@ mod tests { assert!(!help.contains("do-not-print")); clear_env(); } + + #[test] + #[serial] + fn tool_cli_has_no_clock_input() { + clear_env(); + // A generic supervisor setting is invisible to the tool launch parser. + // SAFETY: serialized test; see clear_env. + unsafe { std::env::set_var(env::CLOCK, "simulation") }; + let launch = parse_tool_from(&["tool-bin"]).unwrap(); + assert_eq!(launch.clock, ClockMode::Real); + + let mut help = Vec::new(); + command_for::("default-id", "robot") + .write_long_help(&mut help) + .unwrap(); + let help = String::from_utf8(help).unwrap(); + assert!(!help.contains("--clock")); + assert!(!help.contains(env::CLOCK)); + + for arguments in [ + vec!["tool-bin", "--clock", "simulation"], + vec!["tool-bin", "--simulation"], + ] { + let error = command_for::("default-id", "robot") + .try_get_matches_from(arguments) + .unwrap_err(); + assert_eq!(error.kind(), ErrorKind::UnknownArgument); + } + + let mut programmatic = ParticipantLaunch::local("tool", "robot"); + programmatic.clock = ClockMode::Simulation; + assert_eq!( + ToolParticipantLaunch::clock_mode(&programmatic), + ClockMode::Real + ); + assert_eq!( + ClockedParticipantLaunch::clock_mode(&programmatic), + ClockMode::Simulation + ); + clear_env(); + } } diff --git a/phoxal/src/participant/runner.rs b/phoxal/src/participant/runner.rs index ac218d0..029afdc 100644 --- a/phoxal/src/participant/runner.rs +++ b/phoxal/src/participant/runner.rs @@ -78,7 +78,7 @@ use crate::participant::bus_log::{self, BusLogState}; use crate::participant::clock::{ClockSource, RealClock}; use crate::participant::context::{SetupContext, ShutdownContext, StepContext}; use crate::participant::heartbeat::{self, HeartbeatPublisher}; -use crate::participant::launch::{ClockMode, ParticipantLaunch}; +use crate::participant::launch::{ClockMode, ParticipantLaunch, ParticipantLaunchPolicy}; use crate::participant::managed::{ManagedTaskExit, ManagedTasks}; use crate::participant::process_metrics::{self, ProcessMetricsPublisher}; use crate::participant::scheduler::{ @@ -104,27 +104,16 @@ pub fn run() -> crate::Result<()> { /// Async host runner for custom Tokio mains /// (`phoxal::tokio::run::().await`). pub async fn run_async() -> crate::Result<()> { - let launch = ParticipantLaunch::from_cli(R::ID, "robot")?; + let launch = R::LaunchPolicy::from_cli(R::ID, "robot")?; init_tracing(); - let clock = launch_clock(&launch)?; - run_with::(launch, clock, shutdown_signal()).await -} - -/// The caller-provided clock for a launch. Real mode uses it directly for -/// timestamps and scheduler anchoring. Simulation mode replaces it with the -/// clock derived from its [`SimulationScheduler`] so step release and every -/// lifecycle timestamp share the authoritative `simulation/clock` domain. -pub(crate) fn launch_clock(launch: &ParticipantLaunch) -> crate::Result { - match launch.clock { - ClockMode::Real | ClockMode::Simulation => Ok(RealClock::new()), - } + run_with::(launch, shutdown_signal()).await } /// Select the step scheduler for `clock_mode` (D34/#09): the seam that /// answers "when should the next `#[step]` tick fire", separate from the -/// [`ClockSource`] used for timestamps (see [`launch_clock`]). +/// [`ClockSource`] used for timestamps. /// /// [`ClockMode::Real`] preserves the runner's pre-#09 wall-clock cadence /// exactly, returning no driving handle (`None`). [`ClockMode::Simulation`] @@ -226,16 +215,12 @@ pub(crate) fn spawn_simulation_clock_feed( })) } -/// Run a participant against an explicit launch, clock, and shutdown trigger. The -/// seam the test harness + integration tests drive (D41). -pub async fn run_with( - launch: ParticipantLaunch, - clock: C, - shutdown: S, -) -> crate::Result<()> +/// Run a participant against an explicit launch and shutdown trigger. The +/// runner owns the host clock; tools therefore have no clock parameter to +/// receive or override. +pub async fn run_with(launch: ParticipantLaunch, shutdown: S) -> crate::Result<()> where R: ParticipantLifecycle, - C: ClockSource, S: Future, { init_tracing(); @@ -249,7 +234,7 @@ where }) .await?; - let result = run_with_bus::(&bus, launch, clock, shutdown).await; + let result = run_with_bus::(&bus, launch, shutdown).await; if let Err(e) = bus.close().await { tracing::warn!(target: "phoxal.runtime", error = %e, "bus close failed"); @@ -257,7 +242,7 @@ where result } -/// Run a participant on a **caller-owned** bus, against an explicit launch, clock, and +/// Run a participant on a **caller-owned** bus, against an explicit launch and /// shutdown trigger. Unlike [`run_with`], this does not open or close the bus - the /// caller controls its lifecycle. /// @@ -268,7 +253,39 @@ where /// sharing one [`Bus`] publish under that bus's participant id, so distinct /// per-participant source attribution still requires a bus per participant. The /// `launch` here drives config, robot-model, and component-instance resolution. -pub async fn run_with_bus( +pub async fn run_with_bus( + bus: &Bus, + launch: ParticipantLaunch, + shutdown: S, +) -> crate::Result<()> +where + R: ParticipantLifecycle, + S: Future, +{ + run_with_bus_inner::(bus, launch, RealClock::new(), shutdown).await +} + +/// Deterministic clock-injection seam for checked graph participants. A tool's +/// fixed [`ToolParticipantLaunch`](crate::participant::launch::ToolParticipantLaunch) +/// policy excludes it even if user code manually adds the public +/// [`TypedGraphSurface`](crate::participant::TypedGraphSurface) marker. +#[doc(hidden)] +pub async fn run_with_bus_clock( + bus: &Bus, + launch: ParticipantLaunch, + clock: C, + shutdown: S, +) -> crate::Result<()> +where + R: ParticipantLifecycle + + crate::participant::TypedGraphSurface, + C: ClockSource, + S: Future, +{ + run_with_bus_inner::(bus, launch, clock, shutdown).await +} + +async fn run_with_bus_inner( bus: &Bus, launch: ParticipantLaunch, clock: C, @@ -300,7 +317,8 @@ where S: Future, { let schedule = R::__step_schedule(); - let (scheduler, clock_handle) = step_scheduler_for(launch.clock, schedule, clock.now()); + let clock_mode = R::LaunchPolicy::clock_mode(&launch); + let (scheduler, clock_handle) = step_scheduler_for(clock_mode, schedule, clock.now()); let effective_clock = Arc::new(match &scheduler { AnyStepScheduler::Simulation(sim) => RunnerClock::Simulation(sim.simulation_clock()), AnyStepScheduler::Real(_) => RunnerClock::Delegated(clock), @@ -336,9 +354,9 @@ where /// The runner's effective timestamp clock, chosen once the scheduler is built. /// -/// In real mode it simply delegates to the caller-provided [`ClockSource`] (the -/// host [`RealClock`], or a [`TestClock`](crate::participant::clock::TestClock) -/// under the test harness). In simulation mode it is the +/// In real mode it delegates to the runner-owned host [`RealClock`] (or a +/// [`TestClock`](crate::participant::clock::TestClock) through the checked-graph +/// test seam). In simulation mode it is the /// [`SimulationClock`](crate::participant::clock::SimulationClock) that shares /// the `SimulationScheduler`'s live `simulation/clock` feed, so stamped time and /// step-release time stay in the one simulation domain (see the `SimulationClock` @@ -391,7 +409,6 @@ where // `api::topic::internal::new(cap)`). The runner is the only minter. let mut ctx = SetupContext::::new( bus.clone(), - Arc::clone(&clock) as Arc, ::phoxal_bus::OwnerCap::__mint(), robot, launch.robot_root.clone(), @@ -876,17 +893,6 @@ pub(crate) fn bus_log_state() -> Arc { mod tests { use super::*; - #[test] - fn simulation_clock_launch_is_accepted() { - // #09: PHOXAL_CLOCK=simulation is no longer rejected - the runner - // selects a `SimulationScheduler` for it (see `step_scheduler_for`) - // instead of bailing before the bus even opens. - let mut launch = ParticipantLaunch::local("participant", "robot"); - launch.clock = ClockMode::Simulation; - - launch_clock(&launch).expect("simulation clock launch should be accepted"); - } - #[test] fn step_scheduler_for_selects_real_or_simulation_by_clock_mode() { let now = LogicalTime::new(0, 0); diff --git a/phoxal/tests/interop.rs b/phoxal/tests/interop.rs index 6fddb36..67b34e1 100644 --- a/phoxal/tests/interop.rs +++ b/phoxal/tests/interop.rs @@ -10,7 +10,7 @@ use std::sync::atomic::{AtomicU32, AtomicU64, Ordering}; use std::time::Duration; -use phoxal::participant::{ParticipantLaunch, RealClock}; +use phoxal::participant::ParticipantLaunch; use phoxal::prelude::*; use phoxal::raw::{Bus, BusConfig, run_with_bus}; use phoxal_api::v1 as api; @@ -108,16 +108,14 @@ async fn two_runtimes_exchange_a_contract_on_one_bus() { // Run both runtimes concurrently on the shared bus; each stops after a short // wall-clock window (enough for the 50 Hz producer to emit many samples). - let producer = run_with_bus::( + let producer = run_with_bus::( &bus, ParticipantLaunch::local("producer-1", "robot"), - RealClock::new(), async { tokio::time::sleep(Duration::from_millis(500)).await }, ); - let consumer = run_with_bus::( + let consumer = run_with_bus::( &bus, ParticipantLaunch::local("consumer-1", "robot"), - RealClock::new(), async { tokio::time::sleep(Duration::from_millis(500)).await }, ); diff --git a/phoxal/tests/interop_dynamic.rs b/phoxal/tests/interop_dynamic.rs index 9074d20..f4d9ac9 100644 --- a/phoxal/tests/interop_dynamic.rs +++ b/phoxal/tests/interop_dynamic.rs @@ -12,7 +12,7 @@ use std::sync::atomic::{AtomicU32, AtomicU64, Ordering}; use std::time::Duration; -use phoxal::participant::{ParticipantLaunch, RealClock}; +use phoxal::participant::ParticipantLaunch; use phoxal::prelude::*; use phoxal::raw::{Bus, BusConfig, run_with_bus}; use phoxal_api::v1 as api; @@ -118,16 +118,14 @@ async fn two_runtimes_exchange_a_dynamic_topic_on_one_bus() { .await .expect("open shared bus"); - let producer = run_with_bus::( + let producer = run_with_bus::( &bus, ParticipantLaunch::local("encoder-producer-1", "robot"), - RealClock::new(), async { tokio::time::sleep(Duration::from_millis(500)).await }, ); - let consumer = run_with_bus::( + let consumer = run_with_bus::( &bus, ParticipantLaunch::local("encoder-consumer-1", "robot"), - RealClock::new(), async { tokio::time::sleep(Duration::from_millis(500)).await }, ); diff --git a/phoxal/tests/managed_tasks.rs b/phoxal/tests/managed_tasks.rs index cf40b08..f11e8b4 100644 --- a/phoxal/tests/managed_tasks.rs +++ b/phoxal/tests/managed_tasks.rs @@ -6,7 +6,7 @@ use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::time::Duration; -use phoxal::participant::{ManagedTaskPolicy, ParticipantLaunch, RealClock, run_with}; +use phoxal::participant::{ManagedTaskPolicy, ParticipantLaunch, run_with}; use phoxal::prelude::*; static NAMESPACE_SEQ: AtomicU64 = AtomicU64::new(0); @@ -49,7 +49,7 @@ async fn managed_task_panic_faults_the_runtime() { let result = tokio::time::timeout( Duration::from_secs(5), - run_with::(launch, RealClock::new(), shutdown), + run_with::(launch, shutdown), ) .await .expect("runner must not hang after a managed task panics"); @@ -92,7 +92,7 @@ async fn managed_task_early_return_faults_by_default() { let result = tokio::time::timeout( Duration::from_secs(5), - run_with::(launch, RealClock::new(), shutdown), + run_with::(launch, shutdown), ) .await .expect("runner must not hang after a managed task returns early"); @@ -138,8 +138,7 @@ async fn managed_task_early_return_under_allow_exit_does_not_fault() { tokio::time::sleep(Duration::from_millis(200)).await; }; - let result = - run_with::(launch, RealClock::new(), shutdown).await; + let result = run_with::(launch, shutdown).await; assert!(EARLY_RETURN_ALLOWED_TASK_RAN.load(Ordering::Relaxed)); result.expect("an AllowExit managed task returning early must not fault the runtime"); @@ -183,7 +182,7 @@ async fn managed_task_panic_under_allow_exit_does_not_fault() { tokio::time::sleep(Duration::from_millis(200)).await; }; - let result = run_with::(launch, RealClock::new(), shutdown).await; + let result = run_with::(launch, shutdown).await; assert!(ALLOWED_PANIC_TASK_RAN.load(Ordering::Relaxed)); result.expect("an AllowExit managed task panicking must not fault the runtime"); @@ -233,7 +232,7 @@ async fn clean_shutdown_cancels_and_joins_managed_tasks() { let result = tokio::time::timeout( Duration::from_secs(5), - run_with::(launch, RealClock::new(), shutdown), + run_with::(launch, shutdown), ) .await .expect("runner must not hang joining a cooperative managed task"); @@ -279,7 +278,7 @@ async fn setup_failure_cancels_spawned_managed_tasks() { let result = tokio::time::timeout( Duration::from_secs(5), - run_with::(launch, RealClock::new(), shutdown), + run_with::(launch, shutdown), ) .await .expect("runner must not hang cleaning up a managed task after setup fails"); @@ -342,7 +341,7 @@ async fn shutdown_reports_unjoined_managed_tasks_after_grace_elapses() { let started = std::time::Instant::now(); let result = tokio::time::timeout( Duration::from_secs(10), - run_with::(launch, RealClock::new(), shutdown), + run_with::(launch, shutdown), ) .await .expect("runner must not hang past its own timeout budget waiting on a stuck task"); diff --git a/phoxal/tests/robot_root.rs b/phoxal/tests/robot_root.rs index 42077c4..f9454b8 100644 --- a/phoxal/tests/robot_root.rs +++ b/phoxal/tests/robot_root.rs @@ -5,7 +5,7 @@ use std::path::PathBuf; use std::sync::OnceLock; use std::time::Duration; -use phoxal::participant::{ParticipantLaunch, RealClock, run_with}; +use phoxal::participant::{ParticipantLaunch, run_with}; use phoxal::prelude::*; use serial_test::serial; @@ -40,7 +40,7 @@ async fn setup_reads_robot_model_from_the_root() { tokio::time::sleep(Duration::from_millis(50)).await; }; - run_with::(launch, RealClock::new(), shutdown) + run_with::(launch, shutdown) .await .expect("runner should complete with a robot root"); @@ -58,7 +58,7 @@ async fn robot_is_absent_without_a_root() { let shutdown = async { tokio::time::sleep(Duration::from_millis(50)).await; }; - let result = run_with::(launch, RealClock::new(), shutdown).await; + let result = run_with::(launch, shutdown).await; assert!( result.is_err(), "setup should fail when no robot model is bound" diff --git a/phoxal/tests/runner.rs b/phoxal/tests/runner.rs index e1a0848..58022d9 100644 --- a/phoxal/tests/runner.rs +++ b/phoxal/tests/runner.rs @@ -29,9 +29,9 @@ use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::time::Duration; use phoxal::bus::{DEFAULT_QUERY_TIMEOUT, Latest, LogicalTime, OwnerCap, Publisher, Querier}; -use phoxal::participant::{ClockMode, ParticipantLaunch, RealClock, TestClock}; +use phoxal::participant::{ClockMode, ParticipantLaunch, TestClock}; use phoxal::prelude::*; -use phoxal::raw::{Bus, BusConfig, run_with_bus}; +use phoxal::raw::{Bus, BusConfig, run_with_bus, run_with_bus_clock}; use phoxal_api::ContractBody; use phoxal_api::v1 as api; use phoxal_api::v2; @@ -42,6 +42,8 @@ static COUNTER_STEPS: AtomicU64 = AtomicU64::new(0); static SHUTDOWN_CALLED: AtomicBool = AtomicBool::new(false); static SLOW_SHUTDOWN_COMPLETED: AtomicBool = AtomicBool::new(false); static SIM_CLOCK_STEPS: AtomicU64 = AtomicU64::new(0); +static HOST_TOOL_TICKS: AtomicU64 = AtomicU64::new(0); +static HOST_TOOL_MESSAGES: AtomicU64 = AtomicU64::new(0); static SIM_CLOCK_CONTEXTS: Mutex> = Mutex::new(Vec::new()); /// A fresh in-process namespace per test invocation, so concurrently-run @@ -190,7 +192,7 @@ async fn new_model_participant_runs_through_a_real_bus() { .expect("build submap querier"); let launch = ParticipantLaunch::local("wall-follower-v2-1", "robot"); - let runner = run_with_bus::(&bus, launch, RealClock::new(), async { + let runner = run_with_bus::(&bus, launch, async { tokio::time::sleep(Duration::from_millis(600)).await }); @@ -432,7 +434,7 @@ async fn subscriber_and_latest_survive_the_owned_arc_split() { .expect("build submap querier"); let launch = ParticipantLaunch::local("drain-proof-v2-1", "robot"); - let runner = run_with_bus::(&bus, launch, RealClock::new(), async { + let runner = run_with_bus::(&bus, launch, async { tokio::time::sleep(Duration::from_millis(800)).await }); @@ -695,6 +697,39 @@ impl RobotInspector { } } +#[phoxal::tool(id = "host-driven-tool")] +struct HostDrivenTool; + +#[phoxal::behavior] +impl HostDrivenTool { + #[setup] + async fn setup(ctx: &mut SetupContext) -> Result<(Self, Self::Api)> { + let bus = ctx.raw_bus(); + let manual = phoxal::raw::Subscriber::::new( + &bus, + &api::topic::internal::new(ctx.owner_capability()) + .motion() + .manual(), + 8, + ) + .await?; + ctx.spawn_managed("host-ticker", async { + let mut interval = tokio::time::interval(Duration::from_millis(10)); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + loop { + interval.tick().await; + HOST_TOOL_TICKS.fetch_add(1, Ordering::Relaxed); + } + }); + ctx.spawn_managed("raw-subscriber", async move { + while manual.recv().await.is_ok() { + HOST_TOOL_MESSAGES.fetch_add(1, Ordering::Relaxed); + } + }); + Ok((Self, ())) + } +} + #[derive(serde::Deserialize, phoxal::Config)] struct ConfiguredInspectorConfig { label: String, @@ -718,24 +753,83 @@ impl ConfiguredInspector { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn configless_tool_accepts_absent_config_but_configured_tool_rejects_it() { let configless = ParticipantLaunch::local("robot-inspector", "robot"); - phoxal::participant::run_with::(configless, RealClock::new(), async {}) + phoxal::participant::run_with::(configless, async {}) .await .expect("a tool with omitted config type should accept absent PHOXAL_CONFIG"); let configured = ParticipantLaunch::local("configured-inspector", "robot"); - let error = phoxal::participant::run_with::( - configured, - RealClock::new(), - async {}, - ) - .await - .expect_err("a tool with an explicit non-optional config should require PHOXAL_CONFIG"); + let error = phoxal::participant::run_with::(configured, async {}) + .await + .expect_err("a tool with an explicit non-optional config should require PHOXAL_CONFIG"); assert!( error.to_string().contains("invalid type: null"), "unexpected absent-config error: {error:#}" ); } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn clockless_tool_keeps_host_work_and_raw_subscriptions_running() { + HOST_TOOL_TICKS.store(0, Ordering::Relaxed); + HOST_TOOL_MESSAGES.store(0, Ordering::Relaxed); + let participant_id = "host-driven-tool-1"; + let namespace = unique_namespace("host-driven-tool"); + let mut bus_config = BusConfig::in_process(namespace.clone(), "robot"); + bus_config.participant = participant_id.to_string(); + let bus = Bus::open(bus_config).await.expect("bus should open"); + let heartbeat_topic = api::topic::internal::new(OwnerCap::__mint()) + .presence() + .heartbeat(); + let heartbeats = Subscriber::::new(&bus, &heartbeat_topic, 8) + .await + .expect("heartbeat subscriber should attach"); + let manual = phoxal::raw::Publisher::new(bus.clone(), &api::topic::new().motion().manual()) + .expect("manual publisher should attach"); + + let mut launch = ParticipantLaunch::local(participant_id, "robot"); + launch.namespace = namespace; + run_with_bus::(&bus, launch, async move { + tokio::time::sleep(Duration::from_millis(20)).await; + manual + .publish_at( + LogicalTime::new(7, 42), + api::motion::ManualCommand { + linear_x_mps: 0.2, + angular_z_radps: 0.0, + }, + ) + .await + .expect("raw tool input should publish"); + tokio::time::sleep(Duration::from_millis(80)).await; + }) + .await + .expect("tool should run without a logical clock input"); + + assert!( + HOST_TOOL_TICKS.load(Ordering::Relaxed) >= 2, + "host ticker must run without a logical clock" + ); + assert_eq!( + HOST_TOOL_MESSAGES.load(Ordering::Relaxed), + 1, + "raw tool subscriptions must run without a logical clock" + ); + let mut produced_at = Vec::new(); + while let Some(received) = heartbeats.try_recv() { + if received.body.participant == participant_id { + produced_at.push(received.metadata.produced_at_ns); + } + } + assert!( + !produced_at.is_empty(), + "tool heartbeats should be observed" + ); + assert!( + produced_at.iter().all(|at| *at > 1_000_000_000_000_000_000), + "tool lifecycle timestamps must stay in host time, got {produced_at:?}" + ); + bus.close().await.expect("bus should close"); +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn runner_runs_steps_then_shuts_down_cleanly() { let launch = ParticipantLaunch::local("counter-1", "robot"); @@ -743,7 +837,7 @@ async fn runner_runs_steps_then_shuts_down_cleanly() { tokio::time::sleep(Duration::from_millis(200)).await; }; - phoxal::participant::run_with::(launch, RealClock::new(), shutdown) + phoxal::participant::run_with::(launch, shutdown) .await .expect("runner should complete cleanly"); @@ -774,7 +868,7 @@ async fn runner_publishes_presence_heartbeats_from_idle_loop() { let mut launch = ParticipantLaunch::local(participant_id, "robot"); launch.namespace = namespace; - let runner = run_with_bus::(&bus, launch, RealClock::new(), async { + let runner = run_with_bus::(&bus, launch, async { tokio::time::sleep(Duration::from_millis(2200)).await; }); let collector = async { @@ -846,7 +940,7 @@ async fn simulation_setup_failure_heartbeats_use_simulation_time() { let injected_clock = TestClock::new(); injected_clock.advance(Duration::from_secs(123)); - let error = run_with_bus::(&bus, launch, injected_clock, async {}) + let error = run_with_bus_clock::(&bus, launch, injected_clock, async {}) .await .expect_err("setup should fail"); assert!(error.to_string().contains("intentional setup failure")); @@ -886,7 +980,7 @@ async fn slow_shutdown_hook_is_bounded_by_grace() { let started = std::time::Instant::now(); tokio::time::timeout( Duration::from_secs(10), - phoxal::participant::run_with::(launch, RealClock::new(), shutdown), + phoxal::participant::run_with::(launch, shutdown), ) .await .expect("runner must not hang on a slow shutdown hook") @@ -963,7 +1057,7 @@ async fn simulation_mode_step_advances_only_with_the_clock_feed() { // mode must ignore it for StepContext and publication timestamps. let injected_clock = TestClock::new(); injected_clock.advance(Duration::from_secs(123)); - let runner = run_with_bus::(&bus, launch, injected_clock, async { + let runner = run_with_bus_clock::(&bus, launch, injected_clock, async { // No steps should have released yet: the feed has not published a // single sample, so the scheduler's logical time has never left // `start`. @@ -1090,7 +1184,7 @@ async fn driver_reads_its_bound_component_instance() { tokio::time::sleep(Duration::from_millis(50)).await; }; - phoxal::participant::run_with::(launch, RealClock::new(), shutdown) + phoxal::participant::run_with::(launch, shutdown) .await .expect("driver should read its bound component instance and run cleanly"); } diff --git a/phoxal/tests/trybuild/fail/default_bus_module_cannot_open_raw_bus.rs b/phoxal/tests/trybuild/fail/default_bus_module_cannot_open_raw_bus.rs index d0a7ed5..099e785 100644 --- a/phoxal/tests/trybuild/fail/default_bus_module_cannot_open_raw_bus.rs +++ b/phoxal/tests/trybuild/fail/default_bus_module_cannot_open_raw_bus.rs @@ -15,7 +15,7 @@ impl DefaultRawBusOpen { async fn setup(_ctx: &mut SetupContext) -> Result<(Self, Self::Api)> { let _config = phoxal::bus::BusConfig::in_process("dev", "robot"); let _open = phoxal::bus::Bus::open; - let _run_with_bus = phoxal::participant::run_with_bus::; + let _run_with_bus = phoxal::participant::run_with_bus::; Ok((Self, ())) } } diff --git a/phoxal/tests/trybuild/fail/default_bus_module_cannot_open_raw_bus.stderr b/phoxal/tests/trybuild/fail/default_bus_module_cannot_open_raw_bus.stderr index c9f69f0..6347bb8 100644 --- a/phoxal/tests/trybuild/fail/default_bus_module_cannot_open_raw_bus.stderr +++ b/phoxal/tests/trybuild/fail/default_bus_module_cannot_open_raw_bus.stderr @@ -13,22 +13,19 @@ error[E0433]: cannot find `Bus` in `bus` error[E0425]: cannot find value `run_with_bus` in module `phoxal::participant` --> tests/trybuild/fail/default_bus_module_cannot_open_raw_bus.rs:18:50 | -18 | let _run_with_bus = phoxal::participant::run_with_bus::; +18 | let _run_with_bus = phoxal::participant::run_with_bus::; | ^^^^^^^^^^^^ | ::: src/participant/runner.rs | - | / pub async fn run_with( - | | launch: ParticipantLaunch, - | | clock: C, - | | shutdown: S, -... | - | | C: ClockSource, + | / pub async fn run_with(launch: ParticipantLaunch, shutdown: S) -> crate::Result<()> + | | where + | | R: ParticipantLifecycle, | | S: Future, | |___________________________- similarly named function `run_with` defined here | help: a function with a similar name exists | -18 - let _run_with_bus = phoxal::participant::run_with_bus::; -18 + let _run_with_bus = phoxal::participant::run_with::; +18 - let _run_with_bus = phoxal::participant::run_with_bus::; +18 + let _run_with_bus = phoxal::participant::run_with::; | diff --git a/phoxal/tests/trybuild/fail/tool_cannot_access_clock.rs b/phoxal/tests/trybuild/fail/tool_cannot_access_clock.rs new file mode 100644 index 0000000..2cc71ca --- /dev/null +++ b/phoxal/tests/trybuild/fail/tool_cannot_access_clock.rs @@ -0,0 +1,25 @@ +use phoxal::prelude::*; + +#[phoxal::tool(id = "clocked-tool")] +struct ClockedTool; + +// Even manually adding the public graph marker cannot change the launch policy +// that #[phoxal::tool] fixed in the Participant impl. +impl phoxal::participant::TypedGraphSurface for ClockedTool {} + +#[phoxal::behavior] +impl ClockedTool { + #[setup] + async fn setup(ctx: &mut SetupContext) -> Result<(Self, Self::Api)> { + let _clock = ctx.clock(); + Ok((Self, ())) + } +} + +fn main() { + let _inject = phoxal::raw::run_with_bus_clock::< + ClockedTool, + phoxal::participant::TestClock, + std::future::Ready<()>, + >; +} diff --git a/phoxal/tests/trybuild/fail/tool_cannot_access_clock.stderr b/phoxal/tests/trybuild/fail/tool_cannot_access_clock.stderr new file mode 100644 index 0000000..c70e6b4 --- /dev/null +++ b/phoxal/tests/trybuild/fail/tool_cannot_access_clock.stderr @@ -0,0 +1,26 @@ +error[E0599]: no method named `clock` found for mutable reference `&mut phoxal::participant::SetupContext` in the current scope + --> tests/trybuild/fail/tool_cannot_access_clock.rs:14:26 + | +14 | let _clock = ctx.clock(); + | ^^^^^ method not found in `&mut phoxal::participant::SetupContext` + +error[E0271]: type mismatch resolving `::LaunchPolicy == ClockedParticipantLaunch` + --> tests/trybuild/fail/tool_cannot_access_clock.rs:21:9 + | +21 | ClockedTool, + | ^^^^^^^^^^^ type mismatch resolving `::LaunchPolicy == ClockedParticipantLaunch` + | +note: expected this to be `phoxal::participant::launch::ClockedParticipantLaunch` + --> tests/trybuild/fail/tool_cannot_access_clock.rs:3:1 + | + 3 | #[phoxal::tool(id = "clocked-tool")] + | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ +note: required by a bound in `phoxal::raw::run_with_bus_clock` + --> src/participant/runner.rs + | + | pub async fn run_with_bus_clock( + | ------------------ required by a bound in this function +... + | R: ParticipantLifecycle + | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ required by this bound in `run_with_bus_clock` + = note: this error originates in the attribute macro `phoxal::tool` (in Nightly builds, run with -Z macro-backtrace for more info) diff --git a/service/motion/Cargo.toml b/service/motion/Cargo.toml index 8932584..1077527 100644 --- a/service/motion/Cargo.toml +++ b/service/motion/Cargo.toml @@ -21,6 +21,7 @@ anyhow = { workspace = true } [dev-dependencies] serde_json = { workspace = true } +tokio = { workspace = true } [lints] workspace = true diff --git a/service/motion/src/arbitration.rs b/service/motion/src/arbitration.rs index b5af843..90ff028 100644 --- a/service/motion/src/arbitration.rs +++ b/service/motion/src/arbitration.rs @@ -1,12 +1,20 @@ //! Body-twist arbitration for manual and navigation candidates. +use std::time::{Duration, Instant}; + use phoxal::bus::LogicalTime; use phoxal::model::robot::v0::MotionLimits; use phoxal_api::v1 as api; -pub(crate) const MANUAL_STALE_NS: u64 = 500_000_000; +pub(crate) const MANUAL_STALE: Duration = Duration::from_millis(150); pub(crate) const AUTONOMOUS_STALE_NS: u64 = 500_000_000; +#[derive(Clone)] +pub(crate) struct ManualCandidate { + pub(crate) body: api::motion::ManualCommand, + pub(crate) received_at: Instant, +} + #[derive(Clone)] pub(crate) struct Timed { pub(crate) body: T, @@ -21,12 +29,13 @@ pub(crate) struct Arbitration { } pub(crate) fn arbitrate( - manual: Option<&Timed>, + manual: Option<&ManualCandidate>, autonomous: Option<&Timed>, emergency_stop_engaged: bool, safety: Option<&Timed>, limits: MotionLimits, now: LogicalTime, + host_now: Instant, ) -> Arbitration { if emergency_stop_engaged { return zero( @@ -35,16 +44,16 @@ pub(crate) fn arbitrate( ); } - if manual.is_some_and(|candidate| { - candidate.at.epoch() == now.epoch() && candidate.at.time_ns() > now.time_ns() - }) { + if manual.is_some_and(|candidate| candidate.received_at > host_now) { return zero( Some(api::motion::Source::Manual), api::motion::ZeroReason::ManualCandidateFromFuture, ); } - if let Some(manual) = manual.filter(|candidate| is_fresh(candidate.at, now, MANUAL_STALE_NS)) { + if let Some(manual) = + manual.filter(|candidate| host_now.duration_since(candidate.received_at) <= MANUAL_STALE) + { // Manual teleoperation is the recovery/commissioning path and must not // depend on the still-experimental world-safety provider being ready. // A valid protective stop still wins, and every manual command remains @@ -166,6 +175,16 @@ pub(crate) fn candidate_age_ns(candidate: Option<&Timed>, now: LogicalTime }) } +pub(crate) fn manual_candidate_age_ns( + candidate: Option<&ManualCandidate>, + now: Instant, +) -> Option { + candidate.and_then(|candidate| { + now.checked_duration_since(candidate.received_at) + .map(|age| u64::try_from(age.as_nanos()).unwrap_or(u64::MAX)) + }) +} + fn select( source: api::motion::Source, linear_x_mps: f64, @@ -235,6 +254,20 @@ mod tests { LogicalTime::new(0, NOW_NS) } + fn host_now() -> Instant { + Instant::now() + } + + fn manual(linear_x_mps: f64, angular_z_radps: f64, received_at: Instant) -> ManualCandidate { + ManualCandidate { + body: api::motion::ManualCommand { + linear_x_mps, + angular_z_radps, + }, + received_at, + } + } + fn safety() -> Timed { Timed { body: api::safety::MotionConstraints { @@ -285,7 +318,15 @@ mod tests { let mut safety = safety(); safety.body.constraints.push(safety_constraint(true)); let autonomous = autonomous(); - let result = arbitrate(None, Some(&autonomous), false, Some(&safety), LIMITS, now()); + let result = arbitrate( + None, + Some(&autonomous), + false, + Some(&safety), + LIMITS, + now(), + host_now(), + ); assert_eq!( result.zero_reason, Some(api::motion::ZeroReason::SafetyConstraintsUnavailable) @@ -310,7 +351,15 @@ mod tests { mutate(&mut constraint); safety.body.constraints.push(constraint); let autonomous = autonomous(); - let result = arbitrate(None, Some(&autonomous), false, Some(&safety), LIMITS, now()); + let result = arbitrate( + None, + Some(&autonomous), + false, + Some(&safety), + LIMITS, + now(), + host_now(), + ); assert_eq!( result.zero_reason, Some(api::motion::ZeroReason::SafetyConstraintsUnavailable) @@ -320,7 +369,15 @@ mod tests { #[test] fn missing_candidates_are_explained() { - let result = arbitrate(None, None, false, Some(&safety()), LIMITS, now()); + let result = arbitrate( + None, + None, + false, + Some(&safety()), + LIMITS, + now(), + host_now(), + ); assert_eq!(result.source, None); assert_eq!( result.zero_reason, @@ -330,14 +387,17 @@ mod tests { #[test] fn stale_manual_is_distinct_from_missing_manual() { - let manual = Timed { - body: api::motion::ManualCommand { - linear_x_mps: 0.2, - angular_z_radps: 0.1, - }, - at: LogicalTime::new(0, NOW_NS - MANUAL_STALE_NS - 1), - }; - let result = arbitrate(Some(&manual), None, false, Some(&safety()), LIMITS, now()); + let host_now = host_now(); + let manual = manual(0.2, 0.1, host_now - MANUAL_STALE - Duration::from_nanos(1)); + let result = arbitrate( + Some(&manual), + None, + false, + Some(&safety()), + LIMITS, + now(), + host_now, + ); assert_eq!( result.zero_reason, Some(api::motion::ZeroReason::ManualCandidateStale) @@ -346,14 +406,8 @@ mod tests { #[test] fn emergency_stop_overrides_manual_and_navigation() { - let manual = Timed { - body: api::motion::ManualCommand { - linear_x_mps: 0.2, - angular_z_radps: 0.1, - }, - at: LogicalTime::new(0, NOW_NS), - }; - let result = arbitrate(Some(&manual), None, true, None, LIMITS, now()); + let manual = manual(0.2, 0.1, host_now()); + let result = arbitrate(Some(&manual), None, true, None, LIMITS, now(), host_now()); assert_eq!(result.selected, zero_target()); assert_eq!( result.zero_reason, @@ -363,14 +417,16 @@ mod tests { #[test] fn manual_wins_and_is_clamped_to_robot_limits() { - let manual = Timed { - body: api::motion::ManualCommand { - linear_x_mps: 9.0, - angular_z_radps: -9.0, - }, - at: LogicalTime::new(0, NOW_NS), - }; - let result = arbitrate(Some(&manual), None, false, Some(&safety()), LIMITS, now()); + let manual = manual(9.0, -9.0, host_now()); + let result = arbitrate( + Some(&manual), + None, + false, + Some(&safety()), + LIMITS, + now(), + host_now(), + ); assert_eq!(result.source, Some(api::motion::Source::Manual)); assert_eq!(result.selected.linear_x_mps, 0.6); assert_eq!(result.selected.angular_z_radps, -2.0); @@ -378,14 +434,16 @@ mod tests { #[test] fn non_finite_candidate_stops() { - let manual = Timed { - body: api::motion::ManualCommand { - linear_x_mps: f64::NAN, - angular_z_radps: 0.0, - }, - at: LogicalTime::new(0, NOW_NS), - }; - let result = arbitrate(Some(&manual), None, false, Some(&safety()), LIMITS, now()); + let manual = manual(f64::NAN, 0.0, host_now()); + let result = arbitrate( + Some(&manual), + None, + false, + Some(&safety()), + LIMITS, + now(), + host_now(), + ); assert_eq!(result.selected, zero_target()); assert_eq!( result.zero_reason, @@ -395,14 +453,17 @@ mod tests { #[test] fn future_dated_manual_candidate_stops() { - let manual = Timed { - body: api::motion::ManualCommand { - linear_x_mps: 0.2, - angular_z_radps: 0.0, - }, - at: LogicalTime::new(0, NOW_NS + 1), - }; - let result = arbitrate(Some(&manual), None, false, Some(&safety()), LIMITS, now()); + let host_now = host_now(); + let manual = manual(0.2, 0.0, host_now + Duration::from_nanos(1)); + let result = arbitrate( + Some(&manual), + None, + false, + Some(&safety()), + LIMITS, + now(), + host_now, + ); assert_eq!( result.zero_reason, Some(api::motion::ZeroReason::ManualCandidateFromFuture) @@ -411,37 +472,33 @@ mod tests { #[test] fn large_finite_candidate_stays_finite_after_f32_conversion() { - let manual = Timed { - body: api::motion::ManualCommand { - linear_x_mps: 1.0e300, - angular_z_radps: -1.0e300, - }, - at: LogicalTime::new(0, NOW_NS), - }; + let manual = manual(1.0e300, -1.0e300, host_now()); let limits = MotionLimits { max_linear_speed_mps: f64::from(f32::MAX), max_angular_speed_radps: f64::from(f32::MAX), }; - let result = arbitrate(Some(&manual), None, false, Some(&safety()), limits, now()); + let result = arbitrate( + Some(&manual), + None, + false, + Some(&safety()), + limits, + now(), + host_now(), + ); assert!(result.selected.linear_x_mps.is_finite()); assert!(result.selected.angular_z_radps.is_finite()); } #[test] - fn prior_epoch_candidate_cannot_resurrect_after_clock_reset() { + fn logical_epoch_reset_does_not_expire_fresh_manual_input() { let limits = MotionLimits { max_linear_speed_mps: 0.6, max_angular_speed_radps: 2.0, }; - let prior_epoch = Timed { - body: api::motion::ManualCommand { - linear_x_mps: 0.4, - angular_z_radps: 0.2, - }, - at: LogicalTime::new(1, NOW_NS), - }; + let manual = manual(0.4, 0.2, host_now()); let result = arbitrate( - Some(&prior_epoch), + Some(&manual), None, false, Some(&Timed { @@ -450,24 +507,16 @@ mod tests { }), limits, LogicalTime::new(2, NOW_NS), + host_now(), ); - assert_eq!( - result.zero_reason, - Some(api::motion::ZeroReason::ManualCandidateStale) - ); - assert_eq!(result.selected, zero_target()); + assert_eq!(result.source, Some(api::motion::Source::Manual)); + assert_eq!(result.selected.linear_x_mps, 0.4); } #[test] fn manual_works_without_safety_and_uses_valid_constraints_when_available() { - let manual = Timed { - body: api::motion::ManualCommand { - linear_x_mps: 0.5, - angular_z_radps: 0.4, - }, - at: now(), - }; - let missing = arbitrate(Some(&manual), None, false, None, LIMITS, now()); + let manual = manual(0.5, 0.4, host_now()); + let missing = arbitrate(Some(&manual), None, false, None, LIMITS, now(), host_now()); assert_eq!(missing.source, Some(api::motion::Source::Manual)); assert_eq!(missing.selected.linear_x_mps, 0.5); @@ -483,6 +532,7 @@ mod tests { Some(&constrained), LIMITS, now(), + host_now(), ); assert_eq!(result.selected.linear_x_mps, 0.1); assert_eq!(result.selected.angular_z_radps, 0.4); @@ -491,7 +541,15 @@ mod tests { #[test] fn autonomous_still_requires_safety_constraints() { let autonomous = autonomous(); - let missing = arbitrate(None, Some(&autonomous), false, None, LIMITS, now()); + let missing = arbitrate( + None, + Some(&autonomous), + false, + None, + LIMITS, + now(), + host_now(), + ); assert_eq!( missing.zero_reason, Some(api::motion::ZeroReason::SafetyConstraintsUnavailable) @@ -500,17 +558,19 @@ mod tests { #[test] fn valid_safety_protective_stop_still_blocks_manual() { - let manual = Timed { - body: api::motion::ManualCommand { - linear_x_mps: 0.5, - angular_z_radps: 0.4, - }, - at: now(), - }; + let manual = manual(0.5, 0.4, host_now()); let mut stopped = safety(); stopped.body.stop = true; stopped.body.constraints.push(safety_constraint(true)); - let result = arbitrate(Some(&manual), None, false, Some(&stopped), LIMITS, now()); + let result = arbitrate( + Some(&manual), + None, + false, + Some(&stopped), + LIMITS, + now(), + host_now(), + ); assert_eq!(result.selected, zero_target()); assert_eq!( result.zero_reason, diff --git a/service/motion/src/main.rs b/service/motion/src/main.rs index f89c902..9853bad 100644 --- a/service/motion/src/main.rs +++ b/service/motion/src/main.rs @@ -11,6 +11,9 @@ mod arbitration; +use std::sync::{Arc, Mutex}; +use std::time::Instant; + use anyhow::Result; use phoxal::model::component::v0::capability::Capability; use phoxal::model::robot::v0::MotionLimits; @@ -18,7 +21,9 @@ use phoxal::model::v0::Robot; use phoxal::prelude::*; use phoxal_api::v1 as api; -use crate::arbitration::{Timed, arbitrate, candidate_age_ns, safety_is_usable}; +use crate::arbitration::{ + ManualCandidate, Timed, arbitrate, candidate_age_ns, manual_candidate_age_ns, safety_is_usable, +}; const COMPONENT_ESTOP_STALE_NS: u64 = 1_000_000_000; @@ -88,6 +93,9 @@ impl EmergencyStopLatch { #[derive(phoxal::Api)] struct Api { + // Declares the typed graph subscription. The managed receiver owns the + // only draining clone so logical-step pauses cannot accumulate a replay + // backlog in this handle. manual: Subscriber, autonomous: Subscriber, software_estop: Subscriber, @@ -100,7 +108,7 @@ struct Api { #[phoxal::service(id = "motion", config = ())] struct Motion { limits: MotionLimits, - last_manual: Option>, + latest_manual: Arc>>, last_autonomous: Option>, estop: EmergencyStopLatch, last_safety_constraints: Option>, @@ -114,6 +122,15 @@ impl Motion { let robot = ctx.robot()?; let limits = robot.manifest.robot.motion_limits.validate()?; let estop_bindings = emergency_stop_bindings(robot); + let manual_subscriber = ctx + .subscriber(api::topic::internal::new(cap).motion().manual(), 32) + .await?; + let latest_manual = Arc::new(Mutex::new(None)); + let manual_slot = Arc::clone(&latest_manual); + let manual_receiver = manual_subscriber.clone(); + ctx.spawn_managed("manual-receiver", async move { + receive_manual_forever(manual_receiver, manual_slot).await; + }); let mut component_estops = Vec::with_capacity(estop_bindings.len()); for binding in &estop_bindings { @@ -132,15 +149,13 @@ impl Motion { Ok(( Self { limits, - last_manual: None, + latest_manual, last_autonomous: None, estop: EmergencyStopLatch::new(estop_bindings.len()), last_safety_constraints: None, }, Self::Api { - manual: ctx - .subscriber(api::topic::internal::new(cap).motion().manual(), 32) - .await?, + manual: manual_subscriber, autonomous: ctx .subscriber(api::topic::new().navigation().candidate(), 32) .await?, @@ -161,12 +176,8 @@ impl Motion { #[step(hz = 20)] async fn step(&mut self, api: &mut Self::Api, step: StepContext) -> Result<()> { - while let Some(received) = api.manual.try_recv() { - self.last_manual = Some(Timed { - body: received.body, - at: LogicalTime::new(received.metadata.epoch, received.metadata.produced_at_ns), - }); - } + let manual = latest_manual(&self.latest_manual); + let host_now = Instant::now(); while let Some(received) = api.autonomous.try_recv() { self.last_autonomous = Some(Timed { body: received.body, @@ -202,12 +213,13 @@ impl Motion { api::motion::SafetyRuntime::Absent }; let arbitration = arbitrate( - self.last_manual.as_ref(), + manual.as_ref(), self.last_autonomous.as_ref(), self.estop.engaged(step.time()), self.last_safety_constraints.as_ref(), self.limits, step.time(), + host_now, ); self.estop.finish_cycle(); @@ -218,10 +230,7 @@ impl Motion { .publish_at( step.time(), api::motion::State { - manual_candidate_age_ns: candidate_age_ns( - self.last_manual.as_ref(), - step.time(), - ), + manual_candidate_age_ns: manual_candidate_age_ns(manual.as_ref(), host_now), autonomous_candidate_age_ns: candidate_age_ns( self.last_autonomous.as_ref(), step.time(), @@ -253,6 +262,36 @@ impl Motion { } } +async fn receive_manual_forever( + subscriber: Subscriber, + latest: Arc>>, +) { + loop { + match subscriber.recv().await { + Ok(received) => store_manual(&latest, received.body, Instant::now()), + Err(_) => return, + } + } +} + +fn store_manual( + latest: &Mutex>, + body: api::motion::ManualCommand, + received_at: Instant, +) { + *latest + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) = + Some(ManualCandidate { body, received_at }); +} + +fn latest_manual(latest: &Mutex>) -> Option { + latest + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .clone() +} + fn state_target(target: &api::drive::Target) -> api::motion::Target { api::motion::Target { linear_x_mps: target.linear_x_mps, @@ -342,6 +381,92 @@ mod tests { assert_eq!(state.curvature_limit_radpm, drive.curvature_limit_radpm); } + #[test] + fn manual_receiver_slot_is_bounded_and_newest_wins() { + let latest = Mutex::new(None); + let first_at = Instant::now(); + store_manual( + &latest, + api::motion::ManualCommand { + linear_x_mps: 0.1, + angular_z_radps: 0.2, + }, + first_at, + ); + let second_at = first_at + std::time::Duration::from_millis(10); + store_manual( + &latest, + api::motion::ManualCommand { + linear_x_mps: 0.3, + angular_z_radps: 0.4, + }, + second_at, + ); + + let manual = latest_manual(&latest).expect("latest command should be retained"); + assert_eq!(manual.body.linear_x_mps, 0.3); + assert_eq!(manual.body.angular_z_radps, 0.4); + assert_eq!(manual.received_at, second_at); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn manual_receiver_runs_without_logical_steps_and_keeps_newest() { + let namespace = format!( + "test/motion-manual-receiver/{}-{}", + std::process::id(), + phoxal::raw::host_time().time_ns() + ); + let bus = phoxal::raw::Bus::open(phoxal::raw::BusConfig::in_process(namespace, "robot")) + .await + .expect("bus should open"); + let publisher = + phoxal::raw::Publisher::new(bus.clone(), &api::topic::new().motion().manual()) + .expect("manual publisher should attach"); + let subscriber = Subscriber::new( + &bus, + &api::topic::internal::new(phoxal::raw::OwnerCap::__mint()) + .motion() + .manual(), + 32, + ) + .await + .expect("manual subscriber should attach"); + let latest = Arc::new(Mutex::new(None)); + let receiver_slot = Arc::clone(&latest); + let receiver = tokio::spawn(async move { + receive_manual_forever(subscriber, receiver_slot).await; + }); + + for linear_x_mps in [0.1, 0.2, 0.3] { + publisher + .publish_at( + phoxal::raw::host_time(), + api::motion::ManualCommand { + linear_x_mps, + angular_z_radps: 0.0, + }, + ) + .await + .expect("manual command should publish"); + } + + tokio::time::timeout(std::time::Duration::from_secs(2), async { + loop { + if latest_manual(&latest).is_some_and(|manual| manual.body.linear_x_mps == 0.3) { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("receiver should advance independently of logical steps"); + + let manual = latest_manual(&latest).expect("latest manual command should be retained"); + assert_eq!(manual.body.linear_x_mps, 0.3); + receiver.abort(); + bus.close().await.expect("bus should close"); + } + #[test] fn software_and_component_estops_latch_independently() { let mut latch = EmergencyStopLatch::new(2); diff --git a/service/navigation/src/main.rs b/service/navigation/src/main.rs index 869d87f..8f6dfe7 100644 --- a/service/navigation/src/main.rs +++ b/service/navigation/src/main.rs @@ -570,7 +570,7 @@ mod tests { use phoxal::participant::{ ClockSource, ContractRole, Participant, ParticipantApi, ParticipantLaunch, TestClock, }; - use phoxal::raw::{Bus, BusConfig, OwnerCap, Publisher, Subscriber, run_with_bus}; + use phoxal::raw::{Bus, BusConfig, OwnerCap, Publisher, Subscriber, run_with_bus_clock}; use phoxal_api::ContractBody; use super::*; @@ -706,7 +706,7 @@ mod tests { let clock = TestClock::new(); let runner_clock = clock.clone(); - let runner = run_with_bus::( + let runner = run_with_bus_clock::( &bus, ParticipantLaunch::local("navigation-1", "robot"), runner_clock, diff --git a/tool/joypad/src/main.rs b/tool/joypad/src/main.rs index f33bca1..5389a5e 100644 --- a/tool/joypad/src/main.rs +++ b/tool/joypad/src/main.rs @@ -2,7 +2,7 @@ use std::collections::{HashMap, HashSet}; use gilrs::{Button, EventType, Gamepad, GamepadId, Gilrs}; use phoxal::prelude::*; -use phoxal::raw::{Publisher, Subscriber}; +use phoxal::raw::{Publisher, Subscriber, host_time}; use phoxal_api::v1 as motion_api; use phoxal_api::v2 as api; @@ -29,7 +29,6 @@ impl ToolJoypad { async fn setup(ctx: &mut SetupContext) -> Result<(Self, Self::Api)> { let cap = ctx.owner_capability(); let bus = ctx.raw_bus(); - let clock = ctx.clock(); let manual_publisher = Publisher::new(bus.clone(), &motion_api::topic::new().motion().manual())?; @@ -48,7 +47,6 @@ impl ToolJoypad { devices_publisher, connect_subscriber, rescan_subscriber, - clock, ) .await }); @@ -93,7 +91,6 @@ async fn run_joypad( devices_publisher: Publisher, connect_subscriber: Subscriber, rescan_subscriber: Subscriber, - clock: std::sync::Arc, ) { let mut gilrs = match Gilrs::new() { Ok(gilrs) => Some(gilrs), @@ -109,29 +106,35 @@ async fn run_joypad( } else { registry.last_error = Some("gamepad backend unavailable".to_string()); } - publish_devices(&devices_publisher, ®istry, clock.now()).await; + publish_devices(&devices_publisher, ®istry, host_time()).await; let mut ticker = tokio::time::interval(std::time::Duration::from_secs_f64(1.0 / POLL_HZ)); - ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); loop { tokio::select! { _ = ticker.tick() => { let mut changed = false; + let mut selected_disconnected = false; if let Some(gilrs) = gilrs.as_mut() { while let Some(event) = gilrs.next_event() { gilrs.update(&event); - changed |= apply_event(gilrs, &mut registry, event.id, &event.event); + let outcome = apply_event(gilrs, &mut registry, event.id, &event.event); + changed |= outcome.changed; + selected_disconnected |= outcome.selected_disconnected; } } + if selected_disconnected { + publish_zero(&manual_publisher).await; + } if changed { - publish_devices(&devices_publisher, ®istry, clock.now()).await; + publish_devices(&devices_publisher, ®istry, host_time()).await; } let Some(gilrs) = gilrs.as_ref() else { continue }; let Some(selected_id) = selected_gilrs_id(®istry) else { continue }; let command = command_from_gamepad(&gilrs.gamepad(selected_id)); - if let Err(error) = manual_publisher.publish_at(clock.now(), command).await { + if let Err(error) = manual_publisher.publish_at(host_time(), command).await { tracing::warn!(target: "tool_joypad", error = %error, "publish failed"); } } @@ -143,7 +146,7 @@ async fn run_joypad( } else { registry.last_error = Some("gamepad backend unavailable".to_string()); } - publish_devices(&devices_publisher, ®istry, clock.now()).await; + publish_devices(&devices_publisher, ®istry, host_time()).await; } Err(error) => { tracing::warn!(target: "tool_joypad", error = %error, "connect subscription failed"); @@ -155,11 +158,13 @@ async fn run_joypad( match received { Ok(_received) => { if let Some(gilrs) = gilrs.as_ref() { - rescan(gilrs, &mut registry); + if rescan(gilrs, &mut registry) { + publish_zero(&manual_publisher).await; + } } else { registry.last_error = Some("gamepad backend unavailable".to_string()); } - publish_devices(&devices_publisher, ®istry, clock.now()).await; + publish_devices(&devices_publisher, ®istry, host_time()).await; } Err(error) => { tracing::warn!(target: "tool_joypad", error = %error, "rescan subscription failed"); @@ -171,6 +176,16 @@ async fn run_joypad( } } +async fn publish_zero(publisher: &Publisher) { + let zero = motion_api::motion::ManualCommand { + linear_x_mps: 0.0, + angular_z_radps: 0.0, + }; + if let Err(error) = publisher.publish_at(host_time(), zero).await { + tracing::warn!(target: "tool_joypad", error = %error, "disconnect zero publish failed"); + } +} + fn has_compatible_mapping(gamepad: &Gamepad<'_>) -> bool { mapping_supports_manual_controls([ gamepad.button_code(Button::LeftTrigger).is_some(), @@ -221,7 +236,18 @@ fn devices_snapshot(registry: &Registry) -> api::joypad::Devices { /// Handle a decoded gilrs hotplug event, mutating `registry` in place. /// Returns `true` if the device set or selection changed (Devices should be /// republished). -fn apply_event(gilrs: &Gilrs, registry: &mut Registry, id: GamepadId, event: &EventType) -> bool { +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +struct RegistryChange { + changed: bool, + selected_disconnected: bool, +} + +fn apply_event( + gilrs: &Gilrs, + registry: &mut Registry, + id: GamepadId, + event: &EventType, +) -> RegistryChange { match event { EventType::Connected => { let stable_id = observe(gilrs, registry, id); @@ -233,10 +259,13 @@ fn apply_event(gilrs: &Gilrs, registry: &mut Registry, id: GamepadId, event: &Ev registry.selected = Some(stable_id); registry.last_error = None; } - true + RegistryChange { + changed: true, + selected_disconnected: false, + } } EventType::Disconnected => on_disconnected(registry, id), - _ => false, + _ => RegistryChange::default(), } } @@ -268,7 +297,8 @@ fn handle_connect(registry: &mut Registry, id: &str) { /// previously known set so a still-connected pad keeps its stable id, a /// newly seen pad is assigned one, and a pad no longer present is marked /// disconnected (not removed - it may reconnect later). -fn rescan(gilrs: &Gilrs, registry: &mut Registry) { +fn rescan(gilrs: &Gilrs, registry: &mut Registry) -> bool { + let selected_before = registry.selected.clone(); let mut seen: HashSet = HashSet::new(); for (id, _gamepad) in gilrs.gamepads() { seen.insert(id); @@ -283,6 +313,12 @@ fn rescan(gilrs: &Gilrs, registry: &mut Registry) { } } reconcile_selection(gilrs, registry); + selected_before.is_some_and(|selected| { + registry + .entries + .get(&selected) + .is_some_and(|entry| !entry.connected) + }) } /// If the current selection is no longer connected, clear it (reporting why) @@ -380,24 +416,32 @@ fn observe(gilrs: &Gilrs, registry: &mut Registry, id: GamepadId) -> String { stable_id } -fn on_disconnected(registry: &mut Registry, id: GamepadId) -> bool { - let mut disconnected: Option = None; - for (stable_id, entry) in registry.entries.iter_mut() { - if entry.gilrs_id == Some(id) { - entry.gilrs_id = None; - entry.connected = false; - disconnected = Some(stable_id.clone()); - break; - } - } - let Some(stable_id) = disconnected else { - return false; +fn on_disconnected(registry: &mut Registry, id: GamepadId) -> RegistryChange { + let stable_id = registry + .entries + .iter() + .find_map(|(stable_id, entry)| (entry.gilrs_id == Some(id)).then(|| stable_id.clone())); + let Some(stable_id) = stable_id else { + return RegistryChange::default(); + }; + disconnect_stable_id(registry, &stable_id) +} + +fn disconnect_stable_id(registry: &mut Registry, stable_id: &str) -> RegistryChange { + let Some(entry) = registry.entries.get_mut(stable_id) else { + return RegistryChange::default(); }; - if registry.selected.as_deref() == Some(stable_id.as_str()) { + entry.gilrs_id = None; + entry.connected = false; + let selected_disconnected = registry.selected.as_deref() == Some(stable_id); + if selected_disconnected { registry.selected = None; registry.last_error = Some(format!("selected device '{stable_id}' disconnected")); } - true + RegistryChange { + changed: true, + selected_disconnected, + } } /// Derive a STABLE wire id for a pad from its identity - NOT the @@ -626,6 +670,67 @@ mod tests { ); } + #[test] + fn selected_device_disconnect_requests_an_immediate_zero() { + let mut registry = Registry { + selected: Some("pad".to_string()), + ..Registry::default() + }; + registry.entries.insert( + "pad".to_string(), + PadEntry { + base_id: "pad".to_string(), + gilrs_id: None, + name: "Pad".to_string(), + connected: true, + mapped: true, + }, + ); + + let outcome = disconnect_stable_id(&mut registry, "pad"); + + assert_eq!( + outcome, + RegistryChange { + changed: true, + selected_disconnected: true, + } + ); + assert!(registry.selected.is_none()); + assert!(!registry.entries["pad"].connected); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn disconnect_zero_is_published_on_the_manual_contract() { + let bus = phoxal::raw::Bus::open(phoxal::raw::BusConfig::in_process( + format!("test/joypad-disconnect/{}", std::process::id()), + "robot", + )) + .await + .expect("bus should open"); + let publisher = Publisher::new(bus.clone(), &motion_api::topic::new().motion().manual()) + .expect("manual publisher should attach"); + let subscriber = Subscriber::new( + &bus, + &motion_api::topic::internal::new(phoxal::raw::OwnerCap::__mint()) + .motion() + .manual(), + 1, + ) + .await + .expect("manual subscriber should attach"); + + publish_zero(&publisher).await; + let received = tokio::time::timeout(std::time::Duration::from_secs(2), subscriber.recv()) + .await + .expect("zero command should arrive") + .expect("zero command should decode"); + + assert_eq!(received.body.linear_x_mps, 0.0); + assert_eq!(received.body.angular_z_radps, 0.0); + bus.close().await.expect("bus should close"); + } + #[test] fn all_manual_controls_are_required_for_compatibility() { assert!(!mapping_supports_manual_controls([true, true, true, false])); diff --git a/tool/router/src/main.rs b/tool/router/src/main.rs index 452567a..c7e2f35 100644 --- a/tool/router/src/main.rs +++ b/tool/router/src/main.rs @@ -3,7 +3,7 @@ use std::sync::{Arc, Mutex}; use std::time::Duration; use phoxal::prelude::*; -use phoxal::raw::{Bus, BusMetadata, LogicalTime, OwnerCap, Publisher}; +use phoxal::raw::{Bus, BusMetadata, OwnerCap, Publisher, host_time}; use phoxal_api::v1 as api; use phoxal_api::v2 as preview_api; @@ -240,7 +240,7 @@ async fn spawn_metrics(ctx: &mut SetupContext, router: &zenoh::Sessi ManagedTaskPolicy::FaultOnExit, async move { let mut ticker = tokio::time::interval(METRICS_WINDOW); - ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); // Tokio intervals yield their first tick immediately. Consume that // bootstrap tick so the first published snapshot covers a real // window instead of reporting a near-zero interval as one second. @@ -252,7 +252,7 @@ async fn spawn_metrics(ctx: &mut SetupContext, router: &zenoh::Sessi let samples = drain_window(&publish_counters); let metrics = build_metrics(&samples, window_ended.duration_since(window_started)); window_started = window_ended; - if let Err(error) = metrics_publisher.publish_at(now(), metrics).await { + if let Err(error) = metrics_publisher.publish_at(host_time(), metrics).await { tracing::warn!(target: "tool_router", error = %error, "router metrics publish failed"); } } @@ -460,17 +460,10 @@ async fn publish_uplink_state( publisher: &Publisher, state: api::bus::uplink::State, ) -> Result<()> { - publisher.publish_at(now(), state).await?; + publisher.publish_at(host_time(), state).await?; Ok(()) } -fn now() -> LogicalTime { - let elapsed = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default(); - LogicalTime::new(0, u64::try_from(elapsed.as_nanos()).unwrap_or(u64::MAX)) -} - fn main() -> phoxal::Result<()> { phoxal::run::() } diff --git a/tool/telemetry/src/main.rs b/tool/telemetry/src/main.rs index 7de3f93..d85dc81 100644 --- a/tool/telemetry/src/main.rs +++ b/tool/telemetry/src/main.rs @@ -1,7 +1,7 @@ use std::time::Duration; use phoxal::prelude::*; -use phoxal::raw::{Bus, LogicalTime, OwnerCap, Publisher}; +use phoxal::raw::{Bus, OwnerCap, Publisher, host_time}; use phoxal_api::v2 as api; use sysinfo::{CpuRefreshKind, MemoryRefreshKind, RefreshKind, System}; @@ -51,6 +51,7 @@ async fn sample_host_forever(publisher: Publisher) { .with_memory(MemoryRefreshKind::nothing().with_ram()), ); let mut interval = tokio::time::interval(SAMPLE_INTERVAL); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); // `sysinfo`'s CPU usage is a delta against the PREVIOUS refresh, so the // very first refresh has nothing to diff against and reads inaccurately @@ -64,7 +65,7 @@ async fn sample_host_forever(publisher: Publisher) { loop { interval.tick().await; let sample = sample_host(&mut system); - if let Err(error) = publisher.publish_at(now(), sample).await { + if let Err(error) = publisher.publish_at(host_time(), sample).await { tracing::warn!(target: "tool_telemetry", error = %error, "host telemetry publish failed"); } } @@ -86,13 +87,6 @@ fn sample_host(system: &mut System) -> api::telemetry::Host { } } -fn now() -> LogicalTime { - let elapsed = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default(); - LogicalTime::new(0, u64::try_from(elapsed.as_nanos()).unwrap_or(u64::MAX)) -} - fn main() -> phoxal::Result<()> { phoxal::run::() } diff --git a/xtask/src/workspace.rs b/xtask/src/workspace.rs index 11c011d..0188d68 100644 --- a/xtask/src/workspace.rs +++ b/xtask/src/workspace.rs @@ -567,6 +567,8 @@ pub fn require_nonempty_artifacts(artifacts: &[OfficialArtifact]) -> Result<()> #[cfg(test)] mod tests { + use std::fs; + use serde_json::json; use super::*; @@ -579,6 +581,70 @@ mod tests { classify_manifest_path(&root(), &root().join(relative)) } + fn visit_rust_sources(directory: &Path, sources: &mut Vec) -> Result<()> { + for entry in fs::read_dir(directory) + .with_context(|| format!("failed to read {}", directory.display()))? + { + let path = entry?.path(); + if path.is_dir() { + visit_rust_sources(&path, sources)?; + } else if path.extension().is_some_and(|extension| extension == "rs") { + sources.push(path); + } + } + Ok(()) + } + + #[test] + fn official_tools_do_not_import_logical_clock_surfaces() -> Result<()> { + let workspace_root = Path::new(env!("CARGO_MANIFEST_DIR")) + .parent() + .context("xtask manifest directory has no workspace parent")?; + let mut sources = Vec::new(); + visit_rust_sources(&workspace_root.join("tool"), &mut sources)?; + let forbidden = [ + "ClockMode", + "ClockSource", + "SimulationClock", + "StepContext", + "simulation().clock()", + ".clock()", + ]; + + let mut violations = Vec::new(); + for source in sources { + let body = fs::read_to_string(&source)?; + for token in forbidden { + if body.contains(token) { + violations.push(format!("{} imports or calls {token}", source.display())); + } + } + } + + assert!( + violations.is_empty(), + "tools must remain host/event driven:\n{}", + violations.join("\n") + ); + Ok(()) + } + + #[test] + fn official_periodic_tools_skip_missed_ticks() -> Result<()> { + let workspace_root = Path::new(env!("CARGO_MANIFEST_DIR")) + .parent() + .context("xtask manifest directory has no workspace parent")?; + for tool in ["joypad", "router", "telemetry"] { + let source = workspace_root.join("tool").join(tool).join("src/main.rs"); + let body = fs::read_to_string(&source)?; + assert!( + body.contains("set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip)"), + "periodic tool {tool} must skip missed ticks instead of catching up" + ); + } + Ok(()) + } + #[test] fn real_workspace_release_scope_is_valid() -> Result<()> { let workspace_manifest = Path::new(env!("CARGO_MANIFEST_DIR"))