Start the signaling send timeout at the first write attempt

app/src/WebSocketClient.cpp

@@ -330,7 +330,7 @@ bool WebSocketClient::send_frame(uint8_t opcode, const uint8_t* payload, size_t
         frame.push_back(payload[i] ^ mask_key[i % 4]);
     }
 
-    return tx_queue_.enqueue(std::move(frame), opennow::websocket::WriteQueue::Clock::now());
+    return tx_queue_.enqueue(std::move(frame));
 }
 
 void WebSocketClient::send_message(const std::string& msg) {

app/src/websocket_write_queue.hpp

@@ -5,6 +5,7 @@
 #include <cstddef>
 #include <cstdint>
 #include <deque>
+#include <optional>
 #include <utility>
 #include <vector>
 
@@ -35,12 +36,12 @@ class WriteQueue
                bytes <= MaximumBytes - queued_bytes_;
     }
 
-    bool enqueue(std::vector<uint8_t>&& bytes, Clock::time_point now)
+    bool enqueue(std::vector<uint8_t>&& bytes)
     {
         if (!can_enqueue(bytes.size()))
             return false;
         queued_bytes_ += bytes.size();
-        frames_.push_back({std::move(bytes), 0, now + SendTimeout});
+        frames_.push_back({std::move(bytes), 0, std::nullopt});
         return true;
     }
 
