e3ap: surface assigned message ids for dApp sends and relayed xApp control

VERSION

@@ -1 +1 @@
-0.1.1
+0.1.2

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<uint8_t> action_data);
+                           std::vector<uint8_t> 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<uint8_t> 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<uint8_t> report_data,
+                          uint32_t* out_message_id = nullptr);
 
     /** Send a ReleaseMessage. The dApp stays running until stop() is called. */
     ErrorCode release();

include/libe3/e3_interface.hpp

@@ -189,11 +189,16 @@ class E3Interface {
                                          std::optional<uint32_t> 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<uint8_t> action_data);
+                                        std::vector<uint8_t> action_data,
+                                        uint32_t* out_message_id = nullptr);
     ErrorCode queue_dapp_report(uint32_t ran_function_id,
-                                std::vector<uint8_t> report_data);
+                                std::vector<uint8_t> report_data,
+                                uint32_t* out_message_id = nullptr);
     ErrorCode queue_release_message();
 
     // Block until the setup handshake completes (or times out / fails).

include/libe3/types.hpp

@@ -300,6 +300,9 @@ struct XAppControlAction {
     uint32_t dapp_identifier{0};
     uint32_t ran_function_identifier{0};
     std::vector<uint8_t> 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};
 };
 
 /**

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<uint8_t> action_data) {
+                                std::vector<uint8_t> 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<uint8_t> report_data) {
+ErrorCode E3Agent::send_report(uint32_t ran_function_id,
+                               std::vector<uint8_t> 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() {

src/core/e3_interface.cpp

@@ -1129,7 +1129,10 @@ void E3Interface::inbound_loop_dapp() {
             }
             case PduType::XAPP_CONTROL_ACTION: {
                 auto* a = std::get_if<XAppControlAction>(&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<uint8_t> action_data
+    std::vector<uint8_t> 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<uint8_t> report_data
+    std::vector<uint8_t> 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() {

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<uint8_t> action_data) {
-    return static_cast<int>(
-        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<int>(message_id);
+    }
+    return static_cast<int>(rc);
 }
 
 int DAppSession::send_report(uint32_t ran_function_id, std::vector<uint8_t> report_data) {
-    return static_cast<int>(
-        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<int>(message_id);
+    }
+    return static_cast<int>(rc);
 }
 
 int DAppSession::send_message_ack(uint32_t request_id, int response_code) {

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<uint8_t> 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<uint8_t> 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<uint8_t> 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);

tests/integration/test_role_pair_zmq_ipc.cpp

@@ -105,13 +105,15 @@ 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));
     }
 
     std::atomic<int> control_count{0};
     std::atomic<uint32_t> last_control_id{0};
     std::atomic<size_t> last_action_size{0};
+    std::atomic<uint32_t> last_request_message_id{0};
 
 private:
     std::atomic<bool> 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<uint32_t> 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<std::mutex> lk(mu);
+        xapp_control_message_ids.push_back(a.message_id);
+        cv.notify_all();
+    });
 
     ASSERT_TRUE(dapp.start() == ErrorCode::SUCCESS);
 
@@ -209,14 +219,33 @@ TEST(role_pair_full_handshake_indication_control_report_release) {
     // Send a control action and verify the SM saw it
     std::vector<uint8_t> 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) {
         std::this_thread::sleep_for(50ms);
     }
     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<std::mutex> 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);