Public Access
jaap-jan
c8848ec1e3
The events endpoint cancelled its pump token and only then wrote the close frame. The token was also the one passed to socket.ReceiveAsync and SendAsync, and cancelling a pending operation on the ManagedWebSocket that Kestrel hands over aborts it. CloseAsync then saw an Aborted socket and returned, so in production a client whose token expired got a TCP reset rather than 4401 "reconnect with a fresh token", and a shutdown sent no 1001 either. The tests missed it because TestServer's in-memory socket survives that cancellation: the same scenario through TestServer closes with 4401. PumpAsync now does what the relay's pump does. No WebSocket operation is cancelled; the token stops only the outbound queue, the clock and the access refresh. The sender is let finish its last write, the close goes out, and the receiver ends on the client's answering close. Each wait is bounded by the close timeout, and the socket is aborted only when a peer stops answering. ApiFixture gains ConnectEventsOverTcpAsync, which runs a second host on real Kestrel over loopback, sharing the container and identity stub. The new AnExpiringToken_ClosesTheSocketWith4401 fails with a reset before this change and passes after it.
575 lines
24 KiB
C#
575 lines
24 KiB
C#
using System.Collections.Frozen;
|
|
using System.Globalization;
|
|
using System.Net.WebSockets;
|
|
using System.Security.Claims;
|
|
using System.Text.Json;
|
|
using DodoSSH.Api.Authorization;
|
|
using DodoSSH.Api.Setup;
|
|
using DodoSSH.Contracts;
|
|
using DodoSSH.Domain.Authorization;
|
|
using FastEndpoints;
|
|
using Microsoft.Extensions.Options;
|
|
|
|
namespace DodoSSH.Api.Features.Events;
|
|
|
|
/// <summary>
|
|
/// The socket that says "pull now" so a client does not have to wait for its timer.
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// <para>
|
|
/// Everything this endpoint sends is a <em>notice</em>. It never carries an item, a payload or a
|
|
/// cursor: the client's answer to a notice is the delta pull it would have run on its own anyway, so
|
|
/// there is exactly one code path that applies a change and this is not it. See ADR 0012 for why
|
|
/// pushing the items themselves is refused.
|
|
/// </para>
|
|
/// <para>
|
|
/// The bearer token authorises the upgrade, unlike the relay's ticket in ADR 0004. The relay's socket
|
|
/// is a byte pipe whose whole authorization decision is made before it opens; this one is a view of
|
|
/// the caller's own vault list and has to keep answering "what may this account read" for as long as
|
|
/// it is held. Its two bounds on that — the token's own expiry, and a periodic re-resolve — are in
|
|
/// <see cref="MindAsync"/>.
|
|
/// </para>
|
|
/// </remarks>
|
|
internal sealed class VaultEventsEndpoint(
|
|
IServiceScopeFactory scopes,
|
|
VaultEventHub hub,
|
|
IOptions<EventsOptions> options,
|
|
TimeProvider clock,
|
|
IHostApplicationLifetime lifetime,
|
|
ILogger<VaultEventsEndpoint> logger)
|
|
: EndpointWithoutRequest
|
|
{
|
|
/// <summary>
|
|
/// The largest message this endpoint will read from a client.
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// A client sends nothing but <c>ping</c>, so the cap is three orders of magnitude of headroom and
|
|
/// still small enough that a hostile client cannot make the server buffer anything worth having.
|
|
/// </remarks>
|
|
private const int MaxInboundFrameBytes = 4 * 1024;
|
|
|
|
/// <summary>How long to wait for the close handshake before dropping the socket.</summary>
|
|
private static readonly TimeSpan CloseTimeout = TimeSpan.FromSeconds(5);
|
|
|
|
/// <inheritdoc />
|
|
public override void Configure()
|
|
{
|
|
Get(VaultEvents.Path);
|
|
|
|
// Enrolled, matching sync. A caller with no identity key holds no vault key either, so every
|
|
// notice this socket could send is about ciphertext they cannot read.
|
|
Policies(Auth.EnrolledPolicy);
|
|
|
|
Description(b => b
|
|
.WithName("VaultEvents")
|
|
.WithSummary("Pushes a notice when a vault the caller can read has changed.")
|
|
.WithTags("Events"));
|
|
}
|
|
|
|
/// <inheritdoc />
|
|
public override async Task HandleAsync(CancellationToken ct)
|
|
{
|
|
if (await RefusedAsync().ConfigureAwait(false))
|
|
{
|
|
return;
|
|
}
|
|
|
|
var (userId, vaults) = await ResolveAccessAsync(ct).ConfigureAwait(false);
|
|
|
|
// Admitted before the upgrade so a refusal costs nothing, but answered *through* the socket
|
|
// rather than as an HTTP status: a constrained WebSocket client cannot read the status of a
|
|
// failed upgrade, and "you have too many open" is precisely the case where the client needs
|
|
// to know to back off rather than retry. Same reasoning as ADR 0004 on request headers.
|
|
var connection = hub.TryAdmit(userId, vaults);
|
|
|
|
try
|
|
{
|
|
using var socket = await HttpContext.WebSockets
|
|
.AcceptWebSocketAsync(new WebSocketAcceptContext { SubProtocol = VaultEvents.SubProtocol })
|
|
.ConfigureAwait(false);
|
|
|
|
if (connection is null)
|
|
{
|
|
await CloseAsync(
|
|
socket,
|
|
new Closure(
|
|
VaultEvents.TooManyConnectionsCloseCode, "Too many open event sockets."))
|
|
.ConfigureAwait(false);
|
|
|
|
return;
|
|
}
|
|
|
|
await PumpAsync(socket, connection, ct).ConfigureAwait(false);
|
|
}
|
|
finally
|
|
{
|
|
// Inside a try that starts *before* the upgrade, because an accept that throws — a client
|
|
// that abandoned the handshake — would otherwise leave an admitted connection in the hub
|
|
// for the life of the process, counting against this account's cap and taking a slot from
|
|
// the sockets that did open.
|
|
if (connection is not null)
|
|
{
|
|
hub.Remove(connection);
|
|
}
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Answers the requests that are not an event socket at all, as ordinary HTTP.
|
|
/// </summary>
|
|
/// <returns>Whether a response was sent and the handler should stop.</returns>
|
|
/// <remarks>
|
|
/// All three answers are problem documents rather than bare statuses, because each one has a
|
|
/// different remedy and a client that cannot tell them apart would retry the two that will never
|
|
/// succeed. Answered before the upgrade, so a caller that got the handshake wrong reads why in a
|
|
/// body rather than inferring it from a socket that closed.
|
|
/// </remarks>
|
|
private async Task<bool> RefusedAsync()
|
|
{
|
|
if (!options.Value.Enabled)
|
|
{
|
|
// 404 rather than 501: the feature is absent from this deployment, and /api/v1/meta does
|
|
// not advertise it. A client that dialled anyway keeps polling, which is correct.
|
|
await Send.ResultAsync(Problems.Coded(
|
|
StatusCodes.Status404NotFound,
|
|
ProblemCodes.EventsUnavailable,
|
|
"This server does not push vault changes. Synchronise on a timer instead; "
|
|
+ "GET /api/v1/meta lists the features it does offer."))
|
|
.ConfigureAwait(false);
|
|
|
|
return true;
|
|
}
|
|
|
|
if (!HttpContext.WebSockets.IsWebSocketRequest)
|
|
{
|
|
await Send.ResultAsync(Problems.Coded(
|
|
StatusCodes.Status400BadRequest,
|
|
ProblemCodes.MalformedRequest,
|
|
"This endpoint is a WebSocket. Send an upgrade request offering the "
|
|
+ $"'{VaultEvents.SubProtocol}' subprotocol."))
|
|
.ConfigureAwait(false);
|
|
|
|
return true;
|
|
}
|
|
|
|
// The subprotocol is this API's version negotiation for the socket, so an upgrade that does
|
|
// not offer it is refused rather than accepted and answered in a dialect the caller may not
|
|
// read. See VaultEvents.SubProtocol.
|
|
if (!HttpContext.WebSockets.WebSocketRequestedProtocols
|
|
.Contains(VaultEvents.SubProtocol, StringComparer.Ordinal))
|
|
{
|
|
await Send.ResultAsync(Problems.Coded(
|
|
StatusCodes.Status400BadRequest,
|
|
ProblemCodes.MalformedRequest,
|
|
$"This server speaks '{VaultEvents.SubProtocol}', which the upgrade request did "
|
|
+ "not offer."))
|
|
.ConfigureAwait(false);
|
|
|
|
return true;
|
|
}
|
|
|
|
return false;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Reads who the caller is and which vaults they may follow, in a scope of its own.
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// A fresh scope, disposed at once, rather than services injected into this endpoint — which is
|
|
/// the lesson ADR 0004 records paying for on the relay. This handler runs for as long as the
|
|
/// socket is open, so anything scoped it held would be a <c>DbContext</c> alive for hours, and a
|
|
/// few hundred of those exhaust the connection pool. The database is touched here and in
|
|
/// <see cref="RefreshAsync"/>, briefly, and nowhere else.
|
|
/// </remarks>
|
|
private async Task<(Guid UserId, FrozenSet<Guid> Vaults)> ResolveAccessAsync(
|
|
CancellationToken cancellationToken)
|
|
{
|
|
var scope = scopes.CreateAsyncScope();
|
|
await using var _ = scope.ConfigureAwait(false);
|
|
|
|
var currentUser = scope.ServiceProvider.GetRequiredService<ICurrentUserContext>();
|
|
var vaultAccess = scope.ServiceProvider.GetRequiredService<IVaultAccessService>();
|
|
|
|
var user = await currentUser.GetOrProvisionAsync(cancellationToken).ConfigureAwait(false);
|
|
|
|
return (user.Id, await ReachableAsync(vaultAccess, user.Id, cancellationToken).ConfigureAwait(false));
|
|
}
|
|
|
|
private static async Task<FrozenSet<Guid>> ReachableAsync(
|
|
IVaultAccessService vaultAccess,
|
|
Guid userId,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
var accessible = await vaultAccess.ListAsync(userId, cancellationToken).ConfigureAwait(false);
|
|
|
|
return accessible
|
|
.Where(access => access.Vault is not null
|
|
&& access.Permissions.HasFlag(PermissionFlags.Read))
|
|
.Select(access => access.Vault!.Id)
|
|
.ToFrozenSet();
|
|
}
|
|
|
|
/// <summary>Runs the socket until something ends it, then closes it politely.</summary>
|
|
/// <remarks>
|
|
/// <para>
|
|
/// Three loops rather than one: reading a socket and writing to it are independent waits, and the
|
|
/// clock is a third. They are joined by <see cref="Task.WhenAny(Task[])"/>, and the sender is
|
|
/// finished before the close is written, because a close frame racing a notice frame is a protocol
|
|
/// violation that presents as a client dropping its connection for no visible reason.
|
|
/// </para>
|
|
/// <para>
|
|
/// <b>No operation on the WebSocket is ever cancelled.</b> Cancelling a pending receive or send on
|
|
/// the socket Kestrel hands over aborts it, and an aborted socket cannot send the close frame that
|
|
/// is the only way a client learns <em>why</em> it was dropped — 4401 "reconnect with a fresh
|
|
/// token" against 1001 "the server is going away" against a failure to back off from. This was
|
|
/// once the case here, and it went unnoticed because <c>TestServer</c>'s in-memory socket survives
|
|
/// the cancellation that a real one does not. So the token stops the queue and the clock; the
|
|
/// sender is let finish its last write; the close goes out; and the receiver ends on the client's
|
|
/// answering close. Each wait is bounded, and the socket is aborted only when a peer has stopped
|
|
/// answering altogether. The relay's pump does the same, for the same reason.
|
|
/// </para>
|
|
/// </remarks>
|
|
private async Task PumpAsync(
|
|
WebSocket socket,
|
|
VaultEventConnection connection,
|
|
CancellationToken requestAborted)
|
|
{
|
|
var settings = options.Value;
|
|
|
|
using var pump = CancellationTokenSource.CreateLinkedTokenSource(
|
|
requestAborted, lifetime.ApplicationStopping);
|
|
|
|
var closure = new Closure(
|
|
(int)WebSocketCloseStatus.NormalClosure, string.Empty);
|
|
|
|
connection.TryEnqueue(new VaultEvent(
|
|
VaultEventKinds.Hello,
|
|
ServerTime: clock.GetUtcNow(),
|
|
HeartbeatSeconds: (int)settings.HeartbeatInterval.TotalSeconds,
|
|
VaultCount: connection.VaultCount));
|
|
|
|
var sending = SendAsync(socket, connection, pump.Token);
|
|
var receiving = ReceiveAsync(socket, connection);
|
|
var minding = MindAsync(connection, closure, pump.Token);
|
|
|
|
await Task.WhenAny(sending, receiving, minding).ConfigureAwait(false);
|
|
|
|
await pump.CancelAsync().ConfigureAwait(false);
|
|
|
|
// Nothing may still be mid-send when the close frame goes out. A send that does not finish is
|
|
// a client that has stopped reading, which will not read a close frame either.
|
|
if (!await SettlesAsync(sending).ConfigureAwait(false))
|
|
{
|
|
socket.Abort();
|
|
}
|
|
|
|
await SettlesAsync(minding).ConfigureAwait(false);
|
|
|
|
if (lifetime.ApplicationStopping.IsCancellationRequested)
|
|
{
|
|
// 1001 "going away", so a client knows to reconnect immediately rather than treating a
|
|
// rolling deployment as a server that has broken.
|
|
closure.Set((int)WebSocketCloseStatus.EndpointUnavailable, "The server is shutting down.");
|
|
}
|
|
|
|
EventsLog.ClosingConnection(logger, connection.UserId, closure.Reason);
|
|
|
|
await CloseAsync(socket, closure).ConfigureAwait(false);
|
|
|
|
// The client's answering close is what ends the receiver. Returning before it arrives would
|
|
// dispose the socket under a close frame still in flight.
|
|
if (!await SettlesAsync(receiving).ConfigureAwait(false))
|
|
{
|
|
socket.Abort();
|
|
await SettlesAsync(receiving).ConfigureAwait(false);
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Writes queued frames to the socket, one at a time.
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// The <em>only</em> writer, which is what makes concurrent sends impossible without a lock: the
|
|
/// heartbeat and the pong both go into the same queue rather than to the socket. A WebSocket
|
|
/// permits one send at a time and faults permanently on a second, so this is not a tidiness
|
|
/// preference.
|
|
/// </remarks>
|
|
private async Task SendAsync(
|
|
WebSocket socket,
|
|
VaultEventConnection connection,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
await foreach (var frame in connection.Outbound
|
|
.ReadAllAsync(cancellationToken)
|
|
.ConfigureAwait(false))
|
|
{
|
|
// Re-resolved before the notice is forwarded, not after: the client's answer to this frame
|
|
// is to re-read its vault list, and the point of a newly shared vault is that the *next*
|
|
// change to it produces a notice too. A socket that forwarded first would not follow the
|
|
// new vault until its next periodic refresh.
|
|
if (string.Equals(frame.Kind, VaultEventKinds.VaultsChanged, StringComparison.Ordinal))
|
|
{
|
|
await RefreshAsync(connection, cancellationToken).ConfigureAwait(false);
|
|
}
|
|
|
|
var bytes = JsonSerializer.SerializeToUtf8Bytes(frame, DodoSshJsonContext.Default.VaultEvent);
|
|
|
|
// Not cancellable, deliberately. See PumpAsync.
|
|
await socket
|
|
.SendAsync(bytes, WebSocketMessageType.Text, endOfMessage: true, CancellationToken.None)
|
|
.ConfigureAwait(false);
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Reads what the client sends, which in this version is heartbeats and a close.
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// <para>
|
|
/// A client cannot ask to follow a vault, and that is deliberate rather than unfinished: a
|
|
/// <c>subscribe(vaultId)</c> frame is an existence oracle for vault ids, which is the disclosure
|
|
/// <c>SyncPullEndpoint</c> answers 404 rather than 403 to avoid. Subscription is decided from the
|
|
/// caller's access and nothing else.
|
|
/// </para>
|
|
/// <para>
|
|
/// It takes no token: it ends on the client's close, a frame it refuses, or the socket failing —
|
|
/// and when the server is the one ending the connection, the close it sends is what prompts the
|
|
/// first of those. See <see cref="PumpAsync"/>.
|
|
/// </para>
|
|
/// </remarks>
|
|
private static async Task ReceiveAsync(WebSocket socket, VaultEventConnection connection)
|
|
{
|
|
var buffer = new byte[MaxInboundFrameBytes];
|
|
|
|
while (true)
|
|
{
|
|
// Not cancellable, deliberately. See PumpAsync.
|
|
var received = await socket.ReceiveAsync(buffer, CancellationToken.None).ConfigureAwait(false);
|
|
|
|
if (received.MessageType == WebSocketMessageType.Close)
|
|
{
|
|
return;
|
|
}
|
|
|
|
// Oversized, or split across frames. Nothing this protocol sends is either, so the client
|
|
// is broken or probing; ending the socket is cheaper than reassembling for it.
|
|
if (!received.EndOfMessage)
|
|
{
|
|
return;
|
|
}
|
|
|
|
// Binary is unused in v1 and skipped rather than refused, because ADR 0012 reserves it for
|
|
// shared-session data — an older server meeting a newer client must ignore those, not
|
|
// close on them.
|
|
if (received.MessageType != WebSocketMessageType.Text)
|
|
{
|
|
continue;
|
|
}
|
|
|
|
if (Kind(buffer.AsSpan(0, received.Count)) is VaultEventKinds.Ping)
|
|
{
|
|
connection.TryEnqueue(new VaultEvent(VaultEventKinds.Pong));
|
|
}
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Reads a frame's kind, or null if it is not one this server understands.
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// A frame that will not parse is skipped rather than closing the socket. This is a control
|
|
/// channel whose failure mode is "the client polls instead", so tolerating a frame from a newer
|
|
/// client costs nothing and refusing one costs that client its push for the whole session.
|
|
/// </remarks>
|
|
private static string? Kind(ReadOnlySpan<byte> utf8)
|
|
{
|
|
try
|
|
{
|
|
return JsonSerializer.Deserialize(utf8, DodoSshJsonContext.Default.VaultEvent)?.Kind;
|
|
}
|
|
catch (JsonException)
|
|
{
|
|
return null;
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Keeps the heartbeat going, the vault set current, and the socket inside its token's lifetime.
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// <para>
|
|
/// The deadline is the earlier of the access token's <c>exp</c> and a hard cap on how long any one
|
|
/// socket may live. Closing on expiry is what keeps a long-lived connection from outliving the
|
|
/// short-lived credential that authorised it; the client answers by reconnecting with a fresh
|
|
/// token, which is a sub-second gap in a channel that degrades to polling anyway.
|
|
/// </para>
|
|
/// <para>
|
|
/// The wait is the shorter of the heartbeat and the time left, so the deadline is met to within a
|
|
/// tick rather than to within a heartbeat.
|
|
/// </para>
|
|
/// </remarks>
|
|
private async Task MindAsync(
|
|
VaultEventConnection connection,
|
|
Closure closure,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
var settings = options.Value;
|
|
var started = clock.GetUtcNow();
|
|
|
|
var deadline = TokenExpiry() is { } expiry && expiry < started + settings.MaxConnectionDuration
|
|
? (Expiry: expiry, ForToken: true)
|
|
: (Expiry: started + settings.MaxConnectionDuration, ForToken: false);
|
|
|
|
var refreshed = started;
|
|
|
|
while (!cancellationToken.IsCancellationRequested)
|
|
{
|
|
var now = clock.GetUtcNow();
|
|
var remaining = deadline.Expiry - now;
|
|
|
|
if (remaining <= TimeSpan.Zero)
|
|
{
|
|
closure.Set(
|
|
deadline.ForToken
|
|
? VaultEvents.TokenExpiredCloseCode
|
|
: (int)WebSocketCloseStatus.NormalClosure,
|
|
deadline.ForToken
|
|
? "The access token has expired. Reconnect with a fresh one."
|
|
: "This connection reached its maximum lifetime.");
|
|
|
|
return;
|
|
}
|
|
|
|
var wait = settings.HeartbeatInterval < remaining ? settings.HeartbeatInterval : remaining;
|
|
|
|
await Task.Delay(wait, clock, cancellationToken).ConfigureAwait(false);
|
|
|
|
now = clock.GetUtcNow();
|
|
|
|
if (now - refreshed >= settings.AccessRefreshInterval)
|
|
{
|
|
// The backstop for a grant withdrawn while this socket was open. What it bounds is
|
|
// metadata — that a vault changed — because that is all a notice carries and reading
|
|
// the vault still needs a key this server has never held. See ADR 0012.
|
|
await RefreshAsync(connection, cancellationToken).ConfigureAwait(false);
|
|
refreshed = now;
|
|
}
|
|
|
|
connection.TryEnqueue(new VaultEvent(VaultEventKinds.Ping, ServerTime: now));
|
|
}
|
|
}
|
|
|
|
/// <summary>Re-reads which vaults this socket may follow.</summary>
|
|
/// <remarks>
|
|
/// A failure is logged and swallowed. The alternative is dropping a working socket because one
|
|
/// database call timed out, which would trade an occasionally stale vault set for an outage.
|
|
/// </remarks>
|
|
private async Task RefreshAsync(VaultEventConnection connection, CancellationToken cancellationToken)
|
|
{
|
|
try
|
|
{
|
|
var scope = scopes.CreateAsyncScope();
|
|
await using var _ = scope.ConfigureAwait(false);
|
|
|
|
var vaultAccess = scope.ServiceProvider.GetRequiredService<IVaultAccessService>();
|
|
|
|
connection.Resubscribe(
|
|
await ReachableAsync(vaultAccess, connection.UserId, cancellationToken)
|
|
.ConfigureAwait(false));
|
|
}
|
|
catch (OperationCanceledException)
|
|
{
|
|
// The socket is closing.
|
|
}
|
|
catch (Exception exception) when (exception is not OutOfMemoryException)
|
|
{
|
|
EventsLog.AccessRefreshFailed(logger, connection.UserId, exception);
|
|
}
|
|
}
|
|
|
|
/// <remarks>
|
|
/// <c>MapInboundClaims</c> is off — see <see cref="Auth"/> — so the claim is spelled as the
|
|
/// provider issued it rather than as a WS-Federation URI. Null is treated as "no bound from the
|
|
/// token", which the bearer handler's <c>RequireExpirationTime</c> should make unreachable; the
|
|
/// lifetime cap covers it either way.
|
|
/// </remarks>
|
|
private DateTimeOffset? TokenExpiry() =>
|
|
long.TryParse(
|
|
HttpContext.User.FindFirstValue("exp"),
|
|
NumberStyles.Integer,
|
|
CultureInfo.InvariantCulture,
|
|
out var seconds)
|
|
? DateTimeOffset.FromUnixTimeSeconds(seconds)
|
|
: null;
|
|
|
|
private static async Task CloseAsync(WebSocket socket, Closure closure)
|
|
{
|
|
if (socket.State is not (WebSocketState.Open or WebSocketState.CloseReceived))
|
|
{
|
|
return;
|
|
}
|
|
|
|
using var timeout = new CancellationTokenSource(CloseTimeout);
|
|
|
|
try
|
|
{
|
|
await socket
|
|
.CloseOutputAsync((WebSocketCloseStatus)closure.Code, closure.Reason, timeout.Token)
|
|
.ConfigureAwait(false);
|
|
}
|
|
catch (Exception exception)
|
|
when (exception is OperationCanceledException or WebSocketException or ObjectDisposedException)
|
|
{
|
|
// The peer is already gone. There is nothing to tell it and nothing to recover.
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Waits for a pump loop to finish, for at most the close timeout.
|
|
/// </summary>
|
|
/// <returns>Whether it finished.</returns>
|
|
/// <remarks>
|
|
/// Every one of these loops ends by being cancelled or by the socket going away, so an exception
|
|
/// here is the expected shape of "this connection is over" rather than a fault to propagate — and
|
|
/// propagating it would skip the close frame the other side is waiting for.
|
|
/// </remarks>
|
|
private static async Task<bool> SettlesAsync(Task loop)
|
|
{
|
|
try
|
|
{
|
|
await loop.WaitAsync(CloseTimeout).ConfigureAwait(false);
|
|
|
|
return true;
|
|
}
|
|
catch (TimeoutException)
|
|
{
|
|
return false;
|
|
}
|
|
catch (Exception exception)
|
|
when (exception is OperationCanceledException or WebSocketException or ObjectDisposedException)
|
|
{
|
|
return true;
|
|
}
|
|
}
|
|
|
|
/// <summary>Why the socket is being closed, decided by whichever loop ended first.</summary>
|
|
/// <remarks>
|
|
/// Mutable and shared, and safe without a lock for one specific reason: it is written only by
|
|
/// <see cref="MindAsync"/>, and read only after that loop has been awaited, which is a memory
|
|
/// barrier. If that loop outlived the close timeout, a late write races the read, and either
|
|
/// answer is still a true one.
|
|
/// </remarks>
|
|
private sealed class Closure(int code, string reason)
|
|
{
|
|
internal int Code { get; private set; } = code;
|
|
|
|
internal string Reason { get; private set; } = reason;
|
|
|
|
internal void Set(int code, string reason)
|
|
{
|
|
Code = code;
|
|
Reason = reason;
|
|
}
|
|
}
|
|
}
|