Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 11 additions & 0 deletions Resgrid.Audio.Relay.Console/Program.cs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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();
});
Comment on lines +93 to +97

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

kody code-review Bug critical

PlatformNotSupportedException thrown at startup on Windows because PosixSignalRegistration.Create(PosixSignal.SIGTERM) runs without a platform guard and sits outside the try block in Main, causing an unhandled exception since the Engine targets net10.0-windows. Guard the registration with OperatingSystem.IsLinux() or OperatingSystem.IsMacOS() so the SIGTERM handler only registers on supported platforms.

PosixSignalRegistration sigTermRegistration = null;
if (OperatingSystem.IsLinux() || OperatingSystem.IsMacOS())
{
	sigTermRegistration = PosixSignalRegistration.Create(PosixSignal.SIGTERM, context =>
	{
		context.Cancel = true;
		cancellationTokenSource.Cancel();
	});
}
Prompt for LLM

File Resgrid.Audio.Relay.Console/Program.cs:

Line 93 to 97:

WHAT: PosixSignalRegistration.Create(PosixSignal.SIGTERM) is called without a platform guard and sits outside any try/catch, but SIGTERM is not a supported PosixSignal on Windows — .NET throws PlatformNotSupportedException. WHY: The Engine targets net10.0-windows, so running the CLI on Windows now crashes at startup (the throw is before the try block, so Main receives an unhandled exception). HOW: Guard the registration with OperatingSystem.IsLinux() (or wrap in try/catch PlatformNotSupportedException) so the SIGTERM handler only registers on platforms that support it.

Suggested Code:

PosixSignalRegistration sigTermRegistration = null;
if (OperatingSystem.IsLinux() || OperatingSystem.IsMacOS())
{
	sigTermRegistration = PosixSignalRegistration.Create(PosixSignal.SIGTERM, context =>
	{
		context.Cancel = true;
		cancellationTokenSource.Cancel();
	});
}

Talk to Kody by mentioning @kody

Was this suggestion helpful? React with 👍 or 👎 to help Kody learn from this interaction.


try
{
// The engine owns all mode wiring; the console just builds the service and runs it.
Expand Down
139 changes: 139 additions & 0 deletions Resgrid.Audio.Voice.Tests/RelayResilienceTests.cs
Original file line number Diff line number Diff line change
@@ -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);
}
}
}
}
1 change: 1 addition & 0 deletions Resgrid.Audio.Voice.Tests/Resgrid.Audio.Voice.Tests.csproj
Original file line number Diff line number Diff line change
Expand Up @@ -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>
18 changes: 18 additions & 0 deletions Resgrid.Relay.Engine/Configuration/RelayHostOptions.cs
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
4 changes: 4 additions & 0 deletions Resgrid.Relay.Engine/Properties/AssemblyInfo.cs
Original file line number Diff line number Diff line change
@@ -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")]
2 changes: 2 additions & 0 deletions Resgrid.Relay.Engine/Services/DispatchRelayService.cs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
2 changes: 2 additions & 0 deletions Resgrid.Relay.Engine/Services/RadioRelayService.cs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
2 changes: 2 additions & 0 deletions Resgrid.Relay.Engine/Services/RecordRelayService.cs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
85 changes: 83 additions & 2 deletions Resgrid.Relay.Engine/Services/RelayServiceBase.cs
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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;

Expand All @@ -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;
Expand Down Expand Up @@ -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);
}
}
Expand Down Expand Up @@ -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)
Expand Down
19 changes: 19 additions & 0 deletions Resgrid.Relay.Engine/Telemetry/IRelayModeTelemetry.cs
Original file line number Diff line number Diff line change
@@ -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);
}
}
Loading
Loading