Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 11 additions & 12 deletions include/libe3/e3_interface.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -218,10 +218,9 @@ class E3Interface {
std::atomic<AgentState> state_{AgentState::UNINITIALIZED};
std::atomic<bool> should_stop_{false};

// Monotonic source for E3-MessageID (1..1000). A wrapping counter gives
// unique ids across the small window of in-flight requests, unlike the
// former random draw which could collide (birthday hazard) and defeat
// request/response correlation.
// Monotonic source for E3-MessageID (1..4294967295). A counter gives
// unique ids, unlike the former random draw which could collide (birthday
// hazard) and defeat request/response correlation.
std::atomic<uint32_t> next_message_id_{0};

// Core components
Expand Down Expand Up @@ -397,22 +396,22 @@ class E3Interface {

public:
/**
* @brief Generate the next message ID, monotonic within 1..1000.
* @brief Generate the next message ID, monotonic within 1..4294967295.
*
* Wraps a lock-free counter into the ASN.1 E3-MessageID (1..1000) range.
* Monotonic (not random) so an id is unique across the in-flight window,
* letting a caller correlate a response by the id it was assigned.
* Wraps a lock-free counter into the ASN.1 E3-MessageID range, skipping 0.
* Monotonic (not random) so an id is unique, letting a caller correlate a
* response by the id it was assigned.
*/
uint32_t generate_message_id();

/**
* @brief Return a request id that can be echoed into a response.
*
* The decoders do not enforce integer range constraints, so a
* peer-supplied message id outside E3-MessageID's 1..1000 can reach
* the handlers; echoing it into a response would fail the encode.
* In-range ids pass through, anything else is replaced with a fresh
* generated id (dApps do not correlate setup replies by ID).
* peer-supplied message id of 0, outside E3-MessageID's 1..4294967295,
* can reach the handlers; echoing it into a response would fail the
* encode. Every non-zero id passes through unchanged, 0 is replaced
* with a fresh generated id.
*/
uint32_t sanitize_request_message_id(uint32_t request_id);
};
Expand Down
6 changes: 0 additions & 6 deletions include/libe3/subscription_manager.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -247,12 +247,6 @@ class SubscriptionManager {
// Callback for SM lifecycle events
SmLifecycleCallback sm_lifecycle_callback_;

// Next dApp ID to assign (1-100 per spec)
uint32_t next_dapp_id_{1};

// Next subscription ID to assign
uint32_t next_subscription_id_{1};

// Helper to create composite key for subscription
static uint64_t make_sub_key(uint32_t dapp_id, uint32_t ran_func_id) {
return (static_cast<uint64_t>(dapp_id) << 32) | ran_func_id;
Expand Down
18 changes: 13 additions & 5 deletions src/core/e3_interface.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -146,7 +146,9 @@ void E3Interface::forget_subscription_op(uint32_t request_id) {
}

uint32_t E3Interface::sanitize_request_message_id(uint32_t request_id) {
if (request_id >= 1 && request_id <= 1000) {
// E3-MessageID is 1..4294967295: every non-zero id can be echoed, a peer
// that counts across message types (Aerial) passes 1000 within seconds.
if (request_id != 0) {
return request_id;
}
return generate_message_id();
Expand Down Expand Up @@ -796,7 +798,7 @@ void E3Interface::handle_setup_request(const SetupRequest& request, uint32_t req
response_code,
config_.e3ap_version,
// A failed registration assigns no id; engaging the optional with 0
// would violate E3-DAppID (1..100) at encode.
// would violate E3-DAppID (1..65535) at encode.
response_code == ResponseCode::POSITIVE
? std::optional<uint32_t>(assigned_dapp_id)
: std::nullopt,
Expand Down Expand Up @@ -828,9 +830,8 @@ void E3Interface::send_negative_setup_reply(uint32_t request_id) {
// (ranFunctionList included), so with the fixed, mutually known
// encoding it always encodes. The mandatory fields have to satisfy
// their schema constraints, though: requestId is E3-MessageID
// INTEGER (1..1000) — an id that cannot be echoed (0 from undecodable
// bytes, or out of range from a peer the decoder did not constrain) is
// replaced with a freshly generated one — and ranIdentifier is
// INTEGER (1..4294967295) — an id that cannot be echoed (0, from
// undecodable bytes) is replaced with a freshly generated one — and ranIdentifier is
// SIZE (1..64), hence the placeholder when the config left it empty.
uint32_t message_id = generate_message_id();
request_id = sanitize_request_message_id(request_id);
Expand Down Expand Up @@ -1451,6 +1452,13 @@ ErrorCode E3Interface::queue_dapp_report(
uint32_t sequence_id,
uint32_t* out_message_id
) {
// E3-SequenceID is 1..4294967295 and mandatory on a report: with 0 every
// encode fails, so refuse it here where the caller sees it.
if (sequence_id == 0) {
E3_LOG_ERROR(LOG_TAG) << "dApp report refused: sequence_id must be non-zero (E3-SequenceID is 1..4294967295)";
return ErrorCode::INVALID_PARAM;
}

if (!dapp_state_) return ErrorCode::STATE_ERROR;
auto id = dapp_id();
if (!id) return ErrorCode::NOT_INITIALIZED;
Expand Down
50 changes: 23 additions & 27 deletions src/core/subscription_manager.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -15,8 +15,18 @@ namespace libe3 {

namespace {
constexpr const char* LOG_TAG = "SubMgr";
constexpr uint32_t MAX_DAPP_ID = 100;
constexpr uint32_t MIN_DAPP_ID = 1;
// E3-DAppID and the subscription id are both INTEGER (1..65535).
constexpr uint32_t MIN_ID = 1;
constexpr uint32_t MAX_ID = 65535;

// Lowest id in MIN_ID..MAX_ID that `taken` does not hold, 0 when all are in use.
template <typename Map>
uint32_t lowest_free_id(const Map& taken) {
for (uint32_t id = MIN_ID; id <= MAX_ID; ++id) {
if (taken.count(id) == 0) return id;
}
return 0;
}
}

SubscriptionManager::SubscriptionManager() {
Expand Down Expand Up @@ -45,22 +55,12 @@ void SubscriptionManager::set_sm_lifecycle_callback(SmLifecycleCallback callback
std::pair<ErrorCode, uint32_t> SubscriptionManager::register_dapp() {
std::unique_lock lock(mutex_);

// Find the next available dApp ID
uint32_t assigned_id = next_dapp_id_;
uint32_t attempts = 0;

// Search for an available ID (handle wrap-around and gaps from unregistered dApps)
while (registered_dapps_.count(assigned_id) > 0 && attempts <= MAX_DAPP_ID) {
assigned_id++;
if (assigned_id > MAX_DAPP_ID) {
assigned_id = MIN_DAPP_ID;
}
attempts++;
}

if (attempts > MAX_DAPP_ID) {
E3_LOG_ERROR(LOG_TAG) << "No available dApp IDs (range " << MIN_DAPP_ID
<< ".." << MAX_DAPP_ID << ")";
// Hand out the lowest free id, so a dApp that reconnects gets the same id back
// and a live id is never reused.
const uint32_t assigned_id = lowest_free_id(registered_dapps_);
if (assigned_id == 0) {
E3_LOG_ERROR(LOG_TAG) << "No available dApp IDs (range " << MIN_ID
<< ".." << MAX_ID << ")";
return {ErrorCode::INTERNAL_ERROR, 0};
}

Expand All @@ -72,12 +72,6 @@ std::pair<ErrorCode, uint32_t> SubscriptionManager::register_dapp() {
// Initialize empty subscription set for this dApp
dapp_subscriptions_[assigned_id] = {};

// Update next_dapp_id_ for the next registration
next_dapp_id_ = assigned_id + 1;
if (next_dapp_id_ > MAX_DAPP_ID) {
next_dapp_id_ = MIN_DAPP_ID;
}

E3_LOG_INFO(LOG_TAG) << "dApp registered successfully with ID " << assigned_id;
return {ErrorCode::SUCCESS, assigned_id};
}
Expand Down Expand Up @@ -186,9 +180,11 @@ std::pair<ErrorCode, uint32_t> SubscriptionManager::add_subscription(
bool had_subscribers = !ran_function_subscribers_[ran_function_id].empty();

// Assign subscription ID
uint32_t subscription_id = next_subscription_id_++;
if (next_subscription_id_ > 100) {
next_subscription_id_ = 1; // Wrap around per spec (0-100 range)
const uint32_t subscription_id = lowest_free_id(subscription_ids_);
if (subscription_id == 0) {
E3_LOG_ERROR(LOG_TAG) << "No available subscription IDs (range " << MIN_ID
<< ".." << MAX_ID << ")";
return {ErrorCode::INTERNAL_ERROR, 0};
}

// Add subscription
Expand Down
16 changes: 8 additions & 8 deletions swig/e3_dapp_session.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -196,22 +196,22 @@ std::vector<uint8_t> DAppSession::setup_ran_function_data(std::size_t i) const {
return impl_->setup.ran_function_list[i].ran_function_data;
}

int DAppSession::subscribe(uint32_t ran_function_id,
std::vector<uint32_t> telemetry_ids,
std::vector<uint32_t> control_ids,
int sub_time_ms,
int periodicity) {
int64_t DAppSession::subscribe(uint32_t ran_function_id,
std::vector<uint32_t> telemetry_ids,
std::vector<uint32_t> control_ids,
int sub_time_ms,
int periodicity) {
uint32_t request_id = 0;
ErrorCode rc = impl_->agent->subscribe(
ran_function_id, std::move(telemetry_ids), std::move(control_ids),
opt_u32(sub_time_ms), opt_u32(periodicity), &request_id);
// On success return the assigned request id (positive, 1..1000) so Python
// On success return the assigned request id (positive, 1..4294967295) so Python
// can correlate the SubscriptionResponse by id; on failure return the
// ErrorCode (all error codes are negative, SUCCESS is 0).
if (rc == ErrorCode::SUCCESS) {
return static_cast<int>(request_id);
return static_cast<int64_t>(request_id);
}
return static_cast<int>(rc);
return static_cast<int64_t>(rc);
}

int DAppSession::unsubscribe(uint32_t ran_function_id) {
Expand Down
11 changes: 6 additions & 5 deletions swig/e3_dapp_session.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -146,11 +146,11 @@ class DAppSession {
* @brief Subscribe to a RAN function (after a positive setup).
* @param sub_time_ms subscription time, or -1 for unset.
* @param periodicity delivery periodicity (µs), or -1 for unset.
* @return the assigned request id (positive, 1..1000) on success, so the
* caller can correlate the SubscriptionResponse by its request_id;
* a negative ErrorCode on failure.
* @return the assigned request id (positive, 1..4294967295) on success, so
* the caller can correlate the SubscriptionResponse by its
* request_id; a negative ErrorCode on failure.
*/
int subscribe(uint32_t ran_function_id,
int64_t subscribe(uint32_t ran_function_id,
std::vector<uint32_t> telemetry_ids,
std::vector<uint32_t> control_ids,
int sub_time_ms,
Expand All @@ -174,7 +174,8 @@ class DAppSession {
/**
* @brief Send a dApp report to the RAN.
* @param sequence_id correlation id this dApp assigns to the detection;
* mandatory on the wire.
* mandatory on the wire, 1..4294967295 (0 is refused with
* INVALID_PARAM).
* @return the assigned message id (positive, 1..4294967295) on success, so
* the caller can correlate a later ack/response by id; a negative
* ErrorCode on failure.
Expand Down
2 changes: 1 addition & 1 deletion tests/test_asn1_size.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -143,7 +143,7 @@ TEST(Asn1Size_IndicationMessage_largePayload) {
auto payload = make_payload(8192);

Pdu pdu(PduType::INDICATION_MESSAGE);
pdu.message_id = 234; // E3-MessageID is constrained to INTEGER (1..1000)
pdu.message_id = 234; // E3-MessageID is constrained to INTEGER (1..4294967295)
IndicationMessage msg;
msg.dapp_identifier = 1;
msg.ran_function_identifier = 1;
Expand Down
16 changes: 16 additions & 0 deletions tests/test_e3_interface.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -214,6 +214,22 @@ TEST(DAppStatusChanged_no_spurious_calls) {
ASSERT_EQ(count, 0);
}

/**
* E3-SequenceID is 1..4294967295 and mandatory on a report; 0 would fail every
* encode, so the API refuses it. A control may carry 0 (the field is optional).
*/
TEST(DAppReport_sequence_id_zero_is_refused) {
E3Config cfg = make_test_config();
cfg.role = E3Role::DAPP;
E3Interface iface(cfg);
ASSERT_EQ(error_to_int(iface.init()), error_to_int(ErrorCode::SUCCESS));

std::vector<uint8_t> data{1, 2, 3};
ASSERT_EQ(error_to_int(iface.queue_dapp_report(1, data, 0)), error_to_int(ErrorCode::INVALID_PARAM));
ASSERT_NE(error_to_int(iface.queue_dapp_report(1, data, 1)), error_to_int(ErrorCode::INVALID_PARAM));
ASSERT_NE(error_to_int(iface.queue_dapp_control_action(1, 1, data, 0)), error_to_int(ErrorCode::INVALID_PARAM));
}

int main() {
return RUN_ALL_TESTS();
}
4 changes: 2 additions & 2 deletions tests/test_encoder_equivalence.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -212,8 +212,8 @@ struct Sample {
};

/// Every field set to a value distinguishable from its default, and every
/// value inside the ASN.1 constraints: E3-DAppID and the RAN-function and
/// control identifiers are 1..100, E3-MessageID is 1..1000, E3-Version is at
/// value inside the ASN.1 constraints: E3-DAppID is 1..65535, the RAN-function and
/// control identifiers are 1..65535, E3-MessageID is 1..4294967295, E3-Version is at
/// most 11 characters, octet strings are 1..32768 bytes.
static std::vector<Sample> sample_pdus() {
std::vector<Sample> samples;
Expand Down
25 changes: 9 additions & 16 deletions tests/test_setup_bad_request.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -164,19 +164,14 @@ TEST(SetupChannel_garbageRequest_repliesAndChannelSurvives) {

#if defined(LIBE3_ENABLE_JSON)
/**
* A well-formed SetupRequest whose message id lies outside E3-MessageID's
* 1..1000 must still receive a decodable positive SetupResponse: the agent
* substitutes an in-range id instead of failing the response encode.
*
* JSON-only: the JSON codec is the one that lets an out-of-range id through
* to the handler (the APER encoder rejects it on the sender side, but the
* APER decoder does not range-check either, so the substitution protects
* both encodings).
* A SetupRequest whose message id is far above 1000 (a peer that counts across
* message types, like Aerial, gets there within seconds) must get a positive
* SetupResponse that echoes that id unchanged: E3-MessageID is 1..4294967295.
*/
TEST(SetupChannel_outOfRangeRequestId_substitutedAndAnswered) {
const std::string setup_ep = unique_ipc("setup_oor");
const std::string sub_ep = unique_ipc("inbound_oor");
const std::string pub_ep = unique_ipc("outbound_oor");
TEST(SetupChannel_largeRequestId_isEchoed) {
const std::string setup_ep = unique_ipc("setup_big");
const std::string sub_ep = unique_ipc("inbound_big");
const std::string pub_ep = unique_ipc("outbound_big");

E3Config cfg;
cfg.role = E3Role::RAN;
Expand Down Expand Up @@ -209,7 +204,7 @@ TEST(SetupChannel_outOfRangeRequestId_substitutedAndAnswered) {
ASSERT_TRUE(encoder != nullptr);

auto setup = encoder->encode_setup_request(
4096, "1.0.0", "badsetup-dapp", "0.0.1", "test-vendor");
3000000000u, "1.0.0", "badsetup-dapp", "0.0.1", "test-vendor");
ASSERT_TRUE(setup.has_value());
ASSERT_GE(zmq_send(req, setup->buffer.data(), setup->buffer.size(), 0), 0);

Expand All @@ -225,9 +220,7 @@ TEST(SetupChannel_outOfRangeRequestId_substitutedAndAnswered) {
ASSERT_EQ(static_cast<int>(resp->response_code),
static_cast<int>(ResponseCode::POSITIVE));
ASSERT_TRUE(resp->dapp_identifier.has_value());
// The unencodable id was substituted with one inside E3-MessageID.
ASSERT_GE(resp->request_id, 1u);
ASSERT_LE(resp->request_id, 1000u);
ASSERT_EQ(resp->request_id, 3000000000u);

zmq_close(req);
zmq_ctx_destroy(ctx);
Expand Down
47 changes: 47 additions & 0 deletions tests/test_subscription_manager.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@

#include "test_framework.hpp"
#include "libe3/subscription_manager.hpp"
#include <algorithm>
#include <thread>
#include <atomic>

Expand Down Expand Up @@ -47,6 +48,52 @@ TEST(SubscriptionManager_register_multiple_dapps_assigns_unique_ids) {
ASSERT_NE(id1, id3);
}

TEST(SubscriptionManager_dapp_ids_are_the_lowest_free_ones) {
SubscriptionManager mgr;
auto [r1, id1] = mgr.register_dapp();
auto [r2, id2] = mgr.register_dapp();
auto [r3, id3] = mgr.register_dapp();
ASSERT_EQ(id1, 1u);
ASSERT_EQ(id2, 2u);
ASSERT_EQ(id3, 3u);

// A dApp that disconnects and reconnects gets its id back, however often it does.
for (int i = 0; i < 5; ++i) {
ASSERT_TRUE(mgr.unregister_dapp(2) == ErrorCode::SUCCESS);
auto [r, id] = mgr.register_dapp();
ASSERT_TRUE(r == ErrorCode::SUCCESS);
ASSERT_EQ(id, 2u);
}
}

TEST(SubscriptionManager_serves_more_than_100_dapps_with_distinct_ids) {
SubscriptionManager mgr;
std::unordered_set<uint32_t> dapp_ids, sub_ids;
for (int i = 0; i < 300; ++i) {
auto [r, dapp_id] = mgr.register_dapp();
ASSERT_TRUE(r == ErrorCode::SUCCESS);
ASSERT_TRUE(dapp_ids.insert(dapp_id).second);
auto [sr, sub_id] = mgr.add_subscription(dapp_id, 1);
ASSERT_TRUE(sr == ErrorCode::SUCCESS);
ASSERT_TRUE(sub_ids.insert(sub_id).second); // a live subscription id is never handed out again
}
ASSERT_EQ(mgr.dapp_count(), 300u);
ASSERT_EQ(*std::max_element(dapp_ids.begin(), dapp_ids.end()), 300u);
ASSERT_EQ(*std::max_element(sub_ids.begin(), sub_ids.end()), 300u);
}

TEST(SubscriptionManager_subscription_id_of_a_deleted_subscription_is_reused) {
SubscriptionManager mgr;
auto [r, dapp_id] = mgr.register_dapp();
auto [r1, s1] = mgr.add_subscription(dapp_id, 1);
auto [r2, s2] = mgr.add_subscription(dapp_id, 2);
ASSERT_EQ(s1, 1u);
ASSERT_EQ(s2, 2u);
ASSERT_TRUE(mgr.remove_subscription(dapp_id, 1) == ErrorCode::SUCCESS);
auto [r3, s3] = mgr.add_subscription(dapp_id, 3);
ASSERT_EQ(s3, 1u);
}

TEST(SubscriptionManager_unregister_dapp) {
SubscriptionManager mgr;

Expand Down
Loading