From c75cf06c2dedbda03d31d33b9b19cac522bd1a7d Mon Sep 17 00:00:00 2001 From: Kevin Ahrendt Date: Wed, 7 Oct 2026 08:39:23 -0400 Subject: [PATCH 1/3] Stamp messages longer than one TCP segment when they finish arriving A message up to SendspinConnection::SINGLE_SEGMENT_MESSAGE_BYTES keeps the stamp from when the transport first saw it; a longer one gets the completion stamp the transport now passes to end_inbound_message(). --- src/connection.cpp | 25 ++++++--- src/connection.h | 14 ++++- src/esp/client_connection.cpp | 6 +-- src/esp/server_connection.cpp | 7 +-- src/host/client_connection.cpp | 2 +- src/host/server_connection.cpp | 2 +- tests/loopback_connection.h | 1 + tests/test_client_lifecycle.cpp | 2 +- tests/test_noise_transport.cpp | 94 +++++++++++++++++++++++++++++---- 9 files changed, 126 insertions(+), 27 deletions(-) diff --git a/src/connection.cpp b/src/connection.cpp index e4fd07f7..c12e06f8 100644 --- a/src/connection.cpp +++ b/src/connection.cpp @@ -603,6 +603,7 @@ SendspinConnection::InboundTarget SendspinConnection::route_inbound_message(size header->receive_time_us = stamp; header->kind = kind; this->inbound_item_ = item; + this->inbound_item_len_ = len; return {inbound_item_bytes(item), InboundRoute::RECEIVE}; } @@ -664,13 +665,17 @@ void SendspinConnection::note_message_completed() { std::memory_order_relaxed); } -void SendspinConnection::end_inbound_message(bool received) { +void SendspinConnection::end_inbound_message(bool received, int64_t complete_time_us) { if (received) { this->note_message_completed(); } if (this->inbound_item_ != nullptr) { void* item = std::exchange(this->inbound_item_, nullptr); if (received) { + if (this->inbound_item_len_ > SINGLE_SEGMENT_MESSAGE_BYTES) { + inbound_item_header(item)->receive_time_us = + static_cast(complete_time_us); + } this->inbound_ring_->complete(item); } else { // FreeRTOS cannot cancel an acquire: complete it as DISCARD, which take() returns @@ -681,7 +686,13 @@ void SendspinConnection::end_inbound_message(bool received) { } } else if (this->inbound_to_fallback_) { this->inbound_to_fallback_ = false; - if (!received || !this->inbound_gate_.publish_pending_message()) { + if (!received) { + return; + } + if (this->fallback_len_ > SINGLE_SEGMENT_MESSAGE_BYTES) { + this->fallback_receive_time_us_ = static_cast(complete_time_us); + } + if (!this->inbound_gate_.publish_pending_message()) { return; } } else { @@ -692,7 +703,7 @@ void SendspinConnection::end_inbound_message(bool received) { void SendspinConnection::abandon_inbound_message() { if (this->inbound_item_ != nullptr) { - this->end_inbound_message(false); + this->end_inbound_message(false, 0); } this->inbound_to_fallback_ = false; this->fragment_dropping_ = false; @@ -763,7 +774,7 @@ SendspinConnection::InboundTarget SendspinConnection::begin_inbound_fragment( return {this->fallback_buf_.data() + this->fallback_len_, InboundRoute::RECEIVE}; } -void SendspinConnection::end_inbound_fragment(size_t len, bool last) { +void SendspinConnection::end_inbound_fragment(size_t len, bool last, int64_t complete_time_us) { if (last) { this->fragment_assembly_open_ = false; this->note_message_completed(); @@ -780,7 +791,7 @@ void SendspinConnection::end_inbound_fragment(size_t len, bool last) { } if (!this->inbound_gate_.is_admitted()) { this->inbound_to_fallback_ = true; - this->end_inbound_message(true); + this->end_inbound_message(true, complete_time_us); return; } // Admitted: copy the assembled message into a ring item, the one copy this path costs over @@ -794,13 +805,13 @@ void SendspinConnection::end_inbound_fragment(size_t len, bool last) { if (this->inbound_item_ == nullptr) { // Routed to the fallback buffer it is already in: longer than the ring takes, or the // admission flag cleared in between. - this->end_inbound_message(true); + this->end_inbound_message(true, complete_time_us); return; } std::memcpy(target.data, this->fallback_buf_.data(), total); this->fallback_buf_.reset(); this->fallback_len_ = 0; - this->end_inbound_message(true); + this->end_inbound_message(true, complete_time_us); } // ============================================================================ diff --git a/src/connection.h b/src/connection.h index bc937a5f..f9a120b1 100644 --- a/src/connection.h +++ b/src/connection.h @@ -898,6 +898,11 @@ class SendspinConnection : public std::enable_shared_from_thischunk_dest_ = nullptr; - this->end_inbound_message(true); + this->end_inbound_message(true, receive_time); } return; } @@ -294,9 +294,9 @@ void SendspinClientConnection::handle_data(const esp_websocket_event_data_t* dat if (chunk_len > 0) { std::memcpy(target.data, data->data_ptr, chunk_len); } - this->end_inbound_fragment(chunk_len, last); + this->end_inbound_fragment(chunk_len, last, receive_time); } else if (target.route == InboundRoute::DROP) { - this->end_inbound_fragment(0, last); + this->end_inbound_fragment(0, last, receive_time); } } diff --git a/src/esp/server_connection.cpp b/src/esp/server_connection.cpp index c174614b..d7e9ff2b 100644 --- a/src/esp/server_connection.cpp +++ b/src/esp/server_connection.cpp @@ -18,6 +18,7 @@ #include "platform/compiler.h" #include "platform/logging.h" #include "platform/memory.h" +#include "platform/time.h" #include "platform/types.h" #include "protocol_messages.h" #include "sendspin/types.h" @@ -292,7 +293,7 @@ SS_HOT esp_err_t SendspinServerConnection::handle_data(httpd_req_t* req, int64_t return SendspinServerConnection::discard_frame_payload(req, ws_pkt, server); } ret = SendspinServerConnection::receive_frame_payload(req, ws_pkt, target.data); - this->end_inbound_message(ret == ESP_OK); + this->end_inbound_message(ret == ESP_OK, platform_time_us()); return ret; } @@ -304,7 +305,7 @@ SS_HOT esp_err_t SendspinServerConnection::handle_data(httpd_req_t* req, int64_t } if (target.route == InboundRoute::DROP) { ret = SendspinServerConnection::discard_frame_payload(req, ws_pkt, server); - this->end_inbound_fragment(0, ws_pkt.final); + this->end_inbound_fragment(0, ws_pkt.final, platform_time_us()); return ret; } ret = SendspinServerConnection::receive_frame_payload(req, ws_pkt, target.data); @@ -312,7 +313,7 @@ SS_HOT esp_err_t SendspinServerConnection::handle_data(httpd_req_t* req, int64_t // httpd closes the session over the error; nothing assembled so far is published. return ret; } - this->end_inbound_fragment(ws_pkt.len, ws_pkt.final); + this->end_inbound_fragment(ws_pkt.len, ws_pkt.final, platform_time_us()); return ESP_OK; } diff --git a/src/host/client_connection.cpp b/src/host/client_connection.cpp index 91e718ca..c4fd9ccf 100644 --- a/src/host/client_connection.cpp +++ b/src/host/client_connection.cpp @@ -172,7 +172,7 @@ void SendspinClientConnection::setup_callbacks() { this->begin_inbound_message(data.size(), !msg->binary, receive_time); if (target.route == InboundRoute::RECEIVE) { std::copy(data.begin(), data.end(), target.data); - this->end_inbound_message(true); + this->end_inbound_message(true, receive_time); } break; } diff --git a/src/host/server_connection.cpp b/src/host/server_connection.cpp index 831cb98c..21b37dac 100644 --- a/src/host/server_connection.cpp +++ b/src/host/server_connection.cpp @@ -105,7 +105,7 @@ void SendspinServerConnection::handle_message(const std::string& data, bool is_b return; } std::copy(data.begin(), data.end(), target.data); - this->end_inbound_message(true); + this->end_inbound_message(true, receive_time); } } // namespace sendspin diff --git a/tests/loopback_connection.h b/tests/loopback_connection.h index 0e967367..091f53ab 100644 --- a/tests/loopback_connection.h +++ b/tests/loopback_connection.h @@ -94,6 +94,7 @@ class TestConnection : public SendspinConnection { using SendspinConnection::end_inbound_message; using SendspinConnection::InboundRoute; using SendspinConnection::InboundTarget; + using SendspinConnection::SINGLE_SEGMENT_MESSAGE_BYTES; // --- Test helpers --- diff --git a/tests/test_client_lifecycle.cpp b/tests/test_client_lifecycle.cpp index 02c2f969..d31cc1eb 100644 --- a/tests/test_client_lifecycle.cpp +++ b/tests/test_client_lifecycle.cpp @@ -1949,7 +1949,7 @@ TEST(NextDeadline, AFallbackMessageHeldBehindARingItemRunsTheNextTickAtOnce) { const auto target = conn->begin_inbound_message(len, false, platform_time_us()); ASSERT_EQ(target.route, SendspinConnection::InboundRoute::RECEIVE); std::memset(target.data, 0x5A, len); - conn->end_inbound_message(true); + conn->end_inbound_message(true, platform_time_us()); }; receive(3); receive(client.inbound_ring_->max_message_bytes() + row.extra_len); diff --git a/tests/test_noise_transport.cpp b/tests/test_noise_transport.cpp index db128f96..e7dce536 100644 --- a/tests/test_noise_transport.cpp +++ b/tests/test_noise_transport.cpp @@ -2117,7 +2117,7 @@ struct InboundHarness { this->conn.begin_inbound_message(bytes.size(), /*is_text=*/false, platform_time_us()); if (target.route == TestConnection::InboundRoute::RECEIVE) { std::memcpy(target.data, bytes.data(), bytes.size()); - this->conn.end_inbound_message(true); + this->conn.end_inbound_message(true, platform_time_us()); } return target.route; } @@ -2320,7 +2320,7 @@ TEST(InboundReceive, AMultiFrameMessageIsCappedOnItsRunningTotal) { first_len, /*first=*/true, /*is_text=*/false, platform_time_us()); ASSERT_EQ(first.route, TestConnection::InboundRoute::RECEIVE); std::memset(first.data, 0x11, first_len); - h.conn.end_inbound_fragment(first_len, /*last=*/false); + h.conn.end_inbound_fragment(first_len, /*last=*/false, platform_time_us()); const TestConnection::InboundTarget second = h.conn.begin_inbound_fragment( row.second_len, /*first=*/false, /*is_text=*/false, platform_time_us()); @@ -2331,7 +2331,7 @@ TEST(InboundReceive, AMultiFrameMessageIsCappedOnItsRunningTotal) { continue; } std::memset(second.data, 0x22, row.second_len); - h.conn.end_inbound_fragment(row.second_len, /*last=*/true); + h.conn.end_inbound_fragment(row.second_len, /*last=*/true, platform_time_us()); // The whole message reaches the protocol task: in a ring item once admitted (the one // copy this path costs), as the pending message before. @@ -2444,7 +2444,7 @@ TEST(InboundReceive, AnAbandonedRingItemIsNeverHandedToTheProtocolTask) { } else { std::memcpy(target.data + bytes.size() / 2, bytes.data() + bytes.size() / 2, bytes.size() - bytes.size() / 2); - h.conn.end_inbound_message(true); + h.conn.end_inbound_message(true, platform_time_us()); } size_t len = 0; @@ -2504,7 +2504,8 @@ TEST(InboundReceive, AFrameOutOfFragmentSequenceClosesTheConnection) { first.size(), /*first=*/true, /*is_text=*/false, platform_time_us()); ASSERT_EQ(target.route, TestConnection::InboundRoute::RECEIVE); std::memcpy(target.data, first.data(), first.size()); - h.conn.end_inbound_fragment(first.size(), row.before == Before::PENDING_ASSEMBLED); + h.conn.end_inbound_fragment(first.size(), row.before == Before::PENDING_ASSEMBLED, + platform_time_us()); } InboundMessage pending; const uint8_t* pending_data = nullptr; @@ -2521,7 +2522,7 @@ TEST(InboundReceive, AFrameOutOfFragmentSequenceClosesTheConnection) { route = target.route; if (route == TestConnection::InboundRoute::RECEIVE) { std::memcpy(target.data, second.data(), second.size()); - h.conn.end_inbound_fragment(second.size(), /*last=*/true); + h.conn.end_inbound_fragment(second.size(), /*last=*/true, platform_time_us()); } } else { route = h.receive(second); @@ -2605,7 +2606,7 @@ TEST(InboundReceive, TheLivenessStampIsTakenWhenAMessageCompletes) { std::memcpy(target.data, bytes.data(), bytes.size()); EXPECT_EQ(h.conn.get_last_receive_time_us(), 0U) << "stamped before completion"; if (row.complete) { - h.conn.end_inbound_message(true); + h.conn.end_inbound_message(true, platform_time_us()); } } } else { @@ -2617,13 +2618,13 @@ TEST(InboundReceive, TheLivenessStampIsTakenWhenAMessageCompletes) { std::memcpy(target.data, bytes.data(), bytes.size()); received = bytes.size(); } - h.conn.end_inbound_fragment(received, /*last=*/false); + h.conn.end_inbound_fragment(received, /*last=*/false, platform_time_us()); EXPECT_EQ(h.conn.get_last_receive_time_us(), 0U) << "stamped before completion"; if (row.complete) { const TestConnection::InboundTarget last = h.conn.begin_inbound_fragment( 0, /*first=*/false, /*is_text=*/false, platform_time_us()); ASSERT_EQ(last.route, route); - h.conn.end_inbound_fragment(0, /*last=*/true); + h.conn.end_inbound_fragment(0, /*last=*/true, platform_time_us()); } } EXPECT_EQ(h.conn.get_last_receive_time_us() != 0U, row.stamped); @@ -2637,3 +2638,78 @@ TEST(InboundReceive, TheLivenessStampIsTakenWhenAMessageCompletes) { } } } + +// A message up to SINGLE_SEGMENT_MESSAGE_BYTES keeps its first stamp, a longer one its completion +// stamp, on every route. +TEST(InboundReceive, AMessageIsStampedAtFirstSightUpToOneSegmentAndAtCompletionPastIt) { + enum class Shape : uint8_t { SINGLE_FRAME, MULTI_FRAME }; + struct Row { + const char* name; + bool admitted; // admitted: a ring item; unadmitted: the fallback buffer + Shape shape; + size_t len; + bool completion_stamp; + }; + constexpr size_t BOUND = TestConnection::SINGLE_SEGMENT_MESSAGE_BYTES; + const Row rows[] = { + {"ring item at the bound", true, Shape::SINGLE_FRAME, BOUND, false}, + {"ring item past the bound", true, Shape::SINGLE_FRAME, BOUND + 1, true}, + {"fallback buffer at the bound", false, Shape::SINGLE_FRAME, BOUND, false}, + {"fallback buffer past the bound", false, Shape::SINGLE_FRAME, BOUND + 1, true}, + {"multi-frame into a ring item at the bound", true, Shape::MULTI_FRAME, BOUND, false}, + {"multi-frame into a ring item past the bound", true, Shape::MULTI_FRAME, BOUND + 1, + true}, + {"multi-frame into the fallback buffer at the bound", false, Shape::MULTI_FRAME, BOUND, + false}, + {"multi-frame into the fallback buffer past the bound", false, Shape::MULTI_FRAME, + BOUND + 1, true}, + }; + constexpr int64_t FIRST_US = 1'000; + constexpr int64_t MIDDLE_US = 3'000; + constexpr int64_t COMPLETE_US = 5'000; + + for (const Row& row : rows) { + SCOPED_TRACE(row.name); + InboundHarness h; + h.conn.set_admitted(row.admitted); + const std::vector bytes(row.len, 0x42); + + if (row.shape == Shape::SINGLE_FRAME) { + const TestConnection::InboundTarget target = + h.conn.begin_inbound_message(bytes.size(), /*is_text=*/false, FIRST_US); + ASSERT_EQ(target.route, TestConnection::InboundRoute::RECEIVE); + std::memcpy(target.data, bytes.data(), bytes.size()); + h.conn.end_inbound_message(true, COMPLETE_US); + } else { + const size_t first_len = bytes.size() / 2; + const TestConnection::InboundTarget first = h.conn.begin_inbound_fragment( + first_len, /*first=*/true, /*is_text=*/false, FIRST_US); + ASSERT_EQ(first.route, TestConnection::InboundRoute::RECEIVE); + std::memcpy(first.data, bytes.data(), first_len); + h.conn.end_inbound_fragment(first_len, /*last=*/false, MIDDLE_US); + const TestConnection::InboundTarget second = h.conn.begin_inbound_fragment( + bytes.size() - first_len, /*first=*/false, /*is_text=*/false, MIDDLE_US); + ASSERT_EQ(second.route, TestConnection::InboundRoute::RECEIVE); + std::memcpy(second.data, bytes.data() + first_len, bytes.size() - first_len); + h.conn.end_inbound_fragment(bytes.size() - first_len, /*last=*/true, COMPLETE_US); + } + + const uint32_t expected = + static_cast(row.completion_stamp ? COMPLETE_US : FIRST_US); + if (row.admitted) { + size_t len = 0; + void* item = h.ring.take(&len, 0); + ASSERT_NE(item, nullptr) << "the admitted message skipped the ring"; + EXPECT_EQ(len, bytes.size()); + EXPECT_EQ(inbound_item_header(item)->receive_time_us, expected); + h.conn.inbound_gate().note_item_taken(); + h.ring.return_item(item); + } else { + InboundMessage pending; + ASSERT_TRUE(h.conn.pending_message(pending)); + EXPECT_EQ(pending.len, bytes.size()); + EXPECT_EQ(pending.receive_time_us, expected); + h.conn.consume_pending_message(); + } + } +} From 0d43b4118c9a738436f5b626c2285a586043e9b6 Mon Sep 17 00:00:00 2001 From: Kevin Ahrendt Date: Wed, 7 Oct 2026 09:36:19 -0400 Subject: [PATCH 2/3] Stamp a multi-frame message at completion whatever its length Its fragments may arrive in separate segments, so the first fragment's stamp can be early; begin_inbound_fragment() no longer takes a stamp. --- src/connection.cpp | 7 ++++--- src/connection.h | 10 ++++------ src/esp/client_connection.cpp | 3 +-- src/esp/server_connection.cpp | 3 +-- tests/test_noise_transport.cpp | 26 +++++++++++++------------- 5 files changed, 23 insertions(+), 26 deletions(-) diff --git a/src/connection.cpp b/src/connection.cpp index c12e06f8..405277a5 100644 --- a/src/connection.cpp +++ b/src/connection.cpp @@ -710,8 +710,8 @@ void SendspinConnection::abandon_inbound_message() { this->fragment_assembly_open_ = false; } -SendspinConnection::InboundTarget SendspinConnection::begin_inbound_fragment( - size_t len, bool first, bool is_text, int64_t receive_time_us) { +SendspinConnection::InboundTarget SendspinConnection::begin_inbound_fragment(size_t len, bool first, + bool is_text) { // The rare path (see the declaration): a multi-frame WebSocket message is assembled in the // fallback buffer whatever the admission state, and routed when its last bytes arrive. if (first && this->fragment_assembly_open_) { @@ -739,7 +739,6 @@ SendspinConnection::InboundTarget SendspinConnection::begin_inbound_fragment( // connection's pending message through them. this->fallback_len_ = 0; this->fallback_kind_ = is_text ? InboundKind::TEXT : InboundKind::BINARY; - this->fallback_receive_time_us_ = static_cast(receive_time_us); } } else if (!this->fragment_assembly_open_) { // RFC 6455 section 5.4: a continuation frame continues a fragmented message, so one with @@ -789,6 +788,8 @@ void SendspinConnection::end_inbound_fragment(size_t len, bool last, int64_t com if (!last) { return; } + // A multi-frame message may span segments. + this->fallback_receive_time_us_ = static_cast(complete_time_us); if (!this->inbound_gate_.is_admitted()) { this->inbound_to_fallback_ = true; this->end_inbound_message(true, complete_time_us); diff --git a/src/connection.h b/src/connection.h index f9a120b1..c6fcdcd5 100644 --- a/src/connection.h +++ b/src/connection.h @@ -898,9 +898,9 @@ class SendspinConnection : public std::enable_shared_from_thisfin && frame_done; - const InboundTarget target = - this->begin_inbound_fragment(chunk_len, first, is_text, receive_time); + const InboundTarget target = this->begin_inbound_fragment(chunk_len, first, is_text); if (target.route == InboundRoute::RECEIVE) { if (chunk_len > 0) { std::memcpy(target.data, data->data_ptr, chunk_len); diff --git a/src/esp/server_connection.cpp b/src/esp/server_connection.cpp index d7e9ff2b..fb7009e2 100644 --- a/src/esp/server_connection.cpp +++ b/src/esp/server_connection.cpp @@ -298,8 +298,7 @@ SS_HOT esp_err_t SendspinServerConnection::handle_data(httpd_req_t* req, int64_t } // A frame of a multi-frame message (the rare path; see begin_inbound_fragment()). - const InboundTarget target = - this->begin_inbound_fragment(ws_pkt.len, !continuation, is_text, receive_time); + const InboundTarget target = this->begin_inbound_fragment(ws_pkt.len, !continuation, is_text); if (target.route == InboundRoute::CLOSE) { return ESP_FAIL; } diff --git a/tests/test_noise_transport.cpp b/tests/test_noise_transport.cpp index e7dce536..a489ccd1 100644 --- a/tests/test_noise_transport.cpp +++ b/tests/test_noise_transport.cpp @@ -2317,13 +2317,13 @@ TEST(InboundReceive, AMultiFrameMessageIsCappedOnItsRunningTotal) { const size_t first_len = row.cap - 10; const TestConnection::InboundTarget first = h.conn.begin_inbound_fragment( - first_len, /*first=*/true, /*is_text=*/false, platform_time_us()); + first_len, /*first=*/true, /*is_text=*/false); ASSERT_EQ(first.route, TestConnection::InboundRoute::RECEIVE); std::memset(first.data, 0x11, first_len); h.conn.end_inbound_fragment(first_len, /*last=*/false, platform_time_us()); const TestConnection::InboundTarget second = h.conn.begin_inbound_fragment( - row.second_len, /*first=*/false, /*is_text=*/false, platform_time_us()); + row.second_len, /*first=*/false, /*is_text=*/false); EXPECT_EQ(second.route, row.route); EXPECT_EQ(h.conn.close_transport_now_calls_, row.route == TestConnection::InboundRoute::CLOSE ? 1 : 0); @@ -2501,7 +2501,7 @@ TEST(InboundReceive, AFrameOutOfFragmentSequenceClosesTheConnection) { ASSERT_EQ(h.receive(single), TestConnection::InboundRoute::RECEIVE); } else { const TestConnection::InboundTarget target = h.conn.begin_inbound_fragment( - first.size(), /*first=*/true, /*is_text=*/false, platform_time_us()); + first.size(), /*first=*/true, /*is_text=*/false); ASSERT_EQ(target.route, TestConnection::InboundRoute::RECEIVE); std::memcpy(target.data, first.data(), first.size()); h.conn.end_inbound_fragment(first.size(), row.before == Before::PENDING_ASSEMBLED, @@ -2518,7 +2518,7 @@ TEST(InboundReceive, AFrameOutOfFragmentSequenceClosesTheConnection) { if (row.second != Second::SINGLE_FRAME) { const TestConnection::InboundTarget target = h.conn.begin_inbound_fragment( second.size(), /*first=*/row.second == Second::FIRST_FRAGMENT, - /*is_text=*/false, platform_time_us()); + /*is_text=*/false); route = target.route; if (route == TestConnection::InboundRoute::RECEIVE) { std::memcpy(target.data, second.data(), second.size()); @@ -2611,7 +2611,7 @@ TEST(InboundReceive, TheLivenessStampIsTakenWhenAMessageCompletes) { } } else { const TestConnection::InboundTarget target = h.conn.begin_inbound_fragment( - bytes.size(), /*first=*/true, /*is_text=*/false, platform_time_us()); + bytes.size(), /*first=*/true, /*is_text=*/false); ASSERT_EQ(target.route, route); size_t received = 0; if (target.route == TestConnection::InboundRoute::RECEIVE) { @@ -2622,7 +2622,7 @@ TEST(InboundReceive, TheLivenessStampIsTakenWhenAMessageCompletes) { EXPECT_EQ(h.conn.get_last_receive_time_us(), 0U) << "stamped before completion"; if (row.complete) { const TestConnection::InboundTarget last = h.conn.begin_inbound_fragment( - 0, /*first=*/false, /*is_text=*/false, platform_time_us()); + 0, /*first=*/false, /*is_text=*/false); ASSERT_EQ(last.route, route); h.conn.end_inbound_fragment(0, /*last=*/true, platform_time_us()); } @@ -2639,9 +2639,9 @@ TEST(InboundReceive, TheLivenessStampIsTakenWhenAMessageCompletes) { } } -// A message up to SINGLE_SEGMENT_MESSAGE_BYTES keeps its first stamp, a longer one its completion -// stamp, on every route. -TEST(InboundReceive, AMessageIsStampedAtFirstSightUpToOneSegmentAndAtCompletionPastIt) { +// A single-frame message up to SINGLE_SEGMENT_MESSAGE_BYTES keeps its first stamp; a longer one, +// or a multi-frame message of any length, gets its completion stamp. +TEST(InboundReceive, ASmallSingleFrameMessageKeepsItsFirstStampAndOthersTheirCompletion) { enum class Shape : uint8_t { SINGLE_FRAME, MULTI_FRAME }; struct Row { const char* name; @@ -2656,11 +2656,11 @@ TEST(InboundReceive, AMessageIsStampedAtFirstSightUpToOneSegmentAndAtCompletionP {"ring item past the bound", true, Shape::SINGLE_FRAME, BOUND + 1, true}, {"fallback buffer at the bound", false, Shape::SINGLE_FRAME, BOUND, false}, {"fallback buffer past the bound", false, Shape::SINGLE_FRAME, BOUND + 1, true}, - {"multi-frame into a ring item at the bound", true, Shape::MULTI_FRAME, BOUND, false}, + {"multi-frame into a ring item at the bound", true, Shape::MULTI_FRAME, BOUND, true}, {"multi-frame into a ring item past the bound", true, Shape::MULTI_FRAME, BOUND + 1, true}, {"multi-frame into the fallback buffer at the bound", false, Shape::MULTI_FRAME, BOUND, - false}, + true}, {"multi-frame into the fallback buffer past the bound", false, Shape::MULTI_FRAME, BOUND + 1, true}, }; @@ -2683,12 +2683,12 @@ TEST(InboundReceive, AMessageIsStampedAtFirstSightUpToOneSegmentAndAtCompletionP } else { const size_t first_len = bytes.size() / 2; const TestConnection::InboundTarget first = h.conn.begin_inbound_fragment( - first_len, /*first=*/true, /*is_text=*/false, FIRST_US); + first_len, /*first=*/true, /*is_text=*/false); ASSERT_EQ(first.route, TestConnection::InboundRoute::RECEIVE); std::memcpy(first.data, bytes.data(), first_len); h.conn.end_inbound_fragment(first_len, /*last=*/false, MIDDLE_US); const TestConnection::InboundTarget second = h.conn.begin_inbound_fragment( - bytes.size() - first_len, /*first=*/false, /*is_text=*/false, MIDDLE_US); + bytes.size() - first_len, /*first=*/false, /*is_text=*/false); ASSERT_EQ(second.route, TestConnection::InboundRoute::RECEIVE); std::memcpy(second.data, bytes.data() + first_len, bytes.size() - first_len); h.conn.end_inbound_fragment(bytes.size() - first_len, /*last=*/true, COMPLETE_US); From 4efac8ea3d18542da0050b812c2082c29ebeb02b Mon Sep 17 00:00:00 2001 From: Kevin Ahrendt Date: Wed, 7 Oct 2026 11:44:06 -0400 Subject: [PATCH 3/3] Stamp every message when its last byte arrives A message that fits one TCP segment can still arrive in more than one (the sender may write the header and payload separately), so the first-sight stamp is dropped: the transport's completion stamp is every message's receive time, and begin_inbound_message() no longer takes a stamp. --- src/connection.cpp | 32 +++++++------------------ src/connection.h | 18 ++++---------- src/esp/client_connection.cpp | 3 +-- src/esp/server_connection.cpp | 5 ++-- src/esp/server_connection.h | 3 +-- src/esp/ws_server.cpp | 5 +--- src/host/client_connection.cpp | 3 +-- src/host/server_connection.cpp | 2 +- tests/loopback_connection.h | 1 - tests/test_client_lifecycle.cpp | 2 +- tests/test_noise_transport.cpp | 42 ++++++++++----------------------- 11 files changed, 35 insertions(+), 81 deletions(-) diff --git a/src/connection.cpp b/src/connection.cpp index 405277a5..8a9149ef 100644 --- a/src/connection.cpp +++ b/src/connection.cpp @@ -495,8 +495,8 @@ void SendspinConnection::fail_inbound() { } } -SS_HOT SendspinConnection::InboundTarget SendspinConnection::begin_inbound_message( - size_t len, bool is_text, int64_t receive_time_us) { +SS_HOT SendspinConnection::InboundTarget SendspinConnection::begin_inbound_message(size_t len, + bool is_text) { if (this->fragment_assembly_open_) { // RFC 6455 section 5.4: the fragments of one message are not interleaved with another // data message. Failing here also keeps the assembly's continuations off a fallback @@ -506,8 +506,7 @@ SS_HOT SendspinConnection::InboundTarget SendspinConnection::begin_inbound_messa return {nullptr, InboundRoute::CLOSE}; } const InboundTarget target = - this->route_inbound_message(len, is_text ? InboundKind::TEXT : InboundKind::BINARY, - static_cast(receive_time_us)); + this->route_inbound_message(len, is_text ? InboundKind::TEXT : InboundKind::BINARY); if (target.route == InboundRoute::DROP) { // A dropped message is read and discarded without holding anything, so its start // stands in for its completion as proof the peer is alive. @@ -522,8 +521,7 @@ SS_HOT SendspinConnection::InboundTarget SendspinConnection::begin_inbound_messa } SendspinConnection::InboundTarget SendspinConnection::route_inbound_message(size_t len, - InboundKind kind, - uint32_t stamp) { + InboundKind kind) { if (this->inbound_ring_ == nullptr || this->inbound_gate_.is_detached()) { return {nullptr, InboundRoute::DROP}; } @@ -534,7 +532,7 @@ SendspinConnection::InboundTarget SendspinConnection::route_inbound_message(size return {nullptr, InboundRoute::DROP}; } if (len <= InboundGate::PRE_ADMISSION_MESSAGE_BYTES) { - return this->route_to_fallback(len, kind, stamp, /*admitted=*/false); + return this->route_to_fallback(len, kind, /*admitted=*/false); } // The message still pending may be the server/activate that admits this connection, so // the cap is judged once the protocol task has consumed it. @@ -562,7 +560,7 @@ SendspinConnection::InboundTarget SendspinConnection::route_inbound_message(size // ring items (see InboundGate). The rare path; the buffer is allocated for it and released at // the next ring write. if (len > this->inbound_ring_->max_message_bytes()) { - return this->route_to_fallback(len, kind, stamp, /*admitted=*/true); + return this->route_to_fallback(len, kind, /*admitted=*/true); } // A pre-admission message still pending from before the admission holds every later write // back, so the protocol task sees this connection's messages in order. @@ -600,16 +598,13 @@ SendspinConnection::InboundTarget SendspinConnection::route_inbound_message(size } InboundItemHeader* header = inbound_item_header(item); header->connection_id = static_cast(this->instance_id); - header->receive_time_us = stamp; header->kind = kind; this->inbound_item_ = item; - this->inbound_item_len_ = len; return {inbound_item_bytes(item), InboundRoute::RECEIVE}; } SendspinConnection::InboundTarget SendspinConnection::route_to_fallback(size_t len, InboundKind kind, - uint32_t stamp, bool admitted) { if (admitted) { // An admitted connection's message waits for the buffer no longer than one waits for @@ -642,7 +637,6 @@ SendspinConnection::InboundTarget SendspinConnection::route_to_fallback(size_t l } this->fallback_len_ = len; this->fallback_kind_ = kind; - this->fallback_receive_time_us_ = stamp; this->inbound_to_fallback_ = true; return {this->fallback_buf_.data(), InboundRoute::RECEIVE}; } @@ -672,10 +666,7 @@ void SendspinConnection::end_inbound_message(bool received, int64_t complete_tim if (this->inbound_item_ != nullptr) { void* item = std::exchange(this->inbound_item_, nullptr); if (received) { - if (this->inbound_item_len_ > SINGLE_SEGMENT_MESSAGE_BYTES) { - inbound_item_header(item)->receive_time_us = - static_cast(complete_time_us); - } + inbound_item_header(item)->receive_time_us = static_cast(complete_time_us); this->inbound_ring_->complete(item); } else { // FreeRTOS cannot cancel an acquire: complete it as DISCARD, which take() returns @@ -689,9 +680,7 @@ void SendspinConnection::end_inbound_message(bool received, int64_t complete_tim if (!received) { return; } - if (this->fallback_len_ > SINGLE_SEGMENT_MESSAGE_BYTES) { - this->fallback_receive_time_us_ = static_cast(complete_time_us); - } + this->fallback_receive_time_us_ = static_cast(complete_time_us); if (!this->inbound_gate_.publish_pending_message()) { return; } @@ -788,8 +777,6 @@ void SendspinConnection::end_inbound_fragment(size_t len, bool last, int64_t com if (!last) { return; } - // A multi-frame message may span segments. - this->fallback_receive_time_us_ = static_cast(complete_time_us); if (!this->inbound_gate_.is_admitted()) { this->inbound_to_fallback_ = true; this->end_inbound_message(true, complete_time_us); @@ -798,8 +785,7 @@ void SendspinConnection::end_inbound_fragment(size_t len, bool last, int64_t com // Admitted: copy the assembled message into a ring item, the one copy this path costs over // a single-frame message, and release the buffer. const size_t total = this->fallback_len_; - const InboundTarget target = - this->route_inbound_message(total, this->fallback_kind_, this->fallback_receive_time_us_); + const InboundTarget target = this->route_inbound_message(total, this->fallback_kind_); if (target.route != InboundRoute::RECEIVE) { return; } diff --git a/src/connection.h b/src/connection.h index c6fcdcd5..fe167cd9 100644 --- a/src/connection.h +++ b/src/connection.h @@ -898,24 +898,19 @@ class SendspinConnection : public std::enable_shared_from_thisbegin_inbound_message(frame_len, is_text, receive_time); + const InboundTarget target = this->begin_inbound_message(frame_len, is_text); this->chunk_dest_ = target.route == InboundRoute::RECEIVE ? target.data : nullptr; } if (this->chunk_dest_ == nullptr) { diff --git a/src/esp/server_connection.cpp b/src/esp/server_connection.cpp index fb7009e2..f31be643 100644 --- a/src/esp/server_connection.cpp +++ b/src/esp/server_connection.cpp @@ -260,8 +260,7 @@ void SendspinServerConnection::trigger_close() { httpd_sess_trigger_close(this->server_, this->sockfd_); } -SS_HOT esp_err_t SendspinServerConnection::handle_data(httpd_req_t* req, int64_t receive_time, - SendspinWsServer* server) { +SS_HOT esp_err_t SendspinServerConnection::handle_data(httpd_req_t* req, SendspinWsServer* server) { // The connection was delivered (and wired to the inbound ring) from the upgrade GET before any // frame can arrive; a frame on a never-delivered or released connection is dropped by the // inbound routing (begin_inbound_message()). @@ -285,7 +284,7 @@ SS_HOT esp_err_t SendspinServerConnection::handle_data(httpd_req_t* req, int64_t if (!continuation && ws_pkt.final) { // A single-frame message, the only kind a conforming peer sends: received straight into // its destination, a ring item once the connection is admitted. - const InboundTarget target = this->begin_inbound_message(ws_pkt.len, is_text, receive_time); + const InboundTarget target = this->begin_inbound_message(ws_pkt.len, is_text); if (target.route == InboundRoute::CLOSE) { return ESP_FAIL; } diff --git a/src/esp/server_connection.h b/src/esp/server_connection.h index 1fdf72b3..f33c248a 100644 --- a/src/esp/server_connection.h +++ b/src/esp/server_connection.h @@ -150,10 +150,9 @@ class SendspinServerConnection : public SendspinConnection { /// message straight into its ring item (or, before admission, the fallback buffer), a frame of /// a multi-frame message into the fallback buffer it is assembled in. On the httpd task. /// @param req The httpd request containing the WebSocket frame. - /// @param receive_time Timestamp when the data was received. /// @param server The server whose discard buffer a dropped frame is read into. /// @return ESP_OK on success; an error makes httpd close the session. - esp_err_t handle_data(httpd_req_t* req, int64_t receive_time, SendspinWsServer* server); + esp_err_t handle_data(httpd_req_t* req, SendspinWsServer* server); protected: /// @brief Queues the frame like send_binary_message(), carrying `before_write` to the httpd diff --git a/src/esp/ws_server.cpp b/src/esp/ws_server.cpp index 4ea2715d..c43a8815 100644 --- a/src/esp/ws_server.cpp +++ b/src/esp/ws_server.cpp @@ -320,9 +320,6 @@ void SendspinWsServer::close_callback(httpd_handle_t handle, int sockfd) { } SS_HOT esp_err_t SendspinWsServer::websocket_handler(httpd_req_t* req) { - // Capture timestamp immediately for accurate time synchronization - int64_t receive_time = esp_timer_get_time(); - // Look up the connection via httpd's per-session ctx. The slot was pinned in open_callback and // is freed only after this handler unwinds for a given session, so the shared_ptr copy below // is always valid. Copying the shared_ptr keeps the conn alive for the duration of dispatch @@ -355,7 +352,7 @@ SS_HOT esp_err_t SendspinWsServer::websocket_handler(httpd_req_t* req) { // inbound gate, so a still-alive session-pinned conn does not leak messages into freshly-reset // role queues. A frame on a never-delivered connection (its session outliving a // tick() reap by a moment) has no inbound ring to go to and is dropped the same way. - return conn->handle_data(req, receive_time, server); + return conn->handle_data(req, server); } } // namespace sendspin diff --git a/src/host/client_connection.cpp b/src/host/client_connection.cpp index c4fd9ccf..297a428c 100644 --- a/src/host/client_connection.cpp +++ b/src/host/client_connection.cpp @@ -168,8 +168,7 @@ void SendspinClientConnection::setup_callbacks() { // single-frame path: one copy, from IXWebSocket's string into a ring item (or, // before admission, into the fallback buffer). const std::string& data = msg->str; - const InboundTarget target = - this->begin_inbound_message(data.size(), !msg->binary, receive_time); + const InboundTarget target = this->begin_inbound_message(data.size(), !msg->binary); if (target.route == InboundRoute::RECEIVE) { std::copy(data.begin(), data.end(), target.data); this->end_inbound_message(true, receive_time); diff --git a/src/host/server_connection.cpp b/src/host/server_connection.cpp index 21b37dac..82dfad32 100644 --- a/src/host/server_connection.cpp +++ b/src/host/server_connection.cpp @@ -100,7 +100,7 @@ void SendspinServerConnection::handle_message(const std::string& data, bool is_b // IXWebSocket hands over each message reassembled, so every message takes the single-frame // path: one copy, from IXWebSocket's string into a ring item (or, before admission, into // the fallback buffer). - const InboundTarget target = this->begin_inbound_message(data.size(), !is_binary, receive_time); + const InboundTarget target = this->begin_inbound_message(data.size(), !is_binary); if (target.route != InboundRoute::RECEIVE) { return; } diff --git a/tests/loopback_connection.h b/tests/loopback_connection.h index 091f53ab..0e967367 100644 --- a/tests/loopback_connection.h +++ b/tests/loopback_connection.h @@ -94,7 +94,6 @@ class TestConnection : public SendspinConnection { using SendspinConnection::end_inbound_message; using SendspinConnection::InboundRoute; using SendspinConnection::InboundTarget; - using SendspinConnection::SINGLE_SEGMENT_MESSAGE_BYTES; // --- Test helpers --- diff --git a/tests/test_client_lifecycle.cpp b/tests/test_client_lifecycle.cpp index d31cc1eb..97ddce6b 100644 --- a/tests/test_client_lifecycle.cpp +++ b/tests/test_client_lifecycle.cpp @@ -1946,7 +1946,7 @@ TEST(NextDeadline, AFallbackMessageHeldBehindARingItemRunsTheNextTickAtOnce) { // The transport side: a ring item, then the second message. const auto receive = [&](size_t len) { - const auto target = conn->begin_inbound_message(len, false, platform_time_us()); + const auto target = conn->begin_inbound_message(len, false); ASSERT_EQ(target.route, SendspinConnection::InboundRoute::RECEIVE); std::memset(target.data, 0x5A, len); conn->end_inbound_message(true, platform_time_us()); diff --git a/tests/test_noise_transport.cpp b/tests/test_noise_transport.cpp index a489ccd1..78a16cc8 100644 --- a/tests/test_noise_transport.cpp +++ b/tests/test_noise_transport.cpp @@ -2114,7 +2114,7 @@ struct InboundHarness { // Receives one complete message as a single-frame transport does. TestConnection::InboundRoute receive(const std::vector& bytes) { const TestConnection::InboundTarget target = - this->conn.begin_inbound_message(bytes.size(), /*is_text=*/false, platform_time_us()); + this->conn.begin_inbound_message(bytes.size(), /*is_text=*/false); if (target.route == TestConnection::InboundRoute::RECEIVE) { std::memcpy(target.data, bytes.data(), bytes.size()); this->conn.end_inbound_message(true, platform_time_us()); @@ -2435,7 +2435,7 @@ TEST(InboundReceive, AnAbandonedRingItemIsNeverHandedToTheProtocolTask) { h.conn.set_admitted(true); const TestConnection::InboundTarget target = - h.conn.begin_inbound_message(bytes.size(), /*is_text=*/false, platform_time_us()); + h.conn.begin_inbound_message(bytes.size(), /*is_text=*/false); ASSERT_EQ(target.route, TestConnection::InboundRoute::RECEIVE); ASSERT_TRUE(in_ring_storage(h.ring, target.data)) << "an admitted message skipped the ring"; std::memcpy(target.data, bytes.data(), bytes.size() / 2); @@ -2600,7 +2600,7 @@ TEST(InboundReceive, TheLivenessStampIsTakenWhenAMessageCompletes) { if (row.shape == Shape::SINGLE_FRAME) { const TestConnection::InboundTarget target = - h.conn.begin_inbound_message(bytes.size(), /*is_text=*/false, platform_time_us()); + h.conn.begin_inbound_message(bytes.size(), /*is_text=*/false); ASSERT_EQ(target.route, route); if (target.route == TestConnection::InboundRoute::RECEIVE) { std::memcpy(target.data, bytes.data(), bytes.size()); @@ -2639,44 +2639,32 @@ TEST(InboundReceive, TheLivenessStampIsTakenWhenAMessageCompletes) { } } -// A single-frame message up to SINGLE_SEGMENT_MESSAGE_BYTES keeps its first stamp; a longer one, -// or a multi-frame message of any length, gets its completion stamp. -TEST(InboundReceive, ASmallSingleFrameMessageKeepsItsFirstStampAndOthersTheirCompletion) { +// A message's receive stamp is its completion stamp on every route. +TEST(InboundReceive, AMessageIsStampedWhenItFinishesArriving) { enum class Shape : uint8_t { SINGLE_FRAME, MULTI_FRAME }; struct Row { const char* name; bool admitted; // admitted: a ring item; unadmitted: the fallback buffer Shape shape; - size_t len; - bool completion_stamp; }; - constexpr size_t BOUND = TestConnection::SINGLE_SEGMENT_MESSAGE_BYTES; const Row rows[] = { - {"ring item at the bound", true, Shape::SINGLE_FRAME, BOUND, false}, - {"ring item past the bound", true, Shape::SINGLE_FRAME, BOUND + 1, true}, - {"fallback buffer at the bound", false, Shape::SINGLE_FRAME, BOUND, false}, - {"fallback buffer past the bound", false, Shape::SINGLE_FRAME, BOUND + 1, true}, - {"multi-frame into a ring item at the bound", true, Shape::MULTI_FRAME, BOUND, true}, - {"multi-frame into a ring item past the bound", true, Shape::MULTI_FRAME, BOUND + 1, - true}, - {"multi-frame into the fallback buffer at the bound", false, Shape::MULTI_FRAME, BOUND, - true}, - {"multi-frame into the fallback buffer past the bound", false, Shape::MULTI_FRAME, - BOUND + 1, true}, + {"single frame into a ring item", true, Shape::SINGLE_FRAME}, + {"single frame into the fallback buffer", false, Shape::SINGLE_FRAME}, + {"multi-frame into a ring item", true, Shape::MULTI_FRAME}, + {"multi-frame into the fallback buffer", false, Shape::MULTI_FRAME}, }; - constexpr int64_t FIRST_US = 1'000; constexpr int64_t MIDDLE_US = 3'000; constexpr int64_t COMPLETE_US = 5'000; + const std::vector bytes(64, 0x42); for (const Row& row : rows) { SCOPED_TRACE(row.name); InboundHarness h; h.conn.set_admitted(row.admitted); - const std::vector bytes(row.len, 0x42); if (row.shape == Shape::SINGLE_FRAME) { const TestConnection::InboundTarget target = - h.conn.begin_inbound_message(bytes.size(), /*is_text=*/false, FIRST_US); + h.conn.begin_inbound_message(bytes.size(), /*is_text=*/false); ASSERT_EQ(target.route, TestConnection::InboundRoute::RECEIVE); std::memcpy(target.data, bytes.data(), bytes.size()); h.conn.end_inbound_message(true, COMPLETE_US); @@ -2694,21 +2682,17 @@ TEST(InboundReceive, ASmallSingleFrameMessageKeepsItsFirstStampAndOthersTheirCom h.conn.end_inbound_fragment(bytes.size() - first_len, /*last=*/true, COMPLETE_US); } - const uint32_t expected = - static_cast(row.completion_stamp ? COMPLETE_US : FIRST_US); if (row.admitted) { size_t len = 0; void* item = h.ring.take(&len, 0); ASSERT_NE(item, nullptr) << "the admitted message skipped the ring"; - EXPECT_EQ(len, bytes.size()); - EXPECT_EQ(inbound_item_header(item)->receive_time_us, expected); + EXPECT_EQ(inbound_item_header(item)->receive_time_us, COMPLETE_US); h.conn.inbound_gate().note_item_taken(); h.ring.return_item(item); } else { InboundMessage pending; ASSERT_TRUE(h.conn.pending_message(pending)); - EXPECT_EQ(pending.len, bytes.size()); - EXPECT_EQ(pending.receive_time_us, expected); + EXPECT_EQ(pending.receive_time_us, COMPLETE_US); h.conn.consume_pending_message(); } }