Stamp every inbound message when its last byte arrives

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<uint32_t>(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,15 +598,13 @@ SendspinConnection::InboundTarget SendspinConnection::route_inbound_message(size
     }
     InboundItemHeader* header = inbound_item_header(item);
     header->connection_id = static_cast<uint32_t>(this->instance_id);
-    header->receive_time_us = stamp;
     header->kind = kind;
     this->inbound_item_ = item;
     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
@@ -641,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};
 }
@@ -664,13 +659,14 @@ 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) {
+            inbound_item_header(item)->receive_time_us = static_cast<uint32_t>(complete_time_us);
             this->inbound_ring_->complete(item);
         } else {
             // FreeRTOS cannot cancel an acquire: complete it as DISCARD, which take() returns
@@ -681,7 +677,11 @@ 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;
+        }
+        this->fallback_receive_time_us_ = static_cast<uint32_t>(complete_time_us);
+        if (!this->inbound_gate_.publish_pending_message()) {
             return;
         }
     } else {
@@ -692,15 +692,15 @@ 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;
     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_) {
@@ -728,7 +728,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<uint32_t>(receive_time_us);
         }
     } else if (!this->fragment_assembly_open_) {
         // RFC 6455 section 5.4: a continuation frame continues a fragmented message, so one with
@@ -763,7 +762,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,27 +779,26 @@ 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
     // 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;
     }
     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);
 }
 
 // ============================================================================

src/connection.h

@@ -903,14 +903,15 @@ class SendspinConnection : public std::enable_shared_from_this<SendspinConnectio
     /// is being assembled closes the connection (RFC 6455 section 5.4).
     /// @param len Message length in bytes.
     /// @param is_text Whether the message arrived in a text frame.
-    /// @param receive_time_us platform_time_us() when the transport received it.
-    InboundTarget begin_inbound_message(size_t len, bool is_text, int64_t receive_time_us);
+    InboundTarget begin_inbound_message(size_t len, bool is_text);
 
     /// @brief Ends the message begin_inbound_message() routed to RECEIVE: publishes it to the
     /// protocol task and wakes it. Transport thread.
     /// @param received false when the receive failed after the start: a ring item is completed
     ///        as InboundKind::DISCARD, a fallback message is not published.
-    void end_inbound_message(bool received);
+    /// @param complete_time_us platform_time_us() when the message finished arriving: its
+    ///        receive time.
+    void end_inbound_message(bool received, int64_t complete_time_us);
 
     /// @brief Starts receiving `len` more bytes of a multi-frame WebSocket message (a message
     /// sent as a frame plus continuation frames), assembled in the fallback buffer. Transport
@@ -926,15 +927,14 @@ class SendspinConnection : public std::enable_shared_from_this<SendspinConnectio
     /// @param len Bytes this frame (or this chunk of it) carries.
     /// @param first Whether these are the first bytes of the message.
     /// @param is_text Whether the message's first frame is a text frame; read when `first`.
-    /// @param receive_time_us platform_time_us() when the transport received these bytes.
-    InboundTarget begin_inbound_fragment(size_t len, bool first, bool is_text,
-                                         int64_t receive_time_us);
+    InboundTarget begin_inbound_fragment(size_t len, bool first, bool is_text);
 
     /// @brief Commits the bytes begin_inbound_fragment() routed to RECEIVE and, on the message's
     /// last bytes, publishes the assembled message. Transport thread.
     /// @param len Bytes received into the target.
     /// @param last Whether these complete the message.
-    void end_inbound_fragment(size_t len, bool last);
+    /// @param complete_time_us platform_time_us() when these bytes finished arriving.
+    void end_inbound_fragment(size_t len, bool last, int64_t complete_time_us);
 
     /// @brief Gives up on the message being received, for a transport that stops part-way
     /// through one (a message delivered in several chunks): a ring item is completed as
@@ -955,7 +955,7 @@ class SendspinConnection : public std::enable_shared_from_this<SendspinConnectio
     /// the message still pending to be consumed (it may be the one that admits the connection,
     /// wait_until_writable()) and closes the connection if it is still unadmitted then. A
     /// detached or unattached connection drops everything.
-    InboundTarget route_inbound_message(size_t len, InboundKind kind, uint32_t stamp);
+    InboundTarget route_inbound_message(size_t len, InboundKind kind);
 
     /// @brief Routes a complete message of `len` bytes to the fallback buffer, in order with the
     /// connection's ring items, once the protocol task has consumed the previous one, allocating
@@ -967,7 +967,7 @@ class SendspinConnection : public std::enable_shared_from_this<SendspinConnectio
     /// closes the connection with a warning, for the same reason a full ring does (a detached
     /// connection's message is dropped instead).
     /// @return RECEIVE into the buffer; DROP or CLOSE as above; an allocation failure closes.
-    InboundTarget route_to_fallback(size_t len, InboundKind kind, uint32_t stamp, bool admitted);
+    InboundTarget route_to_fallback(size_t len, InboundKind kind, bool admitted);
 
     /// @brief InboundGate::wait_until_writable() bounded by InboundGate::WRITABLE_WAIT_MS; a
     /// timeout on a connection that is not detached closes it through fail_inbound(). Transport

src/esp/client_connection.cpp

@@ -268,8 +268,7 @@ void SendspinClientConnection::handle_data(const esp_websocket_event_data_t* dat
         // connection is admitted. The destination is chosen on the first chunk, from the frame's
         // full length.
         if (offset == 0) {
-            const InboundTarget target =
-                this->begin_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) {
@@ -280,23 +279,22 @@ void SendspinClientConnection::handle_data(const esp_websocket_event_data_t* dat
         }
         if (frame_done) {
             this->chunk_dest_ = nullptr;
-            this->end_inbound_message(true);
+            this->end_inbound_message(true, receive_time);
         }
         return;
     }
 
     // A chunk of a multi-frame message (the rare path; see begin_inbound_fragment()).
     const bool first = !continuation && offset == 0;
     const bool last = data->fin && 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);
         }
-        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);
     }
 }
 

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"
@@ -259,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()).
@@ -284,35 +284,34 @@ 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;
         }
         if (target.route == InboundRoute::DROP) {
             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;
     }
 
     // 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;
     }
     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);
     if (ret != ESP_OK) {
         // 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;
 }
 

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

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