DodoTech-Public/DodoSSH
Public Access
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity

Merge branch 'main' into the desktop updater, and give way on two numbers

Browse Source 
Main landed a realtime push feature while this branch was building the updater,
and the two collided in three places. Every one of them resolves the same way:
main got there first, so this branch moves.

**Two ADRs were both numbered 0012.** Main's is realtime push; this one is now
[ADR 0013](docs/adr/0013-desktop-distribution-and-updates.md). Git did not call
this a conflict — the filenames differ — so it would have merged quietly and left
the directory with two 0012s and every cross-reference ambiguous. Renumbered here
along with the nine places that point at it.

**Two manual-check phases were both numbered 15**, and that one git did catch.
Main's "Changes that arrive without a timer" keeps 15; installing and updating
the desktop client becomes Phase 16, with its checks and every reference to them
renumbered. The file's own rule is that a number is for life, which is exactly
why the one that had not been pushed is the one that gives way.

**The merge rewrote several files with CRLF**, and `.editorconfig` asks for LF on
everything except `*.ps1`. That is not cosmetic here: IDE0055 is an error and
`EnforceCodeStyleInBuild` is on, so it failed the build on three lines of
App.axaml.cs whose only change in this branch was an ADR number in a comment.
Forty-six files normalised back to LF; the release script keeps CRLF, which is
what `.gitattributes` and `.editorconfig` both already say for a PowerShell file.

Nothing else conflicted. The updater does not touch the sync loop or the event
stream, and the one file both sides edited heavily — MainWindowViewModel — merged
without a hunk in common.

