fix(quic): stop ~ClientQUIC racing the callback to close the connection
MsQuicConnectionClose was reachable twice for one handle: ~ClientQUIC called it unconditionally, and the SHUTDOWN_COMPLETE callback also called it whenever AppCloseInProgress was clear -- which it is for the whole window between the destructor's ConnectionShutdown and its ConnectionClose. The second call trips CXPLAT_TEL_ASSERT(!Connection->State.HandleClosed) and aborts the process. Clearing impl->connection from the callback did not help: the destructor had already loaded it. `connection` is now a claim token, taken under the mutex, and only whoever wins the claim closes. The callback keeps closing peer-dropped connections so they do not leak while their ClientQUIC lives on. Separately, the destructor now waits (bounded, 5s) for the connection's streams to reach StreamClose before closing it. QUICStream::Stop only initiates a shutdown whose async completion does the close, so a connection could otherwise be closed with streams msquic still considers open. The count lives in a shared StreamRegistry rather than behind the ClientQUIC*, so a stream finalising after the wait gave up decrements a live object. A failed handshake also closes its connection and configuration now. Throwing from the constructor body means ~ClientQUIC never runs, so both handles leaked -- and the connection callback holds the impl that was about to be destroyed with the half-built object. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
parent
25e8c84794
commit
42a34de244
2 changed files with 138 additions and 16 deletions
|
|
@ -50,6 +50,47 @@ namespace {
|
|||
return r;
|
||||
}
|
||||
|
||||
// Open-stream bookkeeping, shared by a connection and every stream on it.
|
||||
// A stream counts as open from the moment msquic has a callback handler
|
||||
// for it until FinalizeClose has run StreamClose. ~ClientQUIC drains this
|
||||
// before closing the connection: QUICStream::Stop only *initiates* a
|
||||
// shutdown whose async completion does the actual close, so without the
|
||||
// wait a connection can be closed with streams still open on the msquic
|
||||
// side.
|
||||
//
|
||||
// Reached through a shared_ptr rather than through the owning ClientQUIC*
|
||||
// so that a stream finalising after the (bounded) wait gave up decrements
|
||||
// a live object instead of a freed connection.
|
||||
struct StreamRegistry {
|
||||
std::mutex mtx;
|
||||
std::condition_variable cv;
|
||||
int open = 0;
|
||||
|
||||
void Add() {
|
||||
std::lock_guard lk(mtx);
|
||||
++open;
|
||||
}
|
||||
|
||||
void Remove() {
|
||||
{
|
||||
std::lock_guard lk(mtx);
|
||||
--open;
|
||||
}
|
||||
cv.notify_all();
|
||||
}
|
||||
|
||||
// False if the timeout expired with streams still open.
|
||||
bool WaitForDrain(std::chrono::milliseconds timeout) {
|
||||
std::unique_lock lk(mtx);
|
||||
return cv.wait_for(lk, timeout, [&]{ return open == 0; });
|
||||
}
|
||||
};
|
||||
|
||||
// Bound on how long ~ClientQUIC waits for the connection's streams to
|
||||
// finish closing. The decrements arrive on msquic worker threads, so an
|
||||
// unbounded wait would turn a dropped peer into a hung destructor.
|
||||
constexpr std::chrono::milliseconds streamDrainTimeout{ 5000 };
|
||||
|
||||
// Encode an ALPN string into the wire format msquic expects: a length
|
||||
// byte followed by the ASCII characters. Lifetime of the returned buffer
|
||||
// matches the caller's storage in `out`.
|
||||
|
|
@ -85,15 +126,29 @@ struct QUICStream::Impl {
|
|||
// ref (wrapper / in-flight callback copy) drops.
|
||||
std::shared_ptr<Impl>* selfRef = nullptr;
|
||||
|
||||
// The owning connection's open-stream count. Incremented where the msquic
|
||||
// handler is installed (QUICStream's handle constructor / OpenStream),
|
||||
// decremented by FinalizeClose.
|
||||
std::shared_ptr<StreamRegistry> registry;
|
||||
|
||||
// The connection's registry. Defined below ClientQUIC::Impl (which owns
|
||||
// it); a member of QUICStream::Impl rather than a free function so it
|
||||
// inherits QUICStream's friendship with ClientQUIC.
|
||||
static std::shared_ptr<StreamRegistry> RegistryOf(ClientQUIC* connection);
|
||||
|
||||
// Detach the msquic handler, close the stream, and release the callback's
|
||||
// strong ref. Called once, from SHUTDOWN_COMPLETE (normal path) or the
|
||||
// OpenStream error path — never racing another close of the same stream.
|
||||
static void FinalizeClose(Impl* self, HQUIC stream) {
|
||||
Runtime().api->SetCallbackHandler(stream, nullptr, nullptr);
|
||||
Runtime().api->StreamClose(stream);
|
||||
std::shared_ptr<StreamRegistry> registry = std::move(self->registry);
|
||||
std::shared_ptr<Impl>* ref = self->selfRef;
|
||||
self->selfRef = nullptr;
|
||||
delete ref;
|
||||
// Last: a destructor blocked in WaitForDrain may run ConnectionClose
|
||||
// the moment this hits zero, and msquic wants StreamClose first.
|
||||
if (registry) registry->Remove();
|
||||
}
|
||||
|
||||
static QUIC_STATUS QUIC_API Callback(HQUIC stream, void* ctx, QUIC_STREAM_EVENT* ev) {
|
||||
|
|
@ -163,6 +218,8 @@ QUICStream::QUICStream(HQUIC handle, ClientQUIC* connection)
|
|||
{
|
||||
impl->handle = handle;
|
||||
impl->connection = connection;
|
||||
impl->registry = Impl::RegistryOf(connection);
|
||||
if (impl->registry) impl->registry->Add();
|
||||
impl->selfRef = new std::shared_ptr<Impl>(impl); // strong ref owned by the callback
|
||||
Runtime().api->SetCallbackHandler(handle, reinterpret_cast<void*>(&Impl::Callback), impl->selfRef);
|
||||
}
|
||||
|
|
@ -355,8 +412,38 @@ struct ClientQUIC::Impl {
|
|||
// H3_MISSING_SETTINGS on the peer side.
|
||||
std::deque<QUICStream> pendingStreams;
|
||||
|
||||
// Streams currently open on this connection. Shared with each QUICStream
|
||||
// so the count survives a stream that outlives us.
|
||||
std::shared_ptr<StreamRegistry> streams = std::make_shared<StreamRegistry>();
|
||||
|
||||
ClientQUIC* outer = nullptr;
|
||||
|
||||
// Take the connection handle away from whoever else might close it.
|
||||
// `connection` is the single owner token: exactly one of ~ClientQUIC and
|
||||
// the SHUTDOWN_COMPLETE callback wins the claim, and only the winner calls
|
||||
// ConnectionClose. Without this the two race and msquic bugchecks on the
|
||||
// second close (CXPLAT_TEL_ASSERT(!Connection->State.HandleClosed)).
|
||||
HQUIC ClaimConnection() {
|
||||
std::lock_guard lk(mtx);
|
||||
HQUIC claimed = connection;
|
||||
connection = nullptr;
|
||||
return claimed;
|
||||
}
|
||||
|
||||
// Shut the connection down and close it, waiting (bounded) for every
|
||||
// stream on it to reach StreamClose first. No-op if someone else already
|
||||
// claimed the handle. Never holds `mtx` across an msquic call: the
|
||||
// connection callback takes it too.
|
||||
void CloseConnection() {
|
||||
HQUIC claimed = ClaimConnection();
|
||||
if (!claimed) return;
|
||||
// Aborts any stream still open; each one's SHUTDOWN_COMPLETE runs
|
||||
// FinalizeClose, which closes it and drops the count below.
|
||||
Runtime().api->ConnectionShutdown(claimed, QUIC_CONNECTION_SHUTDOWN_FLAG_NONE, 0);
|
||||
streams->WaitForDrain(streamDrainTimeout);
|
||||
Runtime().api->ConnectionClose(claimed);
|
||||
}
|
||||
|
||||
static QUIC_STATUS QUIC_API Callback(HQUIC conn, void* ctx, QUIC_CONNECTION_EVENT* ev) {
|
||||
auto* self = static_cast<Impl*>(ctx);
|
||||
switch (ev->Type) {
|
||||
|
|
@ -386,9 +473,16 @@ struct ClientQUIC::Impl {
|
|||
self->closed = true;
|
||||
}
|
||||
self->cv.notify_all();
|
||||
if (ev->SHUTDOWN_COMPLETE.AppCloseInProgress == 0) {
|
||||
// AppCloseInProgress means ~ClientQUIC is already inside
|
||||
// ConnectionClose; otherwise close the handle here so a
|
||||
// connection dropped by the peer does not leak while the
|
||||
// owning ClientQUIC lives on. ClaimConnection settles which
|
||||
// of the two actually does it. Every stream on the connection
|
||||
// has already been shut down (and so closed by FinalizeClose)
|
||||
// by the time msquic reports the connection complete.
|
||||
if (ev->SHUTDOWN_COMPLETE.AppCloseInProgress == 0
|
||||
&& self->ClaimConnection() != nullptr) {
|
||||
Runtime().api->ConnectionClose(conn);
|
||||
self->connection = nullptr;
|
||||
}
|
||||
return QUIC_STATUS_SUCCESS;
|
||||
}
|
||||
|
|
@ -521,11 +615,21 @@ ClientQUIC::ClientQUIC(const char* host, std::uint16_t port, std::string alpnIn,
|
|||
throw QUICException(std::format("ConnectionStart failed: 0x{:x}", static_cast<unsigned>(s)));
|
||||
}
|
||||
|
||||
std::unique_lock lk(impl->mtx);
|
||||
impl->cv.wait(lk, [&]{ return impl->connected || impl->closed; });
|
||||
if (!impl->connected) {
|
||||
throw QUICException(std::format("QUIC handshake failed: 0x{:x}", static_cast<unsigned>(impl->shutdownStatus)));
|
||||
QUIC_STATUS handshakeStatus = QUIC_STATUS_SUCCESS;
|
||||
{
|
||||
std::unique_lock lk(impl->mtx);
|
||||
impl->cv.wait(lk, [&]{ return impl->connected || impl->closed; });
|
||||
if (impl->connected) return;
|
||||
handshakeStatus = impl->shutdownStatus;
|
||||
}
|
||||
// Throwing from here means ~ClientQUIC never runs, so tear the msquic
|
||||
// handles down by hand — the connection callback holds `impl`, which is
|
||||
// about to be destroyed with the half-built object.
|
||||
impl->CloseConnection();
|
||||
Runtime().api->ConfigurationClose(impl->configuration);
|
||||
impl->configuration = nullptr;
|
||||
throw QUICException(std::format("QUIC handshake failed: 0x{:x}",
|
||||
static_cast<unsigned>(handshakeStatus)));
|
||||
}
|
||||
|
||||
ClientQUIC::ClientQUIC(std::string host, std::uint16_t port, std::string alpnIn, QUICClientCredentials creds)
|
||||
|
|
@ -552,11 +656,7 @@ ClientQUIC::ClientQUIC(ClientQUIC&& other) noexcept
|
|||
|
||||
ClientQUIC::~ClientQUIC() {
|
||||
if (!impl) return;
|
||||
if (impl->connection) {
|
||||
Runtime().api->ConnectionShutdown(impl->connection, QUIC_CONNECTION_SHUTDOWN_FLAG_NONE, 0);
|
||||
Runtime().api->ConnectionClose(impl->connection);
|
||||
impl->connection = nullptr;
|
||||
}
|
||||
impl->CloseConnection();
|
||||
if (impl->configuration && impl->ownsConfiguration) {
|
||||
Runtime().api->ConfigurationClose(impl->configuration);
|
||||
impl->configuration = nullptr;
|
||||
|
|
@ -564,8 +664,15 @@ ClientQUIC::~ClientQUIC() {
|
|||
}
|
||||
|
||||
void ClientQUIC::Stop() {
|
||||
if (!impl || !impl->connection) return;
|
||||
Runtime().api->ConnectionShutdown(impl->connection, QUIC_CONNECTION_SHUTDOWN_FLAG_NONE, 0);
|
||||
if (!impl) return;
|
||||
// Read the handle under the lock: the SHUTDOWN_COMPLETE callback may be
|
||||
// closing the connection concurrently on an msquic worker.
|
||||
HQUIC conn = nullptr;
|
||||
{
|
||||
std::lock_guard lk(impl->mtx);
|
||||
conn = impl->connection;
|
||||
}
|
||||
if (conn) Runtime().api->ConnectionShutdown(conn, QUIC_CONNECTION_SHUTDOWN_FLAG_NONE, 0);
|
||||
}
|
||||
|
||||
QUICStream ClientQUIC::OpenStream(bool unidirectional) {
|
||||
|
|
@ -573,6 +680,11 @@ QUICStream ClientQUIC::OpenStream(bool unidirectional) {
|
|||
QUICStream stream;
|
||||
stream.impl = std::make_shared<QUICStream::Impl>();
|
||||
stream.impl->connection = this;
|
||||
// Count the stream as open before StreamOpen registers the callback, so a
|
||||
// callback can never observe a half-initialised Impl. The two failure
|
||||
// paths below undo it.
|
||||
stream.impl->registry = impl->streams;
|
||||
stream.impl->registry->Add();
|
||||
stream.impl->selfRef = new std::shared_ptr<QUICStream::Impl>(stream.impl);
|
||||
QUIC_STREAM_OPEN_FLAGS openFlags = unidirectional
|
||||
? QUIC_STREAM_OPEN_FLAG_UNIDIRECTIONAL
|
||||
|
|
@ -583,6 +695,9 @@ QUICStream ClientQUIC::OpenStream(bool unidirectional) {
|
|||
if (QUIC_FAILED(s)) {
|
||||
delete stream.impl->selfRef;
|
||||
stream.impl->selfRef = nullptr;
|
||||
// No msquic stream exists, so FinalizeClose is not the right undo.
|
||||
std::shared_ptr<StreamRegistry> registry = std::move(stream.impl->registry);
|
||||
registry->Remove();
|
||||
throw QUICException(std::format("StreamOpen failed: 0x{:x}", static_cast<unsigned>(s)));
|
||||
}
|
||||
stream.handle = streamHandle;
|
||||
|
|
@ -658,3 +773,8 @@ std::vector<char> ClientQUIC::RecieveDatagramSync() {
|
|||
}
|
||||
|
||||
HQUIC ClientQUIC::GetHandle() const { return impl ? impl->connection : nullptr; }
|
||||
|
||||
std::shared_ptr<StreamRegistry> QUICStream::Impl::RegistryOf(ClientQUIC* connection) {
|
||||
if (!connection || !connection->impl) return nullptr;
|
||||
return connection->impl->streams;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -149,9 +149,11 @@ namespace Crafter {
|
|||
// via OnStream()).
|
||||
// - Unreliable, unordered datagrams (SendDatagram() / OnDatagram()).
|
||||
//
|
||||
// Lifetime: ~ClientQUIC closes the connection. Streams obtained from
|
||||
// OpenStream() are scoped to the connection and must be destroyed (or
|
||||
// moved out) before the ClientQUIC.
|
||||
// Lifetime: ~ClientQUIC shuts the connection down and closes it, waiting
|
||||
// (bounded — 5s) for every stream on it to finish closing first, since
|
||||
// QUICStream::Stop only initiates a shutdown that completes asynchronously.
|
||||
// Streams outliving the connection are therefore safe to hold, though they
|
||||
// are dead once it goes; prefer destroying or moving them out first.
|
||||
//
|
||||
// Browser build: the only QUIC-shaped API the browser exposes is
|
||||
// WebTransport, which is HTTP/3-based and reached at a fixed URL. Here:
|
||||
|
|
|
|||
Loading…
Reference in a new issue