diff --git a/.clang-tidy b/.clang-tidy index c856aaf..29ff45e 100644 --- a/.clang-tidy +++ b/.clang-tidy @@ -83,6 +83,7 @@ Checks: >- -cert-dcl03-c, -readability-else-after-return, -llvm-prefer-isa-or-dyn-cast-in-conditionals, + -*-trailing-comma, CheckOptions: - key: misc-const-correctness.TransformPointersAsValues value: 'true' diff --git a/cy/cy.c b/cy/cy.c index 80233e1..4be7b50 100644 --- a/cy/cy.c +++ b/cy/cy.c @@ -85,6 +85,18 @@ struct cy_tree_t // Soft states associated remotes will be discarded when stale for this long. #define SESSION_LIFETIME (60 * MEGA) +// How long completed reliable-request ACK state is retained to answer retransmits. +// This is a last resort knob for extremely memory-starved systems, for expert tuning only. +// +// Applications that are memory-constrained, request-intensive, and receive reliable responses may choose to downsize +// this to reduce the memory pressure from retained ACK markers in case an ACK transmission has failed. +// Small values introduce the risk of the server reading a NACK for a retried reliable response where the original +// response was an ACK; some applications, in particular those that don't use streaming, are tolerant to this which +// enables memory savings. +#ifndef CY_CONFIG_REQUEST_ACK_RETENTION_us +#define CY_CONFIG_REQUEST_ACK_RETENTION_us SESSION_LIFETIME +#endif + // Largest backward monotonic-counter jump still treated as delayed traffic from the current session. #define SESSION_COUNTER_MAX_BACKWARD_LAG 100000ULL @@ -1015,6 +1027,11 @@ struct cy_topic_t // Similar to publish futures but referencing request_future_t. cy_tree_t* request_futures_by_tag; + // Ack records left behind by destroyed request futures to answer retransmitted reliable responses. + // Ordered by dead_at: re-headed only at handoff and never touched afterward, so a tail sweep is exact. + cy_tree_t* request_acks_by_tag; + cy_list_t request_acks_by_expiry; + // Subscriber-related states. // // The subject reader exists only as long as there are active subscriptions to avoid unrelated traffic. @@ -1222,7 +1239,7 @@ static int_fast8_t topic_lage(const cy_topic_t* const topic, const cy_us_t now) } // CRDT merge operator on the topic log-age. Shift ts_origin into the past if needed. -static void topic_merge_lage(cy_topic_t* const topic, const cy_us_t now, int_fast8_t r_lage) +static void topic_merge_lage(cy_topic_t* const topic, const cy_us_t now, const int_fast8_t r_lage) { topic->ts_origin = sooner(topic->ts_origin, now - lage_to_us(r_lage)); } @@ -1596,6 +1613,8 @@ static cy_err_t topic_new(cy_t* const cy, topic->pub_futures_by_tag = NULL; topic->request_futures_by_tag = NULL; + topic->request_acks_by_tag = NULL; + topic->request_acks_by_expiry = LIST_EMPTY; topic->user_context = CY_USER_CONTEXT_EMPTY; @@ -2517,6 +2536,7 @@ typedef struct uint64_t seqno_top; bitmap_t seqno_acked[BITMAP_WORDS(REQUEST_FUTURE_HISTORY)]; // bit 0 = seqno_top, bit 1 = seqno_top-1, ... } request_future_remote_t; +static_assert((sizeof(void*) > 4) || (sizeof(request_future_remote_t) <= (64 - 8)), "o1heap block spill"); typedef struct { @@ -2524,24 +2544,44 @@ typedef struct uint64_t remote_id; } request_future_remote_factory_context_t; +// Answers retransmitted reliable responses after the application has destroyed the request future. +// It holds only the fields the ack decision consults, so the future is handed over at disposal and freed. +// Per-remote states are never removed while the future lives, assuming futures are short-lived and/or the +// responder set is mostly constant; after handoff the record is query-only and can no longer grow. States: +// solo -- remote solo_remote_id acked seqno 0; the common single-responder shape, no tree, no second alloc +// !solo, tree -- promoted: per-remote bitmaps, same rule as the live path +// !solo, NULL -- nothing acked, answers NACK to everything +typedef struct +{ + cy_tree_t index; // Keyed by tag. MUST be the first field for ptr equivalence. + cy_list_member_t expiry; // Ordered by dead_at; head is newest. Enlisted only at handoff. + uint64_t tag; + cy_us_t dead_at; + union + { + uint64_t solo_remote_id; // iff solo + cy_tree_t* tree; // iff !solo; NULL means NONE + } u; + bool solo; +} request_ack_t; +static_assert((sizeof(void*) > 4) || (sizeof(request_ack_t) <= (64 - 8)), "o1heap block spill"); + typedef struct { cy_future_t base; // The key is the tag. cy_topic_t* topic; cy_us_t liveness_timeout; // Inter-response timeout for stream liveness monitoring. - bool finalized; // Staying behind to handle possible duplicate responses to ack/nack correctly. cy_future_t* publish; cy_err_t error; // Most recently seen error. uint64_t response_count; // Unique responses after deduplication from all remotes combined. cy_response_t last_response; // Overwritten when new responses arrive. - // States per remote node that is responding to this request using reliable response delivery. - // States are never removed assuming that futures are short-lived and/or the responder set is mostly constant. - // This is used to deduplicate responses (when reliable response is delivered but ack is lost, remote retransmits) - // and to keep track which ones need to be acked when duplicates arrive. - cy_tree_t* remote_by_id; + // Deduplication state for remotes responding with reliable delivery; created on the first such response. + // Handed over to the topic at disposal so it can keep answering retransmits after the future is gone. + // NULL until the first reliable response arrives. + request_ack_t* ack; } request_future_t; static int32_t request_future_remote_cavl_compare(const void* const user, const cy_tree_t* const node) @@ -2561,11 +2601,156 @@ static cy_tree_t* request_future_remote_cavl_factory(void* const user) return (cy_tree_t*)node; } +typedef enum +{ + response_rx_ack, + response_rx_nack, + response_rx_silent, // Transient local drop: no ACK/NACK, keep the future pending. +} response_rx_t; + +static int32_t request_ack_cavl_compare(const void* const user, const cy_tree_t* const node) +{ + const uint64_t outer = *(const uint64_t*)user; + const uint64_t inner = ((const request_ack_t*)node)->tag; + return (outer == inner) ? 0 : ((outer > inner) ? +1 : -1); +} + +static request_ack_t* request_ack_new(const cy_t* const cy, const uint64_t tag) +{ + request_ack_t* const self = (request_ack_t*)mem_alloc_zero(cy, sizeof(request_ack_t)); + if (self != NULL) { + self->index = TREE_NULL; + self->expiry = LIST_MEMBER_NULL; + self->tag = tag; + self->dead_at = BIG_BANG; // No deadline while the future owns it; request_ack_retain() sets the real one. + self->u.tree = NULL; // Names the active union member; do not infer it from the zeroed storage. + self->solo = false; + } + return self; +} + +// Serves detached (future-owned), swept, and torn-down records alike. +static void request_ack_destroy(cy_topic_t* const owner, request_ack_t* const self) +{ + const cy_t* const cy = owner->cy; + CY_ASSERT(is_listed(&owner->request_acks_by_expiry, &self->expiry) == + cavl2_is_inserted(owner->request_acks_by_tag, &self->index)); + delist(&owner->request_acks_by_expiry, &self->expiry); + cavl2_remove_if(&owner->request_acks_by_tag, &self->index); + if (!self->solo) { + while (self->u.tree != NULL) { + request_future_remote_t* const remote = (request_future_remote_t*)self->u.tree; + cavl2_remove(&self->u.tree, self->u.tree); + mem_free(cy, remote); + } + } + mem_free(cy, self); +} + +static void request_ack_drop_stale(cy_topic_t* const owner, const cy_us_t now) +{ + while (true) { + request_ack_t* const ack = LIST_TAIL(owner->request_acks_by_expiry, request_ack_t, expiry); + if ((ack == NULL) || (ack->dead_at >= now)) { + break; + } + CY_TRACE(owner->cy, "🧹 T%016jx tag=%016jx", (uintmax_t)owner->hash, (uintmax_t)ack->tag); + request_ack_destroy(owner, ack); + } +} + +static bool request_ack_test(const request_ack_t* const self, const uint64_t remote_id, const uint64_t seqno) +{ + if (self == NULL) { + return false; + } + if (self->solo) { + return (self->u.solo_remote_id == remote_id) && (seqno == 0); + } + const request_future_remote_t* const remote = + (request_future_remote_t*)cavl2_find(self->u.tree, &remote_id, request_future_remote_cavl_compare); + return (remote != NULL) && (seqno <= remote->seqno_top) && + bitmap_test_bounded(remote->seqno_acked, REQUEST_FUTURE_HISTORY, remote->seqno_top - seqno); +} + +// Sets out_fresh iff the response is genuinely new and must reach the app. +// The solo slot encodes exactly "R acked seqno 0", so it can only be claimed while the record is still empty. +static response_rx_t request_ack_admit(request_ack_t* const self, + cy_topic_t* const topic, + const uint64_t remote_id, + const uint64_t seqno, + bool* const out_fresh) +{ + cy_t* const cy = topic->cy; + *out_fresh = false; + if (self->solo) { + if ((self->u.solo_remote_id == remote_id) && (seqno == 0)) { + return response_rx_ack; // Duplicate of the inlined ack; no promotion needed. + } + cy_tree_t* promoted = NULL; + request_future_remote_factory_context_t solo_ctx = { .cy = cy, .remote_id = self->u.solo_remote_id }; + request_future_remote_t* const node = + (request_future_remote_t*)cavl2_find_or_insert(&promoted, + &solo_ctx.remote_id, + request_future_remote_cavl_compare, + &solo_ctx, + request_future_remote_cavl_factory); + if (node == NULL) { + ON_ASYNC_ERROR(cy, topic, CY_ERR_MEMORY); + return response_rx_silent; // Still solo, nothing lost. + } + // A zero-allocated node means "known, nothing acked", so the inlined ack must be re-stated explicitly. + bitmap_set(node->seqno_acked, 0); // seqno_top is already zero, and the inlined ack was for seqno 0. + self->solo = false; + self->u.tree = promoted; + } else if ((self->u.tree == NULL) && (seqno == 0)) { + self->solo = true; + self->u.solo_remote_id = remote_id; + *out_fresh = true; + return response_rx_ack; + } + // Generic per-remote path: find or create the remote state, then update its seqno frontier bitmap. + request_future_remote_factory_context_t factory_ctx = { .cy = cy, .remote_id = remote_id }; + request_future_remote_t* const remote = (request_future_remote_t*)cavl2_find_or_insert( + &self->u.tree, &remote_id, request_future_remote_cavl_compare, &factory_ctx, request_future_remote_cavl_factory); + if (remote == NULL) { + ON_ASYNC_ERROR(cy, topic, CY_ERR_MEMORY); + return response_rx_silent; + } + if (seqno > remote->seqno_top) { // Pushes the frontier, need to shift the bitmap. + bitmap_shift(remote->seqno_acked, REQUEST_FUTURE_HISTORY, (intmax_t)(seqno - remote->seqno_top)); + bitmap_set(remote->seqno_acked, 0); // 0th bit is always set, redundant but simple + remote->seqno_top = seqno; + } else { // earlier seqno below the frontier, which might be new if delivered out of order + const uint64_t dist = remote->seqno_top - seqno; + if (dist >= REQUEST_FUTURE_HISTORY) { + return response_rx_nack; // too old, exceeds history, probably sender misbehaving, do not accept + } + if (bitmap_test(remote->seqno_acked, (size_t)dist)) { + return response_rx_ack; // duplicate, probably lost ack + } + bitmap_set(remote->seqno_acked, (size_t)dist); // genuinely new response just arrived out of order + } + CY_ASSERT(remote->seqno_top >= seqno); + *out_fresh = true; + return response_rx_ack; +} + +// Hand over to the topic to answer retransmits after the future is gone. Allocation-free: dispose must be infallible. +static void request_ack_retain(cy_topic_t* const owner, request_ack_t* const self, const cy_us_t now) +{ + self->dead_at = now + (CY_CONFIG_REQUEST_ACK_RETENTION_us); + const cy_tree_t* const ins = cavl2_find_or_insert( + &owner->request_acks_by_tag, &self->tag, request_ack_cavl_compare, self, cavl2_trivial_factory); + CY_ASSERT(ins == &self->index); // Tags are unique per topic. + (void)ins; + enlist_head(&owner->request_acks_by_expiry, &self->expiry); +} + static void request_publish_callback(cy_future_t* const fut) { request_future_t* const self = (request_future_t*)cy_future_context(fut).ptr[0]; CY_ASSERT(self->publish == fut); - CY_ASSERT(!self->finalized); const cy_err_t err = cy_future_error(fut); if (cy_future_done(fut)) { // In case there are intermediate updates. May be uncoverable. cy_future_destroy(fut); @@ -2576,17 +2761,10 @@ static void request_publish_callback(cy_future_t* const fut) } if (err != CY_OK) { // Report every error. self->error = err; - future_notify(&self->base); // Invalidates self; expect finalization. + future_notify(&self->base); // Invalidates self; expect disposal. } } -typedef enum -{ - response_rx_ack, - response_rx_nack, - response_rx_silent, // Transient local drop: no ACK/NACK, keep the future pending. -} response_rx_t; - // Invalidates the future because it may be destroyed. static response_rx_t request_on_response(request_future_t* const self, const uint64_t seqno, @@ -2599,50 +2777,22 @@ static response_rx_t request_on_response(request_future_t* const self, CY_ASSERT(message.content != NULL); cy_t* const cy = self->base.cy; - // Zombie mode -- the application has destroyed the future and is no longer accepting responses. - // We are left behind only to retransmit acks for reliable responses if any are lost. - if (self->finalized) { - if (reliable) { - const request_future_remote_t* const remote = - (request_future_remote_t*)cavl2_find(self->remote_by_id, &lane.id, request_future_remote_cavl_compare); - if ((remote != NULL) && (seqno <= remote->seqno_top)) { - return bitmap_test_bounded(remote->seqno_acked, REQUEST_FUTURE_HISTORY, remote->seqno_top - seqno) - ? response_rx_ack - : response_rx_nack; - } - } - return response_rx_nack; // Do not proceed to the acceptance path, we're already dead. - } - // The transport deduplicates messages, meaning that at this level only reliable responses require deduplication, // because the remote would retransmit if our acks are lost. We need to shield the application from that. if (reliable) { - request_future_remote_factory_context_t factory_ctx = { .cy = cy, .remote_id = lane.id }; - request_future_remote_t* const remote = - (request_future_remote_t*)cavl2_find_or_insert(&self->remote_by_id, - &lane.id, - request_future_remote_cavl_compare, - &factory_ctx, - request_future_remote_cavl_factory); - if (remote == NULL) { - ON_ASYNC_ERROR(cy, self->topic, CY_ERR_MEMORY); - return response_rx_silent; - } - if (seqno > remote->seqno_top) { // Pushes the frontier, need to shift the bitmap. - bitmap_shift(remote->seqno_acked, REQUEST_FUTURE_HISTORY, (intmax_t)(seqno - remote->seqno_top)); - bitmap_set(remote->seqno_acked, 0); // 0th bit is always set, redundant bit simple - remote->seqno_top = seqno; - } else { // earlier seqno below the frontier, which might be new if delivered out of order - const uint64_t dist = remote->seqno_top - seqno; - if (dist >= REQUEST_FUTURE_HISTORY) { - return response_rx_nack; // too old, exceeds history, probably sender misbehaving, do not accept + if (self->ack == NULL) { + self->ack = request_ack_new(cy, self->base.key); + if (self->ack == NULL) { + ON_ASYNC_ERROR(cy, self->topic, CY_ERR_MEMORY); + return response_rx_silent; } - if (bitmap_test(remote->seqno_acked, (size_t)dist)) { - return response_rx_ack; // duplicate, probably lost ack - } - bitmap_set(remote->seqno_acked, (size_t)dist); // genuinely new response just arrived out of order } - CY_ASSERT(remote->seqno_top >= seqno); + bool fresh = false; + const response_rx_t verdict = request_ack_admit(self->ack, self->topic, lane.id, seqno, &fresh); + if (!fresh) { + return verdict; // Duplicate, too old, or transient failure; the application must not see it. + } + CY_ASSERT(verdict == response_rx_ack); // A fresh response is always acked; the fall-through relies on it. } // At this point, the response is known to be unique. Rewrite the last stored response. @@ -2658,30 +2808,13 @@ static response_rx_t request_on_response(request_future_t* const self, // Notify the application that a new response is available. self->error = CY_OK; - future_notify(&self->base); // Invalidates self; expect finalization. + future_notify(&self->base); // Invalidates self; expect disposal. return response_rx_ack; } -static void request_future_destroy(request_future_t* const self) -{ - cy_future_t* const base = &self->base; - CY_ASSERT(self->finalized); - CY_ASSERT(self->publish == NULL); - future_deadline_disarm(base); - cy_message_refcount_dec(self->last_response.message.content); // NULL-safe - future_index_remove(base, &self->topic->request_futures_by_tag); - while (self->remote_by_id != NULL) { - request_future_remote_t* const remote = (request_future_remote_t*)self->remote_by_id; - cavl2_remove(&self->remote_by_id, self->remote_by_id); - mem_free(base->cy, remote); - } - mem_free(base->cy, self); -} - static bool request_future_done(const cy_future_t* const base) { const request_future_t* const self = (const request_future_t*)base; - CY_ASSERT(!self->finalized); // use after free? return (self->last_response.message.content != NULL) || !future_deadline_armed(base); // got response or timed out } static cy_err_t request_future_error(const cy_future_t* const base) { return ((const request_future_t*)base)->error; } @@ -2692,33 +2825,35 @@ static void request_future_timeout(cy_future_t* const base, const cy_us_t schedu (void)now; request_future_t* const self = (request_future_t*)base; CY_ASSERT(!future_deadline_armed(base)); - if (!self->finalized) { - self->error = CY_ERR_LIVENESS; - future_notify(base); // Expect finalization call. - } else { - request_future_destroy(self); - } + self->error = CY_ERR_LIVENESS; + future_notify(base); // Expect disposal. } static void request_future_dispose(cy_future_t* const base) { - request_future_t* const self = (request_future_t*)base; - CY_ASSERT(!self->finalized); + request_future_t* const self = (request_future_t*)base; + cy_t* const cy = base->cy; + cy_topic_t* const topic = self->topic; if (self->publish != NULL) { cy_future_destroy(self->publish); self->publish = NULL; } - self->finalized = true; + future_deadline_disarm(base); + const cy_us_t now = cy_now(cy); // sampled before deindexing to avoid vtable access during teardown + future_index_remove(base, &topic->request_futures_by_tag); + // A message destructor re-entering the RX path here would find no record yet; no sane transport does that. + cy_message_refcount_dec(self->last_response.message.content); // NULL-safe // The acks that we sent for reliable responses may have been lost, in which case the remote would retransmit. // In that case we will need to respond the same way we did the first time without involving the application. - // To facilitate that, we leave a pending finalized future behind. It will be destroyed after some timeout. - if (self->remote_by_id != NULL) { // Stayin' alive because we need to continue processing possible duplicates. - cy_message_refcount_dec(self->last_response.message.content); // Release memory early (NULL-safe) - self->last_response.message.content = NULL; - future_deadline_arm(base, cy_now(base->cy) + (SESSION_LIFETIME / 2)); - } else { // If we didn't ack any reliable responses, there is no need to leave a finalized future behind. - request_future_destroy(self); + // A record that acked nothing can only ever answer NACK, so it is freed rather than retained. + if (self->ack != NULL) { + if (self->ack->solo || (self->ack->u.tree != NULL)) { + request_ack_retain(topic, self->ack, now); + } else { + request_ack_destroy(topic, self->ack); + } } + mem_free(cy, self); } static const cy_future_vtable_t request_future_vtable = { .done = request_future_done, @@ -2747,7 +2882,7 @@ cy_future_t* cy_request(cy_publisher_t* const pub, fut->liveness_timeout = response_timeout; fut->last_response.message.timestamp = BIG_BANG; fut->last_response.message.content = NULL; - fut->remote_by_id = NULL; + fut->ack = NULL; // Once fallible preparations are done, send the request. // Reliable publication is quite a can of worms but we use it as a black box here. @@ -4197,16 +4332,16 @@ static void topic_destroy(cy_topic_t* const topic) CY_ASSERT(topic->sub_list_dedup_by_recency.head == NULL); CY_ASSERT(topic->sub_list_dedup_by_recency.tail == NULL); - // Remove any zombie request futures that may be left behind to manage retransmissions. - // This is lifetime-safe because the API contract requires that the application must destroy pending futures - // before destroying their publisher, and the topic cannot be destroyed as long as it has at least one live - // publisher (or subscriber or whatever). To wit, topics are recycled after some timeout, and by the time it - // expires the zombie request futures are likely going to be destroyed on timeout anyway. - while (topic->request_futures_by_tag != NULL) { - request_future_t* const future = (request_future_t*)topic->request_futures_by_tag; - CY_ASSERT(future->finalized); // Otherwise, the application forgot to destroy the future! - request_future_destroy(future); + // The application must destroy pending futures before destroying their publisher, and the topic cannot be + // destroyed while it has a live publisher; so a non-empty index here means the application forgot. + CY_ASSERT(topic->request_futures_by_tag == NULL); + + // Remove the ack records left behind by destroyed request futures to manage retransmissions. + while (topic->request_acks_by_expiry.head != NULL) { + request_ack_destroy(topic, LIST_MEMBER(topic->request_acks_by_expiry.head, request_ack_t, expiry)); } + CY_ASSERT(topic->request_acks_by_expiry.tail == NULL); + CY_ASSERT(topic->request_acks_by_tag == NULL); // Release gossip shard reader/writer from registry. if (topic->gossip_writer != NULL) { @@ -4556,6 +4691,7 @@ static cy_us_t poll(cy_t* const cy, cy_us_t* const out_now) } if (cy->topic_iter != NULL) { dedup_drop_stale(cy->topic_iter, now); + request_ack_drop_stale(cy->topic_iter, now); // Do we accept the full subscriber scan here? const cy_topic_coupling_t* cpl = cy->topic_iter->couplings; while (cpl != NULL) { @@ -4698,7 +4834,8 @@ cy_topic_t* cy_topic_iter_next(cy_topic_t* const topic) { return (cy_topic_t*)ca cy_str_t cy_topic_name(const cy_topic_t* const topic) { - if (topic != NULL) { + // The name index is absent while a topic is being torn down, and in tests that synthesize bare topic objects. + if ((topic != NULL) && (topic->index_name != NULL)) { return (cy_str_t){ .len = topic->index_name->key_len, .str = topic->name }; } return (cy_str_t){ .len = 0, .str = "" }; @@ -4952,11 +5089,10 @@ void cy_on_message(cy_platform_t* const platform, // edge case when we ack a response and destroy the future immediately afterward with the subsequent loss // of the ack, the remote would retransmit, and the next time we will respond with a nack because the // future is already destroyed. There are many ways to avoid this, such as keeping a log of recently - // acked responses, etc. Here, we choose to keep futures alive for a brief time after the application - // destroys them such that we could delegate the ack/nack decision to them, because it appears to be - // the simplest solution with minimal state keeping. Note that futures that acknowledged no responses - // do not need to be retained since the outcome is always the same -- always nack. - // This is an implementation detail that does not affect wire semantics of course. + // acked responses, etc. Here, when the application destroys the future we hand its deduplication state + // over to the topic as a compact record that outlives it and answers retransmits on its own. + // Note that futures that acknowledged no responses leave no record since the outcome is always the + // same -- always nack. This is an implementation detail that does not affect wire semantics of course. response_rx_t response = response_rx_nack; cy_topic_t* const topic = cy_topic_find_by_hash(cy, hash); if (topic != NULL) { @@ -4964,6 +5100,10 @@ void cy_on_message(cy_platform_t* const platform, (request_future_t*)future_index_lookup(topic->request_futures_by_tag, message_tag); if (future != NULL) { response = request_on_response(future, seqno, message, reliable, lane); + } else if (reliable) { // The future may be gone but its ack record may still answer for it. + const request_ack_t* const ack = + (request_ack_t*)cavl2_find(topic->request_acks_by_tag, &message_tag, request_ack_cavl_compare); + response = request_ack_test(ack, lane.id, seqno) ? response_rx_ack : response_rx_nack; } } if (reliable && (response != response_rx_silent)) { diff --git a/tests/src/test_api_rpc_e2e.cpp b/tests/src/test_api_rpc_e2e.cpp index aca2d3e..5156369 100644 --- a/tests/src/test_api_rpc_e2e.cpp +++ b/tests/src/test_api_rpc_e2e.cpp @@ -793,7 +793,57 @@ void test_api_rpc_e2e_r06_reliable_response_history_nack_and_unknown_nack() cleanup_case(net, now, { request }, { server_sub }, { client }); } -void test_api_rpc_e2e_r07_zombie_ack_seen_nack_unseen() +// r07/r14 both open with seqno 5, so neither reaches the single-responder/seqno-0 shape that a client sees in +// practice. This is the black-box counterpart: after the future is destroyed, the retained state must still ACK +// the exact response it accepted, and NACK a different seqno or a different responder. +void test_api_rpc_e2e_r29_retained_solo_ack_and_nack() +{ + e2e::sim_net_t net{}; + TEST_ASSERT_EQUAL_INT(CY_OK, e2e::sim_net_init(net)); + cy_us_t now = 0; + + static constexpr const char* topic_name = "rpc/r29/topic"; + cy_publisher_t* const client = make_client(net, topic_name); + server_context_t server{}; + server.responses_per_request = 0U; + cy_future_t* const server_sub = make_server_subscriber(net, topic_name, server); + + e2e::set_now(net, now); + cy_future_t* const request = request_once(client, now, 6U, 1U, 220'000, 220'000); + TEST_ASSERT_NOT_NULL(request); + const request_wire_info_t req_wire = last_request_wire(net); + const auto payload = e2e::app_payload_pack(929U, 1U); + const std::vector body(payload.begin(), payload.end()); + + // First reliable response carries seqno 0 -- the inlined single-responder shape. + inject_response_wire(net, header_rsp_rel, 0x60U, 0U, req_wire.topic_hash, req_wire.tag, body, now + 1); + const cy_response_t first = cy_response_move(request); + TEST_ASSERT_NOT_NULL(first.message.content); + cy_message_refcount_dec(first.message.content); + + cy_future_destroy(request); + + // Same responder, same seqno -> ACK: destroying the future must not retract an accepted response. + std::size_t before = e2e::sim_net_captures(net).size(); + inject_response_wire(net, header_rsp_rel, 0x60U, 0U, req_wire.topic_hash, req_wire.tag, body, now + 2); + std::vector controls = response_controls_since(net, before); + TEST_ASSERT_EQUAL_size_t(1U, controls.size()); + TEST_ASSERT_EQUAL_UINT8(header_rsp_ack, controls.at(0).header_type); + + // Same responder, seqno 1 -> NACK: never seen, so never acked. + before = e2e::sim_net_captures(net).size(); + inject_response_wire(net, header_rsp_rel, 0x61U, 1U, req_wire.topic_hash, req_wire.tag, body, now + 3); + controls = response_controls_since(net, before); + TEST_ASSERT_EQUAL_size_t(1U, controls.size()); + TEST_ASSERT_EQUAL_UINT8(header_rsp_nack, controls.at(0).header_type); + + // The unknown-responder case is r14's; it cannot be asserted here because the control frame for a forged + // lane is addressed back to that lane, which response_controls_since() filters out. + + cleanup_case(net, now, {}, { server_sub }, { client }); +} + +void test_api_rpc_e2e_r07_retained_ack_seen_nack_unseen() { e2e::sim_net_t net{}; TEST_ASSERT_EQUAL_INT(CY_OK, e2e::sim_net_init(net)); @@ -1063,7 +1113,7 @@ void test_api_rpc_e2e_r13_reliable_response_unknown_request_tag_nack() cleanup_case(net, now, { request }, { server_sub }, { client }); } -void test_api_rpc_e2e_r14_zombie_unseen_remote_reliable_response_nack() +void test_api_rpc_e2e_r14_retained_unseen_remote_reliable_response_nack() { e2e::sim_net_t net{}; TEST_ASSERT_EQUAL_INT(CY_OK, e2e::sim_net_init(net)); @@ -1814,14 +1864,15 @@ int main() RUN_TEST(test_api_rpc_e2e_r04_failure_then_late_success_transition); RUN_TEST(test_api_rpc_e2e_r05_reliable_response_ack_and_duplicate_ack); RUN_TEST(test_api_rpc_e2e_r06_reliable_response_history_nack_and_unknown_nack); - RUN_TEST(test_api_rpc_e2e_r07_zombie_ack_seen_nack_unseen); + RUN_TEST(test_api_rpc_e2e_r29_retained_solo_ack_and_nack); + RUN_TEST(test_api_rpc_e2e_r07_retained_ack_seen_nack_unseen); RUN_TEST(test_api_rpc_e2e_r08_multicast_response_is_rejected); RUN_TEST(test_api_rpc_e2e_r09_request_callback_status_transitions); RUN_TEST(test_api_rpc_e2e_r10_initial_publish_failure_returns_null); RUN_TEST(test_api_rpc_e2e_r11_request_publish_fails_without_response); RUN_TEST(test_api_rpc_e2e_r12_concurrent_requests_are_correlated); RUN_TEST(test_api_rpc_e2e_r13_reliable_response_unknown_request_tag_nack); - RUN_TEST(test_api_rpc_e2e_r14_zombie_unseen_remote_reliable_response_nack); + RUN_TEST(test_api_rpc_e2e_r14_retained_unseen_remote_reliable_response_nack); RUN_TEST(test_api_rpc_e2e_r15_publish_failure_after_response_keeps_future_alive_until_liveness); RUN_TEST(test_api_rpc_e2e_r16_request_future_allocation_failure_returns_null); RUN_TEST(test_api_rpc_e2e_r17_server_reliable_response_ack_success); diff --git a/tests/src/test_intrusive_future_notify_destroy.c b/tests/src/test_intrusive_future_notify_destroy.c index fc702d9..b77baa8 100644 --- a/tests/src/test_intrusive_future_notify_destroy.c +++ b/tests/src/test_intrusive_future_notify_destroy.c @@ -4,8 +4,12 @@ #include "intrusive_fixture_utils.h" #include "message.h" #include +#include #include +static_assert(CY_CONFIG_REQUEST_ACK_RETENTION_us == SESSION_LIFETIME, // NOLINT(misc-redundant-expression) + "request ACK retention should default to the session lifetime"); + typedef struct { cy_platform_t platform; @@ -699,7 +703,14 @@ static void test_request_notify_on_response_reliable_destroy(void) TEST_ASSERT_EQUAL_size_t(1U, cap.calls); TEST_ASSERT_TRUE(cap.saw_done); TEST_ASSERT_EQUAL_INT(CY_OK, cap.last_error); - fixture_spin_to(&fixture, fixture.now + (SESSION_LIFETIME / 2) + 1); + + // The callback destroyed the future during the FIRST reliable response, so the record it hands over was + // created moments earlier in the very same call. It is reaped by poll(); this fixture has exactly one + // topic and cy_spin_once() runs poll() once, so a single spin past dead_at visits it. + cy_topic_t* const topic = cy_publisher_topic(pub); + TEST_ASSERT_NOT_NULL(topic->request_acks_by_tag); + fixture_spin_to(&fixture, fixture.now + (CY_CONFIG_REQUEST_ACK_RETENTION_us) + 1); + TEST_ASSERT_NULL(topic->request_acks_by_tag); // proves the poll sweep, not just fixture_deinit's teardown cy_unadvertise(pub); fixture_deinit(&fixture); @@ -719,7 +730,8 @@ static void test_request_notify_on_response_reliable_oom_silent(void) destroy_capture_t cap = { 0 }; set_destroy_callback(fut, &cap); - fixture_fail_alloc_size(&fixture, sizeof(request_future_remote_t), 1U); + // The response below carries seqno 0, i.e. the solo path, whose only allocation is the ack record itself. + fixture_fail_alloc_size(&fixture, sizeof(request_ack_t), 1U); dispatch_response_message(&fixture, UINT64_C(0xB003), header_rsp_rel, @@ -730,14 +742,17 @@ static void test_request_notify_on_response_reliable_oom_silent(void) 0xADU, fixture.now + 1); + TEST_ASSERT_EQUAL_size_t(0U, fixture.fail_size_count); // the injection fired rather than being absorbed TEST_ASSERT_EQUAL_size_t(0U, cap.calls); TEST_ASSERT_FALSE(cy_future_done(fut)); TEST_ASSERT_EQUAL_INT(CY_OK, cy_future_error(fut)); TEST_ASSERT_EQUAL_size_t(0U, fixture.unicast_count); TEST_ASSERT_TRUE(fixture.async_error_count > 0U); TEST_ASSERT_EQUAL_INT(CY_ERR_MEMORY, fixture.last_async_error); + TEST_ASSERT_NULL(((const request_future_t*)fut)->ack); // nothing half-built survived cy_future_destroy(fut); + TEST_ASSERT_NULL(cy_publisher_topic(pub)->request_acks_by_tag); // acked nothing -> no record retained cy_unadvertise(pub); fixture_deinit(&fixture); } @@ -780,29 +795,54 @@ static void test_request_notify_timeout_no_remote_destroy(void) fixture_deinit(&fixture); } +// Liveness timeout -> notify -> the callback destroys -> a record is STILL handed over, because reliable +// responses were acked earlier. Built on a real advertised topic: a stack topic that is never inserted into +// topics_by_hash is unreachable by poll(), by cy_destroy() and by dispatch, so its record would leak. static void test_request_notify_timeout_with_remote_destroy(void) { fixture_t fixture; fixture_init(&fixture); - cy_topic_t topic; - request_future_t* fut = make_request_future_manual(&fixture, &topic, UINT64_C(0xCC02), 1000); - request_future_remote_factory_context_t fac = { .cy = fixture.cy, .remote_id = UINT64_C(0xD001) }; - request_future_remote_t* const remote = (request_future_remote_t*)cavl2_find_or_insert( - &fut->remote_by_id, &fac.remote_id, request_future_remote_cavl_compare, &fac, request_future_remote_cavl_factory); - TEST_ASSERT_NOT_NULL(remote); + cy_publisher_t* const pub = cy_advertise_client(fixture.cy, cy_str("notify/request/timeout_rec"), 16U); + TEST_ASSERT_NOT_NULL(pub); + cy_priority_set(pub, cy_prio_exceptional); + const cy_bytes_t msg = { .size = 1U, .data = "K", .next = NULL }; + cy_future_t* const fut = cy_request(pub, fixture.now + 150000, 60000, msg); + TEST_ASSERT_NOT_NULL(fut); + cy_topic_t* const topic = cy_publisher_topic(pub); + + // Accept a reliable response and consume it BEFORE arming the destroy-on-notify callback, so that the + // callback is triggered by the liveness timeout rather than by the response itself. + dispatch_response_message(&fixture, + UINT64_C(0xD001), + header_rsp_rel, + 0x14U, + 0U, + last_outgoing_hash_multicast(&fixture), + last_outgoing_tag_multicast(&fixture), + 0xAEU, + fixture.now + 1); + TEST_ASSERT_EQUAL_UINT64(1U, cy_response_count(fut)); + const cy_response_t moved = cy_response_move(fut); + TEST_ASSERT_NOT_NULL(moved.message.content); + cy_message_refcount_dec(moved.message.content); + TEST_ASSERT_NOT_NULL(((const request_future_t*)fut)->ack); + TEST_ASSERT_NULL(topic->request_acks_by_tag); // still owned by the live future, NOT on the topic destroy_capture_t cap = { 0 }; - set_destroy_callback(&fut->base, &cap); + set_destroy_callback(fut, &cap); - fixture_spin_to(&fixture, fixture.now + 1001); + fixture_spin_to(&fixture, fixture.now + 60002); TEST_ASSERT_EQUAL_size_t(1U, cap.calls); TEST_ASSERT_TRUE(cap.saw_done); TEST_ASSERT_EQUAL_INT(CY_ERR_LIVENESS, cap.last_error); - TEST_ASSERT_NOT_NULL(topic.request_futures_by_tag); + TEST_ASSERT_NULL(topic->request_futures_by_tag); + TEST_ASSERT_NOT_NULL(topic->request_acks_by_tag); // handed over on the way out - fixture_spin_to(&fixture, fixture.now + (SESSION_LIFETIME / 2) + 1); - TEST_ASSERT_NULL(topic.request_futures_by_tag); + // Reaped by poll(); one topic exists, so a single spin past dead_at suffices. + fixture_spin_to(&fixture, fixture.now + (CY_CONFIG_REQUEST_ACK_RETENTION_us) + 1); + TEST_ASSERT_NULL(topic->request_acks_by_tag); + cy_unadvertise(pub); fixture_deinit(&fixture); } @@ -842,15 +882,125 @@ static void test_publish_pending_future_destroy_then_unadvertise_clean(void) fixture_deinit(&fixture); } -// Regression (L5): destroying a request future that acked a reliable response leaves a finalized "zombie" -// in request_futures_by_tag (to absorb duplicate retransmits). Tearing the node down while it is still pending -// must let topic_destroy reap it -- exercising the retained library-owned cleanup via cy_destroy, not a timeout. -static void test_request_zombie_reaped_by_cy_destroy_after_unadvertise(void) +// Invariant: a record is on the topic's expiry structures IFF its future is gone. A live future can easily +// outlive CY_CONFIG_REQUEST_ACK_RETENTION_us because every response re-arms the liveness timer, so if it were enlisted +// at creation instead of at handoff, the poll sweep would free it out from under the live future. +static void test_request_live_future_outlives_retention_window(void) +{ + fixture_t fixture; + fixture_init(&fixture); + + cy_publisher_t* const pub = cy_advertise_client(fixture.cy, cy_str("notify/request/long_lived"), 16U); + TEST_ASSERT_NOT_NULL(pub); + cy_priority_set(pub, cy_prio_exceptional); + const cy_bytes_t msg = { .size = 1U, .data = "L", .next = NULL }; + cy_future_t* const fut = cy_request(pub, fixture.now + 150000, SESSION_LIFETIME, msg); + TEST_ASSERT_NOT_NULL(fut); + cy_topic_t* const topic = cy_publisher_topic(pub); + const uint64_t hash = last_outgoing_hash_multicast(&fixture); + const uint64_t tag = last_outgoing_tag_multicast(&fixture); + const uint64_t remote_id = UINT64_C(0xD100); + + dispatch_response_message(&fixture, remote_id, header_rsp_rel, 0x30U, 0U, hash, tag, 0xB0U, fixture.now + 1); + TEST_ASSERT_EQUAL_UINT64(1U, cy_response_count(fut)); + const cy_response_t moved = cy_response_move(fut); + cy_message_refcount_dec(moved.message.content); + TEST_ASSERT_NOT_NULL(((const request_future_t*)fut)->ack); + TEST_ASSERT_NULL(topic->request_acks_by_tag); // owned by the live future, unreachable from the topic + + // Spin well past the retention window, re-arming liveness mid-way with a fresh response so that the future + // stays PENDING by the stated mechanism rather than merely by nobody destroying it. The sweep runs on every + // poll() and must not touch the live record. The two 50 s legs exceed the 60 s default retention together, + // while each stays short of the 60 s liveness timeout that would otherwise fire. + for (unsigned leg = 0; leg < 2U; leg++) { + for (unsigned i = 0; i < 5U; i++) { + fixture_spin_to(&fixture, fixture.now + (SESSION_LIFETIME / 6)); + TEST_ASSERT_NULL(topic->request_acks_by_tag); + TEST_ASSERT_NOT_NULL(topic->request_futures_by_tag); + TEST_ASSERT_FALSE(cy_future_done(fut)); // liveness still armed -- the re-arm is what keeps it so + } + if (leg == 0U) { // A second response resets the liveness window, buying another full leg. + dispatch_response_message( + &fixture, remote_id, header_rsp_rel, 0x31U, 1U, hash, tag, 0xB1U, fixture.now + 1); + TEST_ASSERT_EQUAL_UINT64(2U, cy_response_count(fut)); + const cy_response_t second = cy_response_move(fut); + cy_message_refcount_dec(second.message.content); + } + } + // Total elapsed is ~100 s, well over the 60 s liveness timeout: without the mid-way re-arm the future would + // have materialized long ago, so this assertion is what proves the mechanism. + TEST_ASSERT_FALSE(cy_future_done(fut)); + TEST_ASSERT_NOT_NULL(((const request_future_t*)fut)->ack); + + // A retransmit arriving long after the window is still acked from the live future's own state, and is + // still deduplicated -- the application must not see it twice. + const size_t unicast_before = fixture.unicast_count; + dispatch_response_message(&fixture, remote_id, header_rsp_rel, 0x30U, 0U, hash, tag, 0xB0U, fixture.now + 1); + TEST_ASSERT_EQUAL_size_t(unicast_before + 1U, fixture.unicast_count); + TEST_ASSERT_EQUAL_UINT8(header_rsp_ack, fixture.last_unicast[0]); + TEST_ASSERT_EQUAL_UINT64(2U, cy_response_count(fut)); // deduplicated -- still just the two genuine responses + + // The record is promoted (seqno 0 then 1 from the same remote), and is drained by topic_destroy via + // cy_destroy rather than by a sweep; fixture_deinit's heap check is the verdict. + cy_future_destroy(fut); + TEST_ASSERT_NOT_NULL(topic->request_acks_by_tag); // only now does it reach the topic + cy_unadvertise(pub); + fixture_deinit(&fixture); +} + +// poll() sweeps one topic per call, so records on several topics are reaped over successive spins rather than +// all at once. Every other test here relies on exactly one topic existing, which cannot show that. +static void test_request_ack_records_reaped_across_topics(void) +{ + fixture_t fixture; + fixture_init(&fixture); + + cy_publisher_t* pub[3]; + cy_topic_t* topic[3]; + for (unsigned i = 0; i < 3U; i++) { + char name[32]; + (void)snprintf(name, sizeof(name), "sweep/topic/%u", i); + pub[i] = cy_advertise_client(fixture.cy, cy_str(name), 16U); + TEST_ASSERT_NOT_NULL(pub[i]); + cy_priority_set(pub[i], cy_prio_exceptional); + const cy_bytes_t msg = { .size = 1U, .data = "M", .next = NULL }; + cy_future_t* const fut = cy_request(pub[i], fixture.now + 150000, 60000, msg); + TEST_ASSERT_NOT_NULL(fut); + topic[i] = cy_publisher_topic(pub[i]); + dispatch_response_message(&fixture, + UINT64_C(0xE000) + i, + header_rsp_rel, + (uint8_t)(0x40U + i), + 0U, + last_outgoing_hash_multicast(&fixture), + last_outgoing_tag_multicast(&fixture), + 0xC0U, + fixture.now + 1); + cy_future_destroy(fut); // hands the record to this topic + TEST_ASSERT_NOT_NULL(topic[i]->request_acks_by_tag); + } + + fixture_spin_to(&fixture, fixture.now + (CY_CONFIG_REQUEST_ACK_RETENTION_us) + 1); + + for (unsigned i = 0; i < 8U; i++) { + fixture_spin_to(&fixture, fixture.now + 1); + } + for (unsigned i = 0; i < 3U; i++) { + TEST_ASSERT_NULL(topic[i]->request_acks_by_tag); // the cursor reached every topic + cy_unadvertise(pub[i]); + } + fixture_deinit(&fixture); +} + +// Regression (L5): destroying a request future that acked a reliable response leaves an ack record behind +// in request_acks_by_tag (to absorb duplicate retransmits). Tearing the node down while the record is still +// live must let topic_destroy reap it -- exercising the library-owned cleanup via cy_destroy, not a sweep. +static void test_request_ack_record_reaped_by_cy_destroy_after_unadvertise(void) { fixture_t fixture; fixture_init(&fixture); - cy_publisher_t* const pub = cy_advertise_client(fixture.cy, cy_str("cleanup/request/zombie"), 16U); + cy_publisher_t* const pub = cy_advertise_client(fixture.cy, cy_str("cleanup/request/ack_record"), 16U); TEST_ASSERT_NOT_NULL(pub); cy_priority_set(pub, cy_prio_exceptional); @@ -858,29 +1008,41 @@ static void test_request_zombie_reaped_by_cy_destroy_after_unadvertise(void) cy_future_t* const fut = cy_request(pub, fixture.now + 150000, 60000, msg); TEST_ASSERT_NOT_NULL(fut); - // Reliable response so the request future acks it and records the remote -- this is what spawns the zombie. - dispatch_response_message(&fixture, - UINT64_C(0xB010), - header_rsp_rel, - 0x21U, - 0U, - last_outgoing_hash_multicast(&fixture), - last_outgoing_tag_multicast(&fixture), - 0xAEU, - fixture.now + 1); + // Two distinct responders, so the record is promoted and owns a multi-node remote tree. topic_destroy must + // drain that tree, not just free the record. + const uint64_t hash = last_outgoing_hash_multicast(&fixture); + const uint64_t tag = last_outgoing_tag_multicast(&fixture); + dispatch_response_message(&fixture, UINT64_C(0xB010), header_rsp_rel, 0x21U, 0U, hash, tag, 0xAEU, fixture.now + 1); TEST_ASSERT_TRUE(cy_future_done(fut)); TEST_ASSERT_EQUAL_INT(CY_OK, cy_future_error(fut)); + dispatch_response_message(&fixture, UINT64_C(0xB011), header_rsp_rel, 0x22U, 0U, hash, tag, 0xAFU, fixture.now + 2); + TEST_ASSERT_EQUAL_UINT64(2U, cy_response_count(fut)); cy_topic_t* const topic = cy_publisher_topic(pub); TEST_ASSERT_NOT_NULL(topic); - // Destroying the future leaves a finalized zombie behind. + // Destroying the future hands its deduplication state over as an ack record and frees the future itself. cy_future_destroy(fut); - TEST_ASSERT_NOT_NULL(topic->request_futures_by_tag); - const request_future_t* const zombie = (const request_future_t*)topic->request_futures_by_tag; - TEST_ASSERT_TRUE(zombie->finalized); - - // Reaped by cy_destroy -> topic_destroy without waiting for the zombie's timeout; deinit checks heaps clean. + TEST_ASSERT_NULL(topic->request_futures_by_tag); + TEST_ASSERT_NOT_NULL(topic->request_acks_by_tag); + const request_ack_t* const record = (const request_ack_t*)topic->request_acks_by_tag; + TEST_ASSERT_FALSE(record->solo); // promoted by the second responder + TEST_ASSERT_NOT_NULL(record->u.tree); + + // The retained multi-remote record answers over the wire: both known remotes ack, a third nacks. + const size_t unicast_before = fixture.unicast_count; + dispatch_response_message(&fixture, UINT64_C(0xB010), header_rsp_rel, 0x23U, 0U, hash, tag, 0xB0U, fixture.now + 3); + TEST_ASSERT_EQUAL_UINT8(header_rsp_ack, fixture.last_unicast[0]); + dispatch_response_message(&fixture, UINT64_C(0xB011), header_rsp_rel, 0x24U, 0U, hash, tag, 0xB1U, fixture.now + 4); + TEST_ASSERT_EQUAL_UINT8(header_rsp_ack, fixture.last_unicast[0]); + dispatch_response_message(&fixture, UINT64_C(0xB0FF), header_rsp_rel, 0x25U, 0U, hash, tag, 0xB2U, fixture.now + 5); + TEST_ASSERT_EQUAL_UINT8(header_rsp_nack, fixture.last_unicast[0]); // responder that appeared after the future + TEST_ASSERT_EQUAL_size_t(unicast_before + 3U, fixture.unicast_count); + // Query-only: the unknown responder was answered without being recorded. + const uint64_t unknown = UINT64_C(0xB0FF); + TEST_ASSERT_NULL(cavl2_find(record->u.tree, &unknown, request_future_remote_cavl_compare)); + + // Reaped by cy_destroy -> topic_destroy without waiting for any sweep; deinit checks the heaps are clean. cy_unadvertise(pub); TEST_ASSERT_EQUAL_size_t(0U, fixture.async_error_count); fixture_deinit(&fixture); @@ -1247,7 +1409,9 @@ int main(void) RUN_TEST(test_request_notify_timeout_no_remote_destroy); RUN_TEST(test_request_notify_timeout_with_remote_destroy); RUN_TEST(test_publish_pending_future_destroy_then_unadvertise_clean); - RUN_TEST(test_request_zombie_reaped_by_cy_destroy_after_unadvertise); + RUN_TEST(test_request_live_future_outlives_retention_window); + RUN_TEST(test_request_ack_records_reaped_across_topics); + RUN_TEST(test_request_ack_record_reaped_by_cy_destroy_after_unadvertise); RUN_TEST(test_request_callback_set_after_done_immediate_destroy); RUN_TEST(test_subscriber_notify_arrival_unordered_destroy); RUN_TEST(test_subscriber_notify_arrival_ordered_ejection_destroy); diff --git a/tests/src/test_intrusive_publish_reliable.c b/tests/src/test_intrusive_publish_reliable.c index 924328b..7a9fab8 100644 --- a/tests/src/test_intrusive_publish_reliable.c +++ b/tests/src/test_intrusive_publish_reliable.c @@ -10,6 +10,7 @@ typedef struct cy_platform_vtable_t vtable; cy_t cy; guarded_heap_t heap; + cy_us_t now; size_t fail_after; ///< Fail N-th new allocation if new_alloc_count >= fail_after. size_t new_alloc_count; ///< Counts new allocations only, excludes realloc/free. @@ -25,6 +26,10 @@ typedef struct byte_t last_unicast[HEADER_BYTES]; } fixture_t; +// Tracing calls cy_now() on every CY_TRACE, so the clock hook must exist even though this fixture drives +// time directly. Without it a CY_CONFIG_TRACE=1 build calls through a null pointer. +static cy_us_t fixture_now(cy_platform_t* const platform) { return ((const fixture_t*)platform)->now; } + static void* fixture_realloc(cy_platform_t* const platform, void* const ptr, const size_t size) { fixture_t* const self = (fixture_t*)platform; @@ -85,6 +90,7 @@ static void fixture_init(fixture_t* const self) self->platform.subject_id_modulus = (uint32_t)CY_SUBJECT_ID_MODULUS_16bit; self->platform.cy = &self->cy; self->vtable.realloc = fixture_realloc; + self->vtable.now = fixture_now; self->vtable.unicast = fixture_unicast_send; self->cy.platform = &self->platform; self->diag = (cy_diag_t){ .next = NULL, .user_context = CY_USER_CONTEXT_EMPTY, .vtable = &fixture_diag_vtable }; diff --git a/tests/src/test_intrusive_reordering.c b/tests/src/test_intrusive_reordering.c index 5a14960..fbbf782 100644 --- a/tests/src/test_intrusive_reordering.c +++ b/tests/src/test_intrusive_reordering.c @@ -29,6 +29,7 @@ typedef struct { reorder_fixture_t fixture; subscriber_root_t root; + wkv_node_t root_name_node; // topic_couple() traces the root name under CY_CONFIG_TRACE subscriber_t sub; cy_topic_t topic; reordering_t rr; @@ -106,7 +107,8 @@ static void reorder_env_init(reorder_env_t* const self) self->fixture.last_async_error_line = 0; olga_init(&self->fixture.cy.olga, &self->fixture.cy, olga_now); - self->root.cy = &self->fixture.cy; + self->root.cy = &self->fixture.cy; + self->root.index_name = &self->root_name_node; self->sub.base.index = TREE_NULL; self->sub.base.key = 0; diff --git a/tests/src/test_intrusive_rpc.c b/tests/src/test_intrusive_rpc.c index e7ed6c4..39217b6 100644 --- a/tests/src/test_intrusive_rpc.c +++ b/tests/src/test_intrusive_rpc.c @@ -1,3 +1,4 @@ +#define CY_CONFIG_REQUEST_ACK_RETENTION_us 7000000LL #include // NOLINT(bugprone-suspicious-include) #include #include "guarded_heap.h" @@ -13,7 +14,8 @@ typedef struct guarded_heap_t heap; size_t fail_size; - size_t fail_size_count; + size_t fail_size_skip; // Matching allocations to let through before failing; both promotion allocations + size_t fail_size_count; // are the same size, so failing only the second one requires a skip. cy_us_t now; uint64_t random_state; @@ -43,13 +45,19 @@ typedef struct static size_t g_dummy_publish_dispose_count = 0U; // NOLINT(*-non-const-global-variables) +static_assert(sizeof(request_ack_t) != sizeof(request_future_remote_t), "size-keyed OOM injection is ambiguous"); + static void* fixture_realloc(cy_platform_t* const platform, void* const ptr, const size_t size) { fixture_t* const self = (fixture_t*)platform; if ((ptr == NULL) && (size > 0U)) { if ((self->fail_size_count > 0U) && (self->fail_size == size)) { - self->fail_size_count--; - return NULL; + if (self->fail_size_skip > 0U) { + self->fail_size_skip--; + } else { + self->fail_size_count--; + return NULL; + } } } return guarded_heap_realloc(&self->heap, ptr, size); @@ -146,12 +154,21 @@ static void fixture_init(fixture_t* const self) self->async_error_count = 0U; } -static void fixture_fail_alloc_size(fixture_t* const self, const size_t size, const size_t count) +static void fixture_fail_alloc_size_after(fixture_t* const self, + const size_t size, + const size_t skip, + const size_t count) { self->fail_size = size; + self->fail_size_skip = skip; self->fail_size_count = count; } +static void fixture_fail_alloc_size(fixture_t* const self, const size_t size, const size_t count) +{ + fixture_fail_alloc_size_after(self, size, 0U, count); +} + static void fixture_advance_to(fixture_t* const self, const cy_us_t now) { self->now = now; @@ -202,7 +219,7 @@ static request_future_t* make_request_future(fixture_t* const fixture, out->liveness_timeout = liveness_timeout; out->last_response.message.timestamp = BIG_BANG; out->last_response.message.content = NULL; - out->remote_by_id = NULL; + out->ack = NULL; const bool insert_ok = future_index_insert(&out->base, &topic->request_futures_by_tag, key); TEST_ASSERT_TRUE(insert_ok); future_deadline_arm(&out->base, fixture->now + liveness_timeout); @@ -267,9 +284,26 @@ static cy_future_t* dummy_publish_new(cy_t* const cy) return &out->base; } +// Returns the promoted per-remote node, or NULL if there is no record yet or it is still in the inlined solo shape. +// Use request_ack_is_solo() to assert the solo shape positively rather than inferring it from a NULL here. static request_future_remote_t* request_remote_find(const request_future_t* const fut, const uint64_t remote_id) { - return (request_future_remote_t*)cavl2_find(fut->remote_by_id, &remote_id, request_future_remote_cavl_compare); + if ((fut->ack == NULL) || fut->ack->solo) { + return NULL; + } + return (request_future_remote_t*)cavl2_find(fut->ack->u.tree, &remote_id, request_future_remote_cavl_compare); +} + +static bool request_ack_is_solo(const request_future_t* const fut, const uint64_t remote_id) +{ + return (fut->ack != NULL) && fut->ack->solo && (fut->ack->u.solo_remote_id == remote_id); +} + +// The intrusive fixture drives olga directly and never runs poll(), so retained records are never swept for us. +// This is strictly beyond any record retained at or before the fixture's current time. +static void reap_request_acks(const fixture_t* const fixture, cy_topic_t* const topic) +{ + request_ack_drop_stale(topic, fixture->now + (CY_CONFIG_REQUEST_ACK_RETENTION_us) + 1); } static cy_breadcrumb_t make_test_breadcrumb(const fixture_t* const fixture, @@ -806,7 +840,7 @@ static void test_request_future_destroy_releases_last_response(void) fixture_assert_clean(&fixture); } -static void test_request_future_dispose_zombie_releases_last_response_early(void) +static void test_request_future_dispose_hands_over_and_releases_last_response(void) { fixture_t fixture; fixture_init(&fixture); @@ -816,18 +850,25 @@ static void test_request_future_dispose_zombie_releases_last_response_early(void const cy_lane_t lane = make_lane(88U); const cy_message_ts_t msg = make_message(&fixture, fixture.now + 6U, 0x41U); - TEST_ASSERT_EQUAL_INT(response_rx_ack, request_on_response(fut, 9U, msg, true, lane)); // creates remote state. + TEST_ASSERT_EQUAL_INT(response_rx_ack, request_on_response(fut, 9U, msg, true, lane)); // creates the ack record cy_message_refcount_dec(msg.content); // release local copy assert_message_counters(0U, 1U); + TEST_ASSERT_NOT_NULL(fut->ack); + // The future is freed outright; the record is handed to the topic. Nothing of `fut` may be read after this. cy_future_destroy(&fut->base); - TEST_ASSERT_TRUE(fut->finalized); - TEST_ASSERT_NULL(fut->last_response.message.content); - TEST_ASSERT_NOT_NULL(topic.request_futures_by_tag); + TEST_ASSERT_NULL(topic.request_futures_by_tag); + TEST_ASSERT_NOT_NULL(topic.request_acks_by_tag); + TEST_ASSERT_NOT_NULL(topic.request_acks_by_expiry.head); assert_message_counters(1U, 0U); // dispose() released the retained response immediately. - fixture_advance_to(&fixture, fixture.now + (SESSION_LIFETIME / 2) + 1); - TEST_ASSERT_NULL(topic.request_futures_by_tag); + // seqno 9 was the first response from this remote, so the record is promoted, not solo. + const request_ack_t* const ack = (const request_ack_t*)topic.request_acks_by_tag; + TEST_ASSERT_FALSE(ack->solo); + TEST_ASSERT_EQUAL_INT64(fixture.now + (CY_CONFIG_REQUEST_ACK_RETENTION_us), ack->dead_at); + + reap_request_acks(&fixture, &topic); + TEST_ASSERT_NULL(topic.request_acks_by_tag); assert_message_counters(1U, 0U); fixture_assert_clean(&fixture); } @@ -872,50 +913,382 @@ static void test_request_on_response_reliable_dedup_and_ordering(void) cy_message_refcount_dec(msg.content); TEST_ASSERT_EQUAL_UINT64(2U, fut->response_count); - cy_future_destroy(&fut->base); // becomes zombie because remote states exist - TEST_ASSERT_NOT_NULL(topic.request_futures_by_tag); - fixture_advance_to(&fixture, fixture.now + (SESSION_LIFETIME / 2) + 1); + cy_future_destroy(&fut->base); // hands over the ack record because reliable responses were acked TEST_ASSERT_NULL(topic.request_futures_by_tag); + TEST_ASSERT_NOT_NULL(topic.request_acks_by_tag); + reap_request_acks(&fixture, &topic); + TEST_ASSERT_NULL(topic.request_acks_by_tag); fixture_assert_clean(&fixture); } -static void test_request_on_response_zombie_ack_seen_nack_unseen(void) +// After the future is gone the retained record answers, and it is query-only: it must never insert a remote, +// never shift a bitmap and never set a bit. +static void test_request_ack_record_ack_seen_nack_unseen(void) { fixture_t fixture; fixture_init(&fixture); + const uint64_t topic_hash = UINT64_C(0x5150515051505150); + const uint64_t message_tag = UINT64_C(1003); + const uint64_t remote_id = 555U; cy_topic_t topic; - request_future_t* fut = make_request_future(&fixture, &topic, UINT64_C(1003), 20000); - const cy_lane_t lane = make_lane(555U); + request_future_t* fut = make_indexed_request_future(&fixture, &topic, message_tag, 20000, topic_hash); cy_message_ts_t msg = make_message(&fixture, fixture.now + 1U, 20U); - TEST_ASSERT_EQUAL_INT(response_rx_ack, request_on_response(fut, 5U, msg, true, lane)); + TEST_ASSERT_EQUAL_INT(response_rx_ack, request_on_response(fut, 5U, msg, true, make_lane(remote_id))); + cy_message_refcount_dec(msg.content); + + cy_future_destroy(&fut->base); // `fut` is freed here; only the record survives. + TEST_ASSERT_NULL(topic.request_futures_by_tag); + TEST_ASSERT_NOT_NULL(topic.request_acks_by_tag); + request_ack_t* const ack = (request_ack_t*)topic.request_acks_by_tag; + + TEST_ASSERT_TRUE(request_ack_test(ack, remote_id, 5U)); // seen -> ack + TEST_ASSERT_FALSE(request_ack_test(ack, remote_id, 6U)); // above the frontier -> nack + TEST_ASSERT_FALSE(request_ack_test(ack, remote_id, 4U)); // below the frontier, never acked -> nack + TEST_ASSERT_FALSE(request_ack_test(ack, 556U, 5U)); // unknown remote -> nack + + request_future_remote_t* const remote = (request_future_remote_t*)ack->u.tree; + TEST_ASSERT_NOT_NULL(remote); + + // Same verdicts through the real wire path, including the best-effort case which emits nothing, and one + // dispatch from a remote the record has never seen -- the case that would insert a node if the path mutated. + const size_t sent_before = fixture.unicast_send_count; + dispatch_response_control( + &fixture, (byte_t)header_rsp_rel, 0x11U, 5U, topic_hash, message_tag, remote_id, fixture.now + 2U, false); + TEST_ASSERT_EQUAL_size_t(sent_before + 1U, fixture.unicast_send_count); + TEST_ASSERT_EQUAL_UINT8(header_rsp_ack, fixture.last_unicast[0]); + dispatch_response_control( + &fixture, (byte_t)header_rsp_rel, 0x12U, 6U, topic_hash, message_tag, remote_id, fixture.now + 3U, false); + TEST_ASSERT_EQUAL_size_t(sent_before + 2U, fixture.unicast_send_count); + TEST_ASSERT_EQUAL_UINT8(header_rsp_nack, fixture.last_unicast[0]); + dispatch_response_control( + &fixture, (byte_t)header_rsp_rel, 0x14U, 0U, topic_hash, message_tag, 556U, fixture.now + 4U, false); + TEST_ASSERT_EQUAL_size_t(sent_before + 3U, fixture.unicast_send_count); + TEST_ASSERT_EQUAL_UINT8(header_rsp_nack, fixture.last_unicast[0]); + dispatch_response_control( + &fixture, (byte_t)header_rsp_be, 0x13U, 0U, topic_hash, message_tag, remote_id, fixture.now + 5U, false); + TEST_ASSERT_EQUAL_size_t(sent_before + 3U, fixture.unicast_send_count); // best-effort: no control frame at all + + // Query-only: after everything above, including the unknown-remote dispatch, the record is unchanged. + TEST_ASSERT_FALSE(ack->solo); + TEST_ASSERT_EQUAL_PTR(remote, (request_future_remote_t*)ack->u.tree); + TEST_ASSERT_NULL(cavl2_next_greater(&remote->index_by_remote_id)); // remote 556 was not inserted + TEST_ASSERT_NULL(remote->index_by_remote_id.lr[0]); // ...on either side + TEST_ASSERT_EQUAL_UINT64(5U, remote->seqno_top); // frontier not advanced by seqno 6 + TEST_ASSERT_TRUE(bitmap_test(remote->seqno_acked, 0U)); + TEST_ASSERT_FALSE(bitmap_test(remote->seqno_acked, 1U)); // seqno 4 did not set a bit + + reap_request_acks(&fixture, &topic); + TEST_ASSERT_NULL(topic.request_acks_by_tag); + unindex_request_topic(&fixture, &topic); + fixture_assert_clean(&fixture); +} + +// A single responder whose first reliable response carries seqno 0 is stored inline: no tree node, no bitmap, +// no second allocation. This is the shape the whole optimization exists for. +static void test_request_ack_solo_claim_and_duplicate(void) +{ + fixture_t fixture; + fixture_init(&fixture); + + cy_topic_t topic; + request_future_t* fut = make_request_future(&fixture, &topic, UINT64_C(2001), 20000); + const cy_lane_t lane = make_lane(0xA1U); + + cy_message_ts_t msg = make_message(&fixture, fixture.now + 1U, 1U); + const size_t frags_before = guarded_heap_allocated_fragments(&fixture.heap); + TEST_ASSERT_EQUAL_INT(response_rx_ack, request_on_response(fut, 0U, msg, true, lane)); + const size_t frags_after = guarded_heap_allocated_fragments(&fixture.heap); + cy_message_refcount_dec(msg.content); // release before asserting so a failure cannot leak into tearDown + TEST_ASSERT_EQUAL_size_t(frags_before + 1U, frags_after); // the record only -- no per-remote node + TEST_ASSERT_TRUE(request_ack_is_solo(fut, lane.id)); + TEST_ASSERT_EQUAL_UINT64(1U, fut->response_count); + + // A duplicate of seqno 0 is acked straight from the inlined slot and must not promote. + msg = make_message(&fixture, fixture.now + 2U, 2U); + TEST_ASSERT_EQUAL_INT(response_rx_ack, request_on_response(fut, 0U, msg, true, lane)); cy_message_refcount_dec(msg.content); + TEST_ASSERT_TRUE(request_ack_is_solo(fut, lane.id)); + TEST_ASSERT_EQUAL_UINT64(1U, fut->response_count); // deduplicated, the app saw it once + // Both NACK branches of the retained solo shape. Neither is reachable through the promoted shape, and line + // coverage cannot distinguish them from the ACK case above because they share the return statement. cy_future_destroy(&fut->base); - TEST_ASSERT_NOT_NULL(topic.request_futures_by_tag); - TEST_ASSERT_TRUE(fut->finalized); - TEST_ASSERT_NULL(fut->last_response.message.content); - TEST_ASSERT_TRUE(future_deadline_armed(&fut->base)); + const request_ack_t* const ack = (const request_ack_t*)topic.request_acks_by_tag; + TEST_ASSERT_NOT_NULL(ack); + TEST_ASSERT_TRUE(ack->solo); + TEST_ASSERT_TRUE(request_ack_test(ack, lane.id, 0U)); // the inlined ack + TEST_ASSERT_FALSE(request_ack_test(ack, lane.id, 1U)); // right remote, wrong seqno + TEST_ASSERT_FALSE(request_ack_test(ack, lane.id + 1U, 0U)); // wrong remote, right seqno + + reap_request_acks(&fixture, &topic); + fixture_assert_clean(&fixture); +} + +// Promotion by the same remote must reconstruct the inlined ack losslessly: seqno 0's bit is shifted to index s. +static void test_request_ack_solo_promotes_same_remote(void) +{ + fixture_t fixture; + fixture_init(&fixture); + + cy_topic_t topic; + request_future_t* fut = make_request_future(&fixture, &topic, UINT64_C(2002), 20000); + const cy_lane_t lane = make_lane(0xA2U); + + cy_message_ts_t msg = make_message(&fixture, fixture.now + 1U, 1U); + TEST_ASSERT_EQUAL_INT(response_rx_ack, request_on_response(fut, 0U, msg, true, lane)); + cy_message_refcount_dec(msg.content); + TEST_ASSERT_TRUE(request_ack_is_solo(fut, lane.id)); + + msg = make_message(&fixture, fixture.now + 2U, 2U); + TEST_ASSERT_EQUAL_INT(response_rx_ack, request_on_response(fut, 3U, msg, true, lane)); + cy_message_refcount_dec(msg.content); + TEST_ASSERT_FALSE(request_ack_is_solo(fut, lane.id)); + const request_future_remote_t* const remote = request_remote_find(fut, lane.id); + TEST_ASSERT_NOT_NULL(remote); + TEST_ASSERT_EQUAL_UINT64(3U, remote->seqno_top); + TEST_ASSERT_TRUE(bitmap_test(remote->seqno_acked, 0U)); // seqno 3 + TEST_ASSERT_TRUE(bitmap_test(remote->seqno_acked, 3U)); // seqno 0, carried across the shift + TEST_ASSERT_FALSE(bitmap_test(remote->seqno_acked, 1U)); + TEST_ASSERT_FALSE(bitmap_test(remote->seqno_acked, 2U)); + + // The original seqno-0 ack is still honoured after promotion. + msg = make_message(&fixture, fixture.now + 3U, 3U); + TEST_ASSERT_EQUAL_INT(response_rx_ack, request_on_response(fut, 0U, msg, true, lane)); + cy_message_refcount_dec(msg.content); + TEST_ASSERT_EQUAL_UINT64(2U, fut->response_count); // seqno 0 and 3 only + + cy_future_destroy(&fut->base); + reap_request_acks(&fixture, &topic); + fixture_assert_clean(&fixture); +} + +// A second responder promotes too. The solo remote is migrated FIRST so that a failure on the second +// allocation cannot lose its ack. +static void test_request_ack_solo_promotes_second_remote(void) +{ + fixture_t fixture; + fixture_init(&fixture); + + cy_topic_t topic; + request_future_t* fut = make_request_future(&fixture, &topic, UINT64_C(2003), 20000); + const cy_lane_t r = make_lane(0xA3U); + const cy_lane_t s = make_lane(0xB3U); + + cy_message_ts_t msg = make_message(&fixture, fixture.now + 1U, 1U); + TEST_ASSERT_EQUAL_INT(response_rx_ack, request_on_response(fut, 0U, msg, true, r)); + cy_message_refcount_dec(msg.content); + TEST_ASSERT_TRUE(request_ack_is_solo(fut, r.id)); + + msg = make_message(&fixture, fixture.now + 2U, 2U); + TEST_ASSERT_EQUAL_INT(response_rx_ack, request_on_response(fut, 0U, msg, true, s)); + cy_message_refcount_dec(msg.content); + TEST_ASSERT_FALSE(request_ack_is_solo(fut, r.id)); + + const request_future_remote_t* const rr = request_remote_find(fut, r.id); + const request_future_remote_t* const ss = request_remote_find(fut, s.id); + TEST_ASSERT_NOT_NULL(rr); + TEST_ASSERT_NOT_NULL(ss); + TEST_ASSERT_EQUAL_UINT64(0U, rr->seqno_top); + TEST_ASSERT_TRUE(bitmap_test(rr->seqno_acked, 0U)); // the migrated solo ack + TEST_ASSERT_EQUAL_UINT64(0U, ss->seqno_top); + TEST_ASSERT_TRUE(bitmap_test(ss->seqno_acked, 0U)); + + // Both remotes' seqno-0 acks survive. + msg = make_message(&fixture, fixture.now + 3U, 3U); + TEST_ASSERT_EQUAL_INT(response_rx_ack, request_on_response(fut, 0U, msg, true, r)); + cy_message_refcount_dec(msg.content); + msg = make_message(&fixture, fixture.now + 4U, 4U); + TEST_ASSERT_EQUAL_INT(response_rx_ack, request_on_response(fut, 0U, msg, true, s)); + cy_message_refcount_dec(msg.content); + TEST_ASSERT_EQUAL_UINT64(2U, fut->response_count); + + cy_future_destroy(&fut->base); + TEST_ASSERT_NOT_NULL(topic.request_acks_by_tag); + reap_request_acks(&fixture, &topic); // must drain BOTH remote nodes, not just the record + fixture_assert_clean(&fixture); +} + +// bitmap_shift() resets the whole bitmap when the jump is >= REQUEST_FUTURE_HISTORY. 191 keeps the old ack, +// 192 drops it. This is the shift-side boundary, distinct from the receive-side dist>=192 rejection. +static void test_request_ack_shift_boundary_191_192(void) +{ + for (unsigned k = 0; k < 2U; k++) { + const uint64_t jump = (k == 0U) ? (REQUEST_FUTURE_HISTORY - 1U) : REQUEST_FUTURE_HISTORY; + const bool expect = (k == 0U); // 191 -> still acked; 192 -> forgotten + fixture_t fixture; + fixture_init(&fixture); + + cy_topic_t topic; + request_future_t* fut = make_request_future(&fixture, &topic, UINT64_C(2004) + k, 20000); + const cy_lane_t lane = make_lane(0xA4U); + + cy_message_ts_t msg = make_message(&fixture, fixture.now + 1U, 1U); + TEST_ASSERT_EQUAL_INT(response_rx_ack, request_on_response(fut, 0U, msg, true, lane)); + cy_message_refcount_dec(msg.content); + + msg = make_message(&fixture, fixture.now + 2U, 2U); + TEST_ASSERT_EQUAL_INT(response_rx_ack, request_on_response(fut, jump, msg, true, lane)); + cy_message_refcount_dec(msg.content); + const request_future_remote_t* const remote = request_remote_find(fut, lane.id); + TEST_ASSERT_NOT_NULL(remote); + TEST_ASSERT_EQUAL_UINT64(jump, remote->seqno_top); + TEST_ASSERT_EQUAL_INT(expect, bitmap_test_bounded(remote->seqno_acked, REQUEST_FUTURE_HISTORY, jump)); + + msg = make_message(&fixture, fixture.now + 3U, 3U); + TEST_ASSERT_EQUAL_INT(expect ? response_rx_ack : response_rx_nack, + request_on_response(fut, 0U, msg, true, lane)); + cy_message_refcount_dec(msg.content); + + cy_future_destroy(&fut->base); + reap_request_acks(&fixture, &topic); + fixture_assert_clean(&fixture); + } +} - msg = make_message(&fixture, fixture.now + 2U, 21U); - TEST_ASSERT_EQUAL_INT(response_rx_ack, request_on_response(fut, 5U, msg, true, lane)); // seen -> ack +// Every allocation-failure exit reports exactly one async error, leaves the liveness deadline untouched, and +// preserves whatever was already acked. Retrying after each partial failure must converge. +static void test_request_ack_promotion_oom_preserves_state(void) +{ + fixture_t fixture; + fixture_init(&fixture); + + cy_topic_t topic; + request_future_t* fut = make_request_future(&fixture, &topic, UINT64_C(2010), 20000); + const cy_lane_t r = make_lane(0xA5U); + const cy_lane_t s = make_lane(0xB5U); + + cy_message_ts_t msg = make_message(&fixture, fixture.now + 1U, 1U); + TEST_ASSERT_EQUAL_INT(response_rx_ack, request_on_response(fut, 0U, msg, true, r)); cy_message_refcount_dec(msg.content); + TEST_ASSERT_TRUE(request_ack_is_solo(fut, r.id)); + const cy_us_t deadline_base = fut->base.timeout.deadline; - msg = make_message(&fixture, fixture.now + 3U, 22U); - TEST_ASSERT_EQUAL_INT(response_rx_nack, request_on_response(fut, 6U, msg, true, lane)); // unseen -> nack + // (a) Same-remote promotion fails on the reconstruction allocation -> silent, still solo, ack intact. + fixture_fail_alloc_size(&fixture, sizeof(request_future_remote_t), 1U); + msg = make_message(&fixture, fixture.now + 2U, 2U); + TEST_ASSERT_EQUAL_INT(response_rx_silent, request_on_response(fut, 7U, msg, true, r)); cy_message_refcount_dec(msg.content); + TEST_ASSERT_EQUAL_size_t(0U, fixture.fail_size_count); + TEST_ASSERT_EQUAL_size_t(1U, fixture.async_error_count); + TEST_ASSERT_EQUAL_INT(CY_ERR_MEMORY, fixture.last_async_error); + TEST_ASSERT_TRUE(request_ack_is_solo(fut, r.id)); + TEST_ASSERT_EQUAL_INT64(deadline_base, fut->base.timeout.deadline); // liveness not extended by a dropped response + TEST_ASSERT_EQUAL_UINT64(1U, fut->response_count); - msg = make_message(&fixture, fixture.now + 4U, 23U); - TEST_ASSERT_EQUAL_INT(response_rx_nack, request_on_response(fut, 4U, msg, true, lane)); // unseen older -> nack + // (b) Second-remote promotion fails while reconstructing the solo node -> silent, still solo. + fixture_fail_alloc_size(&fixture, sizeof(request_future_remote_t), 1U); + msg = make_message(&fixture, fixture.now + 3U, 3U); + TEST_ASSERT_EQUAL_INT(response_rx_silent, request_on_response(fut, 0U, msg, true, s)); + cy_message_refcount_dec(msg.content); + TEST_ASSERT_EQUAL_size_t(0U, fixture.fail_size_count); + TEST_ASSERT_EQUAL_size_t(2U, fixture.async_error_count); + TEST_ASSERT_TRUE(request_ack_is_solo(fut, r.id)); + + // (c) Second-remote node fails AFTER the solo node was reconstructed -> silent, record promoted, + // and the solo remote's ack survives losslessly. The naive migration order would lose it here. + fixture_fail_alloc_size_after(&fixture, sizeof(request_future_remote_t), 1U, 1U); + msg = make_message(&fixture, fixture.now + 4U, 4U); + TEST_ASSERT_EQUAL_INT(response_rx_silent, request_on_response(fut, 0U, msg, true, s)); + cy_message_refcount_dec(msg.content); + TEST_ASSERT_EQUAL_size_t(0U, fixture.fail_size_count); + TEST_ASSERT_EQUAL_size_t(3U, fixture.async_error_count); + TEST_ASSERT_FALSE(request_ack_is_solo(fut, r.id)); // partial promotion persists + const request_future_remote_t* const rr = request_remote_find(fut, r.id); + TEST_ASSERT_NOT_NULL(rr); + TEST_ASSERT_TRUE(bitmap_test(rr->seqno_acked, 0U)); // R's seqno-0 ack preserved across the failure + TEST_ASSERT_NULL(request_remote_find(fut, s.id)); + + // (d) Retry after the partial promotion converges, and R's original ack is still honoured. + msg = make_message(&fixture, fixture.now + 5U, 5U); + TEST_ASSERT_EQUAL_INT(response_rx_ack, request_on_response(fut, 0U, msg, true, s)); + cy_message_refcount_dec(msg.content); + TEST_ASSERT_NOT_NULL(request_remote_find(fut, s.id)); + msg = make_message(&fixture, fixture.now + 6U, 6U); + TEST_ASSERT_EQUAL_INT(response_rx_ack, request_on_response(fut, 0U, msg, true, r)); // duplicate -> ack cy_message_refcount_dec(msg.content); + TEST_ASSERT_EQUAL_UINT64(2U, fut->response_count); // R@0 and S@0 - msg = make_message(&fixture, fixture.now + 5U, 24U); - TEST_ASSERT_EQUAL_INT(response_rx_nack, request_on_response(fut, 0U, msg, false, lane)); // zombie reject + cy_future_destroy(&fut->base); + reap_request_acks(&fixture, &topic); + fixture_assert_clean(&fixture); +} + +// A first reliable response with seqno > 0 needs two allocations. If the remote node fails after the record +// was allocated, the record is left in the NONE state, which answers NACK and is NOT retained at disposal. +static void test_request_ack_none_state_not_retained(void) +{ + fixture_t fixture; + fixture_init(&fixture); + + cy_topic_t topic; + request_future_t* fut = make_request_future(&fixture, &topic, UINT64_C(2011), 20000); + const cy_lane_t lane = make_lane(0xA6U); + + fixture_fail_alloc_size(&fixture, sizeof(request_future_remote_t), 1U); + cy_message_ts_t msg = make_message(&fixture, fixture.now + 1U, 1U); + TEST_ASSERT_EQUAL_INT(response_rx_silent, request_on_response(fut, 5U, msg, true, lane)); // seqno > 0 cy_message_refcount_dec(msg.content); + TEST_ASSERT_EQUAL_size_t(0U, fixture.fail_size_count); + TEST_ASSERT_EQUAL_size_t(1U, fixture.async_error_count); + TEST_ASSERT_NOT_NULL(fut->ack); // the record was allocated... + TEST_ASSERT_FALSE(fut->ack->solo); + TEST_ASSERT_NULL(fut->ack->u.tree); // ...and is in the NONE state + TEST_ASSERT_FALSE(request_ack_test(fut->ack, lane.id, 5U)); - fixture_advance_to(&fixture, fixture.now + (SESSION_LIFETIME / 2) + 1); + // A NONE record can only ever answer NACK, so disposal frees it instead of retaining it. + cy_future_destroy(&fut->base); TEST_ASSERT_NULL(topic.request_futures_by_tag); + TEST_ASSERT_NULL(topic.request_acks_by_tag); + TEST_ASSERT_NULL(topic.request_acks_by_expiry.head); + fixture_assert_clean(&fixture); +} + +// The expiry list is re-headed only at handoff, so it stays sorted by dead_at even when futures are disposed +// in a different order than their first responses arrived. The tail sweep depends on that. +static void test_request_ack_expiry_order_follows_disposal(void) +{ + fixture_t fixture; + fixture_init(&fixture); + + cy_topic_t topic_a; + cy_topic_t topic_b; + request_future_t* fa = make_request_future(&fixture, &topic_a, UINT64_C(2020), 20000); + // Both records must live on the same topic for a single expiry list, so re-point the second future. + request_future_t* fb = make_request_future(&fixture, &topic_b, UINT64_C(2021), 20000); + future_index_remove(&fb->base, &topic_b.request_futures_by_tag); + fb->topic = &topic_a; + const bool insert_ok = future_index_insert(&fb->base, &topic_a.request_futures_by_tag, UINT64_C(2021)); + TEST_ASSERT_TRUE(insert_ok); + + cy_message_ts_t msg = make_message(&fixture, fixture.now + 1U, 1U); + TEST_ASSERT_EQUAL_INT(response_rx_ack, request_on_response(fa, 0U, msg, true, make_lane(0xC1U))); + cy_message_refcount_dec(msg.content); + msg = make_message(&fixture, fixture.now + 2U, 2U); + TEST_ASSERT_EQUAL_INT(response_rx_ack, request_on_response(fb, 0U, msg, true, make_lane(0xC2U))); + cy_message_refcount_dec(msg.content); + + // Dispose in the reverse order and advance the clock in between, so dead_at differs. + cy_future_destroy(&fb->base); + fixture_advance_to(&fixture, fixture.now + 1000); + cy_future_destroy(&fa->base); + + // Head is the most recent handoff (fa), tail is the oldest (fb) -- i.e. sorted by dead_at ascending at the tail. + const request_ack_t* const head = LIST_MEMBER(topic_a.request_acks_by_expiry.head, request_ack_t, expiry); + const request_ack_t* const tail = LIST_MEMBER(topic_a.request_acks_by_expiry.tail, request_ack_t, expiry); + TEST_ASSERT_EQUAL_UINT64(UINT64_C(2020), head->tag); + TEST_ASSERT_EQUAL_UINT64(UINT64_C(2021), tail->tag); + TEST_ASSERT_TRUE(tail->dead_at < head->dead_at); + + // Sweeping at a time between the two deadlines must reap only the older one. + request_ack_drop_stale(&topic_a, tail->dead_at + 1); + TEST_ASSERT_NOT_NULL(topic_a.request_acks_by_tag); + TEST_ASSERT_EQUAL_UINT64(UINT64_C(2020), + LIST_MEMBER(topic_a.request_acks_by_expiry.tail, request_ack_t, expiry)->tag); + + reap_request_acks(&fixture, &topic_a); + TEST_ASSERT_NULL(topic_a.request_acks_by_tag); fixture_assert_clean(&fixture); } @@ -934,16 +1307,19 @@ static void test_request_on_response_reliable_oom_stays_pending_silent(void) const size_t callback_base = cap.count; const cy_us_t deadline_base = fut->base.timeout.deadline; + // seqno 0 is the first response from this remote, so it takes the solo path: the only allocation is the + // ack record itself. Failing sizeof(request_future_remote_t) here would not fire at all. cy_message_ts_t msg = make_message(&fixture, fixture.now + 10U, 30U); - fixture_fail_alloc_size(&fixture, sizeof(request_future_remote_t), 1U); + fixture_fail_alloc_size(&fixture, sizeof(request_ack_t), 1U); TEST_ASSERT_EQUAL_INT(response_rx_silent, request_on_response(fut, 0U, msg, true, make_lane(1000U))); cy_message_refcount_dec(msg.content); + TEST_ASSERT_EQUAL_size_t(0U, fixture.fail_size_count); // the injection was consumed, not absorbed elsewhere TEST_ASSERT_EQUAL_size_t(callback_base, cap.count); TEST_ASSERT_TRUE(future_deadline_armed(&fut->base)); TEST_ASSERT_EQUAL_INT64(deadline_base, fut->base.timeout.deadline); TEST_ASSERT_EQUAL_UINT64(0U, fut->response_count); - TEST_ASSERT_NULL(fut->remote_by_id); + TEST_ASSERT_NULL(fut->ack); TEST_ASSERT_EQUAL_size_t(1U, fixture.async_error_count); TEST_ASSERT_EQUAL_INT(CY_ERR_MEMORY, fixture.last_async_error); TEST_ASSERT_FALSE(cy_future_done(&fut->base)); @@ -973,13 +1349,15 @@ static void test_response_wire_reliable_oom_silent_then_retransmit(void) const size_t callback_base = cap.count; const cy_us_t deadline_base = fut->base.timeout.deadline; - fixture_fail_alloc_size(&fixture, sizeof(request_future_remote_t), 1U); + // seqno is 0, so this is the solo path and the ack record is the only allocation on it. + fixture_fail_alloc_size(&fixture, sizeof(request_ack_t), 1U); dispatch_response_control( &fixture, (byte_t)header_rsp_rel, tag, seqno, topic_hash, message_tag, remote_id, fixture.now + 10U, false); + TEST_ASSERT_EQUAL_size_t(0U, fixture.fail_size_count); // the injection was consumed, not absorbed elsewhere TEST_ASSERT_EQUAL_size_t(0U, fixture.unicast_send_count); TEST_ASSERT_EQUAL_size_t(callback_base, cap.count); TEST_ASSERT_EQUAL_UINT64(0U, fut->response_count); - TEST_ASSERT_NULL(fut->remote_by_id); + TEST_ASSERT_NULL(fut->ack); TEST_ASSERT_TRUE(future_deadline_armed(&fut->base)); TEST_ASSERT_EQUAL_INT64(deadline_base, fut->base.timeout.deadline); TEST_ASSERT_FALSE(cy_future_done(&fut->base)); @@ -1008,7 +1386,10 @@ static void test_response_wire_reliable_oom_silent_then_retransmit(void) cy_message_refcount_dec(moved.message.content); cy_future_destroy(&fut->base); - TEST_ASSERT_NOT_NULL(topic.request_futures_by_tag); + TEST_ASSERT_NULL(topic.request_futures_by_tag); + TEST_ASSERT_NOT_NULL(topic.request_acks_by_tag); + // The single responder answered seqno 0 first, so the record stays in the inlined solo shape: no tree node. + TEST_ASSERT_TRUE(((const request_ack_t*)topic.request_acks_by_tag)->solo); dispatch_response_control( &fixture, (byte_t)header_rsp_rel, tag, seqno, topic_hash, message_tag, remote_id, fixture.now + 30U, false); @@ -1028,8 +1409,19 @@ static void test_response_wire_reliable_oom_silent_then_retransmit(void) TEST_ASSERT_EQUAL_UINT64(topic_hash, deserialize_u64(&fixture.last_unicast[8])); TEST_ASSERT_EQUAL_UINT64(message_tag, deserialize_u64(&fixture.last_unicast[16])); - fixture_advance_to(&fixture, fixture.now + (SESSION_LIFETIME / 2) + 1); - TEST_ASSERT_NULL(topic.request_futures_by_tag); + // Retention is a floor: past dead_at the record still answers ACK until a sweep actually runs. + fixture_advance_to(&fixture, fixture.now + (CY_CONFIG_REQUEST_ACK_RETENTION_us) + 1); + dispatch_response_control( + &fixture, (byte_t)header_rsp_rel, tag, seqno, topic_hash, message_tag, remote_id, fixture.now + 50U, false); + TEST_ASSERT_EQUAL_size_t(4U, fixture.unicast_send_count); + TEST_ASSERT_EQUAL_UINT8(header_rsp_ack, fixture.last_unicast[0]); + + reap_request_acks(&fixture, &topic); + TEST_ASSERT_NULL(topic.request_acks_by_tag); + dispatch_response_control( + &fixture, (byte_t)header_rsp_rel, tag, seqno, topic_hash, message_tag, remote_id, fixture.now + 60U, false); + TEST_ASSERT_EQUAL_size_t(5U, fixture.unicast_send_count); + TEST_ASSERT_EQUAL_UINT8(header_rsp_nack, fixture.last_unicast[0]); // record gone -> nack unindex_request_topic(&fixture, &topic); fixture_assert_clean(&fixture); } @@ -1096,7 +1488,7 @@ static void test_request_publish_callback_pending_update_noop(void) request_publish_callback(fut->publish); // pending status branch: no state change expected TEST_ASSERT_NOT_NULL(fut->publish); - TEST_ASSERT_FALSE(fut->finalized); + TEST_ASSERT_NULL(fut->ack); // no response seen, so no deduplication state was created TEST_ASSERT_TRUE(future_deadline_armed(&fut->base)); TEST_ASSERT_NOT_NULL(topic.request_futures_by_tag); TEST_ASSERT_EQUAL_size_t(0U, g_dummy_publish_dispose_count); @@ -1175,9 +1567,16 @@ int main(void) RUN_TEST(test_message_refcount_primitives_destroy_once); RUN_TEST(test_request_on_response_best_effort_overwrite_and_callback); RUN_TEST(test_request_future_destroy_releases_last_response); - RUN_TEST(test_request_future_dispose_zombie_releases_last_response_early); + RUN_TEST(test_request_future_dispose_hands_over_and_releases_last_response); RUN_TEST(test_request_on_response_reliable_dedup_and_ordering); - RUN_TEST(test_request_on_response_zombie_ack_seen_nack_unseen); + RUN_TEST(test_request_ack_record_ack_seen_nack_unseen); + RUN_TEST(test_request_ack_solo_claim_and_duplicate); + RUN_TEST(test_request_ack_solo_promotes_same_remote); + RUN_TEST(test_request_ack_solo_promotes_second_remote); + RUN_TEST(test_request_ack_shift_boundary_191_192); + RUN_TEST(test_request_ack_promotion_oom_preserves_state); + RUN_TEST(test_request_ack_none_state_not_retained); + RUN_TEST(test_request_ack_expiry_order_follows_disposal); RUN_TEST(test_request_on_response_reliable_oom_stays_pending_silent); RUN_TEST(test_response_wire_reliable_oom_silent_then_retransmit); RUN_TEST(test_response_wire_reliable_client_gone_nacks); diff --git a/tests/src/test_intrusive_topic_allocation.c b/tests/src/test_intrusive_topic_allocation.c index 732dcdf..9382ed3 100644 --- a/tests/src/test_intrusive_topic_allocation.c +++ b/tests/src/test_intrusive_topic_allocation.c @@ -1487,7 +1487,9 @@ static void test_topic_destroy_error_rollback_like_path_on_coupling_oom(void) subscriber_root_t root = { 0 }; wkv_node_t pattern_node = { 0 }; + wkv_node_t name_node = { 0 }; // topic_couple() traces the root name under CY_CONFIG_TRACE root.cy = fix.cy; + root.index_name = &name_node; root.index_pattern = &pattern_node; // non-NULL means pattern root static const wkv_substitution_t subst = { .str = { .len = 1U, .str = "x" }, .ordinal = 0U, .next = NULL };