Skip to content

chore: Simplify bridge heartbeat and disconnect handling #1554

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
hatayama merged 2 commits into from
Jul 6, 2026
Merged

chore: Simplify bridge heartbeat and disconnect handling #1554

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
32a8303
Extract bridge heartbeat and disconnect monitor services
hatayama Jul 6, 2026
4d499a0
Restore bridge monitor extraction comments
hatayama Jul 6, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 4 additions & 2 deletions Assets/Tests/Editor/JsonRpcHeartbeatTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,8 +55,9 @@ public async Task SendHeartbeatsAsync_WhenRunning_WritesFramesUntilCancelled()
// without leaving background work behind.
int writtenFrameCount = 0;
using CancellationTokenSource cancellationSource = new();
UnityCliLoopBridgeHeartbeatService heartbeatService = new();

Task heartbeatTask = UnityCliLoopBridgeServer.SendHeartbeatsAsync(
Task heartbeatTask = heartbeatService.SendHeartbeatsAsync(
() => "{}",
_ =>
{
Expand All @@ -79,8 +80,9 @@ public async Task SendHeartbeatsAsync_WhenWriteThrowsIOException_StopsWithoutFau
// 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();
UnityCliLoopBridgeHeartbeatService heartbeatService = new();

Task heartbeatTask = UnityCliLoopBridgeServer.SendHeartbeatsAsync(
Task heartbeatTask = heartbeatService.SendHeartbeatsAsync(
() => "{}",
_ => throw new System.IO.IOException("broken pipe"),
TimeSpan.FromMilliseconds(1),
Expand Down
56 changes: 56 additions & 0 deletions Packages/src/Editor/Infrastructure/UnityCliLoopBridgeClientDisconnectMonitor.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,56 @@
using System;
using System.Threading;
using System.Threading.Tasks;

namespace io.github.hatayama.UnityCliLoop.Infrastructure
{
/// <summary>
/// Cancels accepted project IPC requests when their client connection disappears.
/// </summary>
internal sealed class UnityCliLoopBridgeClientDisconnectMonitor
{
private const int ClientDisconnectMonitorPollMilliseconds = 100;

/// <summary>
/// Monitors an accepted client connection and cancels the request token source when the client disconnects.
/// </summary>
internal async Task MonitorClientDisconnectAsync(
BridgeClientConnection client,
CancellationTokenSource requestCancellationTokenSource)
{
while (!requestCancellationTokenSource.IsCancellationRequested)
{
if (!client.IsConnected)
{
requestCancellationTokenSource.Cancel();
return;
}

try
{
await Task.Delay(ClientDisconnectMonitorPollMilliseconds, requestCancellationTokenSource.Token);
}
catch (OperationCanceledException)
{
// Cancellation is the normal stop signal from StopClientDisconnectMonitorAsync.
// Without the token the delay always ran to completion, adding one poll
// interval of tail latency to every request teardown and server shutdown.
return;
}
}
}

internal async Task StopClientDisconnectMonitorAsync(
Task clientDisconnectMonitorTask,
CancellationTokenSource requestCancellationTokenSource)
{
if (clientDisconnectMonitorTask == null)
{
return;
}

requestCancellationTokenSource.Cancel();
await clientDisconnectMonitorTask;
}
}
}
11 changes: 11 additions & 0 deletions Packages/src/Editor/Infrastructure/UnityCliLoopBridgeClientDisconnectMonitor.cs.meta
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

58 changes: 58 additions & 0 deletions Packages/src/Editor/Infrastructure/UnityCliLoopBridgeHeartbeatService.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,58 @@
using System;
using System.IO;
using System.Threading;
using System.Threading.Tasks;

namespace io.github.hatayama.UnityCliLoop.Infrastructure
{
/// <summary>
/// Sends and stops project IPC heartbeat frames for an accepted client request.
/// </summary>
internal sealed class UnityCliLoopBridgeHeartbeatService
{
/// <summary>
/// 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.
/// </summary>
internal async Task SendHeartbeatsAsync(
Func<string> createHeartbeatJson,
Func<string, Task> 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;
}
}
}

internal async Task StopHeartbeatsAsync(
Task heartbeatTask,
CancellationTokenSource heartbeatCancellationSource)
{
if (heartbeatTask == null)
{
return;
}

heartbeatCancellationSource?.Cancel();
await heartbeatTask;
}
}
}
11 changes: 11 additions & 0 deletions Packages/src/Editor/Infrastructure/UnityCliLoopBridgeHeartbeatService.cs.meta
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

117 changes: 25 additions & 92 deletions Packages/src/Editor/Infrastructure/UnityCliLoopBridgeServer.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,12 @@ internal UnityCliLoopBridgeServerInstanceFactory(IDomainReloadDetectionService d

public IUnityCliLoopServerInstance Create()
{
UnityCliLoopBridgeServer server = new(_domainReloadDetectionService);
UnityCliLoopBridgeHeartbeatService heartbeatService = new();
UnityCliLoopBridgeClientDisconnectMonitor clientDisconnectMonitor = new();
UnityCliLoopBridgeServer server = new(
_domainReloadDetectionService,
heartbeatService,
clientDisconnectMonitor);
server.ServerLoopExited += NotifyServerLoopExited;

return server;
Expand All @@ -57,6 +62,8 @@ public class UnityCliLoopBridgeServer : IUnityCliLoopServerInstance
// Subscribers must marshal to main thread before accessing Unity APIs.
public event Action ServerLoopExited;
private readonly IDomainReloadDetectionService _domainReloadDetectionService;
private readonly UnityCliLoopBridgeHeartbeatService _heartbeatService;
private readonly UnityCliLoopBridgeClientDisconnectMonitor _clientDisconnectMonitor;

// HResult error codes for normal disconnection detection
private static readonly HashSet<int> NormalDisconnectionHResults = new()
Expand All @@ -79,14 +86,22 @@ public class UnityCliLoopBridgeServer : IUnityCliLoopServerInstance
private readonly ConcurrentDictionary<string, Stream> _clientStreams = new();
private readonly ConcurrentDictionary<int, Task> _clientTasks = new();
private int _nextClientTaskId;
private const int ClientDisconnectMonitorPollMilliseconds = 100;

internal UnityCliLoopBridgeServer(IDomainReloadDetectionService domainReloadDetectionService)
internal UnityCliLoopBridgeServer(
IDomainReloadDetectionService domainReloadDetectionService,
UnityCliLoopBridgeHeartbeatService heartbeatService,
UnityCliLoopBridgeClientDisconnectMonitor clientDisconnectMonitor)
{
System.Diagnostics.Debug.Assert(domainReloadDetectionService != null, "domainReloadDetectionService must not be null");
System.Diagnostics.Debug.Assert(heartbeatService != null, "heartbeatService must not be null");
System.Diagnostics.Debug.Assert(clientDisconnectMonitor != null, "clientDisconnectMonitor must not be null");

_domainReloadDetectionService = domainReloadDetectionService
?? throw new ArgumentNullException(nameof(domainReloadDetectionService));
_heartbeatService = heartbeatService
?? throw new ArgumentNullException(nameof(heartbeatService));
_clientDisconnectMonitor = clientDisconnectMonitor
?? throw new ArgumentNullException(nameof(clientDisconnectMonitor));
}

/// <summary>
Expand Down Expand Up @@ -616,7 +631,7 @@ await WriteJsonResponseLockedAsync(
// server token an in-flight write could ignore StopHeartbeatsAsync
// and stall the final response behind a slow client.
CancellationToken heartbeatToken = heartbeatCancellationSource.Token;
heartbeatTask = SendHeartbeatsAsync(
heartbeatTask = _heartbeatService.SendHeartbeatsAsync(
createHeartbeatJson,
heartbeatJson => WriteJsonResponseLockedAsync(
stream, streamWriteLock, heartbeatJson, heartbeatToken),
Expand All @@ -630,72 +645,29 @@ await WriteJsonResponseLockedAsync(
}

clientDisconnectMonitorTask =
MonitorClientDisconnectAsync(client, requestCancellationTokenSource);
_clientDisconnectMonitor.MonitorClientDisconnectAsync(
client,
requestCancellationTokenSource);
});

// 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);
await _heartbeatService.StopHeartbeatsAsync(heartbeatTask, heartbeatCancellationSource);
heartbeatTask = null;

await WriteJsonResponseLockedAsync(stream, streamWriteLock, responseJson, serverCancellationToken);
}
finally
{
await StopHeartbeatsAsync(heartbeatTask, heartbeatCancellationSource);
await _heartbeatService.StopHeartbeatsAsync(heartbeatTask, heartbeatCancellationSource);
heartbeatCancellationSource?.Dispose();
await StopClientDisconnectMonitorAsync(
await _clientDisconnectMonitor.StopClientDisconnectMonitorAsync(
clientDisconnectMonitorTask,
requestCancellationTokenSource);
}
}
}

/// <summary>
/// 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.
/// </summary>
internal static async Task SendHeartbeatsAsync(
Func<string> createHeartbeatJson,
Func<string, Task> 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,
Expand All @@ -716,45 +688,6 @@ private async Task WriteJsonResponseLockedAsync(
}
}

private static async Task MonitorClientDisconnectAsync(
BridgeClientConnection client,
CancellationTokenSource requestCancellationTokenSource)
{
while (!requestCancellationTokenSource.IsCancellationRequested)
{
if (!client.IsConnected)
{
requestCancellationTokenSource.Cancel();
return;
}

try
{
await Task.Delay(ClientDisconnectMonitorPollMilliseconds, requestCancellationTokenSource.Token);
}
catch (OperationCanceledException)
{
// Cancellation is the normal stop signal from StopClientDisconnectMonitorAsync.
// Without the token the delay always ran to completion, adding one poll
// interval of tail latency to every request teardown and server shutdown.
return;
}
}
}

private static async Task StopClientDisconnectMonitorAsync(
Task clientDisconnectMonitorTask,
CancellationTokenSource requestCancellationTokenSource)
{
if (clientDisconnectMonitorTask == null)
{
return;
}

requestCancellationTokenSource.Cancel();
await clientDisconnectMonitorTask;
}

/// <summary>
/// Determines if the given exception represents a normal client disconnection.
/// </summary>
Expand Down
Loading