fix: drop hub control commands that waited too long to send

Common/Constants/Constants.cs

@@ -22,5 +22,7 @@ public static class Duration
     public static readonly TimeSpan DeviceKeepAliveInitialTimeout = TimeSpan.FromSeconds(65);
     public static readonly TimeSpan DeviceKeepAliveTimeout = TimeSpan.FromSeconds(35);
 
+    public static readonly TimeSpan HubControlMaxQueueAge = TimeSpan.FromSeconds(5);
+
     public static readonly TimeSpan DeactivatedAccountRetentionTime = TimeSpan.FromDays(14);
 }

Common/Websocket/WebsockBaseController.cs

@@ -1,4 +1,5 @@
-using System.Net.WebSockets;
+using System.Diagnostics;
+using System.Net.WebSockets;
 using System.Text.Json;
 using System.Threading.Channels;
 using Microsoft.AspNetCore.Mvc;
@@ -45,7 +46,12 @@ public abstract class WebsocketBaseController<T> : OpenShockControllerBase, IAsy
     /// <summary>
     /// A message waiting to be sent, <paramref name="Sent"/> is completed with whether it was written to the websocket
     /// </summary>
-    protected readonly record struct QueuedMessage(T Data, TaskCompletionSource<bool>? Sent);
+    /// <param name="Data"></param>
+    /// <param name="Sent"></param>
+    /// <param name="ExpiresAt">
+    /// <see cref="Stopwatch.GetTimestamp"/> after which the message is dropped instead of written, 0 for never.
+    /// </param>
+    protected readonly record struct QueuedMessage(T Data, TaskCompletionSource<bool>? Sent, long ExpiresAt = 0);
 
 #pragma warning disable IDISP008
     protected WebSocket? WebSocket;
@@ -73,6 +79,23 @@ public ValueTask QueueMessage(T data)
         return Channel.Writer.WriteAsync(new QueuedMessage(data, null), LinkedToken);
     }
 
+    /// <summary>
+    /// Queue a message that is only worth sending promptly. If the client is so far behind that it has
+    /// waited longer than <paramref name="maxQueueAge"/> when its turn comes, it is dropped.
+    /// </summary>
+    [NonAction]
+    public ValueTask QueueMessage(T data, TimeSpan maxQueueAge)
+    {
+        if (!CanSend())
+        {
+            Logger.LogDebug("WebSocket is null, abort or closed, not sending message");
+            return ValueTask.CompletedTask;
+        }
+
+        var expiresAt = Stopwatch.GetTimestamp() + (long)(maxQueueAge.TotalSeconds * Stopwatch.Frequency);
+        return Channel.Writer.WriteAsync(new QueuedMessage(data, null, expiresAt), LinkedToken);
+    }
+
     /// <summary>
     /// Queue a message and wait until it has been written to the websocket
     /// </summary>
@@ -207,6 +230,13 @@ private async Task MessageLoop()
         {
             await foreach (var msg in Channel.Reader.ReadAllAsync(LinkedToken))
             {
+                if (msg.ExpiresAt != 0 && Stopwatch.GetTimestamp() > msg.ExpiresAt)
+                {
+                    Logger.LogDebug("Dropping a message that waited too long in the send queue");
+                    msg.Sent?.TrySetResult(false);
+                    continue;
+                }
+
                 try
                 {
                     await SendWebSocketMessage(msg.Data, WebSocket!, LinkedToken);

LiveControlGateway/Controllers/HubV1Controller.cs

@@ -3,6 +3,7 @@
 using Microsoft.AspNetCore.Mvc;
 using Microsoft.AspNetCore.SignalR;
 using OpenShock.Common.Authentication;
+using OpenShock.Common.Constants;
 using OpenShock.Common.Hubs;
 using OpenShock.Common.Models;
 using OpenShock.Common.OpenShockDb;
@@ -172,7 +173,7 @@ await otaService.Error(CurrentHubId, payload.BootStatus.OtaUpdateId, false,
     /// <inheritdoc />
     [NonAction]
     public override ValueTask Control(IList<ShockerCommand> controlCommands)
-        => QueueMessage(ControlMessage(controlCommands));
+        => QueueMessage(ControlMessage(controlCommands), Duration.HubControlMaxQueueAge);
 
     /// <inheritdoc />
     [NonAction]

LiveControlGateway/Controllers/HubV2Controller.cs

@@ -202,7 +202,7 @@ await otaService.Error(CurrentHubId, payload.BootStatus.OtaUpdateId, false,
 
     /// <inheritdoc />
     public override ValueTask Control(IList<ShockerCommand> controlCommands)
-        => QueueMessage(ControlMessage(controlCommands));
+        => QueueMessage(ControlMessage(controlCommands), Duration.HubControlMaxQueueAge);
 
     /// <inheritdoc />
     public override Task<bool> SendControl(IList<ShockerCommand> controlCommands)