Ensure generated timestamps are session-monotonic

CMakeLists.txt

@@ -733,6 +733,7 @@ if(UNIX OR MSVC)
     add_executable(z_keyexpr_test ${PROJECT_SOURCE_DIR}/tests/z_keyexpr_test.c)
     add_executable(z_api_null_drop_test ${PROJECT_SOURCE_DIR}/tests/z_api_null_drop_test.c)
     add_executable(z_api_double_drop_test ${PROJECT_SOURCE_DIR}/tests/z_api_double_drop_test.c)
+    add_executable(z_api_timestamp_test ${PROJECT_SOURCE_DIR}/tests/z_api_timestamp_test.c)
     add_executable(z_test_fragment_tx ${PROJECT_SOURCE_DIR}/tests/z_test_fragment_tx.c)
     add_executable(z_test_fragment_rx ${PROJECT_SOURCE_DIR}/tests/z_test_fragment_rx.c)
     add_executable(z_perf_tx ${PROJECT_SOURCE_DIR}/tests/z_perf_tx.c)
@@ -767,6 +768,8 @@ if(UNIX OR MSVC)
     target_link_libraries(z_keyexpr_test zenohpico::lib)
     target_link_libraries(z_api_null_drop_test zenohpico::lib)
     target_link_libraries(z_api_double_drop_test zenohpico::lib)
+    target_link_libraries(z_api_timestamp_test zenohpico::lib)
+    target_compile_definitions(z_api_timestamp_test PRIVATE Z_TEST_HOOKS=1)
     target_link_libraries(z_test_fragment_tx zenohpico::lib)
     target_link_libraries(z_test_fragment_rx zenohpico::lib)
     target_link_libraries(z_perf_tx zenohpico::lib)
@@ -827,6 +830,7 @@ if(UNIX OR MSVC)
     add_test(z_keyexpr_test ${CMAKE_RUNTIME_OUTPUT_DIRECTORY}/z_keyexpr_test)
     add_test(z_api_null_drop_test ${CMAKE_RUNTIME_OUTPUT_DIRECTORY}/z_api_null_drop_test)
     add_test(z_api_double_drop_test ${CMAKE_RUNTIME_OUTPUT_DIRECTORY}/z_api_double_drop_test)
+    add_test(z_api_timestamp_test ${CMAKE_RUNTIME_OUTPUT_DIRECTORY}/z_api_timestamp_test)
     add_test(z_bytes_test ${CMAKE_RUNTIME_OUTPUT_DIRECTORY}/z_bytes_test)
     add_test(z_api_bytes_test ${CMAKE_RUNTIME_OUTPUT_DIRECTORY}/z_api_bytes_test)
     add_test(z_api_encoding_test ${CMAKE_RUNTIME_OUTPUT_DIRECTORY}/z_api_encoding_test)

include/zenoh-pico/api/primitives.h

@@ -907,17 +907,29 @@
 z_result_t z_bytes_writer_append(z_loaned_bytes_writer_t *writer, z_moved_bytes_t *bytes);
 
 /**
- * Create timestamp.
+ * Create a timestamp for the session.
+ *
+ * The generated timestamp is based on the platform wall-clock time. Timestamps
+ * generated by the same session are guaranteed to be strictly increasing. If the
+ * platform clock returns a value that is not greater than the previous timestamp
+ * generated by this session, the returned timestamp is advanced from the
+ * previous timestamp instead.
  *
  * Parameters:
  *   ts: An uninitialized :c:type:`z_timestamp_t`.
- *   zs: Pointer to a :c:type:`z_loaned_session_t` to get the id from.
+ *   zs: Pointer to a :c:type:`z_loaned_session_t`.
  *
  * Return:
- *   ``0`` if encode is successful, ``negative value`` otherwise (for example if RTC is not available on the system).
+ *   ``0`` if timestamp creation is successful, ``negative value`` otherwise
+ *   (for example if RTC is not available on the system).
  */
 z_result_t z_timestamp_new(z_timestamp_t *ts, const z_loaned_session_t *zs);
 
