From 2404b598416629f0214f759e76acee32e052ad49 Mon Sep 17 00:00:00 2001 From: Andrea Lacava Date: Sat, 22 Aug 2026 00:13:44 -0400 Subject: [PATCH 1/2] feat(e3ap): surface assigned message ids for dApp sends and relayed xApp control MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes #67 and #68: both are correlation gaps around E3-MessageID that already exists internally but never reaches the API surface. - send_control()/send_report() (E3Agent + E3Interface) gain a trailing uint32_t* out_message_id = nullptr, mirroring subscribe()'s existing out_request_id. The generated id was already computed before queueing; this only plumbs it out. Source-compatible: no existing caller needs to change, and the ErrorCode return is unchanged in both the success and failure case. - XAppControlAction gains a decode-time-only message_id field, set by the dApp inbound loop from the wrapping E3-PDU's id (already decoded into Pdu::message_id, never forwarded before). No ASN.1/JSON/Protobuf grammar changes — the id was always on the wire, just not threaded through the C++ handler call. This is ABI-breaking (struct layout changes) but not source-breaking for any of the four existing XAppControlHandler registrations (examples/simple_dapp.cpp, swig/e3_dapp_session.cpp, tests/integration/bench_latrec_load.cpp, plus the header/setter declarations) — none construct or destructure the struct positionally. The SWIG binding follows both fixes through to Python: send_control()/ send_report() now return the assigned message id (positive, 1..1000) on success instead of always 0, matching subscribe()'s existing convention; E3_EVENT_XAPP_CONTROL events now carry the id in the existing request_id field. This is a Python-visible behavior change — updating the dApp library's outbound loop (which currently checks `rc != SUCCESS` for these two calls) is left as a follow-up in that repo. Extends tests/integration/test_role_pair_zmq_ipc.cpp's existing RAN+dApp round trip: asserts send_control's out_message_id matches the id the RAN sees as request_message_id, and that two RAN -> dApp xApp control relays deliver distinct, nonzero message ids to the dApp handler. Assisted-by: Claude:claude-sonnet-5 Signed-off-by: Andrea Lacava --- include/libe3/e3_agent.hpp | 23 ++++++++++++--- include/libe3/e3_interface.hpp | 9 ++++-- include/libe3/types.hpp | 3 ++ src/core/e3_agent.cpp | 11 ++++--- src/core/e3_interface.cpp | 29 +++++++++++++----- swig/e3_dapp_session.cpp | 22 +++++++++++--- swig/e3_dapp_session.hpp | 16 ++++++++-- tests/integration/test_role_pair_zmq_ipc.cpp | 31 +++++++++++++++++++- 8 files changed, 119 insertions(+), 25 deletions(-) diff --git a/include/libe3/e3_agent.hpp b/include/libe3/e3_agent.hpp index 4c71c4d..51dff05 100644 --- a/include/libe3/e3_agent.hpp +++ b/include/libe3/e3_agent.hpp @@ -277,13 +277,28 @@ class E3Agent { /** Send a SubscriptionDelete for a previously subscribed RAN function. */ ErrorCode unsubscribe(uint32_t ran_function_id); - /** Send a dApp control action to the RAN. */ + /** + * @brief Send a dApp control action to the RAN. + * + * @param out_message_id when non-null, receives the assigned E3-MessageID on + * success, so the caller can correlate a later ack/response back to + * this send. + */ ErrorCode send_control(uint32_t ran_function_id, uint32_t control_id, - std::vector action_data); + std::vector action_data, + uint32_t* out_message_id = nullptr); - /** Send a dApp report to the RAN. */ - ErrorCode send_report(uint32_t ran_function_id, std::vector report_data); + /** + * @brief Send a dApp report to the RAN. + * + * @param out_message_id when non-null, receives the assigned E3-MessageID on + * success, so the caller can correlate a later ack/response back to + * this send. + */ + ErrorCode send_report(uint32_t ran_function_id, + std::vector report_data, + uint32_t* out_message_id = nullptr); /** Send a ReleaseMessage. The dApp stays running until stop() is called. */ ErrorCode release(); diff --git a/include/libe3/e3_interface.hpp b/include/libe3/e3_interface.hpp index 8f7f17d..697411e 100644 --- a/include/libe3/e3_interface.hpp +++ b/include/libe3/e3_interface.hpp @@ -189,11 +189,16 @@ class E3Interface { std::optional periodicity = std::nullopt, uint32_t* out_request_id = nullptr); ErrorCode queue_subscription_delete(uint32_t ran_function_id); + // out_message_id, when non-null, receives the assigned E3-MessageID on + // success, so the caller can correlate a later ack/response back to this + // send. ErrorCode queue_dapp_control_action(uint32_t ran_function_id, uint32_t control_id, - std::vector action_data); + std::vector action_data, + uint32_t* out_message_id = nullptr); ErrorCode queue_dapp_report(uint32_t ran_function_id, - std::vector report_data); + std::vector report_data, + uint32_t* out_message_id = nullptr); ErrorCode queue_release_message(); // Block until the setup handshake completes (or times out / fails). diff --git a/include/libe3/types.hpp b/include/libe3/types.hpp index 197f86d..83d7c26 100644 --- a/include/libe3/types.hpp +++ b/include/libe3/types.hpp @@ -300,6 +300,9 @@ struct XAppControlAction { uint32_t dapp_identifier{0}; uint32_t ran_function_identifier{0}; std::vector xapp_control_data; + // Set by the inbound loop from the wrapping E3-PDU's id field; not + // serialized, so it carries no ASN.1/JSON/Protobuf grammar of its own. + uint32_t message_id{0}; }; /** diff --git a/src/core/e3_agent.cpp b/src/core/e3_agent.cpp index faf3d5c..a7ad271 100644 --- a/src/core/e3_agent.cpp +++ b/src/core/e3_agent.cpp @@ -426,17 +426,20 @@ ErrorCode E3Agent::unsubscribe(uint32_t ran_function_id) { ErrorCode E3Agent::send_control(uint32_t ran_function_id, uint32_t control_id, - std::vector action_data) { + std::vector action_data, + uint32_t* out_message_id) { if (impl_->config.role != E3Role::DAPP) return ErrorCode::STATE_ERROR; if (!impl_->interface || !impl_->interface->is_running()) return ErrorCode::NOT_INITIALIZED; return impl_->interface->queue_dapp_control_action( - ran_function_id, control_id, std::move(action_data)); + ran_function_id, control_id, std::move(action_data), out_message_id); } -ErrorCode E3Agent::send_report(uint32_t ran_function_id, std::vector report_data) { +ErrorCode E3Agent::send_report(uint32_t ran_function_id, + std::vector report_data, + uint32_t* out_message_id) { if (impl_->config.role != E3Role::DAPP) return ErrorCode::STATE_ERROR; if (!impl_->interface || !impl_->interface->is_running()) return ErrorCode::NOT_INITIALIZED; - return impl_->interface->queue_dapp_report(ran_function_id, std::move(report_data)); + return impl_->interface->queue_dapp_report(ran_function_id, std::move(report_data), out_message_id); } ErrorCode E3Agent::release() { diff --git a/src/core/e3_interface.cpp b/src/core/e3_interface.cpp index de8df4e..2c85494 100644 --- a/src/core/e3_interface.cpp +++ b/src/core/e3_interface.cpp @@ -1129,7 +1129,10 @@ void E3Interface::inbound_loop_dapp() { } case PduType::XAPP_CONTROL_ACTION: { auto* a = std::get_if(&pdu.choice); - if (a) handle_xapp_control_action(*a, seq); + if (a) { + a->message_id = pdu.message_id; + handle_xapp_control_action(*a, seq); + } break; } case PduType::MESSAGE_ACK: { @@ -1389,7 +1392,8 @@ ErrorCode E3Interface::queue_subscription_delete(uint32_t ran_function_id) { ErrorCode E3Interface::queue_dapp_control_action( uint32_t ran_function_id, uint32_t control_id, - std::vector action_data + std::vector action_data, + uint32_t* out_message_id ) { if (!dapp_state_) return ErrorCode::STATE_ERROR; auto id = dapp_id(); @@ -1402,18 +1406,24 @@ ErrorCode E3Interface::queue_dapp_control_action( a.control_identifier = control_id; a.action_data = std::move(action_data); pdu.choice = std::move(a); - pdu.message_id = generate_message_id(); + const uint32_t mid = generate_message_id(); + pdu.message_id = mid; // Mirrors E3Agent::send_indication: this is an emit path, so it stamps // EMIT_ENTER itself rather than leaving it to queue_outbound, which only // allocates enqueue_seq for a Pdu that arrives with none. pdu.enqueue_seq = latrec_seq_next(); latrec_tstamp(pdu.enqueue_seq, LATREC_EMIT_ENTER, latrec_ctx(), ran_function_id); - return queue_outbound(std::move(pdu)); + ErrorCode rc = queue_outbound(std::move(pdu)); + if (rc == ErrorCode::SUCCESS && out_message_id) { + *out_message_id = mid; + } + return rc; } ErrorCode E3Interface::queue_dapp_report( uint32_t ran_function_id, - std::vector report_data + std::vector report_data, + uint32_t* out_message_id ) { if (!dapp_state_) return ErrorCode::STATE_ERROR; auto id = dapp_id(); @@ -1425,8 +1435,13 @@ ErrorCode E3Interface::queue_dapp_report( r.ran_function_identifier = ran_function_id; r.report_data = std::move(report_data); pdu.choice = std::move(r); - pdu.message_id = generate_message_id(); - return queue_outbound(std::move(pdu)); + const uint32_t mid = generate_message_id(); + pdu.message_id = mid; + ErrorCode rc = queue_outbound(std::move(pdu)); + if (rc == ErrorCode::SUCCESS && out_message_id) { + *out_message_id = mid; + } + return rc; } ErrorCode E3Interface::queue_release_message() { diff --git a/swig/e3_dapp_session.cpp b/swig/e3_dapp_session.cpp index 7ca3fd1..b6f894d 100644 --- a/swig/e3_dapp_session.cpp +++ b/swig/e3_dapp_session.cpp @@ -83,6 +83,7 @@ DAppSession::DAppSession(libe3::E3Config config, std::size_t queue_capacity) { ev.dapp_id = a.dapp_identifier; ev.ran_function_id = a.ran_function_identifier; ev.payload = a.xapp_control_data; + ev.request_id = a.message_id; impl->enqueue(std::move(ev)); }); @@ -217,13 +218,26 @@ int DAppSession::unsubscribe(uint32_t ran_function_id) { int DAppSession::send_control(uint32_t ran_function_id, uint32_t control_id, std::vector action_data) { - return static_cast( - impl_->agent->send_control(ran_function_id, control_id, std::move(action_data))); + uint32_t message_id = 0; + ErrorCode rc = impl_->agent->send_control( + ran_function_id, control_id, std::move(action_data), &message_id); + // On success return the assigned message id (positive, 1..1000) so Python + // can correlate a later ack/response by id; on failure return the + // ErrorCode (all error codes are negative, SUCCESS is 0). + if (rc == ErrorCode::SUCCESS) { + return static_cast(message_id); + } + return static_cast(rc); } int DAppSession::send_report(uint32_t ran_function_id, std::vector report_data) { - return static_cast( - impl_->agent->send_report(ran_function_id, std::move(report_data))); + uint32_t message_id = 0; + ErrorCode rc = impl_->agent->send_report(ran_function_id, std::move(report_data), &message_id); + // Same success/failure encoding as send_control() above. + if (rc == ErrorCode::SUCCESS) { + return static_cast(message_id); + } + return static_cast(rc); } int DAppSession::send_message_ack(uint32_t request_id, int response_code) { diff --git a/swig/e3_dapp_session.hpp b/swig/e3_dapp_session.hpp index 5b008c8..3c76d3e 100644 --- a/swig/e3_dapp_session.hpp +++ b/swig/e3_dapp_session.hpp @@ -77,7 +77,7 @@ struct E3Event { uint32_t dapp_id{0}; ///< dApp identifier the message targets uint32_t ran_function_id{0}; ///< RAN function (indication / xApp control) uint32_t subscription_id{0}; ///< subscription id (subscription response) - uint32_t request_id{0}; ///< request id (subscription response / ack) + uint32_t request_id{0}; ///< request/message id (subscription response / ack / xApp control) int response_code{-1}; ///< 0=positive, 1=negative, -1=n/a std::vector payload; ///< opaque E3SM bytes (indication / xApp control) uint64_t trace_seq{0}; ///< set when queued; keys the [latrec] LQ-stage records @@ -156,10 +156,20 @@ class DAppSession { /** @brief Delete a previously created subscription. @return ErrorCode as int. */ int unsubscribe(uint32_t ran_function_id); - /** @brief Send a dApp control action to the RAN. @return ErrorCode as int. */ + /** + * @brief Send a dApp control action to the RAN. + * @return the assigned message id (positive, 1..1000) on success, so the + * caller can correlate a later ack/response by id; a negative + * ErrorCode on failure. + */ int send_control(uint32_t ran_function_id, uint32_t control_id, std::vector action_data); - /** @brief Send a dApp report to the RAN. @return ErrorCode as int. */ + /** + * @brief Send a dApp report to the RAN. + * @return the assigned message id (positive, 1..1000) on success, so the + * caller can correlate a later ack/response by id; a negative + * ErrorCode on failure. + */ int send_report(uint32_t ran_function_id, std::vector report_data); /** @brief Acknowledge a request. @param response_code 0=positive,1=negative. @return ErrorCode as int. */ int send_message_ack(uint32_t request_id, int response_code); diff --git a/tests/integration/test_role_pair_zmq_ipc.cpp b/tests/integration/test_role_pair_zmq_ipc.cpp index 190d31e..5df7c8a 100644 --- a/tests/integration/test_role_pair_zmq_ipc.cpp +++ b/tests/integration/test_role_pair_zmq_ipc.cpp @@ -105,6 +105,7 @@ class TestSimpleSM : public ServiceModel { ++control_count; last_control_id = action.control_identifier; last_action_size = action.action_data.size(); + last_request_message_id = request_message_id; Pdu ack = make_message_ack_pdu(request_message_id, ResponseCode::POSITIVE); return emit_outbound(std::move(ack)); } @@ -112,6 +113,7 @@ class TestSimpleSM : public ServiceModel { std::atomic control_count{0}; std::atomic last_control_id{0}; std::atomic last_action_size{0}; + std::atomic last_request_message_id{0}; private: std::atomic running_{false}; @@ -157,6 +159,7 @@ TEST(role_pair_full_handshake_indication_control_report_release) { std::condition_variable cv; int indications = 0; bool sub_resp_ok = false; + std::vector xapp_control_message_ids; E3Agent dapp(make_dapp_config(ep)); dapp.set_indication_handler([&](const IndicationMessage& msg) { @@ -172,6 +175,13 @@ TEST(role_pair_full_handshake_indication_control_report_release) { sub_resp_ok = (r.response_code == ResponseCode::POSITIVE); cv.notify_all(); }); + // #68: the E3-PDU's id, decoded into Pdu::message_id, must be forwarded + // into XAppControlAction::message_id by the inbound loop. + dapp.set_xapp_control_handler([&](const XAppControlAction& a) { + std::lock_guard lk(mu); + xapp_control_message_ids.push_back(a.message_id); + cv.notify_all(); + }); ASSERT_TRUE(dapp.start() == ErrorCode::SUCCESS); @@ -209,7 +219,11 @@ TEST(role_pair_full_handshake_indication_control_report_release) { // Send a control action and verify the SM saw it std::vector ctrl; ASSERT_TRUE(libe3_examples::encode_simple_control(42, ctrl)); - ASSERT_TRUE(dapp.send_control(1, 1, ctrl) == ErrorCode::SUCCESS); + // #67: send_control's out_message_id must be the same id the RAN sees as + // request_message_id in handle_control_action. + uint32_t sent_message_id = 0; + ASSERT_TRUE(dapp.send_control(1, 1, ctrl, &sent_message_id) == ErrorCode::SUCCESS); + ASSERT_TRUE(sent_message_id != 0); // Allow the SM up to 2s to receive and ack the control for (int i = 0; i < 40 && sm_ptr->control_count.load() == 0; ++i) { @@ -217,6 +231,21 @@ TEST(role_pair_full_handshake_indication_control_report_release) { } ASSERT_GE(sm_ptr->control_count.load(), 1); ASSERT_EQ(sm_ptr->last_control_id.load(), 1u); + ASSERT_EQ(sm_ptr->last_request_message_id.load(), sent_message_id); + + // #68: two RAN -> dApp xApp control relays must each carry a distinct, + // nonzero message id through to the dApp's registered handler. + auto dapp_id = dapp.dapp_id(); + ASSERT_TRUE(dapp_id.has_value()); + ASSERT_TRUE(ran.send_xapp_control(*dapp_id, TestSimpleSM::RAN_FUNCTION_ID, ctrl) == ErrorCode::SUCCESS); + ASSERT_TRUE(ran.send_xapp_control(*dapp_id, TestSimpleSM::RAN_FUNCTION_ID, ctrl) == ErrorCode::SUCCESS); + { + std::unique_lock lk(mu); + ASSERT_TRUE(cv.wait_for(lk, 3s, [&]() { return xapp_control_message_ids.size() >= 2; })); + } + ASSERT_TRUE(xapp_control_message_ids[0] != 0); + ASSERT_TRUE(xapp_control_message_ids[1] != 0); + ASSERT_TRUE(xapp_control_message_ids[0] != xapp_control_message_ids[1]); // Release + stop ASSERT_TRUE(dapp.release() == ErrorCode::SUCCESS); From 92130a1273ddd670c5bd265b1fa2daf35e35281f Mon Sep 17 00:00:00 2001 From: Andrea Lacava Date: Sat, 22 Aug 2026 00:13:51 -0400 Subject: [PATCH 2/2] build: bump VERSION to 0.1.2 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Patch bump for the additive/source-compatible API surfaced in the previous commit (#67, #68) — same class of change as 1ad9ced, which bumped 0.1.0 -> 0.1.1 for the analogous subscribe() out_request_id. Jumps to 0.1.2 rather than 0.1.1 because feat/latency-tracing (open, unmerged) already carries VERSION 0.1.1 on its own branch; this keeps the two in sequence instead of colliding whenever that branch lands. Assisted-by: Claude:claude-sonnet-5 Signed-off-by: Andrea Lacava --- VERSION | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/VERSION b/VERSION index 6da28dd..8294c18 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.1.1 \ No newline at end of file +0.1.2 \ No newline at end of file