loopback: support locality and local delivery for same-session messages

- Deliver PUT/DELETE, query, reply, and reply-final messages for a key
expression that is defined in the same local session locally, without
causing network activity.
- Extend z_api_source_info_test to validate deliveries both to remote and
local subscribers.
- Support locality of samples to be received by subscribers or targeted by
publishers.

User can now set allowed_origin / allowed_destination in
z_put_options_t, z_publisher_put_options_t, z_subscriber_options_t,
z_queryable_options_t, etc. with the following values:
- ZP_LOCALITY_SESSION_LOCAL - stay inside the session (don't send
  anything to network)
- ZP_LOCALITY_REMOTE - send/accept only over the transport (no
  same-session samples)
- ZP_LOCALITY_ANY - allow both,
exactly as with zenoh-c's zc_locality_t.

Author: steils

CMakeLists.txt

@@ -684,6 +684,7 @@ if(UNIX OR MSVC)
       add_executable(z_api_source_info_test ${PROJECT_SOURCE_DIR}/tests/z_api_source_info_test.c)
       add_executable(z_api_queryable_test ${PROJECT_SOURCE_DIR}/tests/z_api_queryable_test.c)
       add_executable(z_api_scheduler_test ${PROJECT_SOURCE_DIR}/tests/z_api_scheduler_test.c)
+      add_executable(z_local_loopback_test ${PROJECT_SOURCE_DIR}/tests/z_local_loopback_test.c)
 if(UNIX)
       add_executable(z_api_advanced_pubsub_test ${PROJECT_SOURCE_DIR}/tests/z_api_advanced_pubsub_test.c
                                                 ${PROJECT_SOURCE_DIR}/tests/utils/tcp_proxy.c)
@@ -701,6 +702,15 @@ endif()
       target_link_libraries(z_api_source_info_test zenohpico::lib)
       target_link_libraries(z_api_queryable_test zenohpico::lib)
       target_link_libraries(z_api_scheduler_test zenohpico::lib)
+      target_link_libraries(z_local_loopback_test zenohpico::lib)
+      target_include_directories(z_local_loopback_test PRIVATE ${PROJECT_SOURCE_DIR}/include)
+      target_compile_definitions(z_local_loopback_test PRIVATE Z_LOOPBACK_TESTING=1)
+      if(PICO_SHARED)
+        target_compile_definitions(${Libname}_shared PRIVATE Z_LOOPBACK_TESTING=1)
+      endif()
+      if(PICO_STATIC)
+        target_compile_definitions(${Libname}_static PRIVATE Z_LOOPBACK_TESTING=1)
+      endif()
       target_link_libraries(z_api_advanced_pubsub_test zenohpico::lib)
 
       configure_file(${PROJECT_SOURCE_DIR}/tests/routed.sh ${CMAKE_RUNTIME_OUTPUT_DIRECTORY}/routed.sh COPYONLY)
@@ -722,6 +732,7 @@ endif()
       add_test(z_api_source_info_test bash ${CMAKE_RUNTIME_OUTPUT_DIRECTORY}/api.sh z_api_source_info_test)
       add_test(z_api_queryable_test bash ${CMAKE_RUNTIME_OUTPUT_DIRECTORY}/api.sh z_api_queryable_test)
       add_test(z_api_scheduler_test bash ${CMAKE_RUNTIME_OUTPUT_DIRECTORY}/api.sh z_api_scheduler_test)
+      add_test(z_local_loopback_test bash ${CMAKE_RUNTIME_OUTPUT_DIRECTORY}/api.sh z_local_loopback_test)
       add_test(z_api_advanced_pubsub_test bash ${CMAKE_RUNTIME_OUTPUT_DIRECTORY}/api.sh z_api_advanced_pubsub_test)
     endif()
   endif()

include/zenoh-pico/api/constants.h

@@ -14,6 +14,8 @@
 #ifndef ZENOH_PICO_API_CONSTANTS_H
 #define ZENOH_PICO_API_CONSTANTS_H
 
+#include <stdbool.h>
+
 #define Z_SELECTOR_TIME "_time="
 #define Z_SELECTOR_QUERY_MATCH "_anyke"
 
@@ -54,6 +56,44 @@ typedef enum z_whatami_t {
 } z_whatami_t;
 #define Z_WHATAMI_DEFAULT Z_WHATAMI_ROUTER;
 
