feat(api): add node.timestamp() accessor for the node's HLC (rescue of #1789)

Rescues @oceanusxiv's #1789 against current main. The PR sat for 18 days
with no reviews and trivially-clean cherry-pick against `main`, so this
is a near-verbatim re-application plus expanded docstrings and a
regression test.

Why
===

`DoraNode` already maintains an internal `uhlc::HLC` clock that stamps
every outgoing message, but user code had no way to read that clock
directly. Without this method, a node that wants to measure per-event
processing latency has to fall back to `std::time::SystemTime::now()` —
a different physical clock from the HLC, so subtracting that from
`event.metadata.timestamp` is meaningless across daemons.

Three concrete use cases motivated #1789:

  1. `latency = node.timestamp() - event.metadata.timestamp` measured
     against the same clock dora's data plane uses.
  2. Correlating log lines / external observations with the HLC.
  3. Profiling node processing time without introducing a second
     clock source.

What's added
============

* `apis/rust/node/src/node/mod.rs` — `DoraNode::timestamp() ->
  uhlc::Timestamp`. Single-line body: `self.clock.new_timestamp()`.
  Docstring spells out the "use this against `event.metadata.timestamp`,
  NOT against `SystemTime::now()`" rule.

* `apis/python/node/src/lib.rs` — `Node.timestamp() -> datetime.datetime`.
  Converts HLC physical time to a UTC-aware `datetime`. The Python
  docstring explicitly notes the resolution caveats (microsecond
  precision, HLC logical counter dropped) and points strict-ordering
  callers at the Rust API.

* `apis/python/operator/src/lib.rs` — promote `datetime_module()` from
  private to `pub` so the node crate can reuse the cached
  `PyModule::import("datetime")` instead of re-importing on every
  call. Same cache pattern dora already uses for its own outgoing
  metadata conversions in the operator crate.

* `apis/python/node/dora/__init__.pyi` — type stub for `timestamp()`.

What was added on top of #1789
==============================

* Expanded both docstrings to call out the load-bearing "use the HLC,
  not SystemTime" guidance and the Python resolution caveats.
