diff --git a/.github/workflows/build-and-test.yml b/.github/workflows/build-and-test.yml index 2a0f8b66c3..c89827ab91 100644 --- a/.github/workflows/build-and-test.yml +++ b/.github/workflows/build-and-test.yml @@ -72,6 +72,20 @@ jobs: - name: Checkout repository uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd + - name: Setup Go + uses: actions/setup-go@4a3601121dd01d1626a1e23e37211e3254c1c06c + with: + go-version-file: cli/.go-version + cache: false + + # Windows-only code paths (named pipe transport, winio deadlines, typed error + # classification) are otherwise never executed by CI. Scoped to the transport + # package: the wider test suite still has Windows-incompatible tests (path + # separators, missing .exe suffixes, TCP-only fixtures) that predate this job. + - name: Test native Go CLI transport on Windows + shell: bash + run: cd cli && go test ./internal/unityipc/ + - name: Test install release filter in Git Bash shell: bash run: sh scripts/test-install-release-filter.sh diff --git a/Assets/Tests/Editor/BridgeTransportListenerTests.cs b/Assets/Tests/Editor/BridgeTransportListenerTests.cs new file mode 100644 index 0000000000..426dcfe26e --- /dev/null +++ b/Assets/Tests/Editor/BridgeTransportListenerTests.cs @@ -0,0 +1,108 @@ +using System; +using System.IO; +using System.Threading; +using NUnit.Framework; + +using io.github.hatayama.UnityCliLoop.Infrastructure; + +namespace io.github.hatayama.UnityCliLoop.Tests.Editor +{ + /// + /// Test fixture that verifies the Unix domain socket listener's stop behavior. + /// + public class BridgeTransportListenerTests + { + private string _tempProjectRoot; + private BridgeTransportEndpoint _endpoint; + + [SetUp] + public void SetUp() + { + _tempProjectRoot = Path.Combine(Path.GetTempPath(), "uloop-listener-test-" + Guid.NewGuid().ToString("N")); + Directory.CreateDirectory(_tempProjectRoot); + _endpoint = BridgeTransportEndpoint.CreateProjectIpc(_tempProjectRoot); + if (_endpoint.Kind != BridgeTransportKind.UnixDomainSocket) + { + Assert.Ignore("Unix domain socket listener tests only run on macOS/Linux."); + } + } + + [TearDown] + public void TearDown() + { + if (_endpoint != null && File.Exists(_endpoint.Path)) + { + File.Delete(_endpoint.Path); + } + + if (Directory.Exists(_tempProjectRoot)) + { + Directory.Delete(_tempProjectRoot, recursive: true); + } + } + + [Test] + public void Stop_WhenCalledTwice_DoesNotThrowAndRemovesSocketFile() + { + // Tests that Stop is idempotent: Dispose() also calls Stop, so double stops happen + // routinely during shutdown and must not throw on the already-closed socket. + UnixDomainSocketBridgeTransportListener listener = new(_endpoint); + listener.Start(); + Assert.That(File.Exists(_endpoint.Path), Is.True); + + listener.Stop(); + Assert.DoesNotThrow(() => listener.Stop()); + Assert.That(File.Exists(_endpoint.Path), Is.False); + } + + [Test] + public void AcceptClient_AfterStop_ThrowsObjectDisposedException() + { + // Tests that accepting on a stopped listener surfaces as disposal, which the server + // loop treats as an exit-and-recover signal, instead of NullReferenceException + // which would make the accept loop spin. + UnixDomainSocketBridgeTransportListener listener = new(_endpoint); + listener.Start(); + listener.Stop(); + + Assert.Throws( + () => listener.AcceptClient(CancellationToken.None)); + } + } + + /// + /// Test fixture that verifies the Windows named pipe listener's stopped-state behavior. + /// These paths run before any pipe is created, so they are platform-independent. + /// + public class WindowsNamedPipeBridgeTransportListenerTests + { + [Test] + public void Stop_WhenCalledTwice_DoesNotThrow() + { + // Tests that Stop is idempotent: Dispose() also calls Stop, so double stops happen + // routinely during shutdown and only one caller may dispose the active pipe. + BridgeTransportEndpoint endpoint = BridgeTransportEndpoint.CreateProjectIpc( + System.IO.Path.GetTempPath()); + WindowsNamedPipeBridgeTransportListener listener = new(endpoint); + listener.Start(); + + listener.Stop(); + Assert.DoesNotThrow(() => listener.Stop()); + } + + [Test] + public void AcceptClient_AfterStop_ThrowsObjectDisposedException() + { + // Tests that a stopped listener refuses to accept instead of creating a fresh + // pipe nobody can wake, so the server loop exits and recovers. + BridgeTransportEndpoint endpoint = BridgeTransportEndpoint.CreateProjectIpc( + System.IO.Path.GetTempPath()); + WindowsNamedPipeBridgeTransportListener listener = new(endpoint); + listener.Start(); + listener.Stop(); + + Assert.Throws( + () => listener.AcceptClient(CancellationToken.None)); + } + } +} diff --git a/Assets/Tests/Editor/BridgeTransportListenerTests.cs.meta b/Assets/Tests/Editor/BridgeTransportListenerTests.cs.meta new file mode 100644 index 0000000000..ed894ca1aa --- /dev/null +++ b/Assets/Tests/Editor/BridgeTransportListenerTests.cs.meta @@ -0,0 +1,11 @@ +fileFormatVersion: 2 +guid: 3ff6bd689fdbb4f2f9600db948281d9c +MonoImporter: + externalObjects: {} + serializedVersion: 2 + defaultReferences: [] + executionOrder: 0 + icon: {instanceID: 0} + userData: + assetBundleName: + assetBundleVariant: diff --git a/Assets/Tests/Editor/JsonRpcHeartbeatTests.cs b/Assets/Tests/Editor/JsonRpcHeartbeatTests.cs new file mode 100644 index 0000000000..303671bbdd --- /dev/null +++ b/Assets/Tests/Editor/JsonRpcHeartbeatTests.cs @@ -0,0 +1,106 @@ +using System; +using System.Threading; +using System.Threading.Tasks; +using Newtonsoft.Json.Linq; +using NUnit.Framework; + +using io.github.hatayama.UnityCliLoop.Infrastructure; + +namespace io.github.hatayama.UnityCliLoop.Tests.Editor +{ + /// + /// Test fixture that verifies heartbeat negotiation and frame emission for the bridge server. + /// + public class JsonRpcHeartbeatTests + { + [Test] + public void CreateDispatchAcceptedResponse_WhenHeartbeatNegotiated_AdvertisesInterval() + { + // Tests that the dispatch ack tells a heartbeat-capable CLI which interval to expect. + string response = JsonRpcProcessor.CreateDispatchAcceptedResponse(1, 10); + + JObject parsed = JObject.Parse(response); + Assert.That(parsed["uloop"]["phase"].ToString(), Is.EqualTo(JsonRpcResponsePhases.Accepted)); + Assert.That(parsed["uloop"]["heartbeatIntervalSeconds"].Value(), Is.EqualTo(10)); + } + + [Test] + public void CreateDispatchAcceptedResponse_WithoutHeartbeat_OmitsInterval() + { + // Tests that older CLIs that did not negotiate heartbeats get the legacy ack shape, + // because they would treat unexpected extra frames as the final response. + string response = JsonRpcProcessor.CreateDispatchAcceptedResponse(1, 0); + + JObject parsed = JObject.Parse(response); + Assert.That(parsed["uloop"]["phase"].ToString(), Is.EqualTo(JsonRpcResponsePhases.Accepted)); + Assert.That(parsed["uloop"]["heartbeatIntervalSeconds"], Is.Null); + } + + [Test] + public void CreateHeartbeatResponse_WhenSerialized_CarriesPhaseAndStallSeconds() + { + // Tests the heartbeat frame shape the CLI parses for freeze diagnosis. + string response = JsonRpcProcessor.CreateHeartbeatResponse(7, 12.5); + + JObject parsed = JObject.Parse(response); + Assert.That(parsed["uloop"]["phase"].ToString(), Is.EqualTo(JsonRpcResponsePhases.Heartbeat)); + Assert.That(parsed["uloop"]["mainThreadStallSeconds"].Value(), Is.EqualTo(12.5)); + Assert.That(parsed["id"].Value(), Is.EqualTo(7)); + } + + [Test] + public async Task SendHeartbeatsAsync_WhenRunning_WritesFramesUntilCancelled() + { + // Tests that the heartbeat loop emits frames on the interval and stops on cancellation + // without leaving background work behind. + int writtenFrameCount = 0; + using CancellationTokenSource cancellationSource = new(); + + Task heartbeatTask = UnityCliLoopBridgeServer.SendHeartbeatsAsync( + () => "{}", + _ => + { + Interlocked.Increment(ref writtenFrameCount); + return Task.CompletedTask; + }, + TimeSpan.FromMilliseconds(10), + cancellationSource.Token); + + await Task.Delay(100); + cancellationSource.Cancel(); + await heartbeatTask; + + Assert.That(Volatile.Read(ref writtenFrameCount), Is.GreaterThanOrEqualTo(1)); + } + + [Test] + public async Task SendHeartbeatsAsync_WhenWriteThrowsIOException_StopsWithoutFaulting() + { + // Tests that a broken connection ends the heartbeat loop silently; teardown is + // owned by the read loop, not the heartbeat writer. + using CancellationTokenSource cancellationSource = new(); + + Task heartbeatTask = UnityCliLoopBridgeServer.SendHeartbeatsAsync( + () => "{}", + _ => throw new System.IO.IOException("broken pipe"), + TimeSpan.FromMilliseconds(1), + cancellationSource.Token); + + await heartbeatTask; + + Assert.That(heartbeatTask.IsCompletedSuccessfully, Is.True); + } + + [Test] + public void EditorMainThreadLivenessTracker_AfterRegistration_ReportsSmallStall() + { + // Tests that the tracker reports near-zero stall right after a recorded tick. + EditorMainThreadLivenessTracker.RegisterForEditorStartup(); + + double stallSeconds = EditorMainThreadLivenessTracker.SecondsSinceLastMainThreadTick(); + + Assert.That(stallSeconds, Is.GreaterThanOrEqualTo(0)); + Assert.That(stallSeconds, Is.LessThan(60)); + } + } +} diff --git a/Assets/Tests/Editor/JsonRpcHeartbeatTests.cs.meta b/Assets/Tests/Editor/JsonRpcHeartbeatTests.cs.meta new file mode 100644 index 0000000000..9ff6ce01e9 --- /dev/null +++ b/Assets/Tests/Editor/JsonRpcHeartbeatTests.cs.meta @@ -0,0 +1,11 @@ +fileFormatVersion: 2 +guid: 07f19653643224ffaaa45e4a954dfc13 +MonoImporter: + externalObjects: {} + serializedVersion: 2 + defaultReferences: [] + executionOrder: 0 + icon: {instanceID: 0} + userData: + assetBundleName: + assetBundleVariant: diff --git a/Assets/Tests/Editor/JsonRpcProcessorCliVersionGateTests.cs b/Assets/Tests/Editor/JsonRpcProcessorCliVersionGateTests.cs index d77efe7df4..0ea907e602 100644 --- a/Assets/Tests/Editor/JsonRpcProcessorCliVersionGateTests.cs +++ b/Assets/Tests/Editor/JsonRpcProcessorCliVersionGateTests.cs @@ -105,14 +105,14 @@ public async Task ProcessRequest_WhenFirstToolWaitsForMainThread_ReturnsServerBu firstResponseTask = JsonRpcProcessor.ProcessRequestWithEarlyResponseAsync( BuildToolRequest(SingleFlightTestTool.Name, 1), CancellationToken.None, - (_, _) => Task.CompletedTask); + (_, _, _) => Task.CompletedTask); Assert.That(dispatcher.PendingContinuationCount, Is.EqualTo(1)); secondResponseTask = JsonRpcProcessor.ProcessRequestWithEarlyResponseAsync( BuildToolRequest(SingleFlightTestTool.Name, 2), CancellationToken.None, - (_, _) => Task.CompletedTask); + (_, _, _) => Task.CompletedTask); string secondResponse = await AwaitWithTimeout(secondResponseTask, TimeSpan.FromMilliseconds(200)); JObject error = ParseError(secondResponse); @@ -163,14 +163,14 @@ public async Task ProcessRequest_WhenExecuteDynamicCodeWaitsForMainThread_Allows firstDynamicCodeTask = JsonRpcProcessor.ProcessRequestWithEarlyResponseAsync( BuildToolRequest(UnityCliLoopConstants.TOOL_NAME_EXECUTE_DYNAMIC_CODE, 1), CancellationToken.None, - (_, _) => Task.CompletedTask); + (_, _, _) => Task.CompletedTask); Assert.That(dispatcher.PendingContinuationCount, Is.EqualTo(1)); secondDynamicCodeTask = JsonRpcProcessor.ProcessRequestWithEarlyResponseAsync( BuildToolRequest(UnityCliLoopConstants.TOOL_NAME_EXECUTE_DYNAMIC_CODE, 2), CancellationToken.None, - (_, _) => Task.CompletedTask); + (_, _, _) => Task.CompletedTask); Assert.That(dispatcher.PendingContinuationCount, Is.EqualTo(2)); Assert.That(secondDynamicCodeTask.IsCompleted, Is.False); @@ -178,7 +178,7 @@ public async Task ProcessRequest_WhenExecuteDynamicCodeWaitsForMainThread_Allows otherToolTask = JsonRpcProcessor.ProcessRequestWithEarlyResponseAsync( BuildToolRequest(SingleFlightTestTool.Name, 3), CancellationToken.None, - (_, _) => Task.CompletedTask); + (_, _, _) => Task.CompletedTask); string otherToolResponse = await AwaitWithTimeout(otherToolTask, TimeSpan.FromMilliseconds(200)); JObject error = ParseError(otherToolResponse); @@ -227,7 +227,7 @@ public async Task ProcessRequest_WhenCompileWaitsForDomainReload_KeepsAcceptedRe "{\"WaitForDomainReload\":true}", 1), CancellationToken.None, - (_, shouldCancelOnClientDisconnect) => + (_, shouldCancelOnClientDisconnect, _) => { cancelOnClientDisconnect = shouldCancelOnClientDisconnect; return Task.CompletedTask; @@ -263,7 +263,7 @@ public async Task ProcessRequest_WhenCompileOmitsReloadWait_KeepsAcceptedRequest string response = await JsonRpcProcessor.ProcessRequestWithEarlyResponseAsync( BuildToolRequest(UnityCliLoopConstants.TOOL_NAME_COMPILE, 1), CancellationToken.None, - (_, shouldCancelOnClientDisconnect) => + (_, shouldCancelOnClientDisconnect, _) => { cancelOnClientDisconnect = shouldCancelOnClientDisconnect; return Task.CompletedTask; @@ -302,7 +302,7 @@ public async Task ProcessRequest_WhenCompileDoesNotWaitForDomainReload_CancelsOn "{\"WaitForDomainReload\":false}", 1), CancellationToken.None, - (_, shouldCancelOnClientDisconnect) => + (_, shouldCancelOnClientDisconnect, _) => { cancelOnClientDisconnect = shouldCancelOnClientDisconnect; return Task.CompletedTask; @@ -341,7 +341,7 @@ public async Task ProcessRequest_WhenCompileUsesCamelCaseNoReloadWait_CancelsOnC "{\"waitForDomainReload\":false}", 1), CancellationToken.None, - (_, shouldCancelOnClientDisconnect) => + (_, shouldCancelOnClientDisconnect, _) => { cancelOnClientDisconnect = shouldCancelOnClientDisconnect; return Task.CompletedTask; @@ -384,7 +384,7 @@ public async Task ProcessRequest_AfterGetHierarchyReturns_AllowsImmediateGetLogs "{\"MaxDepth\":0,\"IncludeComponents\":false}", 1), CancellationToken.None, - (_, _) => Task.CompletedTask); + (_, _, _) => Task.CompletedTask); Assert.That(dispatcher.PendingContinuationCount, Is.EqualTo(1)); dispatcher.RunContinuations(); @@ -400,7 +400,7 @@ public async Task ProcessRequest_AfterGetHierarchyReturns_AllowsImmediateGetLogs "{\"MaxCount\":0}", 2), CancellationToken.None, - (_, _) => Task.CompletedTask); + (_, _, _) => Task.CompletedTask); Assert.That(dispatcher.PendingContinuationCount, Is.EqualTo(1)); dispatcher.RunContinuations(); @@ -434,7 +434,7 @@ public async Task ProcessRequest_WhenInternalBridgeCommandRuns_SwitchesToMainThr responseTask = JsonRpcProcessor.ProcessRequestWithEarlyResponseAsync( BuildToolRequest(UnityCliLoopConstants.COMMAND_NAME_GET_VERSION, 1), CancellationToken.None, - (_, _) => Task.CompletedTask); + (_, _, _) => Task.CompletedTask); Assert.That(dispatcher.PendingContinuationCount, Is.EqualTo(1)); Assert.That(responseTask.IsCompleted, Is.False); @@ -478,7 +478,7 @@ public async Task ProcessRequest_WhenMainThreadSwitchIsCanceled_ReleasesExecutio canceledResponseTask = JsonRpcProcessor.ProcessRequestWithEarlyResponseAsync( BuildToolRequest(SingleFlightTestTool.Name, 1), cancellationSource.Token, - (_, _) => Task.CompletedTask); + (_, _, _) => Task.CompletedTask); Assert.That(dispatcher.PendingContinuationCount, Is.EqualTo(1)); @@ -488,7 +488,7 @@ public async Task ProcessRequest_WhenMainThreadSwitchIsCanceled_ReleasesExecutio secondResponseTask = JsonRpcProcessor.ProcessRequestWithEarlyResponseAsync( BuildToolRequest(SingleFlightTestTool.Name, 2), CancellationToken.None, - (_, _) => Task.CompletedTask); + (_, _, _) => Task.CompletedTask); Assert.That(secondResponseTask.IsCompleted, Is.False); diff --git a/Assets/Tests/Editor/UnityCliLoopBridgeServerShutdownTests.cs b/Assets/Tests/Editor/UnityCliLoopBridgeServerShutdownTests.cs new file mode 100644 index 0000000000..a94a9b479a --- /dev/null +++ b/Assets/Tests/Editor/UnityCliLoopBridgeServerShutdownTests.cs @@ -0,0 +1,51 @@ +using System; +using System.Threading; +using System.Threading.Tasks; +using NUnit.Framework; + +using io.github.hatayama.UnityCliLoop.Infrastructure; + +namespace io.github.hatayama.UnityCliLoop.Tests.Editor +{ + /// + /// Test fixture that verifies shutdown disposal of the bridge server's cancellation source. + /// + public class UnityCliLoopBridgeServerShutdownTests + { + [Test] + public async Task DisposeCancellationSourceAfterServerTaskAsync_WhenAllTasksComplete_DisposesSource() + { + // Tests that the cancellation source is disposed once every shutdown-wait task has finished. + CancellationTokenSource cancellationTokenSource = new(); + + await UnityCliLoopBridgeServer.DisposeCancellationSourceAfterServerTaskAsync( + Task.CompletedTask, + new[] { Task.CompletedTask }, + cancellationTokenSource, + TimeSpan.FromSeconds(1)); + + Assert.Throws(() => cancellationTokenSource.Cancel()); + } + + [Test] + public async Task DisposeCancellationSourceAfterServerTaskAsync_WhenTaskOutlivesTimeout_SkipsDisposal() + { + // Tests that a straggling client task keeps the cancellation source alive so the task + // can still observe its token without hitting ObjectDisposedException. + CancellationTokenSource cancellationTokenSource = new(); + TaskCompletionSource stragglingTask = new(); + + await UnityCliLoopBridgeServer.DisposeCancellationSourceAfterServerTaskAsync( + Task.CompletedTask, + new[] { stragglingTask.Task }, + cancellationTokenSource, + TimeSpan.FromMilliseconds(50)); + + Assert.DoesNotThrow(() => cancellationTokenSource.Cancel()); + + // Complete the straggler so the test leaves no pending work behind. + stragglingTask.SetResult(true); + cancellationTokenSource.Dispose(); + } + } +} diff --git a/Assets/Tests/Editor/UnityCliLoopBridgeServerShutdownTests.cs.meta b/Assets/Tests/Editor/UnityCliLoopBridgeServerShutdownTests.cs.meta new file mode 100644 index 0000000000..ebeb169a05 --- /dev/null +++ b/Assets/Tests/Editor/UnityCliLoopBridgeServerShutdownTests.cs.meta @@ -0,0 +1,11 @@ +fileFormatVersion: 2 +guid: 02618fbff103949eb85531d97c686af6 +MonoImporter: + externalObjects: {} + serializedVersion: 2 + defaultReferences: [] + executionOrder: 0 + icon: {instanceID: 0} + userData: + assetBundleName: + assetBundleVariant: diff --git a/Assets/Tests/Editor/UnityCliLoopServerControllerStartupLockTests.cs b/Assets/Tests/Editor/UnityCliLoopServerControllerStartupLockTests.cs index ac5f3112cb..7dddb63154 100644 --- a/Assets/Tests/Editor/UnityCliLoopServerControllerStartupLockTests.cs +++ b/Assets/Tests/Editor/UnityCliLoopServerControllerStartupLockTests.cs @@ -220,6 +220,93 @@ public async Task ProbeReadinessWithTimeoutAsync_WhenProbeDoesNotComplete_Throws Assert.That(readinessProbe.CallCount, Is.EqualTo(1)); } + [Test] + public async Task ExecuteTrackedRecoveryAsync_WhenRecoveryFailsOnce_RetriesAfterBackoffAndSucceeds() + { + // Tests that a transient recovery failure (e.g. readiness timeout during a heavy import) + // is retried with backoff instead of leaving the server down until the next domain reload. + System.Collections.Generic.List recordedWaits = new(); + int recoveryAttempts = 0; + UnityCliLoopServerControllerService service = CreateControllerService( + new TestReadinessProbe(), + (delayMilliseconds, ct) => + { + recordedWaits.Add(delayMilliseconds); + return Task.CompletedTask; + }); + + await service.ExecuteTrackedRecoveryAsync(() => + { + recoveryAttempts++; + if (recoveryAttempts == 1) + { + throw new System.TimeoutException("readiness probe timed out"); + } + + return Task.CompletedTask; + }); + + Assert.That(recoveryAttempts, Is.EqualTo(2)); + Assert.That(recordedWaits, Is.EqualTo(new[] + { + UnityCliLoopServerConfig.RECOVERY_RETRY_DELAYS_MS[0] + })); + } + + [Test] + public void ExecuteTrackedRecoveryAsync_WhenRecoveryKeepsFailing_GivesUpAfterAllRetriesAndClearsSession() + { + // Tests that persistent recovery failure exhausts the full backoff schedule before + // surfacing the error and clearing the server session. + _sessionStateService.MarkServerStarted(); + System.Collections.Generic.List recordedWaits = new(); + int recoveryAttempts = 0; + UnityCliLoopServerControllerService service = CreateControllerService( + new TestReadinessProbe(), + (delayMilliseconds, ct) => + { + recordedWaits.Add(delayMilliseconds); + return Task.CompletedTask; + }); + + System.InvalidOperationException exception = + Assert.ThrowsAsync(async () => + await service.ExecuteTrackedRecoveryAsync(() => + { + recoveryAttempts++; + throw new System.TimeoutException("readiness probe timed out"); + })); + + Assert.That(exception.Message, Does.Contain("recovery failed")); + Assert.That(recoveryAttempts, Is.EqualTo(UnityCliLoopServerConfig.RECOVERY_RETRY_DELAYS_MS.Length + 1)); + Assert.That( + recordedWaits, + Is.EqualTo(UnityCliLoopServerConfig.RECOVERY_RETRY_DELAYS_MS)); + Assert.That(_sessionStateService.GetIsServerRunning(), Is.False); + } + + [Test] + public async Task ExecuteTrackedRecoveryAsync_WhenServerManuallyStoppedDuringBackoff_AbandonsRetry() + { + // Tests that an explicit Stop Server during the retry backoff wins over automatic recovery. + int recoveryAttempts = 0; + UnityCliLoopServerControllerService service = CreateControllerService( + new TestReadinessProbe(), + (delayMilliseconds, ct) => + { + _sessionStateService.MarkServerManuallyStopped(); + return Task.CompletedTask; + }); + + await service.ExecuteTrackedRecoveryAsync(() => + { + recoveryAttempts++; + throw new System.TimeoutException("readiness probe timed out"); + }); + + Assert.That(recoveryAttempts, Is.EqualTo(1)); + } + [Test] public async Task RestoreServerStateIfNeeded_WhenServerWasManuallyStopped_ShouldSkipStartupRecovery() { @@ -311,12 +398,9 @@ public async Task StartServerWithUseCaseAsync_WhenRestartCleanupStopFails_Should Assert.That(_sessionStateService.GetIsServerManuallyStopped(), Is.False); } - private UnityCliLoopServerControllerService CreateControllerService() - { - return CreateControllerService(new TestReadinessProbe()); - } - - private UnityCliLoopServerControllerService CreateControllerService(TestReadinessProbe readinessProbe) + private UnityCliLoopServerControllerService CreateControllerService( + TestReadinessProbe readinessProbe = null, + System.Func waitBeforeRecoveryRetryAsync = null) { TestServerInstanceFactory serverInstanceFactory = new(); UnityCliLoopServerLifecycleRegistryService lifecycleRegistry = @@ -326,8 +410,9 @@ private UnityCliLoopServerControllerService CreateControllerService(TestReadines lifecycleRegistry, new DomainReloadDetectionFileService(_sessionStateService), _sessionStateService, - readinessProbe, - new TestDomainReloadLifecycle()); + readinessProbe ?? new TestReadinessProbe(), + new TestDomainReloadLifecycle(), + waitBeforeRecoveryRetryAsync: waitBeforeRecoveryRetryAsync); } /// diff --git a/Packages/src/Editor/Domain/CliConstants.cs b/Packages/src/Editor/Domain/CliConstants.cs index 4d96111a46..79443441dd 100644 --- a/Packages/src/Editor/Domain/CliConstants.cs +++ b/Packages/src/Editor/Domain/CliConstants.cs @@ -6,7 +6,7 @@ namespace io.github.hatayama.UnityCliLoop.Domain public static class CliConstants { public const string EXECUTABLE_NAME = "uloop"; - public const string MINIMUM_REQUIRED_CLI_VERSION = "3.0.0-beta.29"; + public const string MINIMUM_REQUIRED_CLI_VERSION = "3.0.0-beta.30"; public const string MINIMUM_REQUIRED_CLI_RELEASE_TAG = CLI_RELEASE_TAG_PREFIX + MINIMUM_REQUIRED_CLI_VERSION; public const string VERSION_FLAG = "--version"; public const string SHORT_VERSION_FLAG = "-v"; diff --git a/Packages/src/Editor/Infrastructure/Api/JsonRpcProcessor.cs b/Packages/src/Editor/Infrastructure/Api/JsonRpcProcessor.cs index 4a395b0d1d..f35d2458e3 100644 --- a/Packages/src/Editor/Infrastructure/Api/JsonRpcProcessor.cs +++ b/Packages/src/Editor/Infrastructure/Api/JsonRpcProcessor.cs @@ -35,7 +35,8 @@ public static class JsonRpcProcessor internal delegate Task JsonRpcEarlyResponseWriter( string responseJson, - bool cancelOnClientDisconnect); + bool cancelOnClientDisconnect, + Func createHeartbeatJson); /// /// Process JSON-RPC request and generate response @@ -85,6 +86,7 @@ private static JsonRpcRequest ParseRequest(string jsonRequest) Params = request["params"], ClientCliVersion = ReadClientCliVersion(request), AcceptsDispatchAck = ReadAcceptsDispatchAck(request), + AcceptsHeartbeat = ReadAcceptsHeartbeat(request), Id = request["id"]?.ToObject() }; } @@ -112,6 +114,17 @@ private static bool ReadAcceptsDispatchAck(JObject request) return metadata["acceptsDispatchAck"]?.Value() ?? false; } + private static bool ReadAcceptsHeartbeat(JObject request) + { + JObject metadata = request["uloop"] as JObject; + if (metadata == null) + { + return false; + } + + return metadata["acceptsHeartbeat"]?.Value() ?? false; + } + /// /// Process notification (fire-and-forget) /// @@ -166,9 +179,18 @@ private static async Task ProcessRpcRequest( if (request.AcceptsDispatchAck && earlyResponseWriter != null) { + int heartbeatIntervalSeconds = request.AcceptsHeartbeat + ? UnityCliLoopServerConfig.HEARTBEAT_INTERVAL_SECONDS + : 0; + Func createHeartbeatJson = request.AcceptsHeartbeat + ? () => CreateHeartbeatResponse( + request.Id, + EditorMainThreadLivenessTracker.SecondsSinceLastMainThreadTick()) + : null; await earlyResponseWriter( - CreateDispatchAcceptedResponse(request.Id), - ShouldCancelAcceptedRequestOnClientDisconnect(request)); + CreateDispatchAcceptedResponse(request.Id, heartbeatIntervalSeconds), + ShouldCancelAcceptedRequestOnClientDisconnect(request), + createHeartbeatJson); } Stopwatch requestStopwatch = Stopwatch.StartNew(); @@ -284,8 +306,22 @@ private static string CreateCliUpdateRequiredResponse(object id, string currentC return JsonConvert.SerializeObject(errorResponse, Formatting.None, settings); } - private static string CreateDispatchAcceptedResponse(object id) + internal static string CreateDispatchAcceptedResponse(object id, int heartbeatIntervalSeconds) { + // heartbeatIntervalSeconds is only advertised when the client negotiated + // heartbeats; older CLIs would treat unexpected extra frames as the final + // response, so the field doubles as the negotiation answer. + object uloopMetadata = heartbeatIntervalSeconds > 0 + ? new + { + phase = JsonRpcResponsePhases.Accepted, + heartbeatIntervalSeconds + } + : (object)new + { + phase = JsonRpcResponsePhases.Accepted + }; + object response = new { jsonrpc = UnityCliLoopServerConfig.JSONRPC_VERSION, @@ -294,9 +330,32 @@ private static string CreateDispatchAcceptedResponse(object id) { accepted = true }, + uloop = uloopMetadata + }; + + JsonSerializerSettings settings = new() + { + ReferenceLoopHandling = ReferenceLoopHandling.Ignore, + MaxDepth = UnityCliLoopServerConfig.DEFAULT_JSON_MAX_DEPTH + }; + + return JsonConvert.SerializeObject(response, Formatting.None, settings); + } + + internal static string CreateHeartbeatResponse(object id, double mainThreadStallSeconds) + { + object response = new + { + jsonrpc = UnityCliLoopServerConfig.JSONRPC_VERSION, + id, + result = new + { + alive = true + }, uloop = new { - phase = JsonRpcResponsePhases.Accepted + phase = JsonRpcResponsePhases.Heartbeat, + mainThreadStallSeconds } }; @@ -445,6 +504,7 @@ public static class JsonRpcErrorTypes public static class JsonRpcResponsePhases { public const string Accepted = "accepted"; + public const string Heartbeat = "heartbeat"; } /// diff --git a/Packages/src/Editor/Infrastructure/Api/JsonRpcRequest.cs b/Packages/src/Editor/Infrastructure/Api/JsonRpcRequest.cs index ab5c9679f1..3c30b1f6bf 100644 --- a/Packages/src/Editor/Infrastructure/Api/JsonRpcRequest.cs +++ b/Packages/src/Editor/Infrastructure/Api/JsonRpcRequest.cs @@ -20,6 +20,8 @@ internal class JsonRpcRequest public bool AcceptsDispatchAck { get; set; } + public bool AcceptsHeartbeat { get; set; } + /// /// JSON-RPC 2.0 spec requires id type to match the request. /// Must be string, number, or null - same as received. diff --git a/Packages/src/Editor/Infrastructure/BridgeTransportListener.cs b/Packages/src/Editor/Infrastructure/BridgeTransportListener.cs index 197a0dcf45..acaa4b369d 100644 --- a/Packages/src/Editor/Infrastructure/BridgeTransportListener.cs +++ b/Packages/src/Editor/Infrastructure/BridgeTransportListener.cs @@ -115,10 +115,18 @@ public BridgeClientConnection AcceptClient(CancellationToken ct) { ct.ThrowIfCancellationRequested(); using CancellationTokenRegistration cancellationRegistration = ct.Register(Stop); + Socket listener = _listener; + if (listener == null) + { + // Why: a stopped listener must surface as disposal so ServerLoopAsync exits the + // accept loop and triggers recovery, instead of spinning on NullReferenceException. + throw new ObjectDisposedException(nameof(UnixDomainSocketBridgeTransportListener)); + } + Socket client; try { - client = _listener.Accept(); + client = listener.Accept(); } catch (ObjectDisposedException) when (ct.IsCancellationRequested) { @@ -138,8 +146,16 @@ public BridgeClientConnection AcceptClient(CancellationToken ct) public void Stop() { - _listener?.Close(); - _listener = null; + // Why: Stop races between the accept-loop cancellation registration, StopServer on + // the main thread, and unexpected-exit cleanup on the thread pool. Interlocked hands + // the socket to exactly one caller so Close and the socket-file delete run once. + Socket listener = Interlocked.Exchange(ref _listener, null); + if (listener == null) + { + return; + } + + listener.Close(); if (File.Exists(Endpoint.Path)) { File.Delete(Endpoint.Path); @@ -164,6 +180,9 @@ internal sealed class WindowsNamedPipeBridgeTransportListener : IBridgeTransport { private NamedPipeServerStream _activePipe; private long _nextClientId; + // Why: without a stopped state, an accept racing with Stop could create a fresh + // pipe that nobody can wake, leaving the accept loop blocked forever. + private volatile bool _stopped; public BridgeTransportEndpoint Endpoint { get; } @@ -179,6 +198,7 @@ public void Start() public BridgeClientConnection AcceptClient(CancellationToken ct) { ct.ThrowIfCancellationRequested(); + ThrowIfStopped(); NamedPipeServerStream pipe = new( Endpoint.PipeName, PipeDirection.InOut, @@ -186,7 +206,15 @@ public BridgeClientConnection AcceptClient(CancellationToken ct) PipeTransmissionMode.Byte, PipeOptions.Asynchronous); _activePipe = pipe; - using CancellationTokenRegistration cancellationRegistration = ct.Register(DisposeActivePipe); + // Why: Stop may have run between the stopped check and the assignment above; + // re-checking after publishing the pipe guarantees one side disposes it. + if (_stopped) + { + Interlocked.Exchange(ref _activePipe, null)?.Dispose(); + ThrowIfStopped(); + } + + using CancellationTokenRegistration cancellationRegistration = ct.Register(Stop); bool connected = false; try { @@ -209,10 +237,7 @@ public BridgeClientConnection AcceptClient(CancellationToken ct) } finally { - if (ReferenceEquals(_activePipe, pipe)) - { - _activePipe = null; - } + Interlocked.CompareExchange(ref _activePipe, null, pipe); if (!connected) { @@ -223,8 +248,12 @@ public BridgeClientConnection AcceptClient(CancellationToken ct) public void Stop() { - _activePipe?.Dispose(); - _activePipe = null; + // Why: Stop races between the accept-loop cancellation registration, StopServer + // on the main thread, and unexpected-exit cleanup on the thread pool; Interlocked + // hands the pipe to exactly one caller so Dispose runs once. + _stopped = true; + NamedPipeServerStream pipe = Interlocked.Exchange(ref _activePipe, null); + pipe?.Dispose(); } public void Dispose() @@ -232,9 +261,14 @@ public void Dispose() Stop(); } - private void DisposeActivePipe() + private void ThrowIfStopped() { - _activePipe?.Dispose(); + if (_stopped) + { + // Surface as disposal so ServerLoopAsync exits the accept loop and triggers + // recovery instead of re-listening on a stopped listener. + throw new ObjectDisposedException(nameof(WindowsNamedPipeBridgeTransportListener)); + } } } } diff --git a/Packages/src/Editor/Infrastructure/InfrastructureEditorStartup.cs b/Packages/src/Editor/Infrastructure/InfrastructureEditorStartup.cs index 780d4e6a46..57121166f5 100644 --- a/Packages/src/Editor/Infrastructure/InfrastructureEditorStartup.cs +++ b/Packages/src/Editor/Infrastructure/InfrastructureEditorStartup.cs @@ -13,6 +13,7 @@ internal static void Initialize(UnityCliLoopEditorSettingsService editorSettings UnityCliLoopPackageRemovalSettingsResetter packageRemovalSettingsResetter = new(editorSettingsService); packageRemovalSettingsResetter.RegisterForEditorStartup(); UnityCliLoopEditorSettingsRecoveryScheduler.ScheduleForEditorStartup(editorSettingsService); + EditorMainThreadLivenessTracker.RegisterForEditorStartup(); } } } diff --git a/Packages/src/Editor/Infrastructure/Server/UnityCliLoopServerController.cs b/Packages/src/Editor/Infrastructure/Server/UnityCliLoopServerController.cs index fa378d5e19..c5cea58f11 100644 --- a/Packages/src/Editor/Infrastructure/Server/UnityCliLoopServerController.cs +++ b/Packages/src/Editor/Infrastructure/Server/UnityCliLoopServerController.cs @@ -36,6 +36,7 @@ private enum ServerStopIntent private readonly IUnityCliLoopServerDomainReloadLifecycle _domainReloadLifecycle; private readonly Func _isReadinessProbeBlocked; private readonly Func _waitBeforeReadinessRetryAsync; + private readonly Func _waitBeforeRecoveryRetryAsync; private readonly int _readinessIdleTimeoutMilliseconds; private IUnityCliLoopServerInstance _bridgeServer; private readonly SemaphoreSlim _startupSemaphore = new SemaphoreSlim(1, 1); @@ -51,6 +52,7 @@ internal UnityCliLoopServerControllerService( IUnityCliLoopServerDomainReloadLifecycle domainReloadLifecycle, Func isReadinessProbeBlocked = null, Func waitBeforeReadinessRetryAsync = null, + Func waitBeforeRecoveryRetryAsync = null, int readinessIdleTimeoutMilliseconds = UnityCliLoopServerConfig.READINESS_PROBE_TIMEOUT_MS) { System.Diagnostics.Debug.Assert(serverInstanceFactory != null, "serverInstanceFactory must not be null"); @@ -69,6 +71,7 @@ internal UnityCliLoopServerControllerService( _domainReloadLifecycle = domainReloadLifecycle ?? throw new ArgumentNullException(nameof(domainReloadLifecycle)); _isReadinessProbeBlocked = isReadinessProbeBlocked ?? IsEditorBusyForReadinessProbe; _waitBeforeReadinessRetryAsync = waitBeforeReadinessRetryAsync ?? TimerDelay.Wait; + _waitBeforeRecoveryRetryAsync = waitBeforeRecoveryRetryAsync ?? TimerDelay.Wait; _readinessIdleTimeoutMilliseconds = readinessIdleTimeoutMilliseconds; _sessionRecoveryService = new SessionRecoveryService( this, @@ -207,9 +210,12 @@ private void CompleteScheduledStartupRecovery( scheduledRecoveryCompletionSource.SetResult(true); } - public async void StartServer() + public void StartServer() { - await StartServerWithUseCaseAsync(); + // Why: an async void body lets exceptions (e.g. a readiness probe failure thrown by + // MarkServerReadyAsync) escape to the Unity synchronization context as unhandled. + // Forget() observes the task and logs the exception instead. + StartServerWithUseCaseAsync().Forget(); } /// @@ -262,9 +268,11 @@ internal async Task StartServerWithUseCaseAsync() /// /// Stops the server. /// - public async void StopServer() + public void StopServer() { - await StopServerWithUseCaseAsync(); + // Why: same as StartServer — Forget() keeps shutdown failures observed and logged + // instead of crashing through an async void boundary. + StopServerWithUseCaseAsync().Forget(); } /// @@ -466,22 +474,52 @@ private Task ScheduleTrackedRecovery(Func recoveryAction) return recoveryTask; } - private async Task ExecuteTrackedRecoveryAsync(Func recoveryAction) + /// + /// Runs a recovery action, retrying with backoff so one transient failure + /// (e.g. readiness timeout during a heavy import) does not leave the server + /// down until the next domain reload. + /// + internal async Task ExecuteTrackedRecoveryAsync(Func recoveryAction) { Debug.Assert(recoveryAction != null, "recoveryAction must not be null"); - try - { - await recoveryAction(); - } - catch (Exception ex) + int failedAttemptCount = 0; + while (true) { - string message = $"Unity CLI Loop server recovery failed before the bridge became ready. {ex.GetBaseException().Message}"; - VibeLogger.LogError( - "server_recovery_failed", - message); - _sessionStateService.ClearServerSession(); - throw new InvalidOperationException(message, ex); + try + { + await recoveryAction(); + return; + } + catch (Exception ex) + { + if (failedAttemptCount >= UnityCliLoopServerConfig.RECOVERY_RETRY_DELAYS_MS.Length) + { + string message = $"Unity CLI Loop server recovery failed before the bridge became ready. {ex.GetBaseException().Message}"; + VibeLogger.LogError( + "server_recovery_failed", + message); + _sessionStateService.ClearServerSession(); + throw new InvalidOperationException(message, ex); + } + + int delayMilliseconds = UnityCliLoopServerConfig.RECOVERY_RETRY_DELAYS_MS[failedAttemptCount]; + failedAttemptCount++; + VibeLogger.LogWarning( + "server_recovery_retry_scheduled", + $"Recovery attempt {failedAttemptCount} failed; retrying in {delayMilliseconds}ms. {ex.GetBaseException().Message}"); + await _waitBeforeRecoveryRetryAsync(delayMilliseconds, CancellationToken.None); + + // Why: an explicit Stop Server issued during the backoff must win over + // automatic recovery, otherwise the retry would silently restart the server. + if (_sessionStateService.GetIsServerManuallyStopped()) + { + VibeLogger.LogInfo( + "server_recovery_retry_abandoned", + "Recovery retry abandoned because the server was manually stopped."); + return; + } + } } } diff --git a/Packages/src/Editor/Infrastructure/Threading/EditorMainThreadLivenessTracker.cs b/Packages/src/Editor/Infrastructure/Threading/EditorMainThreadLivenessTracker.cs new file mode 100644 index 0000000000..bffac6ed38 --- /dev/null +++ b/Packages/src/Editor/Infrastructure/Threading/EditorMainThreadLivenessTracker.cs @@ -0,0 +1,39 @@ +using System; +using System.Threading; +using UnityEditor; + +namespace io.github.hatayama.UnityCliLoop.Infrastructure +{ + /// + /// Records when the editor main thread last pumped EditorApplication.update so + /// background threads (the IPC heartbeat sender) can report main-thread stalls. + /// + internal static class EditorMainThreadLivenessTracker + { + // Initialized to "now" so stall reports stay sane before the first update tick. + private static long _lastTickUtcTicks = DateTime.UtcNow.Ticks; + + internal static void RegisterForEditorStartup() + { + EditorApplication.update -= RecordTick; + EditorApplication.update += RecordTick; + RecordTick(); + } + + private static void RecordTick() + { + Volatile.Write(ref _lastTickUtcTicks, DateTime.UtcNow.Ticks); + } + + /// + /// Returns how long the main thread has gone without an update tick. + /// Safe to call from any thread. + /// + internal static double SecondsSinceLastMainThreadTick() + { + long lastTickTicks = Volatile.Read(ref _lastTickUtcTicks); + double seconds = (DateTime.UtcNow - new DateTime(lastTickTicks, DateTimeKind.Utc)).TotalSeconds; + return seconds < 0 ? 0 : seconds; + } + } +} diff --git a/Packages/src/Editor/Infrastructure/Threading/EditorMainThreadLivenessTracker.cs.meta b/Packages/src/Editor/Infrastructure/Threading/EditorMainThreadLivenessTracker.cs.meta new file mode 100644 index 0000000000..e3fa7a7aea --- /dev/null +++ b/Packages/src/Editor/Infrastructure/Threading/EditorMainThreadLivenessTracker.cs.meta @@ -0,0 +1,11 @@ +fileFormatVersion: 2 +guid: 0f214dcb789fe4ae69fe8bf8ea7ecfd6 +MonoImporter: + externalObjects: {} + serializedVersion: 2 + defaultReferences: [] + executionOrder: 0 + icon: {instanceID: 0} + userData: + assetBundleName: + assetBundleVariant: diff --git a/Packages/src/Editor/Infrastructure/UnityCliLoopBridgeServer.cs b/Packages/src/Editor/Infrastructure/UnityCliLoopBridgeServer.cs index ef36537570..dce0e7a60d 100644 --- a/Packages/src/Editor/Infrastructure/UnityCliLoopBridgeServer.cs +++ b/Packages/src/Editor/Infrastructure/UnityCliLoopBridgeServer.cs @@ -210,7 +210,11 @@ public void StopServer() Task serverTask = _serverTask; Task[] clientTasks = GetActiveClientTasks(); _serverTask = null; - DisposeCancellationSourceAfterServerTaskAsync(serverTask, clientTasks, cancellationTokenSource).Forget(); + DisposeCancellationSourceAfterServerTaskAsync( + serverTask, + clientTasks, + cancellationTokenSource, + TimeSpan.FromSeconds(UnityCliLoopServerConfig.SHUTDOWN_TIMEOUT_SECONDS)).Forget(); } private CancellationTokenSource TakeCancellationTokenSource() @@ -218,17 +222,29 @@ private CancellationTokenSource TakeCancellationTokenSource() return Interlocked.Exchange(ref _cancellationTokenSource, null); } - private static async Task DisposeCancellationSourceAfterServerTaskAsync( + internal static async Task DisposeCancellationSourceAfterServerTaskAsync( Task serverTask, Task[] clientTasks, - CancellationTokenSource cancellationTokenSource) + CancellationTokenSource cancellationTokenSource, + TimeSpan shutdownTimeout) { Task[] tasks = BuildShutdownWaitTasks(serverTask, clientTasks); if (tasks.Length > 0) { - await Task.WhenAny( - Task.WhenAll(tasks), - Task.Delay(TimeSpan.FromSeconds(UnityCliLoopServerConfig.SHUTDOWN_TIMEOUT_SECONDS))); + Task allTasksCompleted = Task.WhenAll(tasks); + Task firstCompleted = await Task.WhenAny( + allTasksCompleted, + Task.Delay(shutdownTimeout)); + if (firstCompleted != allTasksCompleted) + { + // Why: a straggling task may still observe the token after this timeout, and + // disposing the source under it surfaces ObjectDisposedException inside that + // task. Leaking one CancellationTokenSource is harmless; disposing early is not. + VibeLogger.LogWarning( + "server_shutdown_cts_leaked", + "Shutdown timed out waiting for server/client tasks; skipping CancellationTokenSource disposal."); + return; + } } cancellationTokenSource?.Dispose(); @@ -459,6 +475,10 @@ private async Task HandleClientAsync(BridgeClientConnection client, Cancellation // Initialize new framing components bufferManager = new DynamicBufferManager(); messageReassembler = new MessageReassembler(bufferManager); + + // Not disposed deliberately: a heartbeat writer could still be releasing it + // during teardown, and an undisposed SemaphoreSlim holds no unmanaged state. + SemaphoreSlim streamWriteLock = new(1, 1); // Start with initial buffer size byte[] buffer = bufferManager.GetBuffer(BufferConfig.INITIAL_BUFFER_SIZE); @@ -482,12 +502,16 @@ private async Task HandleClientAsync(BridgeClientConnection client, Cancellation { if (string.IsNullOrWhiteSpace(requestJson)) continue; - await ProcessRequestFrameAsync(client, stream, requestJson, cancellationToken); + await ProcessRequestFrameAsync(client, stream, streamWriteLock, requestJson, cancellationToken); } - // Validate reassembler state and clear if needed + // Why: false means the reassembler was disposed and can no longer make + // progress; closing the connection lets the CLI observe the disconnect + // immediately instead of waiting silently for its own timeout. + // (Corrupted framing state throws from ValidateState and is handled below.) if (!messageReassembler.ValidateState()) { + break; } } } @@ -575,6 +599,7 @@ private async Task WriteJsonResponseAsync( private async Task ProcessRequestFrameAsync( BridgeClientConnection client, Stream stream, + SemaphoreSlim streamWriteLock, string requestJson, CancellationToken serverCancellationToken) { @@ -582,14 +607,34 @@ private async Task ProcessRequestFrameAsync( CancellationTokenSource.CreateLinkedTokenSource(serverCancellationToken)) { Task clientDisconnectMonitorTask = null; + Task heartbeatTask = null; + CancellationTokenSource heartbeatCancellationSource = null; try { string responseJson = await JsonRpcProcessor.ProcessRequestWithEarlyResponseAsync( requestJson, requestCancellationTokenSource.Token, - async (responseJsonValue, cancelOnClientDisconnect) => + async (responseJsonValue, cancelOnClientDisconnect, createHeartbeatJson) => { - await WriteJsonResponseAsync(stream, responseJsonValue, serverCancellationToken); + await WriteJsonResponseLockedAsync( + stream, streamWriteLock, responseJsonValue, serverCancellationToken); + + if (createHeartbeatJson != null) + { + heartbeatCancellationSource = CancellationTokenSource.CreateLinkedTokenSource( + requestCancellationTokenSource.Token); + // Why: the heartbeat token must govern the write as well; with the + // server token an in-flight write could ignore StopHeartbeatsAsync + // and stall the final response behind a slow client. + CancellationToken heartbeatToken = heartbeatCancellationSource.Token; + heartbeatTask = SendHeartbeatsAsync( + createHeartbeatJson, + heartbeatJson => WriteJsonResponseLockedAsync( + stream, streamWriteLock, heartbeatJson, heartbeatToken), + TimeSpan.FromSeconds(UnityCliLoopServerConfig.HEARTBEAT_INTERVAL_SECONDS), + heartbeatToken); + } + if (!cancelOnClientDisconnect) { return; @@ -599,10 +644,17 @@ private async Task ProcessRequestFrameAsync( MonitorClientDisconnectAsync(client, requestCancellationTokenSource); }); - await WriteJsonResponseAsync(stream, responseJson, serverCancellationToken); + // Stop heartbeats before the final response so no heartbeat frame can be + // queued after the response the CLI stops reading at. + await StopHeartbeatsAsync(heartbeatTask, heartbeatCancellationSource); + heartbeatTask = null; + + await WriteJsonResponseLockedAsync(stream, streamWriteLock, responseJson, serverCancellationToken); } finally { + await StopHeartbeatsAsync(heartbeatTask, heartbeatCancellationSource); + heartbeatCancellationSource?.Dispose(); await StopClientDisconnectMonitorAsync( clientDisconnectMonitorTask, requestCancellationTokenSource); @@ -610,6 +662,71 @@ await StopClientDisconnectMonitorAsync( } } + /// + /// Sends heartbeat frames at the given interval until cancelled. Write failures end + /// the loop silently because the connection teardown is owned by the read loop. + /// + internal static async Task SendHeartbeatsAsync( + Func createHeartbeatJson, + Func writeFrameAsync, + TimeSpan interval, + CancellationToken ct) + { + while (true) + { + try + { + await Task.Delay(interval, ct); + await writeFrameAsync(createHeartbeatJson()); + } + catch (OperationCanceledException) + { + return; + } + catch (IOException) + { + return; + } + catch (ObjectDisposedException) + { + return; + } + } + } + + private static async Task StopHeartbeatsAsync( + Task heartbeatTask, + CancellationTokenSource heartbeatCancellationSource) + { + if (heartbeatTask == null) + { + return; + } + + heartbeatCancellationSource?.Cancel(); + await heartbeatTask; + } + + private async Task WriteJsonResponseLockedAsync( + Stream stream, + SemaphoreSlim streamWriteLock, + string responseJson, + CancellationToken ct) + { + // Why: heartbeat frames are written from a background timer while the final + // response is written by the request task; interleaved writes would corrupt + // Content-Length framing, so all frame writes share one lock per connection. + await streamWriteLock.WaitAsync(ct); + try + { + await WriteJsonResponseAsync(stream, responseJson, ct); + } + finally + { + streamWriteLock.Release(); + } + } + private static async Task MonitorClientDisconnectAsync( BridgeClientConnection client, CancellationTokenSource requestCancellationTokenSource) diff --git a/Packages/src/Editor/ToolContracts/UnityCliLoopServerConfig.cs b/Packages/src/Editor/ToolContracts/UnityCliLoopServerConfig.cs index 5561b6cd6f..b06ffd4492 100644 --- a/Packages/src/Editor/ToolContracts/UnityCliLoopServerConfig.cs +++ b/Packages/src/Editor/ToolContracts/UnityCliLoopServerConfig.cs @@ -9,6 +9,16 @@ public static class UnityCliLoopServerConfig public const int READINESS_PROBE_TIMEOUT_MS = 30000; + // Why: a single failed recovery (e.g. readiness timeout during a heavy import right after + // a domain reload) must not leave the server down until the next reload. The total backoff + // stays well inside the CLI's 180s readiness polling window so a late success is still picked up. + public static readonly int[] RECOVERY_RETRY_DELAYS_MS = { 5000, 15000, 30000 }; + + // Heartbeats prove to the CLI that the server process is alive during long-running + // commands; each frame also reports how long the editor main thread has gone without + // an update tick so the CLI can distinguish "busy" from "frozen". + public const int HEARTBEAT_INTERVAL_SECONDS = 10; + public const string JSONRPC_VERSION = "2.0"; public const int INTERNAL_ERROR_CODE = -32603; diff --git a/cli/contract.json b/cli/contract.json index f3667f0b58..210e458fe1 100644 --- a/cli/contract.json +++ b/cli/contract.json @@ -1,4 +1,4 @@ { "schemaVersion": 1, - "cliVersion": "3.0.0-beta.29" + "cliVersion": "3.0.0-beta.30" } diff --git a/cli/internal/cli/compile_wait.go b/cli/internal/cli/compile_wait.go index a0364023c7..34f5456e5a 100644 --- a/cli/internal/cli/compile_wait.go +++ b/cli/internal/cli/compile_wait.go @@ -6,7 +6,6 @@ import ( "encoding/hex" "encoding/json" "fmt" - "strings" "time" "github.com/hatayama/unity-cli-loop/cli/internal/unityipc" @@ -189,19 +188,6 @@ func shouldWaitForCompileStatus(err error, outcome unityipc.UnitySendOutcome) bo return outcome.RequestAccepted && isFinalResponseTimeoutError(err) } -func isTransportDisconnectError(err error) bool { - message := err.Error() - return message == "UNITY_NO_RESPONSE" || - strings.Contains(message, "EOF") || - strings.Contains(message, "connection reset") || - strings.Contains(message, "broken pipe") || - strings.Contains(message, "use of closed network connection") -} - -func isFinalResponseTimeoutError(err error) bool { - return strings.Contains(err.Error(), "i/o timeout") -} - func logCliDebugModeResolved(connection unityipc.Connection, command string) { if !isCliVibeLogEnabled() { return diff --git a/cli/internal/cli/connection_retry.go b/cli/internal/cli/connection_retry.go index d791c1566c..a72fc37f00 100644 --- a/cli/internal/cli/connection_retry.go +++ b/cli/internal/cli/connection_retry.go @@ -20,6 +20,18 @@ var ( const focusRestoreTimeout = 2 * time.Second +// Why: domain reload on large projects can keep the IPC endpoint down well past the +// base retry window, and an undispatched dial failure is always safe to retry. While +// a running Unity process is confirmed, the dial-retry window stretches to +// base * factor (60s by default) so an external reload does not fail commands +// spuriously. Busy responses stay bounded by the base window: busy proves the server +// is alive and surfacing it quickly lets the caller decide. +const serverConnectionRetryUnityAliveFactor = 6 + +func unityAliveRetryWindow() time.Duration { + return serverConnectionRetryTimeout * serverConnectionRetryUnityAliveFactor +} + type unityServerNotRespondingError struct { projectRoot string endpoint string @@ -69,7 +81,8 @@ func sendWithTransientConnectionRetryAndResponseTimeout( progress unityipc.ProgressFunc, responseTimeout time.Duration, ) (unityipc.UnitySendOutcome, error) { - retryContext, cancel := context.WithTimeout(ctx, serverConnectionRetryTimeout) + startedAt := time.Now() + retryContext, cancel := context.WithTimeout(ctx, unityAliveRetryWindow()) defer cancel() var lastOutcome unityipc.UnitySendOutcome @@ -90,12 +103,23 @@ func sendWithTransientConnectionRetryAndResponseTimeout( if responseTimeout > 0 { client = client.WithResponseTimeout(responseTimeout) } - outcome, err := client.SendWithProgressOutcomeAcceptContext(ctx, retryContext, method, params, progress) + // Each attempt keeps the base accept bound: a dispatched request that never gets + // acked must still fail at the base window. Only the dial-retry loop as a whole + // uses the extended unity-alive window. + attemptContext, cancelAttempt := context.WithTimeout(retryContext, serverConnectionRetryTimeout) + outcome, err := client.SendWithProgressOutcomeAcceptContext(ctx, attemptContext, method, params, progress) + cancelAttempt() if isUnityServerBusyRPCError(err) { // Busy means the request was never executed, so a bounded retry is safe and // usually absorbs back-to-back tool calls without bothering the caller. lastOutcome = outcome lastErr = err + if time.Since(startedAt) >= serverConnectionRetryTimeout { + if ctx.Err() != nil { + return lastOutcome, ctx.Err() + } + return lastOutcome, lastErr + } select { case <-retryContext.Done(): if ctx.Err() != nil { @@ -165,6 +189,16 @@ func sendWithTransientConnectionRetryAndResponseTimeout( lastOutcome = outcome lastErr = err + if time.Since(startedAt) >= unityAliveRetryWindow() { + if ctx.Err() != nil { + return lastOutcome, ctx.Err() + } + return lastOutcome, unityServerNotRespondingError{ + projectRoot: connection.ProjectRoot, + endpoint: connection.Endpoint.Address, + cause: lastErr, + } + } select { case <-retryContext.Done(): if ctx.Err() != nil { diff --git a/cli/internal/cli/connection_retry_test.go b/cli/internal/cli/connection_retry_test.go index a424a77f21..21abf8f21f 100644 --- a/cli/internal/cli/connection_retry_test.go +++ b/cli/internal/cli/connection_retry_test.go @@ -527,6 +527,169 @@ func TestSendWithTransientConnectionRetryReturnsBusyAfterRetryWindow(t *testing. } } +// Verifies that undispatched dial failures keep retrying past the base window while a +// running Unity process is confirmed, so a domain reload longer than the base window +// (e.g. on large projects) does not fail the command spuriously. +func TestSendWithTransientConnectionRetryExtendsWindowWhileUnityProcessRuns(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("TCP endpoint injection is only used by this non-Windows client test") + } + + originalFinder := findRunningUnityProcessForConnectionRetry + originalFocus := focusUnityProcessForConnectionRetry + originalTimeout := serverConnectionRetryTimeout + originalPoll := serverConnectionRetryPoll + findRunningUnityProcessForConnectionRetry = func(context.Context, string) (*unityProcess, error) { + return &unityProcess{pid: 123}, nil + } + focusUnityProcessForConnectionRetry = func(context.Context, int) (restoreFocusFunc, error) { + return func(context.Context) error { return nil }, nil + } + // Base window expires well before the server comes up; only the extended + // unity-alive window (base * factor) allows the late dial to succeed. + serverConnectionRetryTimeout = 100 * time.Millisecond + serverConnectionRetryPoll = 10 * time.Millisecond + t.Cleanup(func() { + findRunningUnityProcessForConnectionRetry = originalFinder + focusUnityProcessForConnectionRetry = originalFocus + serverConnectionRetryTimeout = originalTimeout + serverConnectionRetryPoll = originalPoll + }) + + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("failed to listen: %v", err) + } + address := listener.Addr().String() + // Simulate the IPC endpoint being down during a domain reload: nothing accepts + // until the "reload" finishes at 250ms, past the 100ms base window. + _ = listener.Close() + + serverReady := make(chan net.Listener, 1) + go func() { + time.Sleep(250 * time.Millisecond) + lateListener, listenErr := net.Listen("tcp", address) + if listenErr != nil { + serverReady <- nil + return + } + serverReady <- lateListener + success := `{"jsonrpc":"2.0","id":1,"result":{"ok":true}}` + for { + conn, acceptErr := lateListener.Accept() + if acceptErr != nil { + return + } + go func() { + defer func() { + _ = conn.Close() + }() + if _, readErr := unityipc.Read(bufio.NewReader(conn)); readErr != nil { + return + } + _ = unityipc.Write(conn, []byte(success)) + }() + } + }() + t.Cleanup(func() { + if lateListener := <-serverReady; lateListener != nil { + _ = lateListener.Close() + } + }) + + connection := unityipc.Connection{ + Endpoint: unityipc.Endpoint{ + Network: "tcp", + Address: address, + }, + ProjectRoot: t.TempDir(), + } + + outcome, err := sendWithTransientConnectionRetry( + context.Background(), + connection, + "get-logs", + map[string]any{}, + nil) + if err != nil { + t.Fatalf("expected success after server came back inside the extended window, got %v", err) + } + if len(outcome.Result) == 0 { + t.Fatal("expected a result from the recovered server") + } +} + +// Verifies busy responses still stop retrying at the base window even though the +// unity-alive window for dial failures is longer. +func TestSendWithTransientConnectionRetryKeepsBusyBoundedByBaseWindow(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("TCP endpoint injection is only used by this non-Windows client test") + } + + originalTimeout := serverConnectionRetryTimeout + originalPoll := serverConnectionRetryPoll + serverConnectionRetryTimeout = 200 * time.Millisecond + serverConnectionRetryPoll = 10 * time.Millisecond + t.Cleanup(func() { + serverConnectionRetryTimeout = originalTimeout + serverConnectionRetryPoll = originalPoll + }) + + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("failed to listen: %v", err) + } + defer func() { + _ = listener.Close() + }() + + busy := `{"jsonrpc":"2.0","id":1,"error":{"code":-32603,"message":"Unity is busy running 'compile'.","data":{"type":"server_busy","runningToolName":"compile","requestedToolName":"get-logs","message":"busy"}}}` + go func() { + for { + conn, acceptErr := listener.Accept() + if acceptErr != nil { + return + } + go func() { + defer func() { + _ = conn.Close() + }() + if _, readErr := unityipc.Read(bufio.NewReader(conn)); readErr != nil { + return + } + _ = unityipc.Write(conn, []byte(busy)) + }() + } + }() + + connection := unityipc.Connection{ + Endpoint: unityipc.Endpoint{ + Network: "tcp", + Address: listener.Addr().String(), + }, + ProjectRoot: t.TempDir(), + } + + startedAt := time.Now() + _, err = sendWithTransientConnectionRetry( + context.Background(), + connection, + "get-logs", + map[string]any{}, + nil) + elapsed := time.Since(startedAt) + + var rpcErr *unityipc.RPCError + if !errors.As(err, &rpcErr) { + t.Fatalf("busy must surface as the original RPC error, got: %v", err) + } + // Generous CI margin: the assertion only needs to prove busy did not run for the + // full extended window (base * factor = 1200ms here). + if elapsed >= unityAliveRetryWindow() { + t.Fatalf("busy retries ran into the extended window: %v", elapsed) + } +} + // Verifies a dispatched RPC failure arriving after the retry window expires is not // masked by a busy response seen earlier in the window. func TestSendWithTransientConnectionRetrySurfacesDispatchedFailureAfterBusy(t *testing.T) { diff --git a/cli/internal/cli/control_play_mode_wait.go b/cli/internal/cli/control_play_mode_wait.go index e539d5f4b8..ddbcb4e50d 100644 --- a/cli/internal/cli/control_play_mode_wait.go +++ b/cli/internal/cli/control_play_mode_wait.go @@ -53,9 +53,7 @@ func runControlPlayModeWithStateWait( connection, controlPlayModeCommandName, params, - func(string) { - spinner.Update("Executing control-play-mode...") - }, + newSpinnerProgressFunc(spinner, "Executing control-play-mode..."), ) initialResponse := controlPlayModeResponse{} diff --git a/cli/internal/cli/run.go b/cli/internal/cli/run.go index 96ad4f67ab..f1477dab99 100644 --- a/cli/internal/cli/run.go +++ b/cli/internal/cli/run.go @@ -170,9 +170,7 @@ func runTool(ctx context.Context, connection unityipc.Connection, command string connection, command, params, - func(string) { - spinner.Update(fmt.Sprintf("Executing %s...", command)) - }, + newSpinnerProgressFunc(spinner, fmt.Sprintf("Executing %s...", command)), ) spinner.Stop() if err != nil { @@ -197,9 +195,7 @@ func runExecuteDynamicCodeWithDomainReloadWait(ctx context.Context, connection u connection, executeDynamicCodeCommandName, params, - func(string) { - spinner.Update("Executing execute-dynamic-code...") - }, + newSpinnerProgressFunc(spinner, "Executing execute-dynamic-code..."), ) if err != nil { if shouldWaitForExecuteDynamicCodeDisconnect(err, outcome) { @@ -261,9 +257,7 @@ func runCompileWithDomainReloadWait(ctx context.Context, connection unityipc.Con connection, compileCommandName, params, - func(string) { - spinner.Update("Executing compile...") - }, + newSpinnerProgressFunc(spinner, "Executing compile..."), compileResponseTimeout, ) logCompileRequestSendResult(connection, requestID, outcome, err, startedAt) @@ -330,9 +324,7 @@ func runList(ctx context.Context, connection unityipc.Connection, stdout io.Writ connection, "get-tool-details", map[string]any{}, - func(string) { - spinner.Update("Fetching tool list...") - }, + newSpinnerProgressFunc(spinner, "Fetching tool list..."), ) spinner.Stop() if err != nil { @@ -353,9 +345,7 @@ func runSync(ctx context.Context, connection unityipc.Connection, stdout io.Writ connection, "get-tool-details", map[string]any{}, - func(string) { - spinner.Update("Syncing tools...") - }, + newSpinnerProgressFunc(spinner, "Syncing tools..."), ) spinner.Stop() if err != nil { diff --git a/cli/internal/cli/spinner.go b/cli/internal/cli/spinner.go index 008a499718..87c4c1d00c 100644 --- a/cli/internal/cli/spinner.go +++ b/cli/internal/cli/spinner.go @@ -6,6 +6,8 @@ import ( "os" "sync" "time" + + "github.com/hatayama/unity-cli-loop/cli/internal/unityipc" ) const ( @@ -37,6 +39,20 @@ func newLaunchSpinner(stdout io.Writer, stderr io.Writer) *terminalSpinner { return newSpinner(stderr, isTerminalWriter(stderr), "Waiting for Unity to finish starting...") } +// Maps connection-stage progress events to the contextual executing message +// and passes display-ready payloads (such as the heartbeat main-thread stall +// notice) through to the spinner verbatim. Without this mapping the stall +// diagnosis built by the IPC client would never reach the user. +func newSpinnerProgressFunc(spinner *terminalSpinner, executingMessage string) unityipc.ProgressFunc { + return func(message string) { + if message == unityipc.ProgressEventConnected || message == unityipc.ProgressEventAccepted { + spinner.Update(executingMessage) + return + } + spinner.Update(message) + } +} + func newSpinner(writer io.Writer, enabled bool, message string) *terminalSpinner { spinner := &terminalSpinner{ writer: writer, diff --git a/cli/internal/cli/spinner_test.go b/cli/internal/cli/spinner_test.go index aeed591106..d741ac124b 100644 --- a/cli/internal/cli/spinner_test.go +++ b/cli/internal/cli/spinner_test.go @@ -4,6 +4,8 @@ import ( "bytes" "strings" "testing" + + "github.com/hatayama/unity-cli-loop/cli/internal/unityipc" ) func TestSpinnerDoesNotWriteWhenDisabled(t *testing.T) { @@ -50,6 +52,40 @@ func TestLaunchSpinnerWritesStartupMessage(t *testing.T) { } } +func TestSpinnerProgressFuncShowsStallMessageVerbatim(t *testing.T) { + // Verifies that heartbeat stall payloads reach the spinner verbatim. + var stderr bytes.Buffer + spinner := newSpinner(&stderr, true, "Connecting to Unity...") + progress := newSpinnerProgressFunc(spinner, "Executing get-logs...") + + progress("unity editor main thread busy for 38s...") + spinner.Stop() + + if !strings.Contains(stderr.String(), "unity editor main thread busy for 38s...") { + t.Fatalf("spinner output did not include stall message: %q", stderr.String()) + } +} + +func TestSpinnerProgressFuncMapsConnectionEventsToExecutingMessage(t *testing.T) { + // Verifies that connection-stage events show the contextual executing message + // instead of the raw event token. + var stderr bytes.Buffer + spinner := newSpinner(&stderr, true, "Connecting to Unity...") + progress := newSpinnerProgressFunc(spinner, "Executing get-logs...") + + progress(unityipc.ProgressEventConnected) + progress(unityipc.ProgressEventAccepted) + spinner.Stop() + + output := stderr.String() + if !strings.Contains(output, "Executing get-logs...") { + t.Fatalf("spinner output did not include executing message: %q", output) + } + if strings.Contains(output, unityipc.ProgressEventConnected) || strings.Contains(output, unityipc.ProgressEventAccepted) { + t.Fatalf("spinner output leaked a raw progress event token: %q", output) + } +} + func TestToolFeedbackSkipsExecuteDynamicCode(t *testing.T) { // Verifies that execute-dynamic-code keeps the CLI hot path quiet. if shouldShowToolFeedback(executeDynamicCodeCommandName) { diff --git a/cli/internal/cli/transport_errors.go b/cli/internal/cli/transport_errors.go new file mode 100644 index 0000000000..209e5a346f --- /dev/null +++ b/cli/internal/cli/transport_errors.go @@ -0,0 +1,56 @@ +package cli + +import ( + "errors" + "io" + "net" + "os" + "strings" + "syscall" +) + +// Reports whether the error means the transport dropped mid-request. Domain reload +// recovery (status polling) depends on this classification, so typed causes are +// checked first: error strings differ across platforms and locales (notably Windows +// named pipes), and a missed match would fail the command instead of recovering. +func isTransportDisconnectError(err error) bool { + if err == nil { + return false + } + if errors.Is(err, io.EOF) || + errors.Is(err, io.ErrUnexpectedEOF) || + errors.Is(err, io.ErrClosedPipe) || + errors.Is(err, net.ErrClosed) || + errors.Is(err, syscall.ECONNRESET) || + errors.Is(err, syscall.ECONNABORTED) || + errors.Is(err, syscall.EPIPE) { + return true + } + + // Fallback for errors that expose no typed cause. + message := err.Error() + return message == "UNITY_NO_RESPONSE" || + strings.Contains(message, "EOF") || + strings.Contains(message, "connection reset") || + strings.Contains(message, "broken pipe") || + strings.Contains(message, "use of closed network connection") +} + +// Reports whether the error is a connection deadline expiry. The Timeout() probe +// runs through the unwrap chain because go-winio's named pipe deadline error is not +// os.ErrDeadlineExceeded and os.IsTimeout does not unwrap fmt.Errorf("%w") wrapping. +func isFinalResponseTimeoutError(err error) bool { + if err == nil { + return false + } + if errors.Is(err, os.ErrDeadlineExceeded) || os.IsTimeout(err) { + return true + } + var timeoutCause interface{ Timeout() bool } + if errors.As(err, &timeoutCause) && timeoutCause.Timeout() { + return true + } + + // Fallback for errors that expose no typed cause. + return strings.Contains(err.Error(), "i/o timeout") +} diff --git a/cli/internal/cli/transport_errors_test.go b/cli/internal/cli/transport_errors_test.go new file mode 100644 index 0000000000..dc88498cfa --- /dev/null +++ b/cli/internal/cli/transport_errors_test.go @@ -0,0 +1,108 @@ +package cli + +import ( + "fmt" + "io" + "net" + "os" + "syscall" + "testing" +) + +// Test support type that exposes a typed cause behind an opaque message, simulating +// platform-specific transport errors (e.g. Windows named pipes) whose text does not +// contain the POSIX phrases the string fallback looks for. +type opaqueCauseError struct { + cause error +} + +func (err opaqueCauseError) Error() string { + return "transport failure (code 0x6d)" +} + +func (err opaqueCauseError) Unwrap() error { + return err.cause +} + +// Test support type that reports Timeout() without a recognizable message, +// matching how go-winio surfaces named pipe deadline expiry. +type timeoutOnlyError struct{} + +func (timeoutOnlyError) Error() string { + return "operation did not finish in time" +} + +func (timeoutOnlyError) Timeout() bool { + return true +} + +// Verifies that disconnect classification matches typed causes instead of relying on +// platform- or locale-dependent error strings. +func TestIsTransportDisconnectErrorMatchesTypedCauses(t *testing.T) { + typedCauses := []error{ + io.EOF, + io.ErrUnexpectedEOF, + io.ErrClosedPipe, + net.ErrClosed, + syscall.ECONNRESET, + syscall.ECONNABORTED, + syscall.EPIPE, + } + + for _, cause := range typedCauses { + if !isTransportDisconnectError(opaqueCauseError{cause: cause}) { + t.Fatalf("typed cause was not classified as disconnect: %v", cause) + } + } +} + +// Verifies that non-transport errors are not misclassified as disconnects. +func TestIsTransportDisconnectErrorRejectsUnrelatedErrors(t *testing.T) { + unrelated := []error{ + fmt.Errorf("unity error: compilation failed"), + opaqueCauseError{cause: fmt.Errorf("some inner failure")}, + } + + for _, err := range unrelated { + if isTransportDisconnectError(err) { + t.Fatalf("unrelated error was classified as disconnect: %v", err) + } + } +} + +// Verifies the legacy string fallback still classifies errors without a typed cause. +func TestIsTransportDisconnectErrorKeepsStringFallback(t *testing.T) { + fallbackErrors := []error{ + fmt.Errorf("UNITY_NO_RESPONSE"), + fmt.Errorf("read tcp 127.0.0.1:1: connection reset by peer"), + } + + for _, err := range fallbackErrors { + if !isTransportDisconnectError(err) { + t.Fatalf("string fallback did not classify: %v", err) + } + } +} + +// Verifies that final-response timeout classification matches typed deadline errors +// and Timeout()-reporting errors instead of relying on the "i/o timeout" message. +func TestIsFinalResponseTimeoutErrorMatchesTypedCauses(t *testing.T) { + if !isFinalResponseTimeoutError(opaqueCauseError{cause: os.ErrDeadlineExceeded}) { + t.Fatal("wrapped deadline error was not classified as timeout") + } + if !isFinalResponseTimeoutError(timeoutOnlyError{}) { + t.Fatal("Timeout()-reporting error was not classified as timeout") + } + if isFinalResponseTimeoutError(fmt.Errorf("unity error: busy")) { + t.Fatal("unrelated error was classified as timeout") + } +} + +// Verifies that a Timeout()-reporting error stays classified even when wrapped, which +// is how go-winio's named pipe deadline error reaches the caller on Windows. +func TestIsFinalResponseTimeoutErrorUnwrapsTimeoutCauses(t *testing.T) { + wrapped := opaqueCauseError{cause: timeoutOnlyError{}} + if !isFinalResponseTimeoutError(wrapped) { + t.Fatal("wrapped Timeout()-reporting error was not classified as timeout") + } +} diff --git a/cli/internal/unityipc/client.go b/cli/internal/unityipc/client.go index 814cd815e9..610e5e20ad 100644 --- a/cli/internal/unityipc/client.go +++ b/cli/internal/unityipc/client.go @@ -4,8 +4,10 @@ import ( "bufio" "context" "encoding/json" + "errors" "fmt" "net" + "os" "time" ) @@ -14,7 +16,22 @@ const ( finalResponseTimeout = 30 * time.Minute ) -const rpcResponsePhaseAccepted = "accepted" +const ( + rpcResponsePhaseAccepted = "accepted" + rpcResponsePhaseHeartbeat = "heartbeat" +) + +// Why: heartbeat silence means the server process died without closing the socket or +// its sender stalled; six missed heartbeats is well past scheduling jitter. +const heartbeatSilenceGraceFactor = 6 + +// Why: editor main-thread work running five minutes without a single update tick +// almost always means a frozen editor. Failing here with a diagnosis beats sitting on +// the 30-minute absolute deadline while the editor needs a restart. +const defaultMainThreadStallLimit = 5 * time.Minute + +// Stall reports below this are normal editor busyness and not worth surfacing. +const mainThreadStallProgressThresholdSeconds = 30 type Client struct { connection Connection @@ -22,10 +39,21 @@ type Client struct { clientVersion string acceptTimeout time.Duration responseTimeout time.Duration + // Test seams: zero means "use the derived/default values". + heartbeatSilenceOverride time.Duration + mainThreadStallLimit time.Duration } type ProgressFunc = func(message string) +// Connection-stage progress events. Consumers map these tokens to their own +// contextual message; any other progress payload is display-ready text such +// as the main-thread stall notice. +const ( + ProgressEventConnected = "connected" + ProgressEventAccepted = "accepted" +) + type rpcRequest struct { JSONRPC string `json:"jsonrpc"` Method string `json:"method"` @@ -37,6 +65,7 @@ type rpcRequest struct { type rpcClientMetadata struct { CLIVersion string `json:"cliVersion"` AcceptsDispatchAck bool `json:"acceptsDispatchAck"` + AcceptsHeartbeat bool `json:"acceptsHeartbeat"` } type rpcResponse struct { @@ -48,7 +77,21 @@ type rpcResponse struct { } type rpcResponseMeta struct { - Phase string `json:"phase,omitempty"` + Phase string `json:"phase,omitempty"` + HeartbeatIntervalSeconds int `json:"heartbeatIntervalSeconds,omitempty"` + MainThreadStallSeconds float64 `json:"mainThreadStallSeconds,omitempty"` +} + +// Reports that Unity's editor main thread stopped pumping while the IPC connection +// stayed alive — the freeze case heartbeats exist to expose. +type EditorUnresponsiveError struct { + StallSeconds float64 +} + +func (err *EditorUnresponsiveError) Error() string { + return fmt.Sprintf( + "unity editor main thread has been unresponsive for %.0f seconds; the editor is likely frozen. Restart it with 'uloop launch -r'", + err.StallSeconds) } type rpcError struct { @@ -128,7 +171,7 @@ func (client *Client) SendWithProgressOutcomeAcceptContext( }() if progress != nil { - progress("connected") + progress(ProgressEventConnected) } client.requestID++ @@ -139,6 +182,7 @@ func (client *Client) SendWithProgressOutcomeAcceptContext( ULoop: rpcClientMetadata{ CLIVersion: client.clientVersion, AcceptsDispatchAck: true, + AcceptsHeartbeat: true, }, ID: client.requestID, } @@ -172,11 +216,13 @@ func (client *Client) SendWithProgressOutcomeAcceptContext( if response.ULoop.Phase == rpcResponsePhaseAccepted { outcome.RequestAccepted = true if progress != nil { - progress("accepted") + progress(ProgressEventAccepted) } cancelAccept() - if err := setConnectionDeadlineFromNow(conn, client.getResponseTimeout()); err != nil { + heartbeatSilence := client.getHeartbeatSilence(response.ULoop.HeartbeatIntervalSeconds) + absoluteDeadline := time.Now().Add(client.getResponseTimeout()) + if err := applyPostAcceptDeadline(conn, heartbeatSilence, absoluteDeadline); err != nil { timing.Total = time.Since(startedAt) outcome.Timing = timing return outcome, err @@ -184,14 +230,46 @@ func (client *Client) SendWithProgressOutcomeAcceptContext( stopCancelWatcher := watchConnectionCancellation(ctx, conn) defer stopCancelWatcher() - response, err = readRPCResponse(reader, &timing) - if err != nil { - timing.Total = time.Since(startedAt) - outcome.Timing = timing + for { + response, err = readRPCResponse(reader, &timing) + if err != nil { + timing.Total = time.Since(startedAt) + outcome.Timing = timing + if ctx.Err() != nil { + return outcome, ctx.Err() + } + if heartbeatSilence > 0 && isDeadlineExpiry(err) && time.Now().Before(absoluteDeadline) { + return outcome, fmt.Errorf( + "no response or heartbeat from Unity for %s; the connection or server stalled: %w", + heartbeatSilence, err) + } + return outcome, err + } + if response.ULoop.Phase != rpcResponsePhaseHeartbeat { + break + } if ctx.Err() != nil { + timing.Total = time.Since(startedAt) + outcome.Timing = timing return outcome, ctx.Err() } - return outcome, err + + stallSeconds := response.ULoop.MainThreadStallSeconds + if stallSeconds >= client.getMainThreadStallLimit().Seconds() { + timing.Total = time.Since(startedAt) + outcome.Timing = timing + return outcome, &EditorUnresponsiveError{StallSeconds: stallSeconds} + } + if progress != nil && stallSeconds >= mainThreadStallProgressThresholdSeconds { + progress(fmt.Sprintf("unity editor main thread busy for %.0fs...", stallSeconds)) + } + if heartbeatSilence > 0 { + if err := applyPostAcceptDeadline(conn, heartbeatSilence, absoluteDeadline); err != nil { + timing.Total = time.Since(startedAt) + outcome.Timing = timing + return outcome, err + } + } } } @@ -230,8 +308,52 @@ func (client *Client) getResponseTimeout() time.Duration { return finalResponseTimeout } -func setConnectionDeadlineFromNow(conn net.Conn, timeout time.Duration) error { - return conn.SetDeadline(time.Now().Add(timeout)) +// Derives the sliding silence window for a negotiated heartbeat connection. +// Zero disables sliding: an explicit response timeout (e.g. compile's quick fallback +// to status polling) must stay an absolute deadline, and servers that did not +// negotiate heartbeats keep the legacy absolute deadline too. +func (client *Client) getHeartbeatSilence(heartbeatIntervalSeconds int) time.Duration { + if client.responseTimeout > 0 { + return 0 + } + if heartbeatIntervalSeconds <= 0 { + return 0 + } + if client.heartbeatSilenceOverride > 0 { + return client.heartbeatSilenceOverride + } + return time.Duration(heartbeatIntervalSeconds) * time.Second * heartbeatSilenceGraceFactor +} + +func (client *Client) getMainThreadStallLimit() time.Duration { + if client.mainThreadStallLimit > 0 { + return client.mainThreadStallLimit + } + return defaultMainThreadStallLimit +} + +// Reports whether the error is a connection deadline expiry. go-winio's named pipe +// deadline error is not os.ErrDeadlineExceeded, so a Timeout() probe through the +// unwrap chain is required for Windows. +func isDeadlineExpiry(err error) bool { + if errors.Is(err, os.ErrDeadlineExceeded) || os.IsTimeout(err) { + return true + } + var timeoutCause interface{ Timeout() bool } + return errors.As(err, &timeoutCause) && timeoutCause.Timeout() +} + +// Sets the connection deadline to the sliding heartbeat-silence window, capped by the +// absolute response deadline so heartbeats can never extend the total wait past it. +func applyPostAcceptDeadline(conn net.Conn, heartbeatSilence time.Duration, absoluteDeadline time.Time) error { + deadline := absoluteDeadline + if heartbeatSilence > 0 { + slidingDeadline := time.Now().Add(heartbeatSilence) + if slidingDeadline.Before(deadline) { + deadline = slidingDeadline + } + } + return conn.SetDeadline(deadline) } func watchConnectionCancellation(ctx context.Context, conn net.Conn) func() { diff --git a/cli/internal/unityipc/client_heartbeat_test.go b/cli/internal/unityipc/client_heartbeat_test.go new file mode 100644 index 0000000000..b6263329d7 --- /dev/null +++ b/cli/internal/unityipc/client_heartbeat_test.go @@ -0,0 +1,237 @@ +package unityipc + +import ( + "bufio" + "context" + "encoding/json" + "errors" + "net" + "os" + "runtime" + "strings" + "testing" + "time" +) + +// Test support server that acks with heartbeat negotiation and then runs the given +// frame script (heartbeats and/or a final response) over one accepted connection. +func startHeartbeatTestServer(t *testing.T, serve func(conn net.Conn)) Connection { + t.Helper() + if runtime.GOOS == "windows" { + t.Skip("TCP endpoint injection is only used by this non-Windows client test") + } + + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("failed to listen: %v", err) + } + t.Cleanup(func() { + _ = listener.Close() + }) + + go func() { + conn, acceptErr := listener.Accept() + if acceptErr != nil { + return + } + defer func() { + _ = conn.Close() + }() + if _, readErr := Read(bufio.NewReader(conn)); readErr != nil { + return + } + serve(conn) + }() + + return Connection{ + Endpoint: Endpoint{ + Network: "tcp", + Address: listener.Addr().String(), + }, + ProjectRoot: t.TempDir(), + } +} + +func writeFrame(t *testing.T, conn net.Conn, payload string) { + t.Helper() + if err := Write(conn, []byte(payload)); err != nil { + t.Errorf("failed to write frame: %v", err) + } +} + +const heartbeatAck = `{"jsonrpc":"2.0","id":1,"result":{"accepted":true},"uloop":{"phase":"accepted","heartbeatIntervalSeconds":1}}` + +// Verifies that the request advertises heartbeat support so the server can negotiate. +func TestSendAdvertisesHeartbeatSupport(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("TCP endpoint injection is only used by this non-Windows client test") + } + + captured := make(chan map[string]any, 1) + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("failed to listen: %v", err) + } + defer func() { + _ = listener.Close() + }() + go func() { + conn, acceptErr := listener.Accept() + if acceptErr != nil { + return + } + defer func() { + _ = conn.Close() + }() + payload, readErr := Read(bufio.NewReader(conn)) + if readErr != nil { + return + } + var request map[string]any + if json.Unmarshal(payload, &request) == nil { + captured <- request + } + _ = Write(conn, []byte(`{"jsonrpc":"2.0","id":1,"result":{"ok":true}}`)) + }() + connection := Connection{ + Endpoint: Endpoint{Network: "tcp", Address: listener.Addr().String()}, + ProjectRoot: t.TempDir(), + } + + client := NewClient(connection, "9.9.9") + if _, err := client.Send(context.Background(), "get-version", map[string]any{}); err != nil { + t.Fatalf("send failed: %v", err) + } + + request := <-captured + metadata, ok := request["uloop"].(map[string]any) + if !ok { + t.Fatalf("request misses uloop metadata: %#v", request) + } + if metadata["acceptsHeartbeat"] != true { + t.Fatalf("acceptsHeartbeat not advertised: %#v", metadata) + } +} + +// Verifies that heartbeat frames slide the silence deadline so a slow final response +// arrives even though the initial silence window alone would have expired. +func TestSendKeepsWaitingWhileHeartbeatsArrive(t *testing.T) { + heartbeat := `{"jsonrpc":"2.0","id":1,"result":{"alive":true},"uloop":{"phase":"heartbeat","mainThreadStallSeconds":0}}` + connection := startHeartbeatTestServer(t, func(conn net.Conn) { + writeFrame(t, conn, heartbeatAck) + for i := 0; i < 6; i++ { + time.Sleep(50 * time.Millisecond) + writeFrame(t, conn, heartbeat) + } + writeFrame(t, conn, `{"jsonrpc":"2.0","id":1,"result":{"ok":true}}`) + }) + + client := NewClient(connection, "9.9.9") + client.heartbeatSilenceOverride = 200 * time.Millisecond + + outcome, err := client.SendWithProgressOutcome(context.Background(), "run-tests", map[string]any{}, nil) + if err != nil { + t.Fatalf("expected success with heartbeats sliding the deadline, got %v", err) + } + if !outcome.RequestAccepted { + t.Fatal("expected accepted outcome") + } + if len(outcome.Result) == 0 { + t.Fatal("expected final result") + } +} + +// Verifies that a negotiated connection fails with a heartbeat-silence diagnosis when +// frames stop arriving, instead of waiting for the 30-minute absolute deadline. +func TestSendFailsWithDiagnosisWhenHeartbeatsStop(t *testing.T) { + heartbeat := `{"jsonrpc":"2.0","id":1,"result":{"alive":true},"uloop":{"phase":"heartbeat","mainThreadStallSeconds":0}}` + done := make(chan struct{}) + connection := startHeartbeatTestServer(t, func(conn net.Conn) { + writeFrame(t, conn, heartbeatAck) + writeFrame(t, conn, heartbeat) + <-done + }) + defer close(done) + + client := NewClient(connection, "9.9.9") + client.heartbeatSilenceOverride = 150 * time.Millisecond + + startedAt := time.Now() + _, err := client.SendWithProgressOutcome(context.Background(), "run-tests", map[string]any{}, nil) + if err == nil { + t.Fatal("expected heartbeat silence error") + } + if !strings.Contains(err.Error(), "heartbeat") { + t.Fatalf("error misses heartbeat diagnosis: %v", err) + } + if !errors.Is(err, os.ErrDeadlineExceeded) && !os.IsTimeout(err) { + t.Fatalf("silence error must stay classified as a timeout: %v", err) + } + if elapsed := time.Since(startedAt); elapsed > 5*time.Second { + t.Fatalf("silence detection took too long: %v", elapsed) + } +} + +// Verifies that a heartbeat reporting a main-thread stall beyond the limit fails with +// an editor-unresponsive diagnosis while the connection is still alive. +func TestSendFailsWhenMainThreadStallExceedsLimit(t *testing.T) { + stalledHeartbeat := `{"jsonrpc":"2.0","id":1,"result":{"alive":true},"uloop":{"phase":"heartbeat","mainThreadStallSeconds":3}}` + done := make(chan struct{}) + connection := startHeartbeatTestServer(t, func(conn net.Conn) { + writeFrame(t, conn, heartbeatAck) + writeFrame(t, conn, stalledHeartbeat) + <-done + }) + defer close(done) + + client := NewClient(connection, "9.9.9") + client.heartbeatSilenceOverride = 5 * time.Second + client.mainThreadStallLimit = 2 * time.Second + + _, err := client.SendWithProgressOutcome(context.Background(), "run-tests", map[string]any{}, nil) + var unresponsiveErr *EditorUnresponsiveError + if !errors.As(err, &unresponsiveErr) { + t.Fatalf("expected EditorUnresponsiveError, got %v", err) + } + if !strings.Contains(err.Error(), "uloop launch -r") { + t.Fatalf("error misses restart hint: %v", err) + } +} + +// Verifies that an explicit response timeout stays an absolute deadline: heartbeats +// are skipped but must not extend it (compile relies on this to fall back to polling). +func TestSendKeepsExplicitResponseTimeoutDespiteHeartbeats(t *testing.T) { + heartbeat := `{"jsonrpc":"2.0","id":1,"result":{"alive":true},"uloop":{"phase":"heartbeat","mainThreadStallSeconds":0}}` + done := make(chan struct{}) + connection := startHeartbeatTestServer(t, func(conn net.Conn) { + writeFrame(t, conn, heartbeatAck) + for { + select { + case <-done: + return + case <-time.After(20 * time.Millisecond): + // The client is expected to time out and close the connection while + // this loop is still writing, so write failures end the loop silently + // instead of failing the test through writeFrame's t.Errorf. + if err := Write(conn, []byte(heartbeat)); err != nil { + return + } + } + } + }) + defer close(done) + + client := NewClient(connection, "9.9.9").WithResponseTimeout(150 * time.Millisecond) + + startedAt := time.Now() + _, err := client.SendWithProgressOutcome(context.Background(), "compile", map[string]any{}, nil) + if err == nil { + t.Fatal("expected timeout despite heartbeats") + } + if !os.IsTimeout(err) && !errors.Is(err, os.ErrDeadlineExceeded) { + t.Fatalf("expected deadline error, got %v", err) + } + if elapsed := time.Since(startedAt); elapsed > 5*time.Second { + t.Fatalf("explicit response timeout was extended by heartbeats: %v", elapsed) + } +} diff --git a/cli/internal/unityipc/client_windows_test.go b/cli/internal/unityipc/client_windows_test.go new file mode 100644 index 0000000000..1a732ab082 --- /dev/null +++ b/cli/internal/unityipc/client_windows_test.go @@ -0,0 +1,153 @@ +//go:build windows + +package unityipc + +import ( + "bufio" + "context" + "errors" + "fmt" + "io" + "net" + "os" + "strings" + "testing" + "time" + + "github.com/Microsoft/go-winio" +) + +// These tests exercise the real Windows named pipe transport (go-winio) so the +// platform assumptions behind heartbeat deadlines and typed disconnect +// classification are verified on CI's Windows runners, not just on TCP. + +func testPipeConnection(t *testing.T) (Connection, net.Listener) { + t.Helper() + pipePath := fmt.Sprintf(`\\.\pipe\uloop-test-%d-%s`, os.Getpid(), strings.ReplaceAll(t.Name(), "/", "_")) + listener, err := winio.ListenPipe(pipePath, nil) + if err != nil { + t.Fatalf("failed to listen on named pipe: %v", err) + } + t.Cleanup(func() { + _ = listener.Close() + }) + + connection := Connection{ + Endpoint: Endpoint{ + Network: "pipe", + Address: pipePath, + }, + ProjectRoot: t.TempDir(), + } + return connection, listener +} + +func serveOnePipeRequest(t *testing.T, listener net.Listener, serve func(conn net.Conn)) { + t.Helper() + go func() { + conn, acceptErr := listener.Accept() + if acceptErr != nil { + return + } + defer func() { + _ = conn.Close() + }() + if _, readErr := Read(bufio.NewReader(conn)); readErr != nil { + return + } + serve(conn) + }() +} + +// Verifies that heartbeat frames slide the silence deadline over a real named pipe, +// proving go-winio honors repeated SetDeadline calls the way the client relies on. +func TestPipeSendKeepsWaitingWhileHeartbeatsArrive(t *testing.T) { + heartbeat := `{"jsonrpc":"2.0","id":1,"result":{"alive":true},"uloop":{"phase":"heartbeat","mainThreadStallSeconds":0}}` + connection, listener := testPipeConnection(t) + serveOnePipeRequest(t, listener, func(conn net.Conn) { + writeFrame(t, conn, heartbeatAck) + for i := 0; i < 6; i++ { + time.Sleep(50 * time.Millisecond) + writeFrame(t, conn, heartbeat) + } + writeFrame(t, conn, `{"jsonrpc":"2.0","id":1,"result":{"ok":true}}`) + }) + + client := NewClient(connection, "9.9.9") + client.heartbeatSilenceOverride = 200 * time.Millisecond + + outcome, err := client.SendWithProgressOutcome(context.Background(), "run-tests", map[string]any{}, nil) + if err != nil { + t.Fatalf("expected success with heartbeats sliding the deadline, got %v", err) + } + if len(outcome.Result) == 0 { + t.Fatal("expected final result") + } +} + +// Verifies that heartbeat silence over a named pipe fails with the diagnosis and a +// timeout-classified cause, proving the winio deadline error is recognized. +func TestPipeSendFailsWithDiagnosisWhenHeartbeatsStop(t *testing.T) { + done := make(chan struct{}) + connection, listener := testPipeConnection(t) + serveOnePipeRequest(t, listener, func(conn net.Conn) { + writeFrame(t, conn, heartbeatAck) + <-done + }) + defer close(done) + + client := NewClient(connection, "9.9.9") + client.heartbeatSilenceOverride = 150 * time.Millisecond + + _, err := client.SendWithProgressOutcome(context.Background(), "run-tests", map[string]any{}, nil) + if err == nil { + t.Fatal("expected heartbeat silence error") + } + if !strings.Contains(err.Error(), "heartbeat") { + t.Fatalf("error misses heartbeat diagnosis: %v", err) + } + if !isDeadlineExpiry(err) { + t.Fatalf("winio deadline expiry was not classified as timeout: %v", err) + } +} + +// Verifies that closing the pipe after the ack surfaces as typed io.EOF, the premise +// the transport disconnect classification relies on for Windows domain reloads. +func TestPipeServerCloseAfterAckYieldsTypedEOF(t *testing.T) { + connection, listener := testPipeConnection(t) + serveOnePipeRequest(t, listener, func(conn net.Conn) { + writeFrame(t, conn, heartbeatAck) + }) + + client := NewClient(connection, "9.9.9") + + _, err := client.SendWithProgressOutcome(context.Background(), "run-tests", map[string]any{}, nil) + if err == nil { + t.Fatal("expected disconnect error after server closed the pipe") + } + if !errors.Is(err, io.EOF) && !errors.Is(err, io.ErrUnexpectedEOF) { + t.Fatalf("pipe close did not surface as typed EOF: %v", err) + } +} + +// Verifies that dialing a missing pipe fails immediately with a dial error, the +// premise behind treating it as a retryable undispatched connection attempt. +func TestPipeDialMissingPipeFailsFast(t *testing.T) { + connection := Connection{ + Endpoint: Endpoint{ + Network: "pipe", + Address: fmt.Sprintf(`\\.\pipe\uloop-test-missing-%d`, os.Getpid()), + }, + ProjectRoot: t.TempDir(), + } + + client := NewClient(connection, "9.9.9") + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + _, err := client.SendWithProgressOutcome(ctx, "get-logs", map[string]any{}, nil) + var connectionErr *ConnectionAttemptError + if !errors.As(err, &connectionErr) { + t.Fatalf("expected ConnectionAttemptError for missing pipe, got %v", err) + } +}