diff --git a/Common/Constants/Constants.cs b/Common/Constants/Constants.cs index 659c9b30..d4ff3203 100644 --- a/Common/Constants/Constants.cs +++ b/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); } diff --git a/Common/Websocket/WebsockBaseController.cs b/Common/Websocket/WebsockBaseController.cs index 6d8fe2b6..3b3d22a7 100644 --- a/Common/Websocket/WebsockBaseController.cs +++ b/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 : OpenShockControllerBase, IAsy /// /// A message waiting to be sent, is completed with whether it was written to the websocket /// - protected readonly record struct QueuedMessage(T Data, TaskCompletionSource? Sent); + /// + /// + /// + /// after which the message is dropped instead of written, 0 for never. + /// + protected readonly record struct QueuedMessage(T Data, TaskCompletionSource? 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); } + /// + /// Queue a message that is only worth sending promptly. If the client is so far behind that it has + /// waited longer than when its turn comes, it is dropped. + /// + [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); + } + /// /// Queue a message and wait until it has been written to the websocket /// @@ -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); diff --git a/LiveControlGateway/Controllers/HubV1Controller.cs b/LiveControlGateway/Controllers/HubV1Controller.cs index 9430fa9a..2fd90185 100644 --- a/LiveControlGateway/Controllers/HubV1Controller.cs +++ b/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, /// [NonAction] public override ValueTask Control(IList controlCommands) - => QueueMessage(ControlMessage(controlCommands)); + => QueueMessage(ControlMessage(controlCommands), Duration.HubControlMaxQueueAge); /// [NonAction] diff --git a/LiveControlGateway/Controllers/HubV2Controller.cs b/LiveControlGateway/Controllers/HubV2Controller.cs index d3a12fa4..147ddbac 100644 --- a/LiveControlGateway/Controllers/HubV2Controller.cs +++ b/LiveControlGateway/Controllers/HubV2Controller.cs @@ -202,7 +202,7 @@ await otaService.Error(CurrentHubId, payload.BootStatus.OtaUpdateId, false, /// public override ValueTask Control(IList controlCommands) - => QueueMessage(ControlMessage(controlCommands)); + => QueueMessage(ControlMessage(controlCommands), Duration.HubControlMaxQueueAge); /// public override Task SendControl(IList controlCommands)