Efficient frame coalescing and ack for viewer delivery

Introduce ClientSendGrain to queue and coalesce screen frames per viewer, ensuring only the latest frame is delivered after explicit client ack. Update client/server protocol to support frame acknowledgments. Add integration tests for coalescing logic. Improve logging and refactor message delivery for reliability and backpressure.

Author: haefele

src/RemoteViewer.Client/RemoteViewer.Client.csproj

@@ -7,6 +7,10 @@
     <AvaloniaUseCompiledBindingsByDefault>true</AvaloniaUseCompiledBindingsByDefault>
   </PropertyGroup>
 
+  <ItemGroup>
+    <InternalsVisibleTo Include="RemoteViewer.IntegrationTests" />
+  </ItemGroup>
+
   <ItemGroup>
     <AvaloniaResource Include="Assets\**" />
   </ItemGroup>

src/RemoteViewer.Client/Services/HubClient/Connection.cs

@@ -463,6 +463,10 @@ async void IConnectionImpl.OnMessageReceived(string senderClientId, string messa
                     {
                         var message = ProtocolSerializer.Deserialize<FrameMessage>(data);
                         ((IViewerServiceImpl)this.ViewerService!).HandleFrame(message.DisplayId, message.FrameNumber, message.Codec, message.Regions);
+
+                        if (this.Owner.Options.SuppressAutoFrameAck is false)
+                            await this.Owner.SendAckFrameAsync();
+
                         break;
                     }
 

src/RemoteViewer.Client/Services/HubClient/ConnectionHubClient.cs

@@ -23,13 +23,15 @@ public ConnectionHubClient(
     {
         this._logger = logger;
         this._serviceProvider = serviceProvider;
+        this.Options = options.Value;
 
         this._connection = new HubConnectionBuilder()
-            .WithUrl($"{options.Value.BaseUrl}/connection", httpOptions =>
+            .WithUrl($"{this.Options.BaseUrl}/connection", httpOptions =>
             {
                 httpOptions.Headers.Add("X-Client-Version", ThisAssembly.AssemblyInformationalVersion);
                 httpOptions.Headers.Add("X-Display-Name", this.DisplayName);
             })
+
             .WithAutomaticReconnect()
             .AddMessagePackProtocol(Witness.GeneratedTypeShapeProvider)
             .Build();
@@ -174,6 +176,8 @@ private void CloseAllConnections()
         this._connections.Clear();
     }
 
+    public ConnectionHubClientOptions Options { get; }
+
     public string? ClientId { get; private set; }
     public string? Username { get; private set; }
     public string? Password { get; private set; }
@@ -260,9 +264,9 @@ public async Task ConnectToHub()
 
             return error;
         }
-        catch (Exception ex) when (!this.IsConnected)
+        catch (Exception ex)
         {
-            this._logger.LogWarning(ex, "Failed to connect to device - hub disconnected");
+            this._logger.LogWarning(ex, "Failed to connect to device");
             return null;
         }
     }
@@ -278,9 +282,9 @@ public async Task GenerateNewPassword()
             await this._connection.InvokeAsync("GenerateNewPassword");
             this._logger.LogInformation("New password generated");
         }
-        catch (Exception ex) when (!this.IsConnected)
+        catch (Exception ex)
         {
-            this._logger.LogWarning(ex, "Failed to generate new password - hub disconnected");
+            this._logger.LogWarning(ex, "Failed to generate new password");
         }
     }
 
@@ -297,9 +301,9 @@ public async Task SetDisplayName(string displayName)
             await this._connection.InvokeAsync("SetDisplayName", displayName);
             this._logger.LogInformation("Display name set successfully");
         }
-        catch (Exception ex) when (!this.IsConnected)
+        catch (Exception ex)
         {
-            this._logger.LogWarning(ex, "Failed to set display name - hub disconnected");
+            this._logger.LogWarning(ex, "Failed to set display name");
         }
     }
 
@@ -321,9 +325,9 @@ internal async Task SendMessageAsync(string connectionId, string messageType, Re
             await this._connection.SendAsync("SendMessage", connectionId, messageType, data, destination, targetClientIds);
             this._logger.LogDebug("Message sent successfully");
         }
-        catch (Exception ex) when (!this.IsConnected)
+        catch (Exception ex)
         {
-            this._logger.LogWarning(ex, "Failed to send message - hub disconnected");
+            this._logger.LogWarning(ex, "Failed to send message");
         }
     }
 
@@ -338,9 +342,9 @@ internal async Task DisconnectAsync(string connectionId)
             await this._connection.InvokeAsync("Disconnect", connectionId);
             this._logger.LogInformation("Disconnected from connection: {ConnectionId}", connectionId);
         }
-        catch (Exception ex) when (!this.IsConnected)
+        catch (Exception ex)
         {
-            this._logger.LogWarning(ex, "Failed to disconnect - hub disconnected");
+            this._logger.LogWarning(ex, "Failed to disconnect");
         }
     }
 
