diff --git a/src/dapps/dapps.core.tests/MeshCoreBearerTests.cs b/src/dapps/dapps.core.tests/MeshCoreBearerTests.cs index af173e8..6f6a200 100644 --- a/src/dapps/dapps.core.tests/MeshCoreBearerTests.cs +++ b/src/dapps/dapps.core.tests/MeshCoreBearerTests.cs @@ -151,16 +151,28 @@ public void ChannelMonitor_OccupancyRisesWithTrafficThenPrunes() } [Fact] - public void TxBudget_Refund_ReturnsTheLastReservation() + public void TxBudget_Refund_ReturnsTheReservation() { var now = new DateTime(2026, 1, 1, 12, 0, 0, DateTimeKind.Utc); var b = new TxBudget(secondsPerHour: 1.0); // 1000 ms/hr - b.TryReserve(700, now, out _).Should().BeTrue(); - b.TryReserve(700, now, out _).Should().BeFalse("1400ms > 1000ms budget"); - b.Refund(); // give the 700ms back + b.TryReserve(700, now, out _, out var t1).Should().BeTrue(); + b.TryReserve(700, now, out _, out _).Should().BeFalse("1400ms > 1000ms budget"); + b.Refund(t1); b.UsedSeconds(now).Should().BeApproximately(0, 0.001); - b.TryReserve(900, now, out _).Should().BeTrue("budget was refunded"); + b.TryReserve(900, now, out _, out _).Should().BeTrue("budget was refunded"); + } + + [Fact] + public void TxBudget_Refund_RemovesTheSpecificReservation_NotJustTheLast() + { + var now = new DateTime(2026, 1, 1, 12, 0, 0, DateTimeKind.Utc); + var b = new TxBudget(secondsPerHour: 1.0); + + b.TryReserve(300, now, out _, out var a).Should().BeTrue(); + b.TryReserve(300, now, out _, out _).Should().BeTrue(); + b.Refund(a); // refund the FIRST reservation, not the most recent + b.UsedSeconds(now).Should().BeApproximately(0.3, 0.001, "only the second 300ms reservation remains"); } [Fact] @@ -170,6 +182,54 @@ public void ChannelData_ParseRecv_ShortFrame_ThrowsInvalidData() act.Should().Throw(); } + [Fact] + public void Reliability_TrackThenAck_ConfirmsAndClears() + { + var r = new MeshCoreReliability(); + var now = new DateTime(2026, 1, 1, 12, 0, 0, DateTimeKind.Utc); + var m = SampleMessage("hi") with { Id = "abc0001", Originator = "GB7A-1", Destination = "GB7B-1" }; + + r.Track(m, "GB7A-1", now); + r.PendingCount.Should().Be(1); + r.OnAck("abc0001").Should().BeTrue(); + r.PendingCount.Should().Be(0); + r.Confirmed.Should().Be(1); + r.OnAck("missing").Should().BeFalse(); + } + + [Fact] + public void Reliability_DueResends_RespectBackoffAndDeadline() + { + var r = new MeshCoreReliability(new MeshCoreReliability.Options( + BaseBackoff: TimeSpan.FromSeconds(20), Multiplier: 2.0, + MaxBackoff: TimeSpan.FromSeconds(120), MaxLifetime: TimeSpan.FromSeconds(60))); + var now = new DateTime(2026, 1, 1, 12, 0, 0, DateTimeKind.Utc); + var m = SampleMessage("hi") with { Id = "abc0002", Ttl = 60, Originator = "GB7A-1" }; + + r.Track(m, "GB7A-1", now); + r.DueResends(now).Should().BeEmpty("first backoff is 20s away"); + r.DueResends(now.AddSeconds(21)).Should().ContainSingle(); + r.DueResends(now.AddSeconds(21)).Should().ContainSingle("DueResends must not advance backoff on its own"); + r.MarkResent("abc0002", now.AddSeconds(21)); + r.DueResends(now.AddSeconds(22)).Should().BeEmpty("backoff advanced after MarkResent"); + r.DueResends(now.AddSeconds(120)).Should().BeEmpty("past the lifetime deadline"); + r.DropExpired(now.AddSeconds(120)).Should().ContainSingle(); + r.Expired.Should().Be(1); + } + + [Fact] + public void Reliability_BuildAck_IsAckAddressedToOriginator() + { + var m = SampleMessage("data") with { Id = "data001", Originator = "GB7A-1", Destination = "GB7B-1" }; + var ack = MeshCoreReliability.BuildAck(m, localCallsign: "GB7B-1", ackId: "ack0001"); + + MeshCoreReliability.IsAck(ack).Should().BeTrue(); + MeshCoreReliability.AckedId(ack).Should().Be("data001"); + ack.Destination.Should().Be("GB7A-1"); + ack.Originator.Should().Be("GB7B-1"); + MeshCoreReliability.IsAck(m).Should().BeFalse(); + } + [Fact] public void RouteBuilder_CopiesMeshCoreChannelHint() { diff --git a/src/dapps/dapps.core.tests/MeshCoreConfigTests.cs b/src/dapps/dapps.core.tests/MeshCoreConfigTests.cs index 505d9c4..d51a861 100644 --- a/src/dapps/dapps.core.tests/MeshCoreConfigTests.cs +++ b/src/dapps/dapps.core.tests/MeshCoreConfigTests.cs @@ -51,6 +51,7 @@ public async Task MeshCoreOptions_RoundTripThroughStore() opts.MeshCoreCompress = false; opts.MeshCoreCongestionBackoffFraction = 0.25; opts.MeshCoreLbtGuardMs = 250; + opts.MeshCoreReliableDelivery = false; await store.SaveAsync(opts); @@ -68,5 +69,6 @@ public async Task MeshCoreOptions_RoundTripThroughStore() reloaded.MeshCoreCompress.Should().BeFalse(); reloaded.MeshCoreCongestionBackoffFraction.Should().Be(0.25); reloaded.MeshCoreLbtGuardMs.Should().Be(250); + reloaded.MeshCoreReliableDelivery.Should().BeFalse(); } } diff --git a/src/dapps/dapps.core/Models/SystemOptions.cs b/src/dapps/dapps.core/Models/SystemOptions.cs index 3f3b7b6..1cbd182 100644 --- a/src/dapps/dapps.core/Models/SystemOptions.cs +++ b/src/dapps/dapps.core/Models/SystemOptions.cs @@ -129,6 +129,10 @@ public class SystemOptions /// Listen-before-talk guard in ms (#157). 0 disables. public int MeshCoreLbtGuardMs { get; set; } = 400; + /// End-to-end reliability (#26): ACK received messages + resend our own + /// unacked messages until acked or their lifetime expires. + public bool MeshCoreReliableDelivery { get; set; } = true; + /// /// When true, app-interface clients (MQTT and REST) must present a /// valid token; topic / endpoint scope is also enforced against the diff --git a/src/dapps/dapps.core/Services/DbStartup.cs b/src/dapps/dapps.core/Services/DbStartup.cs index 4f7ce1e..62926d1 100644 --- a/src/dapps/dapps.core/Services/DbStartup.cs +++ b/src/dapps/dapps.core/Services/DbStartup.cs @@ -103,6 +103,7 @@ private static readonly (string Key, string Default)[] SeededOptions = ("MeshCoreCompress", "true"), ("MeshCoreCongestionBackoffFraction", "0.5"), ("MeshCoreLbtGuardMs", "400"), + ("MeshCoreReliableDelivery", "true"), ("AuthRequired", "false"), ("UpdateCheckEnabled", "true"), ("RoutingAlgorithm", "passive-flood"), diff --git a/src/dapps/dapps.core/Services/MeshCoreBearer.cs b/src/dapps/dapps.core/Services/MeshCoreBearer.cs index b57cc26..3b7be96 100644 --- a/src/dapps/dapps.core/Services/MeshCoreBearer.cs +++ b/src/dapps/dapps.core/Services/MeshCoreBearer.cs @@ -58,6 +58,7 @@ public async Task RunAsync(CancellationToken ct) var opts = BuildOptions(s); var budget = new TxBudget(opts.AirtimeBudgetSecPerHour); + var reliability = opts.ReliableDelivery ? new MeshCoreReliability() : null; _link = new MeshCoreLink(opts, _loggerFactory.CreateLogger()); try @@ -70,12 +71,45 @@ public async Task RunAsync(CancellationToken ct) return; } - _backhaul = new MeshCoreCompanionBackhaul( - _link, opts, budget, _loggerFactory.CreateLogger(), _txGate); - _inbound = new MeshCoreInbound(_link, _inbox, _loggerFactory.CreateLogger()); + var backhaul = new MeshCoreCompanionBackhaul( + _link, opts, budget, _loggerFactory.CreateLogger(), _txGate, reliability); + _backhaul = backhaul; + _inbound = new MeshCoreInbound( + _link, _inbox, _loggerFactory.CreateLogger(), + reliability, + sendAck: (ack, c) => backhaul.ResendAsync(ack, opts.LocalCallsign, c), + localCallsign: opts.LocalCallsign); Enabled = true; - await _inbound.RunAsync(ct); + // Reliability resend loop runs alongside the inbound drain loop. + var resendTask = reliability is not null + ? Task.Run(() => ResendLoopAsync(reliability, backhaul, ct), ct) + : Task.CompletedTask; + try { await _inbound.RunAsync(ct); } + finally { try { await resendTask; } catch { /* shutdown */ } } + } + + private async Task ResendLoopAsync(MeshCoreReliability reliability, MeshCoreCompanionBackhaul backhaul, CancellationToken ct) + { + while (!ct.IsCancellationRequested) + { + try { await Task.Delay(TimeSpan.FromSeconds(5), ct); } + catch (OperationCanceledException) { break; } + + var now = DateTime.UtcNow; + foreach (var m in reliability.DropExpired(now)) + _log.LogWarning("MeshCore: reliable delivery gave up on {0} (unacked past lifetime)", m.Id); + foreach (var (msg, local) in reliability.DueResends(now)) + { + try + { + var res = await backhaul.ResendAsync(msg, local, ct); + if (res.Accepted) reliability.MarkResent(msg.Id, DateTime.UtcNow); + else _log.LogDebug("MeshCore: resend of {0} deferred: {1}", msg.Id, res.Error); + } + catch (Exception ex) { _log.LogDebug("MeshCore: resend of {0} failed: {1}", msg.Id, ex.Message); } + } + } } public bool CanHandle(BackhaulRoute route) => @@ -100,6 +134,8 @@ public Task SendAsync( Compress = s.MeshCoreCompress, CongestionBackoffFraction = s.MeshCoreCongestionBackoffFraction, LbtGuardMs = s.MeshCoreLbtGuardMs, + ReliableDelivery = s.MeshCoreReliableDelivery, + LocalCallsign = s.Callsign, AppName = "dapps", }; diff --git a/src/dapps/dapps.core/Services/SystemOptionsStore.cs b/src/dapps/dapps.core/Services/SystemOptionsStore.cs index 01fb0e4..df549d0 100644 --- a/src/dapps/dapps.core/Services/SystemOptionsStore.cs +++ b/src/dapps/dapps.core/Services/SystemOptionsStore.cs @@ -118,6 +118,7 @@ public async Task SaveAsync(SystemOptions options) await Upsert(connection, existing, nameof(options.MeshCoreCompress), options.MeshCoreCompress.ToString()); await Upsert(connection, existing, nameof(options.MeshCoreCongestionBackoffFraction), options.MeshCoreCongestionBackoffFraction.ToString(CultureInfo.InvariantCulture)); await Upsert(connection, existing, nameof(options.MeshCoreLbtGuardMs), options.MeshCoreLbtGuardMs.ToString()); + await Upsert(connection, existing, nameof(options.MeshCoreReliableDelivery), options.MeshCoreReliableDelivery.ToString()); Reload(); } @@ -198,6 +199,7 @@ private static SystemOptions Parse(Dictionary r) MeshCoreCompress = TryGetBool(r, nameof(SystemOptions.MeshCoreCompress), true), MeshCoreCongestionBackoffFraction = TryGetDouble(r, nameof(SystemOptions.MeshCoreCongestionBackoffFraction), 0.5, min: 0, max: 1), MeshCoreLbtGuardMs = TryGetInt(r, nameof(SystemOptions.MeshCoreLbtGuardMs), 400, min: 0), + MeshCoreReliableDelivery = TryGetBool(r, nameof(SystemOptions.MeshCoreReliableDelivery), true), }; } diff --git a/src/dapps/dapps.meshcore.soak/Program.cs b/src/dapps/dapps.meshcore.soak/Program.cs index c6e6c5b..e607152 100644 --- a/src/dapps/dapps.meshcore.soak/Program.cs +++ b/src/dapps/dapps.meshcore.soak/Program.cs @@ -34,6 +34,8 @@ Compress = !a.Has("no-compress"), CongestionBackoffFraction = a.GetDouble("congestion", 0.5), LbtGuardMs = a.GetInt("lbt", 400), + ReliableDelivery = !a.Has("no-reliable"), + LocalCallsign = self, AppName = "dapps-soak", }; @@ -55,11 +57,41 @@ try { await link.StartAsync(cts.Token); } catch (Exception ex) { log.LogError(ex, "link failed to start"); return 2; } -var backhaul = new MeshCoreCompanionBackhaul(link, opts, budget, lf.CreateLogger()); -var inbound = new MeshCoreInbound(link, inbox, lf.CreateLogger()); +var reliability = opts.ReliableDelivery ? new MeshCoreReliability() : null; +var backhaul = new MeshCoreCompanionBackhaul( + link, opts, budget, lf.CreateLogger(), reliability: reliability); + +// Optional induced loss to exercise reliability resends (soak only). +double dropPct = a.GetDouble("drop-pct", 0); +Func? drop = dropPct > 0 ? (_ => Random.Shared.NextDouble() * 100 < dropPct) : null; + +var inbound = new MeshCoreInbound( + link, inbox, lf.CreateLogger(), + reliability, + sendAck: (ack, c) => backhaul.ResendAsync(ack, self, c), + localCallsign: self, + dropForTest: drop); var route = new BackhaulRoute(peer, MeshCoreChannel: opts.ChannelName); var inboundTask = Task.Run(() => inbound.RunAsync(cts.Token)); +var resendTask = reliability is not null ? Task.Run(async () => +{ + while (!cts.IsCancellationRequested) + { + try { await Task.Delay(TimeSpan.FromSeconds(5), cts.Token); } catch { break; } + var now = DateTime.UtcNow; + foreach (var m in reliability.DropExpired(now)) log.LogWarning("reliable delivery gave up on {0}", m.Id); + foreach (var (msg, local) in reliability.DueResends(now)) + { + try + { + var res = await backhaul.ResendAsync(msg, local, cts.Token); + if (res.Accepted) reliability.MarkResent(msg.Id, DateTime.UtcNow); + } + catch { } + } + } +}) : Task.CompletedTask; long sent = 0, accepted = 0, throttled = 0, backedOff = 0, failed = 0; string[] samples = @@ -108,7 +140,7 @@ // Run for the duration, then stop. try { await Task.Delay(TimeSpan.FromSeconds(durationSec), cts.Token); } catch { } cts.Cancel(); -try { await Task.WhenAll(senderTask, inboundTask); } catch { } +try { await Task.WhenAll(senderTask, inboundTask, resendTask); } catch { } var (recv, maxSeq, distinct) = inbox.Snapshot(); double lossPct = maxSeq >= 0 ? 100.0 * (1.0 - (double)distinct / (maxSeq + 1)) : 0; @@ -118,6 +150,9 @@ log.LogInformation("Channel occupancy (trailing 60s): {0:0.0}%", backhaul.Occupancy * 100); log.LogInformation("Airtime used (trailing hr): {0:0.0}s ({1:0.000}% duty); link resets={2}; link state={3}", budget.UsedSeconds(DateTime.UtcNow), budget.DutyPercent(DateTime.UtcNow), link.ResetCount, link.State); +if (reliability is not null) + log.LogInformation("Reliability: confirmed={0} expired={1} pending={2} (induced drop {3:0.#}%)", + reliability.Confirmed, reliability.Expired, reliability.PendingCount, dropPct); return 0; // ---------------- helpers ---------------- diff --git a/src/dapps/dapps.meshcore/MeshCoreBearerOptions.cs b/src/dapps/dapps.meshcore/MeshCoreBearerOptions.cs index ffeb9c2..39843ad 100644 --- a/src/dapps/dapps.meshcore/MeshCoreBearerOptions.cs +++ b/src/dapps/dapps.meshcore/MeshCoreBearerOptions.cs @@ -29,6 +29,14 @@ public sealed class MeshCoreBearerOptions /// to avoid colliding with an in-progress flood. 0 disables. public int LbtGuardMs { get; set; } = 400; + /// End-to-end reliability (#26): ACK received messages addressed to us + /// and resend our own unacked messages until acked or their lifetime expires. + public bool ReliableDelivery { get; set; } = true; + + /// This node's DAPPS callsign — decides which received messages to ACK + /// (those addressed to us) and is the ACK originator. + public string LocalCallsign { get; set; } = ""; + public RegionPreset ResolveRegion() => Regions.Find(Region) ?? throw new ArgumentException($"unknown MeshCore region '{Region}'"); diff --git a/src/dapps/dapps.meshcore/MeshCoreChannelTransport.cs b/src/dapps/dapps.meshcore/MeshCoreChannelTransport.cs index 3e5201f..30add83 100644 --- a/src/dapps/dapps.meshcore/MeshCoreChannelTransport.cs +++ b/src/dapps/dapps.meshcore/MeshCoreChannelTransport.cs @@ -22,7 +22,8 @@ public sealed class MeshCoreChannelTransport public const int Mtu = 160; private readonly Reassembler _reassembler = new(); - private readonly Dictionary _compressed = new(); + private readonly Dictionary _compressed = new(); + private readonly object _nonceLock = new(); private byte _nonce; /// Encode a BackhaulMessage into one-or-more channel-data payloads. @@ -35,8 +36,14 @@ public IReadOnlyList ToFrames(BackhaulMessage message, DappsCompression. var frames = new List(fragments.Count); foreach (var f in fragments) { - byte hdr = (byte)((_nonce << 1) | (comp ? 1 : 0)); - _nonce = (byte)((_nonce + 1) & 0x7F); + // ToFrames is now called concurrently (OMM send + reliability resend loop + // + inbound ACK emission), so the rolling nonce must be atomic. + byte hdr; + lock (_nonceLock) + { + hdr = (byte)((_nonce << 1) | (comp ? 1 : 0)); + _nonce = (byte)((_nonce + 1) & 0x7F); + } var frame = new byte[1 + f.Length]; frame[0] = hdr; f.CopyTo(frame, 1); @@ -60,11 +67,11 @@ public Result Ingest(byte[] dataPayload, DateTime now) try { header = Packetiser.ParseHeader(fragment); } catch (InvalidDataException) { return new Result(Kind.Bad, null, null); } - _compressed[header.Id] = comp; + _compressed[header.Id] = (comp, now); var assembled = _reassembler.Accept(fragment, now); if (assembled is null) return new Result(Kind.FragmentPartial, null, header); - var compressed = _compressed.TryGetValue(header.Id, out var c) && c; + var compressed = _compressed.TryGetValue(header.Id, out var c) && c.comp; _compressed.Remove(header.Id); try { @@ -77,7 +84,12 @@ public Result Ingest(byte[] dataPayload, DateTime now) } } - /// Drop reassembly state for messages whose first fragment is older - /// than . - public int DropStale(DateTime cutoff) => _reassembler.DropOlderThan(cutoff); + /// Drop reassembly state (and the matching compressed-flag entries) for + /// messages whose first fragment is older than . + public int DropStale(DateTime cutoff) + { + foreach (var k in _compressed.Where(kv => kv.Value.seen < cutoff).Select(kv => kv.Key).ToList()) + _compressed.Remove(k); + return _reassembler.DropOlderThan(cutoff); + } } diff --git a/src/dapps/dapps.meshcore/MeshCoreCompanionBackhaul.cs b/src/dapps/dapps.meshcore/MeshCoreCompanionBackhaul.cs index 7028560..58e1819 100644 --- a/src/dapps/dapps.meshcore/MeshCoreCompanionBackhaul.cs +++ b/src/dapps/dapps.meshcore/MeshCoreCompanionBackhaul.cs @@ -29,18 +29,21 @@ public sealed class MeshCoreCompanionBackhaul : IDappsBackhaul private readonly ILogger _log; private readonly MeshCoreChannelTransport _tx = new(); private readonly ChannelMonitor _monitor; + private readonly MeshCoreReliability? _reliability; private readonly double _congestionThreshold; private readonly Dictionary _recent = new(); private readonly object _recentLock = new(); public MeshCoreCompanionBackhaul( - MeshCoreLink link, MeshCoreBearerOptions opts, TxBudget budget, ILogger log, IDappsTxGate? txGate = null) + MeshCoreLink link, MeshCoreBearerOptions opts, TxBudget budget, ILogger log, + IDappsTxGate? txGate = null, MeshCoreReliability? reliability = null) { _link = link; _opts = opts; _budget = budget; _log = log; _txGate = txGate ?? AlwaysOpenTxGate.Instance; + _reliability = reliability; _region = opts.ResolveRegion(); _monitor = new ChannelMonitor(_region); link.PacketHeard += (len, _) => _monitor.RecordHeard(len, DateTime.UtcNow); @@ -58,13 +61,31 @@ public MeshCoreCompanionBackhaul( public async Task SendAsync( BackhaulMessage message, BackhaulRoute route, string localCallsign, CancellationToken ct) + { + var result = await SendCoreAsync(message, localCallsign, coalesce: true, ct); + // Reliable delivery (#26): track a successfully-sent data message so we + // resend it until the destination ACKs (or its lifetime expires). + if (result.Accepted && _reliability is not null && !MeshCoreReliability.IsAck(message)) + _reliability.Track(message, localCallsign, DateTime.UtcNow); + return result; + } + + /// Re-broadcast a message (a reliability resend, or an ACK) bypassing + /// the broadcast-coalescing dedup — still gated by the governor + adaptive + /// controls. Not tracked for reliability. + public Task ResendAsync(BackhaulMessage message, string localCallsign, CancellationToken ct) + => SendCoreAsync(message, localCallsign, coalesce: false, ct); + + private async Task SendCoreAsync( + BackhaulMessage message, string localCallsign, bool coalesce, CancellationToken ct) { if (!_txGate.TxAllowed) return BackhaulSendResult.Fail($"tx-stopped: {_txGate.BlockReason ?? "(no reason)"}"); // Broadcast coalescing: a channel send reaches every member, so the same // message offered for multiple neighbours need only go on air once. - if (AlreadyBroadcast(message.Id)) + // Resends deliberately bypass this (coalesce=false). + if (coalesce && AlreadyBroadcast(message.Id)) return BackhaulSendResult.Ok(); // Adaptive congestion backoff (#157): don't pile onto a busy shared channel. @@ -79,22 +100,23 @@ public async Task SendAsync( for (var i = 0; i < frames.Count; i++) { var airMs = LoRaAirtime.FrameMs(frames[i].Length, _region); - if (!_budget.TryReserve(airMs, DateTime.UtcNow, out var reason)) + if (!_budget.TryReserve(airMs, DateTime.UtcNow, out var reason, out var token)) return BackhaulSendResult.Fail(reason); await ListenBeforeTalkAsync(ct); // avoid colliding with an in-progress flood bool sent; try { sent = await _link.SendDataAsync(frames[i], ct); } - catch (Exception ex) { _budget.Refund(); return BackhaulSendResult.Fail($"meshcore send failed: {ex.Message}"); } - if (!sent) { _budget.Refund(); return BackhaulSendResult.Fail($"meshcore link not ready ({_link.State})"); } + catch (Exception ex) { _budget.Refund(token); return BackhaulSendResult.Fail($"meshcore send failed: {ex.Message}"); } + if (!sent) { _budget.Refund(token); return BackhaulSendResult.Fail($"meshcore link not ready ({_link.State})"); } if (i < frames.Count - 1) { try { await Task.Delay(FramePace, ct); } catch { } } } - MarkBroadcast(message.Id); - _log.LogInformation("MeshCore: broadcast {0} ({1} frame(s)) dst={2} from={3} duty={4:0.00}% occ={5:0.0}%", - message.Id, frames.Count, message.Destination, localCallsign, _budget.DutyPercent(DateTime.UtcNow), occ * 100); + if (coalesce) MarkBroadcast(message.Id); + _log.LogInformation("MeshCore: {0} {1} ({2} frame(s)) dst={3} from={4} duty={5:0.00}% occ={6:0.0}%", + coalesce ? "broadcast" : "resend", message.Id, frames.Count, message.Destination, localCallsign, + _budget.DutyPercent(DateTime.UtcNow), occ * 100); return BackhaulSendResult.Ok(); } diff --git a/src/dapps/dapps.meshcore/MeshCoreInbound.cs b/src/dapps/dapps.meshcore/MeshCoreInbound.cs index 023e05a..2d8b002 100644 --- a/src/dapps/dapps.meshcore/MeshCoreInbound.cs +++ b/src/dapps/dapps.meshcore/MeshCoreInbound.cs @@ -23,15 +23,33 @@ public sealed class MeshCoreInbound private readonly ILogger _log; private readonly MeshCoreChannelTransport _rx = new(); private readonly SemaphoreSlim _wake = new(0); + private readonly MeshCoreReliability? _reliability; + private readonly Func>? _sendAck; + private readonly string _localCallsign; + private readonly Func? _dropForTest; + // Idempotency (#26): ids already delivered to the app, so a resend (after a lost + // ACK) isn't delivered twice. Single-threaded (drained on one loop); window > the + // reliability lifetime so we remember long enough to cover resends. + private readonly Dictionary _delivered = new(StringComparer.Ordinal); + private static readonly TimeSpan DeliveredDedupWindow = TimeSpan.FromMinutes(10); /// Count of fully-decoded BackhaulMessages delivered (observability). public long Delivered { get; private set; } - public MeshCoreInbound(MeshCoreLink link, IBackhaulInbox inbox, ILogger log) + public MeshCoreInbound( + MeshCoreLink link, IBackhaulInbox inbox, ILogger log, + MeshCoreReliability? reliability = null, + Func>? sendAck = null, + string? localCallsign = null, + Func? dropForTest = null) { _link = link; _inbox = inbox; _log = log; + _reliability = reliability; + _sendAck = sendAck; + _localCallsign = localCallsign ?? ""; + _dropForTest = dropForTest; _link.MessageWaiting += () => { try { _wake.Release(); } catch { } }; } @@ -58,23 +76,68 @@ public async Task RunAsync(CancellationToken ct) { foreach (var d in batch.Data) { - var r = _rx.Ingest(d.Payload, DateTime.UtcNow); + var now = DateTime.UtcNow; + var r = _rx.Ingest(d.Payload, now); if (r.Kind != MeshCoreChannelTransport.Kind.BackhaulComplete) continue; - var msg = r.Message!; + + // ACK control frames are consumed here UNCONDITIONALLY (never + // delivered to the app), whether or not local reliability is on. + if (MeshCoreReliability.IsAck(msg)) + { + var acked = MeshCoreReliability.AckedId(msg); + if (acked is not null && _reliability is not null && _reliability.OnAck(acked)) + _log.LogInformation("MeshCore: ACK confirmed {0}", acked); + continue; + } + + // Test-only induced loss (soak): drop before deliver + ack. + if (_dropForTest is not null && _dropForTest(msg)) + { + _log.LogWarning("MeshCore: [test] dropped {0}", msg.Id); + continue; + } + var source = !string.IsNullOrEmpty(msg.LinkSourceCallsign) ? msg.LinkSourceCallsign! : UnknownSourceCallsign; - try + + // Idempotency (#26): a resend after a lost ACK reassembles into the + // same id — deliver to the app only once, but still ACK every copy + // so the sender can stop resending. + if (MarkDeliveredIfNew(msg.Id, now)) { - await _inbox.DeliverAsync(msg, source, ct); - Delivered++; - _log.LogInformation("MeshCore: delivered {0} from {1} (dst={2}, snr={3:0.0}dB)", - msg.Id, source, msg.Destination, d.SnrDb); + try + { + await _inbox.DeliverAsync(msg, source, ct); + Delivered++; + _log.LogInformation("MeshCore: delivered {0} from {1} (dst={2}, snr={3:0.0}dB)", + msg.Id, source, msg.Destination, d.SnrDb); + } + catch (Exception ex) + { + _log.LogError(ex, "MeshCore inbox delivery failed for {0}", msg.Id); + } } - catch (Exception ex) + else { - _log.LogError(ex, "MeshCore inbox delivery failed for {0}", msg.Id); + _log.LogDebug("MeshCore: duplicate {0} already delivered - re-ACK only", msg.Id); + } + + // ACK a data message addressed to us (even duplicates). + if (_reliability is not null && _sendAck is not null && IsForLocal(msg)) + { + try + { + var ack = MeshCoreReliability.BuildAck(msg, _localCallsign, Guid.NewGuid().ToString("N")[..7]); + var res = await _sendAck(ack, ct); + if (!res.Accepted) + _log.LogDebug("MeshCore: ACK for {0} not sent: {1}", msg.Id, res.Error); + } + catch (Exception ex) + { + _log.LogWarning("MeshCore: failed to ACK {0}: {1}", msg.Id, ex.Message); + } } } } @@ -86,4 +149,23 @@ public async Task RunAsync(CancellationToken ct) } } } + + private bool IsForLocal(BackhaulMessage m) => + !string.IsNullOrEmpty(_localCallsign) + && string.Equals(m.Destination, _localCallsign, StringComparison.OrdinalIgnoreCase); + + /// True if this id hasn't been delivered within the dedup window (and + /// records it); false if it's a duplicate. Prunes stale ids opportunistically. + private bool MarkDeliveredIfNew(string id, DateTime now) + { + if (_delivered.Count > 0) + { + var cutoff = now - DeliveredDedupWindow; + foreach (var k in _delivered.Where(kv => kv.Value < cutoff).Select(kv => kv.Key).ToList()) + _delivered.Remove(k); + } + if (_delivered.ContainsKey(id)) return false; + _delivered[id] = now; + return true; + } } diff --git a/src/dapps/dapps.meshcore/MeshCoreReliability.cs b/src/dapps/dapps.meshcore/MeshCoreReliability.cs new file mode 100644 index 0000000..a15c571 --- /dev/null +++ b/src/dapps/dapps.meshcore/MeshCoreReliability.cs @@ -0,0 +1,132 @@ +using dapps.client.Backhaul; + +namespace dapps.meshcore; + +/// +/// End-to-end reliability for the MeshCore bearer (#26). Channel messages are +/// fire-and-forget floods with no link-layer ACK and significant loss, so DAPPS +/// adds its own: the receiver ACKs any data message addressed to it (a small +/// control message carrying the acked id in the ), and the +/// sender tracks each unacked message and resends it on an exponential backoff +/// until it is acked or its lifetime expires. +/// +/// This manager is pure bookkeeping (no I/O); the bearer drives it: it calls +/// on send, when an ACK arrives, and a +/// loop polls / . Thread-safe. +/// +public sealed class MeshCoreReliability +{ + /// Header key carrying the acked message id. Its presence marks a + /// message as an ACK (control), not app data. + public const string AckHeader = "mc-ack"; + + public sealed record Options(TimeSpan BaseBackoff, double Multiplier, TimeSpan MaxBackoff, TimeSpan MaxLifetime) + { + public static Options Default => new(TimeSpan.FromSeconds(20), 1.6, TimeSpan.FromSeconds(120), TimeSpan.FromMinutes(5)); + } + + private sealed class Pending + { + public required BackhaulMessage Message; + public required string LocalCallsign; + public DateTime DeadlineUtc; + public DateTime NextResendUtc; + public int Attempts; + } + + private readonly Options _opts; + private readonly Dictionary _pending = new(StringComparer.Ordinal); + private readonly object _lock = new(); + + public long Confirmed { get; private set; } + public long Expired { get; private set; } + public int PendingCount { get { lock (_lock) return _pending.Count; } } + + public MeshCoreReliability(Options? opts = null) => _opts = opts ?? Options.Default; + + public static bool IsAck(BackhaulMessage m) => m.Headers is not null && m.Headers.ContainsKey(AckHeader); + + public static string? AckedId(BackhaulMessage m) => + m.Headers is not null && m.Headers.TryGetValue(AckHeader, out var v) ? v : null; + + /// Build the ACK control message for a received data message. + public static BackhaulMessage BuildAck(BackhaulMessage received, string localCallsign, string ackId) => new( + Id: ackId, + Destination: received.Originator ?? received.LinkSourceCallsign ?? "MESHCORE", + Salt: null, + Ttl: 60, + Payload: [], + Originator: localCallsign, + LinkSourceCallsign: localCallsign, + Headers: new Dictionary { [AckHeader] = received.Id }); + + /// Register a sent data message as awaiting an ACK (no-op for ACKs). + public void Track(BackhaulMessage m, string localCallsign, DateTime now) + { + if (IsAck(m)) return; + var lifetime = _opts.MaxLifetime; + if (m.Ttl is int t and > 0) + lifetime = TimeSpan.FromSeconds(Math.Min(t, _opts.MaxLifetime.TotalSeconds)); + lock (_lock) + { + _pending[m.Id] = new Pending + { + Message = m, + LocalCallsign = localCallsign, + DeadlineUtc = now + lifetime, + NextResendUtc = now + _opts.BaseBackoff, + Attempts = 0, + }; + } + } + + /// Mark a message confirmed delivered. Returns true if it was pending. + public bool OnAck(string ackedId) + { + lock (_lock) + { + if (_pending.Remove(ackedId)) { Confirmed++; return true; } + return false; + } + } + + /// Messages currently due for a resend. Does NOT advance backoff — call + /// only after a resend actually goes on air, so a refused + /// resend (congestion / budget) doesn't burn a retransmit slot. + public List<(BackhaulMessage message, string localCallsign)> DueResends(DateTime now) + { + var due = new List<(BackhaulMessage, string)>(); + lock (_lock) + foreach (var p in _pending.Values) + if (now >= p.NextResendUtc && now < p.DeadlineUtc) + due.Add((p.Message, p.LocalCallsign)); + return due; + } + + /// Record that a message was actually re-sent: advance its attempt count + /// and next-resend time (exponential backoff). + public void MarkResent(string id, DateTime now) + { + lock (_lock) + { + if (!_pending.TryGetValue(id, out var p)) return; + p.Attempts++; + var backoffMs = Math.Min( + _opts.BaseBackoff.TotalMilliseconds * Math.Pow(_opts.Multiplier, p.Attempts), + _opts.MaxBackoff.TotalMilliseconds); + p.NextResendUtc = now + TimeSpan.FromMilliseconds(backoffMs); + } + } + + /// Remove and return messages whose lifetime elapsed unacked (gave up). + public List DropExpired(DateTime now) + { + lock (_lock) + { + var expired = _pending.Where(kv => now >= kv.Value.DeadlineUtc).ToList(); + foreach (var e in expired) _pending.Remove(e.Key); + Expired += expired.Count; + return expired.Select(e => e.Value.Message).ToList(); + } + } +} diff --git a/src/dapps/dapps.meshcore/README.md b/src/dapps/dapps.meshcore/README.md index c286abf..70e7940 100644 --- a/src/dapps/dapps.meshcore/README.md +++ b/src/dapps/dapps.meshcore/README.md @@ -21,6 +21,12 @@ evidence this is built on. `LOG_RX_DATA` (0x88) overheard-packet events, then does listen-before-talk and refuses sends when the channel is congested (a *dynamic* good-citizen control on top of the static budget). A per-node threshold jitter keeps two contending nodes from backing off in lockstep. +- **End-to-end reliability** (`MeshCoreReliability`, #26) — the channel is a fire-and-forget flood + with no link ACK, so DAPPS adds its own, **datagram-style, not session-based**: the receiver ACKs + any data message addressed to it (a tiny `mc-ack` control broadcast), the sender resends unacked + messages on exponential backoff until acked or their lifetime expires, and the receiver **dedups by + message id** so a resend after a lost ACK is delivered to the app only once (idempotent). Resends + and ACKs are ordinary channel traffic, subject to the governor + adaptive controls. - **Broadcast semantics** — a private channel is one shared medium, so a message is broadcast once and the addressee self-selects (the inbox `IsLocal` gate). Identical message ids offered for multiple neighbours are coalesced within a window so we don't re-broadcast. @@ -51,6 +57,7 @@ Configure via `DAPPS_MESHCORE_*` env vars (or the `systemoptions` table): | `DAPPS_MESHCORE_COMPRESS` | `true` | zstd-dict compression | | `DAPPS_MESHCORE_CONGESTION_BACKOFF_FRACTION` | `0.5` | adaptive: refuse sends when channel occupancy ≥ this (0 disables) | | `DAPPS_MESHCORE_LBT_GUARD_MS` | `400` | adaptive: listen-before-talk guard in ms (0 disables) | +| `DAPPS_MESHCORE_RELIABLE_DELIVERY` | `true` | end-to-end ACK + resend of unacked messages (#26) | Inbound is fully wired: received messages are decoded and delivered to `IBackhaulInbox` (DB + MQTT), sender derived from the in-band `LinkSourceCallsign`. Outbound is selected for routes carrying a diff --git a/src/dapps/dapps.meshcore/TxBudget.cs b/src/dapps/dapps.meshcore/TxBudget.cs index b0c87d7..906e503 100644 --- a/src/dapps/dapps.meshcore/TxBudget.cs +++ b/src/dapps/dapps.meshcore/TxBudget.cs @@ -12,9 +12,10 @@ public sealed class TxBudget public const double DefaultSecondsPerHour = 30; // ≈0.83% duty private readonly double _budgetMs; - private readonly List<(DateTime when, double ms)> _window = new(); + private readonly List<(long token, DateTime when, double ms)> _window = new(); private readonly object _lock = new(); private double _sumMs; + private long _nextToken; public TxBudget(double secondsPerHour) => _budgetMs = secondsPerHour * 1000.0; @@ -30,9 +31,10 @@ public double DutyPercent(DateTime now) lock (_lock) { Prune(now); return _sumMs / 36_000.0; } } - /// Reserve airtime for one transmission; returns false (changes - /// nothing) if it would exceed the trailing-hour budget. - public bool TryReserve(double airtimeMs, DateTime now, out string reason) + /// Reserve airtime for one transmission. Returns false (changing nothing) + /// if it would exceed the trailing-hour budget; otherwise + /// identifies the reservation for a later . + public bool TryReserve(double airtimeMs, DateTime now, out string reason, out long token) { lock (_lock) { @@ -40,26 +42,34 @@ public bool TryReserve(double airtimeMs, DateTime now, out string reason) if (_sumMs + airtimeMs > _budgetMs) { reason = $"airtime budget exceeded: used {_sumMs / 1000:0.0}s + {airtimeMs / 1000:0.00}s > {_budgetMs / 1000:0.0}s/hr"; + token = 0; return false; } - _window.Add((now, airtimeMs)); + token = ++_nextToken; + _window.Add((token, now, airtimeMs)); _sumMs += airtimeMs; reason = ""; return true; } } - /// Return the most recent reservation to the budget — called when a - /// send that reserved airtime did not actually go on air (link not ready / - /// exception), so a failed attempt doesn't consume the duty budget. - public void Refund() + /// Convenience overload for callers that never refund (e.g. tests). + public bool TryReserve(double airtimeMs, DateTime now, out string reason) + => TryReserve(airtimeMs, now, out reason, out _); + + /// Return a specific reservation (by ) — called + /// when a send that reserved airtime did not go on air. Concurrency-safe: removes + /// exactly that reservation, not merely the most recent (which, under concurrent + /// sends, could belong to a different in-flight send). + public void Refund(long token) { lock (_lock) { - if (_window.Count > 0) + var i = _window.FindIndex(e => e.token == token); + if (i >= 0) { - _sumMs -= _window[^1].ms; - _window.RemoveAt(_window.Count - 1); + _sumMs -= _window[i].ms; + _window.RemoveAt(i); } if (_window.Count == 0 || _sumMs < 0) _sumMs = 0; }