Merge pull request #16 from Resgrid/develop

RR1-T102 Fixes and hardening and systemd support for cli

Author: ucswift

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<int> 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.

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
+{
+	/// <summary>
+	/// Behavioral coverage for the LiveKit-mode resilience loop in
+	/// <see cref="RelayServiceBase"/>: exponential back-off retries, the consecutive-failure
+	/// circuit breaker, and graceful stop while retrying. Uses a tiny test subclass with a
+	/// scripted <c>ExecuteAsync</c> and near-zero back-off so the tests run fast.
+	/// </summary>
+	[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");
+		}
+
+		/// <summary>Minimal LiveKit-mode service whose run is scripted by the test.</summary>
+		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<ScriptedRelayService, CancellationToken, Task> ExecuteBehavior { get; set; }
+
+			protected override async Task ExecuteAsync(CancellationToken token)
+			{
+				Interlocked.Increment(ref _attempts);
+				await ExecuteBehavior(this, token).ConfigureAwait(false);
+			}
+		}
+	}
+}

Resgrid.Audio.Voice.Tests/Resgrid.Audio.Voice.Tests.csproj

@@ -16,5 +16,6 @@
 
   <ItemGroup>
     <ProjectReference Include="..\Resgrid.Audio.Voice\Resgrid.Audio.Voice.csproj" />
+    <ProjectReference Include="..\Resgrid.Relay.Engine\Resgrid.Relay.Engine.csproj" />
   </ItemGroup>
 </Project>

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();
+	}
+
+	/// <summary>
+	/// 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.
+	/// </summary>
+	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

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")]

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;

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;

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;

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;
 		}
 
+		/// <summary>
+		/// 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.
+		/// </summary>
+		protected virtual bool IsLiveKitMode => false;
+
 		public string Mode { get; }
 		public RelayServiceState State => _state;
 		public event EventHandler<RelayStateChangedEventArgs> StateChanged;
@@ -78,21 +90,24 @@ 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)
 			{
 				// 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)
 			{
 				// 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);
+		}
+
+		/// <summary>
+		/// Runs <see cref="ExecuteAsync"/>, 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 <see cref="StartAsync"/> faults the service) once <see cref="ResilienceOptions.MaxConsecutiveFailures"/>
+		/// failures occur without a healthy run resetting the counter. Non-LiveKit modes,
+		/// and any mode with resilience disabled, run <see cref="ExecuteAsync"/> once directly.
+		/// </summary>
+		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);
+				}
+			}
+		}
+
+		/// <summary>
+		/// Exponential back-off (doubling from <see cref="ResilienceOptions.InitialBackoffSeconds"/>,
+		/// capped at <see cref="ResilienceOptions.MaxBackoffSeconds"/>) with ±20% jitter, floored at
+		/// 0.5s so retries never busy-spin.
+		/// </summary>
+		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)

Resgrid.Relay.Engine/Telemetry/IRelayModeTelemetry.cs

@@ -0,0 +1,19 @@
+using System;
+using System.Threading.Tasks;
+
+namespace Resgrid.Relay.Engine.Telemetry
+{
+	/// <summary>
+	/// 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.
+	/// </summary>
+	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);
+	}
+}