@@ -354,9 +358,9 @@ internal async Task SetConnectionPropertiesAsync(string connectionId, Connection
             this._logger.LogDebug("Setting connection properties - ConnectionId: {ConnectionId}", connectionId);
             await this._connection.InvokeAsync("SetConnectionProperties", connectionId, properties);
         }
-        catch (Exception ex) when (!this.IsConnected)
+        catch (Exception ex)
         {
-            this._logger.LogWarning(ex, "Failed to set connection properties - hub disconnected");
+            this._logger.LogWarning(ex, "Failed to set connection properties");
         }
     }
 
@@ -370,12 +374,27 @@ internal async Task SetConnectionPropertiesAsync(string connectionId, Connection
             this._logger.LogDebug("Generating IPC auth token for connection: {ConnectionId}", connectionId);
             return await this._connection.InvokeAsync<string?>("GenerateIpcAuthToken", connectionId);
         }
-        catch (Exception ex) when (!this.IsConnected)
+        catch (Exception ex)
         {
-            this._logger.LogWarning(ex, "Failed to generate IPC auth token - hub disconnected");
+            this._logger.LogWarning(ex, "Failed to generate IPC auth token");
             return null;
         }
     }
+
+    internal async Task SendAckFrameAsync()
+    {
+        if (!this.IsConnected || this.IsReconnecting)
+            return;
+
+        try
+        {
+            await this._connection.SendAsync("AckFrame");
+        }
+        catch (Exception ex)
+        {
+            this._logger.LogWarning(ex, "Failed to send frame ack");
+        }
+    }
 }
 
 #region EventArgs Classes

src/RemoteViewer.Client/Services/HubClient/ConnectionHubClientOptions.cs

@@ -1,6 +1,5 @@
-using Microsoft.AspNetCore.Http.Connections;
+namespace RemoteViewer.Client.Services.HubClient;
 
-namespace RemoteViewer.Client.Services.HubClient;
 
 public class ConnectionHubClientOptions
 {
@@ -9,4 +8,6 @@ public class ConnectionHubClientOptions
 #else
     public string BaseUrl { get; set; } = "https://rdp.xemio.net";
 #endif
+
+    public bool SuppressAutoFrameAck { get; set; }
 }

src/RemoteViewer.Server/Hubs/ConnectionHub.cs

@@ -79,6 +79,11 @@ public async Task SendMessage(string connectionId, string messageType, byte[] da
         await clientsService.SendMessage(this.Context.ConnectionId, connectionId, messageType, data, destination, targetClientIds);
     }
 
+    public Task AckFrame()
+    {
+        return clientsService.AckFrame(this.Context.ConnectionId);
+    }
+
     public async Task SetConnectionProperties(string connectionId, ConnectionProperties properties)
     {
         await clientsService.SetConnectionProperties(this.Context.ConnectionId, connectionId, properties);

src/RemoteViewer.Server/Orleans/Grains/ClientGrain.cs

@@ -35,6 +35,8 @@ public sealed partial class ClientGrain(ILogger<ClientGrain> logger, IHubContext
 
     private string _displayName = string.Empty;
 
+    private IClientSendGrain? _sendGrain;
+
     private IConnectionGrain? _presenterConnectionGrain;
     private readonly List<IConnectionGrain> _viewerConnectionGrains = [];
 
@@ -67,6 +69,8 @@ public async Task Initialize(string? displayName)
             this.LogUsernameCollision(attempts);
         }
 
+        this._sendGrain = this.GrainFactory.GetGrain<IClientSendGrain>(this.GetPrimaryKeyString());
+
         this.LogClientInitialized(this._clientId, this._usernameGrain.GetPrimaryKeyString());
 
         await hubContext.Clients
@@ -93,6 +97,7 @@ public async Task Deactivate()
             await connection.Internal_RemoveClient(this.AsReference<IClientGrain>());
         }
 
+        await this._sendGrain.Disconnect();
         await this._usernameGrain.ReleaseAsync(this.GetPrimaryKeyString());
 
         this.DeactivateOnIdle();
@@ -208,14 +213,15 @@ private static string FormatUsername(string username)
         }
         return sb.ToString();
     }
-    [MemberNotNull(nameof(_clientId), nameof(_usernameGrain), nameof(_password))]
+    [MemberNotNull(nameof(_clientId), nameof(_usernameGrain), nameof(_sendGrain), nameof(_password))]
     private void EnsureInitialized()
     {
-        if (this._clientId is null || this._usernameGrain is null || this._password is null)
+        if (this._clientId is null || this._usernameGrain is null || this._sendGrain is null || this._password is null)
         {
             throw new InvalidOperationException(
                 $"ClientGrain not initialized: clientId={(this._clientId is null ? "null" : "set")}, " +
                 $"usernameGrain={(this._usernameGrain is null ? "null" : "set")}, " +
+                $"sendGrain={(this._sendGrain is null ? "null" : "set")}, " +
                 $"password={(this._password is null ? "null" : "set")}");
         }
     }