loopback: support 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.

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/session/loopback.h

@@ -0,0 +1,66 @@
+//
+// 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 "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);
+
+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, _z_n_qos_t qos);
+
+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_has_remote_targets(const _z_session_t *zn);
+bool _z_session_deliver_reply_final_locally(_z_session_t *zn, _z_zint_t rid);
+
+#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 */

src/net/primitives.c

@@ -33,6 +33,7 @@
 #include "zenoh-pico/protocol/keyexpr.h"
 #include "zenoh-pico/session/interest.h"
 #include "zenoh-pico/session/liveliness.h"
+#include "zenoh-pico/session/loopback.h"
 #include "zenoh-pico/session/query.h"
 #include "zenoh-pico/session/queryable.h"
 #include "zenoh-pico/session/resource.h"
@@ -195,8 +196,15 @@ z_result_t _z_write(_z_session_t *zn, const _z_keyexpr_t *keyexpr, const _z_byte
                     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) {
     z_result_t ret = _Z_RES_OK;
-    _z_network_message_t msg;
     _z_qos_t qos = _z_n_qos_make(is_express, cong_ctrl == Z_CONGESTION_CONTROL_BLOCK, priority);
+
+    bool handled_locally = _z_session_deliver_push_locally(zn, keyexpr, payload, encoding, kind, qos, timestamp,
+                                                           attachment, reliability, source_info);
+    if (handled_locally && !_z_session_has_remote_targets(zn)) {
+        return _Z_RES_OK;
+    }
+
+    _z_network_message_t msg;
     switch (kind) {
         case Z_SAMPLE_KIND_PUT:
             _z_n_msg_make_push_put(&msg, keyexpr, payload, encoding, qos, timestamp, attachment, reliability,
@@ -378,6 +386,12 @@ z_result_t _z_send_reply(const _z_query_t *query, const _z_session_rc_t *zsrc, c
     }
     // Build the reply context decorator. This is NOT the final reply.
     _z_n_qos_t qos = _z_n_qos_make(is_express, cong_ctrl == Z_CONGESTION_CONTROL_BLOCK, priority);
+    bool handled_locally = _z_session_deliver_reply_locally(query, zsrc, keyexpr, payload, encoding, kind, qos,
+                                                            timestamp, att, source_info);
+    if (handled_locally && !_z_session_has_remote_targets(zn)) {
+        return _Z_RES_OK;
+    }
+
     _z_zenoh_message_t z_msg;
     switch (kind) {
         case Z_SAMPLE_KIND_PUT:
@@ -408,6 +422,11 @@ z_result_t _z_send_reply_err(const _z_query_t *query, const _z_session_rc_t *zsr
     // Build the reply context decorator. This is NOT the final reply.
     _z_n_qos_t qos = _z_n_qos_make(false, true, Z_PRIORITY_DEFAULT);
     _z_source_info_t source_info = _z_source_info_null();
+    bool handled_locally = _z_session_deliver_reply_err_locally(query, zsrc, payload, encoding, qos);
+    if (handled_locally && !_z_session_has_remote_targets(zn)) {
+        return _Z_RES_OK;
+    }
+
     _z_zenoh_message_t msg;
     _z_n_msg_make_reply_err(&msg, &zn->_local_zid, query->_request_id, Z_RELIABILITY_DEFAULT, qos, payload, encoding,
                             &source_info);
@@ -501,6 +520,13 @@ z_result_t _z_query(_z_session_t *zn, const _z_keyexpr_t *keyexpr, const char *p
     _z_n_msg_make_query(&z_msg, keyexpr, &params, pq->_id, Z_RELIABILITY_DEFAULT, pq->_consolidation, payload, encoding,
                         timeout_ms, attachment, qos, &source_info);
 
+    bool handled_locally = _z_session_deliver_query_locally(zn, keyexpr, &params, pq->_consolidation, payload, encoding,
+                                                            attachment, &source_info, pq->_id, qos);
+    if (handled_locally && !_z_session_has_remote_targets(zn)) {
+        _z_trigger_query_reply_final(zn, pq->_id);
+        return _Z_RES_OK;
+    }
+
     if (_z_send_n_msg(zn, &z_msg, Z_RELIABILITY_RELIABLE, cong_ctrl, NULL) != _Z_RES_OK) {
         _z_unregister_pending_query(zn, pq);
         _Z_ERROR_RETURN(_Z_ERR_TRANSPORT_TX_FAILED);

src/net/query.c

@@ -14,6 +14,7 @@
 #include "zenoh-pico/net/query.h"
 
 #include "zenoh-pico/net/session.h"
+#include "zenoh-pico/session/loopback.h"
 #include "zenoh-pico/transport/common/tx.h"
 #include "zenoh-pico/utils/logging.h"
 
@@ -31,10 +32,16 @@ z_result_t _z_query_send_reply_final(_z_query_t *q) {
     if (_Z_RC_IS_NULL(&sess_rc)) {
         _Z_ERROR_RETURN(_Z_ERR_TRANSPORT_TX_FAILED);
     }
+
+    _z_session_t *session = _Z_RC_IN_VAL(&sess_rc);
+    if (_z_session_deliver_reply_final_locally(session, q->_request_id)) {
+        _z_session_rc_drop(&sess_rc);
+        return _Z_RES_OK;
+    }
+
     _z_zenoh_message_t z_msg;
     _z_n_msg_make_response_final(&z_msg, q->_request_id);
-    z_result_t ret =
-        _z_send_n_msg(_Z_RC_IN_VAL(&sess_rc), &z_msg, Z_RELIABILITY_RELIABLE, Z_CONGESTION_CONTROL_BLOCK, NULL);
+    z_result_t ret = _z_send_n_msg(session, &z_msg, Z_RELIABILITY_RELIABLE, Z_CONGESTION_CONTROL_BLOCK, NULL);
     _z_msg_clear(&z_msg);
     _z_session_rc_drop(&sess_rc);
     return ret;