Move websocket internal events to channels, misc fixes

.github/workflows/build.yml

@@ -57,6 +57,22 @@ jobs:
       tests_failed: ${{ steps.covstats.outputs.tests_failed }}
       tests_skipped: ${{ steps.covstats.outputs.tests_skipped }}
     steps:
+
+      - name: Trace pkill calls
+        run: |
+          mkdir -p "$HOME/bin"
+          cat > "$HOME/bin/pkill" <<'SHIM'
+          #!/bin/bash
+          {
+            echo "[pkill-trace] args: $*"
+            echo "[pkill-trace] caller pid=$PPID"
+            ps -o pid=,ppid=,comm=,args= -p "$PPID" 2>/dev/null
+          } >&2
+          exec /usr/bin/pkill "$@"
+          SHIM
+          chmod +x "$HOME/bin/pkill"
+          echo "$HOME/bin" >> "$GITHUB_PATH"
+
       - name: Checkout repository
         uses: actions/checkout@v6
 
@@ -82,10 +98,37 @@ jobs:
           --configuration Release \
           --logger "console;verbosity=detailed" \
           --logger "trx;LogFileName=test-results.trx" \
+          --blame \
+          --blame-crash \
+          --blame-hang \
+          --blame-hang-timeout 5m \
           --collect:"XPlat Code Coverage" \
           --results-directory ./coverage \
           -- DataCollectionRunSettings.DataCollectors.DataCollector.Configuration.ExcludeByFile="**/Migrations/**;**/*.g.cs;**/SubathonManager.UI/**"
 
+      - name: Show blame / crash diagnostics
+        if: always()
+        shell: bash
+        run: |
+          echo "--- files produced by --blame ---"
+          find ./coverage -iname "Sequence*.xml" -o -iname "*.dmp" -o -iname "core*" || true
+          for f in $(find ./coverage -iname "Sequence*.xml" 2>/dev/null); do
+            echo "===== $f ====="
+            cat "$f"
+          done
+
+      - name: Upload blame / crash diagnostics
+        if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          name: test-blame-diagnostics
+          path: |
+            coverage/**/Sequence*.xml
+            coverage/**/*.dmp
+            coverage/**/*.trx
+          if-no-files-found: ignore
+          retention-days: 7
+
       - name: List raw coverage files
         if: always()
         shell: bash

.gitignore

