Remove topic hash from the transport layer; remove ordered delivery and use simplified ACK matching based on the transfer-ID alone (#72)

Ordered delivery cannot be correct given the risk of multiple topics
temporarily sharing the same subject.

Author: pavel-kirienko

.idea/dictionaries/project.xml

@@ -24,15 +24,13 @@ next-generation intelligent vehicles: manned and unmanned aircraft, spacecraft,
 - ≤1-copy TX pipeline with deduplication across multiple interfaces and scattered input buffer support.
 - Support for redundant network interfaces with seamless interface aggregation and zero fail-over delay.
 - Robust message reassembler supporting highly distorted datagram streams:
-  out-of-order fragments, message ordering recovery, fragment/message deduplication, interleaving, variable MTU, ...
-- Robust message ordering recovery for ordering-sensitive applications (e.g., state estimators, control loops)
-  with well-defined deterministic recovery in the event of lost messages.
+  out-of-order fragments, fragment/message deduplication, interleaving, variable MTU, ...
 - Packet loss mitigation via:
   - reliable topics (retransmit until acknowledged; callback notifications for successful/failed deliveries).
   - redundant interfaces (packet lost on one interface may be received on another, transparent to the application);
 - Heap not required (but supported); the library can be used with fixed-size block pool allocators.
 - Detailed time complexity and memory requirement models for the benefit of real-time high-integrity applications.
-- Highly scalable: designed to handle thousands of topics and hundreds of concurrent transfers with minimal resources.
+- Scalable: designed to handle thousands of topics and hundreds of concurrent transfers with minimal resources.
 - Runs anywhere out of the box, including extremely resource-constrained baremetal environments with ~100K ROM/RAM.
   No porting required.
 - Partial MISRA C compliance (reach out to <https://forum.opencyphal.org>).

README.md

@@ -0,0 +1,40 @@
+# All Cyphal/UDP traffic is sent to port 9382.
+# The subject multicast group address is composed as 239.0.0.0 (=0xEF000000) + subject_id (23 bits).
+# All frames of a transfer must share the same field values unless otherwise noted.
+# Frames may arrive out-of-order, possibly interleaved with neighboring transfers; implementations must cope.
+
+uint5 version                   #=2 in this version.
+uint3 priority                  # 0=highest, 7=lowest.
+
+uint2 KIND_MSG_BEST_EFFORT = 0  # No ack must be sent.
+uint2 KIND_MSG_RELIABLE    = 1  # Remote must acknowledge reception by sending an ACK frame back.
+uint2 KIND_ACK             = 2  # Sent P2P; the transfer_id is of the acknowledged frame. Payload empty/ignored.
+uint2 kind
+uint6 reserved_incompat         # Discard frame if any incompatibility flags are set that are not understood.
+
+void16                          # Reserved for compatibility flags and fields (transmit zero, ignore on reception).
+
+# Payload reassembly information.
+# We provide both the frame index and the frame payload offset to allow various reassembly strategies depending on the
+# preferences of the implementation. The provided information is sufficient for zero-copy out-of-order reassembly.
+# Offset 4 bytes.
+
+uint24 frame_index              # Zero-based index of the payload fragment carried by this frame.
+void8
+uint32 frame_payload_offset     # The offset of the frame payload relative to the start of the transfer payload.
+uint32 transfer_payload_size    # Total for all frames.
+
+# Transfer identification information.
+# The transfer-ID is a single field that segregates transfers by topic hash and epoch (publisher sequence restarts).
+# Offset 16 bytes.
+
+uint64 transfer_id              # For multi-frame reassembly and dedup. ACK specifies the acked tfer here.
+uint64 sender_uid               # Origin identifier ensures invariance to the source IP address for reassembly.
+
+# Integrity checking information.
+# Offset 32 bytes.
+
+uint32 prefix_crc32c            # crc32c(payload[0:(frame_payload_offset+payload_size)])
+uint32 header_crc32c            # Covers all fields above. Same as the transfer payload CRC.
+
+# End of header at 40 bytes. Payload follows.

cyphal_udp_header.dsdl

@@ -98,7 +98,6 @@ gen_test_single(test_e2e_random "src/test_e2e_random.cpp;${library_dir}/udpard.c
 gen_test_single(test_e2e_edge "src/test_e2e_edge.cpp;${library_dir}/udpard.c")
 gen_test_single(test_e2e_api "src/test_e2e_api.cpp;${library_dir}/udpard.c")
 gen_test_single(test_e2e_responses "src/test_e2e_responses.cpp;${library_dir}/udpard.c")
-gen_test_single(test_e2e_reliable_ordered "src/test_e2e_reliable_ordered.cpp;${library_dir}/udpard.c")
 gen_test_single(test_integration_sockets "src/test_integration_sockets.cpp;${library_dir}/udpard.c")
 
 # Coverage targets. Usage:

libudpard/udpard.c

@@ -31,7 +31,6 @@ struct RxContext
     std::array<udpard_udpip_ep_t, UDPARD_IFACE_COUNT_MAX> sources{};
     uint64_t                                              remote_uid = 0;
     size_t                                                received   = 0;
-    size_t                                                collisions = 0;
 };
 
 // Refcount helpers keep captured datagrams alive.
@@ -115,19 +114,14 @@ void on_message(udpard_rx_t* const rx, udpard_rx_port_t* const port, const udpar
     ctx->received++;
 }
 
-void on_collision(udpard_rx_t* const rx, udpard_rx_port_t* const /*port*/, const udpard_remote_t /*remote*/)
-{
-    auto* ctx = static_cast<RxContext*>(rx->user);
-    ctx->collisions++;
-}
-constexpr udpard_rx_port_vtable_t callbacks{ .on_message = &on_message, .on_collision = &on_collision };
+constexpr udpard_rx_port_vtable_t callbacks{ .on_message = &on_message };
 
 // Ack port frees responses.
 void on_ack_response(udpard_rx_t*, udpard_rx_port_t* port, const udpard_rx_transfer_t tr)
 {
     udpard_fragment_free_all(tr.payload, udpard_make_deleter(port->memory.fragment));
 }
-constexpr udpard_rx_port_vtable_t ack_callbacks{ .on_message = &on_ack_response, .on_collision = &on_collision };
+constexpr udpard_rx_port_vtable_t ack_callbacks{ .on_message = &on_ack_response };
 
 // Reliable delivery must survive data and ack loss.
 // Each node uses exactly one TX and one RX instance as per the library design.
@@ -163,6 +157,7 @@ void test_reliable_delivery_under_losses()
         res = instrumented_allocator_make_resource(&pub_tx_alloc_payload);
     }
     const udpard_rx_mem_resources_t pub_rx_mem{ .session  = instrumented_allocator_make_resource(&pub_rx_alloc_session),
+                                                .slot     = instrumented_allocator_make_resource(&pub_rx_alloc_session),
                                                 .fragment = instrumented_allocator_make_resource(&pub_rx_alloc_frag) };
 
     udpard_tx_mem_resources_t sub_tx_mem{};
@@ -171,6 +166,7 @@ void test_reliable_delivery_under_losses()
         res = instrumented_allocator_make_resource(&sub_tx_alloc_payload);
     }
     const udpard_rx_mem_resources_t sub_rx_mem{ .session  = instrumented_allocator_make_resource(&sub_rx_alloc_session),
+                                                .slot     = instrumented_allocator_make_resource(&sub_rx_alloc_session),
                                                 .fragment = instrumented_allocator_make_resource(&sub_rx_alloc_frag) };
 
     // Publisher node: single TX, single RX (linked to TX for ACK processing).
