From 94f2183057aca38349234379f485a8bf50888994 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?McKayla=20=E3=81=AF=E3=81=AA?= Date: Wed, 2 Sep 2026 00:33:04 +0000 Subject: [PATCH] fix: report failed Coder Connect tunnels --- Tests.Vpn.Service/ManagerTest.cs | 121 ++++++++++++++++++++++++++++++- Vpn.Service/Manager.cs | 36 +++++++-- Vpn.Service/TunnelSupervisor.cs | 64 ++++++++++++---- 3 files changed, 196 insertions(+), 25 deletions(-) diff --git a/Tests.Vpn.Service/ManagerTest.cs b/Tests.Vpn.Service/ManagerTest.cs index f003a0b..0014398 100644 --- a/Tests.Vpn.Service/ManagerTest.cs +++ b/Tests.Vpn.Service/ManagerTest.cs @@ -1,3 +1,4 @@ +using System.Reflection; using Coder.Desktop.Vpn; using Coder.Desktop.Vpn.Proto; using Coder.Desktop.Vpn.Service; @@ -11,6 +12,7 @@ internal class FakeTunnelSupervisor(RpcVersion? negotiatedVersion, Exception? se { public List SentRequests { get; } = []; public TaskCompletionSource Sent { get; } = new(); + public int StopCalls { get; private set; } public RpcVersion? NegotiatedVersion => negotiatedVersion; @@ -26,7 +28,11 @@ public Task StartAsync(string binPath, Speaker.On Speaker.OnErrorDelegate errorHandler, CancellationToken ct = default) => throw new NotImplementedException(); - public Task StopAsync(CancellationToken ct = default) => Task.CompletedTask; + public Task StopAsync(CancellationToken ct = default) + { + StopCalls++; + return Task.CompletedTask; + } public Task SendMessage(ManagerMessage message, CancellationToken ct = default) => throw new NotImplementedException(); @@ -51,9 +57,19 @@ internal class FakeManagerRpc : IManagerRpc public event IManagerRpc.OnReceiveHandler? OnReceive; #pragma warning restore CS0067 + public List Broadcasts { get; } = []; + public TaskCompletionSource Broadcast { get; } = new(); + public Task StopAsync(CancellationToken cancellationToken) => Task.CompletedTask; public Task ExecuteAsync(CancellationToken stoppingToken) => Task.CompletedTask; - public Task BroadcastAsync(ServiceMessage message, CancellationToken ct = default) => Task.CompletedTask; + + public Task BroadcastAsync(ServiceMessage message, CancellationToken ct = default) + { + Broadcasts.Add(message.Clone()); + Broadcast.TrySetResult(); + return Task.CompletedTask; + } + public ValueTask DisposeAsync() => ValueTask.CompletedTask; } @@ -72,11 +88,51 @@ internal class FakeTelemetryEnricher : ITelemetryEnricher [TestFixture] public class ManagerTest { - private static Manager NewManager(ITunnelSupervisor tunnelSupervisor, ISystemResumeMonitor? resumeMonitor = null) + private static Manager NewManager(ITunnelSupervisor tunnelSupervisor, ISystemResumeMonitor? resumeMonitor = null, + IManagerRpc? managerRpc = null) => new(Options.Create(new ManagerConfig()), NullLogger.Instance, new FakeDownloader(), - tunnelSupervisor, new FakeManagerRpc(), new FakeTelemetryEnricher(), + tunnelSupervisor, managerRpc ?? new FakeManagerRpc(), new FakeTelemetryEnricher(), resumeMonitor ?? new FakeSystemResumeMonitor()); + private static long GetTunnelGeneration(Manager manager) + { + var generationField = typeof(Manager).GetField("_tunnelGeneration", + BindingFlags.Instance | BindingFlags.NonPublic); + Assert.That(generationField, Is.Not.Null); + return (long)generationField!.GetValue(manager)!; + } + + private static void RaiseTunnelRpcError(Manager manager, long tunnelGeneration, Exception error) + { + var handler = typeof(Manager).GetMethod("HandleTunnelRpcError", BindingFlags.Instance | BindingFlags.NonPublic); + Assert.That(handler, Is.Not.Null); + handler!.Invoke(manager, [tunnelGeneration, error]); + } + + private static void AddPeer(Manager manager) + { + var handler = typeof(Manager).GetMethod("HandleTunnelMessagePeerUpdate", + BindingFlags.Instance | BindingFlags.NonPublic); + Assert.That(handler, Is.Not.Null); + handler!.Invoke(manager, + [ + new TunnelMessage + { + PeerUpdate = new PeerUpdate + { + UpsertedWorkspaces = + { + new Workspace { Id = Google.Protobuf.ByteString.CopyFrom(new byte[16]), Name = "workspace" }, + }, + UpsertedAgents = + { + new Agent { Id = Google.Protobuf.ByteString.CopyFrom(new byte[16]), Name = "agent" }, + }, + }, + }, + ]); + } + [Test(Description = "Send a wake request to the tunnel on system resume")] [CancelAfter(30_000)] public async Task SendsWakeRequestOnResume(CancellationToken ct) @@ -115,4 +171,61 @@ public async Task IgnoresWakeRequestFailure(CancellationToken ct) Assert.That(supervisor.SentRequests, Has.Count.EqualTo(1)); } + + [Test(Description = "Tunnel RPC errors report the VPN as stopped")] + [CancelAfter(30_000)] + public async Task ReportsStoppedAfterTunnelRpcError(CancellationToken ct) + { + var supervisor = new FakeTunnelSupervisor(new RpcVersion(1, 3)); + var managerRpc = new FakeManagerRpc(); + using var manager = NewManager(supervisor, managerRpc: managerRpc); + AddPeer(manager); + + RaiseTunnelRpcError(manager, GetTunnelGeneration(manager), new IOException("tunnel disconnected")); + await managerRpc.Broadcast.Task.WaitAsync(ct); + + Assert.Multiple(() => + { + Assert.That(supervisor.StopCalls, Is.Zero); + Assert.That(managerRpc.Broadcasts, Has.Count.EqualTo(1)); + Assert.That(managerRpc.Broadcasts[0].Status.Lifecycle, Is.EqualTo(Status.Types.Lifecycle.Stopped)); + Assert.That(managerRpc.Broadcasts[0].Status.PeerUpdate.UpsertedAgents, Is.Empty); + Assert.That(managerRpc.Broadcasts[0].Status.PeerUpdate.UpsertedWorkspaces, Is.Empty); + }); + } + + [Test(Description = "Errors from a replaced tunnel do not change VPN status")] + [CancelAfter(30_000)] + public async Task IgnoresErrorFromReplacedTunnel(CancellationToken ct) + { + var supervisor = new FakeTunnelSupervisor(new RpcVersion(1, 3)); + var managerRpc = new FakeManagerRpc(); + using var manager = NewManager(supervisor, managerRpc: managerRpc); + + RaiseTunnelRpcError(manager, GetTunnelGeneration(manager) - 1, new IOException("old tunnel disconnected")); + await Task.Delay(100, ct); + + Assert.That(managerRpc.Broadcasts, Is.Empty); + } + + [Test(Description = "Stopping the manager reports the VPN as stopped")] + [CancelAfter(30_000)] + public async Task ReportsStoppedWhenManagerStops(CancellationToken ct) + { + var supervisor = new FakeTunnelSupervisor(new RpcVersion(1, 3)); + var managerRpc = new FakeManagerRpc(); + using var manager = NewManager(supervisor, managerRpc: managerRpc); + AddPeer(manager); + + await manager.StopAsync(ct); + + Assert.Multiple(() => + { + Assert.That(supervisor.StopCalls, Is.EqualTo(1)); + Assert.That(managerRpc.Broadcasts, Has.Count.EqualTo(1)); + Assert.That(managerRpc.Broadcasts[0].Status.Lifecycle, Is.EqualTo(Status.Types.Lifecycle.Stopped)); + Assert.That(managerRpc.Broadcasts[0].Status.PeerUpdate.UpsertedAgents, Is.Empty); + Assert.That(managerRpc.Broadcasts[0].Status.PeerUpdate.UpsertedWorkspaces, Is.Empty); + }); + } } diff --git a/Vpn.Service/Manager.cs b/Vpn.Service/Manager.cs index 9c3d396..0a4fc7b 100644 --- a/Vpn.Service/Manager.cs +++ b/Vpn.Service/Manager.cs @@ -39,6 +39,7 @@ public class Manager : IManager private readonly ISystemResumeMonitor _systemResumeMonitor; private volatile TunnelStatus _status = TunnelStatus.Stopped; + private long _tunnelGeneration; // TunnelSupervisor already has protections against concurrent operations, // but all the other stuff before starting the tunnel does not. @@ -75,8 +76,16 @@ public void Dispose() public async Task StopAsync(CancellationToken ct = default) { - await _tunnelSupervisor.StopAsync(ct); - await BroadcastStatus(null, ct); + Interlocked.Increment(ref _tunnelGeneration); + try + { + await _tunnelSupervisor.StopAsync(ct); + } + finally + { + ClearPeers(); + await BroadcastStatus(TunnelStatus.Stopped, ct); + } } private void HandleSystemResumed(object? sender, EventArgs e) @@ -200,6 +209,7 @@ private async ValueTask HandleClientMessageStart(ClientMessage me await BroadcastStatus(TunnelStatus.Starting, ct); _lastStartRequest = message.Start; _lastServerVersion = serverVersion; + var tunnelGeneration = Interlocked.Increment(ref _tunnelGeneration); // TODO: each section of this operation needs a timeout @@ -211,7 +221,7 @@ private async ValueTask HandleClientMessageStart(ClientMessage me await BroadcastStartProgress(StartProgressStage.Finalizing, cancellationToken: ct); await _tunnelSupervisor.StartAsync(_config.TunnelBinaryPath, HandleTunnelRpcMessage, - HandleTunnelRpcError, + error => HandleTunnelRpcError(tunnelGeneration, error), ct); var reply = await _tunnelSupervisor.SendRequestAwaitReply(new ManagerMessage @@ -259,6 +269,7 @@ private async ValueTask HandleClientMessageStop(ClientMessage mess { try { + Interlocked.Increment(ref _tunnelGeneration); ClearPeers(); await BroadcastStatus(TunnelStatus.Stopping, ct); // This will handle sending the Stop message to the tunnel for us. @@ -395,18 +406,27 @@ private async Task FallibleBroadcast(ServiceMessage message, CancellationToken c } } - private void HandleTunnelRpcError(Exception e) + private void HandleTunnelRpcError(long tunnelGeneration, Exception e) { _logger.LogError(e, "Manager<->Tunnel RPC error"); + _ = HandleTunnelRpcErrorAsync(tunnelGeneration); + } + + private async Task HandleTunnelRpcErrorAsync(long tunnelGeneration) + { try { - _tunnelSupervisor.StopAsync(); + // Serialize the failure transition with start and stop operations + // so an in-flight start cannot overwrite it with Started. + using var operationLock = await _tunnelOperationLock.LockAsync(); + if (tunnelGeneration != Interlocked.Read(ref _tunnelGeneration)) return; + ClearPeers(); - BroadcastStatus().Wait(); + await BroadcastStatus(TunnelStatus.Stopped); } - catch (Exception e2) + catch (Exception error) { - _logger.LogError(e2, "Failed to stop tunnel supervisor after RPC error"); + _logger.LogError(error, "Failed to update status after tunnel RPC error"); } } diff --git a/Vpn.Service/TunnelSupervisor.cs b/Vpn.Service/TunnelSupervisor.cs index 16e7e8e..2c11caf 100644 --- a/Vpn.Service/TunnelSupervisor.cs +++ b/Vpn.Service/TunnelSupervisor.cs @@ -22,8 +22,7 @@ public interface ITunnelSupervisor : IAsyncDisposable /// Path to the executable /// Handler to call with each RPC message /// - /// Handler for permanent errors from the RPC Speaker. The recipient should call StopAsync after - /// receiving this. + /// Handler for permanent errors from the RPC Speaker or tunnel subprocess. /// /// Cancellation token public Task StartAsync(string binPath, @@ -135,7 +134,8 @@ public async Task StartAsync(string binPath, // We don't use the supplied CancellationToken here because we want it to only apply to the startup // procedure. - _ = _subprocess.WaitForExitAsync(_cts.Token).ContinueWith(OnProcessExited, CancellationToken.None); + var subprocess = _subprocess; + _ = MonitorProcessExitAsync(subprocess, errorHandler); // Start the RPC Speaker. try @@ -143,7 +143,8 @@ public async Task StartAsync(string binPath, var stream = new BidirectionalPipe(_inPipe, _outPipe); _speaker = new Speaker(stream); _speaker.Receive += messageHandler; - _speaker.Error += errorHandler; + _speaker.Error += error => + _ = HandleSubprocessFailureAsync(subprocess, error, errorHandler); // Handshakes already have a 5-second timeout. await _speaker.StartAsync(ct); } @@ -225,26 +226,63 @@ public async ValueTask SendRequestAwaitReply(ManagerMessage messa public async ValueTask DisposeAsync() { - _cts.Dispose(); - await CleanupAsync(); + await _cts.CancelAsync(); + await _operationLock.WaitAsync(); + try + { + await CleanupAsync(); + } + finally + { + _operationLock.Release(); + _cts.Dispose(); + } GC.SuppressFinalize(this); } - private async Task OnProcessExited(Task task) + private async Task MonitorProcessExitAsync(Process subprocess, + Speaker.OnErrorDelegate errorHandler) { - if (task.IsFaulted) - _logger.LogError(task.Exception, "OnProcessExited: subprocess task exited with an exception"); - if (!await _operationLock.WaitAsync(0)) + try + { + await subprocess.WaitForExitAsync(_cts.Token); + } + catch (OperationCanceledException) { - _logger.LogInformation("OnProcessExited: could not acquire operation lock to perform cleanup"); return; } + catch (Exception e) + { + _logger.LogError(e, "Failed while waiting for tunnel subprocess to exit"); + return; + } + + await HandleSubprocessFailureAsync(subprocess, + new InvalidOperationException("Tunnel subprocess exited unexpectedly"), + errorHandler); + } + private async Task HandleSubprocessFailureAsync(Process subprocess, Exception error, + Speaker.OnErrorDelegate errorHandler) + { + await _operationLock.WaitAsync(); try { + // Cleanup of an old or intentionally stopped process clears this + // reference before releasing the operation lock. + if (!ReferenceEquals(subprocess, _subprocess)) return; + + _logger.LogError(error, "Tunnel subprocess failed"); + try + { + errorHandler(error); + } + catch (Exception handlerError) + { + _logger.LogError(handlerError, "Tunnel subprocess error handler failed"); + } + await CleanupAsync(); - _logger.LogInformation("OnProcessExited: subprocess exited with code {ExitCode}", - _subprocess?.ExitCode ?? -1); } finally {