diff --git a/include/canbus/isotp_fast.h b/include/canbus/isotp_fast.h index 8dad86e9..17c05ba4 100644 --- a/include/canbus/isotp_fast.h +++ b/include/canbus/isotp_fast.h @@ -113,6 +113,14 @@ struct isotp_fast_opts uint8_t stmin; uint8_t flags; enum isotp_fast_addressing_mode addressing_mode; + /** + * Optional override for the RX filter mask installed by + * @ref isotp_fast_bind. 0 means the default mask for the addressing + * mode is used. Use this e.g. to additionally match the sender address + * so that concurrent contexts bound to different peers on one interface + * do not receive each other's traffic + */ + uint32_t rx_mask; }; /** diff --git a/include/thingset++/can/zephyr/ThingSetZephyrCanInterface.hpp b/include/thingset++/can/zephyr/ThingSetZephyrCanInterface.hpp index ab22ec7d..18716939 100644 --- a/include/thingset++/can/zephyr/ThingSetZephyrCanInterface.hpp +++ b/include/thingset++/can/zephyr/ThingSetZephyrCanInterface.hpp @@ -49,6 +49,7 @@ class ThingSetZephyrCanInterface : public _ThingSetZephyrCanInterface AddressClaimWorkItem _addressClaimWork; k_event _events; + k_mutex _bindLock; int _claimFilterId; int _discoverFilterId; diff --git a/include/thingset++/can/zephyr/ThingSetZephyrCanRequestResponseContext.hpp b/include/thingset++/can/zephyr/ThingSetZephyrCanRequestResponseContext.hpp index a906c332..a2578a57 100644 --- a/include/thingset++/can/zephyr/ThingSetZephyrCanRequestResponseContext.hpp +++ b/include/thingset++/can/zephyr/ThingSetZephyrCanRequestResponseContext.hpp @@ -52,6 +52,7 @@ class ThingSetZephyrCanRequestResponseContext { static void onRequestResponseReceived(net_buf *buffer, int remainingLength, isotp_fast_addr address, void *arg); void onRequestResponseReceived(net_buf *buffer, int remainingLength, isotp_fast_addr address); static const isotp_fast_opts flowControlOptions; + static const isotp_fast_opts peerFlowControlOptions; }; } // namespace ThingSet::Can::Zephyr \ No newline at end of file diff --git a/src/ThingSetClient.cpp b/src/ThingSetClient.cpp index d2ecd7fa..4eb6fcd8 100644 --- a/src/ThingSetClient.cpp +++ b/src/ThingSetClient.cpp @@ -9,6 +9,9 @@ namespace ThingSet { +static constexpr uint8_t cborNull = 0xF6; +static constexpr size_t responseHeaderSize = 2; /* status code + CBOR null */ + ThingSetClient::ThingSetClient(ThingSetClientTransport &transport, uint8_t *rxBuffer, size_t rxBufferSize, uint8_t *txBuffer, size_t txBufferSize) : _transport(transport), _rxBuffer(rxBuffer), _rxBufferSize(rxBufferSize), _txBuffer(txBuffer), @@ -22,13 +25,17 @@ bool ThingSetClient::connect() ThingSetResult ThingSetClient::read(uint8_t **responseBuffer, size_t &responseSize) { - responseSize = _transport.read(_rxBuffer, _rxBufferSize); - if (responseSize == 0) { - return ThingSetResult(ThingSetStatusCode::internalServerError); + responseSize = 0; + + int received = _transport.read(_rxBuffer, _rxBufferSize); + if (received <= 0) { + // No response (0) or a transport error such as a receive timeout + // (negative errno). The rx buffer may still hold a previous response + return ThingSetResult(ThingSetStatusCode::gatewayTimeout); } #ifdef DEBUG_LOGGING - for (size_t i = 0; i < responseSize; i++) + for (int i = 0; i < received; i++) { if (i > 0 && i % 16 == 0) { printf("\n"); @@ -43,14 +50,14 @@ ThingSetResult ThingSetClient::read(uint8_t **responseBuffer, size_t &responseSi return result; } - // first value is always a CBOR null - if (_rxBuffer[1] != 0xF6) { + // a successful response carries at least the status code plus a CBOR null + if ((size_t)received < responseHeaderSize || _rxBuffer[1] != cborNull) { return ThingSetResult(ThingSetStatusCode::internalServerError); } // return size having accounted for response code and null - responseSize -= 2; - *responseBuffer = &_rxBuffer[2]; + responseSize = (size_t)received - responseHeaderSize; + *responseBuffer = &_rxBuffer[responseHeaderSize]; return result; } diff --git a/src/can/zephyr/ThingSetZephyrCanInterface.cpp b/src/can/zephyr/ThingSetZephyrCanInterface.cpp index 4d8c6140..e3207162 100644 --- a/src/can/zephyr/ThingSetZephyrCanInterface.cpp +++ b/src/can/zephyr/ThingSetZephyrCanInterface.cpp @@ -50,6 +50,7 @@ ThingSetZephyrCanInterface::ThingSetZephyrCanInterface(const device *const canDe { k_work_init(&_addressClaimWork.work, addressClaimWorkHandler); _addressClaimWork.instance = this; + k_mutex_init(&_bindLock); } ThingSetZephyrCanInterface::~ThingSetZephyrCanInterface() @@ -141,8 +142,27 @@ int ThingSetZephyrCanInterface::addFilter(CanID &canId, void (*callback)(const d return can_add_rx_filter(_canDevice, callback, this, &filter); } +namespace { +struct MutexGuard +{ + k_mutex &_mutex; + + explicit MutexGuard(k_mutex &mutex) : _mutex(mutex) + { + k_mutex_lock(&_mutex, K_FOREVER); + } + + ~MutexGuard() + { + k_mutex_unlock(&_mutex); + } +}; +} // namespace + bool ThingSetZephyrCanInterface::bind(uint8_t nodeAddress) { + MutexGuard guard(_bindLock); + if (_nodeAddress == CanID::broadcastAddress) { _nodeAddress = nodeAddress; LOG_INFO("Starting address claim for CAN interface %s", _canDevice->name); diff --git a/src/can/zephyr/ThingSetZephyrCanRequestResponseContext.cpp b/src/can/zephyr/ThingSetZephyrCanRequestResponseContext.cpp index 6c6acb52..76d70afc 100644 --- a/src/can/zephyr/ThingSetZephyrCanRequestResponseContext.cpp +++ b/src/can/zephyr/ThingSetZephyrCanRequestResponseContext.cpp @@ -35,6 +35,16 @@ const isotp_fast_opts ThingSetZephyrCanRequestResponseContext::flowControlOption .addressing_mode = ISOTP_FAST_ADDRESSING_MODE_FIXED, }; +const isotp_fast_opts ThingSetZephyrCanRequestResponseContext::peerFlowControlOptions = { + .bs = 8, + .stmin = CONFIG_THINGSET_PLUS_PLUS_CAN_FRAME_SEPARATION_TIME, +#ifdef CONFIG_CAN_FD_MODE + .flags = ISOTP_MSG_FDF, +#endif + .addressing_mode = ISOTP_FAST_ADDRESSING_MODE_FIXED, + .rx_mask = ISOTP_FIXED_ADDR_RX_MASK | ISOTP_FIXED_ADDR_SA_MASK, +}; + static void onRequestResponseError(int8_t error, isotp_fast_addr addr, void *arg); static void onRequestResponseSent(int result, isotp_fast_addr addr, void *arg); @@ -68,13 +78,22 @@ bool ThingSetZephyrCanRequestResponseContext::bind(uint8_t otherNodeAddress, std .setMessageType(MessageType::requestResponse) .setMessagePriority(MessagePriority::channel) .setTarget(_canInterface.getNodeAddress()); + const isotp_fast_opts *options = &ThingSetZephyrCanRequestResponseContext::flowControlOptions; if (otherNodeAddress != CanID::broadcastAddress) { canId.setSource(otherNodeAddress); + /* bound to one peer: only accept traffic from that peer */ + options = &ThingSetZephyrCanRequestResponseContext::peerFlowControlOptions; } _inboundRequestCallback = callback; - return isotp_fast_bind(&_requestResponseContext, _canInterface.getDevice(), IsoTpFastAddress(canId), - &ThingSetZephyrCanRequestResponseContext::flowControlOptions, onRequestResponseReceived, - this, onRequestResponseError, onRequestResponseSent) == 0; + int result = isotp_fast_bind(&_requestResponseContext, _canInterface.getDevice(), IsoTpFastAddress(canId), + options, onRequestResponseReceived, this, onRequestResponseError, + onRequestResponseSent); + if (result != 0) { + LOG_ERROR("Failed to bind request/response context for node 0x%x (err %d)", otherNodeAddress, result); + _requestResponseContext.filter_id = THINGSET_PLUS_PLUS_ZEPHYR_CAN_FILTER_ID_NONE; + return false; + } + return true; } bool ThingSetZephyrCanRequestResponseContext::send(const uint8_t otherNodeAddress, uint8_t *buffer, size_t len) diff --git a/src/can/zephyr/isotp/isotp_fast.c b/src/can/zephyr/isotp/isotp_fast.c index 314eb5da..70c1adf4 100644 --- a/src/can/zephyr/isotp/isotp_fast.c +++ b/src/can/zephyr/isotp/isotp_fast.c @@ -85,6 +85,7 @@ static int get_send_ctx(struct isotp_fast_ctx *ctx, struct isotp_fast_addr tx_ad context->stmin = ctx->opts->stmin; context->state = ISOTP_TX_SEND_FF; context->error = 0; + atomic_clear(&context->pending_cb); k_sem_init(&context->sem, 0, 1); k_work_init(&context->work, send_work_handler); k_timer_init(&context->timer, send_timeout_handler, NULL); @@ -173,6 +174,7 @@ static int get_recv_ctx(struct isotp_fast_ctx *ctx, struct isotp_fast_addr rx_ad context->state = ISOTP_RX_STATE_WAIT_FF_SF; context->rx_addr = rx_addr; context->error = 0; + atomic_clear(&context->pending_cb); #ifdef ISOTP_FAST_RECEIVE_QUEUE k_msgq_init(&context->recv_queue, context->recv_queue_pool, sizeof(struct net_buf *), CONFIG_ISOTP_FAST_RX_MAX_PACKET_COUNT); @@ -276,6 +278,8 @@ static void receive_can_tx(const struct device *dev, int error, void *arg) ARG_UNUSED(dev); + atomic_dec(&rctx->pending_cb); + if (error != 0) { LOG_ERR("Error sending FC frame (%d)", error); receive_report_error(rctx, ISOTP_N_ERROR); @@ -304,8 +308,11 @@ static void receive_send_fc(struct isotp_fast_recv_ctx *rctx, uint8_t fs) payload_len = data - frame.data; frame.dlc = can_bytes_to_dlc(payload_len); + atomic_inc(&rctx->pending_cb); ret = can_send(rctx->ctx->can_dev, &frame, K_MSEC(ISOTP_A_TIMEOUT_MS), receive_can_tx, rctx); if (ret) { + /* the completion callback never fires for a frame that was not queued */ + atomic_dec(&rctx->pending_cb); LOG_ERR("Can't send FC, (%d)", ret); receive_report_error(rctx, ISOTP_N_TIMEOUT_A); receive_state_machine(rctx); @@ -738,6 +745,7 @@ static void send_can_tx_callback(const struct device *dev, int error, void *arg) ARG_UNUSED(dev); + atomic_dec(&sctx->pending_cb); sctx->backlog--; k_sem_give(&sctx->sem); @@ -783,8 +791,13 @@ static inline int send_ff(struct isotp_fast_send_ctx *sctx) sctx->rem_len -= size; sctx->data += size; frame.dlc = can_bytes_to_dlc(CAN_MAX_DLEN); + atomic_inc(&sctx->pending_cb); ret = can_send(sctx->ctx->can_dev, &frame, K_MSEC(ISOTP_A_TIMEOUT_MS), send_can_tx_callback, sctx); + if (ret != 0) { + /* the completion callback never fires for a frame that was not queued */ + atomic_dec(&sctx->pending_cb); + } return ret; } @@ -806,6 +819,7 @@ static inline int send_cf(struct isotp_fast_send_ctx *sctx) sctx->data += len; frame.dlc = can_bytes_to_dlc(len + index); + atomic_inc(&sctx->pending_cb); ret = can_send(sctx->ctx->can_dev, &frame, K_MSEC(ISOTP_A_TIMEOUT_MS), send_can_tx_callback, sctx); if (ret == 0) { @@ -813,6 +827,10 @@ static inline int send_cf(struct isotp_fast_send_ctx *sctx) sctx->bs--; sctx->backlog++; } + else { + /* the completion callback never fires for a frame that was not queued */ + atomic_dec(&sctx->pending_cb); + } ret = ret ? ret : sctx->rem_len; return ret; @@ -942,7 +960,7 @@ static inline void prepare_filter(struct can_filter *filter, uint32_t rx_addr, #endif } #endif /* CONFIG_ISOTP_FAST_CUSTOM_ADDRESSING */ - filter->mask = mask; + filter->mask = opts->rx_mask != 0 ? opts->rx_mask : mask; filter->flags = CAN_FILTER_IDE; } @@ -969,6 +987,13 @@ int isotp_fast_bind(struct isotp_fast_ctx *ctx, const struct device *can_dev, struct can_filter filter; prepare_filter(&filter, rx_addr.ext_id, opts); ctx->filter_id = can_add_rx_filter(ctx->can_dev, can_rx_callback, ctx, &filter); + if (ctx->filter_id < 0) { + /* Without this check a full filter table used to be reported as a + * successful bind, and every subsequent exchange timed out */ + LOG_ERR("Failed to add RX filter for %x:%x (err %d)", filter.id, filter.mask, + ctx->filter_id); + return ISOTP_NO_FREE_FILTER; + } LOG_DBG("Successfully bound to %x:%x", filter.id, filter.mask); @@ -986,10 +1011,73 @@ static void free_recv_await_ctx(struct isotp_fast_ctx *ctx, struct isotp_fast_re } #endif +static void orphan_sent_callback(int result, struct isotp_fast_addr addr, void *arg) +{ + ARG_UNUSED(result); + ARG_UNUSED(addr); + ARG_UNUSED(arg); +} + +static void orphan_recv_callback(struct net_buf *buffer, int rem_len, struct isotp_fast_addr addr, + void *arg) +{ + ARG_UNUSED(buffer); + ARG_UNUSED(rem_len); + ARG_UNUSED(addr); + ARG_UNUSED(arg); +} + +static const struct isotp_fast_opts orphanage_opts = { + .bs = 0, + .stmin = 0, + .flags = 0, +}; + +static struct isotp_fast_ctx orphanage = { + .isotp_send_ctx_list = SYS_SLIST_STATIC_INIT(&orphanage.isotp_send_ctx_list), + .isotp_recv_ctx_list = SYS_SLIST_STATIC_INIT(&orphanage.isotp_recv_ctx_list), + .can_dev = NULL, + .filter_id = -1, + .opts = &orphanage_opts, + .recv_callback = orphan_recv_callback, + .recv_cb_arg = NULL, + .recv_error_callback = NULL, + .sent_callback = orphan_sent_callback, +}; + +static void orphan_send_ctx(struct isotp_fast_ctx *owner, struct isotp_fast_send_ctx *sctx) +{ + LOG_WRN("Orphaning send context %x (outstanding CAN callback)", sctx->tx_addr.ext_id); + sys_slist_find_and_remove(&owner->isotp_send_ctx_list, &sctx->node); + /* any later work run takes the TX_ERR path, which frees the context */ + send_report_error(sctx, ISOTP_N_ERROR); + sctx->cb_arg = NULL; + sctx->ctx = &orphanage; + sys_slist_append(&orphanage.isotp_send_ctx_list, &sctx->node); +} + +static void orphan_recv_ctx(struct isotp_fast_ctx *owner, struct isotp_fast_recv_ctx *rctx) +{ + LOG_WRN("Orphaning receive context %x (outstanding CAN callback)", rctx->rx_addr.ext_id); + sys_slist_find_and_remove(&owner->isotp_recv_ctx_list, &rctx->node); + /* no state machine case handles UNBOUND: the context is inert */ + rctx->state = ISOTP_RX_STATE_UNBOUND; + rctx->ctx = &orphanage; + sys_slist_append(&orphanage.isotp_recv_ctx_list, &rctx->node); +} + +/* + * Must be called from thread context, and not from the system work queue: + * it cancels the contexts' work items synchronously + */ int isotp_fast_unbind(struct isotp_fast_ctx *ctx) { + struct k_work_sync sync; + + /* stop accepting new inbound frames before tearing anything down */ if (ctx->filter_id >= 0 && ctx->can_dev) { can_remove_rx_filter(ctx->can_dev, ctx->filter_id); + ctx->filter_id = -1; } #ifdef CONFIG_ISOTP_FAST_BLOCKING_RECEIVE @@ -1000,6 +1088,40 @@ int isotp_fast_unbind(struct isotp_fast_ctx *ctx) free_recv_await_ctx(ctx, actx); } #endif + + struct isotp_fast_recv_ctx *rctx; + struct isotp_fast_recv_ctx *rnext; + SYS_SLIST_FOR_EACH_CONTAINER_SAFE(&ctx->isotp_recv_ctx_list, rctx, rnext, node) + { + k_timer_stop(&rctx->timer); + k_work_cancel_sync(&rctx->work, &sync); + if (atomic_get(&rctx->pending_cb) == 0) { + /* a callback that fired before the check can only have submitted + * work; cancel once more now that no further callback can fire */ + k_work_cancel_sync(&rctx->work, &sync); + free_recv_ctx(rctx); + } + else { + orphan_recv_ctx(ctx, rctx); + } + } + + struct isotp_fast_send_ctx *sctx; + struct isotp_fast_send_ctx *snext; + SYS_SLIST_FOR_EACH_CONTAINER_SAFE(&ctx->isotp_send_ctx_list, sctx, snext, node) + { + k_timer_stop(&sctx->timer); + k_work_cancel_sync(&sctx->work, &sync); + if (atomic_get(&sctx->pending_cb) == 0) { + k_work_cancel_sync(&sctx->work, &sync); + ctx->sent_callback(ISOTP_N_ERROR, sctx->tx_addr, sctx->cb_arg); + free_send_ctx(sctx); + } + else { + orphan_send_ctx(ctx, sctx); + } + } + return ISOTP_N_OK; } @@ -1114,6 +1236,7 @@ int isotp_fast_recv(struct isotp_fast_ctx *ctx, struct can_filter sender, uint8_ static void isotp_fast_sent_single_cb(const struct device *dev, int error, void *arg) { struct isotp_fast_send_ctx *ctx = arg; + atomic_dec(&ctx->pending_cb); ctx->ctx->sent_callback(error, ctx->tx_addr, ctx->cb_arg); free_send_ctx(ctx); } @@ -1144,7 +1267,16 @@ int isotp_fast_send(struct isotp_fast_ctx *ctx, const uint8_t *data, size_t len, return ISOTP_NO_NET_BUF_LEFT; } context->cb_arg = cb_arg; + atomic_inc(&context->pending_cb); ret = can_send(ctx->can_dev, &frame, K_MSEC(ISOTP_A_TIMEOUT_MS), isotp_fast_sent_single_cb, context); + if (ret != 0) { + /* The completion callback (which normally frees the context) + * never fires for a frame that was not queued */ + atomic_dec(&context->pending_cb); + if (context->state == ISOTP_TX_SEND_FF && atomic_get(&context->pending_cb) == 0) { + free_send_ctx(context); + } + } return ret; } else { diff --git a/src/can/zephyr/isotp/isotp_fast_internal.h b/src/can/zephyr/isotp/isotp_fast_internal.h index 0c21c29d..26e654de 100644 --- a/src/can/zephyr/isotp/isotp_fast_internal.h +++ b/src/can/zephyr/isotp/isotp_fast_internal.h @@ -6,6 +6,7 @@ #include "isotp_internal.h" #include +#include #include #ifdef CONFIG_ISOTP_FAST_PER_FRAME_DISPATCH @@ -47,6 +48,13 @@ struct isotp_fast_send_ctx uint8_t sn : 4; /**< sequence number; overflows at 4 bits per spec */ uint8_t backlog; uint8_t stmin; + /** + * Number of outstanding can_send() completion callbacks that hold a + * pointer to this context. A queued CAN frame cannot be cancelled, so a + * context must not be freed while this is non-zero (see + * @ref isotp_fast_unbind) + */ + atomic_t pending_cb; }; /** @@ -75,6 +83,12 @@ struct isotp_fast_recv_ctx #ifdef ISOTP_FAST_RECEIVE_QUEUE bool pending; #endif + /** + * Number of outstanding can_send() completion callbacks (flow control + * frames) that hold a pointer to this context; see + * @ref isotp_fast_send_ctx.pending_cb + */ + atomic_t pending_cb; }; #ifdef CONFIG_ISOTP_FAST_BLOCKING_RECEIVE diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index aecf7222..8567153b 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -36,7 +36,8 @@ target_sources(testapp PRIVATE TestBinaryEncoder.cpp TestBinaryDecodingRecords.cpp TestTextEncodingRecords.cpp TestRequestRewriter.cpp - TestEui.cpp) + TestEui.cpp + TestClient.cpp) # regrettably exlcude this test until we figure out why Socket server is broken on macOS if(NOT APPLE) diff --git a/tests/TestClient.cpp b/tests/TestClient.cpp new file mode 100644 index 00000000..fecec6be --- /dev/null +++ b/tests/TestClient.cpp @@ -0,0 +1,132 @@ +/* + * Copyright (c) 2026 Brill Power. + * + * SPDX-License-Identifier: Apache-2.0 + */ +#include "thingset++/ThingSetClient.hpp" +#include +#include +#include +#include +#include +#include + +using namespace ThingSet; + +namespace { + +/// Scriptable in-memory transport for exercising ThingSetClient error paths. +class FakeClientTransport : public ThingSetClientTransport +{ +public: + bool connectResult = true; + bool writeResult = true; + /// Value returned by read(); when positive, up to that many bytes of + /// `response` are copied into the caller's buffer first. + int readResult = 0; + std::vector response; + + bool connect() override + { + return connectResult; + } + + int read(uint8_t *buffer, size_t len) override + { + if (readResult > 0 && !response.empty()) { + size_t count = std::min({ (size_t)readResult, response.size(), len }); + memcpy(buffer, response.data(), count); + } + return readResult; + } + + bool write(uint8_t *, size_t) override + { + return writeResult; + } +}; + +struct ClientFixture +{ + FakeClientTransport transport; + std::array rxBuffer; + std::array txBuffer; + ThingSetClient client; + + ClientFixture() : client(transport, rxBuffer, txBuffer) + {} +}; + +} // namespace + +TEST(Client, ExecSucceedsOnWellFormedResponse) +{ + ClientFixture f; + f.transport.response = { (uint8_t)ThingSetStatusCode::changed, 0xF6, 0x00 }; + f.transport.readResult = 3; + + int ret = -1; + ThingSetResult result = f.client.exec(0x1234, &ret); + EXPECT_TRUE(result.success()); + EXPECT_EQ(ret, 0); +} + +/// Regression test: a receive timeout after a previous successful exchange +/// must fail, even though the stale success response is still sitting in the +/// client's rx buffer. The negative errno from the transport used to be +/// assigned to a size_t, defeating the empty-response check and decoding the +/// stale bytes as a fresh (successful) response. +TEST(Client, ExecFailsOnReadTimeoutDespiteStaleBuffer) +{ + ClientFixture f; + f.transport.response = { (uint8_t)ThingSetStatusCode::changed, 0xF6, 0x00 }; + f.transport.readResult = 3; + + int ret = -1; + ASSERT_TRUE(f.client.exec(0x1234, &ret).success()); + + // the device stops answering: msgq timeout surfaces as -EAGAIN + f.transport.readResult = -EAGAIN; + + ThingSetResult result = f.client.exec(0x1234, &ret); + EXPECT_FALSE(result.success()); + EXPECT_EQ(result.code(), ThingSetStatusCode::gatewayTimeout); +} + +TEST(Client, ExecFailsOnZeroLengthRead) +{ + ClientFixture f; + f.transport.readResult = 0; + + int ret = -1; + ThingSetResult result = f.client.exec(0x1234, &ret); + EXPECT_FALSE(result.success()); + EXPECT_EQ(result.code(), ThingSetStatusCode::gatewayTimeout); +} + +/// A success status byte with no CBOR null after it must not be decoded. +TEST(Client, ExecFailsOnTruncatedResponse) +{ + ClientFixture f; + f.transport.response = { (uint8_t)ThingSetStatusCode::changed }; + f.transport.readResult = 1; + + int ret = -1; + ThingSetResult result = f.client.exec(0x1234, &ret); + EXPECT_FALSE(result.success()); + EXPECT_EQ(result.code(), ThingSetStatusCode::internalServerError); +} + +/// A single-byte error response is legitimate and must surface the device's +/// own status code (distinguishable from a timeout). +TEST(Client, ExecReturnsDeviceErrorStatus) +{ + ClientFixture f; + f.transport.response = { (uint8_t)ThingSetStatusCode::badRequest }; + f.transport.readResult = 1; + + int ret = -1; + ThingSetResult result = f.client.exec(0x1234, &ret); + EXPECT_FALSE(result.success()); + EXPECT_EQ(result.code(), ThingSetStatusCode::badRequest); +} diff --git a/tests/zephyr/can/src/main.cpp b/tests/zephyr/can/src/main.cpp index 13ef6c5c..0e4b4213 100644 --- a/tests/zephyr/can/src/main.cpp +++ b/tests/zephyr/can/src/main.cpp @@ -69,7 +69,8 @@ static k_tid_t createAndRunClient(k_thread_entry_t runner) } // name needs to be this to make stupid twister check pass -#define ZCLIENT_SERVER_TEST(test_name, Body) \ +// variadic so test bodies may contain top-level commas (e.g. template args) +#define ZCLIENT_SERVER_TEST(test_name, ...) \ ZTEST(ZephyrClientServer, test_name) \ { \ k_sem_init(&serverStarted, 0, 1); \ @@ -89,7 +90,7 @@ ZTEST(ZephyrClientServer, test_name) \ zassert_true(client.connect()); \ LOG_INF("Client connected"); \ \ - Body \ + __VA_ARGS__ \ \ k_sem_give(&clientCompleted); \ }); \ @@ -123,6 +124,99 @@ ZCLIENT_SERVER_TEST(test_update, zassert_equal(25.0f, totalVoltage.getValue()); ) +/* Transport lifecycle: repeatedly create a client to a node that never + * answers, let the request time out, and destroy the transport. Historically + * in-flight context handling could leak send contexts from a 4-deep slab and + * left timers/work items pointing at destroyed (stack-allocated) transports. + * Six cycles (> slab depth) surface a reintroduced leak as ISOTP_NO_CTX_LEFT, + * and the final exchange proves the shared client still works. */ +ZCLIENT_SERVER_TEST(test_client_lifecycle_absent_node, + for (int i = 0; i < 6; i++) { + std::array transportRx; + std::array transportTx; + std::array absentClientRx; + std::array absentClientTx; + ThingSetZephyrCanClientTransport absentTransport(clientInterface, 0x55, transportRx, + transportTx); + ThingSetClient absentClient(absentTransport, absentClientRx, absentClientTx); + zassert_true(absentClient.connect()); + int sum; + auto absentResult = absentClient.exec(0x1000, &sum, 1, 2); + zassert_false(absentResult.success(), "exec to an absent node must not succeed"); + } + + int value; + auto result = client.exec(0x1000, &value, 2, 3); + zassert_true(result.success(), "shared client must still work after lifecycle churn"); + zassert_equal(5, value); +) + +/* Response cross-talk: RR client filters historically masked out the source + * address, so every client on a shared interface matched every peer's + * responses (first-match-wins on real hardware silently starved the loser). + * A client bound to a silent peer must not observe responses addressed to + * another client. */ +ZCLIENT_SERVER_TEST(test_no_response_crosstalk_between_clients, + std::array transportRx; + std::array transportTx; + std::array silentClientRx; + std::array silentClientTx; + ThingSetZephyrCanClientTransport silentTransport(clientInterface, 0x55, transportRx, + transportTx); + ThingSetClient silentClient(silentTransport, silentClientRx, silentClientTx); + zassert_true(silentClient.connect()); + + /* a full exchange with the real server crosses the bus */ + int value; + auto result = client.exec(0x1000, &value, 2, 3); + zassert_true(result.success()); + zassert_equal(5, value); + + /* the silent-peer client must not have captured that response: its own + * request must time out rather than return the stray reply */ + int sum; + auto silentResult = silentClient.exec(0x1000, &sum, 1, 2); + zassert_false(silentResult.success(), + "client bound to a silent peer must not see another client's response"); +) + +/* connect() historically could not fail: isotp_fast_bind ignored the result + * of can_add_rx_filter, so a full filter table was reported as a successful + * bind and every subsequent exchange timed out. Exhaust the controller's RX + * filters and check that connect() now fails -- and recovers once filters + * are freed. */ +ZTEST(ZephyrClientServer, test_connect_fails_when_filters_exhausted) +{ + struct can_filter filter = { + .id = 0x100, + .mask = CAN_EXT_ID_MASK, + .flags = CAN_FILTER_IDE, + }; + int filterIds[128]; + int count = 0; + + while (count < (int)ARRAY_SIZE(filterIds)) { + int id = can_add_rx_filter( + canDevice, [](const struct device *, struct can_frame *, void *) {}, nullptr, &filter); + if (id < 0) { + break; + } + filterIds[count++] = id; + } + zassert_true(count < (int)ARRAY_SIZE(filterIds), "expected to exhaust CAN RX filters"); + + std::array rxBuffer; + std::array txBuffer; + ThingSetZephyrCanClientTransport transport(clientInterface, 0x01, rxBuffer, txBuffer); + zassert_false(transport.connect(), "connect() must fail with no free RX filters"); + + for (int i = 0; i < count; i++) { + can_remove_rx_filter(canDevice, filterIds[i]); + } + + zassert_true(transport.connect(), "connect() must succeed again once filters are free"); +} + static void *testSetup(void) { // Not allowed until interface is bound to address