feat(timestamp-stack): add routing timestamp instrumentation

[YuanYuYuan]

Summary

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

Entirely feature = "unstable"-gated. Zero overhead on uninstrumented messages.

Sister PRs

• Rust core: feat(timestamp-stack): add routing timestamp instrumentation ZettaScaleLabs/zenoh#30
• zenoh-c: feat(timestamp-stack): add routing timestamp instrumentation zenoh-c#1

---

API Design

All new types are importable directly from zenoh:

from zenoh import InterceptionPoint, TimestampInstrumentation

Configuring instrumentation

# Keyword-only constructor — at least one flag must be True
instr = TimestampInstrumentation(send=True, receive=True)
instr = TimestampInstrumentation(send=True, route=True, receive=True)

Returns a TimestampInstrumentation object. Raises ZError if all flags are False.

Session-level custom clock

Registered once at zenoh.open time; applied to every instrumented message on that session:

def my_clock(ctx: TsStackContext) -> bytes:
    # ctx.zid        → ZenohId
    # ctx.whatami    → WhatAmI
    # ctx.interception_point → InterceptionPoint
    return struct.pack("<Q", time.time_ns())

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

Returning b"" (empty bytes) skips stamping at that point. When no callback is registered, Zenoh uses a lazily-initialized UHLC clock.

Attaching instrumentation

timestamp_instrumentation= is accepted as a keyword argument on all write operations:

# One-shot put / delete / get
session.put("demo/key", b"payload",  timestamp_instrumentation=instr)
session.delete("demo/key",           timestamp_instrumentation=instr)
session.get("demo/**",               timestamp_instrumentation=instr)

# Querier
querier = session.declare_querier("demo/**")
querier.get(timestamp_instrumentation=instr)

# Reply from a queryable
query.reply("demo/key", b"payload", timestamp_instrumentation=instr)
query.reply_err(b"error payload",   timestamp_instrumentation=instr)

# Publisher — per-put and per-delete overrides
pub = session.declare_publisher("demo/key")
pub.put(b"message-1",   timestamp_instrumentation=instr)        # per-put
pub.put(b"message-2",   timestamp_instrumentation=send_only)    # override
pub.delete(             timestamp_instrumentation=instr)         # per-delete

# AdvancedPublisher (zenoh-ext)
adv_pub.put(b"payload", timestamp_instrumentation=instr)
adv_pub.delete(         timestamp_instrumentation=instr)

Note: timestamp_instrumentation is not available on session.declare_publisher. The Rust PublisherBuilder has no setter for it — instrumentation is always specified per operation.

Pass None to suppress instrumentation on a specific call even when a session default would otherwise apply.

Reading timestamps

sample.timestamp_stack is None when the message was not instrumented:

stack = sample.timestamp_stack      # TimestampStack | None
if stack:
    for rec in stack.records:       # List[TimestampStackRecord]
        print(rec.point.name)       # "SEND", "ROUTE", "RECEIVE", or "UNKNOWN"
        print(rec.is_custom)        # bool — True when a custom callback produced the bytes
        ts = rec.as_timestamp()     # Timestamp | None (None for custom-format records)
        raw = rec.timestamp()       # bytes — always available regardless of format

Records are in wire order: Send first, Receive last, one or more Route records in between (one per routing hop).

The same .timestamp_stack property is available on ReplyError, Reply, and Query:

reply_error.timestamp_stack    # TimestampStack | None
reply.timestamp_stack          # TimestampStack | None (from the reply's sample)
query.timestamp_stack          # TimestampStack | None (from the incoming get)

InterceptionPoint

InterceptionPoint.SEND      # stamped by the putting session
InterceptionPoint.ROUTE     # stamped by each routing hop
InterceptionPoint.RECEIVE   # stamped on delivery to the subscriber
InterceptionPoint.UNKNOWN   # catch-all for variants added in future Rust releases

UNKNOWN exists for forward compatibility — the Rust enum is #[non_exhaustive].

---

Changes

New module: src/timestamp_stack.rs

• InterceptionPoint — Send / Route / Receive / Unknown with forward-compatibility catch-all; .point and .is_custom exposed as #[getter] properties
• TsStackContext — context passed to a custom clock callback (zid, whatami, interception_point)
• TimestampInstrumentation — keyword-only constructor, is_instrumented(point) query
• TimestampStackRecord — .point, .is_custom, .timestamp(), .as_timestamp()
• TimestampStack — .instrumentation, .records
• py_to_session_ts_callback — bridges a Python callable to Arc<dyn Fn(TsStackContext) -> Vec<u8> + Send + Sync>
• Query.timestamp_stack getter

src/session.rs updated

• Session.put() — timestamp_instrumentation= kwarg
• Session.delete() — timestamp_instrumentation= kwarg
• Session.get() — timestamp_instrumentation= kwarg
• Session.open() — timestamp_callback= kwarg

src/query.rs updated

• Query.reply() — timestamp_instrumentation= kwarg
• Query.reply_err() — timestamp_instrumentation= kwarg

src/pubsub.rs updated

• Publisher.put() — timestamp_instrumentation= kwarg (per-put override)
• Publisher.delete() — timestamp_instrumentation= kwarg (per-delete override)

src/ext.rs updated

• AdvancedPublisher.put() — timestamp_instrumentation= kwarg
• AdvancedPublisher.delete() — timestamp_instrumentation= kwarg

zenoh/__init__.pyi and zenoh/ext.pyi stubs updated

Full class bodies for InterceptionPoint, TsStackContext, TimestampInstrumentationBuilder, TimestampInstrumentation, TimestampStackRecord, TimestampStack, and SessionTimestampCallback. All timestamp_instrumentation= kwargs typed across every write operation. timestamp_stack property typed on Sample, ReplyError, Reply, and Query.

Tests and examples

• tests/test_timestamp_stack.py — 14 integration tests covering put/subscribe, publisher per-put/delete override, session delete, querier get, query/reply paths, custom callback, ReplyError propagation, and Query.timestamp_stack
• examples/z_timestamp_instrumentation.py — three scenarios: put/subscribe, publisher with per-put override, and session-level custom callback

---

Breaking Changes

None. All new APIs are behind unstable feature.

[github-actions]

PR missing one of the required labels: {'bug', 'ci', 'new feature', 'dependencies', 'api-sync', 'breaking-change', 'documentation', 'enhancement', 'internal'}

[github-actions]

PR missing one of the required labels: {'bug', 'documentation', 'ci', 'api-sync', 'dependencies', 'internal', 'breaking-change', 'enhancement', 'new feature'}

[github-actions]

PR missing one of the required labels: {'ci', 'api-sync', 'breaking-change', 'bug', 'documentation', 'enhancement', 'dependencies', 'internal', 'new feature'}