@@ -30,6 +30,8 @@ presets/**
 !.idea/inspectionProfiles/
 !.idea/runConfigurations/
 
+resources/HONSE/*
+
 ## Ignore Visual Studio temporary files, build results, and
 ## files generated by popular Visual Studio add-ons.
 

SubathonManager.Data/SubathonManager.Data.csproj

@@ -13,14 +13,14 @@
     </ItemGroup>
 
     <ItemGroup>
-      <PackageReference Include="Microsoft.EntityFrameworkCore" Version="10.0.9" />
-      <PackageReference Include="Microsoft.EntityFrameworkCore.Design" Version="10.0.9">
+      <PackageReference Include="Microsoft.EntityFrameworkCore" Version="10.0.11" />
+      <PackageReference Include="Microsoft.EntityFrameworkCore.Design" Version="10.0.11">
         <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
         <PrivateAssets>all</PrivateAssets>
       </PackageReference>
-      <PackageReference Include="Microsoft.EntityFrameworkCore.Sqlite" Version="10.0.9" />
+      <PackageReference Include="Microsoft.EntityFrameworkCore.Sqlite" Version="10.0.11" />
       <!-- Overrides EF's transitive SQLitePCLRaw 2.1.11 for CVE-2025-6965 (GHSA-2m69-gcr7-jv3q) TODO -->
-      <PackageReference Include="SQLitePCLRaw.bundle_e_sqlite3" Version="3.0.3" />
+      <PackageReference Include="SQLitePCLRaw.bundle_e_sqlite3" Version="3.0.5" />
     </ItemGroup>
 
     <ItemGroup>

SubathonManager.Integration/SubathonManager.Integration.csproj

@@ -16,7 +16,7 @@
       <PackageReference Include="Fourthwall.Client" Version="0.1.0" />
       <PackageReference Include="KoFi.Client" Version="0.1.1" />
       <PackageReference Include="Agash.YTLiveChat" Version="4.2.0" />
-      <PackageReference Include="GoAffPro.Client" Version="0.4.0" />
+      <PackageReference Include="GoAffPro.Client" Version="0.4.1" />
       <PackageReference Include="obs-websocket-dotnet" Version="5.0.1" />
       <PackageReference Include="StreamElements.WebSocket" Version="1.0.2" />
       <PackageReference Include="Streamlabs.SocketClient" Version="3.0.0" />

SubathonManager.Server/Interfaces/IWebSocketClient.cs

@@ -26,4 +26,10 @@ Task CloseAsync(
         WebSocketCloseStatus closeStatus,
         string statusDescription,
         CancellationToken cancellationToken);
+
+    void Abort();
+    void StartOutbound();
+    bool TryEnqueue(byte[] payload, OutboundCoalesceKey key = OutboundCoalesceKey.None);
+    void CompleteOutbound();
+    Task CompleteOutboundAsync(TimeSpan drainTimeout);
 }

SubathonManager.Server/WebServer.WebSocket.cs

@@ -20,7 +20,7 @@ public partial class WebServer
 {
     private readonly List<IWebSocketClient> _clients = new();
     private readonly object _lock = new();
-    private readonly SemaphoreSlim _sendLock = new(1,1);
+    private static readonly TimeSpan OutboundDrainTimeout = TimeSpan.FromSeconds(2);
 
     private void SetupWebsocketListeners()
     {
@@ -68,6 +68,16 @@ private void StopWebsocketServer()
         WheelEvents.WheelSpinResult -= SendWheelSpinResult;
         WheelEvents.WheelSpinStatusChanged -= SendWheelSpinStatusChanged;
         WheelEvents.WheelDataChanged -= SendWheelDataChanged;
+
+        List<IWebSocketClient> clientsCopy;
+        lock (_lock)
+        {
+            clientsCopy = _clients.ToList();
+            _clients.Clear();
+        }
+
+        foreach (var client in clientsCopy)
+            client.CompleteOutbound();
     }
 
     private void OnPromptStart(SubathonPromptRun subathonPromptRun, SubathonPrompt? subathonPrompt)
@@ -98,7 +108,7 @@ internal void SendWidgetReload(Guid widgetId, float x, float y, int width, int h
             widgetId = widgetId.ToString(),
             x, y, width, height, scaleX, scaleY
         };
-        Task.Run(() => BroadcastAsyncObject(data, WebsocketClientMessageType.Overlay));
+        BroadcastObject(data, WebsocketClientMessageType.Overlay);
     }
 
     internal void SendPromptData(SubathonPromptRun? run, long progress = 0)
@@ -119,7 +129,7 @@ internal void SendPromptData(SubathonPromptRun? run, long progress = 0)
             prompt_eventtype = $"{run?.LinkedPrompt?.FilterEventType}",
             prompt_eventtype_metafilter =  run?.LinkedPrompt?.FilterMeta
         };
-        Task.Run(() => BroadcastAsyncObject(data, WebsocketClientTypeHelper.ConsumersList));
+        BroadcastObject(data, WebsocketClientTypeHelper.ConsumersList);
     }
 
     internal void SendWidgetVarsUpdate(Guid widgetId, 
@@ -133,13 +143,13 @@ internal void SendWidgetVarsUpdate(Guid widgetId,
             jsVars  = jsVars.Select(v => new { name = v.Name, value = v.Value, 
                 injectLine = v.GetInjectLine() })
         };
-        Task.Run(() => BroadcastAsyncObject(data, WebsocketClientTypeHelper.ConsumersList));
+        BroadcastObject(data, WebsocketClientTypeHelper.ConsumersList);
     }
 
     internal void SendSubathonValues(string jsonData)
     {
         var newData = $"{{ \"type\": \"value_config\", \"ws_type\": \"{WebsocketClientMessageType.ValueConfig}\", \"data\": {jsonData} }}";
-        Task.Run(() => BroadcastAsync(newData, WebsocketClientTypeHelper.ConfigConsumersList));
+        Broadcast(newData, OutboundCoalesceKey.None, WebsocketClientTypeHelper.ConfigConsumersList);
     }
 
     internal void SendGoalsUpdated(List<SubathonGoal> goals, long currentPoints, GoalsType type)
@@ -151,7 +161,7 @@ internal void SendGoalsUpdated(List<SubathonGoal> goals, long currentPoints, Goa
             goals = goals.Select(goal => GoalToObject(goal, currentPoints)).ToArray(),
             goals_type = $"{type}"
         };
-        Task.Run(() => BroadcastAsyncObject(data, WebsocketClientTypeHelper.ConsumersList));
+        BroadcastObject(data, OutboundCoalesceKey.GoalsList, WebsocketClientTypeHelper.ConsumersList);
     }
 
     private object GoalToObject(SubathonGoal goal, long currentPoints)
@@ -173,7 +183,7 @@ internal void SendGoalCompleted(SubathonGoal goal, long currentPoints)
             goal_points =  goal.Points,
             points = currentPoints
         };
-        Task.Run(() => BroadcastAsyncObject(data, WebsocketClientTypeHelper.ConsumersList));
+        BroadcastObject(data, WebsocketClientTypeHelper.ConsumersList);
     }
 
     private async Task InitConnection(IWebSocketClient socket)
@@ -343,28 +353,28 @@ internal void SendSubathonEventProcessed(SubathonEvent subathonEvent, bool effec
     {
         bool showOverride = _config.GetBool("App", "ShowLockedEvents", false);
         if (!showOverride && !subathonEvent.ProcessedToSubathon) return;
-        Task.Run(() => BroadcastAsyncObject(SubathonEventToObject(subathonEvent), WebsocketClientTypeHelper.ConsumersList));
+        BroadcastObject(SubathonEventToObject(subathonEvent), WebsocketClientTypeHelper.ConsumersList);
     }
 
     internal void SendSubathonTotals(SubathonTotals totals)
     {
-        Task.Run(() => BroadcastAsyncObject(SubathonTotalsToObject(totals), WebsocketClientTypeHelper.ConsumersList));
+        BroadcastObject(SubathonTotalsToObject(totals), OutboundCoalesceKey.SubathonTotals,
+            WebsocketClientTypeHelper.ConsumersList);
     }
 
     internal void SendSubscriptionTotals(SubscriptionTotals totals)
     {
-        Task.Run(() => BroadcastAsyncObject(SubscriptionTotalsToObject(totals), WebsocketClientTypeHelper.ConsumersList));
+        BroadcastObject(SubscriptionTotalsToObject(totals), OutboundCoalesceKey.SubscriptionTotals,
+            WebsocketClientTypeHelper.ConsumersList);
     }
 
     internal void SendRefreshRequest(Guid id)
     {
-        Task.Run(() =>
-            BroadcastAsyncObject(new
-            {
-                type = "refresh_request",
-                id = id.ToString()
-            }, WebsocketClientMessageType.Overlay)
-        );
+        BroadcastObject(new
+        {
+            type = "refresh_request",
+            id = id.ToString()
+        }, WebsocketClientMessageType.Overlay);
     }
 
     private object SubathonDataToObject(SubathonData subathon)
@@ -408,7 +418,8 @@ private object SubathonDataToObject(SubathonData subathon)
 
     internal void SendSubathonDataUpdate(SubathonData subathon, DateTime time)
     {
-        Task.Run(() => BroadcastAsyncObject(SubathonDataToObject(subathon), WebsocketClientTypeHelper.ConsumersList));
+        BroadcastObject(SubathonDataToObject(subathon), OutboundCoalesceKey.SubathonTimer,
+            WebsocketClientTypeHelper.ConsumersList);
     }
 
     internal void AddSocketClient(IWebSocketClient socket)
@@ -418,6 +429,7 @@ internal void AddSocketClient(IWebSocketClient socket)
             _clients.Add(socket);
             _logger?.LogDebug("{ClientsCount} websocket clients connected", _clients.Count);
         }
+        socket.StartOutbound();
     }
 
     public async Task HandleWebSocketRequestAsync(IHttpContext ctx)
@@ -448,6 +460,8 @@ public async Task HandleWebSocketRequestAsync(IHttpContext ctx)
         }
         finally
         {
+            await client.CompleteOutboundAsync(OutboundDrainTimeout);
+
             foreach (var clientIntegrationSource in client.IntegrationSources)
             {
                 WebServerEvents.RaiseWebSocketIntegrationSourceChange(clientIntegrationSource.ToString(), false);
@@ -486,16 +500,7 @@ private async Task Listen(IWebSocketClient socket)
                     switch (type.GetString())
                     {
                         case "ping":
-                            var pong = Encoding.UTF8.GetBytes("{\"ws_type\":\"pong\"}");
-                            await _sendLock.WaitAsync();
-                            try
-                            {
-                                await socket.SendAsync(pong, WebSocketMessageType.Text, true, CancellationToken.None);
-                            }
-                            finally
-                            {
-                                _sendLock.Release();
-                            }
+                            socket.TryEnqueue(Encoding.UTF8.GetBytes("{\"ws_type\":\"pong\"}"));
                             break;
                         case "hello":
                             _logger?.LogDebug($"[WebSocket] [{socket.ClientId}] Hello from {json.RootElement.GetProperty("origin").GetString()}");
@@ -687,7 +692,7 @@ private void SendWheelSpinStarted(WheelSet wheel, int delaySeconds)
             spin_delay_seconds = delaySeconds,
             timestamp = DateTime.Now
         };
-        Task.Run(() => BroadcastAsyncObject(data, WebsocketClientTypeHelper.ConsumersList));
+        BroadcastObject(data, WebsocketClientTypeHelper.ConsumersList);
     }
 
     private void SendWheelSpinResult(WheelSet wheel, WheelItem? item, WheelSpinHistory history, int _)
@@ -707,7 +712,7 @@ private void SendWheelSpinResult(WheelSet wheel, WheelItem? item, WheelSpinHisto
             },
             timestamp = DateTime.Now
         };
-        Task.Run(() => BroadcastAsyncObject(data, WebsocketClientTypeHelper.ConsumersList));
+        BroadcastObject(data, WebsocketClientTypeHelper.ConsumersList);
     }
 
     private void SendWheelSpinStatusChanged(WheelSpinHistory history, int _)
@@ -721,7 +726,7 @@ private void SendWheelSpinStatusChanged(WheelSpinHistory history, int _)
             updated_at = history.UpdatedAt,
             wheel_item = itemSnapshot
         };
-        Task.Run(() => BroadcastAsyncObject(data, WebsocketClientTypeHelper.ConsumersList));
+        BroadcastObject(data, WebsocketClientTypeHelper.ConsumersList);
     }
 
     private void SendWheelDataChanged(WheelSet wheel, int spinsOwed)
@@ -737,16 +742,16 @@ private void SendWheelDataChanged(WheelSet wheel, int spinsOwed)
             wheel = new { id = wheelId, name = wheelName, spin_count = spinCount, items },
             spins_owed = spinsOwed
         };
-        Task.Run(() => BroadcastAsyncObject(data, WebsocketClientTypeHelper.ConsumersList));
+        BroadcastObject(data, WebsocketClientTypeHelper.ConsumersList);
     }
 
-    private async Task BroadcastAsyncObject(object data, params WebsocketClientMessageType[] types)
-    {
-        string json = JsonSerializer.Serialize(data);
-        await BroadcastAsync(json, types);
-    }
-    
-    private async Task BroadcastAsync(string json, params WebsocketClientMessageType[] types)
+    private void BroadcastObject(object data, OutboundCoalesceKey key, params WebsocketClientMessageType[] types)
+        => Broadcast(JsonSerializer.Serialize(data), key, types);
+
+    private void BroadcastObject(object data, params WebsocketClientMessageType[] types)
+        => Broadcast(JsonSerializer.Serialize(data), OutboundCoalesceKey.None, types);
+
+    private void Broadcast(string json, OutboundCoalesceKey key, params WebsocketClientMessageType[] types)
     {
         byte[] bytes = Encoding.UTF8.GetBytes(json);
 
@@ -757,51 +762,18 @@ private async Task BroadcastAsync(string json, params WebsocketClientMessageType
         foreach (var ws in clientsCopy.Where(ws =>
                      ws.State == WebSocketState.Open && ws.ClientTypes.Any(types.Contains)))
         {
-            await _sendLock.WaitAsync();
-            try
-            {
-                await ws.SendAsync(bytes, WebSocketMessageType.Text, true, CancellationToken.None);
-            }
-            finally
-            {
-                _sendLock.Release();
-            }
+            ws.TryEnqueue(bytes, key);
         }
     }
 
-    internal async Task SelectSendAsync(IWebSocketClient client, object data)
-    {
-        string json = JsonSerializer.Serialize(data);
-        byte[] bytes = Encoding.UTF8.GetBytes(json);
-        if (client.State == WebSocketState.Open)
-        {
-            await _sendLock.WaitAsync();
-            try
-            {
-                await client.SendAsync(bytes, WebSocketMessageType.Text, true, CancellationToken.None);
-            }
-            finally
-            {
-                _sendLock.Release();
-            }
-        }
-    }
-
-    internal async Task SelectSendStringAsync(IWebSocketClient client, string data)
+    internal Task SelectSendAsync(IWebSocketClient client, object data)
+        => SelectSendStringAsync(client, JsonSerializer.Serialize(data));
+
+    internal Task SelectSendStringAsync(IWebSocketClient client, string data)
     {
-        byte[] bytes = Encoding.UTF8.GetBytes(data);
         if (client.State == WebSocketState.Open)
-        {
-            await _sendLock.WaitAsync();
-            try
-            {
-                await client.SendAsync(bytes, WebSocketMessageType.Text, true, CancellationToken.None);
-            }
-            finally
-            {
-                _sendLock.Release();
-            }
-        }
+            client.TryEnqueue(Encoding.UTF8.GetBytes(data));
+        return Task.CompletedTask;
     }
 
     public string GetWebsocketInjectionScript(string? routeId = "")