+/**
+ * The locality of samples to be received by subscribers or targeted by publishers.
+ *
+ * Enumerators:
+ *   ZP_LOCALITY_ANY: Allow both session-local and remote traffic.
+ *   ZP_LOCALITY_SESSION_LOCAL: Allow session-local traffic only.
+ *   ZP_LOCALITY_REMOTE: Allow remote traffic only.
+ */
+typedef enum zp_locality_t {
+    ZP_LOCALITY_ANY = 0,
+    ZP_LOCALITY_SESSION_LOCAL = 1,
+    ZP_LOCALITY_REMOTE = 2,
+} zp_locality_t;
+
+static inline zp_locality_t zp_locality_default(void) { return ZP_LOCALITY_ANY; }
+
+static inline bool zp_locality_allows_local(zp_locality_t locality) {
+    switch (locality) {
+        case ZP_LOCALITY_REMOTE:
+            return false;
+        case ZP_LOCALITY_ANY:
+        case ZP_LOCALITY_SESSION_LOCAL:
+        default:
+            return true;
+    }
+}
+
+static inline bool zp_locality_allows_remote(zp_locality_t locality) {
+    switch (locality) {
+        case ZP_LOCALITY_SESSION_LOCAL:
+            return false;
+        case ZP_LOCALITY_ANY:
+        case ZP_LOCALITY_REMOTE:
+        default:
+            return true;
+    }
+}
+
 /**
  * Status values for keyexpr canonization operation.
  * Used as return value of canonization-related functions,

include/zenoh-pico/api/types.h

@@ -164,7 +164,7 @@ typedef _z_matching_status_t z_matching_status_t;
  * Represents the configuration used to configure a subscriber upon declaration :c:func:`z_declare_subscriber`.
  */
 typedef struct {
-    uint8_t __dummy;  // Just to avoid empty structures that might cause undefined behavior
+    zp_locality_t allowed_origin;
 } z_subscriber_options_t;
 
 /**
@@ -210,6 +210,7 @@ typedef struct {
 #ifdef Z_FEATURE_UNSTABLE_API
     z_reliability_t reliability;
 #endif
+    zp_locality_t allowed_destination;
 } z_publisher_options_t;
 
 /**
@@ -233,6 +234,7 @@ typedef struct z_querier_options_t {
     z_query_consolidation_t consolidation;
     z_congestion_control_t congestion_control;
     bool is_express;
+    zp_locality_t allowed_destination;
     z_priority_t priority;
     uint64_t timeout_ms;
 } z_querier_options_t;
@@ -259,6 +261,7 @@ typedef struct z_querier_get_options_t {
  */
 typedef struct {
     bool complete;
+    zp_locality_t allowed_origin;
 } z_queryable_options_t;
 
 /**
@@ -337,6 +340,7 @@ typedef struct {
     z_timestamp_t *timestamp;
     bool is_express;
     z_moved_bytes_t *attachment;
+    zp_locality_t allowed_destination;
 #ifdef Z_FEATURE_UNSTABLE_API
     z_reliability_t reliability;
     z_moved_source_info_t *source_info;
@@ -359,6 +363,7 @@ typedef struct {
     z_priority_t priority;
     bool is_express;
     z_timestamp_t *timestamp;
+    zp_locality_t allowed_destination;
 #ifdef Z_FEATURE_UNSTABLE_API
     z_reliability_t reliability;
     z_moved_source_info_t *source_info;
@@ -419,6 +424,7 @@ typedef struct {
     z_congestion_control_t congestion_control;
     z_priority_t priority;
     bool is_express;
+    zp_locality_t allowed_destination;
     z_query_target_t target;
     uint64_t timeout_ms;
     z_moved_bytes_t *attachment;

include/zenoh-pico/net/primitives.h

@@ -110,7 +110,7 @@ _z_keyexpr_t _z_update_keyexpr_to_declared(_z_session_t *zs, _z_keyexpr_t keyexp
  */
 _z_publisher_t _z_declare_publisher(const _z_session_rc_t *zn, _z_keyexpr_t keyexpr, _z_encoding_t *encoding,
                                     z_congestion_control_t congestion_control, z_priority_t priority, bool is_express,
-                                    z_reliability_t reliability);
+                                    z_reliability_t reliability, zp_locality_t allowed_destination);
 
 /**
  * Undeclare a :c:type:`_z_publisher_t`.
@@ -147,7 +147,8 @@ z_result_t _z_undeclare_publisher(_z_publisher_t *pub);
 z_result_t _z_write(_z_session_t *zn, const _z_keyexpr_t *keyexpr, const _z_bytes_t *payload,
                     const _z_encoding_t *encoding, const z_sample_kind_t kind, const z_congestion_control_t cong_ctrl,
                     z_priority_t priority, bool is_express, const _z_timestamp_t *timestamp,
-                    const _z_bytes_t *attachment, z_reliability_t reliability, const _z_source_info_t *source_info);
+                    const _z_bytes_t *attachment, z_reliability_t reliability, const _z_source_info_t *source_info,
+                    zp_locality_t allowed_destination);
 #endif
 
 #if Z_FEATURE_SUBSCRIPTION == 1
@@ -165,7 +166,8 @@ z_result_t _z_write(_z_session_t *zn, const _z_keyexpr_t *keyexpr, const _z_byte
  *    The created :c:type:`_z_subscriber_t` (in null state if the declaration failed).
  */
 _z_subscriber_t _z_declare_subscriber(const _z_session_rc_t *zn, _z_keyexpr_t keyexpr,
-                                      _z_closure_sample_callback_t callback, _z_drop_handler_t dropper, void *arg);
+                                      _z_closure_sample_callback_t callback, _z_drop_handler_t dropper, void *arg,
+                                      zp_locality_t allowed_origin);
 
 /**
  * Undeclare a :c:type:`_z_subscriber_t`.
@@ -195,7 +197,8 @@ z_result_t _z_undeclare_subscriber(_z_subscriber_t *sub);
  *    The created :c:type:`_z_queryable_t` (in null state if the declaration failed)..
  */
 _z_queryable_t _z_declare_queryable(const _z_session_rc_t *zn, _z_keyexpr_t keyexpr, bool complete,
-                                    _z_closure_query_callback_t callback, _z_drop_handler_t dropper, void *arg);
+                                    _z_closure_query_callback_t callback, _z_drop_handler_t dropper, void *arg,
+                                    zp_locality_t allowed_origin);
 
 /**
  * Undeclare a :c:type:`_z_queryable_t`.
@@ -270,7 +273,8 @@ z_result_t _z_send_reply_err(const _z_query_t *query, const _z_session_rc_t *zsr
 _z_querier_t _z_declare_querier(const _z_session_rc_t *zn, _z_keyexpr_t keyexpr,
                                 z_consolidation_mode_t consolidation_mode, z_congestion_control_t congestion_control,
                                 z_query_target_t target, z_priority_t priority, bool is_express, uint64_t timeout_ms,
-                                _z_encoding_t *encoding, z_reliability_t reliability);
+                                _z_encoding_t *encoding, z_reliability_t reliability,
+                                zp_locality_t allowed_destination);
 
 /**
  * Undeclare a :c:type:`_z_querier_t`.
@@ -308,7 +312,7 @@ z_result_t _z_query(_z_session_t *zn, const _z_keyexpr_t *keyexpr, const char *p
                     z_query_target_t target, z_consolidation_mode_t consolidation, const _z_bytes_t *payload,
                     const _z_encoding_t *encoding, _z_closure_reply_callback_t callback, _z_drop_handler_t dropper,
                     void *arg, uint64_t timeout_ms, const _z_bytes_t *attachment, _z_n_qos_t qos,
-                    z_congestion_control_t cong_ctrl);
+                    z_congestion_control_t cong_ctrl, zp_locality_t allowed_destination);
 #endif
 
 #if Z_FEATURE_INTEREST == 1

include/zenoh-pico/net/publish.h

@@ -15,6 +15,7 @@
 #ifndef INCLUDE_ZENOH_PICO_NET_PUBLISH_H
 #define INCLUDE_ZENOH_PICO_NET_PUBLISH_H
 
+#include "zenoh-pico/api/constants.h"
 #include "zenoh-pico/net/filtering.h"
 #include "zenoh-pico/net/session.h"
 #include "zenoh-pico/protocol/core.h"
@@ -35,6 +36,7 @@ typedef struct _z_publisher_t {
     z_priority_t _priority;
     z_reliability_t reliability;
     bool _is_express;
+    zp_locality_t _allowed_destination;
     _z_write_filter_t _filter;
 } _z_publisher_t;
 

include/zenoh-pico/net/query.h

@@ -67,6 +67,7 @@ typedef struct _z_querier_t {
     z_reliability_t reliability;
     bool _is_express;
     uint64_t _timeout_ms;
+    zp_locality_t _allowed_destination;
     _z_write_filter_t _filter;
 } _z_querier_t;
 

include/zenoh-pico/session/loopback.h

@@ -0,0 +1,69 @@
+//
+// Copyright (c) 2025 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, <[email protected]>
+//
+
+#ifndef ZENOH_PICO_SESSION_LOOPBACK_H
+#define ZENOH_PICO_SESSION_LOOPBACK_H
+
+#include <stdbool.h>
+#include <stdint.h>
+
+#include "zenoh-pico/api/constants.h"
+#include "zenoh-pico/net/query.h"
+#include "zenoh-pico/protocol/core.h"
+#include "zenoh-pico/protocol/definitions/network.h"
+#include "zenoh-pico/session/session.h"
+#include "zenoh-pico/transport/transport.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+bool _z_session_deliver_push_locally(_z_session_t *zn, const _z_keyexpr_t *keyexpr, const _z_bytes_t *payload,
+                                     const _z_encoding_t *encoding, z_sample_kind_t kind, _z_n_qos_t qos,
+                                     const _z_timestamp_t *timestamp, const _z_bytes_t *attachment,
+                                     z_reliability_t reliability, const _z_source_info_t *source_info,
+                                     zp_locality_t allowed_destination);
+
+bool _z_session_deliver_query_locally(_z_session_t *zn, const _z_keyexpr_t *keyexpr, const _z_slice_t *parameters,
+                                      z_consolidation_mode_t consolidation, const _z_bytes_t *payload,
+                                      const _z_encoding_t *encoding, const _z_bytes_t *attachment,
+                                      const _z_source_info_t *source_info, _z_zint_t qid, uint64_t timeout_ms,
+                                      _z_n_qos_t qos, zp_locality_t allowed_destination);
+
+bool _z_session_deliver_reply_locally(const _z_query_t *query, const _z_session_rc_t *responder,
+                                      const _z_keyexpr_t *keyexpr, const _z_bytes_t *payload,
+                                      const _z_encoding_t *encoding, z_sample_kind_t kind, _z_n_qos_t qos,
+                                      const _z_timestamp_t *timestamp, const _z_bytes_t *attachment,
+                                      const _z_source_info_t *source_info);
+
+bool _z_session_deliver_reply_err_locally(const _z_query_t *query, const _z_session_rc_t *responder,
+                                          const _z_bytes_t *payload, const _z_encoding_t *encoding, _z_n_qos_t qos);
+
+bool _z_session_deliver_reply_final_locally(_z_session_t *zn, _z_zint_t rid, bool allow_remote);
+
+#if defined(Z_LOOPBACK_TESTING)
+typedef _z_transport_common_t *(*_z_session_transport_override_fn)(_z_session_t *);
+void _z_session_set_transport_common_override(_z_session_transport_override_fn fn);
+
+typedef z_result_t (*_z_session_send_override_fn)(_z_session_t *zn, const _z_network_message_t *n_msg,
+                                                  z_reliability_t reliability, z_congestion_control_t cong_ctrl,
+                                                  void *peer, bool *handled);
+void _z_transport_set_send_n_msg_override(_z_session_send_override_fn fn);
+#endif
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /* ZENOH_PICO_SESSION_LOOPBACK_H */

