diff --git a/src/RemSound.App/SelfTest.cs b/src/RemSound.App/SelfTest.cs index d59279b..8a04fdc 100644 --- a/src/RemSound.App/SelfTest.cs +++ b/src/RemSound.App/SelfTest.cs @@ -67,6 +67,7 @@ internal static class SelfTest RunStep(results, "Service profile isolation (location + hidden from pickers)", ServiceProfileIsolation); RunStep(results, "Service send host (headless stream + yield)", ServiceSendHostStream); RunStep(results, "Service network presence (reachable + shell teardown)", ServiceNetworkPresenceReachable); + RunStep(results, "Service reachability-gated sending (drop dead peers, re-arm recovered)", ServiceReachabilityGating); RunStep(results, "Service registration args", ServiceRegistrationArgs); RunStep(results, "Recording engine (all formats + source gate + mono)", RecordingEngine); RunStep(results, "Recording split tracks (per-peer + own)", RecordingSplitTracks); @@ -1475,6 +1476,43 @@ internal static class SelfTest finally { RemSound.Sender.ProcessLoopbackCapture.ForceSupportedForTest = prev; } } + /// The service must only stream to peers the heartbeat can reach and drop long-unreachable + /// ones (never blast audio into a dead address — issue #8), re-arming a peer the moment it recovers + /// (issue #15) — the same behaviour as the app's RefreshAudioReceivers. Tests the pure arming logic. + private static string? ServiceReachabilityGating() + { + var a = new IPEndPoint(IPAddress.Parse("10.0.0.1"), 47830); + var b = new IPEndPoint(IPAddress.Parse("10.0.0.2"), 47830); + var all = new[] { a, b }; + var prune = TimeSpan.FromSeconds(30); + + var bothHealthy = new List + { + new(a, PeerHealthState.Healthy, 10, TimeSpan.FromSeconds(1)), + new(b, PeerHealthState.Healthy, 12, TimeSpan.FromSeconds(1)), + }; + Check(ServiceSendHost.ComputeArmedEndpoints(all, bothHealthy, prune).Length == 2, "both reachable peers must be armed"); + + var bDeadLong = new List + { + new(a, PeerHealthState.Healthy, 10, TimeSpan.FromSeconds(1)), + new(b, PeerHealthState.Unreachable, null, TimeSpan.FromSeconds(60)), + }; + var armed = ServiceSendHost.ComputeArmedEndpoints(all, bDeadLong, prune); + Check(armed.Length == 1 && armed[0].Equals(a), "a peer unreachable past the grace window must be dropped (never stream into a dead address)"); + + var bDeadGrace = new List + { + new(a, PeerHealthState.Healthy, 10, TimeSpan.FromSeconds(1)), + new(b, PeerHealthState.Unreachable, null, TimeSpan.FromSeconds(10)), + }; + Check(ServiceSendHost.ComputeArmedEndpoints(all, bDeadGrace, prune).Length == 2, "a briefly-unreachable peer stays armed during the grace window"); + + Check(ServiceSendHost.ComputeArmedEndpoints(all, new List(), prune).Length == 2, "with no heartbeat data yet, arm the full set"); + + return "reachable armed; long-unreachable dropped; grace-window kept; recovery re-arms (issues #8/#15)"; + } + private static int FreeUdpPort() { using var s = new System.Net.Sockets.Socket(System.Net.Sockets.AddressFamily.InterNetwork, diff --git a/src/RemSound.App/ServiceNetworkPresence.cs b/src/RemSound.App/ServiceNetworkPresence.cs index 62475a8..88b6511 100644 --- a/src/RemSound.App/ServiceNetworkPresence.cs +++ b/src/RemSound.App/ServiceNetworkPresence.cs @@ -49,6 +49,10 @@ internal sealed class ServiceNetworkPresence : IDisposable /// Test seam: is the well-known-port listener actually bound? internal bool ListenerBound => receiver.IsListenerRunning; + /// Current per-peer heartbeat health (reachable / stale / unreachable + age), so the host can + /// gate the audio send on reachability — the same signal the interactive app uses. Empty when down. + public IReadOnlyList PeerHealthSnapshot() => heartbeat?.GetAllPeerHealth() ?? Array.Empty(); + /// Bring the service up as a discoverable, reachable, send-only peer on , /// tracking (the profile's peers) for heartbeat/pairing and unicast /// announcements. Idempotent: re-applies cleanly if already up. Never throws. diff --git a/src/RemSound.App/ServiceSendHost.cs b/src/RemSound.App/ServiceSendHost.cs index 9875ae7..35d5bdf 100644 --- a/src/RemSound.App/ServiceSendHost.cs +++ b/src/RemSound.App/ServiceSendHost.cs @@ -43,6 +43,14 @@ public sealed class ServiceSendHost : IDisposable private volatile bool wantSending; // true while the app is absent and we intend to stream private long lastDeviceChangeTick; + // Reachability-gated sending (issues #8 / #15): stream ONLY to peers the heartbeat can reach, and drop + // any that stay unreachable — never blast audio into a dead address forever. The heartbeat keeps + // probing the FULL set, so a peer that comes back is re-armed on the next refresh. Mirrors the app's + // RefreshAudioReceivers. Send-only, so there's no "actively receiving" carve-out. + private static readonly TimeSpan PruneUnreachableAfter = TimeSpan.FromSeconds(30); + private IPEndPoint[] allEndpoints = []; + private string? armedSignature; + /// Supplies the current service profile (re-read on each resume so edits /// are picked up). Returns null if none is configured. /// Optional diagnostic sink. @@ -94,6 +102,10 @@ public sealed class ServiceSendHost : IDisposable sender.ConfigureCodec(profile.Codec, MainForm.EffectiveOpusFrameSamples(profile.Codec, profile.OpusFrameSamplesPerChannel, profile.SendRate)); sender.SetSendRate(profile.SendRate); sender.SetTightLatency(profile.TightLatencyMode); + // Arm the full set to begin with (nothing is known-dead yet); RefreshSendArming then prunes any + // peer the heartbeat can't reach and re-arms it when it recovers. + allEndpoints = endpoints.ToArray(); + armedSignature = null; sender.SetReceivers(endpoints); sender.Configure(specs); sender.Start(); @@ -130,6 +142,39 @@ public sealed class ServiceSendHost : IDisposable return ApplyProfile(profile); } + /// Pure, testable: which endpoints to actively stream to — the full set minus any peer the + /// heartbeat reports as continuously unreachable for longer than . A peer + /// that's reachable, or still within the grace window, stays armed. Mirrors the app's RefreshAudioReceivers. + internal static IPEndPoint[] ComputeArmedEndpoints(IReadOnlyList all, IReadOnlyList health, TimeSpan pruneAfter) + { + HashSet? dead = null; + foreach (var ph in health) + { + if (ph.State == PeerHealthState.Unreachable && ph.AgeOfLastPong is { } age && age > pruneAfter) + (dead ??= new HashSet(StringComparer.OrdinalIgnoreCase)).Add($"{ph.AudioEndpoint.Address}:{ph.AudioEndpoint.Port}"); + } + return dead is null ? all.ToArray() : all.Where(ep => !dead.Contains($"{ep.Address}:{ep.Port}")).ToArray(); + } + + /// Re-arm the sender to only the reachable peers, using the heartbeat health. Cheap and + /// idempotent — only touches the sender when the armed set actually changes. Called on the service's + /// existing poll tick while streaming, so there's no extra timer. + private void RefreshSendArming() + { + lock (gate) + { + if (!running || allEndpoints.Length == 0) return; + var armed = ComputeArmedEndpoints(allEndpoints, presence.PeerHealthSnapshot(), PruneUnreachableAfter); + var sig = string.Join("|", armed.Select(ep => $"{ep.Address}:{ep.Port}").OrderBy(s => s, StringComparer.OrdinalIgnoreCase)); + if (sig == armedSignature) return; + armedSignature = sig; + sender.SetReceivers(armed); + log?.Invoke(armed.Length == 0 + ? $"service: 0 reachable peers — holding audio (heartbeat still probing {allEndpoints.Length})" + : $"service: streaming to {armed.Length}/{allEndpoints.Length} reachable peer(s)"); + } + } + /// The service's main loop: watch the interactive-presence token and hand the send back and /// forth. Starts sending immediately if no app is present. A short settle delay before resuming /// stops rapid app open/close from thrashing the engine. Returns when is @@ -174,6 +219,9 @@ public sealed class ServiceSendHost : IDisposable if (!triedThisAbsence && !IsSending) { Resume(); triedThisAbsence = true; } } } + // While streaming, re-arm to only the reachable peers (drop dead ones, pick up recovered ones). + // Piggybacks this existing tick — no extra timer, no background pile-up. + if (IsSending) RefreshSendArming(); appWasPresent = appPresent; ct.WaitHandle.WaitOne(pollMs); }