+#if defined(Z_TEST_HOOKS)
+typedef z_result_t (*_z_timestamp_time_since_epoch_override_fn)(_z_time_since_epoch *, void *);
+void _z_timestamp_set_time_since_epoch_override(_z_timestamp_time_since_epoch_override_fn fn, void *arg);
+#endif
+
 /**
  * Returns NTP64 time associated with this timestamp.
  *

include/zenoh-pico/net/session.h

@@ -139,6 +139,10 @@ typedef struct _z_session_t {
     uint32_t _entity_id;
     _z_zint_t _query_id;
     _z_zint_t _interest_id;
+    _z_ntp64_t _last_timestamp;
+#if Z_FEATURE_MULTI_THREAD == 1
+    _z_mutex_t _mutex_last_timestamp;
+#endif
 
     // Session declarations
     _z_resource_slist_t *_local_resources;

src/api/api.c

@@ -403,13 +403,48 @@ z_result_t z_bytes_writer_append(z_loaned_bytes_writer_t *writer, z_moved_bytes_
     return _z_bytes_writer_append_z_bytes(writer, &bytes->_this._val);
 }
 
+#if defined(Z_TEST_HOOKS)
+static _z_timestamp_time_since_epoch_override_fn _z_timestamp_time_since_epoch_override = NULL;
+static void *_z_timestamp_time_since_epoch_override_arg = NULL;
+
+void _z_timestamp_set_time_since_epoch_override(_z_timestamp_time_since_epoch_override_fn fn, void *arg) {
+    _z_timestamp_time_since_epoch_override = fn;
+    _z_timestamp_time_since_epoch_override_arg = arg;
+}
+#endif
+
 z_result_t z_timestamp_new(z_timestamp_t *ts, const z_loaned_session_t *zs) {
+    if ((ts == NULL) || (zs == NULL) || _Z_RC_IS_NULL(zs)) {
+        _Z_ERROR_RETURN(_Z_ERR_INVALID);
+    }
     *ts = _z_timestamp_null();
     _z_time_since_epoch t;
+#if defined(Z_TEST_HOOKS)
+    if (_z_timestamp_time_since_epoch_override != NULL) {
+        _Z_RETURN_IF_ERR(_z_timestamp_time_since_epoch_override(&t, _z_timestamp_time_since_epoch_override_arg));
+    } else {
+        _Z_RETURN_IF_ERR(_z_get_time_since_epoch(&t));
+    }
+#else
     _Z_RETURN_IF_ERR(_z_get_time_since_epoch(&t));
+#endif
+
+    _z_session_t *s = _Z_RC_IN_VAL(zs);
+    _z_ntp64_t time = _z_timestamp_ntp64_from_time(t.secs, t.nanos);
+    _Z_RETURN_IF_ERR(_z_mutex_lock(&s->_mutex_last_timestamp));
+    if (time <= s->_last_timestamp) {
+        if (s->_last_timestamp == UINT64_MAX) {
+            _z_mutex_unlock(&s->_mutex_last_timestamp);
+            _Z_ERROR_RETURN(_Z_ERR_GENERIC);
+        }
+        time = s->_last_timestamp + 1;
+    }
+    s->_last_timestamp = time;
+    _z_mutex_unlock(&s->_mutex_last_timestamp);
+
     ts->valid = true;
-    ts->time = _z_timestamp_ntp64_from_time(t.secs, t.nanos);
-    ts->id = _Z_RC_IN_VAL(zs)->_local_zid;
+    ts->time = time;
+    ts->id = s->_local_zid;
     return _Z_RES_OK;
 }
 

src/session/utils.c

@@ -61,10 +61,16 @@
 #if Z_FEATURE_MULTI_THREAD == 1
     _Z_RETURN_IF_ERR(_z_mutex_init(&zn->_mutex_inner));
     ret = _z_mutex_rec_init(&zn->_mutex_transport);
     if (ret != _Z_RES_OK) {
         _z_mutex_drop(&zn->_mutex_inner);
         _Z_ERROR_RETURN(ret);
     }
+    ret = _z_mutex_init(&zn->_mutex_last_timestamp);
+    if (ret != _Z_RES_OK) {
+        _z_mutex_rec_drop(&zn->_mutex_transport);
+        _z_mutex_drop(&zn->_mutex_inner);
+        _Z_ERROR_RETURN(ret);
+    }
 #if Z_FEATURE_ADMIN_SPACE == 1
     ret = _z_mutex_init(&zn->_mutex_admin_space);
     if (ret != _Z_RES_OK) {
@@ -80,6 +86,7 @@
     zn->_entity_id = 1;
     zn->_resource_id = 1;
     zn->_query_id = 1;
+    zn->_last_timestamp = 0;
 
 #if Z_FEATURE_AUTO_RECONNECT == 1
     _z_config_init(&zn->_config);
@@ -150,6 +157,7 @@
         _z_mutex_drop(&zn->_mutex_admin_space);
 #endif
         _z_mutex_rec_drop(&zn->_mutex_transport);
+        _z_mutex_drop(&zn->_mutex_last_timestamp);
         _z_mutex_drop(&zn->_mutex_inner);
 #endif
         _z_sync_group_drop(&zn->_callback_drop_sync_group);
@@ -231,7 +239,7 @@
     _z_session_close(zn);
     _z_runtime_clear(&zn->_runtime);
 #if Z_FEATURE_AUTO_RECONNECT == 1
     _z_config_clear(&zn->_config);
 #endif
     _z_session_transport_mutex_lock(zn);
     _z_transport_clear(&zn->_tp);
@@ -242,6 +250,7 @@
     _z_mutex_drop(&zn->_mutex_admin_space);
 #endif
     _z_mutex_rec_drop(&zn->_mutex_transport);
+    _z_mutex_drop(&zn->_mutex_last_timestamp);
     _z_mutex_drop(&zn->_mutex_inner);
 #endif  // Z_FEATURE_MULTI_THREAD == 1
     _z_sync_group_drop(&zn->_callback_drop_sync_group);

src/system/unix/system.c

@@ -329,9 +329,11 @@
 }
 
 z_result_t _z_get_time_since_epoch(_z_time_since_epoch *t) {
-    z_time_t now;
-    gettimeofday(&now, NULL);
+    struct timespec now;
+    if (clock_gettime(CLOCK_REALTIME, &now) != 0) {
+        return _Z_ERR_SYSTEM_GENERIC;
+    }
     t->secs = (uint32_t)now.tv_sec;
-    t->nanos = (uint32_t)now.tv_usec * 1000;
-    return 0;
+    t->nanos = (uint32_t)now.tv_nsec;
+    return _Z_RES_OK;
 }

tests/z_api_timestamp_test.c

@@ -0,0 +1,114 @@
+//
+// Copyright (c) 2026 ZettaScale Technology
+//
+// This program and the accompanying materials are made available under the
+// terms of the Eclipse Public License 2.0 which is available at
+// http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
+// which is available at https://www.apache.org/licenses/LICENSE-2.0.
+//
+// SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
+//
+// Contributors:
+//   ZettaScale Zenoh Team, <zenoh@zettascale.tech>
+
+#include <stddef.h>
+
+#include "zenoh-pico/api/primitives.h"
+#include "zenoh-pico/api/types.h"
+#include "zenoh-pico/net/session.h"
+#include "zenoh-pico/protocol/core.h"
+#include "zenoh-pico/session/session.h"
+#include "zenoh-pico/session/utils.h"
+
+#undef NDEBUG
+#include <assert.h>
+
+typedef struct {
+    _z_id_t zid;
+    _z_session_t session;
+    _z_session_rc_t session_rc;
+    _z_time_since_epoch clock_value;
+} timestamp_test_fixture_t;
+
+static void setup_session(timestamp_test_fixture_t *fixture) {
+    *fixture = (timestamp_test_fixture_t){0};
+    _z_session_generate_zid(&fixture->zid, Z_ZID_LENGTH);
+    assert(_z_session_init(&fixture->session, &fixture->zid) == _Z_RES_OK);
+    if (_Z_RC_IS_NULL(&fixture->session_rc)) {
+        fixture->session_rc = _z_session_rc_new(&fixture->session);
+        assert(!_Z_RC_IS_NULL(&fixture->session_rc));
+    } else {
+        fixture->session_rc._val = &fixture->session;
+    }
+}
+
+static z_result_t fake_time_since_epoch(_z_time_since_epoch *t, void *arg) {
+    timestamp_test_fixture_t *fixture = (timestamp_test_fixture_t *)arg;
+    *t = fixture->clock_value;
+    return _Z_RES_OK;
+}
+
+static void setup_session_with_fake_clock(timestamp_test_fixture_t *fixture, uint32_t secs, uint32_t nanos) {
+    setup_session(fixture);
+    fixture->clock_value.secs = secs;
+    fixture->clock_value.nanos = nanos;
+    _z_timestamp_set_time_since_epoch_override(fake_time_since_epoch, fixture);
+}
+
+static void cleanup_session(timestamp_test_fixture_t *fixture) {
+    _z_timestamp_set_time_since_epoch_override(NULL, NULL);
+    assert(_z_session_rc_decr(&fixture->session_rc));
+    fixture->session_rc = _z_session_rc_null();
+    _z_session_clear(&fixture->session);
+}
+
+static void test_timestamp_new_with_real_clock(void) {
+    timestamp_test_fixture_t fixture;
+    setup_session(&fixture);
+
+    z_timestamp_t prev;
+    z_timestamp_t next;
+    assert(z_timestamp_new(&prev, &fixture.session_rc) == _Z_RES_OK);
+    assert(_z_timestamp_check(&prev));
+    assert(z_timestamp_ntp64_time(&prev) > 0);
+    assert(_z_id_eq(&prev.id, &fixture.zid));
+
+    for (size_t i = 0; i < 10000; i++) {
+        assert(z_timestamp_new(&next, &fixture.session_rc) == _Z_RES_OK);
+        assert(_z_timestamp_cmp(&prev, &next) < 0);
+        prev = next;
+    }
+
+    cleanup_session(&fixture);
+}
+
+static void test_timestamp_new_with_repeated_time(void) {
+    timestamp_test_fixture_t fixture;
+    setup_session_with_fake_clock(&fixture, 42, 100);
+
+    z_timestamp_t first;
+    z_timestamp_t second;
+    z_timestamp_t third;
+    _z_ntp64_t timestamp = _z_timestamp_ntp64_from_time(42, 100);
+    assert(z_timestamp_new(&first, &fixture.session_rc) == _Z_RES_OK);
+    assert(z_timestamp_new(&second, &fixture.session_rc) == _Z_RES_OK);
+    fixture.clock_value.nanos = 99;
+    assert(z_timestamp_new(&third, &fixture.session_rc) == _Z_RES_OK);
+
+    assert(z_timestamp_ntp64_time(&first) == timestamp);
+    assert(z_timestamp_ntp64_time(&second) == timestamp + 1);
+    assert(z_timestamp_ntp64_time(&third) == timestamp + 2);
+    assert(_z_timestamp_cmp(&first, &second) < 0);
+    assert(_z_timestamp_cmp(&second, &third) < 0);
+    assert(_z_id_eq(&first.id, &fixture.zid));
+    assert(_z_id_eq(&second.id, &fixture.zid));
+    assert(_z_id_eq(&third.id, &fixture.zid));
+
+    cleanup_session(&fixture);
+}
+
+int main(void) {
+    test_timestamp_new_with_real_clock();
+    test_timestamp_new_with_repeated_time();
+    return 0;
+}