MeshCore bearer: end-to-end reliability over the flood (#26)

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<InvalidDataException>();
     }
 
+    [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()
     {

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();
     }
 }

src/dapps/dapps.core/Models/SystemOptions.cs

@@ -129,6 +129,10 @@ public class SystemOptions
     /// <summary>Listen-before-talk guard in ms (#157). 0 disables.</summary>
     public int MeshCoreLbtGuardMs { get; set; } = 400;
 
+    /// <summary>End-to-end reliability (#26): ACK received messages + resend our own
+    /// unacked messages until acked or their lifetime expires.</summary>
+    public bool MeshCoreReliableDelivery { get; set; } = true;
+
     /// <summary>
     /// When true, app-interface clients (MQTT and REST) must present a
     /// valid token; topic / endpoint scope is also enforced against the

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

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<MeshCoreLink>());
 
         try
@@ -70,12 +71,45 @@ public async Task RunAsync(CancellationToken ct)
             return;
         }
 
-        _backhaul = new MeshCoreCompanionBackhaul(
-            _link, opts, budget, _loggerFactory.CreateLogger<MeshCoreCompanionBackhaul>(), _txGate);
-        _inbound = new MeshCoreInbound(_link, _inbox, _loggerFactory.CreateLogger<MeshCoreInbound>());
+        var backhaul = new MeshCoreCompanionBackhaul(
+            _link, opts, budget, _loggerFactory.CreateLogger<MeshCoreCompanionBackhaul>(), _txGate, reliability);
+        _backhaul = backhaul;
+        _inbound = new MeshCoreInbound(
+            _link, _inbox, _loggerFactory.CreateLogger<MeshCoreInbound>(),
+            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<BackhaulSendResult> SendAsync(
         Compress = s.MeshCoreCompress,
         CongestionBackoffFraction = s.MeshCoreCongestionBackoffFraction,
         LbtGuardMs = s.MeshCoreLbtGuardMs,
+        ReliableDelivery = s.MeshCoreReliableDelivery,
+        LocalCallsign = s.Callsign,
         AppName = "dapps",
     };
 

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<string, string> 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),
         };
     }
 

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<MeshCoreCompanionBackhaul>());
-var inbound = new MeshCoreInbound(link, inbox, lf.CreateLogger<MeshCoreInbound>());
+var reliability = opts.ReliableDelivery ? new MeshCoreReliability() : null;
+var backhaul = new MeshCoreCompanionBackhaul(
+    link, opts, budget, lf.CreateLogger<MeshCoreCompanionBackhaul>(), reliability: reliability);
+
+// Optional induced loss to exercise reliability resends (soak only).
+double dropPct = a.GetDouble("drop-pct", 0);
+Func<BackhaulMessage, bool>? drop = dropPct > 0 ? (_ => Random.Shared.NextDouble() * 100 < dropPct) : null;
+
+var inbound = new MeshCoreInbound(
+    link, inbox, lf.CreateLogger<MeshCoreInbound>(),
+    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 ----------------

src/dapps/dapps.meshcore/MeshCoreBearerOptions.cs

@@ -29,6 +29,14 @@ public sealed class MeshCoreBearerOptions
     /// to avoid colliding with an in-progress flood. 0 disables.</summary>
     public int LbtGuardMs { get; set; } = 400;
 
+    /// <summary>End-to-end reliability (#26): ACK received messages addressed to us
+    /// and resend our own unacked messages until acked or their lifetime expires.</summary>
+    public bool ReliableDelivery { get; set; } = true;
+
+    /// <summary>This node's DAPPS callsign — decides which received messages to ACK
+    /// (those addressed to us) and is the ACK originator.</summary>
+    public string LocalCallsign { get; set; } = "";
+
     public RegionPreset ResolveRegion() =>
         Regions.Find(Region) ?? throw new ArgumentException($"unknown MeshCore region '{Region}'");
 

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<string, bool> _compressed = new();
+    private readonly Dictionary<string, (bool comp, DateTime seen)> _compressed = new();
+    private readonly object _nonceLock = new();
     private byte _nonce;
 
     /// <summary>Encode a BackhaulMessage into one-or-more channel-data payloads.</summary>
@@ -35,8 +36,14 @@ public IReadOnlyList<byte[]> ToFrames(BackhaulMessage message, DappsCompression.
         var frames = new List<byte[]>(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)
         }
     }
 
-    /// <summary>Drop reassembly state for messages whose first fragment is older
-    /// than <paramref name="cutoff"/>.</summary>
-    public int DropStale(DateTime cutoff) => _reassembler.DropOlderThan(cutoff);
+    /// <summary>Drop reassembly state (and the matching compressed-flag entries) for
+    /// messages whose first fragment is older than <paramref name="cutoff"/>.</summary>
+    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);
+    }
 }