From 69ae5d15cc0342a09f06761bd143fad633ad45f1 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 13 Oct 2026 18:50:08 +0000 Subject: [PATCH 1/7] Kick: a close that tells the peer why and bounds the wait --- .../opal/eventstream/async_event_stream.h | 2 +- .../include/opal/eventstream/event_stream.h | 49 ++++++- .../opal/eventstream/jsonrpc_stream_socket.h | 2 + runtime/include/opal/http/websocket.h | 54 ++++++++ .../include/opal/server/session_registry.h | 60 ++++++--- .../src/eventstream/jsonrpc_stream_socket.cc | 8 ++ runtime/src/http/beast_transport.cc | 67 ++++++++- runtime/src/http/websocket_pair.cc | 23 +++- .../opal/testing/websocket_contract_test.h | 19 +++ .../eventstream/async_event_stream_test.cc | 46 +++++++ .../eventstream/jsonrpc_stream_socket_test.cc | 12 ++ runtime/tests/http/beast_websocket_test.cc | 127 ++++++++++++++++++ runtime/tests/http/websocket_pair_test.cc | 41 ++++++ runtime/tests/server/session_registry_test.cc | 24 +++- 14 files changed, 498 insertions(+), 36 deletions(-) diff --git a/runtime/include/opal/eventstream/async_event_stream.h b/runtime/include/opal/eventstream/async_event_stream.h index 16b6bbc3..05dbeb9f 100644 --- a/runtime/include/opal/eventstream/async_event_stream.h +++ b/runtime/include/opal/eventstream/async_event_stream.h @@ -404,7 +404,7 @@ class AsyncEventStream { // stream after a Share() closes it, waits out pinned operations, and // leaves every handle failing softly. EventStreamHandle Share() { - return EventStreamHandle(view_.Ensure(socket_.get()), encode_); + return EventStreamHandle(view_.Ensure(socket_.get(), socket_), encode_); } private: diff --git a/runtime/include/opal/eventstream/event_stream.h b/runtime/include/opal/eventstream/event_stream.h index cb31a701..817b902b 100644 --- a/runtime/include/opal/eventstream/event_stream.h +++ b/runtime/include/opal/eventstream/event_stream.h @@ -54,6 +54,9 @@ struct SharedSessionState { std::condition_variable idle; http::WebSocket* socket = nullptr; // null once the stream ended int active = 0; // handle operations inside socket calls + // Engaged when the stream owns its socket (the coroutine adapter): what + // lets a handle end the session without a pin (EventStreamHandle::End). + std::weak_ptr owner; }; // Drops one pin; the revoking stream's drain wait wakes at zero. Free so @@ -86,10 +89,12 @@ class SharedViewOwner { // The state handles share, created on first use (a stream that never // shares carries none and End() is a no-op). - std::shared_ptr Ensure(http::WebSocket* socket) { + std::shared_ptr Ensure(http::WebSocket* socket, + std::weak_ptr owner = {}) { if (state_ == nullptr) { state_ = std::make_shared(); state_->socket = socket; + state_->owner = std::move(owner); socket_ = socket; } return state_; @@ -161,10 +166,15 @@ class EventStreamHandle { // session from outside the handler (the slow-consumer policy): the // owner's blocked Receive/Send unblock exactly as for any other close. void Close() const { - http::WebSocket* socket = Acquire(); - if (socket == nullptr) return; - socket->Close(); - Release(); + End([](http::WebSocket& socket) { socket.Close(); }); + } + + // The socket's Kick (opal/http/websocket.h): Close for a peer the caller + // has given up on — `why` on the close frame, and at most `grace` spent + // waiting for an answer. A no-op once the session ended. + void Kick(const http::CloseReason& why, std::chrono::milliseconds grace) const { + http::CheckCloseReason(why); + End([&why, grace](http::WebSocket& socket) { socket.Kick(why, grace); }); } // The completion-driven send (ADR-0019): the socket's SendAsync contract @@ -227,6 +237,35 @@ class EventStreamHandle { void Release() const { internal::ReleasePin(*state_); } + // Close and Kick. A transport may complete the parked receive inline on + // the closing thread (the in-memory pair does), ending the loop and + // destroying the stream inside this very call — and the stream's + // teardown waits for every pin, this frame's included. So a socket the + // stream owns is kept alive by ownership instead of a pin; only a + // borrowed one is pinned, and a borrowing stream is never torn down by + // its own socket's completions. + template + void End(const Ending& ending) const { + if (state_ == nullptr) return; + std::shared_ptr owned; + http::WebSocket* borrowed = nullptr; + { + const std::scoped_lock lock(state_->mutex); + if (state_->socket == nullptr) return; + owned = state_->owner.lock(); + if (owned == nullptr) { + ++state_->active; + borrowed = state_->socket; + } + } + if (owned != nullptr) { + ending(*owned); + return; + } + ending(*borrowed); + Release(); + } + std::shared_ptr state_; std::function(const Tx&)> encode_; }; diff --git a/runtime/include/opal/eventstream/jsonrpc_stream_socket.h b/runtime/include/opal/eventstream/jsonrpc_stream_socket.h index af52e3d4..5b5ee8ad 100644 --- a/runtime/include/opal/eventstream/jsonrpc_stream_socket.h +++ b/runtime/include/opal/eventstream/jsonrpc_stream_socket.h @@ -75,6 +75,8 @@ class JsonRpcStreamSocket final : public http::WebSocket { Outcome> Receive(std::chrono::milliseconds timeout) override; Outcome Send(const Message& message) override; void Close() override; + void Kick(const http::CloseReason& why, std::chrono::milliseconds grace) override; + std::optional PeerCloseReason() const 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 diff --git a/runtime/include/opal/http/websocket.h b/runtime/include/opal/http/websocket.h index d622460e..d1a551af 100644 --- a/runtime/include/opal/http/websocket.h +++ b/runtime/include/opal/http/websocket.h @@ -2,17 +2,43 @@ #define OPAL_HTTP_WEBSOCKET_H_ #include +#include #include #include #include #include +#include "opal/core/fatal.h" #include "opal/eventstream/frame.h" #include "opal/http/headers.h" #include "opal/http/transport.h" namespace opal::http { +// A close frame's status and reason (RFC 6455 §7.4): 1000 is a normal end, +// 4000-4999 are the application's own (a hub's "superseded", say), and the +// reason is UTF-8 for people. +struct CloseReason { + std::uint16_t code = 1000; + std::string reason; + + friend bool operator==(const CloseReason&, const CloseReason&) = default; +}; + +// Kick's precondition: a code an endpoint may send (1000, or 3000-4999 — +// the protocol's own failure codes are the transport's to send) and a +// reason that fits a control frame beside it. Fails fast (ADR-0009). +inline void CheckCloseReason(const CloseReason& why) { + if (why.code != 1000 && (why.code < 3000 || why.code > 4999)) { + internal::Fatal("websocket Kick: close code " + std::to_string(why.code) + + " is not 1000 or in 3000-4999"); + } + if (why.reason.size() > 123) { + internal::Fatal("websocket Kick: close reason is " + std::to_string(why.reason.size()) + + " bytes; a close frame holds 123"); + } +} + // A blocking, full-duplex event-stream session over one WebSocket // connection (ADR-0015). What travels is eventstream::Message — exactly one // frame per binary WebSocket message; text messages and malformed frames @@ -114,6 +140,34 @@ class WebSocket { // TerminalWaiters. virtual void Close() = 0; + // Ends the session for a peer the caller has given up on (ADR-0022's + // kick): a close frame carrying `why`, then the connection torn down + // after at most `grace` waiting for the peer's answer, where Close() + // waits out the transport's whole closing handshake. A peer that is + // still there reads `why` (a browser's CloseEvent.code and .reason); one + // that is gone costs `grace`, not the handshake timeout. Everything else + // is Close()'s contract: idempotent, non-blocking, safe from any thread, + // and it unblocks and completes what is parked — a peer that does not + // answer in time ends the session with an error rather than a clean + // nullopt. On a session already closing it stops waiting for the peer + // and the frame already sent stands. `why.code` must be 1000 or in + // 3000-4999 and `why.reason` at most 123 bytes (RFC 6455 §5.5, §7.4); + // anything else is a contract violation and fails fast (ADR-0009). + // + // The default is Close(), for an implementation with neither close + // frames nor a handshake to bound. + virtual void Kick(const CloseReason& why, std::chrono::milliseconds grace) { + (void)grace; + CheckCloseReason(why); + Close(); + } + + // The close frame the peer ended the session with: engaged once a close + // the peer started has been read (its code and reason, as sent), and + // never for a session this end closed or one that failed. Safe from any + // thread. + virtual std::optional PeerCloseReason() const { return std::nullopt; } + // --- Completion-driven twins (ADR-0019) ------------------------------ // // Same outcomes as the blocking calls, delivered to a callback that diff --git a/runtime/include/opal/server/session_registry.h b/runtime/include/opal/server/session_registry.h index b0637134..fd31f2d9 100644 --- a/runtime/include/opal/server/session_registry.h +++ b/runtime/include/opal/server/session_registry.h @@ -462,32 +462,41 @@ class SessionRegistry { return Broadcast([&event](const Id&) { return event; }, delivery); } - // Closes id's current session — the kick primitive (ADR-0022): the - // session observes the close and its handler runs its normal exit path, - // making the id admittable on the next dial — freed outright on a - // Remove exit; parked-resumable on a Detach exit, where the next - // ResumeOrAdd reports kResumed and snapshot replay carries the old - // session's identity across. Policy stays - // with the application: this is how a ResumeOrAdd refusal becomes - // actionable when the old session is known dead (a silent partition, an - // operator action) instead of waiting out TCP. On a detached entry it - // closes an already-closed handle — a harmless no-op (detached ids are - // Resume's and Remove's business). False for an unknown id. The handle - // copy is taken under the entry lock (the Resume-swap discipline) and - // closed outside all locks. + // Closes id's current session (ADR-0022): the session observes the + // close and its handler runs its normal exit path, making the id + // admittable on the next dial — freed outright on a Remove exit; + // parked-resumable on a Detach exit, where the next ResumeOrAdd reports + // kResumed and snapshot replay carries the old session's identity + // across. Close runs the transport's whole closing handshake, so a peer + // that never answers holds the id until the handshake times out; Kick, + // below, is the kick for one the application has given up on. On a + // detached entry it closes an already-closed handle — a harmless no-op + // (detached ids are Resume's and Remove's business). False for an + // unknown id. The handle copy is taken under the entry lock (the + // Resume-swap discipline) and closed outside all locks. bool Close(const Id& id) { - std::optional doomed; - { - const std::scoped_lock lock(mutex_); - const auto it = sessions_.find(id); - if (it == sessions_.end()) return false; - const std::scoped_lock entry_lock(it->second->mutex); - doomed.emplace(it->second->handle); - } + std::optional doomed = CurrentHandle(id); + if (!doomed.has_value()) return false; doomed->Close(); return true; } + // The kick primitive (ADR-0022): Close for a session the application + // has given up on — a silent partition, a newer connection under the + // same identity, an operator action. The peer, if still there, reads + // `why` (an application code says why it was dropped, so it can stop + // redialing); one that is gone holds the id for at most `grace` rather + // than the closing handshake's timeout. Policy stays with the + // application: ResumeOrAdd never kicks on its own. Everything else is + // Close's, false for an unknown id included. + bool Kick(const Id& id, const http::CloseReason& why, std::chrono::milliseconds grace) { + http::CheckCloseReason(why); + std::optional doomed = CurrentHandle(id); + if (!doomed.has_value()) return false; + doomed->Kick(why, grace); + return true; + } + // Closes every registered session (idempotent, non-blocking): each // blocked handler wakes and returns, unregistering itself on the way out. void CloseAll() { @@ -532,6 +541,15 @@ class SessionRegistry { } private: + // A copy of id's current handle, taken under the entry lock Resume swaps + // under; nullopt for an unknown id. + std::optional CurrentHandle(const Id& id) { + const std::scoped_lock lock(mutex_); + const auto it = sessions_.find(id); + if (it == sessions_.end()) return std::nullopt; + const std::scoped_lock entry_lock(it->second->mutex); + return it->second->handle; + } struct Queued { Tx event; DeliveryClass delivery; diff --git a/runtime/src/eventstream/jsonrpc_stream_socket.cc b/runtime/src/eventstream/jsonrpc_stream_socket.cc index 199dc225..b530b409 100644 --- a/runtime/src/eventstream/jsonrpc_stream_socket.cc +++ b/runtime/src/eventstream/jsonrpc_stream_socket.cc @@ -129,6 +129,14 @@ Outcome JsonRpcStreamSocket::Send(const Message& message) { void JsonRpcStreamSocket::Close() { inner_->Close(); } +void JsonRpcStreamSocket::Kick(const http::CloseReason& why, std::chrono::milliseconds grace) { + inner_->Kick(why, grace); +} + +std::optional JsonRpcStreamSocket::PeerCloseReason() const { + return inner_->PeerCloseReason(); +} + void JsonRpcStreamSocket::ReceiveAsync(ReceiveCallback callback) { inner_->ReceiveAsync(PoliceAsync(std::move(callback))); } diff --git a/runtime/src/http/beast_transport.cc b/runtime/src/http/beast_transport.cc index b568e781..d5093aaa 100644 --- a/runtime/src/http/beast_transport.cc +++ b/runtime/src/http/beast_transport.cc @@ -578,6 +578,40 @@ class WsSession final : public WebSocketSessionBase, } } + // Close with `why` on the frame and the closing handshake bounded by + // `grace` (Beast's handshake timeout for this close only). A handshake + // already running is Close's own, waiting on the peer for the whole + // request timeout: the kick closes the socket under it instead. + void Kick(const CloseReason& why, std::chrono::milliseconds grace) override { + CheckCloseReason(why); + auto self = this->shared_from_this(); + try { + asio::post(ws_.get_executor(), [self, why, grace] { + std::unique_lock lock(self->mutex_); + if (self->close_started_) { + if (self->failed_ || self->peer_closed_) return; + lock.unlock(); + beast::error_code ignored; + (void)beast::get_lowest_layer(self->ws_).socket().close(ignored); + return; + } + if (!self->close_requested_) { + self->close_reason_ = + bws::close_reason(static_cast(why.code), why.reason); + } + self->close_grace_ = grace; + self->RequestCloseLocked(lock); + }); + } catch (...) { + Abort("websocket: cannot schedule the kick"); + } + } + + std::optional PeerCloseReason() const override { + const std::scoped_lock lock(mutex_); + return peer_close_; + } + // Fails both facade calls immediately and closes the socket out from // under any outstanding op — Stop()-time and client-teardown semantics, // safe from any thread even when the io context is already stopped. @@ -625,6 +659,9 @@ class WsSession final : public WebSocketSessionBase, { const std::scoped_lock lock(mutex_); peer_closed_ = true; + // The peer started this close: its frame is what reason() holds. + peer_close_ = CloseReason{.code = static_cast(ws_.reason().code), + .reason = std::string(ws_.reason().reason)}; waiters = TakeAsyncWaitersLocked(); wake_.notify_all(); } @@ -846,13 +883,23 @@ class WsSession final : public WebSocketSessionBase, void StartCloseLocked(std::unique_lock& lock) { close_started_ = true; - const bws::close_code code = close_code_; + const bws::close_reason why = close_reason_; + const std::optional grace = close_grace_; lock.unlock(); + if (grace.has_value()) { + // Read when the close starts (Beast arms the closing phase's timer + // from it then), so it bounds this handshake and nothing else. + constexpr std::chrono::milliseconds kMaxGrace = std::chrono::hours(24); + bws::stream_base::timeout timeout; + ws_.get_option(timeout); + timeout.handshake_timeout = std::clamp(*grace, std::chrono::milliseconds::zero(), kMaxGrace); + ws_.set_option(timeout); + } auto self = this->shared_from_this(); // Beast's close op performs the whole closing handshake (it reads // until the peer's close arrives, then tears down): its success IS // the session's clean end. - ws_.async_close(code, [self](beast::error_code ec) { + ws_.async_close(why, [self](beast::error_code ec) { AsyncWaiters waiters; bool clean = false; std::string why; @@ -906,7 +953,7 @@ class WsSession final : public WebSocketSessionBase, waiters = TakeAsyncWaitersLocked(); why = error_; wake_.notify_all(); - close_code_ = bws::close_code::protocol_error; + close_reason_ = bws::close_reason(bws::close_code::protocol_error); close_requested_ = true; if (!close_started_ && !wire_write_busy_) { StartCloseLocked(lock); // leaves the lock released @@ -957,7 +1004,7 @@ class WsSession final : public WebSocketSessionBase, std::mutex send_mutex_; // serializes Send callers end to end - std::mutex mutex_; + mutable std::mutex mutex_; std::condition_variable wake_; std::deque received_; // The ADR-0019 parked completions: at most one of each; every terminal @@ -980,7 +1027,11 @@ class WsSession final : public WebSocketSessionBase, bool wire_write_busy_ = false; bool close_requested_ = false; bool close_started_ = false; - bws::close_code close_code_ = bws::close_code::normal; + bws::close_reason close_reason_{bws::close_code::normal}; + // Engaged by Kick: the closing handshake's bound. + std::optional close_grace_; + // The close the peer started, as its frame carried it. + std::optional peer_close_; }; // The websocket_gate and on_websocket callbacks are contained the way @@ -2476,6 +2527,12 @@ class DialedWebSocket final : public WebSocket { return io_->session->Send(message); } void Close() override { io_->session->Close(); } + void Kick(const CloseReason& why, std::chrono::milliseconds grace) override { + io_->session->Kick(why, grace); + } + std::optional PeerCloseReason() const override { + return io_->session->PeerCloseReason(); + } void ReceiveAsync(WebSocket::ReceiveCallback callback) override { io_->session->ReceiveAsync(std::move(callback)); } diff --git a/runtime/src/http/websocket_pair.cc b/runtime/src/http/websocket_pair.cc index 211be218..a9ed0422 100644 --- a/runtime/src/http/websocket_pair.cc +++ b/runtime/src/http/websocket_pair.cc @@ -54,6 +54,9 @@ struct PairState { std::array blocked_receivers{}; std::array blocked_senders{}; bool closed = false; + // Indexed by the end that reads it (its send index): the close the other + // end started, set once by the first Close or Kick. + std::array, 2> peer_close; }; class PairEnd final : public WebSocket { @@ -102,7 +105,23 @@ class PairEnd final : public WebSocket { return Unit{}; } - void Close() override { + void Close() override { End(CloseReason{}); } + + // The pair has no handshake to wait out, so the grace is moot: the kick + // is a close that tells the peer why. + void Kick(const CloseReason& why, std::chrono::milliseconds grace) override { + (void)grace; + CheckCloseReason(why); + End(why); + } + + std::optional PeerCloseReason() const override { + const std::scoped_lock lock(state_->mutex); + return state_->peer_close[send_index_]; + } + + private: + void End(const CloseReason& why) { // One close ends the session for both ends, so both ends' waiters come // out here — one TerminalWaiters each, since the send-before-receive // rule it enforces is per-session-state: what an end's receive @@ -110,6 +129,7 @@ class PairEnd final : public WebSocket { std::array waiters; { const std::scoped_lock lock(state_->mutex); + if (!state_->closed) state_->peer_close[1 - send_index_] = why; state_->closed = true; for (std::size_t end = 0; end < 2; ++end) { WebSocket::SendCallback send; @@ -138,6 +158,7 @@ class PairEnd final : public WebSocket { } } + public: void ReceiveAsync(WebSocket::ReceiveCallback callback) override { ReceiveAsyncWithin(std::nullopt, std::move(callback)); } diff --git a/runtime/testing/include/opal/testing/websocket_contract_test.h b/runtime/testing/include/opal/testing/websocket_contract_test.h index 8e256ce4..1e523d41 100644 --- a/runtime/testing/include/opal/testing/websocket_contract_test.h +++ b/runtime/testing/include/opal/testing/websocket_contract_test.h @@ -312,6 +312,24 @@ TYPED_TEST_P(WebSocketContractTest, ATerminalTransitionCompletesALoneParkedRecei AwaitLoopEnd(loop_ended, "a lone parked receive was left hanging"); } +// A kick ends the session without the peer's help (ADR-0022): the driver's +// peer never reads, so it never answers a close either, and the parked +// receive must still complete within the grace rather than the closing +// handshake's timeout. +TYPED_TEST_P(WebSocketContractTest, AKickCompletesAParkedReceiveWithoutThePeer) { + TypeParam driver; + std::shared_ptr socket = driver.Socket(); + ContractMailbox>> ended; + socket->ReceiveAsync([&ended](Outcome> received) { + ended.Post(std::move(received)); + }); + ASSERT_FALSE(ended.WaitFor(std::chrono::milliseconds(100))) << "nothing has ended the session"; + socket->Kick({.code = 4001, .reason = "superseded"}, std::chrono::milliseconds(200)); + ASSERT_TRUE(ended.WaitFor(std::chrono::seconds(5))) + << "the kick waited for a peer that never answers"; + (void)ended.Wait(); +} + // One outstanding send-class operation per session: the second refuses // inline with Validation rather than queueing behind the first. TYPED_TEST_P(WebSocketContractTest, ASecondSendClassOperationRefusesWhileOneIsParked) { @@ -459,6 +477,7 @@ REGISTER_TYPED_TEST_SUITE_P(WebSocketContractTest, ATerminalTransitionFiresTheParkedSendBeforeTheParkedReceive, ATerminalTransitionCompletesALoneParkedCoroutineSend, ATerminalTransitionCompletesALoneParkedReceive, + AKickCompletesAParkedReceiveWithoutThePeer, ASecondSendClassOperationRefusesWhileOneIsParked, ATimedReceiveOnAQuietWireTimesOutAndReleasesItsSlot, ANonPositiveTimedReceivePolls, diff --git a/runtime/tests/eventstream/async_event_stream_test.cc b/runtime/tests/eventstream/async_event_stream_test.cc index 9ebabc0f..223d4537 100644 --- a/runtime/tests/eventstream/async_event_stream_test.cc +++ b/runtime/tests/eventstream/async_event_stream_test.cc @@ -14,6 +14,8 @@ #include #include #include +#include +#include #include #include #include @@ -669,6 +671,50 @@ TEST(AsyncEventStreamTest, SharedHandlesOutliveTheLoopAndItsStream) { EXPECT_EQ(stale.error().kind(), ErrorKind::kTransport); } +// A hub ending a session from outside the loop (a registry Close or Kick) +// on a transport that completes the parked receive inline on the closing +// thread — the pair's documented behavior: the loop ends and its stream +// is destroyed inside that very call, so the handle's call must not hold +// anything the stream's teardown waits for. +void EndFromOutsideTheParkedLoop(const std::function&)>& end) { + auto [client_socket, server_socket] = http::InMemoryWebSocketPair::Create(); + Mailbox> minted; + std::atomic done{false}; + [](std::shared_ptr socket, Mailbox>* minted, + std::atomic* done) -> Detached { + AsyncServer stream(std::move(socket), EncodePong, DecodePing); + minted->Post(stream.Share()); + (void)co_await stream.Receive(); // parked until the handle ends it + *done = true; + }(server_socket, &minted, &done); + EventStreamHandle handle = minted.Wait(); + + std::promise returned; + std::thread ender([&] { + end(handle); + returned.set_value(); + }); + if (returned.get_future().wait_for(std::chrono::seconds(5)) != std::future_status::ready) { + ender.detach(); // wedged for good; the assertion below names it + FAIL() << "ending the session through its handle deadlocked on the stream's teardown"; + } + ender.join(); + EXPECT_TRUE(done.load()); + auto at_client = client_socket->Receive(); + ASSERT_TRUE(at_client.ok()) << at_client.error().message(); + EXPECT_FALSE(at_client->has_value()); +} + +TEST(AsyncEventStreamTest, AHandleCloseThatEndsTheLoopInlineReturns) { + EndFromOutsideTheParkedLoop([](const EventStreamHandle& handle) { handle.Close(); }); +} + +TEST(AsyncEventStreamTest, AHandleKickThatEndsTheLoopInlineReturns) { + EndFromOutsideTheParkedLoop([](const EventStreamHandle& handle) { + handle.Kick({.code = 4001, .reason = "superseded"}, std::chrono::milliseconds(100)); + }); +} + TEST(AsyncEventStreamTest, HandleSendAsyncCompletesAndFailsSoftlyAfterRevocation) { auto [client_socket, server_socket] = http::InMemoryWebSocketPair::Create(); Mailbox> live; diff --git a/runtime/tests/eventstream/jsonrpc_stream_socket_test.cc b/runtime/tests/eventstream/jsonrpc_stream_socket_test.cc index de581786..03469003 100644 --- a/runtime/tests/eventstream/jsonrpc_stream_socket_test.cc +++ b/runtime/tests/eventstream/jsonrpc_stream_socket_test.cc @@ -62,6 +62,18 @@ TEST(JsonRpcStreamSocketTest, EventsRoundTripBetweenTwoWrappedEnds) { EXPECT_EQ(**echoed, event); } +// The kick and the close it reads are the inner session's: the wrapper +// passes the code through both ways. +TEST(JsonRpcStreamSocketTest, AKickAndThePeersCloseReasonAreTheInnerSessions) { + auto [left, right] = http::InMemoryWebSocketPair::Create(); + auto server = Wrap(right, JsonRpcStreamSocket::Role::kServer); + auto client = Wrap(left, JsonRpcStreamSocket::Role::kClient); + server->Kick({.code = 4001, .reason = "superseded"}, std::chrono::milliseconds(100)); + EXPECT_EQ(left->PeerCloseReason(), (http::CloseReason{.code = 4001, .reason = "superseded"})); + EXPECT_EQ(client->PeerCloseReason(), (http::CloseReason{.code = 4001, .reason = "superseded"})); + EXPECT_EQ(server->PeerCloseReason(), std::nullopt); +} + TEST(JsonRpcStreamSocketTest, TheWireCarriesThePinnedNotificationText) { // One end unwrapped: what it receives is the wire — a headerless // message whose payload is the notification text, byte-pinned. diff --git a/runtime/tests/http/beast_websocket_test.cc b/runtime/tests/http/beast_websocket_test.cc index 67a32476..940eed55 100644 --- a/runtime/tests/http/beast_websocket_test.cc +++ b/runtime/tests/http/beast_websocket_test.cc @@ -21,6 +21,7 @@ #include #include #include +#include #include #include #include @@ -1434,6 +1435,132 @@ TEST(BeastWebSocketTest, AThrowingServeCallbackStillClosesTheSessionCleanly) { server.Stop(); } +// A server whose one session hands the test its socket and then sits in +// Receive, reporting how that receive ended — so the test can end the +// session from outside, as a hub does, and time it. The socket is the +// serve callback's borrow: use it only before the session has ended. +class HeldSessionServer { + public: + struct Ended { + Outcome> received; + std::optional peer_close; // read before the borrow ends + }; + + explicit HeldSessionServer(int request_timeout_seconds = 30) { + BeastServerTransport::Options options; + options.request_timeout_seconds = request_timeout_seconds; + options.on_websocket = [this](const HttpRequest&, WebSocket& socket) { + socket_.set_value(&socket); + auto received = socket.Receive(); + ended_promise_.set_value(Ended{std::move(received), socket.PeerCloseReason()}); + }; + server_ = std::make_unique(options); + EXPECT_TRUE(server_->Start(NotFoundHandler()).ok()); + } + ~HeldSessionServer() { server_->Stop(); } + + int port() const { return server_->port(); } + // The server's end of the session, once a peer has upgraded. + WebSocket& Socket() { + if (held_ == nullptr) { + auto socket = socket_.get_future(); + EXPECT_EQ(socket.wait_for(std::chrono::seconds(5)), std::future_status::ready); + held_ = socket.get(); + } + return *held_; + } + // How the server's receive ended, if it did within `budget`. + std::optional EndedWithin(std::chrono::milliseconds budget) { + if (!ended_.has_value() && ended_future_.wait_for(budget) == std::future_status::ready) { + ended_ = ended_future_.get(); + } + return ended_; + } + + private: + std::promise socket_; + WebSocket* held_ = nullptr; + std::promise ended_promise_; + std::future ended_future_ = ended_promise_.get_future(); + std::optional ended_; + std::unique_ptr server_; +}; + +// Why Kick exists: Close's closing handshake waits for the peer's answer, +// and a peer that is gone never sends one. +TEST(BeastWebSocketTest, ACloseWaitsOutTheHandshakeForAPeerThatNeverAnswers) { + HeldSessionServer server(/*request_timeout_seconds=*/2); + RawWsPeer peer(server.port()); // never reads, so never answers a close + server.Socket().Close(); + EXPECT_FALSE(server.EndedWithin(std::chrono::milliseconds(1000)).has_value()) + << "the close ended before the handshake timeout"; + ASSERT_TRUE(server.EndedWithin(std::chrono::seconds(6)).has_value()); +} + +TEST(BeastWebSocketTest, AKickEndsTheSessionOfAPeerThatNeverAnswersWithinItsGrace) { + HeldSessionServer server; // a 30s closing handshake, were it Close + RawWsPeer peer(server.port()); + const auto kicked = std::chrono::steady_clock::now(); + server.Socket().Kick({.code = 4001, .reason = "superseded"}, std::chrono::milliseconds(200)); + auto ended = server.EndedWithin(std::chrono::seconds(5)); + ASSERT_TRUE(ended.has_value()) << "the kick waited for a peer that never answers"; + EXPECT_FALSE(ended->received.ok()) << "an unanswered close is not a clean end"; + EXPECT_LT(std::chrono::steady_clock::now() - kicked, std::chrono::seconds(2)); +} + +TEST(BeastWebSocketTest, AKickTellsAPeerThatAnswersItsCodeAndReason) { + HeldSessionServer server; + RawWsPeer peer(server.port()); + server.Socket().Kick({.code = 4001, .reason = "superseded"}, std::chrono::seconds(5)); + EXPECT_EQ(peer.DrainToCloseCode(), boost::beast::websocket::error::closed); + EXPECT_EQ(peer.reason().code, 4001); + EXPECT_EQ(std::string(peer.reason().reason), "superseded"); + auto ended = server.EndedWithin(std::chrono::seconds(5)); + ASSERT_TRUE(ended.has_value()); + ASSERT_TRUE(ended->received.ok()) << ended->received.error().message(); + EXPECT_FALSE(ended->received->has_value()) << "an answered kick is a clean end"; +} + +TEST(BeastWebSocketTest, TheClientReadsTheCodeAKickSent) { + HeldSessionServer server; + auto dialed = BeastWebSocketClient::Dial({.host = "127.0.0.1", .port = server.port()}); + ASSERT_TRUE(dialed.ok()) << dialed.error().message(); + server.Socket().Kick({.code = 4001, .reason = "superseded"}, std::chrono::seconds(5)); + auto ended = (*dialed)->Receive(); + ASSERT_TRUE(ended.ok()) << ended.error().message(); + EXPECT_FALSE(ended->has_value()); + EXPECT_EQ((*dialed)->PeerCloseReason(), (CloseReason{.code = 4001, .reason = "superseded"})); +} + +TEST(BeastWebSocketTest, AServerCloseReadsAsANormalClosureAtTheClient) { + HeldSessionServer server; + auto dialed = BeastWebSocketClient::Dial({.host = "127.0.0.1", .port = server.port()}); + ASSERT_TRUE(dialed.ok()) << dialed.error().message(); + EXPECT_EQ((*dialed)->PeerCloseReason(), std::nullopt) << "nothing closed yet"; + server.Socket().Close(); + auto ended = (*dialed)->Receive(); + ASSERT_TRUE(ended.ok()) << ended.error().message(); + EXPECT_FALSE(ended->has_value()); + EXPECT_EQ((*dialed)->PeerCloseReason(), (CloseReason{.code = 1000, .reason = ""})); + // The server started that close: its end read no close of the peer's. + auto server_end = server.EndedWithin(std::chrono::seconds(5)); + ASSERT_TRUE(server_end.has_value()); + EXPECT_EQ(server_end->peer_close, std::nullopt); +} + +TEST(BeastWebSocketTest, AKickOnASessionAlreadyClosingStopsWaitingForThePeer) { + HeldSessionServer server; + RawWsPeer peer(server.port()); + WebSocket& socket = server.Socket(); + socket.Close(); + ASSERT_FALSE(server.EndedWithin(std::chrono::milliseconds(300)).has_value()); + const auto kicked = std::chrono::steady_clock::now(); + socket.Kick({.code = 4001, .reason = "superseded"}, std::chrono::milliseconds(200)); + ASSERT_TRUE(server.EndedWithin(std::chrono::seconds(5)).has_value()) + << "the kick left the close waiting out its handshake"; + EXPECT_LT(std::chrono::steady_clock::now() - kicked, std::chrono::seconds(2)); +} + TEST(BeastWebSocketTest, AVanishedPeerIsDetectedByTheServerIdleTimeout) { // ADR-0015: "a vanished peer is detected without any application ping // protocol." The knob makes it unit-testable: idle timeout 1s, a peer diff --git a/runtime/tests/http/websocket_pair_test.cc b/runtime/tests/http/websocket_pair_test.cc index c3a0c045..3378eb49 100644 --- a/runtime/tests/http/websocket_pair_test.cc +++ b/runtime/tests/http/websocket_pair_test.cc @@ -239,6 +239,47 @@ TEST(WebSocketPairTest, CloseUnblocksASenderStalledOnBackpressure) { EXPECT_TRUE(failed.load()); } +// The peer of a kicked end reads why it was dropped; the end that kicked +// started the close itself and reads nothing. +TEST(WebSocketPairTest, AKickTellsThePeerItsCodeAndReason) { + auto [a, b] = InMemoryWebSocketPair::Create(); + a->Kick({.code = 4001, .reason = "superseded"}, std::chrono::milliseconds(100)); + + auto ended = b->Receive(); + ASSERT_TRUE(ended.ok()) << ended.error().message(); + EXPECT_FALSE(ended->has_value()); + EXPECT_EQ(b->PeerCloseReason(), (CloseReason{.code = 4001, .reason = "superseded"})); + EXPECT_EQ(a->PeerCloseReason(), std::nullopt); +} + +TEST(WebSocketPairTest, APlainCloseReadsAsANormalClosureAtThePeer) { + auto [a, b] = InMemoryWebSocketPair::Create(); + EXPECT_EQ(b->PeerCloseReason(), std::nullopt) << "nothing closed yet"; + a->Close(); + EXPECT_EQ(b->PeerCloseReason(), (CloseReason{.code = 1000, .reason = ""})); + EXPECT_EQ(a->PeerCloseReason(), std::nullopt); +} + +// Only the first close says why: a later kick from either end is the +// no-op any close of a closed session is. +TEST(WebSocketPairTest, AKickAfterTheSessionClosedChangesNothing) { + auto [a, b] = InMemoryWebSocketPair::Create(); + a->Close(); + a->Kick({.code = 4001, .reason = "late"}, std::chrono::milliseconds(100)); + b->Kick({.code = 4002, .reason = "later"}, std::chrono::milliseconds(100)); + EXPECT_EQ(b->PeerCloseReason(), (CloseReason{.code = 1000, .reason = ""})); + EXPECT_EQ(a->PeerCloseReason(), std::nullopt); +} + +TEST(WebSocketPairDeathTest, AKickWithACodeNoEndpointMaySendDies) { + auto [a, b] = InMemoryWebSocketPair::Create(); + EXPECT_DEATH(a->Kick({.code = 1006, .reason = ""}, std::chrono::milliseconds(1)), + "close code 1006 is not 1000 or in 3000-4999"); + EXPECT_DEATH( + a->Kick({.code = 4001, .reason = std::string(124, 'x')}, std::chrono::milliseconds(1)), + "close reason is 124 bytes; a close frame holds 123"); +} + TEST(WebSocketPairTest, AMessageTheCodecRefusesFailsValidationAndSparesTheSession) { auto [a, b] = InMemoryWebSocketPair::Create(); const auto refused = a->Send(Message{.headers = {{"", true}}}); // empty header name diff --git a/runtime/tests/server/session_registry_test.cc b/runtime/tests/server/session_registry_test.cc index ed9c3571..2f4ec861 100644 --- a/runtime/tests/server/session_registry_test.cc +++ b/runtime/tests/server/session_registry_test.cc @@ -1103,9 +1103,9 @@ TEST(SessionRegistryGraceTest, ResumeOrAddConvergesWhenDetachLandsMidRetry) { EXPECT_EQ(NextAt(fresh), "converged"); } -TEST(SessionRegistryGraceTest, CloseKicksTheLiveSessionAndFreesTheId) { - // The kick primitive (ADR-0022): the session observes the close, the - // handler side runs its normal exit path, and the id frees. +TEST(SessionRegistryGraceTest, CloseEndsTheLiveSessionAndFreesTheId) { + // ADR-0022: the session observes the close, the handler side runs its + // normal exit path, and the id frees. Registry registry; Session session = MakeSession(); ASSERT_TRUE(registry.Add("ada", *session.handle)); @@ -1118,6 +1118,24 @@ TEST(SessionRegistryGraceTest, CloseKicksTheLiveSessionAndFreesTheId) { EXPECT_FALSE(registry.Close("ghost")); } +TEST(SessionRegistryGraceTest, KickEndsTheLiveSessionTellingThePeerWhy) { + // The kick primitive (ADR-0022): Close's exit path, and the peer reads + // the application's code — how a superseded client knows not to redial. + Registry registry; + Session session = MakeSession(); + ASSERT_TRUE(registry.Add("ada", *session.handle)); + + EXPECT_TRUE( + registry.Kick("ada", {.code = 4001, .reason = "superseded"}, std::chrono::milliseconds(100))); + EXPECT_EQ(NextAt(session), std::nullopt); + EXPECT_EQ(session.client->PeerCloseReason(), + (http::CloseReason{.code = 4001, .reason = "superseded"})); + EXPECT_TRUE(registry.Remove("ada")); + Session fresh = MakeSession(); + EXPECT_TRUE(registry.Add("ada", *fresh.handle)); + EXPECT_FALSE(registry.Kick("ghost", {.code = 4001, .reason = ""}, std::chrono::milliseconds(1))); +} + TEST(SessionRegistryGraceTest, CloseOnADetachedEntryIsAHarmlessNoOp) { Registry::Options options; options.grace_period = std::chrono::seconds{300}; From d8605c056df2b9ba591ee150baee8ce8a629da4d Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 13 Oct 2026 19:10:22 +0000 Subject: [PATCH 2/7] Kick: ADR-0022 amendment, guide, changelog, consumer test --- CHANGELOG.md | 21 ++++++++ .../adr/0022-registry-admission-primitives.md | 52 ++++++++++++++++++- docs/production-guide.md | 24 +++++++-- docs/runtime.md | 2 +- .../websocket_acceptance_test.cc | 25 +++++++++ .../include/opal/server/session_registry.h | 2 +- 6 files changed, 118 insertions(+), 8 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b87440e4..082d0bb6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -90,6 +90,19 @@ policy in [docs/versioning.md](docs/versioning.md). ### Added +- **`SessionRegistry::Kick(id, why, grace)`: the kick that does not wait + for the peer** (ADR-0022 amendment). `Close(id)` runs the closing + handshake, which against a peer that has gone silent waits out + `request_timeout_seconds` (30s by default) while the id stays taken. + `Kick` puts a close code and reason on the frame and tears the + connection down after at most `grace`. The same call is + `EventStreamHandle::Kick` and `WebSocket::Kick`; + `WebSocket::PeerCloseReason()` reads the close the peer ended a session + with, so a client told 4001 "superseded" can stop redialing. The base + `WebSocket` defaults keep other implementations compiling: `Kick` is + `Close()`, `PeerCloseReason()` reads nothing. A code other than 1000 or + 3000-4999, or a reason over 123 bytes, fails fast. + - **`Observe` labels requests from their headers** (issue #235). An optional fifth argument, `RequestLabeler`, maps a request's headers to `RequestLabels` once, before dispatch; the result rides on both @@ -262,6 +275,14 @@ policy in [docs/versioning.md](docs/versioning.md). ### Fixed +- **A handle's `Close` no longer deadlocks when the transport ends the + loop inline.** The in-memory pair completes a parked receive on the + closing thread, so a hub calling `registry.Close(id)` on a session + parked in `co_await Receive()` destroyed the stream inside that call, + and the stream's teardown waited forever for the pin the call held. A + handle minted by `AsyncEventStream` now keeps the socket alive by + ownership for `Close` and `Kick` rather than by a pin. + - **A dialed WebSocket released on its own io thread no longer terminates the process.** When the last handle to a `BeastWebSocketClient::Dial` socket dropped inside one of its own completions (typically a `Detached` diff --git a/docs/adr/0022-registry-admission-primitives.md b/docs/adr/0022-registry-admission-primitives.md index fa51e3d1..9b874854 100644 --- a/docs/adr/0022-registry-admission-primitives.md +++ b/docs/adr/0022-registry-admission-primitives.md @@ -4,7 +4,8 @@ and ADR-0021 review panels twice deferred. Implemented: `SessionRegistry::{Admission, ResumeOrAdd, Close}`, the three example admission loops deleted, the production guide's canonical block replaced -by the one call. +by the one call. Amended 2026-10-13: `Close` does not bound a silent +peer, so the kick is `Kick` (below). ## Context @@ -53,6 +54,53 @@ convergent path instead of waiting out TCP. On a detached entry it closes an already-closed handle: a harmless no-op — detached entries are `Resume`'s and `Remove`'s business. +## Amendment (2026-10-13): Kick + +`Close` was meant to be the kick for "a silent partition … instead of +waiting out TCP", and on Beast it is not one. Its close is the graceful +handshake: send the close frame, then read until the peer's close +arrives, bounded only by Beast's handshake timeout — the transport's +`request_timeout_seconds`, 30s by default. A peer that is gone never +answers, so the old handler holds the id for that long, and a +`ResumeOrAdd` behind it is refused the whole time. games_hub found it: +a phone that dropped off the network kept its player locked out of a +chess game for half a minute after it came back. + +**`Kick(id, why, grace) -> bool`** is the kick. It ends the session like +`Close`, with two differences, both carried by new `WebSocket` members +(every implementation in the repo overrides them; the base defaults to +`Close()` and reads nothing): + +- **`WebSocket::Kick(CloseReason why, grace)`** puts `why` — a close code + and reason — on the close frame and waits at most `grace` for the + peer's answer before tearing the connection down. On Beast, `grace` is + the handshake timeout for that one close; a kick that finds Close's + handshake already running closes the socket under it. A peer that is + still there reads the code; one that is gone costs `grace`. +- **`WebSocket::PeerCloseReason()`** is the close frame the peer ended the + session with. With an application code (4000-4999) the server can say + *why* it dropped a client — "superseded by a newer connection" — and a + client can stop redialing instead of taking its seat back, which is what + keeps two connections under one identity from trading it forever. + +`EventStreamHandle::Kick` is the handle's form. Codes outside 1000 and +3000-4999, or a reason over 123 bytes, fail fast (ADR-0009): the +protocol's own codes are the transport's to send. + +`Close` keeps its meaning — the polite end of a session whose peer is +expected to answer — and its doc no longer claims the silent-partition +case. + +A handle's `Close` and `Kick` also stopped pinning a socket the stream +owns. A transport may complete the parked receive inline on the closing +thread — the in-memory pair does, as its contract allows — and the loop +then ends and destroys its stream inside the handle's call, where the +stream's teardown waited for the pin that call held. Any hub calling +`Close(id)` on a pair-backed session parked in `Receive` deadlocked. The +coroutine adapter owns its socket, so the handle now holds a strong +reference across the call instead; a borrowed socket is still pinned, and +a borrowing stream is never torn down by its own socket's completions. + ## Non-goals An async admission shape (`co_await ResumeOrAdd`) — admission runs @@ -67,6 +115,6 @@ Changing `Resume`/`Add` themselves: the primitives compose unchanged. Consumers write one call and branch on the enum; the retry cadence and ordering subtleties live (and are tested) in exactly one place. The guide teaches the call instead of the loop. The silent-partition story -gains its missing move. The registry's public surface grows two +gains its missing move: `Kick`, since the amendment above. The registry's public surface grows two methods and one enum, all shallow compositions of existing primitives — no new locking, no new threads, no new states. diff --git a/docs/production-guide.md b/docs/production-guide.md index 63295c8a..d82bc847 100644 --- a/docs/production-guide.md +++ b/docs/production-guide.md @@ -1070,13 +1070,29 @@ deadline (the old hand-rolled recipe's 20 × 50ms) comfortably covers the failure-notice race. The call blocks up to the deadline — legal because admission runs before the handler's first suspension, on the launching thread. A `kRefused` you -*know* is wrong — a half-dead session whose wire never sent a FIN — has a -convergent answer now: `registry.Close(id)` kicks the old session — its +*know* is wrong — a half-dead session whose wire never sent a FIN, or an +older connection the same identity has replaced — has a convergent +answer: `registry.Kick(id, why, grace)` kicks the old session — its handler observes the close and runs the normal exit path, so the id is admittable on the next dial (freed outright after a Remove exit; parked-resumable after a Detach exit, where the redial resumes with the -old identity and gets the snapshot). Kicking stays the application's call; -`ResumeOrAdd` never does it on its own. +old identity and gets the snapshot). `grace` bounds the wait for a peer +that never answers; `Close(id)` would wait out the whole closing +handshake (`request_timeout_seconds`). Put an application close code in +`why` (4000-4999) so a kicked client that is still there can tell it was +superseded and stop redialing — `PeerCloseReason()` on a native client, +`CloseEvent.code` in a browser: + +```cpp +auto admission = registry.ResumeOrAdd(id, mint, std::chrono::milliseconds(0)); +if (admission == Registry::Admission::kRefused) { + registry.Kick(id, {.code = 4001, .reason = "superseded"}, std::chrono::milliseconds(250)); + admission = registry.ResumeOrAdd(id, mint, std::chrono::seconds(1)); +} +``` + +Kicking stays the application's call; `ResumeOrAdd` never does it on its +own. Say the posture out loud in your protocol docs, because it shapes client code: **recovery is snapshot replay, not message replay.** ADR-0016's diff --git a/docs/runtime.md b/docs/runtime.md index a3ddf27a..c5cf76ac 100644 --- a/docs/runtime.md +++ b/docs/runtime.md @@ -17,7 +17,7 @@ crates (PLAN §3.2a). | `//runtime:http_beast` | `opal::http` | `BeastServerTransport` (ADR-0006): the production server transport on BCR modular Boost.Beast/asio — concurrent connections on a thread pool, keep-alive, per-connection timeouts, body- and header-size limits, graceful drain on Stop. WebSocket upgrades ride the same transport (ADR-0015): `Options::on_websocket`/`websocket_gate` serve blocking full-duplex `opal::http::WebSocket` sessions carrying event-stream messages, and `BeastWebSocketClient::Dial` is the client end. `Options::websocket_accept_json_frames` negotiates the ADR-0018 JSON-text wire for clients offering the subprotocol (browsers), transparently to everything above the session; `Options::websocket_raw_text_frames` flips the whole listener to the unnegotiated raw-text wire (ADR-0023: headerless messages as verbatim text frames, one JSON-RPC envelope each — what a generated jsonRpc2 streaming server mounts), with `WebSocketDialRequest::raw_text_frames` as the client end. Separate target so Boost stays out of dep-light builds | | `//runtime:client` | `opal` | `ClientConfig` (endpoint, timeout, user-agent, transport injection, `RetryPolicy`, request-compression threshold, `Interceptor` hooks around every attempt, `bearer_token`/`api_key` credential providers), `SendWithRetries` (full-jitter exponential backoff over transport errors and 429/5xx), `ObserveAttempts` + `PropagateTraceContext` interceptors — see docs/production-guide.md | | `//runtime:compression` | `opal` | `GzipCompress`/`GzipDecompress` (zlib; decompression-bomb guard, trailing-garbage rejection) backing `@requestCompression` | -| `//runtime:server` | `opal::server` | `Router` (literal > label > greedy precedence, 404/405/400), `RequestContext`, `MakeErrorResponse`, `ValidationFailure`, user-supplied `Middleware` + `Chain`, the `Observe` logging/metrics hook, `RequireBearerAuth`/`RequireApiKeyHeader` guards, `WebSocketRouter` — the streaming routes' method + URI-pattern dispatch behind generated `StreamRouter()`, sharing `Router`'s matching by construction (ADR-0016), plus the application-facing shared-seam mount `AddSession`/`ServeSession()` for `on_websocket_session` (ADR-0019, issue #118; one router serves one seam) — `RequireOrigin`, the browser-facing Origin-allowlist `websocket_gate` (ADR-0018) — and `SessionRegistry` — the multi-client fan-out helper over owning `EventStream::Share()` handles: bounded per-session outbound queues, non-blocking `SendTo`/`Broadcast` with per-recipient construction, a slow-consumer policy, and `Drain` for graceful shutdown (ADR-0017); `Options::async_delivery` swaps writer threads for `SendAsync` completion chains (ADR-0019); `Detach`/`Resume` + `Options::grace_period` give abrupt disconnects a reconnect window with exactly-once expiry cleanup (ADR-0020); `ResumeOrAdd` is the blessed admission call and `Close(id)` the kick primitive (ADR-0022) | +| `//runtime:server` | `opal::server` | `Router` (literal > label > greedy precedence, 404/405/400), `RequestContext`, `MakeErrorResponse`, `ValidationFailure`, user-supplied `Middleware` + `Chain`, the `Observe` logging/metrics hook, `RequireBearerAuth`/`RequireApiKeyHeader` guards, `WebSocketRouter` — the streaming routes' method + URI-pattern dispatch behind generated `StreamRouter()`, sharing `Router`'s matching by construction (ADR-0016), plus the application-facing shared-seam mount `AddSession`/`ServeSession()` for `on_websocket_session` (ADR-0019, issue #118; one router serves one seam) — `RequireOrigin`, the browser-facing Origin-allowlist `websocket_gate` (ADR-0018) — and `SessionRegistry` — the multi-client fan-out helper over owning `EventStream::Share()` handles: bounded per-session outbound queues, non-blocking `SendTo`/`Broadcast` with per-recipient construction, a slow-consumer policy, and `Drain` for graceful shutdown (ADR-0017); `Options::async_delivery` swaps writer threads for `SendAsync` completion chains (ADR-0019); `Detach`/`Resume` + `Options::grace_period` give abrupt disconnects a reconnect window with exactly-once expiry cleanup (ADR-0020); `ResumeOrAdd` is the blessed admission call and `Kick(id, why, grace)` the kick primitive (ADR-0022) | ## Design rules diff --git a/examples/bazel-consumer/websocket_acceptance_test.cc b/examples/bazel-consumer/websocket_acceptance_test.cc index b97a448c..0487d86b 100644 --- a/examples/bazel-consumer/websocket_acceptance_test.cc +++ b/examples/bazel-consumer/websocket_acceptance_test.cc @@ -140,4 +140,29 @@ TEST(WebSocketAcceptanceTest, AConsumerBoundsAReceiveAndKeepsTheSession) { server.Stop(); } +// A server dropping a client it has replaced: the kick names why, and the +// consumer's client reads the code off the close — how it knows to stop +// redialing rather than take the identity back. +TEST(WebSocketAcceptanceTest, AKickedClientReadsWhyItWasDropped) { + BeastServerTransport::Options options; + options.on_websocket = [](const HttpRequest&, WebSocket& socket) { + socket.Kick({.code = 4001, .reason = "superseded"}, std::chrono::milliseconds(250)); + while (true) { + auto message = socket.Receive(); + if (!message.ok() || !message->has_value()) return; + } + }; + BeastServerTransport server(options); + ASSERT_TRUE(server.Start([](const opal::http::HttpRequest&) { return HttpResponse{}; }).ok()); + + auto dialed = BeastWebSocketClient::Dial({.host = "127.0.0.1", .port = server.port()}); + ASSERT_TRUE(dialed.ok()) << dialed.error().message(); + auto ended = (*dialed)->Receive(std::chrono::seconds(5)); + ASSERT_TRUE(ended.ok()) << ended.error().message(); + EXPECT_FALSE(ended->has_value()); + EXPECT_EQ((*dialed)->PeerCloseReason(), + (opal::http::CloseReason{.code = 4001, .reason = "superseded"})); + server.Stop(); +} + } // namespace diff --git a/runtime/include/opal/server/session_registry.h b/runtime/include/opal/server/session_registry.h index fd31f2d9..2ed1b1eb 100644 --- a/runtime/include/opal/server/session_registry.h +++ b/runtime/include/opal/server/session_registry.h @@ -106,7 +106,7 @@ struct DeliveryClass { // expire detached entries immediately. One lazy expiry thread serves the // whole registry — a detached session itself holds zero threads. // ResumeOrAdd is the blessed admission call (ADR-0022): resume-or-fresh- -// join with the brief retry the reconnect race needs, and Close(id) is +// join with the brief retry the reconnect race needs, and Kick(id) is // the kick primitive that makes its refusal actionable. // // All methods are safe from any thread, including a handler's own (a From 9069d774a7ec232efda4982cc0e1cbdfec7527a3 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 13 Oct 2026 19:15:57 +0000 Subject: [PATCH 3/7] Lockfiles: resolve under Bazel 9.3.0 --- MODULE.bazel.lock | 14 ++++++++------ examples/bazel-consumer/MODULE.bazel.lock | 10 ++++++---- 2 files changed, 14 insertions(+), 10 deletions(-) diff --git a/MODULE.bazel.lock b/MODULE.bazel.lock index 882e78c5..0ad626e2 100644 --- a/MODULE.bazel.lock +++ b/MODULE.bazel.lock @@ -24,6 +24,7 @@ "https://bcr.bazel.build/modules/apple_support/1.21.1/MODULE.bazel": "5809fa3efab15d1f3c3c635af6974044bac8a4919c62238cce06acee8a8c11f1", "https://bcr.bazel.build/modules/apple_support/1.22.1/MODULE.bazel": "90bd1a660590f3ceffbdf524e37483094b29352d85317060b2327fff8f3f4458", "https://bcr.bazel.build/modules/apple_support/1.24.2/MODULE.bazel": "0e62471818affb9f0b26f128831d5c40b074d32e6dda5a0d3852847215a41ca4", + "https://bcr.bazel.build/modules/apple_support/2.5.2/MODULE.bazel": "bb1b7081e390a8d3885f642cb0073dfcf8e7a36525e079fa8f5835af7079c0fe", "https://bcr.bazel.build/modules/apple_support/2.8.0/MODULE.bazel": "c45f5176057092afba0be6c48a9ea02101666802f8be536947e099e94757f8ed", "https://bcr.bazel.build/modules/apple_support/2.8.0/source.json": "b7f6d87669c206adaa7d82cc44733349c63e341b80a0e049496ebd91d78c3472", "https://bcr.bazel.build/modules/bazel_features/1.1.0/MODULE.bazel": "cfd42ff3b815a5f39554d97182657f8c4b9719568eb7fded2b9135f084bf760b", @@ -302,6 +303,7 @@ "https://bcr.bazel.build/modules/rules_cc/0.2.0/MODULE.bazel": "b5c17f90458caae90d2ccd114c81970062946f49f355610ed89bebf954f5783c", "https://bcr.bazel.build/modules/rules_cc/0.2.13/MODULE.bazel": "eecdd666eda6be16a8d9dc15e44b5c75133405e820f620a234acc4b1fdc5aa37", "https://bcr.bazel.build/modules/rules_cc/0.2.14/MODULE.bazel": "353c99ed148887ee89c54a17d4100ae7e7e436593d104b668476019023b58df8", + "https://bcr.bazel.build/modules/rules_cc/0.2.15/MODULE.bazel": "6a0a4a75a57aa6dc888300d848053a58c6b12a29f89d4304e1c41448514ec6e8", "https://bcr.bazel.build/modules/rules_cc/0.2.16/MODULE.bazel": "9242fa89f950c6ef7702801ab53922e99c69b02310c39fb6e62b2bd30df2a1d4", "https://bcr.bazel.build/modules/rules_cc/0.2.17/MODULE.bazel": "1849602c86cb60da8613d2de887f9566a6d354a6df6d7009f9d04a14402f9a84", "https://bcr.bazel.build/modules/rules_cc/0.2.18/MODULE.bazel": "4460ec36adc8f722a6a2a4ac9374cb91f2acebadaa93fc37966129afb3dece87", @@ -445,7 +447,7 @@ "moduleExtensions": { "@@rules_android+//bzlmod_extensions:apksig.bzl%apksig_extension": { "general": { - "bzlTransitiveDigest": "qEcqhUuyYlPmvAtyYRqS+4Y7cRHA2Vho3JIBJZJJX+s=", + "bzlTransitiveDigest": "0A8O20aQTIZnt20CyMK1wzktVUyJO1Y3jI6cS1WftKQ=", "usagesDigest": "zr/niBQ/s2fHozWAsg4vI70wAxcuFjG+QtM15qGkq9o=", "recordedInputs": [ "REPO_MAPPING:rules_android+,bazel_tools bazel_tools" @@ -463,7 +465,7 @@ }, "@@rules_android+//bzlmod_extensions:com_android_dex.bzl%com_android_dex_extension": { "general": { - "bzlTransitiveDigest": "gTuOqknGgqPw9YQSCY+qXxgxVwiLRGdBCsCe4i4q/xw=", + "bzlTransitiveDigest": "stTLqB1DjqvC1RcosQO9gDBmAlvTegb/9MYDyiRNuFo=", "usagesDigest": "c1Y/KGGjUYCyd8zNIVTUh1bynVXRFz6xGKaSCBpQANM=", "recordedInputs": [ "REPO_MAPPING:rules_android+,bazel_tools bazel_tools" @@ -494,7 +496,7 @@ }, "@@rules_python+//python/extensions:config.bzl%config": { "general": { - "bzlTransitiveDigest": "vb5YWBnUdt851q2e90djwEpUSyvTyGZpeHryMr1NJVM=", + "bzlTransitiveDigest": "sJuhZihyVty+0xOVsXH1eSSGv2ufRPo9x8QCFFNs90s=", "usagesDigest": "EocbSr4I3/Shk4QaFool8b8navUiUFqgzF9bOaaYfFk=", "recordedInputs": [ "REPO_MAPPING:rules_python+,bazel_tools bazel_tools", @@ -662,7 +664,7 @@ }, "@@rules_python+//python/uv:uv.bzl%uv": { "general": { - "bzlTransitiveDigest": "ijW9KS7qsIY+yBVvJ+Nr1mzwQox09j13DnE3iIwaeTM=", + "bzlTransitiveDigest": "IgkgxNrtlAec7OnaFrWoCC5NQhHcY/aIz4jsqcNMm8Y=", "usagesDigest": "yXvWfXAzpBeW71mWgwU3AqAzXX/dFACnx12eYvBsJ8w=", "recordedInputs": [ "REPO_MAPPING:rules_python+,bazel_tools bazel_tools", @@ -692,7 +694,7 @@ }, "@@windows_support+//windows:extensions.bzl%msvc_runtime": { "general": { - "bzlTransitiveDigest": "yv3gSwY6QneXA6YvHLn9tSmuOkNp2EPN5NFXHC48mdE=", + "bzlTransitiveDigest": "/fpjACapwytnUWPsK9hfNxq/3i0vYEbGgvEFP3etVEA=", "usagesDigest": "fe0w49qOZpnzoEX7GmAfDXDCO0BbgNb1z0LOhMxLrgo=", "recordedInputs": [ "REPO_MAPPING:windows_support+,bazel_tools bazel_tools" @@ -723,7 +725,7 @@ }, "@@windows_support+//windows:extensions.bzl%windows_sdk": { "general": { - "bzlTransitiveDigest": "yv3gSwY6QneXA6YvHLn9tSmuOkNp2EPN5NFXHC48mdE=", + "bzlTransitiveDigest": "/fpjACapwytnUWPsK9hfNxq/3i0vYEbGgvEFP3etVEA=", "usagesDigest": "dSf/ZB7rdHRJw0y/SOPpQL1YYGpndOvmCQnHWzuEhvY=", "recordedInputs": [ "REPO_MAPPING:windows_support+,bazel_tools bazel_tools" diff --git a/examples/bazel-consumer/MODULE.bazel.lock b/examples/bazel-consumer/MODULE.bazel.lock index b78c5307..a3df9919 100644 --- a/examples/bazel-consumer/MODULE.bazel.lock +++ b/examples/bazel-consumer/MODULE.bazel.lock @@ -24,6 +24,7 @@ "https://bcr.bazel.build/modules/apple_support/1.21.1/MODULE.bazel": "5809fa3efab15d1f3c3c635af6974044bac8a4919c62238cce06acee8a8c11f1", "https://bcr.bazel.build/modules/apple_support/1.22.1/MODULE.bazel": "90bd1a660590f3ceffbdf524e37483094b29352d85317060b2327fff8f3f4458", "https://bcr.bazel.build/modules/apple_support/1.24.2/MODULE.bazel": "0e62471818affb9f0b26f128831d5c40b074d32e6dda5a0d3852847215a41ca4", + "https://bcr.bazel.build/modules/apple_support/2.5.2/MODULE.bazel": "bb1b7081e390a8d3885f642cb0073dfcf8e7a36525e079fa8f5835af7079c0fe", "https://bcr.bazel.build/modules/apple_support/2.8.0/MODULE.bazel": "c45f5176057092afba0be6c48a9ea02101666802f8be536947e099e94757f8ed", "https://bcr.bazel.build/modules/apple_support/2.8.0/source.json": "b7f6d87669c206adaa7d82cc44733349c63e341b80a0e049496ebd91d78c3472", "https://bcr.bazel.build/modules/bazel_features/1.1.0/MODULE.bazel": "cfd42ff3b815a5f39554d97182657f8c4b9719568eb7fded2b9135f084bf760b", @@ -290,6 +291,7 @@ "https://bcr.bazel.build/modules/rules_cc/0.2.0/MODULE.bazel": "b5c17f90458caae90d2ccd114c81970062946f49f355610ed89bebf954f5783c", "https://bcr.bazel.build/modules/rules_cc/0.2.13/MODULE.bazel": "eecdd666eda6be16a8d9dc15e44b5c75133405e820f620a234acc4b1fdc5aa37", "https://bcr.bazel.build/modules/rules_cc/0.2.14/MODULE.bazel": "353c99ed148887ee89c54a17d4100ae7e7e436593d104b668476019023b58df8", + "https://bcr.bazel.build/modules/rules_cc/0.2.15/MODULE.bazel": "6a0a4a75a57aa6dc888300d848053a58c6b12a29f89d4304e1c41448514ec6e8", "https://bcr.bazel.build/modules/rules_cc/0.2.16/MODULE.bazel": "9242fa89f950c6ef7702801ab53922e99c69b02310c39fb6e62b2bd30df2a1d4", "https://bcr.bazel.build/modules/rules_cc/0.2.17/MODULE.bazel": "1849602c86cb60da8613d2de887f9566a6d354a6df6d7009f9d04a14402f9a84", "https://bcr.bazel.build/modules/rules_cc/0.2.18/MODULE.bazel": "4460ec36adc8f722a6a2a4ac9374cb91f2acebadaa93fc37966129afb3dece87", @@ -418,7 +420,7 @@ "moduleExtensions": { "@@rules_android+//bzlmod_extensions:apksig.bzl%apksig_extension": { "general": { - "bzlTransitiveDigest": "qEcqhUuyYlPmvAtyYRqS+4Y7cRHA2Vho3JIBJZJJX+s=", + "bzlTransitiveDigest": "0A8O20aQTIZnt20CyMK1wzktVUyJO1Y3jI6cS1WftKQ=", "usagesDigest": "zr/niBQ/s2fHozWAsg4vI70wAxcuFjG+QtM15qGkq9o=", "recordedInputs": [ "REPO_MAPPING:rules_android+,bazel_tools bazel_tools" @@ -436,7 +438,7 @@ }, "@@rules_android+//bzlmod_extensions:com_android_dex.bzl%com_android_dex_extension": { "general": { - "bzlTransitiveDigest": "gTuOqknGgqPw9YQSCY+qXxgxVwiLRGdBCsCe4i4q/xw=", + "bzlTransitiveDigest": "stTLqB1DjqvC1RcosQO9gDBmAlvTegb/9MYDyiRNuFo=", "usagesDigest": "c1Y/KGGjUYCyd8zNIVTUh1bynVXRFz6xGKaSCBpQANM=", "recordedInputs": [ "REPO_MAPPING:rules_android+,bazel_tools bazel_tools" @@ -467,7 +469,7 @@ }, "@@rules_python+//python/extensions:config.bzl%config": { "general": { - "bzlTransitiveDigest": "dzD8Q2YmrP3fz8saWLHPmlwPLO91ImtTmP/c9JKTStM=", + "bzlTransitiveDigest": "L+2ZCpb7GIJqvWaXpzkfk/80d916sIFVmSPzrSm6of8=", "usagesDigest": "ZVSXMAGpD+xzVNPuvF1IoLBkty7TROO0+akMapt1pAg=", "recordedInputs": [ "REPO_MAPPING:rules_python+,bazel_tools bazel_tools", @@ -635,7 +637,7 @@ }, "@@rules_python+//python/uv:uv.bzl%uv": { "general": { - "bzlTransitiveDigest": "ijW9KS7qsIY+yBVvJ+Nr1mzwQox09j13DnE3iIwaeTM=", + "bzlTransitiveDigest": "IgkgxNrtlAec7OnaFrWoCC5NQhHcY/aIz4jsqcNMm8Y=", "usagesDigest": "H8dQoNZcoqP+Mu0tHZTi4KHATzvNkM5ePuEqoQdklIU=", "recordedInputs": [ "REPO_MAPPING:rules_python+,bazel_tools bazel_tools", From 92ebe1378d7f8891d03400d4bb9c2ac399a34fdc Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 13 Oct 2026 19:34:17 +0000 Subject: [PATCH 4/7] Kick arms its own deadline; handles hold owned sockets --- CHANGELOG.md | 15 +-- .../adr/0022-registry-admission-primitives.md | 36 +++--- runtime/BUILD.bazel | 1 + .../include/opal/eventstream/event_stream.h | 114 +++++++++--------- runtime/include/opal/http/websocket.h | 5 +- .../include/opal/server/session_registry.h | 6 +- runtime/src/http/beast_transport.cc | 46 ++++--- .../eventstream/async_event_stream_test.cc | 74 +++++++++--- .../tests/eventstream/event_stream_test.cc | 2 +- runtime/tests/http/beast_websocket_test.cc | 92 +++++++++++++- 10 files changed, 255 insertions(+), 136 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 082d0bb6..d946f1b7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -275,13 +275,14 @@ policy in [docs/versioning.md](docs/versioning.md). ### Fixed -- **A handle's `Close` no longer deadlocks when the transport ends the - loop inline.** The in-memory pair completes a parked receive on the - closing thread, so a hub calling `registry.Close(id)` on a session - parked in `co_await Receive()` destroyed the stream inside that call, - and the stream's teardown waited forever for the pin the call held. A - handle minted by `AsyncEventStream` now keeps the socket alive by - ownership for `Close` and `Kick` rather than by a pin. +- **A handle call no longer deadlocks when the transport ends the loop + inline.** The in-memory pair completes parked operations on the calling + thread, so a hub calling `registry.Close(id)` on a session parked in + `co_await Receive()` — or delivering to a peer that answered by closing + — destroyed the stream inside that call, and the stream's teardown + waited forever for the pin the call held. A handle minted by + `AsyncEventStream` now keeps the socket alive by ownership rather than + by a pin, for every operation. - **A dialed WebSocket released on its own io thread no longer terminates the process.** When the last handle to a `BeastWebSocketClient::Dial` diff --git a/docs/adr/0022-registry-admission-primitives.md b/docs/adr/0022-registry-admission-primitives.md index 9b874854..9efe79f2 100644 --- a/docs/adr/0022-registry-admission-primitives.md +++ b/docs/adr/0022-registry-admission-primitives.md @@ -68,15 +68,20 @@ chess game for half a minute after it came back. **`Kick(id, why, grace) -> bool`** is the kick. It ends the session like `Close`, with two differences, both carried by new `WebSocket` members -(every implementation in the repo overrides them; the base defaults to -`Close()` and reads nothing): +(the runtime's transports — Beast, the in-memory pair, the dialed +client, the JSON-RPC wrapper — override them; the base defaults to +`Close()` and reads nothing, which is what test fakes and the consumer +example's socket get): - **`WebSocket::Kick(CloseReason why, grace)`** puts `why` — a close code and reason — on the close frame and waits at most `grace` for the - peer's answer before tearing the connection down. On Beast, `grace` is - the handshake timeout for that one close; a kick that finds Close's - handshake already running closes the socket under it. A peer that is - still there reads the code; one that is gone costs `grace`. + peer's answer before tearing the connection down. On Beast the kick + arms its own deadline that closes the socket `grace` later, whatever + the close is waiting on — the peer's answer, a keep-alive frame + holding Beast's write lock, a Close's handshake already running, or a + close deferred behind a wedged write. Only the first kick arms it, so a + second neither re-arms it nor cuts the first one's frame short. A peer + that is still there reads the code; one that is gone costs `grace`. - **`WebSocket::PeerCloseReason()`** is the close frame the peer ended the session with. With an application code (4000-4999) the server can say *why* it dropped a client — "superseded by a newer connection" — and a @@ -91,15 +96,16 @@ protocol's own codes are the transport's to send. expected to answer — and its doc no longer claims the silent-partition case. -A handle's `Close` and `Kick` also stopped pinning a socket the stream -owns. A transport may complete the parked receive inline on the closing -thread — the in-memory pair does, as its contract allows — and the loop -then ends and destroys its stream inside the handle's call, where the -stream's teardown waited for the pin that call held. Any hub calling -`Close(id)` on a pair-backed session parked in `Receive` deadlocked. The -coroutine adapter owns its socket, so the handle now holds a strong -reference across the call instead; a borrowed socket is still pinned, and -a borrowing stream is never torn down by its own socket's completions. +A handle's operations also stopped pinning a socket the stream owns. A +transport may complete operations inline on the calling thread — the +in-memory pair does, as its contract allows — and the loop then ends and +destroys its stream inside the handle's call, where the stream's teardown +waited for the pin that call held. Any hub calling `Close(id)` on a +pair-backed session parked in `Receive` deadlocked, and so did a delivery +the peer answered by closing. The coroutine adapter owns its socket, so a +handle now holds a strong reference across each call (to completion, for +`SendAsync`) instead; a borrowed socket is still pinned, and a borrowing +stream is never torn down by its own socket's completions. ## Non-goals diff --git a/runtime/BUILD.bazel b/runtime/BUILD.bazel index e008e7ac..e6a50527 100644 --- a/runtime/BUILD.bazel +++ b/runtime/BUILD.bazel @@ -470,6 +470,7 @@ cc_test( ":eventstream_json", ":http", ":http_beast", + ":server", ":test_tls_identity", ":websocket_contract_test_support", # The hostile raw-websocket peer speaks Beast directly, for the diff --git a/runtime/include/opal/eventstream/event_stream.h b/runtime/include/opal/eventstream/event_stream.h index 817b902b..5a628e98 100644 --- a/runtime/include/opal/eventstream/event_stream.h +++ b/runtime/include/opal/eventstream/event_stream.h @@ -53,9 +53,10 @@ struct SharedSessionState { std::mutex mutex; std::condition_variable idle; http::WebSocket* socket = nullptr; // null once the stream ended - int active = 0; // handle operations inside socket calls - // Engaged when the stream owns its socket (the coroutine adapter): what - // lets a handle end the session without a pin (EventStreamHandle::End). + int active = 0; // pinned handle operations inside socket calls + // Engaged when the stream owns its socket (the coroutine adapter): handle + // operations then hold it by reference instead of a pin + // (EventStreamHandle::Acquire). std::weak_ptr owner; }; @@ -102,8 +103,10 @@ class SharedViewOwner { // Revoke-then-drain: null the pointer so no new handle operation starts, // close the socket so any operation already inside a call unblocks, wait - // for the in-flight count to drain. After this returns no handle can - // reach the socket again, so the borrow (or owned socket) may die. + // for the pinned count to drain. After this returns no handle starts an + // operation and none is inside one on a borrowed socket, so the borrow + // may die; one still inside a call on an owned socket holds its own + // reference. void End() { if (state_ == nullptr) return; const std::shared_ptr state = std::move(state_); @@ -131,7 +134,8 @@ class SharedViewOwner { // instead of borrowed references with manual remove-before-every-return // discipline. The handle never extends the session: once the stream that // minted it is gone (for a server handler, when the handler machinery -// returns), Send fails with Error::Transport — exactly what a closed stream +// returns — a call already under way finishes first), Send fails with +// Error::Transport — exactly what a closed stream // returns today — and Close is a no-op, so a stale handle has no new // failure modes, it just misses. (A moved-from handle behaves like a stale // one.) @@ -154,10 +158,10 @@ class EventStreamHandle { "receive-only stream's handle (use Close)"); auto message = encode_(event); if (!message.ok()) return std::move(message).error(); - http::WebSocket* socket = Acquire(); - if (socket == nullptr) return Error::Transport("event stream ended"); - auto sent = socket->Send(*message); - Release(); + const Hold hold = Acquire(); + if (hold.socket == nullptr) return Error::Transport("event stream ended"); + auto sent = hold.socket->Send(*message); + Release(hold); return sent; } @@ -166,7 +170,10 @@ class EventStreamHandle { // session from outside the handler (the slow-consumer policy): the // owner's blocked Receive/Send unblock exactly as for any other close. void Close() const { - End([](http::WebSocket& socket) { socket.Close(); }); + const Hold hold = Acquire(); + if (hold.socket == nullptr) return; + hold.socket->Close(); + Release(hold); } // The socket's Kick (opal/http/websocket.h): Close for a peer the caller @@ -174,7 +181,10 @@ class EventStreamHandle { // waiting for an answer. A no-op once the session ended. void Kick(const http::CloseReason& why, std::chrono::milliseconds grace) const { http::CheckCloseReason(why); - End([&why, grace](http::WebSocket& socket) { socket.Kick(why, grace); }); + const Hold hold = Acquire(); + if (hold.socket == nullptr) return; + hold.socket->Kick(why, grace); + Release(hold); } // The completion-driven send (ADR-0019): the socket's SendAsync contract @@ -192,26 +202,27 @@ class EventStreamHandle { callback(std::move(message).error()); return; } - http::WebSocket* socket = Acquire(); - if (socket == nullptr) { + const Hold hold = Acquire(); + if (hold.socket == nullptr) { callback(Error::Transport("event stream ended")); return; } - socket->SendAsync(*message, - [state = state_, callback = std::move(callback)](Outcome sent) { - internal::ReleasePin(*state); - callback(std::move(sent)); - }); + // The hold spans issue to completion. + hold.socket->SendAsync(*message, [state = state_, owned = hold.owned, + callback = std::move(callback)](Outcome sent) { + if (owned == nullptr) internal::ReleasePin(*state); + callback(std::move(sent)); + }); } // Whether the underlying session implements the async operations — // false for a session that already ended (a stale handle can deliver // nothing either way). bool SupportsAsync() const { - http::WebSocket* socket = Acquire(); - if (socket == nullptr) return false; - const bool supported = socket->SupportsAsync(); - Release(); + const Hold hold = Acquire(); + if (hold.socket == nullptr) return false; + const bool supported = hold.socket->SupportsAsync(); + Release(hold); return supported; } @@ -225,45 +236,30 @@ class EventStreamHandle { std::function(const Tx&)> encode) : state_(std::move(state)), encode_(std::move(encode)) {} - // The socket, pinned against revocation until Release — or null once the - // session ended (or this handle was moved from). - http::WebSocket* Acquire() const { - if (state_ == nullptr) return nullptr; + // The socket for one handle call, or none once the session ended (or + // this handle was moved from). A socket the stream owns is held by + // reference, never pinned: a transport may complete operations inline on + // the calling thread (the in-memory pair does), ending the loop and + // destroying the stream inside the call, and the stream's teardown waits + // for every pin — this frame's included. A borrowed socket is pinned + // until Release; a borrowing stream is never torn down by its own + // socket's completions. + struct Hold { + std::shared_ptr owned; + http::WebSocket* socket = nullptr; + }; + + Hold Acquire() const { + if (state_ == nullptr) return {}; const std::scoped_lock lock(state_->mutex); - if (state_->socket == nullptr) return nullptr; - ++state_->active; - return state_->socket; + if (state_->socket == nullptr) return {}; + Hold hold{.owned = state_->owner.lock(), .socket = state_->socket}; + if (hold.owned == nullptr) ++state_->active; + return hold; } - void Release() const { internal::ReleasePin(*state_); } - - // Close and Kick. A transport may complete the parked receive inline on - // the closing thread (the in-memory pair does), ending the loop and - // destroying the stream inside this very call — and the stream's - // teardown waits for every pin, this frame's included. So a socket the - // stream owns is kept alive by ownership instead of a pin; only a - // borrowed one is pinned, and a borrowing stream is never torn down by - // its own socket's completions. - template - void End(const Ending& ending) const { - if (state_ == nullptr) return; - std::shared_ptr owned; - http::WebSocket* borrowed = nullptr; - { - const std::scoped_lock lock(state_->mutex); - if (state_->socket == nullptr) return; - owned = state_->owner.lock(); - if (owned == nullptr) { - ++state_->active; - borrowed = state_->socket; - } - } - if (owned != nullptr) { - ending(*owned); - return; - } - ending(*borrowed); - Release(); + void Release(const Hold& hold) const { + if (hold.owned == nullptr) internal::ReleasePin(*state_); } std::shared_ptr state_; diff --git a/runtime/include/opal/http/websocket.h b/runtime/include/opal/http/websocket.h index d1a551af..ecc60e2b 100644 --- a/runtime/include/opal/http/websocket.h +++ b/runtime/include/opal/http/websocket.h @@ -149,8 +149,9 @@ class WebSocket { // is Close()'s contract: idempotent, non-blocking, safe from any thread, // and it unblocks and completes what is parked — a peer that does not // answer in time ends the session with an error rather than a clean - // nullopt. On a session already closing it stops waiting for the peer - // and the frame already sent stands. `why.code` must be 1000 or in + // nullopt. On a session already closing — a Close's handshake, a close + // deferred behind a wedged write — the requested frame stands and the + // kick only bounds the wait; a second kick changes nothing. `why.code` must be 1000 or in // 3000-4999 and `why.reason` at most 123 bytes (RFC 6455 §5.5, §7.4); // anything else is a contract violation and fails fast (ADR-0009). // diff --git a/runtime/include/opal/server/session_registry.h b/runtime/include/opal/server/session_registry.h index 2ed1b1eb..81721d55 100644 --- a/runtime/include/opal/server/session_registry.h +++ b/runtime/include/opal/server/session_registry.h @@ -106,8 +106,8 @@ struct DeliveryClass { // expire detached entries immediately. One lazy expiry thread serves the // whole registry — a detached session itself holds zero threads. // ResumeOrAdd is the blessed admission call (ADR-0022): resume-or-fresh- -// join with the brief retry the reconnect race needs, and Kick(id) is -// the kick primitive that makes its refusal actionable. +// join with the brief retry the reconnect race needs, and Kick(id, why, +// grace) is the kick primitive that makes its refusal actionable. // // All methods are safe from any thread, including a handler's own (a // handler may SendTo itself) and the slow-consumer callback (which runs @@ -368,7 +368,7 @@ class SessionRegistry { // cannot occur. kRefused at the deadline almost always means the id is // live on another connection (rarely: a detached entry past its grace // that expiry has not yet claimed — a beat later the same call would - // report kAdded); whether to kick a live one (Close) is the + // report kAdded); whether to kick a live one (Kick) is the // application's call — ResumeOrAdd never kicks on its own. Admission ResumeOrAdd(const Id& id, const std::function& mint, std::chrono::milliseconds deadline) { diff --git a/runtime/src/http/beast_transport.cc b/runtime/src/http/beast_transport.cc index d5093aaa..a427d531 100644 --- a/runtime/src/http/beast_transport.cc +++ b/runtime/src/http/beast_transport.cc @@ -578,28 +578,35 @@ class WsSession final : public WebSocketSessionBase, } } - // Close with `why` on the frame and the closing handshake bounded by - // `grace` (Beast's handshake timeout for this close only). A handshake - // already running is Close's own, waiting on the peer for the whole - // request timeout: the kick closes the socket under it instead. + // Close with `why` on the frame, plus a deadline of its own: `grace` + // after the kick, whatever the close is waiting on (the peer's answer, a + // stalled control frame holding Beast's write lock, a close deferred + // behind a wedged write), the socket closes under it. Only the first + // kick arms it, so a second cannot cut the first one's frame short. A + // close already requested keeps its frame. void Kick(const CloseReason& why, std::chrono::milliseconds grace) override { CheckCloseReason(why); auto self = this->shared_from_this(); try { asio::post(ws_.get_executor(), [self, why, grace] { std::unique_lock lock(self->mutex_); - if (self->close_started_) { - if (self->failed_ || self->peer_closed_) return; - lock.unlock(); - beast::error_code ignored; - (void)beast::get_lowest_layer(self->ws_).socket().close(ignored); - return; - } + if (self->kick_deadline_.has_value() || self->peer_closed_) return; + constexpr std::chrono::milliseconds kMaxGrace = std::chrono::hours(24); + self->kick_deadline_.emplace(self->ws_.get_executor()); + self->kick_deadline_->expires_after( + std::clamp(grace, std::chrono::milliseconds::zero(), kMaxGrace)); + self->kick_deadline_->async_wait( + [weak = self->weak_from_this()](const boost::system::error_code& ec) { + if (ec) return; + if (auto session = weak.lock()) { + beast::error_code ignored; + (void)beast::get_lowest_layer(session->ws_).socket().close(ignored); + } + }); if (!self->close_requested_) { self->close_reason_ = bws::close_reason(static_cast(why.code), why.reason); } - self->close_grace_ = grace; self->RequestCloseLocked(lock); }); } catch (...) { @@ -884,17 +891,7 @@ class WsSession final : public WebSocketSessionBase, void StartCloseLocked(std::unique_lock& lock) { close_started_ = true; const bws::close_reason why = close_reason_; - const std::optional grace = close_grace_; lock.unlock(); - if (grace.has_value()) { - // Read when the close starts (Beast arms the closing phase's timer - // from it then), so it bounds this handshake and nothing else. - constexpr std::chrono::milliseconds kMaxGrace = std::chrono::hours(24); - bws::stream_base::timeout timeout; - ws_.get_option(timeout); - timeout.handshake_timeout = std::clamp(*grace, std::chrono::milliseconds::zero(), kMaxGrace); - ws_.set_option(timeout); - } auto self = this->shared_from_this(); // Beast's close op performs the whole closing handshake (it reads // until the peer's close arrives, then tears down): its success IS @@ -1028,8 +1025,9 @@ class WsSession final : public WebSocketSessionBase, bool close_requested_ = false; bool close_started_ = false; bws::close_reason close_reason_{bws::close_code::normal}; - // Engaged by Kick: the closing handshake's bound. - std::optional close_grace_; + // Armed by the first Kick: closes the socket `grace` later. Touched on + // the connection's executor only. + std::optional kick_deadline_; // The close the peer started, as its frame carried it. std::optional peer_close_; }; diff --git a/runtime/tests/eventstream/async_event_stream_test.cc b/runtime/tests/eventstream/async_event_stream_test.cc index 223d4537..8adf7448 100644 --- a/runtime/tests/eventstream/async_event_stream_test.cc +++ b/runtime/tests/eventstream/async_event_stream_test.cc @@ -671,12 +671,17 @@ TEST(AsyncEventStreamTest, SharedHandlesOutliveTheLoopAndItsStream) { EXPECT_EQ(stale.error().kind(), ErrorKind::kTransport); } -// A hub ending a session from outside the loop (a registry Close or Kick) -// on a transport that completes the parked receive inline on the closing -// thread — the pair's documented behavior: the loop ends and its stream -// is destroyed inside that very call, so the handle's call must not hold -// anything the stream's teardown waits for. -void EndFromOutsideTheParkedLoop(const std::function&)>& end) { +// A hub reaching a session from outside the loop (a registry Close, Kick +// or delivery) on a transport that completes parked operations inline on +// the calling thread — the pair's documented behavior: the loop ends and +// its stream is destroyed inside that very call, so the handle's call must +// hold nothing the stream's teardown waits for, and must keep the socket +// alive itself (the test drops its own reference first). `arm` runs on the +// client end before the handle's call. +void ReachFromOutsideTheParkedLoop( + const std::function&)>& arm, + const std::function&)>& reach, + std::optional peer_reads) { auto [client_socket, server_socket] = http::InMemoryWebSocketPair::Create(); Mailbox> minted; std::atomic done{false}; @@ -684,35 +689,64 @@ void EndFromOutsideTheParkedLoop(const std::function* done) -> Detached { AsyncServer stream(std::move(socket), EncodePong, DecodePing); minted->Post(stream.Share()); - (void)co_await stream.Receive(); // parked until the handle ends it + (void)co_await stream.Receive(); // parked until the handle's call ends it *done = true; - }(server_socket, &minted, &done); + }(std::exchange(server_socket, nullptr), &minted, &done); EventStreamHandle handle = minted.Wait(); + arm(client_socket); std::promise returned; - std::thread ender([&] { - end(handle); + std::thread caller([&] { + reach(handle); returned.set_value(); }); if (returned.get_future().wait_for(std::chrono::seconds(5)) != std::future_status::ready) { - ender.detach(); // wedged for good; the assertion below names it - FAIL() << "ending the session through its handle deadlocked on the stream's teardown"; + // The caller is wedged on this frame's locals; nothing can unwind it. + ADD_FAILURE() << "the handle's call deadlocked on the stream's teardown"; + std::abort(); } - ender.join(); + caller.join(); EXPECT_TRUE(done.load()); - auto at_client = client_socket->Receive(); - ASSERT_TRUE(at_client.ok()) << at_client.error().message(); - EXPECT_FALSE(at_client->has_value()); + EXPECT_EQ(client_socket->PeerCloseReason(), peer_reads); } +void NothingToArm(const std::shared_ptr&) {} + TEST(AsyncEventStreamTest, AHandleCloseThatEndsTheLoopInlineReturns) { - EndFromOutsideTheParkedLoop([](const EventStreamHandle& handle) { handle.Close(); }); + ReachFromOutsideTheParkedLoop( + NothingToArm, [](const EventStreamHandle& handle) { handle.Close(); }, + http::CloseReason{.code = 1000, .reason = ""}); } TEST(AsyncEventStreamTest, AHandleKickThatEndsTheLoopInlineReturns) { - EndFromOutsideTheParkedLoop([](const EventStreamHandle& handle) { - handle.Kick({.code = 4001, .reason = "superseded"}, std::chrono::milliseconds(100)); - }); + ReachFromOutsideTheParkedLoop( + NothingToArm, + [](const EventStreamHandle& handle) { + handle.Kick({.code = 4001, .reason = "superseded"}, std::chrono::milliseconds(100)); + }, + http::CloseReason{.code = 4001, .reason = "superseded"}); +} + +// A delivery the peer answers by closing: the peer's receive completes +// inside the handle's Send, its close completes the loop's receive, and +// the loop's stream is torn down below the Send on the same thread. +TEST(AsyncEventStreamTest, AHandleSendThePeerAnswersByClosingReturns) { + ReachFromOutsideTheParkedLoop( + [](const std::shared_ptr& client) { + client->ReceiveAsync([client](Outcome>) { client->Close(); }); + }, + [](const EventStreamHandle& handle) { (void)handle.Send(Pong{"bye"}); }, std::nullopt); +} + +TEST(AsyncEventStreamTest, AHandleSendAsyncThePeerAnswersByClosingReturns) { + ReachFromOutsideTheParkedLoop( + [](const std::shared_ptr& client) { + client->ReceiveAsync([client](Outcome>) { client->Close(); }); + }, + [](const EventStreamHandle& handle) { + handle.SendAsync(Pong{"bye"}, [](Outcome) {}); + }, + std::nullopt); } TEST(AsyncEventStreamTest, HandleSendAsyncCompletesAndFailsSoftlyAfterRevocation) { diff --git a/runtime/tests/eventstream/event_stream_test.cc b/runtime/tests/eventstream/event_stream_test.cc index 9652dac3..0a2eef39 100644 --- a/runtime/tests/eventstream/event_stream_test.cc +++ b/runtime/tests/eventstream/event_stream_test.cc @@ -298,7 +298,7 @@ TEST(EventStreamHandleTest, HandleWorksOnTheBorrowedServerPath) { } TEST(EventStreamHandleTest, HandleCloseEndsTheSessionForTheStreamOwner) { - // The hub's kick path: Close from a thread that only holds the handle + // A hub ending a session: Close from a thread that only holds the handle // unblocks the owner's Receive with the stream's natural end. auto [client_socket, server_socket] = http::InMemoryWebSocketPair::Create(); ClientStream client(client_socket, EncodePing, DecodePong); diff --git a/runtime/tests/http/beast_websocket_test.cc b/runtime/tests/http/beast_websocket_test.cc index 940eed55..220963f2 100644 --- a/runtime/tests/http/beast_websocket_test.cc +++ b/runtime/tests/http/beast_websocket_test.cc @@ -32,9 +32,11 @@ #include "opal/eventstream/async_event_stream.h" #include "opal/eventstream/envelope.h" +#include "opal/eventstream/event_stream.h" #include "opal/eventstream/json_frame.h" #include "opal/http/beast_transport.h" #include "opal/http/websocket.h" +#include "opal/server/session_registry.h" #include "opal/testing/connection_event_recorder.h" #include "opal/testing/tls_test_identity.h" #include "opal/testing/websocket_contract_test.h" @@ -1453,18 +1455,32 @@ class HeldSessionServer { socket_.set_value(&socket); auto received = socket.Receive(); ended_promise_.set_value(Ended{std::move(received), socket.PeerCloseReason()}); + // The borrow outlives the receive until the test lets go of it. + (void)released_.wait_for(std::chrono::seconds(30)); }; server_ = std::make_unique(options); EXPECT_TRUE(server_->Start(NotFoundHandler()).ok()); } - ~HeldSessionServer() { server_->Stop(); } + ~HeldSessionServer() { + Release(); + server_->Stop(); + } + // Lets the serve callback return, ending the borrow: call once anything + // built on Socket() (a stream, say) is gone. + void Release() { + if (!released_set_) release_.set_value(); + released_set_ = true; + } int port() const { return server_->port(); } // The server's end of the session, once a peer has upgraded. WebSocket& Socket() { if (held_ == nullptr) { auto socket = socket_.get_future(); - EXPECT_EQ(socket.wait_for(std::chrono::seconds(5)), std::future_status::ready); + if (socket.wait_for(std::chrono::seconds(5)) != std::future_status::ready) { + ADD_FAILURE() << "no peer upgraded"; + std::abort(); // nothing to hand back; get() would wait forever + } held_ = socket.get(); } return *held_; @@ -1480,6 +1496,9 @@ class HeldSessionServer { private: std::promise socket_; WebSocket* held_ = nullptr; + std::promise release_; + std::shared_future released_ = release_.get_future().share(); + bool released_set_ = false; std::promise ended_promise_; std::future ended_future_ = ended_promise_.get_future(); std::optional ended_; @@ -1500,8 +1519,9 @@ TEST(BeastWebSocketTest, ACloseWaitsOutTheHandshakeForAPeerThatNeverAnswers) { TEST(BeastWebSocketTest, AKickEndsTheSessionOfAPeerThatNeverAnswersWithinItsGrace) { HeldSessionServer server; // a 30s closing handshake, were it Close RawWsPeer peer(server.port()); + WebSocket& socket = server.Socket(); const auto kicked = std::chrono::steady_clock::now(); - server.Socket().Kick({.code = 4001, .reason = "superseded"}, std::chrono::milliseconds(200)); + socket.Kick({.code = 4001, .reason = "superseded"}, std::chrono::milliseconds(200)); auto ended = server.EndedWithin(std::chrono::seconds(5)); ASSERT_TRUE(ended.has_value()) << "the kick waited for a peer that never answers"; EXPECT_FALSE(ended->received.ok()) << "an unanswered close is not a clean end"; @@ -1526,7 +1546,7 @@ TEST(BeastWebSocketTest, TheClientReadsTheCodeAKickSent) { auto dialed = BeastWebSocketClient::Dial({.host = "127.0.0.1", .port = server.port()}); ASSERT_TRUE(dialed.ok()) << dialed.error().message(); server.Socket().Kick({.code = 4001, .reason = "superseded"}, std::chrono::seconds(5)); - auto ended = (*dialed)->Receive(); + auto ended = (*dialed)->Receive(std::chrono::seconds(5)); ASSERT_TRUE(ended.ok()) << ended.error().message(); EXPECT_FALSE(ended->has_value()); EXPECT_EQ((*dialed)->PeerCloseReason(), (CloseReason{.code = 4001, .reason = "superseded"})); @@ -1538,7 +1558,7 @@ TEST(BeastWebSocketTest, AServerCloseReadsAsANormalClosureAtTheClient) { ASSERT_TRUE(dialed.ok()) << dialed.error().message(); EXPECT_EQ((*dialed)->PeerCloseReason(), std::nullopt) << "nothing closed yet"; server.Socket().Close(); - auto ended = (*dialed)->Receive(); + auto ended = (*dialed)->Receive(std::chrono::seconds(5)); ASSERT_TRUE(ended.ok()) << ended.error().message(); EXPECT_FALSE(ended->has_value()); EXPECT_EQ((*dialed)->PeerCloseReason(), (CloseReason{.code = 1000, .reason = ""})); @@ -1548,6 +1568,68 @@ TEST(BeastWebSocketTest, AServerCloseReadsAsANormalClosureAtTheClient) { EXPECT_EQ(server_end->peer_close, std::nullopt); } +// Two redials under one identity can kick the same session twice: the +// second must not cut the first kick's frame short, or the peer never +// learns it was superseded. +TEST(BeastWebSocketTest, ASecondKickLeavesTheFirstKicksFrameAlone) { + HeldSessionServer server; + RawWsPeer peer(server.port()); + WebSocket& socket = server.Socket(); + socket.Kick({.code = 4001, .reason = "superseded"}, std::chrono::seconds(5)); + socket.Kick({.code = 4002, .reason = "again"}, std::chrono::milliseconds(0)); + EXPECT_EQ(peer.DrainToCloseCode(), boost::beast::websocket::error::closed); + EXPECT_EQ(peer.reason().code, 4001); + ASSERT_TRUE(server.EndedWithin(std::chrono::seconds(5)).has_value()); +} + +// The first kick's deadline governs: a later one with a shorter grace +// neither re-arms nor hurries it. +TEST(BeastWebSocketTest, ASecondKickKeepsTheFirstKicksDeadline) { + HeldSessionServer server; + RawWsPeer peer(server.port()); // never answers + WebSocket& socket = server.Socket(); + socket.Kick({.code = 4001, .reason = "superseded"}, std::chrono::milliseconds(1500)); + socket.Kick({.code = 4002, .reason = "again"}, std::chrono::milliseconds(0)); + EXPECT_FALSE(server.EndedWithin(std::chrono::milliseconds(700)).has_value()) + << "the second kick cut the first one's grace short"; + ASSERT_TRUE(server.EndedWithin(std::chrono::seconds(5)).has_value()); +} + +// A client may kick too: the server's end reads why. +TEST(BeastWebSocketTest, AClientKickTellsTheServerWhy) { + HeldSessionServer server; + auto dialed = BeastWebSocketClient::Dial({.host = "127.0.0.1", .port = server.port()}); + ASSERT_TRUE(dialed.ok()) << dialed.error().message(); + (void)server.Socket(); // the server's receive is parked + (*dialed)->Kick({.code = 4100, .reason = "leaving"}, std::chrono::seconds(5)); + auto ended = server.EndedWithin(std::chrono::seconds(5)); + ASSERT_TRUE(ended.has_value()); + ASSERT_TRUE(ended->received.ok()) << ended->received.error().message(); + EXPECT_FALSE(ended->received->has_value()); + EXPECT_EQ(ended->peer_close, (CloseReason{.code = 4100, .reason = "leaving"})); +} + +// The hub's path end to end: a registry Kick through the session's handle +// carries the grace down to the wire. +TEST(BeastWebSocketTest, ARegistryKickBoundsTheWaitForAPeerThatNeverAnswers) { + HeldSessionServer server; + RawWsPeer peer(server.port()); + { + eventstream::EventStream stream( + server.Socket(), [](const Message& m) -> Outcome { return m; }, nullptr); + server::SessionRegistry registry; + ASSERT_TRUE(registry.Add("ada", stream.Share())); + const auto kicked = std::chrono::steady_clock::now(); + ASSERT_TRUE(registry.Kick("ada", {.code = 4001, .reason = "superseded"}, + std::chrono::milliseconds(200))); + ASSERT_TRUE(server.EndedWithin(std::chrono::seconds(5)).has_value()) + << "the kick waited for a peer that never answers"; + EXPECT_LT(std::chrono::steady_clock::now() - kicked, std::chrono::seconds(2)); + EXPECT_TRUE(registry.Remove("ada")); + } // the stream goes before the borrow it was built on + server.Release(); +} + TEST(BeastWebSocketTest, AKickOnASessionAlreadyClosingStopsWaitingForThePeer) { HeldSessionServer server; RawWsPeer peer(server.port()); From 31ea055032f79b820f5d4f1d756334558575171a Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 13 Oct 2026 19:46:59 +0000 Subject: [PATCH 5/7] Stream move-assignment ends the old session first --- CHANGELOG.md | 7 +++ .../opal/eventstream/async_event_stream.h | 18 ++++++-- .../include/opal/eventstream/event_stream.h | 19 ++++++-- .../tests/eventstream/event_stream_test.cc | 46 +++++++++++++++++++ 4 files changed, 82 insertions(+), 8 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d946f1b7..9bb34842 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -275,6 +275,13 @@ policy in [docs/versioning.md](docs/versioning.md). ### Fixed +- **Move-assigning over a stream that had shared its session no longer + closes a released socket.** The defaulted move-assignment of + `EventStream` and `AsyncEventStream` assigned the socket before the + shared view, so ending the old session (the view's assignment) called + `Close()` on a socket the stream might already have freed. Both now end + the old session first. + - **A handle call no longer deadlocks when the transport ends the loop inline.** The in-memory pair completes parked operations on the calling thread, so a hub calling `registry.Close(id)` on a session parked in diff --git a/runtime/include/opal/eventstream/async_event_stream.h b/runtime/include/opal/eventstream/async_event_stream.h index 05dbeb9f..1ae11960 100644 --- a/runtime/include/opal/eventstream/async_event_stream.h +++ b/runtime/include/opal/eventstream/async_event_stream.h @@ -268,9 +268,18 @@ class AsyncEventStream { // handles minted by Share() fail softly once the stream is gone. ~AsyncEventStream() = default; AsyncEventStream(AsyncEventStream&&) noexcept = default; - // No explicit noexcept: the computed one depends on std::function's - // move-assignment, which the standard does not make noexcept. - AsyncEventStream& operator=(AsyncEventStream&&) = default; + // Not defaulted: that assigns in declaration order, so the socket would + // be released before view_'s assignment ends the old session through it. + // No noexcept: std::function's move-assignment is not required to be. + AsyncEventStream& operator=(AsyncEventStream&& other) { + if (this != &other) { + view_ = std::move(other.view_); + socket_ = std::move(other.socket_); + encode_ = std::move(other.encode_); + decode_ = std::move(other.decode_); + } + return *this; + } AsyncEventStream(const AsyncEventStream&) = delete; AsyncEventStream& operator=(const AsyncEventStream&) = delete; @@ -412,7 +421,8 @@ class AsyncEventStream { Encoder encode_; Decoder decode_; // Last member on purpose: its teardown (End) runs first and needs the - // socket above still alive. + // socket above still alive. Move-assignment takes it first for the same + // reason. internal::SharedViewOwner view_; }; diff --git a/runtime/include/opal/eventstream/event_stream.h b/runtime/include/opal/eventstream/event_stream.h index 5a628e98..843f25ac 100644 --- a/runtime/include/opal/eventstream/event_stream.h +++ b/runtime/include/opal/eventstream/event_stream.h @@ -313,9 +313,19 @@ class EventStream { // session fans out, so the stream itself stays single-owner. ~EventStream() = default; EventStream(EventStream&&) noexcept = default; - // No explicit noexcept: the computed one depends on std::function's - // move-assignment, which the standard leaves unspecified. - EventStream& operator=(EventStream&&) = default; + // Not defaulted: that assigns in declaration order, so the socket would + // be released before view_'s assignment ends the old session through it. + // No noexcept: std::function's move-assignment is not required to be. + EventStream& operator=(EventStream&& other) { + if (this != &other) { + view_ = std::move(other.view_); + owned_ = std::move(other.owned_); + socket_ = other.socket_; + encode_ = std::move(other.encode_); + decode_ = std::move(other.decode_); + } + return *this; + } // Encodes and sends one event, blocking until its frame is on the wire. // An encoder failure surfaces as-is and leaves the session untouched; @@ -390,7 +400,8 @@ class EventStream { Encoder encode_; Decoder decode_; // Last member on purpose: its teardown (End) runs first and needs the - // socket members above still alive. + // socket members above still alive. Move-assignment takes it first for + // the same reason. internal::SharedViewOwner view_; }; diff --git a/runtime/tests/eventstream/event_stream_test.cc b/runtime/tests/eventstream/event_stream_test.cc index 0a2eef39..0ce2d36c 100644 --- a/runtime/tests/eventstream/event_stream_test.cc +++ b/runtime/tests/eventstream/event_stream_test.cc @@ -20,6 +20,7 @@ #include #include +#include "opal/eventstream/async_event_stream.h" #include "opal/eventstream/frame.h" #include "opal/http/websocket.h" #include "opal/http/websocket_pair.h" @@ -297,6 +298,51 @@ TEST(EventStreamHandleTest, HandleWorksOnTheBorrowedServerPath) { EXPECT_EQ(stale.error().kind(), ErrorKind::kTransport); } +// A socket that records, when the stream ends it, whether the stream still +// owned it then — how a test sees a close land on a socket its stream had +// already let go of (on a real one, a use after free) without committing +// it: the test keeps its own reference throughout. +class OwnershipAtClose final : public http::WebSocket, + public std::enable_shared_from_this { + public: + Outcome> Receive() override { return std::optional(); } + Outcome> Receive(std::chrono::milliseconds) override { + return std::optional(); + } + Outcome Send(const Message&) override { return Unit{}; } + void Close() override { + // The test's reference, plus the stream's while it still holds one. + if (!owners_at_close_.has_value()) owners_at_close_ = weak_from_this().use_count(); + } + std::optional owners_at_close() const { return owners_at_close_; } + + private: + std::optional owners_at_close_; +}; + +template +void MoveAssignmentEndsTheOldSessionWhileItStillOwnsTheSocket() { + auto old_socket = std::make_shared(); + auto new_socket = std::make_shared(); + auto identity = [](const Message& m) -> Outcome { return m; }; + Stream stream(old_socket, identity, identity); + auto handle = stream.Share(); // a shared view, so the old session has one to end + stream = Stream(new_socket, identity, identity); + ASSERT_TRUE(old_socket->owners_at_close().has_value()) << "the old session was never ended"; + EXPECT_EQ(*old_socket->owners_at_close(), 2) + << "the old socket was closed after the stream had released it"; + EXPECT_FALSE(handle.Send(Message{}).ok()) << "the old session's handle still reaches a socket"; + EXPECT_FALSE(new_socket->owners_at_close().has_value()) << "the new session was ended"; +} + +TEST(EventStreamTest, MoveAssignmentEndsTheOldSessionWhileItStillOwnsTheSocket) { + MoveAssignmentEndsTheOldSessionWhileItStillOwnsTheSocket>(); +} + +TEST(AsyncEventStreamTest, MoveAssignmentEndsTheOldSessionWhileItStillOwnsTheSocket) { + MoveAssignmentEndsTheOldSessionWhileItStillOwnsTheSocket>(); +} + TEST(EventStreamHandleTest, HandleCloseEndsTheSessionForTheStreamOwner) { // A hub ending a session: Close from a thread that only holds the handle // unblocks the owner's Receive with the stream's natural end. From 5525f0d936fa9a1fba691438c85597ba51311988 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 13 Oct 2026 19:57:24 +0000 Subject: [PATCH 6/7] Stream move-assignments are noexcept --- runtime/include/opal/eventstream/async_event_stream.h | 3 +-- runtime/include/opal/eventstream/event_stream.h | 3 +-- 2 files changed, 2 insertions(+), 4 deletions(-) diff --git a/runtime/include/opal/eventstream/async_event_stream.h b/runtime/include/opal/eventstream/async_event_stream.h index 1ae11960..6cb2dea1 100644 --- a/runtime/include/opal/eventstream/async_event_stream.h +++ b/runtime/include/opal/eventstream/async_event_stream.h @@ -270,8 +270,7 @@ class AsyncEventStream { AsyncEventStream(AsyncEventStream&&) noexcept = default; // Not defaulted: that assigns in declaration order, so the socket would // be released before view_'s assignment ends the old session through it. - // No noexcept: std::function's move-assignment is not required to be. - AsyncEventStream& operator=(AsyncEventStream&& other) { + AsyncEventStream& operator=(AsyncEventStream&& other) noexcept { if (this != &other) { view_ = std::move(other.view_); socket_ = std::move(other.socket_); diff --git a/runtime/include/opal/eventstream/event_stream.h b/runtime/include/opal/eventstream/event_stream.h index 843f25ac..4c4e539f 100644 --- a/runtime/include/opal/eventstream/event_stream.h +++ b/runtime/include/opal/eventstream/event_stream.h @@ -315,8 +315,7 @@ class EventStream { EventStream(EventStream&&) noexcept = default; // Not defaulted: that assigns in declaration order, so the socket would // be released before view_'s assignment ends the old session through it. - // No noexcept: std::function's move-assignment is not required to be. - EventStream& operator=(EventStream&& other) { + EventStream& operator=(EventStream&& other) noexcept { if (this != &other) { view_ = std::move(other.view_); owned_ = std::move(other.owned_); From e88b4c15d69477fb24cd4eb9d6b50dbc0387bf0c Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 13 Oct 2026 22:53:00 +0000 Subject: [PATCH 7/7] EventStream::Share passes its owned socket to handles --- CHANGELOG.md | 7 ++++--- .../adr/0022-registry-admission-primitives.md | 10 +++++---- .../include/opal/eventstream/event_stream.h | 9 ++++---- .../tests/eventstream/event_stream_test.cc | 21 +++++++++++++++++++ 4 files changed, 36 insertions(+), 11 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9bb34842..f1fd18fe 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -287,9 +287,10 @@ policy in [docs/versioning.md](docs/versioning.md). thread, so a hub calling `registry.Close(id)` on a session parked in `co_await Receive()` — or delivering to a peer that answered by closing — destroyed the stream inside that call, and the stream's teardown - waited forever for the pin the call held. A handle minted by - `AsyncEventStream` now keeps the socket alive by ownership rather than - by a pin, for every operation. + waited forever for the pin the call held. A handle on a stream that owns + its socket (`AsyncEventStream`, or an `EventStream` built from a + `shared_ptr`) now keeps the socket alive by ownership rather than by a + pin, for every operation. - **A dialed WebSocket released on its own io thread no longer terminates the process.** When the last handle to a `BeastWebSocketClient::Dial` diff --git a/docs/adr/0022-registry-admission-primitives.md b/docs/adr/0022-registry-admission-primitives.md index 9efe79f2..e7641241 100644 --- a/docs/adr/0022-registry-admission-primitives.md +++ b/docs/adr/0022-registry-admission-primitives.md @@ -102,10 +102,12 @@ in-memory pair does, as its contract allows — and the loop then ends and destroys its stream inside the handle's call, where the stream's teardown waited for the pin that call held. Any hub calling `Close(id)` on a pair-backed session parked in `Receive` deadlocked, and so did a delivery -the peer answered by closing. The coroutine adapter owns its socket, so a -handle now holds a strong reference across each call (to completion, for -`SendAsync`) instead; a borrowed socket is still pinned, and a borrowing -stream is never torn down by its own socket's completions. +the peer answered by closing. When the stream owns its socket (the +coroutine adapter always does; a blocking `EventStream` built from a +`shared_ptr` does too), a handle now holds a strong reference across each +call (to completion, for `SendAsync`) instead; a borrowed socket is still +pinned, and a borrowing stream is never torn down by its own socket's +completions. ## Non-goals diff --git a/runtime/include/opal/eventstream/event_stream.h b/runtime/include/opal/eventstream/event_stream.h index 4c4e539f..01104615 100644 --- a/runtime/include/opal/eventstream/event_stream.h +++ b/runtime/include/opal/eventstream/event_stream.h @@ -54,9 +54,8 @@ struct SharedSessionState { std::condition_variable idle; http::WebSocket* socket = nullptr; // null once the stream ended int active = 0; // pinned handle operations inside socket calls - // Engaged when the stream owns its socket (the coroutine adapter): handle - // operations then hold it by reference instead of a pin - // (EventStreamHandle::Acquire). + // Engaged when the stream owns its socket: handle operations then hold it + // by reference instead of a pin (EventStreamHandle::Acquire). std::weak_ptr owner; }; @@ -374,7 +373,9 @@ class EventStream { // stream closes the session and leaves them failing softly with // Error::Transport. Call from the thread that owns the stream (it mutates // the stream), as many times as you like. - EventStreamHandle Share() { return EventStreamHandle(view_.Ensure(socket_), encode_); } + EventStreamHandle Share() { + return EventStreamHandle(view_.Ensure(socket_, owned_), encode_); + } private: // Both receive overloads: nullopt through, decode the message, and treat diff --git a/runtime/tests/eventstream/event_stream_test.cc b/runtime/tests/eventstream/event_stream_test.cc index 0ce2d36c..c51e62e3 100644 --- a/runtime/tests/eventstream/event_stream_test.cc +++ b/runtime/tests/eventstream/event_stream_test.cc @@ -335,6 +335,27 @@ void MoveAssignmentEndsTheOldSessionWhileItStillOwnsTheSocket() { EXPECT_FALSE(new_socket->owners_at_close().has_value()) << "the new session was ended"; } +// A handle call on a stream that owns its socket holds the socket by +// reference for the call — the test's, the stream's and the call's — never +// by a pin the stream's teardown would wait on. +template +void AHandleCallHoldsAnOwnedSocketByReference() { + auto socket = std::make_shared(); + auto identity = [](const Message& m) -> Outcome { return m; }; + Stream stream(socket, identity, identity); + stream.Share().Close(); + ASSERT_TRUE(socket->owners_at_close().has_value()); + EXPECT_EQ(*socket->owners_at_close(), 3); +} + +TEST(EventStreamTest, AHandleCallHoldsAnOwnedSocketByReference) { + AHandleCallHoldsAnOwnedSocketByReference>(); +} + +TEST(AsyncEventStreamTest, AHandleCallHoldsAnOwnedSocketByReference) { + AHandleCallHoldsAnOwnedSocketByReference>(); +} + TEST(EventStreamTest, MoveAssignmentEndsTheOldSessionWhileItStillOwnsTheSocket) { MoveAssignmentEndsTheOldSessionWhileItStillOwnsTheSocket>(); }