@@ -55,7 +56,9 @@ class WriteQueue
     {
         for (size_t writes = 0; writes < MaximumWritesPerPoll && !frames_.empty(); ++writes) {
             Frame& frame = frames_.front();
-            if (now() >= frame.deadline)
+            if (!frame.deadline)
+                frame.deadline = now() + SendTimeout;
+            if (now() >= *frame.deadline)
                 return DrainResult::TimedOut;
             const size_t length = std::min(MaximumWriteBytes, frame.bytes.size() - frame.offset);
             const WriteResult result = writer(frame.bytes.data() + frame.offset, length);
@@ -79,7 +82,7 @@ class WriteQueue
     {
         std::vector<uint8_t> bytes;
         size_t offset;
-        Clock::time_point deadline;
+        std::optional<Clock::time_point> deadline;
     };
 
     std::deque<Frame> frames_;

tests/websocket_write_queue_test.cpp

@@ -44,26 +44,26 @@ int main()
     {
         Queue queue;
         FakeWriter writer;
-        assert(queue.enqueue({1, 2, 3, 4}, time));
-        assert(queue.enqueue({5, 6}, time));
+        assert(queue.enqueue({1, 2, 3, 4}));
+        assert(queue.enqueue({5, 6}));
         writer.results = {{Status::Progress, 2}, {Status::Again, 0}};
         assert(queue.drain(writer, now) == Result::Pending);
         assert(writer.calls == 2);
         assert((writer.output == std::vector<uint8_t>{1, 2}));
         writer.results = {{Status::Again, 0}};
         assert(queue.drain(writer, now) == Result::Pending);
         assert(writer.calls == 3);
-        assert(queue.enqueue({7, 8}, time));
+        assert(queue.enqueue({7, 8}));
         assert(queue.drain(writer, now) == Result::Complete);
         assert((writer.output == std::vector<uint8_t>{1, 2, 3, 4, 5, 6, 7, 8}));
     }
 
     {
         Queue queue;
         FakeWriter writer;
-        assert(queue.enqueue(std::vector<uint8_t>(Queue::MaximumBytes, 9), time));
+        assert(queue.enqueue(std::vector<uint8_t>(Queue::MaximumBytes, 9)));
         assert(!queue.can_enqueue(1));
-        assert(!queue.enqueue({1}, time));
+        assert(!queue.enqueue({1}));
         assert(queue.drain(writer, now) == Result::Pending);
         assert(writer.calls == Queue::MaximumWritesPerPoll);
         assert(writer.output.size() == Queue::MaximumWritesPerPoll * Queue::MaximumWriteBytes);
@@ -85,19 +85,19 @@ int main()
         Queue queue;
         FakeWriter writer;
         for (size_t i = 0; i < Queue::MaximumFrames; ++i)
-            assert(queue.enqueue({static_cast<uint8_t>(i)}, time));
-        assert(!queue.enqueue({65}, time));
+            assert(queue.enqueue({static_cast<uint8_t>(i)}));
+        assert(!queue.enqueue({65}));
         assert(queue.drain(writer, now) == Result::Pending);
         assert(writer.calls == Queue::MaximumWritesPerPoll);
         assert(queue.can_enqueue(1));
         assert(!queue.can_enqueue(0));
-        assert(!queue.enqueue({}, time));
+        assert(!queue.enqueue({}));
     }
 
     {
         Queue queue;
         FakeWriter writer;
-        assert(queue.enqueue({1, 2, 3}, time));
+        assert(queue.enqueue({1, 2, 3}));
         writer.results = {{Status::Progress, 1}, {Status::Again, 0}};
         assert(queue.drain(writer, now) == Result::Pending);
         time += std::chrono::milliseconds(999);
@@ -110,41 +110,45 @@ int main()
         assert((writer.output == std::vector<uint8_t>{1, 2}));
         queue.reset();
         writer = {};
-        assert(queue.enqueue({4, 5}, time));
+        assert(queue.enqueue({4, 5}));
         assert(queue.drain(writer, now) == Result::Complete);
         assert((writer.output == std::vector<uint8_t>{4, 5}));
     }
 
     {
         Queue queue;
         FakeWriter writer;
-        assert(queue.enqueue({1}, time));
-        assert(queue.enqueue({2}, time));
-        time += Queue::SendTimeout - std::chrono::milliseconds(1);
+        assert(queue.enqueue({1}));
+        assert(queue.enqueue({2}));
+        time += Queue::SendTimeout * 3;
         auto advancing_writer = [&](const uint8_t* data, size_t length) {
             const auto result = writer(data, length);
-            time += std::chrono::milliseconds(1);
+            time += Queue::SendTimeout - std::chrono::milliseconds(1);
             return result;
         };
-        assert(queue.drain(advancing_writer, now) == Result::TimedOut);
-        assert(writer.calls == 1);
-        assert((writer.output == std::vector<uint8_t>{1}));
+        assert(queue.drain(advancing_writer, now) == Result::Complete);
+        assert(writer.calls == 2);
+        assert((writer.output == std::vector<uint8_t>{1, 2}));
     }
 
     {
         Queue queue;
         FakeWriter writer;
-        assert(queue.enqueue({1}, time));
+        assert(queue.enqueue({1}));
+        assert(queue.enqueue({2}));
+        writer.results = {{Status::Again, 0}};
+        assert(queue.drain(writer, now) == Result::Pending);
         time += Queue::SendTimeout;
         assert(queue.drain(writer, now) == Result::TimedOut);
-        assert(writer.calls == 0);
+        assert(writer.calls == 1);
+        assert(writer.output.empty());
     }
 
     for (const auto status : {Status::Error, Status::Progress}) {
         Queue queue;
         FakeWriter writer;
-        assert(queue.enqueue({1, 2}, time));
-        assert(queue.enqueue({3}, time));
+        assert(queue.enqueue({1, 2}));
+        assert(queue.enqueue({3}));
         writer.results = {{status, 0}};
         assert(queue.drain(writer, now) ==
                (status == Status::Error ? Result::WriteError : Result::Stalled));
@@ -159,7 +163,7 @@ int main()
     {
         Queue queue;
         FakeWriter writer;
-        assert(queue.enqueue({1}, time));
+        assert(queue.enqueue({1}));
         writer.results = {{Status::Progress, 2}};
         assert(queue.drain(writer, now) == Result::WriteError);
     }