Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion .github/workflows/source-gates.yml
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@ name: Source Gates
# These ran as seven separate workflows. Each paid ~26s compiling the xtask
# binary and ~13s compiling its test binary to do a few seconds of file walking,
# and all seven shared one cache key — so they raced, and six of every seven
# cache saves were discarded. One job compiles xtask once and runs all eight
# cache saves were discarded. One job compiles xtask once and runs all nine
# gates plus the whole fixture suite.
#
# `check-all-source-gates` runs every gate before reporting, so one job still
Expand Down
2 changes: 2 additions & 0 deletions Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

3 changes: 3 additions & 0 deletions adapters/streamlib-adapter-cuda/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,9 @@ serial_test = "3.2"
# The OPAQUE_FD round-trip tests drive the surface-share daemon directly
# so they exercise the same wire format helper processes consume.
serde_json.workspace = true
# Each test socket gets a directory of its own: the kernel answers uniqueness,
# and the socket is unlinked on drop instead of littering /tmp.
tempfile.workspace = true

[[test]]
name = "conformance"
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -61,21 +61,21 @@ use streamlib_surface_adapter::{
use streamlib_surface_client::{
MAX_DMA_BUF_PLANES, connect_to_surface_share_socket, send_request_with_fds,
};
use tempfile::TempDir;

const W: u32 = 32;
const H: u32 = 32;
const BPP: u32 = 4;
const SURFACE_ID: &str = "stage6-opaque-fd-round-trip";
const RUNTIME_ID: &str = "stage6-test-runtime";

fn tmp_socket_path() -> PathBuf {
let mut p = std::env::temp_dir();
let nanos = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_nanos())
.unwrap_or(0);
p.push(format!("streamlib-stage6-{nanos}.sock"));
p
/// A socket path in a directory of its own, so uniqueness is the kernel's answer.
/// The returned [`TempDir`] owns that lifetime — dropping it early removes the
/// socket out from under the running daemon.
fn tmp_socket_path() -> (TempDir, PathBuf) {
let dir = TempDir::new().expect("temp dir for surface-share socket");
let path = dir.path().join("stage6.sock");
(dir, path)
}

#[test]
Expand Down Expand Up @@ -148,7 +148,7 @@ fn opaque_fd_chain_host_export_to_consumer_import_to_adapter_acquire() {

// ── Phase 2: stand up surface-share daemon ──────────────────────────
let state = SurfaceShareState::new();
let socket_path = tmp_socket_path();
let (_socket_dir, socket_path) = tmp_socket_path();
let mut service = UnixSocketSurfaceService::new(state, socket_path.clone());
service.start().expect("surface-share service start");
std::thread::sleep(Duration::from_millis(50));
Expand Down
26 changes: 24 additions & 2 deletions docs/plan/changes/one-monotonic-clock.md
Original file line number Diff line number Diff line change
Expand Up @@ -125,21 +125,43 @@ Bare patterns — the ship gate greps each line verbatim as a fixed string.

## ADDED

- ADDED: `cargo xtask check-clock-usage` + its workflow — the mechanical guard that keeps
- ADDED: `cargo xtask check-clock-usage` ~~+ its workflow~~ — the mechanical guard that keeps
the wall-clock list from growing by accident. It bans wall-clock reads
(`SystemTime::now`, `chrono::Utc::now`, `time.time_ns`, `Date.now`) outside an explicit
(`SystemTime::now`, `chrono::Utc::now`, `time.time_ns`, ~~`Date.now`~~) outside an explicit
file allowlist holding exactly the four surfaces above, in the shape of the existing
`lint-logging` / `check-no-escalate-in-lifecycle` checks. Recon found a set of
`SystemTime` uses that mint *unique names*, not timestamps (`vulkan_graphics_kernel.rs:3119`,
`surface_share/unix_socket_service.rs:977`, several test helpers); those either move to
a counter/uuid or join the allowlist with a stated reason — final at implementation.

> Landed 2026-08-12 (#1728). **No workflow** — #1857 consolidated the per-gate workflows,
> so the guard is an entry in `ALL_SOURCE_WALKING_GATES` run by the existing `source-gates`
> job. **No `Date.now` arm** — the Deno SDK is deleted and the engine tree holds no
> `.ts`/`.js` source, so that scan root does not exist; Rust and Python only.
>
> The unique-name uses are **all converted, none allowlisted** — 16 sites across 15 files.
> The allowlist is per-file, so an entry for `iceoryx2/output.rs` or `thread_runner.rs`
> would have licensed a wall-clock read in the exact data-plane files the guard exists to
> protect. `mint_machine_global_unique_name_suffix()` (`core/machine_global_unique_name.rs`)
> is the one primitive for machine-global namespaces, built on the `uuid` the engine already
> links. Every test *socket* path takes a `TempDir` instead — `sun_path` is 108 bytes, and a
> unique name in a shared temp dir spends most of that budget before `TMPDIR` is accounted
> for. The api-server's name seed takes the OS CSPRNG, and the dead
> `apple/time.rs::system_time_to_ns` was deleted. The permitted list holds exactly the four surfaces, as five files
> (`host_ts` has two readers).
- ADDED: an epoch-parity test asserting `MediaClock::now()` and the wheel's
`monotonic_now_ns` land in the same domain as a directly-read
`clock_gettime(CLOCK_MONOTONIC)`. The natural home is the existing cross-process gate
`tests/polyglot_linux_monotonic_clock_parity.rs:112`, which today does **not** include
`MediaClock` — but its Python/Deno subprocess arms are doomed with the ripout, so the
surviving shape is host + wheel.

> Landed as the anticipated host + wheel pair, not one cross-process test: the Rust half is
> `core/media_clock.rs::now_lands_in_the_kernel_monotonic_domain` (#1725), which brackets
> `MediaClock::now()` between two direct `clock_gettime(CLOCK_MONOTONIC)` reads, and the
> Python half is `tests/test_clock_and_log.py::test_monotonic_now_ns_reads_the_kernel_monotonic_clock`.
> `polyglot_linux_monotonic_clock_parity.rs` went with the ripout.

## Notes (not tickets)

- No test anywhere asserts a frame timestamp is small or near zero, so the epoch flip
Expand Down
21 changes: 13 additions & 8 deletions runtime/streamlib-api-server/processors/api_server.rs
Original file line number Diff line number Diff line change
Expand Up @@ -89,16 +89,21 @@ const NOUNS: &[&str] = &[
fn generate_runtime_name() -> String {
use std::collections::hash_map::DefaultHasher;
use std::hash::{Hash, Hasher};
use std::time::{SystemTime, UNIX_EPOCH};

// Use time + pid for randomness without adding fastrand dependency
let seed = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_nanos() as u64
^ (std::process::id() as u64);
// The OS CSPRNG this crate already links for bearer tokens.
let mut seed_bytes = [0u8; 8];
if let Err(csprng_unavailable) = getrandom::getrandom(&mut seed_bytes) {
// A display name is not worth failing a runtime over, and the pid still
// separates concurrent runtimes on one host.
tracing::warn!("OS CSPRNG unavailable for runtime naming: {csprng_unavailable}");
seed_bytes = (std::process::id() as u64).to_ne_bytes();
}

// The two indices read disjoint halves, so the seed has to be diffused
// across all 64 bits first: a pid is well under 2^32, which would otherwise
// leave the high half zero and pin every degraded runtime to one noun.
let mut hasher = DefaultHasher::new();
seed.hash(&mut hasher);
seed_bytes.hash(&mut hasher);
let hash = hasher.finish();

let adj = ADJECTIVES[(hash as usize) % ADJECTIVES.len()];
Expand Down
10 changes: 3 additions & 7 deletions runtime/streamlib-engine/benches/output_writer_ffi_hop.rs
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,7 @@ use std::sync::Arc;
use criterion::{Criterion, black_box, criterion_group, criterion_main};
use iceoryx2::prelude::*;

use streamlib_engine::core::machine_global_unique_name::mint_machine_global_unique_name_suffix;
use streamlib_engine::iceoryx2::{
ChannelEgressConfig, ChannelTrustTier, OutputWriter, OutputWriterInner,
TRUSTED_CHANNEL_PAYLOAD_CEILING_BYTES,
Expand All @@ -52,13 +53,8 @@ use streamlib_engine::iceoryx2::{
/// don't collide on iceoryx2's machine-global `/dev/shm` namespace.
fn unique_suffix(tag: &str) -> String {
format!(
"bench/output_writer/{}/{}/{}",
tag,
std::process::id(),
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_nanos()
"bench/output_writer/{tag}/{}",
mint_machine_global_unique_name_suffix()
)
}

Expand Down
13 changes: 0 additions & 13 deletions runtime/streamlib-engine/src/apple/time.rs
Original file line number Diff line number Diff line change
Expand Up @@ -5,8 +5,6 @@
// Review if these high-precision timing functions are needed or can be removed
#![allow(dead_code)]

use std::time::{SystemTime, UNIX_EPOCH};

#[link(name = "CoreServices", kind = "framework")]
extern "C" {
fn mach_absolute_time() -> u64;
Expand Down Expand Up @@ -34,10 +32,6 @@ pub fn mach_now_ns() -> i64 {
}
}

pub fn system_time_to_ns(time: SystemTime) -> i64 {
time.duration_since(UNIX_EPOCH).unwrap().as_nanos() as i64
}

#[cfg(test)]
mod tests {
use super::*;
Expand All @@ -53,11 +47,4 @@ mod tests {
let elapsed = ns2 - ns1;
assert!((10_000_000..20_000_000).contains(&elapsed));
}

#[test]
fn test_system_time_conversion() {
let now = SystemTime::now();
let ns = system_time_to_ns(now);
assert!(ns > 0);
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -750,6 +750,7 @@ mod tests {
use super::*;
use crate::core::execution::ExecutionConfig;
use crate::core::graph::{InputLinkPortRef, OutputLinkPortRef};
use crate::core::machine_global_unique_name::mint_machine_global_unique_name_suffix;
use crate::core::processors::{DynGeneratedProcessor, PROCESSOR_REGISTRY, ProcessorSpec};
use crate::core::{ProcessorDescriptor, RuntimeContextFullAccess, RuntimeContextLimitedAccess};

Expand Down Expand Up @@ -984,13 +985,8 @@ mod tests {
/// the stale service, not as a clean failure.
fn unique_service_name(tag: &str) -> String {
format!(
"test/wiring/{}/{}/{}",
tag,
std::process::id(),
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_nanos()
"test/wiring/{tag}/{}",
mint_machine_global_unique_name_suffix()
)
}

Expand Down
10 changes: 3 additions & 7 deletions runtime/streamlib-engine/src/core/execution/thread_runner.rs
Original file line number Diff line number Diff line change
Expand Up @@ -576,18 +576,14 @@ fn dispatch_on_resume(
#[cfg(all(test, target_os = "linux"))]
mod tests {
use super::*;
use crate::core::machine_global_unique_name::mint_machine_global_unique_name_suffix;
use iceoryx2::prelude::*;
use std::os::fd::{AsRawFd, FromRawFd};

fn unique_suffix(tag: &str) -> String {
format!(
"test/runner/{}/{}/{}",
tag,
std::process::id(),
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_nanos()
"test/runner/{tag}/{}",
mint_machine_global_unique_name_suffix()
)
}

Expand Down
7 changes: 6 additions & 1 deletion runtime/streamlib-engine/src/core/logging/event.rs
Original file line number Diff line number Diff line change
Expand Up @@ -79,8 +79,13 @@ pub struct RuntimeLogEvent {
/// Schema version of this record. Bumped on breaking changes.
pub schema_version: u32,

/// Host monotonic receipt timestamp, nanoseconds since UNIX epoch.
/// Host receipt wall-clock timestamp, nanoseconds since the UNIX epoch.
/// Authoritative sort key across the merged stream.
///
/// Wall clock, not monotonic: a log record's job is correlating StreamLib
/// with the outside world and with other hosts' logs, which monotonic time
/// cannot do. Never compare or subtract this against a media timestamp —
/// they share a unit and are different quantities.
pub host_ts: u64,

/// Unique runtime identifier (from [`RuntimeUniqueId`]).
Expand Down
82 changes: 82 additions & 0 deletions runtime/streamlib-engine/src/core/machine_global_unique_name.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,82 @@
// Copyright (c) 2025 Jonathan Fontanez
// SPDX-License-Identifier: BUSL-1.1

//! Unique-name minting for machine-global namespaces.
//!
//! iceoryx2 service names live in `/dev/shm`, which is machine-global and
//! outlives the process that created an entry. A name that collides with a
//! concurrent process — or with a stale entry an earlier run left behind after
//! the pid was recycled — surfaces as `DoesNotSupportRequestedMinBufferSize`
//! against the wrong service, not as a clean failure.
//!
//! Nothing minted here is a timestamp. A v4 UUID's 122 random bits make a
//! collision between two mints vanishingly improbable — not impossible, but far
//! below the rate at which the namespace itself fails — across processes, runs
//! and reboots alike, so no clock is read and no counter is kept. The version
//! matters: `now_v7` and friends are timestamp-based, and reaching for one would
//! put a wall-clock read back into this module. The pid prefix is diagnostic —
//! it is what lets someone reading a stale `/dev/shm` entry or a leftover `/tmp`
//! file name the process that left it.
//!
//! The suffix carries no separator a path or a file name would reject, so the
//! caller composes it into whatever naming convention its namespace uses.

use uuid::Uuid;

/// A `<pid>-<uuid v4>` suffix that no concurrent process, and no earlier run,
/// will collide with short of exhausting 122 bits of randomness.
pub fn mint_machine_global_unique_name_suffix() -> String {
format!("{}-{}", std::process::id(), Uuid::new_v4())
}

#[cfg(test)]
mod tests {
use super::*;
use std::collections::HashSet;

#[test]
fn successive_mints_never_repeat() {
let minted: HashSet<String> = (0..10_000)
.map(|_| mint_machine_global_unique_name_suffix())
.collect();
assert_eq!(minted.len(), 10_000, "two mints collided");
}

/// The pid is a fixed prefix within one process, so distinctness can only
/// come from the UUID — this fails if that component is ever dropped or
/// made constant, which a whole-string equality test does not catch.
#[test]
fn the_uuid_component_is_what_differs_between_two_mints() {
let first = mint_machine_global_unique_name_suffix();
let second = mint_machine_global_unique_name_suffix();

let pid_prefix = format!("{}-", std::process::id());
let first_uuid = first.strip_prefix(&pid_prefix).expect(&first);
let second_uuid = second.strip_prefix(&pid_prefix).expect(&second);

assert_ne!(first_uuid, second_uuid);

let parsed = Uuid::parse_str(first_uuid).expect(&first);
assert_eq!(
parsed.get_version_num(),
4,
"randomness is the whole contract: the timestamp-based versions would put a \
wall-clock read back into this mint. The `uuid` dependency enabling only `v4` \
is the first guard and the compiler enforces it; this is the one that survives \
another feature being switched on for an unrelated reason"
);
}

#[test]
fn carries_no_separator_a_path_or_file_name_would_reject() {
let minted = mint_machine_global_unique_name_suffix();
assert!(
!minted.contains(['/', '\\', '.', ' ']),
"callers compose this into iceoryx2 service paths and into file names: {minted}"
);
assert!(
minted.starts_with(&format!("{}-", std::process::id())),
"a stale entry names the process that left it: {minted}"
);
}
}
1 change: 1 addition & 0 deletions runtime/streamlib-engine/src/core/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@ pub mod execution;
pub mod graph;
pub mod graph_snapshot;
pub mod json_schema;
pub mod machine_global_unique_name;
pub mod media_clock;
pub mod prelude;
pub mod processors;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@ use super::events::{
Event, EventListener, KeyCode, KeyState, Modifiers, MouseButton, MouseState, ProcessorEvent,
RuntimeEvent, topics,
};
use crate::core::machine_global_unique_name::mint_machine_global_unique_name_suffix;
use crate::iceoryx2::{Iceoryx2Node, MAX_EVENT_PAYLOAD_SIZE};
use parking_lot::Mutex;
use std::sync::Arc;
Expand Down Expand Up @@ -181,7 +182,10 @@ fn test_iceoryx2_direct_delivery() {
// Bypass PubSub layer entirely — verify iceoryx2 pub/sub works in-process
let node = Iceoryx2Node::new().expect("Failed to create iceoryx2 node");

let service_name = format!("streamlib/diag-{}/events/test", uuid::Uuid::new_v4());
let service_name = format!(
"streamlib/diag-{}/events/test",
mint_machine_global_unique_name_suffix()
);

// Create subscriber FIRST (must exist before publisher sends)
let sub_service = node
Expand Down
Loading
Loading