Verified after merging: the solution restores locked and builds clean, and 304
shell, 100 layout, 54 session, 28 client-api and 25 contracts tests pass. The
first two counts are higher than before the merge because main's own tests came
with it and pass alongside these.
This commit is contained in:
jaap-jan
2026-08-04 17:52:57 +02:00
parent 6728a0a597 5a8731ca8c
commit 3ead865f01
52 changed files with 6205 additions and 341 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/DodoSSH.Client.Api/VaultEventStream.cs
+552
View File
@@ -0,0 +1,552 @@
using System.Net.WebSockets;
using System.Security.Cryptography;
using System.Text.Json;
using System.Threading.Channels;
using DodoSSH.Contracts;
namespace DodoSSH.Client.Api;
/// <summary>
/// A server's "pull now" notices, as everything above the transport needs them.
/// </summary>
/// <remarks>
/// <para>
/// A queue to read from rather than an event to subscribe to, and that shape is the point: the one
/// consumer is a synchronisation loop that already waits on a timer, so it can wait on this the same
/// way and keep every continuation on the thread it started from. An event would deliver on whichever
/// thread the socket happened to complete on, which in a user interface is the difference between
/// working and an intermittent rendering fault nobody can reproduce.
/// </para>
/// <para>
/// <b>Reading this is never how a change is applied.</b> A notice says which vault moved and nothing
/// else; the answer to it is the ordinary delta pull. See ADR 0012.
/// </para>
/// </remarks>
public interface IVaultEventStream : IDisposable
{
/// <summary>Whether a socket is currently established.</summary>
/// <remarks>
/// For the interface to say whether it is live, not for a caller to branch on before reading:
/// synchronising is correct whether or not this is true, because the timer is the fallback.
/// </remarks>
bool IsConnected { get; }
/// <summary>
/// Waits for the next notice.
/// </summary>
/// <remarks>
/// Connects on the first call and reconnects for as long as it is read, so a caller neither starts
/// nor restarts anything. A server that cannot be reached is not an error here — it is a wait that
/// has not finished — because the caller's alternative is the timer it is already running.
/// </remarks>
ValueTask<VaultEvent> ReadAsync(CancellationToken cancellationToken);
/// <summary>
/// Takes a notice if one is already waiting, without blocking.
/// </summary>
/// <returns>Whether there was one.</returns>
/// <remarks>
/// How a caller coalesces a burst. Five people saving at once produces five notices whose answer
/// is a single synchronisation pass, so the loop reads one, waits a moment, and swallows the rest
/// rather than running the same pull five times.
/// </remarks>
bool TryRead(out VaultEvent notice);
}
/// <summary>
/// A stream that never delivers anything.
/// </summary>
/// <remarks>
/// For a server that does not advertise the <c>events</c> feature, and for tests. Deliberately waits
/// for ever rather than completing: a caller selecting between this and a timer must fall through to
/// the timer, and a read that returned immediately would spin that loop as fast as the machine allows.
/// </remarks>
public sealed class IdleVaultEventStream : IVaultEventStream
{
/// <summary>The one instance. It holds nothing.</summary>
public static IdleVaultEventStream Instance { get; } = new();
/// <inheritdoc />
public bool IsConnected => false;
/// <inheritdoc />
public async ValueTask<VaultEvent> ReadAsync(CancellationToken cancellationToken)
{
await Task.Delay(System.Threading.Timeout.Infinite, cancellationToken).ConfigureAwait(false);
// Unreachable: the delay above only ever ends by throwing.
return new VaultEvent(VaultEventKinds.Ping);
}
/// <inheritdoc />
public bool TryRead(out VaultEvent notice)
{
notice = new VaultEvent(VaultEventKinds.Ping);
return false;
}
/// <inheritdoc />
public void Dispose()
{
// Nothing is held.
}
}
/// <summary>Tuning for <see cref="VaultEventStream"/>.</summary>
/// <remarks>
/// Every value here bounds a reconnection rather than a feature. With the socket permanently
/// unavailable the client synchronises on its timer, so the cost of getting these wrong is latency,
/// never correctness.
/// </remarks>
public sealed record VaultEventStreamOptions
{
/// <summary>The defaults.</summary>
public static VaultEventStreamOptions Default { get; } = new();
/// <summary>How long to wait before the first reconnection attempt.</summary>
public TimeSpan InitialBackoff { get; init; } = TimeSpan.FromSeconds(1);
/// <summary>The longest the backoff may grow to.</summary>
/// <remarks>
/// A minute, which is the polling interval: past that point reconnecting sooner buys nothing,
/// because the timer has already done the work the socket would have prompted.
/// </remarks>
public TimeSpan MaxBackoff { get; init; } = TimeSpan.FromMinutes(1);
/// <summary>
/// How long a socket may be silent before it is presumed dead.
/// </summary>
/// <remarks>
/// The server pings on an interval it states in its <c>hello</c>, so silence past a multiple of
/// that means the connection is gone rather than idle — which is otherwise indistinguishable, and
/// is exactly what a reverse proxy that quietly drops idle sockets produces. Used only until a
/// <c>hello</c> arrives; after that the server's own figure is trusted.
/// </remarks>
public TimeSpan InitialSilenceTimeout { get; init; } = TimeSpan.FromSeconds(90);
/// <summary>How many notices may be waiting before the oldest are dropped.</summary>
/// <remarks>
/// Small on purpose. A notice means "pull that vault", so a newer one subsumes the one it
/// displaces; a backlog would only make the loop pull repeatedly for work it has already done.
/// </remarks>
public int QueueDepth { get; init; } = 32;
}
/// <summary>
/// Holds a socket to one server open, and hands over what it says.
/// </summary>
/// <remarks>
/// <para>
/// The whole class is a reconnection policy. A dropped socket is the ordinary case — laptops sleep,
/// proxies time out, tokens expire, servers are redeployed — so nothing here treats a failure as
/// exceptional: it backs off and dials again, for as long as somebody is reading.
/// </para>
/// <para>
/// It is safe to have no server at all. Every failure path ends in "wait, then try again", and the
/// caller's synchronisation timer runs regardless, which is what makes it correct for this class to
/// stay silent about problems rather than surface them.
/// </para>
/// </remarks>
public sealed class VaultEventStream : IVaultEventStream, IAsyncDisposable
{
private readonly Uri endpoint;
private readonly IAccessTokenProvider tokens;
private readonly TimeProvider clock;
private readonly VaultEventStreamOptions options;
private readonly Func<Uri, string, CancellationToken, Task<WebSocket>> connect;
private readonly Channel<VaultEvent> notices;
private readonly CancellationTokenSource closing = new();
private readonly Lock starting = new();
private Task? pump;
private bool disposed;
/// <summary>Creates a stream against one server.</summary>
/// <param name="serverUrl">The server's base URL, as an ordinary <c>http</c> or <c>https</c> address.</param>
/// <param name="tokens">Supplies a bearer token, refreshing it when it is due.</param>
/// <param name="clock">Time source, for the backoff and the silence timeout.</param>
/// <param name="options">Tuning, or null for the defaults.</param>
public VaultEventStream(
Uri serverUrl,
IAccessTokenProvider tokens,
TimeProvider clock,
VaultEventStreamOptions? options = null)
: this(serverUrl, tokens, clock, DialAsync, options)
{
}
/// <remarks>
/// The connector is injected so the suite can drive this against a test host's in-memory socket.
/// Reconnection is the entire behaviour of this class, and testing it against a real network would
/// mean testing it against the one thing that cannot be made to fail on demand.
/// </remarks>
internal VaultEventStream(
Uri serverUrl,
IAccessTokenProvider tokens,
TimeProvider clock,
Func<Uri, string, CancellationToken, Task<WebSocket>> connect,
VaultEventStreamOptions? options = null)
{
ArgumentNullException.ThrowIfNull(serverUrl);
ArgumentNullException.ThrowIfNull(tokens);
ArgumentNullException.ThrowIfNull(clock);
ArgumentNullException.ThrowIfNull(connect);
endpoint = EventsUrl(serverUrl);
this.tokens = tokens;
this.clock = clock;
this.connect = connect;
this.options = options ?? VaultEventStreamOptions.Default;
notices = Channel.CreateBounded<VaultEvent>(new BoundedChannelOptions(this.options.QueueDepth)
{
FullMode = BoundedChannelFullMode.DropOldest,
SingleReader = true,
SingleWriter = true,
});
}
/// <inheritdoc />
public bool IsConnected { get; private set; }
/// <inheritdoc />
public ValueTask<VaultEvent> ReadAsync(CancellationToken cancellationToken)
{
ObjectDisposedException.ThrowIf(disposed, this);
Start();
return notices.Reader.ReadAsync(cancellationToken);
}
/// <inheritdoc />
/// <remarks>
/// Does not start the connection, unlike <see cref="ReadAsync"/>: a caller draining a burst has
/// already read one notice, and "is there another right now" is not a reason to dial a server.
/// </remarks>
public bool TryRead(out VaultEvent notice) => notices.Reader.TryRead(out notice!);
/// <summary>
/// Ends the connection, without waiting for it to unwind.
/// </summary>
/// <remarks>
/// What a shell calls when a connection is dropped, from a synchronous path that must not block —
/// <c>IVaultServer</c> is <see cref="IDisposable"/>, and blocking on a socket teardown from the
/// user-interface thread is exactly the sync-over-async this repository bans. Cancelling is enough:
/// every loop reads the token, and the pump has nothing to flush.
/// </remarks>
public void Dispose()
{
if (disposed)
{
return;
}
disposed = true;
closing.Cancel();
// The source is deliberately left undisposed. The pump may still be inside a linked token
// source derived from this one, and disposing a parent out from under a live child is how a
// clean shutdown becomes an ObjectDisposedException on a background thread. It holds no timer
// and no handle once cancelled; DisposeAsync is the path that cleans it up properly.
}
/// <summary>Ends the connection and waits for it to unwind.</summary>
/// <remarks>The deterministic form, for a caller that can await one — tests, mostly.</remarks>
public async ValueTask DisposeAsync()
{
if (disposed)
{
return;
}
disposed = true;
await closing.CancelAsync().ConfigureAwait(false);
if (pump is not null)
{
try
{
await pump.ConfigureAwait(false);
}
catch (OperationCanceledException)
{
// The point of the cancel above.
}
}
closing.Dispose();
}
/// <summary>
/// Turns a server's base URL into its event socket's.
/// </summary>
/// <remarks>
/// The path is replaced rather than appended, matching every other call in this client: request
/// paths here are absolute — <c>/api/v1/…</c> — so a deployment behind a path prefix is already
/// unsupported, and pretending otherwise in this one place would be a difference nobody could act
/// on.
/// </remarks>
private static Uri EventsUrl(Uri serverUrl) =>
new UriBuilder(serverUrl)
{
Scheme = string.Equals(serverUrl.Scheme, Uri.UriSchemeHttps, StringComparison.OrdinalIgnoreCase)
? "wss"
: "ws",
Path = VaultEvents.Path,
Query = string.Empty,
Fragment = string.Empty,
}.Uri;
private static async Task<WebSocket> DialAsync(Uri url, string token, CancellationToken cancellationToken)
{
var socket = new ClientWebSocket();
try
{
socket.Options.AddSubProtocol(VaultEvents.SubProtocol);
// A header rather than the Sec-WebSocket-Protocol smuggling ADR 0004 needs for the relay:
// this client is a native application and can set one, and the token here is the ordinary
// bearer credential rather than a ticket.
socket.Options.SetRequestHeader("Authorization", $"Bearer {token}");
await socket.ConnectAsync(url, cancellationToken).ConfigureAwait(false);
return socket;
}
catch
{
socket.Dispose();
throw;
}
}
private void Start()
{
if (pump is not null)
{
return;
}
lock (starting)
{
pump ??= Task.Run(() => RunAsync(closing.Token), closing.Token);
}
}
/// <summary>Connects, reads until it cannot, waits, and does it again.</summary>
private async Task RunAsync(CancellationToken cancellationToken)
{
var backoff = options.InitialBackoff;
while (!cancellationToken.IsCancellationRequested)
{
var outcome = await AttemptAsync(cancellationToken).ConfigureAwait(false);
// A socket that lived long enough to say hello proves the server is there and willing, so
// the next failure starts from the bottom again rather than inheriting the backoff that
// got us here. Without this a laptop that woke, connected, and then lost its network an
// hour later would wait a full minute before trying, having already proved it need not.
if (outcome == Outcome.Established)
{
backoff = options.InitialBackoff;
}
// The server said this token is spent, which the token provider can fix without waiting.
// Reconnecting at once is the whole reason that close code is distinct.
var wait = outcome == Outcome.TokenExpired ? TimeSpan.Zero : Jitter(backoff);
if (wait > TimeSpan.Zero)
{
try
{
await Task.Delay(wait, clock, cancellationToken).ConfigureAwait(false);
}
catch (OperationCanceledException)
{
return;
}
backoff = backoff < options.MaxBackoff
? Shorter(backoff * 2, options.MaxBackoff)
: options.MaxBackoff;
}
}
}
/// <summary>One connection, from dial to close.</summary>
private async Task<Outcome> AttemptAsync(CancellationToken cancellationToken)
{
WebSocket? socket = null;
try
{
var token = await tokens.GetAccessTokenAsync(cancellationToken).ConfigureAwait(false);
socket = await connect(endpoint, token, cancellationToken).ConfigureAwait(false);
IsConnected = true;
return await PumpAsync(socket, cancellationToken).ConfigureAwait(false);
}
catch (OperationCanceledException)
{
return Outcome.Cancelled;
}
catch (Exception exception) when (exception is not OutOfMemoryException)
{
// Every failure this can meet — no network, a refused upgrade, a server that has not been
// deployed with this feature, a token that cannot be refreshed — has the same remedy, and
// none of them is worth telling a user about. The synchronisation timer is still running.
return Outcome.Failed;
}
finally
{
IsConnected = false;
socket?.Dispose();
}
}
/// <summary>Reads frames until the socket ends or goes quiet.</summary>
private async Task<Outcome> PumpAsync(WebSocket socket, CancellationToken cancellationToken)
{
var buffer = new byte[8 * 1024];
var silence = options.InitialSilenceTimeout;
var established = false;
while (socket.State == WebSocketState.Open && !cancellationToken.IsCancellationRequested)
{
// Rebuilt per frame rather than reset, because a linked source cannot be un-cancelled and
// the deadline is what detects a socket that has silently gone away.
using var deadline = new CancellationTokenSource(silence, clock);
using var quiet = CancellationTokenSource.CreateLinkedTokenSource(
cancellationToken, deadline.Token);
WebSocketReceiveResult received;
try
{
received = await socket.ReceiveAsync(buffer, quiet.Token).ConfigureAwait(false);
}
catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested)
{
// Silent for longer than the server said it would be. The socket is gone in a way that
// only reconnecting can discover, which is what a proxy dropping an idle connection
// looks like from this end.
return Ended(established);
}
if (received.MessageType == WebSocketMessageType.Close)
{
return (int?)received.CloseStatus == VaultEvents.TokenExpiredCloseCode
? Outcome.TokenExpired
: Ended(established);
}
// Binary is reserved by ADR 0012 for shared-session data, and text that arrived in pieces
// is longer than anything this protocol defines. Skipped rather than fatal, so a newer
// server does not cost this client its push for the whole session.
if (received.MessageType != WebSocketMessageType.Text || !received.EndOfMessage)
{
continue;
}
if (Parse(buffer.AsSpan(0, received.Count)) is not { } frame)
{
continue;
}
established = true;
silence = await AbsorbAsync(socket, frame, silence, cancellationToken).ConfigureAwait(false);
}
return Ended(established);
}
/// <summary>Deals with one frame, and says how long the socket may now stay quiet.</summary>
private async Task<TimeSpan> AbsorbAsync(
WebSocket socket,
VaultEvent frame,
TimeSpan silence,
CancellationToken cancellationToken)
{
if (frame.HeartbeatSeconds is > 0 and var seconds)
{
// Three missed heartbeats. Two is within one paused thread of a false positive, and a
// false positive here costs a reconnection rather than anything a user sees.
silence = TimeSpan.FromSeconds(seconds * 3);
}
if (string.Equals(frame.Kind, VaultEventKinds.Ping, StringComparison.Ordinal))
{
await socket.SendAsync(
JsonSerializer.SerializeToUtf8Bytes(
new VaultEvent(VaultEventKinds.Pong), DodoSshJsonContext.Default.VaultEvent),
WebSocketMessageType.Text,
endOfMessage: true,
cancellationToken)
.ConfigureAwait(false);
return silence;
}
// Everything else, including a kind this build has never heard of, goes to the reader — which
// is what makes the frame table extensible. An unrecognised kind is one the caller ignores;
// refusing it here would be this class deciding what a newer server may say.
notices.Writer.TryWrite(frame);
return silence;
}
private static Outcome Ended(bool established) =>
established ? Outcome.Established : Outcome.Failed;
private static VaultEvent? Parse(ReadOnlySpan<byte> utf8)
{
try
{
return JsonSerializer.Deserialize(utf8, DodoSshJsonContext.Default.VaultEvent);
}
catch (JsonException)
{
return null;
}
}
/// <summary>
/// Spreads reconnections out, so a server that restarts is not met by every client at once.
/// </summary>
/// <remarks>
/// <c>RandomNumberGenerator</c> because <c>System.Random</c> is banned repo-wide. Nothing here is
/// security-relevant — the ban exists so that nothing key-, token- or nonce-adjacent can reach for
/// the weak one by habit, and paying a few microseconds to keep that rule absolute is the cheaper
/// side of the trade.
/// </remarks>
private static TimeSpan Jitter(TimeSpan delay)
{
var milliseconds = (int)Math.Clamp(delay.TotalMilliseconds, 1, int.MaxValue / 2);
return TimeSpan.FromMilliseconds(
milliseconds + RandomNumberGenerator.GetInt32(0, Math.Max(1, milliseconds / 2)));
}
private static TimeSpan Shorter(TimeSpan left, TimeSpan right) => left < right ? left : right;
/// <summary>How one connection attempt ended.</summary>
private enum Outcome
{
/// <summary>Never got as far as a frame. Back off.</summary>
Failed,
/// <summary>Ran, and then ended. Back off, but from the bottom.</summary>
Established,
/// <summary>The server closed it because the token expired. Reconnect at once with a new one.</summary>
TokenExpired,
/// <summary>The stream is being disposed.</summary>
Cancelled,
}
}