feat(timestamp-stack): add routing timestamp instrumentation

[YuanYuYuan]

Summary

Adds opt-in timestamp instrumentation for measuring end-to-end message latency in Zenoh. Messages can carry a TsStack wire extension that accumulates Interception records at up to three points along a message's path: Send, Route, and Receive.

The feature is entirely #[cfg(feature = "unstable")]-gated. Uninstrumented messages carry zero overhead — push_ts_interception is a no-op when ext_ts_stack is None.

Sister PRs

Binding	Branch	Status
Python	YuanYuYuan/zenoh-python#1	CI passing
C	YuanYuYuan/zenoh-c#1	CI passing

---

API Design

Rust (zenoh crate, feature = "unstable")

All public types live in zenoh::timestamp_stack.

Configuring instrumentation — choose which points to record per-message:

let instr = TimestampInstrumentationBuilder::new()
    .set_send(true)
    .set_receive(true)
    .build()?;

Session-level custom clock — registered once at open time, applied to every instrumented message:

let cb: SessionTimestampCallback = Arc::new(|ctx: TsStackContext| {
    // ctx.zid, ctx.whatami, ctx.interception_point are available
    my_clock_bytes()
});
let session = zenoh::open(Config::default())
    .with_timestamp_callback(cb)
    .await?;

SessionTimestampCallback is Arc<dyn Fn(TsStackContext) -> Vec<u8> + Send + Sync>. Returning an empty Vec skips stamping that point. When no callback is registered, Zenoh uses a lazily-initialized UHLC.

Attaching instrumentation — timestamp_instrumentation(Option<TimestampInstrumentation>) is available on:

session.put(key, payload).timestamp_instrumentation(Some(instr)).await?;
session.get(selector).timestamp_instrumentation(Some(instr)).await?;
queryable_builder.reply(key, payload).timestamp_instrumentation(Some(instr)).await?;
session.declare_publisher(key).timestamp_instrumentation(Some(instr)).await?;
publisher.put(payload).timestamp_instrumentation(Some(instr)).await?; // per-put override

Reading timestamps off a received Sample, ReplyError, or Query:

if let Some(stack) = sample.timestamp_stack() {
    for rec in stack.records() {
        match rec.timestamp() {
            InstrumentationTimestamp::UHLC(ts) => { /* standard uhlc::Timestamp */ }
            InstrumentationTimestamp::Custom(bytes) => { /* custom-callback bytes */ }
        }
    }
}

TimestampStack::records() returns records in wire order (Send → Route(s) → Receive). rec.is_custom() is a convenience predicate over the enum variant.

AdvancedPublisher (zenoh-ext):

let publisher = session
    .declare_advanced_publisher(key)
    .timestamp_instrumentation(Some(instr))      // default for all puts
    .await?;
publisher.put(payload)
    .timestamp_instrumentation(Some(send_only))  // per-put override
    .await?;

---

Python (zenoh-python)

All types are importable directly from zenoh:

from zenoh import InterceptionPoint, TimestampInstrumentation

Configuring instrumentation:

instr = TimestampInstrumentation(send=True, receive=True)
instr = TimestampInstrumentation(send=True, route=True, receive=True)

Session-level custom clock:

def my_clock(ctx):           # ctx.zid, ctx.whatami, ctx.interception_point
    return struct.pack("<Q", time.time_ns())

with zenoh.open(zenoh.Config(), timestamp_callback=my_clock) as session:
    ...

Attaching instrumentation — timestamp_instrumentation= kwarg on:

session.put(key, payload, timestamp_instrumentation=instr)
session.get(selector, timestamp_instrumentation=instr)
query.reply(key, payload, timestamp_instrumentation=instr)
query.reply_err(err_payload, timestamp_instrumentation=instr)
session.declare_publisher(key, timestamp_instrumentation=instr)  # publisher default
pub.put(payload, timestamp_instrumentation=send_only)            # per-put override

Reading timestamps:

stack = sample.timestamp_stack      # None if not instrumented
if stack:
    for rec in stack.records:
        ts = rec.as_timestamp()     # Timestamp | None  (None for custom-callback records)
        raw = rec.timestamp()       # bytes | None  (bytes for custom-callback records only)

InterceptionPoint is a pyclass enum with values SEND, ROUTE, RECEIVE, UNKNOWN.

---

C (zenoh-c, ZENOHC_BUILD_WITH_UNSTABLE_API=ON)

Types:

C type	Purpose
z_owned_timestamp_instrumentation_t / z_loaned_timestamp_instrumentation_t	instrumentation config
z_interception_point_t	enum: Z_INTERCEPTION_POINT_SEND, Z_INTERCEPTION_POINT_ROUTE, Z_INTERCEPTION_POINT_RECEIVE, Z_INTERCEPTION_POINT_UNKNOWN
z_loaned_timestamp_stack_t	stack carried by a received message
z_loaned_timestamp_stack_record_t	a single interception record
z_owned_session_ts_callback_t / z_moved_session_ts_callback_t	session-level custom clock

Configuring instrumentation:

z_owned_timestamp_instrumentation_t instr;
z_timestamp_instrumentation_new(&instr, /*send=*/true, /*route=*/false, /*receive=*/true);
// ...use...
z_timestamp_instrumentation_drop(z_timestamp_instrumentation_move(&instr));

Session-level custom clock:

z_owned_session_ts_callback_t cb;
z_closure_session_ts_callback(&cb, my_callback_fn, my_drop_fn, my_context);

z_open_options_t opts = z_open_options_default();
opts.timestamp_callback = z_session_ts_callback_move(&cb);
z_open(&session, z_config_move(&config), &opts);

Attaching instrumentation — timestamp_instrumentation field on the relevant _options_t struct:

z_put_options_t opts = z_put_options_default();
opts.timestamp_instrumentation = z_timestamp_instrumentation_loan(&instr);
z_put(z_session_loan(&session), z_view_keyexpr_loan(&ke), z_bytes_move(&payload), &opts);

Reading timestamps:

const z_loaned_timestamp_stack_t* stack = z_sample_timestamp_stack(sample);
if (stack) {
    size_t n = z_timestamp_stack_record_count(stack);
    for (size_t i = 0; i < n; i++) {
        const z_loaned_timestamp_stack_record_t* rec = z_timestamp_stack_record_at(stack, i);
        bool custom = z_timestamp_stack_record_is_custom(rec);
        if (custom) {
            size_t len;
            const uint8_t* bytes = z_timestamp_stack_record_timestamp(rec, &len);
        } else {
            z_timestamp_t ts;
            z_timestamp_stack_record_as_timestamp(rec, &ts);
        }
    }
}

---

Wire Protocol

A new TsStack extension (ID 0x7) is added to Push, Request, and Response messages. It carries:

• conf_flags — bitmask of enabled interception points (set by the sender, unchanged in transit)
• stack — ordered Vec<Interception>, each with a flags byte (point ID + IS_CUSTOM_TS bit) and a length-prefixed timestamp byte vector

The codec bounds timestamp bytes to u16::MAX (65 535) and caps stack depth at 64 to prevent memory exhaustion from crafted wire input.

---

Implementation Points

• Send: session.rs (resolve_put, resolve_get) and builders/reply.rs
• Route: routing/dispatcher/pubsub.rs per-subscriber (inside fan-out loop) and queries.rs
• Receive: WeakSession::send_push_consume and adminspace.rs

Route stamping happens inside the per-subscriber loop so each subscriber gets a timestamp that reflects queueing delay up to that point in fan-out.

---

Tests

• 20 integration tests in zenoh/tests/timestamp_stack.rs: no-instrumentation baseline, single-point (Send / Receive), all-points ordering, custom callback byte verification, callback context correctness, per-message stack independence, query/reply flows, is_custom flag, route-only instrumentation, ReplyError propagation, multiple-subscriber fan-out, and publisher API path
• 49 integration tests in zenoh/tests/timestamp_instrumentation.rs: full pub/sub lifecycle, multiple-session scenarios, callback interaction with instrumentation flags, and end-to-end latency measurement flows
• 4 codec tests: empty stack, known records, random round-trips, and the 64-depth limit

All 73 tests pass on Rust 1.93.0.

---

Breaking Changes

None. All new types and methods are behind feature = "unstable".

Robustness Notes

• get_ts_stack_timestamp no longer panics on UHLC serialization failure: the expect() was replaced with a tracing::warn! and graceful empty return.
• push_ts_interception uses debug_assert! + silent skip for unknown point IDs.
• Codec decode caps at 64 interceptions and u16::MAX bytes per timestamp to prevent memory exhaustion from crafted wire input.

Known Limitations

• Route timestamp is stamped once per routing hop; multi-hop topologies produce multiple ROUTE records (by design)
• QueryCleanup timeout responses carry ext_ts_stack: None — not instrumented (rationale documented in code)

---

🏷️ Label-Based Checklist

Based on the labels applied to this PR, please complete these additional requirements:

Labels: new feature

🆕 New Feature Requirements

Since this PR adds a new feature:

• Feature scope documented - Clear description of what the feature does and why it's needed
• Minimum necessary code - Implementation is as simple as possible, doesn't overcomplicate the system
• New APIs well-designed - Public APIs are intuitive, consistent with existing APIs
• Comprehensive tests - All functionality is tested (happy path + edge cases + error cases)
• Examples provided - examples/z_timestamp_instrumentation.rs added
• Documentation added - New docs explaining the feature, its use cases, and API
• Feature flag considered - Entirely behind feature = "unstable", zero overhead when inactive
• Performance impact assessed - No overhead on uninstrumented messages (push_ts_interception is a no-op when ext_ts_stack is None)
• Integration tested - 73/73 integration + codec tests pass (Rust 1.93.0)

Consider: Can this feature be split into smaller, incremental PRs?

Instructions:

1. Check off items as you complete them (change - [ ] to - [x])
2. The PR checklist CI will verify these are completed

This checklist updates automatically when labels change, but preserves your checked boxes.