include/zenoh-pico/session/query.h

@@ -36,7 +36,7 @@ z_result_t _z_trigger_query_reply_partial(_z_session_t *zn, _z_zint_t reply_cont
                                           _z_transport_peer_common_t *peer);
 z_result_t _z_trigger_query_reply_err(_z_session_t *zn, _z_zint_t id, _z_msg_err_t *msg,
                                       _z_entity_global_id_t *replier_id);
-z_result_t _z_trigger_query_reply_final(_z_session_t *zn, _z_zint_t id);
+z_result_t _z_trigger_query_reply_final(_z_session_t *zn, _z_zint_t id, bool finalize);
 void _z_unregister_pending_query(_z_session_t *zn, _z_pending_query_t *pq);
 void _z_flush_pending_queries(_z_session_t *zn);
 #endif

include/zenoh-pico/session/queryable.h

@@ -39,6 +39,7 @@ typedef struct {
     _z_keyexpr_t ke_out;
     _z_queryable_infos_svec_t infos;
     size_t qle_nb;
+    bool is_remote;
 } _z_queryable_cache_data_t;
 
 void _z_queryable_cache_invalidate(_z_session_t *zn);

include/zenoh-pico/session/session.h

@@ -18,6 +18,7 @@
 #include <stdbool.h>
 #include <stdint.h>
 
+#include "zenoh-pico/api/constants.h"
 #include "zenoh-pico/collections/element.h"
 #include "zenoh-pico/collections/list.h"
 #include "zenoh-pico/collections/refcount.h"
@@ -73,6 +74,7 @@ typedef struct {
     _z_keyexpr_t _declared_key;
     uint16_t _key_id;
     uint32_t _id;
+    zp_locality_t _allowed_origin;
     _z_closure_sample_callback_t _callback;
     _z_drop_handler_t _dropper;
     void *_arg;
@@ -109,6 +111,7 @@ typedef struct {
     _z_drop_handler_t _dropper;
     void *_arg;
     bool _complete;
+    zp_locality_t _allowed_origin;
 } _z_session_queryable_t;
 
 bool _z_session_queryable_eq(const _z_session_queryable_t *one, const _z_session_queryable_t *two);
@@ -137,6 +140,7 @@ typedef struct {
     _z_zint_t _id;
     _z_closure_reply_callback_t _callback;
     _z_drop_handler_t _dropper;
+    zp_locality_t _allowed_destination;
     z_clock_t _start_time;
     uint64_t _timeout;
     void *_arg;