* New regression test
  `node::tests::timestamp_uses_node_clock_and_is_monotonic` that
  asserts two `timestamp()` calls share an HLC ID (i.e., they read
  the *node's* clock, not a freshly-created one) AND are strictly
  monotonic. Without this test, a refactor that gives `timestamp()`
  its own clock would silently break the latency-measurement use case
  in the docstring.

Verification
============

  cargo fmt --all -- --check
  cargo clippy --all --exclude dora-{node-api,operator-api,ros2-bridge}-python -- -D warnings
  cargo test -p dora-node-api --lib
    (83/83 pass; the new test runs in 0.11s)
  cargo check --examples

Co-authored-by: oceanusxiv <oceanusxiv@users.noreply.github.com>
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: 3 people

apis/python/node/dora/__init__.pyi

@@ -1,3 +1,4 @@
+import datetime
 import typing
 
 import pyarrow
@@ -149,6 +150,10 @@ class Node:
 
         Returns 0 on the first run, 1 after the first restart, etc."""
 
+    def timestamp(self) -> "datetime.datetime":
+        """Returns the current timestamp from the node's Hybrid Logical Clock
+        as a UTC datetime object."""
+
     def merge_external_events(self, subscription: dora.Ros2Subscription) -> None:
         """Merge an external event stream with dora main loop.
         This currently only work with ROS2."""

apis/python/node/src/lib.rs

@@ -8,7 +8,9 @@ use arrow::pyarrow::{FromPyArrow, ToPyArrow};
 use dora_node_api::dora_core::config::{DataId, NodeId};
 use dora_node_api::merged::{MergeExternalSend, MergedEvent};
 use dora_node_api::{DataflowId, DoraNode, EventStream, TryRecvError, init_tracing};
-use dora_operator_api_python::{DelayedCleanup, NodeCleanupHandle, PyEvent, pydict_to_metadata};
+use dora_operator_api_python::{
+    DelayedCleanup, NodeCleanupHandle, PyEvent, datetime_module, pydict_to_metadata,
+};
 use dora_ros2_bridge_python::Ros2Subscription;
 use eyre::{Context, ContextCompat};
 
@@ -688,6 +690,45 @@ impl Node {
         self.node.get_mut().restart_count()
     }
 
+    /// Returns the current timestamp from the node's Hybrid Logical Clock
+    /// as a UTC datetime object.
+    ///
+    /// Use this against ``event["metadata"]["timestamp"]`` from an INPUT
+    /// event to compute per-event processing latency against the same
+    /// clock dora uses on its data plane.
+    ///
+    /// **Resolution & ordering caveats** — Python ``datetime`` is
+    /// microsecond-resolution and only carries the HLC's physical
+    /// component; its logical counter is dropped. Two messages produced
+    /// in the same HLC microsecond will compare equal here even though
+    /// the underlying HLC distinguishes them. For strict ordering use
+    /// the Rust API (``DoraNode::timestamp() -> uhlc::Timestamp``).
+    ///
+    /// **Threading** — like every other ``Node`` method, this acquires
+    /// the node's internal mutex via ``try_lock`` and will panic if
+    /// another thread is mid-call into the same ``Node`` instance. The
+    /// Python GIL serialises in-process calls, so this is only
+    /// reachable if a caller releases the GIL (e.g. via
+    /// ``allow_threads``) and another thread re-enters the same Node.
+    /// Don't do that.
+    ///
+    /// :rtype: datetime.datetime
+    pub fn timestamp(&self, py: Python) -> eyre::Result<Py<PyAny>> {
+        let ts = self.node.get_mut().timestamp();
+        let system_time = ts.get_time().to_system_time();
+        let duration_since_epoch = system_time
+            .duration_since(std::time::UNIX_EPOCH)
+            .context("Failed to calculate duration since epoch")?;
+        let total_seconds = duration_since_epoch.as_secs() as f64
+            + duration_since_epoch.subsec_micros() as f64 / 1_000_000.0;
+        let dt_module = datetime_module(py).context("Failed to import datetime module")?;
+        let datetime_class = dt_module.getattr("datetime")?;
+        let utc_timezone = dt_module.getattr("timezone")?.getattr("utc")?;
+        let py_datetime =
+            datetime_class.call_method1("fromtimestamp", (total_seconds, utc_timezone))?;
+        Ok(py_datetime.unbind())
+    }
+
     /// Merge an external event stream with dora main loop.
     /// This currently only work with ROS2.
     ///

apis/python/operator/src/lib.rs

@@ -22,7 +22,7 @@ use std::time::UNIX_EPOCH;
 /// Cached Python `datetime` module to avoid repeated `PyModule::import` on the hot path.
 static DATETIME_MODULE: PyOnceLock<Py<PyModule>> = PyOnceLock::new();
 
-fn datetime_module<'py>(py: Python<'py>) -> PyResult<&'py Bound<'py, PyModule>> {
+pub fn datetime_module<'py>(py: Python<'py>) -> PyResult<&'py Bound<'py, PyModule>> {
     Ok(DATETIME_MODULE
         .get_or_try_init(py, || PyModule::import(py, "datetime").map(|m| m.unbind()))?
         .bind(py))

apis/rust/node/src/node/mod.rs

@@ -1149,6 +1149,20 @@ impl DoraNode {
         self.restart_count
     }
 
+    /// Returns the current timestamp from the node's Hybrid Logical Clock.
+    ///
+    /// This generates a new HLC timestamp, which combines the physical
+    /// wall-clock time with a logical counter to ensure uniqueness and
+    /// monotonicity even across nodes. The HLC is the same clock dora
+    /// stamps every outgoing message with, so this is the right value
+    /// to subtract from an input event's `metadata.timestamp` when
+    /// measuring per-event processing latency — using
+    /// `std::time::SystemTime::now()` instead would mix two unrelated
+    /// clocks and give meaningless results across daemons.
+    pub fn timestamp(&self) -> uhlc::Timestamp {
+        self.clock.new_timestamp()
+    }
+
     /// Send a structured log message.
     ///
     /// Outputs a JSONL line to stdout that the daemon parses automatically.
@@ -1691,6 +1705,38 @@ mod tests {
         uuid::Uuid::parse_str(&id).expect("should be valid UUID");
     }
 
+    /// `DoraNode::timestamp()` must read from the SAME HLC the node
+    /// uses to stamp outgoing messages. If a refactor accidentally
+    /// gives `timestamp()` its own clock, the latency-measurement use
+    /// case in the docstring silently breaks (subtracting against an
+    /// `event.metadata.timestamp` from the data plane would mix two
+    /// unrelated HLCs). Guard by asserting two calls share an HLC ID
+    /// and that the second reads strictly later than the first.
+    ///
+    /// The strict `t2 > t1` assertion holds by HLC construction: if
+    /// the wall clock advanced between calls, the physical component
+    /// strictly increases; if not, the logical counter bumps. The
+    /// lexicographic ordering on `uhlc::Timestamp` puts `t2` strictly
+    /// after `t1` in either case, so this assertion does not flake on
+    /// fast machines whose OS clock rounds both calls to the same tick.
+    #[test]
+    fn timestamp_uses_node_clock_and_is_monotonic() {
+        let (node, events, _rx) = test_node();
+        let t1 = node.timestamp();
+        let t2 = node.timestamp();
+        assert_eq!(
+            t1.get_id(),
+            t2.get_id(),
+            "two timestamp() calls must come from the same HLC instance",
+        );
+        assert!(
+            t2 > t1,
+            "HLC timestamps must be strictly monotonic: {t1:?} >= {t2:?}"
+        );
+        drop(node);
+        drop(events);
+    }
+
     /// Helper: create a minimal test node with a channel output.
     fn test_node() -> (
         DoraNode,