@@ -184,8 +180,7 @@ void test_reliable_delivery_under_losses()
     udpard_rx_t pub_rx{};
     udpard_rx_new(&pub_rx, &pub_tx);
     udpard_rx_port_t pub_p2p_port{};
-    TEST_ASSERT_TRUE(
-      udpard_rx_port_new(&pub_p2p_port, pub_uid, 16, udpard_rx_unordered, 0, pub_rx_mem, &ack_callbacks));
+    TEST_ASSERT_TRUE(udpard_rx_port_new_p2p(&pub_p2p_port, 16, pub_rx_mem, &ack_callbacks));
 
     // Subscriber node: single TX, single RX (linked to TX for sending ACKs).
     constexpr uint64_t         sub_uid = 0xABCDEF0012345678ULL;
@@ -197,8 +192,7 @@ void test_reliable_delivery_under_losses()
     udpard_rx_t sub_rx{};
     udpard_rx_new(&sub_rx, &sub_tx);
     udpard_rx_port_t sub_port{};
-    const uint64_t   topic_hash = 0x0123456789ABCDEFULL;
-    TEST_ASSERT_TRUE(udpard_rx_port_new(&sub_port, topic_hash, 6000, udpard_rx_unordered, 0, sub_rx_mem, &callbacks));
+    TEST_ASSERT_TRUE(udpard_rx_port_new(&sub_port, 6000, sub_rx_mem, &callbacks));
 
     // Endpoints.
     const std::array<udpard_udpip_ep_t, UDPARD_IFACE_COUNT_MAX> publisher_sources{
@@ -235,7 +229,6 @@ void test_reliable_delivery_under_losses()
                                     deadline,
                                     iface_bitmap_all,
                                     udpard_prio_fast,
-                                    topic_hash,
                                     1U,
                                     payload_view,
                                     &record_feedback,
@@ -296,8 +289,6 @@ void test_reliable_delivery_under_losses()
     TEST_ASSERT_EQUAL_size_t(1, fb.count);
     TEST_ASSERT_EQUAL_UINT32(1, fb.acknowledgements);
     TEST_ASSERT_EQUAL_size_t(1, ctx.received);
-    TEST_ASSERT_EQUAL_size_t(0, ctx.collisions);
-
     // Cleanup.
     udpard_rx_port_free(&sub_rx, &sub_port);
     udpard_rx_port_free(&pub_rx, &pub_p2p_port);
@@ -350,7 +341,6 @@ void test_reliable_stats_and_failures()
                                     10,
                                     iface_bitmap_1,
                                     udpard_prio_fast,
-                                    0xABCULL,
                                     5U,
                                     exp_payload,
                                     &record_feedback,
@@ -380,6 +370,7 @@ void test_reliable_stats_and_failures()
     instrumented_allocator_new(&src_alloc_transfer);
     instrumented_allocator_new(&src_alloc_payload);
     const udpard_rx_mem_resources_t rx_mem{ .session  = instrumented_allocator_make_resource(&rx_alloc_session),
+                                            .slot     = instrumented_allocator_make_resource(&rx_alloc_session),
                                             .fragment = instrumented_allocator_make_resource(&rx_alloc_frag) };
     udpard_tx_mem_resources_t       src_mem{};
     src_mem.transfer = instrumented_allocator_make_resource(&src_alloc_transfer);
@@ -399,7 +390,7 @@ void test_reliable_stats_and_failures()
     ctx.expected.assign({ 1U, 2U, 3U, 4U });
     udpard_rx_new(&rx, nullptr);
     rx.user = &ctx;
-    TEST_ASSERT_TRUE(udpard_rx_port_new(&port, 0x12340000ULL, 64, udpard_rx_unordered, 0, rx_mem, &callbacks));
+    TEST_ASSERT_TRUE(udpard_rx_port_new(&port, 64, rx_mem, &callbacks));
 
     const udpard_bytes_scattered_t src_payload = make_scattered(ctx.expected.data(), ctx.expected.size());
     FeedbackState                  fb_ignore{};
@@ -408,7 +399,6 @@ void test_reliable_stats_and_failures()
                                     1000,
                                     iface_bitmap_1,
                                     udpard_prio_fast,
-                                    port.topic_hash,
                                     7U,
                                     src_payload,
                                     &record_feedback,