diff --git a/Resgrid.Audio.Relay.Console/Program.cs b/Resgrid.Audio.Relay.Console/Program.cs index d183cc0..d18fe48 100644 --- a/Resgrid.Audio.Relay.Console/Program.cs +++ b/Resgrid.Audio.Relay.Console/Program.cs @@ -9,6 +9,7 @@ using System.IO; using System.Linq; using System.Reflection; +using System.Runtime.InteropServices; using System.Text.Json; using System.Threading; using System.Threading.Tasks; @@ -85,6 +86,16 @@ private static async Task RunAsync() }; Cli.CancelKeyPress += cancelHandler; + // Also shut down gracefully on SIGTERM — the default signal `systemctl stop` / `docker stop` + // send. Mirrors the Ctrl+C / SIGINT path so in-flight work (recordings, LiveKit) unwinds + // cleanly instead of the process being killed. Disposed before the CTS (reverse using order), + // so the handler can never fire against a disposed token source. + using var sigTermRegistration = PosixSignalRegistration.Create(PosixSignal.SIGTERM, context => + { + context.Cancel = true; + cancellationTokenSource.Cancel(); + }); + try { // The engine owns all mode wiring; the console just builds the service and runs it. diff --git a/Resgrid.Audio.Voice.Tests/RelayResilienceTests.cs b/Resgrid.Audio.Voice.Tests/RelayResilienceTests.cs new file mode 100644 index 0000000..dbe0dbc --- /dev/null +++ b/Resgrid.Audio.Voice.Tests/RelayResilienceTests.cs @@ -0,0 +1,139 @@ +using System; +using System.Threading; +using System.Threading.Tasks; +using FluentAssertions; +using NUnit.Framework; +using Resgrid.Relay.Engine; +using Resgrid.Relay.Engine.Configuration; +using Resgrid.Relay.Engine.Services; + +namespace Resgrid.Audio.Voice.Tests +{ + /// + /// Behavioral coverage for the LiveKit-mode resilience loop in + /// : exponential back-off retries, the consecutive-failure + /// circuit breaker, and graceful stop while retrying. Uses a tiny test subclass with a + /// scripted ExecuteAsync and near-zero back-off so the tests run fast. + /// + [TestFixture] + public class RelayResilienceTests + { + private static RelayHostOptions OptionsWith(ResilienceOptions resilience) + { + return new RelayHostOptions + { + Resilience = resilience, + // No Sentry DSN ⇒ telemetry is the no-op singleton (no SDK init). + Telemetry = new RelayTelemetryOptions() + }; + } + + private static ResilienceOptions FastResilience(int maxFailures, double healthySeconds = 1000) => new ResilienceOptions + { + Enabled = true, + MaxConsecutiveFailures = maxFailures, + InitialBackoffSeconds = 0.01, + MaxBackoffSeconds = 0.05, + HealthyRunSeconds = healthySeconds + }; + + [Test] + public void ComputeBackoff_GrowsExponentially_AndIsCappedWithJitter() + { + var r = new ResilienceOptions + { + InitialBackoffSeconds = 2, + MaxBackoffSeconds = 60, + MaxConsecutiveFailures = 10, + HealthyRunSeconds = 30 + }; + + // failure 1 ≈ 2s ±20% ⇒ [1.6, 2.4]; failure 3 ≈ 8s ±20% ⇒ [6.4, 9.6]; + // failure 10 caps at 60s ±20% ⇒ [48, 72]; floor is always >= 0.5s. + RelayServiceBase.ComputeBackoff(r, 1).TotalSeconds.Should().BeInRange(1.6, 2.4); + RelayServiceBase.ComputeBackoff(r, 3).TotalSeconds.Should().BeInRange(6.4, 9.6); + RelayServiceBase.ComputeBackoff(r, 10).TotalSeconds.Should().BeInRange(48, 72); + RelayServiceBase.ComputeBackoff(r, 1).TotalSeconds.Should().BeGreaterThanOrEqualTo(0.5); + } + + [Test] + public async Task AlwaysFailing_LiveKitMode_TripsBreaker_AndFaults() + { + var svc = new ScriptedRelayService(OptionsWith(FastResilience(maxFailures: 3))); + + // ExecuteAsync throws immediately every time; the breaker opens on the 3rd failure. + svc.ExecuteBehavior = (_, __) => throw new InvalidOperationException("boom"); + + await svc.StartAsync(CancellationToken.None); + + svc.State.Should().Be(RelayServiceState.Faulted); + svc.Attempts.Should().Be(3, "the breaker opens once MaxConsecutiveFailures quick failures pile up"); + svc.Status.LiveKit.Should().Be(ConnectionState.Degraded); + } + + [Test] + public async Task TransientFailures_ThenGracefulStop_EndsStopped_NotFaulted() + { + var svc = new ScriptedRelayService(OptionsWith(FastResilience(maxFailures: 5))); + + // First two attempts fail (transient); the third blocks until cancelled, then + // returns by honoring the token ⇒ graceful Stopped, never reaching the breaker. + svc.ExecuteBehavior = async (s, token) => + { + if (s.Attempts < 3) + throw new InvalidOperationException("transient"); + await Task.Delay(Timeout.Infinite, token).ConfigureAwait(false); + }; + + var run = svc.StartAsync(CancellationToken.None); + + // Wait until the run is parked in the long-lived (3rd) attempt, then stop it. + var spun = SpinWait.SpinUntil(() => svc.Attempts >= 3 && svc.State == RelayServiceState.Running, TimeSpan.FromSeconds(5)); + spun.Should().BeTrue("the service should reach its healthy long-lived run after the transient failures"); + + await svc.StopAsync(); + await run; + + svc.State.Should().Be(RelayServiceState.Stopped); + svc.Attempts.Should().Be(3); + } + + [Test] + public async Task ResilienceDisabled_RunsExecuteOnce_AndPropagatesFault() + { + var resilience = FastResilience(maxFailures: 3); + resilience.Enabled = false; + var svc = new ScriptedRelayService(OptionsWith(resilience)); + + svc.ExecuteBehavior = (_, __) => throw new InvalidOperationException("boom"); + + await svc.StartAsync(CancellationToken.None); + + svc.State.Should().Be(RelayServiceState.Faulted); + svc.Attempts.Should().Be(1, "with resilience disabled ExecuteAsync runs exactly once"); + } + + /// Minimal LiveKit-mode service whose run is scripted by the test. + private sealed class ScriptedRelayService : RelayServiceBase + { + private int _attempts; + + public ScriptedRelayService(RelayHostOptions options) + : base("test", options, null) + { + } + + protected override bool IsLiveKitMode => true; + + public int Attempts => Volatile.Read(ref _attempts); + + public Func ExecuteBehavior { get; set; } + + protected override async Task ExecuteAsync(CancellationToken token) + { + Interlocked.Increment(ref _attempts); + await ExecuteBehavior(this, token).ConfigureAwait(false); + } + } + } +} diff --git a/Resgrid.Audio.Voice.Tests/Resgrid.Audio.Voice.Tests.csproj b/Resgrid.Audio.Voice.Tests/Resgrid.Audio.Voice.Tests.csproj index 744f76a..792f419 100644 --- a/Resgrid.Audio.Voice.Tests/Resgrid.Audio.Voice.Tests.csproj +++ b/Resgrid.Audio.Voice.Tests/Resgrid.Audio.Voice.Tests.csproj @@ -16,5 +16,6 @@ + diff --git a/Resgrid.Relay.Engine/Configuration/RelayHostOptions.cs b/Resgrid.Relay.Engine/Configuration/RelayHostOptions.cs index 8dafc61..5337a9e 100644 --- a/Resgrid.Relay.Engine/Configuration/RelayHostOptions.cs +++ b/Resgrid.Relay.Engine/Configuration/RelayHostOptions.cs @@ -17,6 +17,24 @@ public sealed class RelayHostOptions public RecorderModeOptions Recorder { get; set; } = new RecorderModeOptions(); public DispatchVoiceOptions DispatchVoice { get; set; } = new DispatchVoiceOptions(); public TtsSettings Tts { get; set; } = new TtsSettings(); + + // ─── Fault tolerance for the LiveKit voice modes (retry + circuit-breaker) ─── + public ResilienceOptions Resilience { get; set; } = new ResilienceOptions(); + } + + /// + /// Retry/back-off and circuit-breaker tuning for the long-lived LiveKit voice modes + /// (radio / record / dispatch). A run that faults is restarted with exponential + /// back-off; the breaker opens (faulting the service) once failures pile up without a + /// healthy run in between. + /// + public sealed class ResilienceOptions + { + public bool Enabled { get; set; } = true; + public int MaxConsecutiveFailures { get; set; } = 5; // circuit-breaker trips after this many consecutive quick failures + public double InitialBackoffSeconds { get; set; } = 2; + public double MaxBackoffSeconds { get; set; } = 60; + public double HealthyRunSeconds { get; set; } = 30; // a run lasting >= this resets the consecutive-failure counter } public sealed class RelayTelemetryOptions diff --git a/Resgrid.Relay.Engine/Properties/AssemblyInfo.cs b/Resgrid.Relay.Engine/Properties/AssemblyInfo.cs new file mode 100644 index 0000000..7f743e7 --- /dev/null +++ b/Resgrid.Relay.Engine/Properties/AssemblyInfo.cs @@ -0,0 +1,4 @@ +using System.Runtime.CompilerServices; + +// Exposes engine internals (e.g. RelayServiceBase.ComputeBackoff) to the resilience tests. +[assembly: InternalsVisibleTo("Resgrid.Audio.Voice.Tests")] diff --git a/Resgrid.Relay.Engine/Services/DispatchRelayService.cs b/Resgrid.Relay.Engine/Services/DispatchRelayService.cs index 9e2134f..0be1784 100644 --- a/Resgrid.Relay.Engine/Services/DispatchRelayService.cs +++ b/Resgrid.Relay.Engine/Services/DispatchRelayService.cs @@ -17,6 +17,8 @@ public DispatchRelayService(RelayHostOptions options, ILogger logger) { } + protected override bool IsLiveKitMode => true; + protected override async Task ExecuteAsync(CancellationToken token) { MutableStatus.LiveKit = ConnectionState.Connecting; diff --git a/Resgrid.Relay.Engine/Services/RadioRelayService.cs b/Resgrid.Relay.Engine/Services/RadioRelayService.cs index 8af27e9..d4b25b2 100644 --- a/Resgrid.Relay.Engine/Services/RadioRelayService.cs +++ b/Resgrid.Relay.Engine/Services/RadioRelayService.cs @@ -18,6 +18,8 @@ public RadioRelayService(RelayHostOptions options, ILogger logger) { } + protected override bool IsLiveKitMode => true; + protected override async Task ExecuteAsync(CancellationToken token) { MutableStatus.LiveKit = ConnectionState.Connecting; diff --git a/Resgrid.Relay.Engine/Services/RecordRelayService.cs b/Resgrid.Relay.Engine/Services/RecordRelayService.cs index 59fbc02..3f95687 100644 --- a/Resgrid.Relay.Engine/Services/RecordRelayService.cs +++ b/Resgrid.Relay.Engine/Services/RecordRelayService.cs @@ -17,6 +17,8 @@ public RecordRelayService(RelayHostOptions options, ILogger logger) { } + protected override bool IsLiveKitMode => true; + protected override async Task ExecuteAsync(CancellationToken token) { MutableStatus.LiveKit = ConnectionState.Connecting; diff --git a/Resgrid.Relay.Engine/Services/RelayServiceBase.cs b/Resgrid.Relay.Engine/Services/RelayServiceBase.cs index 2b32620..df0c07e 100644 --- a/Resgrid.Relay.Engine/Services/RelayServiceBase.cs +++ b/Resgrid.Relay.Engine/Services/RelayServiceBase.cs @@ -2,6 +2,7 @@ using System.Threading; using System.Threading.Tasks; using Resgrid.Relay.Engine.Configuration; +using Resgrid.Relay.Engine.Telemetry; using Serilog; namespace Resgrid.Relay.Engine.Services @@ -15,6 +16,7 @@ namespace Resgrid.Relay.Engine.Services public abstract class RelayServiceBase : IRelayService { private readonly object _sync = new object(); + private readonly IRelayModeTelemetry _telemetry; private CancellationTokenSource _cts; private RelayServiceState _state = RelayServiceState.Stopped; @@ -23,8 +25,18 @@ protected RelayServiceBase(string mode, RelayHostOptions options, ILogger logger Mode = mode ?? throw new ArgumentNullException(nameof(mode)); Options = options ?? throw new ArgumentNullException(nameof(options)); Logger = logger; + // Only the LiveKit voice modes wrap their run in the retry/circuit-breaker + // loop and report to Sentry; everything else uses the cheap no-op telemetry. + _telemetry = IsLiveKitMode ? RelayModeTelemetry.Create(options.Telemetry, logger) : NullRelayModeTelemetry.Instance; } + /// + /// Whether this mode is a long-lived LiveKit voice mode (radio / record / dispatch) + /// that should be wrapped in the resilience (retry + circuit-breaker) loop and + /// reported to Sentry. Non-LiveKit modes (e.g. SMTP) leave this false. + /// + protected virtual bool IsLiveKitMode => false; + public string Mode { get; } public RelayServiceState State => _state; public event EventHandler StateChanged; @@ -78,7 +90,8 @@ public async Task StartAsync(CancellationToken token) // Starting during the startup window (it sets Stopping + cancels _cts under _sync). // Atomic, so a stop that wins the race keeps Stopping/Stopped instead of reverting. TryTransition(RelayServiceState.Starting, RelayServiceState.Running); - await ExecuteAsync(_cts.Token).ConfigureAwait(false); + await RunWithResilienceAsync(_cts.Token).ConfigureAwait(false); + _telemetry.ModeStopped(Mode); TransitionTo(RelayServiceState.Stopped); } catch (OperationCanceledException) when (_cts.IsCancellationRequested) @@ -86,6 +99,7 @@ public async Task StartAsync(CancellationToken token) // Graceful stop: only when OUR shutdown token was actually requested. A cancellation // from elsewhere (e.g. a dependency/HttpClient timeout) is a real fault and flows to // the catch below so Program surfaces a failure exit code. + _telemetry.ModeStopped(Mode); TransitionTo(RelayServiceState.Stopped); } catch (Exception ex) @@ -93,6 +107,7 @@ public async Task StartAsync(CancellationToken token) // IRelayService contract: StartAsync returns on fault — surface it via // State/StateChanged rather than throwing back to the caller. Logger?.Error(ex, "Relay mode '{Mode}' faulted", Mode); + _telemetry.ModeFaulted(Mode, ex); TransitionTo(RelayServiceState.Faulted, ex.Message); } } @@ -124,7 +139,73 @@ public virtual async ValueTask DisposeAsync() try { _cts?.Cancel(); } catch (ObjectDisposedException) { } _cts?.Dispose(); - await Task.CompletedTask.ConfigureAwait(false); + if (_telemetry != null) + await _telemetry.DisposeAsync().ConfigureAwait(false); + } + + /// + /// Runs , wrapping the LiveKit voice modes in a + /// retry-with-back-off loop guarded by a simple circuit breaker. A run that faults + /// is restarted after an exponential, jittered delay; the breaker opens (rethrowing + /// so faults the service) once + /// failures occur without a healthy run resetting the counter. Non-LiveKit modes, + /// and any mode with resilience disabled, run once directly. + /// + private async Task RunWithResilienceAsync(CancellationToken token) + { + var r = Options.Resilience ?? new ResilienceOptions(); + if (!IsLiveKitMode || !r.Enabled) + { + await ExecuteAsync(token).ConfigureAwait(false); + return; + } + + _telemetry.ModeStarting(Mode); + var consecutiveFailures = 0; + while (true) + { + token.ThrowIfCancellationRequested(); + var startTs = System.Diagnostics.Stopwatch.GetTimestamp(); + try + { + await ExecuteAsync(token).ConfigureAwait(false); + return; + } + catch (OperationCanceledException) when (token.IsCancellationRequested) + { + throw; + } + catch (Exception ex) + { + var ran = System.Diagnostics.Stopwatch.GetElapsedTime(startTs).TotalSeconds; + if (ran >= r.HealthyRunSeconds) + consecutiveFailures = 0; // it had been healthy ⇒ fresh transient failure + consecutiveFailures++; + MutableStatus.LiveKit = ConnectionState.Degraded; + if (consecutiveFailures >= r.MaxConsecutiveFailures) + { + // Circuit open ⇒ stop retrying and let the outer catch fault the service. + Logger?.Warning(ex, "Relay mode '{Mode}' circuit breaker open after {N} consecutive failures; faulting", Mode, consecutiveFailures); + throw; + } + var delay = ComputeBackoff(r, consecutiveFailures); + _telemetry.ModeRetrying(Mode, ex, consecutiveFailures, delay); + Logger?.Warning(ex, "Relay mode '{Mode}' failed (failure {N}/{Max}); reconnecting in {Delay:n1}s", Mode, consecutiveFailures, r.MaxConsecutiveFailures, delay.TotalSeconds); + await Task.Delay(delay, token).ConfigureAwait(false); + } + } + } + + /// + /// Exponential back-off (doubling from , + /// capped at ) with ±20% jitter, floored at + /// 0.5s so retries never busy-spin. + /// + internal static TimeSpan ComputeBackoff(ResilienceOptions r, int failures) + { + var baseSecs = Math.Min(r.InitialBackoffSeconds * Math.Pow(2, failures - 1), r.MaxBackoffSeconds); + var jitter = baseSecs * 0.2 * (Random.Shared.NextDouble() * 2 - 1); + return TimeSpan.FromSeconds(Math.Max(0.5, baseSecs + jitter)); } private void TransitionTo(RelayServiceState next, string error = null) diff --git a/Resgrid.Relay.Engine/Telemetry/IRelayModeTelemetry.cs b/Resgrid.Relay.Engine/Telemetry/IRelayModeTelemetry.cs new file mode 100644 index 0000000..1e600c3 --- /dev/null +++ b/Resgrid.Relay.Engine/Telemetry/IRelayModeTelemetry.cs @@ -0,0 +1,19 @@ +using System; +using System.Threading.Tasks; + +namespace Resgrid.Relay.Engine.Telemetry +{ + /// + /// Lightweight observability hooks for the long-lived LiveKit voice modes + /// (radio / record / dispatch). Implementations forward lifecycle and + /// fault/retry events to a backend (e.g. Sentry) — or do nothing when + /// monitoring is disabled. + /// + public interface IRelayModeTelemetry : IAsyncDisposable + { + void ModeStarting(string mode); + void ModeRetrying(string mode, Exception ex, int attempt, TimeSpan nextDelay); + void ModeFaulted(string mode, Exception ex); + void ModeStopped(string mode); + } +} diff --git a/Resgrid.Relay.Engine/Telemetry/NullRelayModeTelemetry.cs b/Resgrid.Relay.Engine/Telemetry/NullRelayModeTelemetry.cs new file mode 100644 index 0000000..810b5ad --- /dev/null +++ b/Resgrid.Relay.Engine/Telemetry/NullRelayModeTelemetry.cs @@ -0,0 +1,37 @@ +using System; +using System.Threading.Tasks; + +namespace Resgrid.Relay.Engine.Telemetry +{ + /// + /// No-op used when mode monitoring is disabled + /// (non-LiveKit modes, or LiveKit modes with no Sentry DSN configured). Every call + /// is a cheap nothing so callers never need to null-check the telemetry instance. + /// + public sealed class NullRelayModeTelemetry : IRelayModeTelemetry + { + public static readonly IRelayModeTelemetry Instance = new NullRelayModeTelemetry(); + + private NullRelayModeTelemetry() + { + } + + public void ModeStarting(string mode) + { + } + + public void ModeRetrying(string mode, Exception ex, int attempt, TimeSpan nextDelay) + { + } + + public void ModeFaulted(string mode, Exception ex) + { + } + + public void ModeStopped(string mode) + { + } + + public ValueTask DisposeAsync() => ValueTask.CompletedTask; + } +} diff --git a/Resgrid.Relay.Engine/Telemetry/RelayModeTelemetry.cs b/Resgrid.Relay.Engine/Telemetry/RelayModeTelemetry.cs new file mode 100644 index 0000000..34327ea --- /dev/null +++ b/Resgrid.Relay.Engine/Telemetry/RelayModeTelemetry.cs @@ -0,0 +1,18 @@ +using Resgrid.Relay.Engine.Configuration; +using Serilog; + +namespace Resgrid.Relay.Engine.Telemetry +{ + /// + /// Factory that picks the right for the current + /// configuration: a Sentry-backed reporter when a DSN is present, otherwise the + /// shared no-op singleton. + /// + public static class RelayModeTelemetry + { + public static IRelayModeTelemetry Create(RelayTelemetryOptions telemetry, ILogger logger) => + string.IsNullOrWhiteSpace(telemetry?.Sentry?.Dsn) + ? NullRelayModeTelemetry.Instance + : SentryRelayModeTelemetry.Acquire(telemetry.Sentry, telemetry.Environment, logger); + } +} diff --git a/Resgrid.Relay.Engine/Telemetry/SentryRelayModeTelemetry.cs b/Resgrid.Relay.Engine/Telemetry/SentryRelayModeTelemetry.cs new file mode 100644 index 0000000..9c56206 --- /dev/null +++ b/Resgrid.Relay.Engine/Telemetry/SentryRelayModeTelemetry.cs @@ -0,0 +1,163 @@ +using System; +using System.Collections.Generic; +using System.Threading.Tasks; +using Resgrid.Relay.Engine.Configuration; +using Sentry; +using Serilog; + +namespace Resgrid.Relay.Engine.Telemetry +{ + /// + /// Sends LiveKit voice-mode lifecycle/fault events to Sentry. Mirrors the init, + /// flush, and per-call try/catch guarding of . When + /// the DSN is blank the SDK is never initialized and every method is a no-op, so an + /// empty configuration costs nothing. + /// + public sealed class SentryRelayModeTelemetry : IRelayModeTelemetry + { + // SentrySdk.Init initializes a process-global hub and the returned IDisposable closes it, + // so every LiveKit service shares ONE telemetry instance and ONE SDK handle. The shared + // instance is reference-counted: the SDK is initialized on the first service and only + // flushed/closed when the last service is disposed — otherwise the first service to stop + // would tear Sentry down for the others still running. + private static readonly object _sharedLock = new object(); + private static SentryRelayModeTelemetry _shared; + private static int _refCount; + + private readonly ILogger _logger; + private readonly IDisposable _sentrySdk; + + private SentryRelayModeTelemetry(SentryTelemetryOptions sentry, string environment, ILogger logger) + { + _logger = logger; + + if (!string.IsNullOrWhiteSpace(sentry?.Dsn)) + { + _sentrySdk = SentrySdk.Init(sentryOptions => + { + sentryOptions.Dsn = sentry.Dsn.Trim(); + sentryOptions.Environment = environment; + sentryOptions.Release = sentry.Release; + sentryOptions.ServerName = Environment.MachineName; + sentryOptions.AttachStacktrace = true; + sentryOptions.SendDefaultPii = sentry.SendDefaultPii; + sentryOptions.ShutdownTimeout = TimeSpan.FromSeconds(2); + }); + } + + _logger?.Information( + "Relay mode observability initialized. SentryEnabled={SentryEnabled} Environment={Environment}", + _sentrySdk != null, + environment); + } + + /// + /// Returns the process-wide shared telemetry, initializing the Sentry SDK on first use. + /// Each call increments a reference count that releases, so the + /// SDK stays alive until every LiveKit service sharing it has been disposed. + /// + public static IRelayModeTelemetry Acquire(SentryTelemetryOptions sentry, string environment, ILogger logger) + { + lock (_sharedLock) + { + _shared ??= new SentryRelayModeTelemetry(sentry, environment, logger); + _refCount++; + return _shared; + } + } + + public void ModeStarting(string mode) + { + if (_sentrySdk == null) + return; + + try + { + SentrySdk.AddBreadcrumb($"mode {mode} starting", category: "relay.mode"); + } + catch (Exception sentryException) + { + _logger?.Warning(sentryException, "Failed sending relay mode breadcrumb to Sentry"); + } + } + + public void ModeStopped(string mode) + { + if (_sentrySdk == null) + return; + + try + { + SentrySdk.AddBreadcrumb($"mode {mode} stopped", category: "relay.mode"); + } + catch (Exception sentryException) + { + _logger?.Warning(sentryException, "Failed sending relay mode breadcrumb to Sentry"); + } + } + + public void ModeRetrying(string mode, Exception ex, int attempt, TimeSpan nextDelay) + { + if (_sentrySdk == null) + return; + + try + { + SentrySdk.AddBreadcrumb( + $"mode {mode} retrying", + category: "relay.mode", + level: BreadcrumbLevel.Warning, + data: new Dictionary + { + ["attempt"] = attempt.ToString(), + ["next_delay_seconds"] = nextDelay.TotalSeconds.ToString("n2"), + ["exception"] = ex?.Message ?? string.Empty + }); + } + catch (Exception sentryException) + { + _logger?.Warning(sentryException, "Failed sending relay mode breadcrumb to Sentry"); + } + } + + public void ModeFaulted(string mode, Exception ex) + { + if (_sentrySdk == null || ex == null) + return; + + try + { + SentrySdk.CaptureException(ex, scope => + { + scope.SetTag("relay.mode", mode); + scope.SetTag("relay.scope", "mode_fault"); + }); + } + catch (Exception sentryException) + { + _logger?.Warning(sentryException, "Failed sending relay mode exception to Sentry"); + } + } + + public async ValueTask DisposeAsync() + { + IDisposable sdk; + lock (_sharedLock) + { + // Release this service's reference; only the last one out flushes and closes the + // shared SDK so any services still running keep reporting to Sentry. + if (_refCount == 0 || --_refCount > 0) + return; + + sdk = _sentrySdk; + _shared = null; + } + + if (sdk != null) + { + await SentrySdk.FlushAsync(TimeSpan.FromSeconds(2)).ConfigureAwait(false); + sdk.Dispose(); + } + } + } +} diff --git a/Resgrid.Relay.Engine/Voice/DispatchVoiceMode.cs b/Resgrid.Relay.Engine/Voice/DispatchVoiceMode.cs index 36216e6..045f68c 100644 --- a/Resgrid.Relay.Engine/Voice/DispatchVoiceMode.cs +++ b/Resgrid.Relay.Engine/Voice/DispatchVoiceMode.cs @@ -6,6 +6,7 @@ using Resgrid.Relay.Engine; using Resgrid.Relay.Engine.Configuration; using Resgrid.Audio.Voice; +using Resgrid.Audio.Voice.Abstractions; using Resgrid.Audio.Voice.Connection; using Resgrid.Audio.Voice.LiveKit; using Resgrid.Audio.Voice.ToneOut; @@ -52,6 +53,34 @@ public static async Task RunAsync(RelayHostOptions options, ILogger logger, if (status != null) status.LiveKit = ConnectionState.Connected; + // Signals a hard LiveKit disconnect so the run throws and the resilience layer + // rejoins; a transient SDK "reconnecting" only degrades status. + var disconnect = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + EventHandler onConnectionChanged = (_, change) => + { + if (change.Connected) + { + // A healthy (re)connected state clears a prior "reconnecting" degrade so the + // pill recovers immediately instead of waiting for the next successful publish. + if (status != null) + status.LiveKit = ConnectionState.Connected; + return; + } + if (string.Equals(change.Reason, "reconnecting", StringComparison.OrdinalIgnoreCase)) + { + if (status != null) + status.LiveKit = ConnectionState.Degraded; + return; + } + disconnect.TrySetResult(); + }; + session.ConnectionChanged += onConnectionChanged; + + // Enter the try BEFORE the remaining setup (TTS client, backlog prime) so the finally + // below always detaches the handler and disposes the publisher — a transient API failure + // during startup must not leak the ConnectionChanged subscription or the publisher. + try + { using var tts = new ResgridTtsClient(options.Tts, logger); // TTS reachability is unverified until the first synthesis actually reaches the service: @@ -73,6 +102,10 @@ public static async Task RunAsync(RelayHostOptions options, ILogger logger, while (!cancellationToken.IsCancellationRequested) { + // A hard LiveKit disconnect ends the poll loop so resilience reconnects. + if (disconnect.Task.IsCompleted) + throw new InvalidOperationException("LiveKit dispatch session disconnected; reconnecting"); + try { var calls = await callsApi.GetActiveCallsAsync(deptId, cancellationToken).ConfigureAwait(false); @@ -145,11 +178,17 @@ public static async Task RunAsync(RelayHostOptions options, ILogger logger, logger.Error(ex, "Dispatch tone-out poll failed"); } - try { await Task.Delay(TimeSpan.FromSeconds(pollSeconds), cancellationToken).ConfigureAwait(false); } + // Wake promptly on a hard disconnect instead of sleeping out the full poll interval. + try { await Task.WhenAny(Task.Delay(TimeSpan.FromSeconds(pollSeconds), cancellationToken), disconnect.Task).ConfigureAwait(false); } catch (TaskCanceledException) { break; } } + } + finally + { + session.ConnectionChanged -= onConnectionChanged; + await publisher.DisposeAsync().ConfigureAwait(false); + } - await publisher.DisposeAsync().ConfigureAwait(false); return 0; } diff --git a/Resgrid.Relay.Engine/Voice/RadioMode.cs b/Resgrid.Relay.Engine/Voice/RadioMode.cs index 70f3e7e..09603e5 100644 --- a/Resgrid.Relay.Engine/Voice/RadioMode.cs +++ b/Resgrid.Relay.Engine/Voice/RadioMode.cs @@ -78,29 +78,66 @@ public static async Task RunAsync(RelayHostOptions options, ILogger logger, status.Receiving = bridge.Receiving; }; - await bridge.StartAsync(session, cancellationToken).ConfigureAwait(false); + // A hard LiveKit disconnect cancels the linked token so the bridge unwinds; we then + // rethrow (below) so the resilience layer rejoins. A transient SDK "reconnecting" + // only degrades status and is left to auto-recover. + using var disconnectCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + var disconnected = false; + EventHandler onConnectionChanged = (_, change) => + { + if (change.Connected) + { + // A successful (re)connect clears a prior "reconnecting" degrade so the pill + // recovers to Connected instead of staying stuck at Degraded after auto-recovery. + if (status != null) + status.LiveKit = ConnectionState.Connected; + return; + } + if (string.Equals(change.Reason, "reconnecting", StringComparison.OrdinalIgnoreCase)) + { + if (status != null) + status.LiveKit = ConnectionState.Degraded; + return; + } + disconnected = true; + try { disconnectCts.Cancel(); } + catch (ObjectDisposedException) { } + }; + session.ConnectionChanged += onConnectionChanged; TransmissionRecorder recorder = null; List recorderDisposables = null; ITransmissionLog recorderLog = null; - if (options.Radio.RecordWhileBridging) + try { - var (stores, disposables) = RecordMode.BuildStores(options.Recorder, logger); - recorderDisposables = disposables; - recorderLog = RecordMode.BuildLog(options.Recorder, logger); - recorder = new TransmissionRecorder(session, options.Recorder.Segmentation, stores, recorderLog, logger); - recorder.Start(); - } + await bridge.StartAsync(session, disconnectCts.Token).ConfigureAwait(false); - logger.Information($"Radio bridge running on '{channel.Name}'. Press Ctrl+C to stop."); - await VoiceModeRuntime.WaitForCancellationAsync(cancellationToken).ConfigureAwait(false); + if (options.Radio.RecordWhileBridging) + { + var (stores, disposables) = RecordMode.BuildStores(options.Recorder, logger); + recorderDisposables = disposables; + recorderLog = RecordMode.BuildLog(options.Recorder, logger); + recorder = new TransmissionRecorder(session, options.Recorder.Segmentation, stores, recorderLog, logger); + recorder.Start(); + } - if (recorder != null) - await recorder.DisposeAsync().ConfigureAwait(false); - if (recorderLog != null) - await recorderLog.DisposeAsync().ConfigureAwait(false); - if (recorderDisposables != null) - foreach (var d in recorderDisposables) d.Dispose(); + logger.Information($"Radio bridge running on '{channel.Name}'. Press Ctrl+C to stop."); + await VoiceModeRuntime.WaitForCancellationAsync(disconnectCts.Token).ConfigureAwait(false); + + // Woken by a hard disconnect (not the user shutdown) ⇒ fault so resilience rejoins. + if (disconnected && !cancellationToken.IsCancellationRequested) + throw new InvalidOperationException("LiveKit radio session disconnected; reconnecting"); + } + finally + { + session.ConnectionChanged -= onConnectionChanged; + if (recorder != null) + await recorder.DisposeAsync().ConfigureAwait(false); + if (recorderLog != null) + await recorderLog.DisposeAsync().ConfigureAwait(false); + if (recorderDisposables != null) + foreach (var d in recorderDisposables) d.Dispose(); + } // bridge is disposed by its `await using` scope on method exit — no explicit call. return 0; diff --git a/Resgrid.Relay.Engine/Voice/RecordMode.cs b/Resgrid.Relay.Engine/Voice/RecordMode.cs index abd413f..1e30900 100644 --- a/Resgrid.Relay.Engine/Voice/RecordMode.cs +++ b/Resgrid.Relay.Engine/Voice/RecordMode.cs @@ -42,11 +42,55 @@ public static async Task RunAsync(RelayHostOptions options, ILogger logger, var log = BuildLog(options.Recorder, logger); var recorders = new List(); + // Signals a hard LiveKit disconnect (not a transient SDK reconnect) so the run + // throws and the resilience layer rejoins. Carries the disconnect reason. + var disconnect = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var sessionHandlers = new List<(IVoiceRoomSession session, EventHandler handler)>(); + // The single LiveKit status pill is shared across every recorded channel, so track which + // channels are mid-reconnect and surface the worst state: stay Degraded while ANY channel + // is reconnecting and only return to Connected once they are all back. Connection events + // fire on SDK threads, so guard the set and the status write with a lock. + var degradedChannels = new HashSet(); + var statusLock = new object(); + try { foreach (var channel in channels) { var session = await manager.JoinAsync(channel, cancellationToken).ConfigureAwait(false); + EventHandler handler = (_, change) => + { + if (change.Connected) + { + // This channel recovered; only report the shared link healthy once EVERY + // channel is back — a sibling still reconnecting keeps the pill Degraded. + if (status != null) + lock (statusLock) + { + degradedChannels.Remove(session); + status.LiveKit = degradedChannels.Count == 0 + ? ConnectionState.Connected + : ConnectionState.Degraded; + } + return; + } + // "reconnecting" is the SDK auto-recovering — degrade but keep running. + if (string.Equals(change.Reason, "reconnecting", StringComparison.OrdinalIgnoreCase)) + { + if (status != null) + lock (statusLock) + { + degradedChannels.Add(session); + status.LiveKit = ConnectionState.Degraded; + } + return; + } + // Any other Connected=false is a hard disconnect — restart the recorder. + disconnect.TrySetResult(change.Reason); + }; + session.ConnectionChanged += handler; + sessionHandlers.Add((session, handler)); + var recorder = new TransmissionRecorder(session, options.Recorder.Segmentation, stores, log, logger); if (status != null) recorder.TransmissionRecorded += (_, __) => status.IncrementTransmissionsRecorded(); @@ -54,15 +98,26 @@ public static async Task RunAsync(RelayHostOptions options, ILogger logger, recorders.Add(recorder); } - // All requested channels joined — LiveKit is up. + // All requested channels joined — LiveKit is up, unless a channel already began + // reconnecting during the join loop, in which case keep the shared pill Degraded. if (status != null) - status.LiveKit = ConnectionState.Connected; + lock (statusLock) + status.LiveKit = degradedChannels.Count == 0 + ? ConnectionState.Connected + : ConnectionState.Degraded; logger.Information($"Recording {channels.Count} channel(s) to {options.Recorder.Store}. Press Ctrl+C to stop."); - await VoiceModeRuntime.WaitForCancellationAsync(cancellationToken).ConfigureAwait(false); + var completed = await Task.WhenAny(VoiceModeRuntime.WaitForCancellationAsync(cancellationToken), disconnect.Task).ConfigureAwait(false); + // Only fault on a real disconnect, not when a Ctrl+C/SIGTERM shutdown raced the + // teardown's own disconnect event — a requested shutdown must complete cleanly. + if (completed == disconnect.Task && !cancellationToken.IsCancellationRequested) + throw new InvalidOperationException($"LiveKit session disconnected ({disconnect.Task.Result}); restarting recorder"); } finally { + // Detach connection handlers first so a teardown-time disconnect can't fire. + foreach (var (session, handler) in sessionHandlers) + session.ConnectionChanged -= handler; // Always tear down so a mid-loop JoinAsync failure or cancellation does not // leak already-created recorders, the metadata log, or disposable stores. foreach (var recorder in recorders) diff --git a/deploy/systemd/README.md b/deploy/systemd/README.md new file mode 100644 index 0000000..75bbedc --- /dev/null +++ b/deploy/systemd/README.md @@ -0,0 +1,98 @@ +# Running Resgrid Relay as a Linux service (systemd) + +The **CLI** (`Resgrid.Audio.Relay.Console`) is cross-platform and runs on Linux for the +LiveKit / backend modes. The **desktop GUI** (`Resgrid.Audio.Relay`, WPF) is +**Windows-only** and is not part of this deployment — use the CLI for headless Linux hosts. + +## Modes available on Linux + +| Mode | Linux | What it does | +|------------|:-----:|----------------------------------------------------------| +| `smtp` | ✅ | SMTP dispatch-email relay | +| `record` | ✅ | Save LiveKit PTT-channel transmissions to disk / S3 + log | +| `dispatch` | ✅ | Tone out new calls (tones + Resgrid TTS) to a PTT channel | +| `audio` | ❌ | Windows tone-detect importer (needs a local sound card) | +| `radio` | ❌ | Radio ↔ LiveKit bridge (needs sound card + serial/HID PTT)| + +The Windows-only modes resolve to a "not supported on this platform" service on Linux +(they fault with a clear message rather than crashing). The LiveKit native library +(`liblivekit_ffi.so`, linux-x64 / linux-arm64) ships inside the `Livekit.Rtc.Dotnet` +NuGet package and is included automatically in the published output; its only OS +prerequisite is `libstdc++6`, present on any standard distribution. + +> One service instance runs **one** mode (set by `RESGRID__RELAY__Mode`). To run several +> modes (e.g. `record` **and** `dispatch`), install one unit per mode — see *Multiple modes*. + +## 1. Prerequisites + +- **.NET 10 runtime** (`dotnet`) on the host — or publish self-contained to skip it. +- **`libstdc++6`** — Debian/Ubuntu: `sudo apt-get install -y libstdc++6`. + +## 2. Publish + +Framework-dependent (host needs the .NET 10 runtime): + +```bash +dotnet publish Resgrid.Audio.Relay.Console/Resgrid.Audio.Relay.Console.csproj \ + -c Release -f net10.0 -o ./publish +``` + +Self-contained (no runtime needed on the host) — choose your architecture: + +```bash +dotnet publish Resgrid.Audio.Relay.Console/Resgrid.Audio.Relay.Console.csproj \ + -c Release -f net10.0 -r linux-x64 --self-contained true -o ./publish # or linux-arm64 +``` + +`net10.0` (not `net10.0-windows`) is the cross-platform target; it excludes the Windows +audio stack (NAudio / DtmfDetection) and the WPF GUI. + +## 3. Install + +```bash +# Service account + directories +sudo useradd --system --no-create-home --shell /usr/sbin/nologin resgrid +sudo mkdir -p /opt/resgrid-relay /etc/resgrid-relay + +# App +sudo cp -r ./publish/* /opt/resgrid-relay/ +sudo chown -R resgrid:resgrid /opt/resgrid-relay + +# Config (secrets) — chmod 600 +sudo cp deploy/systemd/relay.env.example /etc/resgrid-relay/relay.env +sudo chmod 600 /etc/resgrid-relay/relay.env +sudo "$EDITOR" /etc/resgrid-relay/relay.env # set Mode + Resgrid creds + mode settings + +# Unit +sudo cp deploy/systemd/resgrid-relay.service /etc/systemd/system/ +``` + +If you published **self-contained**, change `ExecStart` in the unit to run the binary +directly: `ExecStart=/opt/resgrid-relay/Resgrid.Audio.Relay.Console run`. Otherwise confirm +the `dotnet` path (`which dotnet`); the unit assumes `/usr/bin/dotnet`. + +## 4. Enable + run + +```bash +sudo systemctl daemon-reload +sudo systemctl enable --now resgrid-relay +journalctl -u resgrid-relay -f +``` + +## Notes + +- **Graceful stop** — the CLI traps `SIGTERM` (systemd's default stop signal) and `SIGINT`, + cancelling in-flight work so recordings flush and LiveKit disconnects cleanly; + `TimeoutStopSec=30` bounds the drain before systemd escalates to `SIGKILL`. +- **Configuration** — everything is `RESGRID__RELAY__*` environment variables. Run + `dotnet Resgrid.Audio.Relay.Console.dll help` for the complete, authoritative list. +- **Writable state** — the unit grants `/var/lib/resgrid-relay` (via `StateDirectory`). + Point `Recorder__LocalPath` and/or `Resgrid__TokenCachePath` there. If you write + recordings elsewhere, add that path to `ReadWritePaths=` in the unit (it runs under + `ProtectSystem=strict`). +- **Multiple modes** — copy the unit per mode with its own env file, e.g. + `resgrid-relay-record.service` (`EnvironmentFile=/etc/resgrid-relay/record.env`) and + `resgrid-relay-dispatch.service` (`EnvironmentFile=/etc/resgrid-relay/dispatch.env`), + then `enable --now` each. +- **Containers** — a `Dockerfile` already exists at the repo root for the same CLI if you + prefer running it as a container instead of a host service. diff --git a/deploy/systemd/relay.env.example b/deploy/systemd/relay.env.example new file mode 100644 index 0000000..ca8e1e0 --- /dev/null +++ b/deploy/systemd/relay.env.example @@ -0,0 +1,63 @@ +# Resgrid Relay — service configuration. +# Copy to /etc/resgrid-relay/relay.env, fill in your values, then: +# sudo chmod 600 /etc/resgrid-relay/relay.env (this file holds secrets) +# Full authoritative reference: +# dotnet Resgrid.Audio.Relay.Console.dll help +# +# Linux-supported modes: smtp | record | dispatch +# (audio + radio require Windows — they need local sound-card capture / serial PTT.) + +RESGRID__RELAY__Mode=dispatch + +# ── Resgrid API (required for every mode) ──────────────────────────────────── +RESGRID__RELAY__Resgrid__BaseUrl=https://api.resgrid.com +RESGRID__RELAY__Resgrid__ApiVersion=v4 +RESGRID__RELAY__Resgrid__ClientId=your-oidc-client-id +RESGRID__RELAY__Resgrid__ClientSecret=your-oidc-client-secret +# GrantType: RefreshToken | ClientCredentials | SystemApiKey +RESGRID__RELAY__Resgrid__GrantType=SystemApiKey +RESGRID__RELAY__Resgrid__SystemApiKey=your-system-api-key +RESGRID__RELAY__Resgrid__DepartmentId=your-department-id +# --- RefreshToken grant instead of SystemApiKey: --- +# RESGRID__RELAY__Resgrid__GrantType=RefreshToken +# RESGRID__RELAY__Resgrid__RefreshToken=your-refresh-token +# RESGRID__RELAY__Resgrid__TokenCachePath=/var/lib/resgrid-relay/token.cache + +# ── Voice channel (record / dispatch modes) ────────────────────────────────── +RESGRID__RELAY__Voice__Channel=default +RESGRID__RELAY__Voice__DepartmentId=your-department-id + +# ── dispatch mode: tone out new calls with tones + Resgrid TTS ─────────────── +RESGRID__RELAY__Tts__ServiceBaseUrl=https://tts.resgrid.com +RESGRID__RELAY__DispatchVoice__Channel=default +RESGRID__RELAY__DispatchVoice__PollSeconds=15 + +# ── record mode: save PTT-channel transmissions ────────────────────────────── +# RESGRID__RELAY__Recorder__Channel=all +# RESGRID__RELAY__Recorder__Store=local # local | s3 | both +# RESGRID__RELAY__Recorder__LocalPath=/var/lib/resgrid-relay/recordings +# RESGRID__RELAY__Recorder__Log=jsonl # jsonl | sqlite | none +# RESGRID__RELAY__Recorder__S3__Bucket=my-bucket +# RESGRID__RELAY__Recorder__S3__Region=us-east-1 +# RESGRID__RELAY__Recorder__S3__AccessKey=... # (+ SecretKey) + +# ── smtp mode: dispatch-email relay ────────────────────────────────────────── +# RESGRID__RELAY__Smtp__Port=2525 +# RESGRID__RELAY__Smtp__DepartmentAddressDomains__0=dispatch.resgrid.com +# RESGRID__RELAY__Smtp__RedisCache__Enabled=false + +# ── Resilience (LiveKit voice modes: radio / record / dispatch) ────────────── +# Auto-reconnect with exponential back-off; the circuit breaker faults the service +# only after MaxConsecutiveFailures quick failures with no healthy run in between. +RESGRID__RELAY__Resilience__Enabled=true +RESGRID__RELAY__Resilience__MaxConsecutiveFailures=5 +RESGRID__RELAY__Resilience__InitialBackoffSeconds=2 +RESGRID__RELAY__Resilience__MaxBackoffSeconds=60 +RESGRID__RELAY__Resilience__HealthyRunSeconds=30 + +# ── Optional Sentry monitoring (LiveKit voice modes) ───────────────────────── +# Leave Dsn empty to disable Sentry entirely (no SDK init, completely no-op). +RESGRID__RELAY__Telemetry__Environment=production +RESGRID__RELAY__Telemetry__Sentry__Dsn= +RESGRID__RELAY__Telemetry__Sentry__Release= +RESGRID__RELAY__Telemetry__Sentry__SendDefaultPii=false diff --git a/deploy/systemd/resgrid-relay.service b/deploy/systemd/resgrid-relay.service new file mode 100644 index 0000000..1630574 --- /dev/null +++ b/deploy/systemd/resgrid-relay.service @@ -0,0 +1,44 @@ +[Unit] +Description=Resgrid Relay — LiveKit backend (SMTP dispatch / call recording / TTS dispatch tone-out) +Documentation=https://docs.resgrid.com +After=network-online.target +Wants=network-online.target + +[Service] +Type=simple + +# Dedicated unprivileged service account (see README.md to create it). +User=resgrid +Group=resgrid + +# Where you published the app. This assumes a framework-dependent publish run via the +# dotnet host; for a self-contained publish, point ExecStart at the binary directly: +# ExecStart=/opt/resgrid-relay/Resgrid.Audio.Relay.Console run +WorkingDirectory=/opt/resgrid-relay +ExecStart=/usr/bin/dotnet /opt/resgrid-relay/Resgrid.Audio.Relay.Console.dll run + +# All configuration is via RESGRID__RELAY__* env vars; keep secrets here (chmod 600). +EnvironmentFile=/etc/resgrid-relay/relay.env + +# The CLI traps SIGTERM (systemd's default stop signal) and SIGINT, shutting down gracefully — +# cancelling in-flight work so recordings flush and LiveKit disconnects cleanly. No KillSignal +# override is needed; TimeoutStopSec bounds the graceful drain before systemd escalates to KILL. +TimeoutStopSec=30 + +Restart=on-failure +RestartSec=5 + +# Writable state (recorder LocalPath, Resgrid token cache, etc.) → /var/lib/resgrid-relay +StateDirectory=resgrid-relay + +# Sandboxing — safe defaults. Relax these if your recorder/output paths live elsewhere. +NoNewPrivileges=true +ProtectSystem=strict +ProtectHome=true +PrivateTmp=true +ProtectKernelTunables=true +ProtectControlGroups=true +RestrictSUIDSGID=true + +[Install] +WantedBy=multi-user.target