From 722ad210ce4aca14e7e63e21cff2be4135a8f5ed Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 25 Aug 2026 21:37:56 +0000 Subject: [PATCH 01/11] Give the async receive a deadline across the WebSocket contract (#130) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ReceiveAsync(timeout, callback): the blocking Receive(timeout)'s completion-driven twin — the same four outcomes, Error::Timeout when the budget runs out, the slot released and the session untouched, non-positive polls. A refusing default keeps every implementor compiling (the async family is opt-in, unlike the pure-virtual blocking overload); the shared contract suite grows four deadline cases that hold every SupportsAsync() implementation to it. The timeout/delivery race settles exactly once under the session lock via a park generation: every park bumps it and a deadline fires only for the generation it was armed with, so a stale deadline can never time out a later receive. Beast arms a steady_timer on the connection's executor (weakly captured — a deadline never extends a session); the executor-less in-memory pair arms a short-lived watchdog thread on its shared state; the JsonRpcStreamSocket wrapper passes the deadline through its classifier, where a timeout rides untouched — no violation, no close. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_014C7WdBD99mUFWGxMvGSjSU --- .../eventstream/jsonrpc_stream_socket.h | 10 ++ runtime/include/smithy/http/websocket.h | 25 ++++ .../src/eventstream/jsonrpc_stream_socket.cc | 59 +++++---- runtime/src/http/beast_transport.cc | 85 ++++++++++++- runtime/src/http/websocket_pair.cc | 113 ++++++++++++++---- .../smithy/testing/websocket_contract_test.h | 101 ++++++++++++++++ .../eventstream/jsonrpc_stream_socket_test.cc | 34 ++++++ runtime/tests/http/beast_websocket_test.cc | 42 +++++++ 8 files changed, 420 insertions(+), 49 deletions(-) diff --git a/runtime/include/smithy/eventstream/jsonrpc_stream_socket.h b/runtime/include/smithy/eventstream/jsonrpc_stream_socket.h index 5787d20b..ae1cb79c 100644 --- a/runtime/include/smithy/eventstream/jsonrpc_stream_socket.h +++ b/runtime/include/smithy/eventstream/jsonrpc_stream_socket.h @@ -76,6 +76,11 @@ class JsonRpcStreamSocket final : public http::WebSocket { Outcome Send(const Message& message) override; void Close() override; void ReceiveAsync(ReceiveCallback callback) override; + // The async deadline (#130), worn by the wrapper the way the blocking + // one is: a timeout is the inner Error::Timeout verbatim — no envelope + // classification runs on it, nothing is closed, and the stream picks up + // where it left off. + void ReceiveAsync(std::chrono::milliseconds timeout, ReceiveCallback callback) override; void SendAsync(const Message& message, SendCallback callback) override; bool SupportsAsync() const override; @@ -84,6 +89,11 @@ class JsonRpcStreamSocket final : public http::WebSocket { // answer a violation, close if the frame was fatal, hand back the rest. Outcome> Police(Outcome> raw); + // Both async receives: the classifying completion that Police-es one + // inbound outcome before handing it to `callback` — shared so the timed + // and untimed paths cannot drift. + ReceiveCallback PoliceAsync(ReceiveCallback callback); + // Engaged only by the owning form; every call goes through inner_. std::shared_ptr owner_; http::WebSocket* inner_; diff --git a/runtime/include/smithy/http/websocket.h b/runtime/include/smithy/http/websocket.h index fba27e10..dd586425 100644 --- a/runtime/include/smithy/http/websocket.h +++ b/runtime/include/smithy/http/websocket.h @@ -198,6 +198,31 @@ class WebSocket { "check; the blocking Receive/Send still work)")); } + // The blocking deadline's completion-driven twin (#130): the callback + // fires exactly once with the same four outcomes as Receive(timeout) — + // the message, the peer's clean close (nullopt), the session's permanent + // error, or Error::Timeout ("TimeoutError") when the budget runs out + // with nothing to report. Exactly like the blocking overload, a timeout + // is not terminal: the session is untouched and usable, and the receive + // slot is released exactly as a completed receive releases it — the next + // receive (either API) may park again, and a message the wire delivers + // after the deadline waits in the session for it. A non-positive timeout + // polls: it completes inline with what is already in hand, or with the + // timeout. The one-outstanding-receive rule is unchanged. + // + // The timeout must race the completion and settle exactly once — the + // implementation owns that race the same way it owns the parked slot + // (there is nowhere honest to bound the wait from outside; the blocking + // overload's doc block explains why). The refusing default keeps every + // existing implementor compiling; the contract suite holds anything that + // reports SupportsAsync() to overriding this too. + virtual void ReceiveAsync(std::chrono::milliseconds timeout, ReceiveCallback callback) { + (void)timeout; + callback(Error::Validation( + "websocket: this implementation has no deadline-bounded async receive (the untimed " + "ReceiveAsync and the blocking Receive(timeout) may still work)")); + } + virtual void SendAsync(const eventstream::Message& message, SendCallback callback) { (void)message; callback(Error::Validation( diff --git a/runtime/src/eventstream/jsonrpc_stream_socket.cc b/runtime/src/eventstream/jsonrpc_stream_socket.cc index f4c42850..f908feb9 100644 --- a/runtime/src/eventstream/jsonrpc_stream_socket.cc +++ b/runtime/src/eventstream/jsonrpc_stream_socket.cc @@ -125,34 +125,45 @@ Outcome JsonRpcStreamSocket::Send(const Message& message) { void JsonRpcStreamSocket::Close() { inner_->Close(); } void JsonRpcStreamSocket::ReceiveAsync(ReceiveCallback callback) { + inner_->ReceiveAsync(PoliceAsync(std::move(callback))); +} + +void JsonRpcStreamSocket::ReceiveAsync(std::chrono::milliseconds timeout, + ReceiveCallback callback) { + // A timeout arrives as !ok() and rides through the classifier untouched + // (no violation, no close) — the wrapper stays exactly as usable as the + // socket under it, mirroring the blocking deadline overload. + inner_->ReceiveAsync(timeout, PoliceAsync(std::move(callback))); +} + +JsonRpcStreamSocket::ReceiveCallback JsonRpcStreamSocket::PoliceAsync(ReceiveCallback callback) { // State travels by value into the completion: the wrapper may be gone by // the time the transport completes. The owning form pins the inner // socket through `owner`; the borrowing form rides its documented // blocking-seam lifetime contract. - inner_->ReceiveAsync( - [id = id_, role = role_, owner = owner_, inner = inner_, - callback = std::move(callback)](Outcome> raw) mutable { - Inbound inbound = ClassifyInbound(std::move(raw), id, role); - if (inbound.violation_text.has_value()) { - // The close AND the caller's resumption both ride the send's - // completion: the close cannot cancel the terminal write (the - // ADR-0021 lesson), and neither can the caller — a generated driver - // closes the session as it unwinds, which would escalate past the - // in-flight terminal — because it never runs until the frame is on - // the wire and the close is already requested. - auto terminal = std::make_shared(); - terminal->payload = Blob::FromString(*std::move(inbound.violation_text)); - inner->SendAsync(*terminal, - [terminal, owner, inner, result = std::move(inbound.result), - callback = std::move(callback)](const Outcome& /*sent*/) mutable { - inner->Close(); - callback(std::move(result)); - }); - return; - } - if (inbound.close) inner->Close(); - callback(std::move(inbound.result)); - }); + return [id = id_, role = role_, owner = owner_, inner = inner_, + callback = std::move(callback)](Outcome> raw) mutable { + Inbound inbound = ClassifyInbound(std::move(raw), id, role); + if (inbound.violation_text.has_value()) { + // The close AND the caller's resumption both ride the send's + // completion: the close cannot cancel the terminal write (the + // ADR-0021 lesson), and neither can the caller — a generated driver + // closes the session as it unwinds, which would escalate past the + // in-flight terminal — because it never runs until the frame is on + // the wire and the close is already requested. + auto terminal = std::make_shared(); + terminal->payload = Blob::FromString(*std::move(inbound.violation_text)); + inner->SendAsync(*terminal, + [terminal, owner, inner, result = std::move(inbound.result), + callback = std::move(callback)](const Outcome& /*sent*/) mutable { + inner->Close(); + callback(std::move(result)); + }); + return; + } + if (inbound.close) inner->Close(); + callback(std::move(inbound.result)); + }; } void JsonRpcStreamSocket::SendAsync(const Message& message, SendCallback callback) { diff --git a/runtime/src/http/beast_transport.cc b/runtime/src/http/beast_transport.cc index c0f18a5d..c28c84b6 100644 --- a/runtime/src/http/beast_transport.cc +++ b/runtime/src/http/beast_transport.cc @@ -10,6 +10,7 @@ #include #include #include +#include #include #include #include @@ -357,6 +358,24 @@ class WsSession final : public WebSocketSessionBase, // The completion-driven twin (ADR-0019): one outstanding receive-class // operation per session, completion on the connection's executor. void ReceiveAsync(WebSocket::ReceiveCallback callback) override { + ReceiveAsyncWithin(std::nullopt, std::move(callback)); + } + + // The deadline overload (#130): same immediate paths, and a parked + // receive races its deadline instead of waiting forever. + void ReceiveAsync(std::chrono::milliseconds timeout, + WebSocket::ReceiveCallback callback) override { + ReceiveAsyncWithin(timeout, std::move(callback)); + } + + // Both async receives: `timeout` engaged bounds the park with a timer on + // the connection's executor, disengaged parks until Deliver or a + // terminal transition. The park generation settles the timer/delivery + // race exactly once under mutex_: every park bumps it, the timer fires + // only for the generation it was armed with, so a stale deadline can + // never time out a later receive (timed or not). + void ReceiveAsyncWithin(std::optional timeout, + WebSocket::ReceiveCallback callback) { Outcome> immediate = std::optional(); bool ready_message = false; { @@ -376,9 +395,27 @@ class WsSession final : public WebSocketSessionBase, immediate = std::optional(); } else if (failed_) { immediate = Error::Transport("websocket: " + error_); + } else if (timeout.has_value() && *timeout <= std::chrono::milliseconds::zero()) { + // The blocking overload's poll shape: nothing already in hand is a + // timeout, inline like the other immediates below. + immediate = Error::Timeout("websocket: no message within the receive deadline"); } else { pending_receive_ = std::move(callback); - return; // Deliver / the terminal transitions complete it + ++receive_park_generation_; // a stale deadline must not fire this park + if (timeout.has_value()) { + try { + ArmReceiveDeadlineLocked(*timeout); + } catch (...) { + // Could not schedule the deadline: an unbounded park would + // betray the caller's whole ask, so refuse instead. + WebSocket::ReceiveCallback refused = std::exchange(pending_receive_, nullptr); + wake_.notify_all(); + lock.unlock(); + refused(Error::Transport("websocket: cannot schedule the receive deadline")); + return; + } + } + return; // Deliver / the deadline / the terminal transitions complete it } } if (!ready_message) { @@ -403,6 +440,42 @@ class WsSession final : public WebSocketSessionBase, } } + // With mutex_ held: (re)arms the one receive-deadline timer for the park + // that just went in. Emplacing destroys a previous timer, cancelling its + // pending wait (the handler sees operation_aborted and returns) — and the + // generation check makes even an uncancelled stale wait harmless. The + // handler captures the session weakly: a deadline must never extend a + // session's life. + void ArmReceiveDeadlineLocked(std::chrono::milliseconds timeout) { + receive_deadline_.emplace(ws_.get_executor()); + receive_deadline_->expires_after(timeout); + receive_deadline_->async_wait( + [weak = this->weak_from_this(), + generation = receive_park_generation_](const boost::system::error_code& ec) { + if (ec) return; // cancelled: a newer park re-armed, or teardown + if (auto self = weak.lock()) self->OnReceiveDeadline(generation); + }); + } + + void OnReceiveDeadline(std::uint64_t generation) { + WebSocket::ReceiveCallback expired; + { + const std::lock_guard lock(mutex_); + if (!pending_receive_ || receive_park_generation_ != generation) { + return; // the park this deadline bounded already completed + } + expired = std::exchange(pending_receive_, nullptr); + // A blocking receiver may be waiting behind the parked slot; the + // freed slot is part of its wake condition. + wake_.notify_all(); + } + // The timer fired on the connection's executor — the completion + // context — so contain a throwing application callback (ADR-0003). + InvokeCompletion("websocket receive", expired, + Outcome>( + Error::Timeout("websocket: no message within the receive deadline"))); + } + Outcome Send(const eventstream::Message& message) override { auto frame = EncodeForWire(message); if (!frame.ok()) { @@ -873,6 +946,12 @@ class WsSession final : public WebSocketSessionBase, // transition takes and fires them exactly once (TakeAsyncWaitersLocked). WebSocket::ReceiveCallback pending_receive_; WebSocket::SendCallback pending_send_; + // Bumped on every pending_receive_ park; a receive deadline fires only + // for the generation it was armed with (the timed-receive race settles + // here). The timer is optional so it can be (re)constructed on the + // connection's executor per timed park. + std::uint64_t receive_park_generation_ = 0; + std::optional receive_deadline_; int blocked_receivers_ = 0; bool read_paused_ = false; bool peer_closed_ = false; // clean close: Receive's nullopt @@ -2237,6 +2316,10 @@ class DialedWebSocket final : public WebSocket { void ReceiveAsync(WebSocket::ReceiveCallback callback) override { session_->ReceiveAsync(std::move(callback)); } + void ReceiveAsync(std::chrono::milliseconds timeout, + WebSocket::ReceiveCallback callback) override { + session_->ReceiveAsync(timeout, std::move(callback)); + } void SendAsync(const eventstream::Message& message, WebSocket::SendCallback callback) override { session_->SendAsync(message, std::move(callback)); } diff --git a/runtime/src/http/websocket_pair.cc b/runtime/src/http/websocket_pair.cc index ae744ebb..838c6a03 100644 --- a/runtime/src/http/websocket_pair.cc +++ b/runtime/src/http/websocket_pair.cc @@ -4,9 +4,11 @@ #include #include #include +#include #include #include #include +#include #include #include "smithy/eventstream/frame.h" @@ -44,6 +46,10 @@ struct PairState { // two halves, same shape as the Beast session). std::array pending_receive; std::array, 2> pending_send; + // Bumped on every pending_receive park; a receive-deadline watchdog fires + // only for the generation it was armed with, so a stale deadline can + // never time out a later receive (#130). + std::array receive_park_generation{}; std::array blocked_receivers{}; std::array blocked_senders{}; bool closed = false; @@ -132,30 +138,14 @@ class PairEnd final : public WebSocket { } void ReceiveAsync(WebSocket::ReceiveCallback callback) override { - WebSocket::SendCallback absorbed; - Outcome> immediate = std::optional(); - { - const std::lock_guard lock(state_->mutex); - if (state_->pending_receive[send_index_] || state_->blocked_receivers[send_index_] > 0) { - callback(Error::Validation("websocket pair: a receive is already outstanding")); - return; - } - std::deque& inbound = state_->queues[1 - send_index_]; - if (!inbound.empty()) { - eventstream::Message message = std::move(inbound.front()); - inbound.pop_front(); - absorbed = AbsorbPeerPendingSendLocked(); - state_->changed.notify_all(); - immediate = std::optional(std::move(message)); - } else if (state_->closed) { - immediate = std::optional(); - } else { - state_->pending_receive[send_index_] = std::move(callback); - return; // a send or the close completes it - } - } - if (absorbed) absorbed(Unit{}); - callback(std::move(immediate)); + ReceiveAsyncWithin(std::nullopt, std::move(callback)); + } + + // The deadline overload (#130): same immediate paths, and a parked + // receive races its deadline instead of waiting forever. + void ReceiveAsync(std::chrono::milliseconds timeout, + WebSocket::ReceiveCallback callback) override { + ReceiveAsyncWithin(timeout, std::move(callback)); } void SendAsync(const eventstream::Message& message, WebSocket::SendCallback callback) override { @@ -195,6 +185,78 @@ class PairEnd final : public WebSocket { bool SupportsAsync() const override { return true; } private: + // Both async receives: `timeout` engaged bounds the park, disengaged + // parks until a send or the close completes it. The pair has no executor + // to run a timer on, so a timed park arms a detached watchdog thread — + // short-lived (it exits at the deadline or as soon as the park + // completes), and it owns nothing but the shared state it sleeps on. The + // park generation settles the watchdog/delivery race exactly once under + // the state mutex, so a stale deadline can never time out a later + // receive (timed or not). + void ReceiveAsyncWithin(std::optional timeout, + WebSocket::ReceiveCallback callback) { + WebSocket::SendCallback absorbed; + Outcome> immediate = std::optional(); + { + const std::lock_guard lock(state_->mutex); + if (state_->pending_receive[send_index_] || state_->blocked_receivers[send_index_] > 0) { + callback(Error::Validation("websocket pair: a receive is already outstanding")); + return; + } + std::deque& inbound = state_->queues[1 - send_index_]; + if (!inbound.empty()) { + eventstream::Message message = std::move(inbound.front()); + inbound.pop_front(); + absorbed = AbsorbPeerPendingSendLocked(); + state_->changed.notify_all(); + immediate = std::optional(std::move(message)); + } else if (state_->closed) { + immediate = std::optional(); + } else if (timeout.has_value() && *timeout <= std::chrono::milliseconds::zero()) { + // The blocking overload's poll shape: nothing already in hand is a + // timeout, completed inline like the other immediates. + immediate = Error::Timeout("websocket pair: no message within the receive deadline"); + } else { + state_->pending_receive[send_index_] = std::move(callback); + ++state_->receive_park_generation[send_index_]; + if (timeout.has_value()) { + ArmReceiveDeadlineLocked(*timeout); + } + return; // a send, the deadline, or the close completes it + } + } + if (absorbed) absorbed(Unit{}); + callback(std::move(immediate)); + } + + // With the lock held: spawns the watchdog for the park that just went + // in. It sleeps on the shared condition variable, so a completed park + // (delivery, close, or a fresh park's bumped generation) releases it + // early; at the deadline, a park still bearing its generation is timed + // out — the slot is released exactly as a delivery releases it, and the + // session is untouched. + void ArmReceiveDeadlineLocked(std::chrono::milliseconds timeout) { + std::thread([state = state_, end = send_index_, + generation = state_->receive_park_generation[send_index_], + deadline = std::chrono::steady_clock::now() + timeout] { + WebSocket::ReceiveCallback expired; + { + std::unique_lock lock(state->mutex); + state->changed.wait_until(lock, deadline, [&] { + return !state->pending_receive[end] || state->receive_park_generation[end] != generation; + }); + if (!state->pending_receive[end] || state->receive_park_generation[end] != generation) { + return; // the park this deadline bounded already completed + } + expired = std::exchange(state->pending_receive[end], nullptr); + // A blocking receiver may be waiting behind the parked slot; the + // freed slot is part of its wake condition. + state->changed.notify_all(); + } + expired(Error::Timeout("websocket pair: no message within the receive deadline")); + }).detach(); + } + // Both receive overloads: `timeout` engaged bounds the wait, disengaged // is the unbounded blocking call. Outcome> ReceiveWithin( @@ -247,6 +309,9 @@ class PairEnd final : public WebSocket { WebSocket::ReceiveCallback& parked = state_->pending_receive[1 - send_index_]; if (!parked) return nullptr; delivered = message; + // The emptied slot releases a deadline watchdog sleeping on it (#130); + // harmless for everyone else. + state_->changed.notify_all(); return std::exchange(parked, nullptr); } diff --git a/runtime/testing/include/smithy/testing/websocket_contract_test.h b/runtime/testing/include/smithy/testing/websocket_contract_test.h index 04ba78c3..210f8596 100644 --- a/runtime/testing/include/smithy/testing/websocket_contract_test.h +++ b/runtime/testing/include/smithy/testing/websocket_contract_test.h @@ -335,6 +335,103 @@ TYPED_TEST_P(WebSocketContractTest, ASecondSendClassOperationRefusesWhileOneIsPa ender.join(); } +// The receive deadline's async half (#130): on a quiet wire the parked +// receive completes with Error::Timeout — and, exactly like the blocking +// overload, the timeout is not terminal: the slot is released (a second +// timed receive parks and times out the same way) and the session still +// answers a terminal transition afterwards. +TYPED_TEST_P(WebSocketContractTest, ATimedReceiveOnAQuietWireTimesOutAndReleasesItsSlot) { + TypeParam driver; + std::shared_ptr socket = driver.Socket(); + + for (int round = 0; round < 2; ++round) { + ContractMailbox>> timed; + socket->ReceiveAsync( + std::chrono::milliseconds(50), + [&timed](Outcome> got) { timed.Post(std::move(got)); }); + auto expired = timed.Wait(); + ASSERT_FALSE(expired.ok()) << "round " << round; + EXPECT_EQ(expired.error().code(), "TimeoutError") << "round " << round; + } + + // The session survived both deadlines: a parked receive still completes + // through the terminal transition, not as a leak. + ContractMailbox>> parked; + socket->ReceiveAsync( + [&parked](Outcome> got) { parked.Post(std::move(got)); }); + std::thread ender([&] { driver.EndSessionFromPeer(); }); + ender.join(); + auto terminal = parked.Wait(); + if (!terminal.ok()) { + EXPECT_NE(terminal.error().code(), "TimeoutError"); + } +} + +// A non-positive deadline polls: nothing already in hand completes as a +// timeout without waiting (the blocking overload's documented poll shape). +TYPED_TEST_P(WebSocketContractTest, ANonPositiveTimedReceivePolls) { + TypeParam driver; + std::shared_ptr socket = driver.Socket(); + ContractMailbox>> polled; + + socket->ReceiveAsync( + std::chrono::milliseconds(0), + [&polled](Outcome> got) { polled.Post(std::move(got)); }); + auto expired = polled.Wait(); + ASSERT_FALSE(expired.ok()); + EXPECT_EQ(expired.error().code(), "TimeoutError"); + + std::thread ender([&] { driver.EndSessionFromPeer(); }); + ender.join(); +} + +// A terminal transition that arrives before the deadline owns the parked +// timed receive: it completes with the session's terminal outcome, never +// with the timeout — and the deadline that later fires into the emptied +// slot must be a no-op, not a second completion (the mailbox asserts +// exactly one arrival). +TYPED_TEST_P(WebSocketContractTest, ATerminalTransitionBeatsAParkedTimedReceivesDeadline) { + TypeParam driver; + std::shared_ptr socket = driver.Socket(); + ContractMailbox>> parked; + + socket->ReceiveAsync( + std::chrono::seconds(30), + [&parked](Outcome> got) { parked.Post(std::move(got)); }); + ASSERT_TRUE(parked.Empty()) << "the timed receive should park on a quiet wire"; + + std::thread ender([&] { driver.EndSessionFromPeer(); }); + ender.join(); + auto terminal = parked.Wait(); + if (!terminal.ok()) { + EXPECT_NE(terminal.error().code(), "TimeoutError"); + } +} + +// The one-outstanding rule spans the deadline overload: a second receive +// refuses while a timed one is parked, and the parked one still completes. +TYPED_TEST_P(WebSocketContractTest, ASecondReceiveRefusesWhileATimedOneIsParked) { + TypeParam driver; + std::shared_ptr socket = driver.Socket(); + ContractMailbox>> first; + ContractMailbox>> second; + + socket->ReceiveAsync( + std::chrono::seconds(30), + [&first](Outcome> got) { first.Post(std::move(got)); }); + ASSERT_TRUE(first.Empty()) << "the timed receive should park on a quiet wire"; + + socket->ReceiveAsync( + [&second](Outcome> got) { second.Post(std::move(got)); }); + auto refusal = second.Wait(); + ASSERT_FALSE(refusal.ok()); + EXPECT_EQ(refusal.error().kind(), ErrorKind::kValidation); + + std::thread ender([&] { driver.EndSessionFromPeer(); }); + ender.join(); + (void)first.Wait(); +} + // One outstanding receive-class operation per session, same shape. TYPED_TEST_P(WebSocketContractTest, ASecondReceiveClassOperationRefusesWhileOneIsParked) { TypeParam driver; @@ -363,6 +460,10 @@ REGISTER_TYPED_TEST_SUITE_P(WebSocketContractTest, ATerminalTransitionCompletesALoneParkedCoroutineSend, ATerminalTransitionCompletesALoneParkedReceive, ASecondSendClassOperationRefusesWhileOneIsParked, + ATimedReceiveOnAQuietWireTimesOutAndReleasesItsSlot, + ANonPositiveTimedReceivePolls, + ATerminalTransitionBeatsAParkedTimedReceivesDeadline, + ASecondReceiveRefusesWhileATimedOneIsParked, ASecondReceiveClassOperationRefusesWhileOneIsParked); } // namespace smithy::testing diff --git a/runtime/tests/eventstream/jsonrpc_stream_socket_test.cc b/runtime/tests/eventstream/jsonrpc_stream_socket_test.cc index 5cb86152..35fcd743 100644 --- a/runtime/tests/eventstream/jsonrpc_stream_socket_test.cc +++ b/runtime/tests/eventstream/jsonrpc_stream_socket_test.cc @@ -13,6 +13,7 @@ #include #include +#include #include #include #include @@ -320,6 +321,39 @@ TEST(JsonRpcStreamSocketTest, ADeadlineDelegatesAndATimeoutIsNotAViolation) { EXPECT_EQ(**received, event); } +TEST(JsonRpcStreamSocketTest, TheAsyncDeadlineDelegatesAndATimeoutIsNotAViolation) { + // The #130 twin of the blocking case above: the timeout rides through + // the classifier untouched, and the envelope that arrives after it is + // policed for the next receive exactly as if no deadline had passed. + auto [left, right] = http::InMemoryWebSocketPair::Create(); + auto client = Wrap(left, JsonRpcStreamSocket::Role::kClient); + auto server = Wrap(right, JsonRpcStreamSocket::Role::kServer); + + std::promise>> timed; + server->ReceiveAsync(std::chrono::milliseconds(50), [&timed](Outcome> m) { + timed.set_value(std::move(m)); + }); + auto timed_future = timed.get_future(); + ASSERT_EQ(timed_future.wait_for(std::chrono::seconds(5)), std::future_status::ready); + auto nothing = timed_future.get(); + ASSERT_FALSE(nothing.ok()); + EXPECT_EQ(nothing.error().code(), "TimeoutError"); + + const Message event = + MakeEventMessage("message", "application/json", Blob::FromString(R"({"text":"late"})")); + ASSERT_TRUE(client->Send(event).ok()); + std::promise>> next; + server->ReceiveAsync(std::chrono::seconds(30), [&next](Outcome> m) { + next.set_value(std::move(m)); + }); + auto next_future = next.get_future(); + ASSERT_EQ(next_future.wait_for(std::chrono::seconds(5)), std::future_status::ready); + auto received = next_future.get(); + ASSERT_TRUE(received.ok()) << received.error().message(); + ASSERT_TRUE(received->has_value()); + EXPECT_EQ(**received, event); +} + TEST(JsonRpcStreamSocketTest, CloseAndThePeersCleanCloseDelegate) { auto [left, right] = http::InMemoryWebSocketPair::Create(); auto client = Wrap(left, JsonRpcStreamSocket::Role::kClient); diff --git a/runtime/tests/http/beast_websocket_test.cc b/runtime/tests/http/beast_websocket_test.cc index 31a70df0..19ffa5b1 100644 --- a/runtime/tests/http/beast_websocket_test.cc +++ b/runtime/tests/http/beast_websocket_test.cc @@ -172,6 +172,48 @@ TEST(BeastWebSocketTest, ATimedOutReceiveReleasesTheOneOutstandingSlotOnTheWire) server.Stop(); } +TEST(BeastWebSocketTest, AnAsyncReceiveDeadlineExpiresAndDeliveryStillBeatsALaterOne) { + // The #130 async twin over the real wire: the deadline timer lives on + // the connection's executor, so the timer/delivery race the pair pins in + // memory gets its own proof against Beast's io machinery. + BeastServerTransport server(EchoOptions()); + ASSERT_TRUE(server.Start(NotFoundHandler()).ok()); + + auto dialed = BeastWebSocketClient::Dial({.host = "127.0.0.1", .port = server.port()}); + ASSERT_TRUE(dialed.ok()) << dialed.error().message(); + const std::shared_ptr& socket = *dialed; + + // Quiet wire: the parked timed receive completes with the timeout, on + // the executor, and the session is untouched. + std::promise>> timed; + socket->ReceiveAsync(std::chrono::milliseconds(75), [&timed](Outcome> m) { + timed.set_value(std::move(m)); + }); + auto timed_future = timed.get_future(); + ASSERT_EQ(timed_future.wait_for(std::chrono::seconds(5)), std::future_status::ready); + auto expired = timed_future.get(); + ASSERT_FALSE(expired.ok()); + EXPECT_EQ(expired.error().code(), "TimeoutError"); + + // Same session, fresh deadline: the echo arrives well inside it, the + // delivery wins the race, and the stale first timer never fires into + // this park (set_value would abort on a second completion). + std::promise>> delivered; + socket->ReceiveAsync(std::chrono::seconds(30), [&delivered](Outcome> m) { + delivered.set_value(std::move(m)); + }); + ASSERT_TRUE(socket->Send(Text("chat", "beats the deadline")).ok()); + auto delivered_future = delivered.get_future(); + ASSERT_EQ(delivered_future.wait_for(std::chrono::seconds(5)), std::future_status::ready); + auto echo = delivered_future.get(); + ASSERT_TRUE(echo.ok()) << echo.error().message(); + ASSERT_TRUE(echo->has_value()); + EXPECT_EQ((**echo).payload.ToString(), "echo:beats the deadline"); + + socket->Close(); + server.Stop(); +} + TEST(BeastWebSocketTest, AThrowingAsyncReceiveCallbackDoesNotKillTheSession) { // ADR-0003/#109: an application completion callback that throws runs on the // connection's io thread. It must be contained there — never unwind From 054c0147223f9ec6a42af5b304f1575cef729f66 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 25 Aug 2026 21:38:15 +0000 Subject: [PATCH 02/11] co_await stream.Receive(timeout): the coroutine deadline (#130) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The async twin of EventStream's timed Receive, for the Detached loops that are its audience: a bounded await resolves with Error::Timeout ("TimeoutError") on a session that keeps serving — the frame no longer parks forever holding its session on a peer that never sends, and an event delivered after the deadline waits for the next await. ReceiveMessage grows the same overload for the raw pre-stream await (the jsonRpc2 opening-envelope read). Only decoder rejections close the session; the timeout passes through the awaitable untouched. Pinned over the pair at both layers: delivery-beats-deadline, timeout- then-late-message, the stale-deadline no-op, the zero-timeout poll, a serve loop that ticks through timeouts and then echoes, and the raw awaitable's deadline. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_014C7WdBD99mUFWGxMvGSjSU --- .../smithy/eventstream/async_event_stream.h | 62 +++++++- .../eventstream/async_event_stream_test.cc | 149 ++++++++++++++++++ 2 files changed, 204 insertions(+), 7 deletions(-) diff --git a/runtime/include/smithy/eventstream/async_event_stream.h b/runtime/include/smithy/eventstream/async_event_stream.h index a00bb0b4..c5698540 100644 --- a/runtime/include/smithy/eventstream/async_event_stream.h +++ b/runtime/include/smithy/eventstream/async_event_stream.h @@ -3,6 +3,7 @@ #include #include +#include #include #include #include @@ -178,22 +179,29 @@ inline SendMessageAwaitable SendMessage(std::shared_ptr socket, // One-outstanding-per-class passes through; single-shot like its twin. class [[nodiscard]] ReceiveMessageAwaitable { public: - explicit ReceiveMessageAwaitable(std::shared_ptr socket) - : socket_(std::move(socket)) {} + explicit ReceiveMessageAwaitable(std::shared_ptr socket, + std::optional timeout = std::nullopt) + : socket_(std::move(socket)), timeout_(timeout) {} // NOLINTNEXTLINE(readability-convert-member-functions-to-static) bool await_ready() const noexcept { return false; } bool await_suspend(std::coroutine_handle<> coroutine) { // The second-arrival-resumes race, as in AsyncEventStream's awaitables. - socket_->ReceiveAsync([this, coroutine](Outcome> message) { + auto completion = [this, coroutine](Outcome> message) { received_ = std::move(message); if (arrived_.exchange(true)) coroutine.resume(); - }); + }; + if (timeout_.has_value()) { + socket_->ReceiveAsync(*timeout_, std::move(completion)); + } else { + socket_->ReceiveAsync(std::move(completion)); + } return !arrived_.exchange(true); // suspend iff the callback has not run } Outcome> await_resume() noexcept { return std::move(received_); } private: std::shared_ptr socket_; + std::optional timeout_; std::atomic arrived_{false}; // Outcome has no default constructor; the placeholder is overwritten // before any resume. NOLINT(readability-redundant-member-init) @@ -204,6 +212,17 @@ inline ReceiveMessageAwaitable ReceiveMessage(std::shared_ptr s return ReceiveMessageAwaitable(std::move(socket)); } +// The same await under a deadline (#130): a coroutine that must not park +// forever on a peer that may never send. Resolves with the socket's timed +// receive outcomes — the message, the clean close, the terminal error, or +// Error::Timeout ("TimeoutError"), which is not terminal: the session and +// its receive slot are exactly as a completed receive leaves them, so the +// loop can await again, send, or close on its own schedule. +inline ReceiveMessageAwaitable ReceiveMessage(std::shared_ptr socket, + std::chrono::milliseconds timeout) { + return ReceiveMessageAwaitable(std::move(socket), timeout); +} + // The typed session's coroutine adapter (ADR-0019): EventStream's contract // over the completion-driven socket primitives, with `co_await` where the // blocking facade parks a thread. Owns its session (shared_ptr — the async @@ -261,7 +280,9 @@ class AsyncEventStream { // is closed and the error is the awaited result. class [[nodiscard]] ReceiveAwaitable { public: - explicit ReceiveAwaitable(AsyncEventStream* stream) : stream_(stream) {} + explicit ReceiveAwaitable(AsyncEventStream* stream, + std::optional timeout = std::nullopt) + : stream_(stream), timeout_(timeout) {} bool await_ready() const noexcept { return false; } bool await_suspend(std::coroutine_handle<> coroutine) { // The completion may fire before this returns (immediate refusals, @@ -272,13 +293,22 @@ class AsyncEventStream { // and only a truly asynchronous completion resumes from the callback. // The flag can live in the awaitable: the frame dies only after a // resume, and every resume is sequenced after both exchanges. - stream_->socket_->ReceiveAsync([this, coroutine](Outcome> message) { + auto completion = [this, coroutine](Outcome> message) { raw_ = std::move(message); if (arrived_.exchange(true)) coroutine.resume(); - }); + }; + if (timeout_.has_value()) { + stream_->socket_->ReceiveAsync(*timeout_, std::move(completion)); + } else { + stream_->socket_->ReceiveAsync(std::move(completion)); + } return !arrived_.exchange(true); // suspend iff the callback has not run } Outcome> await_resume() { + // A timeout (code "TimeoutError") arrives here as raw_'s error and + // passes through untouched: it is not terminal, nothing is closed, + // and the next co_await picks the stream up where it left off. Only + // a decoder rejection below ends the session. if (!raw_.ok()) return std::move(raw_).error(); std::optional& message = *raw_; if (!message.has_value()) return std::optional(); @@ -292,6 +322,7 @@ class AsyncEventStream { private: AsyncEventStream* stream_; + std::optional timeout_; std::atomic arrived_{false}; // Outcome has no default constructor; the placeholder is overwritten // before any resume. NOLINT(readability-redundant-member-init) @@ -300,6 +331,23 @@ class AsyncEventStream { ReceiveAwaitable Receive() { return ReceiveAwaitable(this); } + // Receive under a deadline (#130) — the async twin of EventStream's + // timed Receive, for a loop that must not park forever on an event the + // peer may never send: + // + // auto event = co_await stream.Receive(std::chrono::seconds(2)); + // if (!event.ok() && event.error().code() == "TimeoutError") { ... } + // + // Same outcomes plus Error::Timeout, and the same rule as the blocking + // overload: a timeout is not terminal — the session stays usable, the + // receive slot is released, and an event the wire delivers after the + // deadline waits in the session for the next await. A non-positive + // timeout polls. The deadline bounds THIS await only; the session's own + // idle timeout still governs a quiet wire. + ReceiveAwaitable Receive(std::chrono::milliseconds timeout) { + return ReceiveAwaitable(this, timeout); + } + // Awaits one event onto the wire. Encoder failures surface without // suspending and leave the session untouched; wire failures are the // socket's (Error::Transport once the session ended). Uncallable when Tx diff --git a/runtime/tests/eventstream/async_event_stream_test.cc b/runtime/tests/eventstream/async_event_stream_test.cc index 365fb830..6a7f86c8 100644 --- a/runtime/tests/eventstream/async_event_stream_test.cc +++ b/runtime/tests/eventstream/async_event_stream_test.cc @@ -133,6 +133,81 @@ TEST(PairAsyncTest, AReadyMessageCompletesImmediately) { EXPECT_EQ((*outcome)->payload.ToString(), "1"); } +// The #130 deadline over the pair: what the transport-neutral contract +// suite cannot pin because its driver has no "peer sends" hook. + +TEST(PairAsyncTest, ATimedReceiveDeliversAMessageThatBeatsTheDeadline) { + auto [a, b] = http::InMemoryWebSocketPair::Create(); + Mailbox>> received; + a->ReceiveAsync(std::chrono::seconds(30), [&](Outcome> message) { + received.Post(std::move(message)); + }); + EXPECT_TRUE(received.Empty()); // parked: nothing sent yet + + ASSERT_TRUE(b->Send(RawPing(7)).ok()); + auto outcome = received.Wait(); + ASSERT_TRUE(outcome.ok() && outcome->has_value()); + EXPECT_EQ((*outcome)->payload.ToString(), "7"); +} + +TEST(PairAsyncTest, ATimedReceiveTimesOutAndALaterMessageWaitsForTheNextReceive) { + auto [a, b] = http::InMemoryWebSocketPair::Create(); + Mailbox>> timed; + a->ReceiveAsync(std::chrono::milliseconds(50), + [&](Outcome> message) { timed.Post(std::move(message)); }); + auto expired = timed.Wait(); + ASSERT_FALSE(expired.ok()); + EXPECT_EQ(expired.error().code(), "TimeoutError"); + + // Nothing was lost to the deadline: the message the peer sends after it + // waits in the session for whoever receives next. + ASSERT_TRUE(b->Send(RawPing(9)).ok()); + Mailbox>> next; + a->ReceiveAsync([&](Outcome> message) { next.Post(std::move(message)); }); + auto outcome = next.Wait(); + ASSERT_TRUE(outcome.ok() && outcome->has_value()); + EXPECT_EQ((*outcome)->payload.ToString(), "9"); +} + +TEST(PairAsyncTest, AStaleDeadlineNeverFiresALaterReceive) { + auto [a, b] = http::InMemoryWebSocketPair::Create(); + // Park with a short deadline and complete it by delivery well inside it. + Mailbox>> first; + a->ReceiveAsync(std::chrono::milliseconds(200), + [&](Outcome> message) { first.Post(std::move(message)); }); + ASSERT_TRUE(b->Send(RawPing(1)).ok()); + ASSERT_TRUE(first.Wait().ok()); + + // A fresh (untimed) park now occupies the slot the expired deadline was + // armed for. The generation guard makes the stale watchdog a no-op: well + // past the original deadline, the new park is still waiting. + Mailbox>> second; + a->ReceiveAsync( + [&](Outcome> message) { second.Post(std::move(message)); }); + std::this_thread::sleep_for(std::chrono::milliseconds(400)); + EXPECT_TRUE(second.Empty()) << "a stale deadline timed out a receive it never bounded"; + + ASSERT_TRUE(b->Send(RawPing(2)).ok()); + auto outcome = second.Wait(); + ASSERT_TRUE(outcome.ok() && outcome->has_value()); + EXPECT_EQ((*outcome)->payload.ToString(), "2"); +} + +TEST(PairAsyncTest, AZeroTimeoutReceivePollsWhatIsAlreadyInHand) { + auto [a, b] = http::InMemoryWebSocketPair::Create(); + ASSERT_TRUE(b->Send(RawPing(3)).ok()); + Mailbox>> polled; + a->ReceiveAsync(std::chrono::milliseconds(0), [&](Outcome> message) { + polled.Post(std::move(message)); + }); + // Inline: the pair has no executor, so the poll completed before the call + // returned — with the queued message, not a timeout. + ASSERT_FALSE(polled.Empty()); + auto outcome = polled.Wait(); + ASSERT_TRUE(outcome.ok() && outcome->has_value()); + EXPECT_EQ((*outcome)->payload.ToString(), "3"); +} + TEST(PairAsyncTest, ASecondOutstandingReceiveIsRefused) { auto [a, b] = http::InMemoryWebSocketPair::Create(); Mailbox>> first; @@ -358,6 +433,80 @@ TEST(AsyncEventStreamTest, ADetachedLoopEchoesAndEndsOnTheCleanClose) { EXPECT_TRUE(done.load()); } +// The #130 watchdog shape: a loop that bounds each await, treats the +// timeout as "nothing yet" rather than an end, and keeps serving. The +// deadline is not terminal — the same stream receives the ping that +// arrives after a timeout and answers it. +TEST(AsyncEventStreamTest, AwaitedReceiveTimesOutAndTheSessionStaysUsable) { + auto [client_socket, server_socket] = http::InMemoryWebSocketPair::Create(); + std::atomic timeouts{0}; + std::atomic done{false}; + + [](std::shared_ptr socket, std::atomic* timeouts, + std::atomic* done) -> Detached { + AsyncServer stream(std::move(socket), EncodePong, DecodePing); + while (true) { + auto ping = co_await stream.Receive(std::chrono::milliseconds(50)); + if (!ping.ok()) { + if (ping.error().code() == "TimeoutError") { + ++*timeouts; // not terminal: the watchdog tick, then wait again + continue; + } + break; + } + if (!ping->has_value()) break; + auto sent = co_await stream.Send(Pong{"pong-" + std::to_string((*ping)->number)}); + if (!sent.ok()) break; + } + *done = true; + }(server_socket, &timeouts, &done); + + // Let at least one deadline expire before the first ping. + for (int i = 0; i < 100 && timeouts.load() == 0; ++i) { + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + EXPECT_GT(timeouts.load(), 0) << "the awaited deadline never fired"; + EXPECT_FALSE(done.load()) << "a timeout must not end the loop"; + + EventStream client(client_socket, EncodePing, DecodePong); + ASSERT_TRUE(client.Send(Ping{42}).ok()); + auto pong = client.Receive(); + ASSERT_TRUE(pong.ok() && pong->has_value()); + EXPECT_EQ((*pong)->text, "pong-42"); + + client.Close(); + for (int i = 0; i < 100 && !done.load(); ++i) { + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + EXPECT_TRUE(done.load()); +} + +// The raw awaitable's deadline (#130): ReceiveMessage with a timeout — the +// ADR-0023 launch-body shape, where a client that never sends its opening +// envelope must not park the serve coroutine forever. +TEST(AsyncEventStreamTest, ReceiveMessageHonorsItsDeadline) { + auto [client_socket, server_socket] = http::InMemoryWebSocketPair::Create(); + Mailbox>> first; + Mailbox>> second; + + [](std::shared_ptr socket, Mailbox>>* first, + Mailbox>>* second) -> Detached { + first->Post(co_await ReceiveMessage(socket, std::chrono::milliseconds(50))); + // The timeout released the slot: the same socket awaits again and gets + // the envelope that arrives late. + second->Post(co_await ReceiveMessage(socket, std::chrono::seconds(30))); + }(server_socket, &first, &second); + + auto expired = first.Wait(); + ASSERT_FALSE(expired.ok()); + EXPECT_EQ(expired.error().code(), "TimeoutError"); + + ASSERT_TRUE(client_socket->Send(RawPing(5)).ok()); + auto arrived = second.Wait(); + ASSERT_TRUE(arrived.ok() && arrived->has_value()); + EXPECT_EQ((*arrived)->payload.ToString(), "5"); +} + TEST(AsyncEventStreamTest, AwaitedSendBackpressuresWithoutAThread) { auto [client_socket, server_socket] = http::InMemoryWebSocketPair::Create(); std::atomic sent_count{0}; From 4d4f25d354d2d3f2cd5d9be6a5685bd88726b8a1 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 25 Aug 2026 21:38:15 +0000 Subject: [PATCH 03/11] Hold the out-of-tree socket and loops to the receive deadline (#130) The consumer's own WebSocket implements the timed ReceiveAsync (the watchdog shape, generation-guarded, holding itself alive via shared_from_this for at most its own deadline) and runs the contract suite's new deadline cases across the module boundary. The acceptance suite adds the stated audience end to end: a hand-written Detached loop over a real Beast socket bounds its await, treats the timeout as a tick, and serves the message that arrives after one. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_014C7WdBD99mUFWGxMvGSjSU --- .../bazel-consumer/async_acceptance_test.cc | 44 ++++++++++++++ .../websocket_contract_consumer_test.cc | 58 ++++++++++++++++++- 2 files changed, 100 insertions(+), 2 deletions(-) diff --git a/examples/bazel-consumer/async_acceptance_test.cc b/examples/bazel-consumer/async_acceptance_test.cc index a0cb08aa..8c7a8e02 100644 --- a/examples/bazel-consumer/async_acceptance_test.cc +++ b/examples/bazel-consumer/async_acceptance_test.cc @@ -101,6 +101,50 @@ Detached Serve(smithy::server::SessionRegistry& registry, std::string i stream.Close(); } +// The #130 watchdog, out of tree and over a real socket: a hand-written +// Detached loop — the deadline's stated audience — bounds its await +// against a peer that says nothing unsolicited, treats the timeout as a +// tick rather than an end, and the same coroutine still serves the +// message that arrives after one. +TEST(AsyncAcceptanceTest, AnAwaitedReceiveDeadlineTicksWithoutEndingTheSession) { + BeastServerTransport::Options options; + options.on_websocket = [](const HttpRequest&, WebSocket& socket) { + while (true) { + auto message = socket.Receive(); + if (!message.ok() || !message->has_value()) return; + if (!socket.Send(Event("echo", "echo:" + (*message)->payload.ToString())).ok()) return; + } + }; + BeastServerTransport server(options); + ASSERT_TRUE(server.Start(NotFound).ok()); + auto dialed = BeastWebSocketClient::Dial({.host = "127.0.0.1", .port = server.port()}); + ASSERT_TRUE(dialed.ok()) << dialed.error().message(); + + std::promise first_timeout_code; + std::promise echoed; + [](std::shared_ptr socket, std::promise* timeout_code, + std::promise* echoed) -> Detached { + AsyncEventStream stream(std::move(socket), Identity, Identity); + // Quiet wire: the bounded await resolves with the timeout, terminally + // for nothing — the session and the loop both continue. + auto nothing = co_await stream.Receive(std::chrono::milliseconds(75)); + timeout_code->set_value(nothing.ok() ? "ok" : nothing.error().code()); + (void)co_await stream.Send(Event("chat", "after the deadline")); + auto echo = co_await stream.Receive(std::chrono::seconds(10)); + echoed->set_value(echo.ok() && echo->has_value() ? (**echo).payload.ToString() : ""); + stream.Close(); + }(*dialed, &first_timeout_code, &echoed); + + auto code = first_timeout_code.get_future(); + ASSERT_EQ(code.wait_for(std::chrono::seconds(5)), std::future_status::ready); + EXPECT_EQ(code.get(), "TimeoutError"); + + auto reply = echoed.get_future(); + ASSERT_EQ(reply.wait_for(std::chrono::seconds(5)), std::future_status::ready); + EXPECT_EQ(reply.get(), "echo:after the deadline"); + server.Stop(); +} + TEST(AsyncAcceptanceTest, ThreeSessionsShareOneHandlerThreadAndAFanOutRegistry) { // Declared before the transport on purpose: sessions reference the // registry from their coroutines, so it must outlive them. diff --git a/examples/bazel-consumer/websocket_contract_consumer_test.cc b/examples/bazel-consumer/websocket_contract_consumer_test.cc index 4587e78a..3910adeb 100644 --- a/examples/bazel-consumer/websocket_contract_consumer_test.cc +++ b/examples/bazel-consumer/websocket_contract_consumer_test.cc @@ -15,10 +15,12 @@ #include #include #include +#include #include #include #include #include +#include #include #include "smithy/eventstream/frame.h" @@ -37,7 +39,7 @@ using smithy::http::WebSocket; // wire but that the terminal transition is expressed with TerminalWaiters, // so this implementation inherits the ordering rule without its author // having to rediscover why the rule exists. -class ConsumerSocket final : public WebSocket { +class ConsumerSocket final : public WebSocket, public std::enable_shared_from_this { public: // Small on purpose: the contract suite wedges the wire by sending, and a // shallow queue gets there in a few messages. @@ -78,7 +80,8 @@ class ConsumerSocket final : public WebSocket { const std::lock_guard lock(mutex_); if (!closed_ && !pending_receive_) { pending_receive_ = std::move(callback); - return; // EndSession completes it + ++receive_park_generation_; // a stale deadline must not fire this park + return; // EndSession completes it } if (!closed_) { immediate = smithy::Error::Validation("consumer socket: a receive is already outstanding"); @@ -87,6 +90,32 @@ class ConsumerSocket final : public WebSocket { callback(std::move(immediate)); } + // The deadline overload (#130): the same park, bounded by a watchdog + // thread racing the terminal transition — the executor-less shape. The + // park generation settles the race exactly once under the lock, so a + // stale deadline can never time out a later receive; the watchdog holds + // the socket alive (shared_from_this) for at most its own deadline. + void ReceiveAsync(std::chrono::milliseconds timeout, ReceiveCallback callback) override { + Outcome> immediate = std::optional(); // the clean end + { + const std::lock_guard lock(mutex_); + if (!closed_ && !pending_receive_ && timeout > std::chrono::milliseconds::zero()) { + pending_receive_ = std::move(callback); + ++receive_park_generation_; + ArmDeadlineLocked(timeout); + return; // EndSession or the deadline completes it + } + if (!closed_ && pending_receive_) { + immediate = smithy::Error::Validation("consumer socket: a receive is already outstanding"); + } else if (!closed_) { + // The non-positive poll — this peer never sends, so nothing is + // ever already in hand. + immediate = smithy::Error::Timeout("consumer socket: no message within the deadline"); + } + } + callback(std::move(immediate)); + } + void SendAsync(const Message& message, SendCallback callback) override { (void)message; Outcome immediate = Unit{}; @@ -126,10 +155,35 @@ class ConsumerSocket final : public WebSocket { } private: + // With the lock held: arms the watchdog for the park that just went in. + // It sleeps on the shared condition variable, so a completed park + // (EndSession, or a fresh park's bumped generation) releases it early; + // at the deadline, a park still bearing its generation is timed out — + // the slot released exactly as a completion releases it. + void ArmDeadlineLocked(std::chrono::milliseconds timeout) { + std::thread([self = shared_from_this(), generation = receive_park_generation_, + deadline = std::chrono::steady_clock::now() + timeout] { + ReceiveCallback expired; + { + std::unique_lock lock(self->mutex_); + self->changed_.wait_until(lock, deadline, [&] { + return !self->pending_receive_ || self->receive_park_generation_ != generation; + }); + if (!self->pending_receive_ || self->receive_park_generation_ != generation) { + return; // the park this deadline bounded already completed + } + expired = std::exchange(self->pending_receive_, nullptr); + self->changed_.notify_all(); + } + expired(smithy::Error::Timeout("consumer socket: no message within the deadline")); + }).detach(); + } + std::mutex mutex_; std::condition_variable changed_; ReceiveCallback pending_receive_; SendCallback pending_send_; + std::uint64_t receive_park_generation_ = 0; std::size_t queued_ = 0; bool closed_ = false; }; From 2cd8dd586a97b87b4fbd047cd656bc11895dff21 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 25 Aug 2026 21:38:15 +0000 Subject: [PATCH 04/11] Pin the #109 numeric bounds out of tree, over the real socket MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The consumer's todo model gains an intEnum priority and a float effortHours, so the narrowing and membership checks run through the exact pipeline consumers ship: generator at build time, regenerated server on the production transport, hostile wire values (2^32+2, an unknown member, 1e300) rejected before the handler with the suite-exact identities, and the generated client serializing valid ones. Optional members — the existing handlers and the model-evolution script are untouched. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_014C7WdBD99mUFWGxMvGSjSU --- examples/bazel-consumer/model/todo.smithy | 12 +++++ .../todo_beast_acceptance_test.cc | 44 +++++++++++++++++++ 2 files changed, 56 insertions(+) diff --git a/examples/bazel-consumer/model/todo.smithy b/examples/bazel-consumer/model/todo.smithy index a10b6bdd..db4512ca 100644 --- a/examples/bazel-consumer/model/todo.smithy +++ b/examples/bazel-consumer/model/todo.smithy @@ -22,6 +22,10 @@ operation AddTask { @required @length(min: 1, max: 256) title: String + + priority: Priority + + effortHours: Float } output := { @@ -55,6 +59,14 @@ operation GetTask { errors: [NoSuchTask] } +/// int32 on the wire: out-of-range values fail the parse and unknown +/// in-range values fail server validation (#109) — the members above give +/// those checks out-of-tree, real-socket coverage. +intEnum Priority { + LOW = 1 + HIGH = 2 +} + @error("client") @httpError(404) structure NoSuchTask { diff --git a/examples/bazel-consumer/todo_beast_acceptance_test.cc b/examples/bazel-consumer/todo_beast_acceptance_test.cc index 4151ee0b..64f20a95 100644 --- a/examples/bazel-consumer/todo_beast_acceptance_test.cc +++ b/examples/bazel-consumer/todo_beast_acceptance_test.cc @@ -107,6 +107,50 @@ TEST_F(TodoBeastAcceptanceTest, RoundTripsAndModeledErrorsWork) { ASSERT_NE(missing.error().detail(), nullptr); } +// The #109 numeric bounds, out of tree and over the real socket: hostile +// wire values for the model's intEnum and float members are rejected by +// the generated server before the handler runs, and the generated client +// serializes valid ones — the same generator output consumers ship. +TEST_F(TodoBeastAcceptanceTest, NumericBoundsAreEnforcedOverTheRealSocket) { + smithy::http::BeastHttpClient raw({.host = "127.0.0.1", .port = transport_->port()}); + const auto post = [&raw](const std::string& body) { + smithy::http::HttpRequest request; + request.method = "POST"; + request.target = "/tasks"; + request.headers.Set("content-type", "application/json"); + request.body = body; + return raw.Send(request); + }; + + // 2^32+2 would alias onto a valid Priority under a truncating cast; the + // parse rejects it instead. + auto aliased = post(R"({"title":"t","priority":4294967298})"); + ASSERT_TRUE(aliased.ok()) << aliased.error().message(); + EXPECT_EQ(aliased->status, 400); + EXPECT_EQ(aliased->headers.Get("x-error-type").value_or(""), "SerializationException"); + + // In range but outside the modeled set: membership validation, with the + // suite-exact message shape. + auto unknown = post(R"({"title":"t","priority":3})"); + ASSERT_TRUE(unknown.ok()) << unknown.error().message(); + EXPECT_EQ(unknown->status, 400); + EXPECT_EQ(unknown->headers.Get("x-error-type").value_or(""), "ValidationException"); + EXPECT_NE(unknown->body.find("Member must satisfy enum value set: [1, 2]"), std::string::npos) + << unknown->body; + + // A finite double beyond float range was UB to cast; it fails the parse. + auto overflow = post(R"({"title":"t","effortHours":1e300})"); + ASSERT_TRUE(overflow.ok()) << overflow.error().message(); + EXPECT_EQ(overflow->status, 400); + EXPECT_EQ(overflow->headers.Get("x-error-type").value_or(""), "SerializationException"); + + // Valid values ride the generated client end to end. + auto added = client_->AddTask(AddTaskInput{ + .title = "well bounded", .priority = acme::todo::Priority::kHigh, .effortHours = 1.5F}); + ASSERT_TRUE(added.ok()) << added.error().message(); + EXPECT_EQ(added->title, "well bounded"); +} + TEST_F(TodoBeastAcceptanceTest, LifecycleStopsAndRestartsAcrossTransportGenerations) { // The rolling-restart pattern from the production guide, composed entirely // from the consumer-visible API: serve, Stop() (bounded — the drain knob From a31f4de584feb9875952bc5e9ec7c499cb2bb906 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 25 Aug 2026 21:38:15 +0000 Subject: [PATCH 05/11] Document the async receive deadline (#130) Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_014C7WdBD99mUFWGxMvGSjSU --- CHANGELOG.md | 16 ++++++++++++++++ docs/production-guide.md | 9 +++++++++ docs/server-guide.md | 6 ++++++ 3 files changed, 31 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 655d1708..a8c561dc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -45,6 +45,22 @@ policy in [docs/versioning.md](docs/versioning.md). ### Added +- **A deadline for the coroutine receive** (#130). + `co_await stream.Receive(timeout)` on `AsyncEventStream` (and + `ReceiveMessage(socket, timeout)` for the raw awaitable) resolves with + `Error::Timeout` (code `TimeoutError`) when nothing arrives in time — the + completion-driven twin of the blocking `Receive(timeout)`, with the same + rule: a timeout is not terminal, the receive slot is released, and a + message the wire delivers after the deadline waits for the next await. A + `Detached` serve loop can now bound every await instead of parking its + frame forever on a peer that never sends. One layer down it is + `WebSocket::ReceiveAsync(timeout, callback)` — a refusing default (the + async family is opt-in), implemented by both transports (a timer on the + connection's executor for Beast; a watchdog for the executor-less + in-memory pair) and passed through by `JsonRpcStreamSocket`, with the + timeout/delivery race settled exactly once by a park generation. The + shared WebSocket contract suite grows four deadline cases every + implementation runs, in-repo and out. - **Servers validate intEnum membership** (#109). String enums already failed request validation outside the modeled value set; intEnum members were accepted silently. They now produce the same suite-exact diff --git a/docs/production-guide.md b/docs/production-guide.md index 2d24a221..775cee5b 100644 --- a/docs/production-guide.md +++ b/docs/production-guide.md @@ -461,6 +461,15 @@ code holding a raw session; a hand-rolled socket (a test fake, an adapter over another WebSocket library) owes its callers a wait that actually ends and an `Error::Timeout` when it does. +The completion-driven half has the same deadline (#130): +`co_await stream.Receive(std::chrono::seconds(2))` on an `AsyncEventStream` +resolves with the same four outcomes and the same rule — a timeout releases +the receive slot and leaves the session usable, and an event the wire +delivers after the deadline waits for the next await. One layer down it is +`WebSocket::ReceiveAsync(timeout, callback)`, a refusing default rather than +a pure virtual (the async family is opt-in), with the shared contract suite +holding every `SupportsAsync()` implementation to it. + Not every `ClientConfig` knob reaches a streaming dial — the upgrade GET is not a unary request: diff --git a/docs/server-guide.md b/docs/server-guide.md index 2f4537dc..8655d174 100644 --- a/docs/server-guide.md +++ b/docs/server-guide.md @@ -140,6 +140,12 @@ while (true) { } ``` +The completion-driven seam has the same watchdog (#130): a `Detached` loop bounds its +await with `co_await stream.Receive(std::chrono::seconds(1))` and gets the same +`TimeoutError` on a session that keeps serving — the loop above works verbatim as a +coroutine. `ReceiveMessage(socket, timeout)` bounds the raw pre-stream await the same +way (the jsonRpc2 opening-envelope read). + Whatever method the operation models, upgrades are always GET — a WebSocket upgrade is a GET on the wire, and the generated routes register accordingly: on the modeled URI for the binding protocols, on the shared `/` endpoint for jsonRpc2 (ADR-0023), whose opening From c7d8a0b95bab86f00d435a4df6e4e6685f139073 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 25 Aug 2026 21:47:45 +0000 Subject: [PATCH 06/11] Post the test mailbox's notify under its lock MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit TSan on the new deadline tests surfaced a latent race in the suite's own Mailbox: notify_all ran after the lock was released, so a poster on a foreign thread — the pair's deadline watchdog, here — raced the waiter's return, and ~Mailbox could destroy the condition variable mid-notify. ContractMailbox already documents and follows the notify-under-the-lock rule; the test Mailbox now matches it. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_014C7WdBD99mUFWGxMvGSjSU --- runtime/tests/eventstream/async_event_stream_test.cc | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/runtime/tests/eventstream/async_event_stream_test.cc b/runtime/tests/eventstream/async_event_stream_test.cc index 6a7f86c8..4c319397 100644 --- a/runtime/tests/eventstream/async_event_stream_test.cc +++ b/runtime/tests/eventstream/async_event_stream_test.cc @@ -73,11 +73,13 @@ template class Mailbox { public: void Post(T value) { - { - const std::lock_guard lock(mutex_); - ASSERT_FALSE(value_.has_value()) << "completion fired twice"; - value_.emplace(std::move(value)); - } + // Notify under the lock, not after it (ContractMailbox's rule): a + // poster on a foreign thread — a deadline watchdog, a peer completion — + // otherwise races the waiter's return and ~Mailbox runs while the + // poster is still inside notify_all. + const std::lock_guard lock(mutex_); + ASSERT_FALSE(value_.has_value()) << "completion fired twice"; + value_.emplace(std::move(value)); ready_.notify_all(); } From c1f1b606278fb94c7dce9db54cf0d22c55900264 Mon Sep 17 00:00:00 2001 From: Andy Aylward Date: Wed, 26 Aug 2026 02:08:39 +0100 Subject: [PATCH 07/11] Harden the deadline arming: spawn failure, lock scope, saturation (review panel) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Survivors of the panel on #198, all verified: (1) a std::thread spawn that throws must not escape ReceiveAsync beside a still-armed park — from a coroutine that is a use-after-free when the completion later fires into the freed frame; the pair and the consumer reference implementation now take the park back and refuse (exactly-once, nothing thrown), under __cpp_exceptions in the pair since //runtime:http is in the -fno-exceptions gate. (2) Watchdogs spawn outside the session lock — pthread_create is too heavy to hold both ends' traffic behind. (3) now + timeout saturates at a year, so milliseconds::max() as "practically forever" waits instead of overflowing into an instant spurious timeout. (4) The poll doc no longer promises inline completion for a ready message (Beast posts ready results by design), and the stale-deadline test gets a full second of delivery margin against sanitizer-loaded runners. The Beast timer's saturation lands in the companion commit. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_014C7WdBD99mUFWGxMvGSjSU --- .../websocket_contract_consumer_test.cc | 54 +++++++++++----- runtime/include/smithy/http/websocket.h | 5 +- runtime/src/http/websocket_pair.cc | 63 ++++++++++++++----- 3 files changed, 89 insertions(+), 33 deletions(-) diff --git a/examples/bazel-consumer/websocket_contract_consumer_test.cc b/examples/bazel-consumer/websocket_contract_consumer_test.cc index 3910adeb..17c73a5f 100644 --- a/examples/bazel-consumer/websocket_contract_consumer_test.cc +++ b/examples/bazel-consumer/websocket_contract_consumer_test.cc @@ -12,6 +12,7 @@ #include +#include #include #include #include @@ -94,18 +95,18 @@ class ConsumerSocket final : public WebSocket, public std::enable_shared_from_th // thread racing the terminal transition — the executor-less shape. The // park generation settles the race exactly once under the lock, so a // stale deadline can never time out a later receive; the watchdog holds - // the socket alive (shared_from_this) for at most its own deadline. + // the socket alive (shared_from_this) for at most its own deadline, and + // it is spawned outside the lock so a park never stalls the session + // behind pthread_create. void ReceiveAsync(std::chrono::milliseconds timeout, ReceiveCallback callback) override { Outcome> immediate = std::optional(); // the clean end + std::uint64_t parked_generation = 0; // 0 = not parked (the counter starts at 1) { const std::lock_guard lock(mutex_); if (!closed_ && !pending_receive_ && timeout > std::chrono::milliseconds::zero()) { pending_receive_ = std::move(callback); - ++receive_park_generation_; - ArmDeadlineLocked(timeout); - return; // EndSession or the deadline completes it - } - if (!closed_ && pending_receive_) { + parked_generation = ++receive_park_generation_; + } else if (!closed_ && pending_receive_) { immediate = smithy::Error::Validation("consumer socket: a receive is already outstanding"); } else if (!closed_) { // The non-positive poll — this peer never sends, so nothing is @@ -113,6 +114,10 @@ class ConsumerSocket final : public WebSocket, public std::enable_shared_from_th immediate = smithy::Error::Timeout("consumer socket: no message within the deadline"); } } + if (parked_generation != 0) { + ArmDeadline(parked_generation, timeout); // EndSession or the deadline completes it + return; + } callback(std::move(immediate)); } @@ -155,14 +160,19 @@ class ConsumerSocket final : public WebSocket, public std::enable_shared_from_th } private: - // With the lock held: arms the watchdog for the park that just went in. - // It sleeps on the shared condition variable, so a completed park - // (EndSession, or a fresh park's bumped generation) releases it early; - // at the deadline, a park still bearing its generation is timed out — - // the slot released exactly as a completion releases it. - void ArmDeadlineLocked(std::chrono::milliseconds timeout) { - std::thread([self = shared_from_this(), generation = receive_park_generation_, - deadline = std::chrono::steady_clock::now() + timeout] { + // Arms the watchdog for the park `generation`. It sleeps on the shared + // condition variable, so a completed park (EndSession, or a fresh park's + // bumped generation) releases it early; at the deadline, a park still + // bearing its generation is timed out — the slot released exactly as a + // completion releases it. A spawn that fails must not leave the park + // unbounded: it is taken back (unless the session ended it first) and + // refused, keeping the callback exactly-once with nothing thrown at the + // caller. The saturation guards milliseconds::max()-style "practically + // forever" deadlines from overflowing into the past. + void ArmDeadline(std::uint64_t generation, std::chrono::milliseconds timeout) { + constexpr std::chrono::milliseconds kMaxWait = std::chrono::hours(24 * 365); + const auto deadline = std::chrono::steady_clock::now() + std::min(timeout, kMaxWait); + const auto watchdog = [self = shared_from_this(), generation, deadline] { ReceiveCallback expired; { std::unique_lock lock(self->mutex_); @@ -176,7 +186,21 @@ class ConsumerSocket final : public WebSocket, public std::enable_shared_from_th self->changed_.notify_all(); } expired(smithy::Error::Timeout("consumer socket: no message within the deadline")); - }).detach(); + }; + ReceiveCallback refused; + try { + std::thread(watchdog).detach(); + return; + } catch (...) { + const std::lock_guard lock(mutex_); + if (pending_receive_ && receive_park_generation_ == generation) { + refused = std::exchange(pending_receive_, nullptr); + changed_.notify_all(); + } + } + if (refused) { + refused(smithy::Error::Transport("consumer socket: cannot arm the receive deadline")); + } } std::mutex mutex_; diff --git a/runtime/include/smithy/http/websocket.h b/runtime/include/smithy/http/websocket.h index dd586425..5ac4a19b 100644 --- a/runtime/include/smithy/http/websocket.h +++ b/runtime/include/smithy/http/websocket.h @@ -207,8 +207,9 @@ class WebSocket { // slot is released exactly as a completed receive releases it — the next // receive (either API) may park again, and a message the wire delivers // after the deadline waits in the session for it. A non-positive timeout - // polls: it completes inline with what is already in hand, or with the - // timeout. The one-outstanding-receive rule is unchanged. + // polls: it completes without waiting — with what is already in hand + // (possibly on the completion context, like any ready result) or inline + // with the timeout. The one-outstanding-receive rule is unchanged. // // The timeout must race the completion and settle exactly once — the // implementation owns that race the same way it owns the parked slot diff --git a/runtime/src/http/websocket_pair.cc b/runtime/src/http/websocket_pair.cc index 838c6a03..6f9b0b31 100644 --- a/runtime/src/http/websocket_pair.cc +++ b/runtime/src/http/websocket_pair.cc @@ -1,5 +1,6 @@ #include "smithy/http/websocket_pair.h" +#include #include #include #include @@ -197,6 +198,8 @@ class PairEnd final : public WebSocket { WebSocket::ReceiveCallback callback) { WebSocket::SendCallback absorbed; Outcome> immediate = std::optional(); + // 0 = not parked (the generation counter starts at 1 on the first park). + std::uint64_t parked_generation = 0; { const std::lock_guard lock(state_->mutex); if (state_->pending_receive[send_index_] || state_->blocked_receivers[send_index_] > 0) { @@ -218,27 +221,33 @@ class PairEnd final : public WebSocket { immediate = Error::Timeout("websocket pair: no message within the receive deadline"); } else { state_->pending_receive[send_index_] = std::move(callback); - ++state_->receive_park_generation[send_index_]; - if (timeout.has_value()) { - ArmReceiveDeadlineLocked(*timeout); - } - return; // a send, the deadline, or the close completes it + parked_generation = ++state_->receive_park_generation[send_index_]; } } + if (parked_generation != 0) { + // A send, the deadline, or the close completes the park. + if (timeout.has_value()) ArmReceiveDeadline(parked_generation, *timeout); + return; + } if (absorbed) absorbed(Unit{}); callback(std::move(immediate)); } - // With the lock held: spawns the watchdog for the park that just went - // in. It sleeps on the shared condition variable, so a completed park - // (delivery, close, or a fresh park's bumped generation) releases it - // early; at the deadline, a park still bearing its generation is timed - // out — the slot is released exactly as a delivery releases it, and the - // session is untouched. - void ArmReceiveDeadlineLocked(std::chrono::milliseconds timeout) { - std::thread([state = state_, end = send_index_, - generation = state_->receive_park_generation[send_index_], - deadline = std::chrono::steady_clock::now() + timeout] { + // Spawns the watchdog for the park `generation` — outside the lock, so + // both ends' traffic never stalls behind pthread_create. It sleeps on + // the shared condition variable, so a completed park (delivery, close, + // or a fresh park's bumped generation) releases it early; at the + // deadline, a park still bearing its generation is timed out — the slot + // released exactly as a delivery releases it, the session untouched. A + // spawn that fails must not leave the park unbounded: the park is taken + // back (unless delivery already beat us to it) and refused, keeping the + // callback exactly-once with no exception escaping. + void ArmReceiveDeadline(std::uint64_t generation, std::chrono::milliseconds timeout) { + // Saturate far-future deadlines (milliseconds::max() as "practically + // forever") instead of overflowing now + timeout into the past. + constexpr std::chrono::milliseconds kMaxWait = std::chrono::hours(24 * 365); + const auto deadline = std::chrono::steady_clock::now() + std::min(timeout, kMaxWait); + const auto watchdog = [state = state_, end = send_index_, generation, deadline] { WebSocket::ReceiveCallback expired; { std::unique_lock lock(state->mutex); @@ -254,7 +263,29 @@ class PairEnd final : public WebSocket { state->changed.notify_all(); } expired(Error::Timeout("websocket pair: no message within the receive deadline")); - }).detach(); + }; + // Under -fno-exceptions a failed thread spawn terminates (nothing can + // throw), which is the fail-fast posture; with exceptions on, contain + // it here so the caller never sees a throw beside a still-armed park. +#if defined(__cpp_exceptions) + WebSocket::ReceiveCallback refused; + try { + std::thread(watchdog).detach(); + return; + } catch (...) { + const std::lock_guard lock(state_->mutex); + if (state_->pending_receive[send_index_] && + state_->receive_park_generation[send_index_] == generation) { + refused = std::exchange(state_->pending_receive[send_index_], nullptr); + state_->changed.notify_all(); + } + } + if (refused) { + refused(Error::Transport("websocket pair: cannot arm the receive deadline")); + } +#else + std::thread(watchdog).detach(); +#endif } // Both receive overloads: `timeout` engaged bounds the wait, disengaged From b18c36d6972d4b459b0eb7e88b0876c8ee3673fc Mon Sep 17 00:00:00 2001 From: Andy Aylward Date: Wed, 26 Aug 2026 02:11:39 +0100 Subject: [PATCH 08/11] Give the stale-deadline test a full second of delivery margin (review panel) The companion test half of the hardening commit: the pair test's first-round deadline grows to a second with a 1300ms probe, so a sanitizer-loaded runner cannot let the deadline win the delivery race the test is not about; and the test Mailbox posts its notify under the lock (ContractMailbox's rule) so a watchdog-thread poster cannot race ~Mailbox. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_014C7WdBD99mUFWGxMvGSjSU --- runtime/tests/eventstream/async_event_stream_test.cc | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/runtime/tests/eventstream/async_event_stream_test.cc b/runtime/tests/eventstream/async_event_stream_test.cc index 4c319397..fb2fb99d 100644 --- a/runtime/tests/eventstream/async_event_stream_test.cc +++ b/runtime/tests/eventstream/async_event_stream_test.cc @@ -173,9 +173,12 @@ TEST(PairAsyncTest, ATimedReceiveTimesOutAndALaterMessageWaitsForTheNextReceive) TEST(PairAsyncTest, AStaleDeadlineNeverFiresALaterReceive) { auto [a, b] = http::InMemoryWebSocketPair::Create(); - // Park with a short deadline and complete it by delivery well inside it. + // Park with a bounded deadline and complete it by delivery well inside + // it — a full second of margin, so a scheduler stall (TSan's slowdown on + // a loaded runner) cannot let the deadline win a race this test is not + // about. Mailbox>> first; - a->ReceiveAsync(std::chrono::milliseconds(200), + a->ReceiveAsync(std::chrono::seconds(1), [&](Outcome> message) { first.Post(std::move(message)); }); ASSERT_TRUE(b->Send(RawPing(1)).ok()); ASSERT_TRUE(first.Wait().ok()); @@ -186,7 +189,7 @@ TEST(PairAsyncTest, AStaleDeadlineNeverFiresALaterReceive) { Mailbox>> second; a->ReceiveAsync( [&](Outcome> message) { second.Post(std::move(message)); }); - std::this_thread::sleep_for(std::chrono::milliseconds(400)); + std::this_thread::sleep_for(std::chrono::milliseconds(1300)); EXPECT_TRUE(second.Empty()) << "a stale deadline timed out a receive it never bounded"; ASSERT_TRUE(b->Send(RawPing(2)).ok()); From 5b12a9693c071ba927e2a998f772e2b69cdce641 Mon Sep 17 00:00:00 2001 From: Andy Aylward Date: Wed, 26 Aug 2026 02:20:50 +0100 Subject: [PATCH 09/11] Saturate the Beast receive deadline against far-future timeouts (review panel) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Beast half of the hardening: ArmReceiveDeadlineLocked clamps the timer's expiry at a year, so milliseconds::max() as "practically forever" waits instead of overflowing now + duration into the past and firing a spurious instant timeout — the same saturation the pair and the consumer reference implementation apply. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_014C7WdBD99mUFWGxMvGSjSU --- runtime/src/http/beast_transport.cc | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/runtime/src/http/beast_transport.cc b/runtime/src/http/beast_transport.cc index c28c84b6..87a1a5fc 100644 --- a/runtime/src/http/beast_transport.cc +++ b/runtime/src/http/beast_transport.cc @@ -447,8 +447,12 @@ class WsSession final : public WebSocketSessionBase, // handler captures the session weakly: a deadline must never extend a // session's life. void ArmReceiveDeadlineLocked(std::chrono::milliseconds timeout) { + // Saturate far-future deadlines (milliseconds::max() as "practically + // forever") instead of overflowing the timer's now + duration into the + // past and firing a spurious instant timeout. + constexpr std::chrono::milliseconds kMaxWait = std::chrono::hours(24 * 365); receive_deadline_.emplace(ws_.get_executor()); - receive_deadline_->expires_after(timeout); + receive_deadline_->expires_after(std::min(timeout, kMaxWait)); receive_deadline_->async_wait( [weak = this->weak_from_this(), generation = receive_park_generation_](const boost::system::error_code& ec) { From d9dd78ed2394fa4af77d038355a421b169400ae7 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 26 Aug 2026 02:57:28 +0000 Subject: [PATCH 10/11] Read the park off the completion slot, not a parallel flag MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The hardening's parked_generation sentinel left clang-tidy unable to correlate "we parked" with "the callback was moved into the session", so bugprone-use-after-move and clang-analyzer-cplusplus.Move both flagged the inline completion on a path that cannot happen — CI lint's exit 123. The pairing is real even if the path is not, so settle it structurally rather than silence it: the completion the call still owes lives in a local, and parking hands it over with the std::exchange the terminal paths already use, which empties the local. Whether this call parked is then read off the slot itself (`if (!deliver)`), so the two outcomes are one variable rather than a flag a reader has to keep in sync with a move. The pair and the consumer reference implementation both take the shape; behavior is unchanged, and the watchdog still spawns outside the lock. Verified: clang-tidy clean on websocket_pair.cc and across the CI sweep, clang-format clean, and websocket_pair / beast_websocket / async_event_stream / jsonrpc_stream_socket plus the out-of-tree consumer contract suite all pass. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_014C7WdBD99mUFWGxMvGSjSU --- .../websocket_contract_consumer_test.cc | 9 ++++++--- runtime/src/http/websocket_pair.cc | 16 +++++++++++----- 2 files changed, 17 insertions(+), 8 deletions(-) diff --git a/examples/bazel-consumer/websocket_contract_consumer_test.cc b/examples/bazel-consumer/websocket_contract_consumer_test.cc index 17c73a5f..515182ca 100644 --- a/examples/bazel-consumer/websocket_contract_consumer_test.cc +++ b/examples/bazel-consumer/websocket_contract_consumer_test.cc @@ -100,11 +100,14 @@ class ConsumerSocket final : public WebSocket, public std::enable_shared_from_th // behind pthread_create. void ReceiveAsync(std::chrono::milliseconds timeout, ReceiveCallback callback) override { Outcome> immediate = std::optional(); // the clean end + // The completion this call still owes; parking hands it to the session + // and leaves this empty, so the slot itself says which happened. + ReceiveCallback deliver = std::move(callback); std::uint64_t parked_generation = 0; // 0 = not parked (the counter starts at 1) { const std::lock_guard lock(mutex_); if (!closed_ && !pending_receive_ && timeout > std::chrono::milliseconds::zero()) { - pending_receive_ = std::move(callback); + pending_receive_ = std::exchange(deliver, nullptr); parked_generation = ++receive_park_generation_; } else if (!closed_ && pending_receive_) { immediate = smithy::Error::Validation("consumer socket: a receive is already outstanding"); @@ -114,11 +117,11 @@ class ConsumerSocket final : public WebSocket, public std::enable_shared_from_th immediate = smithy::Error::Timeout("consumer socket: no message within the deadline"); } } - if (parked_generation != 0) { + if (!deliver) { ArmDeadline(parked_generation, timeout); // EndSession or the deadline completes it return; } - callback(std::move(immediate)); + deliver(std::move(immediate)); } void SendAsync(const Message& message, SendCallback callback) override { diff --git a/runtime/src/http/websocket_pair.cc b/runtime/src/http/websocket_pair.cc index 6f9b0b31..ab72de58 100644 --- a/runtime/src/http/websocket_pair.cc +++ b/runtime/src/http/websocket_pair.cc @@ -198,7 +198,12 @@ class PairEnd final : public WebSocket { WebSocket::ReceiveCallback callback) { WebSocket::SendCallback absorbed; Outcome> immediate = std::optional(); - // 0 = not parked (the generation counter starts at 1 on the first park). + // The completion this call still owes, fired once the lock is released. + // Parking hands it to the session instead and leaves this empty (the + // std::exchange emptying is the same one the terminal paths rely on), + // so which of the two happened is read off the slot itself rather than + // off a flag a reader — or the analyzer — has to correlate with it. + WebSocket::ReceiveCallback deliver; std::uint64_t parked_generation = 0; { const std::lock_guard lock(state_->mutex); @@ -206,6 +211,7 @@ class PairEnd final : public WebSocket { callback(Error::Validation("websocket pair: a receive is already outstanding")); return; } + deliver = std::move(callback); std::deque& inbound = state_->queues[1 - send_index_]; if (!inbound.empty()) { eventstream::Message message = std::move(inbound.front()); @@ -220,17 +226,17 @@ class PairEnd final : public WebSocket { // timeout, completed inline like the other immediates. immediate = Error::Timeout("websocket pair: no message within the receive deadline"); } else { - state_->pending_receive[send_index_] = std::move(callback); + state_->pending_receive[send_index_] = std::exchange(deliver, nullptr); parked_generation = ++state_->receive_park_generation[send_index_]; } } - if (parked_generation != 0) { - // A send, the deadline, or the close completes the park. + if (!deliver) { + // Parked: a send, the deadline, or the close completes it. if (timeout.has_value()) ArmReceiveDeadline(parked_generation, *timeout); return; } if (absorbed) absorbed(Unit{}); - callback(std::move(immediate)); + deliver(std::move(immediate)); } // Spawns the watchdog for the park `generation` — outside the lock, so From d4d33997071c1f71ebf93f47266204c2d6e0c111 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 26 Aug 2026 10:27:00 +0000 Subject: [PATCH 11/11] Drop spent receive deadlines, and one park shape in the consumer socket MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Both non-blocking nits from the review on #198, verified against the code: The Beast timer outlived the park it bounded. Deliver and TakeAsyncWaitersLocked emptied pending_receive_ but left receive_deadline_ armed until it expired or the next timed park emplaced over it, so a session that got its message immediately still woke the executor once at the old deadline. The generation guard already made that wakeup a no-op — this just spares it. Resetting is safe from any thread for the same reason the arming emplace is: every touch of the timer is under mutex_, and the armed handler captures the session weakly rather than the timer, so destroying it only cancels a pending wait whose handler then returns on operation_aborted. The consumer reference implementation had two park styles side by side — the untimed receive early-returning under the lock, the timed one using the deliver-slot shape the pair settled on. Since this file is what a third party copies, the untimed path now takes the same shape; behavior is identical. Verified: clang-format clean; beast_websocket / async_event_stream / websocket_pair / jsonrpc_stream_socket / session_registry pass plain, and the first four again under ASan and under TSan (the lifetime and race questions the timer reset actually raises); all 13 consumer module tests pass out of tree. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_014C7WdBD99mUFWGxMvGSjSU --- .../websocket_contract_consumer_test.cc | 13 ++++++++----- runtime/src/http/beast_transport.cc | 11 +++++++++++ 2 files changed, 19 insertions(+), 5 deletions(-) diff --git a/examples/bazel-consumer/websocket_contract_consumer_test.cc b/examples/bazel-consumer/websocket_contract_consumer_test.cc index 515182ca..5a123f12 100644 --- a/examples/bazel-consumer/websocket_contract_consumer_test.cc +++ b/examples/bazel-consumer/websocket_contract_consumer_test.cc @@ -77,18 +77,21 @@ class ConsumerSocket final : public WebSocket, public std::enable_shared_from_th // lock is released — the seam's documented shapes, nothing more. void ReceiveAsync(ReceiveCallback callback) override { Outcome> immediate = std::optional(); // the clean end + // The same deliver-slot shape as the timed overload below: parking + // hands the completion to the session and empties this, so one park + // style covers both receives. + ReceiveCallback deliver = std::move(callback); { const std::lock_guard lock(mutex_); if (!closed_ && !pending_receive_) { - pending_receive_ = std::move(callback); + pending_receive_ = std::exchange(deliver, nullptr); ++receive_park_generation_; // a stale deadline must not fire this park - return; // EndSession completes it - } - if (!closed_) { + } else if (!closed_) { immediate = smithy::Error::Validation("consumer socket: a receive is already outstanding"); } } - callback(std::move(immediate)); + if (!deliver) return; // parked: EndSession completes it + deliver(std::move(immediate)); } // The deadline overload (#130): the same park, bounded by a watchdog diff --git a/runtime/src/http/beast_transport.cc b/runtime/src/http/beast_transport.cc index 87a1a5fc..f2ff9bc3 100644 --- a/runtime/src/http/beast_transport.cc +++ b/runtime/src/http/beast_transport.cc @@ -704,6 +704,8 @@ class WsSession final : public WebSocketSessionBase, const std::lock_guard lock(mutex_); if (pending_receive_) { receive = std::exchange(pending_receive_, nullptr); + // The park is over, so its deadline has nothing left to bound. + receive_deadline_.reset(); handoff.emplace(std::move(message)); } else { received_.push_back(std::move(message)); @@ -726,6 +728,15 @@ class WsSession final : public WebSocketSessionBase, // can never fire twice and the slots' emptiness stays the busy signal. using AsyncWaiters = WebSocket::TerminalWaiters; AsyncWaiters TakeAsyncWaitersLocked() { + // Whatever deadline bounded the receive being taken is spent with it. + // Destroying the timer cancels its pending wait, so the handler returns + // on operation_aborted instead of waking the executor for a park that + // no longer exists; the generation guard already made such a wakeup a + // no-op, this just spares it. Safe from any thread for the same reason + // ArmReceiveDeadlineLocked's emplace is: every touch of the timer is + // under mutex_, and the armed handler captures the session weakly + // rather than the timer. + receive_deadline_.reset(); return {std::exchange(pending_receive_, nullptr), std::exchange(pending_send_, nullptr)}; }