From ed3e86349954bf83cfe2f078f04e100e3679b434 Mon Sep 17 00:00:00 2001 From: Tom M0LTE <37816024+M0LTE@users.noreply.github.com> Date: Wed, 1 Jul 2026 00:21:07 +0000 Subject: [PATCH] MeshCore bearer: neighbour routing + adaptive airtime + review fixes (#155, #157) #155 - route a configured neighbour over MeshCore: additive DbNeighbour.MeshCoreChannel copied through RouteBuilder, so a neighbour with a channel hint is selected by the bearer. #157 - adaptive airtime + channel monitoring: ChannelMonitor estimates channel occupancy from LOG_RX_DATA (0x88) overheard-packet events; the bearer does listen-before-talk and refuses sends when the channel is congested (a dynamic good-citizen control on top of the static budget), with per-node threshold jitter for fairness. Configurable via DAPPS_MESHCORE_CONGESTION_BACKOFF_FRACTION / _LBT_GUARD_MS. Validated on air (backoff triggered at ~21% occupancy). Fixes from an adversarial review of the bearer: - CRITICAL: SystemOptionsStore.Parse/SaveAsync never read/wrote the MeshCore* keys, so the bearer could not be enabled or configured via the persisted table. Both directions fixed; round-trip regression test added. - MeshCoreInbound: cancel the losing wait so idle iterations don't leak semaphore waiters. - MeshCoreLink: dispose the client if configuration fails after Open (SerialPort/read-loop leak); swallow a dispose-race on send. - MeshCoreFrames: length-guard every parser; DrainAsync skips a malformed frame instead of dropping the whole batch. - TxBudget.Refund: return airtime reserved for a send that didn't go on air. - OutboundMessageManager: audit-label MeshCore forwards as "meshcore". Tests: 17 MeshCore unit tests; full solution builds clean. On-air soak (two Heltec V3s) healthy incl. watchdog reset+recovery and adaptive backoff. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01KLbwvhE2cKCe8WPZNg8k17 --- .../dapps.core.tests/MeshCoreBearerTests.cs | 50 +++++++++++++ .../dapps.core.tests/MeshCoreConfigTests.cs | 72 +++++++++++++++++++ src/dapps/dapps.core/Models/DbRouteHint.cs | 8 +++ src/dapps/dapps.core/Models/SystemOptions.cs | 7 ++ src/dapps/dapps.core/Routing/RouteBuilder.cs | 3 +- src/dapps/dapps.core/Services/DbStartup.cs | 2 + .../dapps.core/Services/MeshCoreBearer.cs | 2 + .../Services/OutboundMessageManager.cs | 6 +- .../dapps.core/Services/SystemOptionsStore.cs | 37 ++++++++++ src/dapps/dapps.meshcore.soak/Program.cs | 8 ++- src/dapps/dapps.meshcore/ChannelMonitor.cs | 64 +++++++++++++++++ .../dapps.meshcore/MeshCoreBearerOptions.cs | 9 +++ src/dapps/dapps.meshcore/MeshCoreClient.cs | 37 ++++++++-- .../MeshCoreCompanionBackhaul.cs | 39 ++++++++-- src/dapps/dapps.meshcore/MeshCoreFrames.cs | 5 ++ src/dapps/dapps.meshcore/MeshCoreInbound.cs | 11 ++- src/dapps/dapps.meshcore/MeshCoreLink.cs | 41 ++++++++--- src/dapps/dapps.meshcore/README.md | 6 ++ src/dapps/dapps.meshcore/TxBudget.cs | 29 ++++++-- 19 files changed, 406 insertions(+), 30 deletions(-) create mode 100644 src/dapps/dapps.core.tests/MeshCoreConfigTests.cs create mode 100644 src/dapps/dapps.meshcore/ChannelMonitor.cs diff --git a/src/dapps/dapps.core.tests/MeshCoreBearerTests.cs b/src/dapps/dapps.core.tests/MeshCoreBearerTests.cs index 3abfd4e..af173e8 100644 --- a/src/dapps/dapps.core.tests/MeshCoreBearerTests.cs +++ b/src/dapps/dapps.core.tests/MeshCoreBearerTests.cs @@ -2,6 +2,8 @@ using System.Text; using AwesomeAssertions; using dapps.client.Backhaul; +using dapps.core.Models; +using dapps.core.Routing; using dapps.meshcore; using Xunit; @@ -129,6 +131,54 @@ public void SelfInfo_ParsesRadioParams() self.Name.Should().Be("DAPPS-R1"); } + [Fact] + public void ChannelMonitor_OccupancyRisesWithTrafficThenPrunes() + { + var region = Regions.Find("uk-test")!; + var m = new ChannelMonitor(region, TimeSpan.FromSeconds(10)); + var t0 = new DateTime(2026, 1, 1, 12, 0, 0, DateTimeKind.Utc); + + m.OccupancyFraction(t0).Should().Be(0); + m.SinceLastHeard(t0).Should().Be(TimeSpan.MaxValue); + + for (var i = 0; i < 5; i++) m.RecordHeard(150, t0.AddMilliseconds(i * 10)); + m.HeardCount.Should().Be(5); + m.OccupancyFraction(t0.AddSeconds(1)).Should().BeGreaterThan(0); + m.SinceLastHeard(t0.AddSeconds(1)).Should().BeLessThan(TimeSpan.FromSeconds(2)); + + // Once the heard packets age past the window, occupancy returns to zero. + m.OccupancyFraction(t0.AddSeconds(30)).Should().Be(0); + } + + [Fact] + public void TxBudget_Refund_ReturnsTheLastReservation() + { + 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.UsedSeconds(now).Should().BeApproximately(0, 0.001); + b.TryReserve(900, now, out _).Should().BeTrue("budget was refunded"); + } + + [Fact] + public void ChannelData_ParseRecv_ShortFrame_ThrowsInvalidData() + { + var act = () => ChannelData.ParseRecv([0x1B, 0, 0]); + act.Should().Throw(); + } + + [Fact] + public void RouteBuilder_CopiesMeshCoreChannelHint() + { + var route = RouteBuilder.FromNeighbour( + new DbNeighbour { Callsign = "GB7XYZ-1", MeshCoreChannel = "dapps" }, defaultBearerPort: null); + route.Callsign.Should().Be("GB7XYZ-1"); + route.MeshCoreChannel.Should().Be("dapps"); + } + private static void AssertEqual(BackhaulMessage? got, BackhaulMessage original) { got.Should().NotBeNull(); diff --git a/src/dapps/dapps.core.tests/MeshCoreConfigTests.cs b/src/dapps/dapps.core.tests/MeshCoreConfigTests.cs new file mode 100644 index 0000000..505d9c4 --- /dev/null +++ b/src/dapps/dapps.core.tests/MeshCoreConfigTests.cs @@ -0,0 +1,72 @@ +using AwesomeAssertions; +using dapps.core.Models; +using dapps.core.Services; +using Microsoft.Extensions.Logging.Abstractions; +using SQLite; +using Xunit; + +namespace dapps.core.tests; + +/// +/// Guards the MeshCore config round-trip through +/// (#154 review): every DAPPS_MESHCORE_* option must survive Save -> reload, so +/// the bearer can actually be enabled and configured via the persisted table +/// (regression guard for the earlier gap where Parse/SaveAsync ignored them). +/// +[Collection(SqliteOverridePathCollection.Name)] +public sealed class MeshCoreConfigTests : IAsyncLifetime +{ + private string dbPath = null!; + + public ValueTask InitializeAsync() + { + dbPath = Path.Combine(Path.GetTempPath(), $"dapps-mc-{Guid.NewGuid():N}.db"); + DbInfo.OverridePath = dbPath; + using var c = new SQLiteConnection(dbPath); + c.CreateTable(); + return ValueTask.CompletedTask; + } + + public ValueTask DisposeAsync() + { + DbInfo.OverridePath = null; + try { File.Delete(dbPath); } catch { /* ignore */ } + return ValueTask.CompletedTask; + } + + [Fact] + public async Task MeshCoreOptions_RoundTripThroughStore() + { + var store = new SystemOptionsStore(NullLogger.Instance); + var opts = store.CurrentValue; + opts.MeshCoreEnabled = true; + opts.MeshCorePort = "/dev/ttyUSB9"; + opts.MeshCoreRegion = "uk-narrow"; + opts.MeshCoreChannelIndex = 3; + opts.MeshCoreChannelName = "testch"; + opts.MeshCoreChannelPsk = "3135135fd198029d689b64f45df2aae9"; + opts.MeshCoreNodeName = "GB7TST-1"; + opts.MeshCoreTxPowerDbm = 14; + opts.MeshCoreAirtimeBudgetSecondsPerHour = 45; + opts.MeshCoreCompress = false; + opts.MeshCoreCongestionBackoffFraction = 0.25; + opts.MeshCoreLbtGuardMs = 250; + + await store.SaveAsync(opts); + + // A fresh store reads the persisted rows via Parse. + var reloaded = new SystemOptionsStore(NullLogger.Instance).CurrentValue; + reloaded.MeshCoreEnabled.Should().BeTrue(); + reloaded.MeshCorePort.Should().Be("/dev/ttyUSB9"); + reloaded.MeshCoreRegion.Should().Be("uk-narrow"); + reloaded.MeshCoreChannelIndex.Should().Be(3); + reloaded.MeshCoreChannelName.Should().Be("testch"); + reloaded.MeshCoreChannelPsk.Should().Be("3135135fd198029d689b64f45df2aae9"); + reloaded.MeshCoreNodeName.Should().Be("GB7TST-1"); + reloaded.MeshCoreTxPowerDbm.Should().Be(14); + reloaded.MeshCoreAirtimeBudgetSecondsPerHour.Should().Be(45); + reloaded.MeshCoreCompress.Should().BeFalse(); + reloaded.MeshCoreCongestionBackoffFraction.Should().Be(0.25); + reloaded.MeshCoreLbtGuardMs.Should().Be(250); + } +} diff --git a/src/dapps/dapps.core/Models/DbRouteHint.cs b/src/dapps/dapps.core/Models/DbRouteHint.cs index 604099c..446c44b 100644 --- a/src/dapps/dapps.core/Models/DbRouteHint.cs +++ b/src/dapps/dapps.core/Models/DbRouteHint.cs @@ -43,4 +43,12 @@ public class DbNeighbour /// direct connection (the usual case). See . /// public string? ConnectScriptJson { get; set; } + + /// + /// Optional MeshCore channel name. When set, this neighbour is reachable over + /// the MeshCore bearer (#154): the backhaul broadcasts on the configured + /// private channel and this neighbour self-selects by destination callsign. + /// Null = not a MeshCore neighbour. (sqlite-net adds this column on upgrade.) + /// + public string? MeshCoreChannel { get; set; } } diff --git a/src/dapps/dapps.core/Models/SystemOptions.cs b/src/dapps/dapps.core/Models/SystemOptions.cs index a4c17ba..3f3b7b6 100644 --- a/src/dapps/dapps.core/Models/SystemOptions.cs +++ b/src/dapps/dapps.core/Models/SystemOptions.cs @@ -122,6 +122,13 @@ public class SystemOptions /// Compress the backhaul payload (zstd + shared dictionary). public bool MeshCoreCompress { get; set; } = true; + /// Adaptive congestion backoff (#157): refuse sends when channel + /// occupancy is at/above this fraction (0..1). 0 disables. + public double MeshCoreCongestionBackoffFraction { get; set; } = 0.5; + + /// Listen-before-talk guard in ms (#157). 0 disables. + public int MeshCoreLbtGuardMs { get; set; } = 400; + /// /// 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/Routing/RouteBuilder.cs b/src/dapps/dapps.core/Routing/RouteBuilder.cs index 3544289..40a015f 100644 --- a/src/dapps/dapps.core/Routing/RouteBuilder.cs +++ b/src/dapps/dapps.core/Routing/RouteBuilder.cs @@ -18,5 +18,6 @@ public static BackhaulRoute FromNeighbour(DbNeighbour neighbour, int? defaultBea Callsign: neighbour.Callsign, BearerPort: neighbour.BearerPort ?? defaultBearerPort, UdpEndpoint: neighbour.UdpEndpoint, - ConnectScript: ConnectScript.FromJson(neighbour.ConnectScriptJson)); + ConnectScript: ConnectScript.FromJson(neighbour.ConnectScriptJson), + MeshCoreChannel: neighbour.MeshCoreChannel); } diff --git a/src/dapps/dapps.core/Services/DbStartup.cs b/src/dapps/dapps.core/Services/DbStartup.cs index 6722f00..4f7ce1e 100644 --- a/src/dapps/dapps.core/Services/DbStartup.cs +++ b/src/dapps/dapps.core/Services/DbStartup.cs @@ -101,6 +101,8 @@ private static readonly (string Key, string Default)[] SeededOptions = ("MeshCoreNodeName", "DAPPS"), ("MeshCoreAirtimeBudgetSecondsPerHour", "30"), ("MeshCoreCompress", "true"), + ("MeshCoreCongestionBackoffFraction", "0.5"), + ("MeshCoreLbtGuardMs", "400"), ("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 aba9913..b57cc26 100644 --- a/src/dapps/dapps.core/Services/MeshCoreBearer.cs +++ b/src/dapps/dapps.core/Services/MeshCoreBearer.cs @@ -98,6 +98,8 @@ public Task SendAsync( NodeName = s.MeshCoreNodeName, AirtimeBudgetSecPerHour = s.MeshCoreAirtimeBudgetSecondsPerHour, Compress = s.MeshCoreCompress, + CongestionBackoffFraction = s.MeshCoreCongestionBackoffFraction, + LbtGuardMs = s.MeshCoreLbtGuardMs, AppName = "dapps", }; diff --git a/src/dapps/dapps.core/Services/OutboundMessageManager.cs b/src/dapps/dapps.core/Services/OutboundMessageManager.cs index bf0e6ff..0433f43 100644 --- a/src/dapps/dapps.core/Services/OutboundMessageManager.cs +++ b/src/dapps/dapps.core/Services/OutboundMessageManager.cs @@ -180,7 +180,8 @@ private async Task ForwardAndObserveAsync( { await ta.RecordAsync( kind: "forward", - bearer: route.UdpEndpoint is not null ? "udp" : "agw", + bearer: route.MeshCoreChannel is not null ? "meshcore" + : route.UdpEndpoint is not null ? "udp" : "agw", channelKey: route.BearerPort?.ToString() ?? "", targetCallsign: route.Callsign, messageId: message.Id, @@ -243,7 +244,8 @@ private async Task FloodAndMarkAsync( { await ta.RecordAsync( kind: "forward-flood", - bearer: route.UdpEndpoint is not null ? "udp" : "agw", + bearer: route.MeshCoreChannel is not null ? "meshcore" + : route.UdpEndpoint is not null ? "udp" : "agw", channelKey: route.BearerPort?.ToString() ?? "", targetCallsign: route.Callsign, messageId: message.Id, diff --git a/src/dapps/dapps.core/Services/SystemOptionsStore.cs b/src/dapps/dapps.core/Services/SystemOptionsStore.cs index aec7b7a..01fb0e4 100644 --- a/src/dapps/dapps.core/Services/SystemOptionsStore.cs +++ b/src/dapps/dapps.core/Services/SystemOptionsStore.cs @@ -1,3 +1,4 @@ +using System.Globalization; using dapps.client; using dapps.core.Models; using Microsoft.Extensions.Options; @@ -105,6 +106,18 @@ public async Task SaveAsync(SystemOptions options) await Upsert(connection, existing, nameof(options.TransmissionAuditRetentionDays), options.TransmissionAuditRetentionDays.ToString()); await Upsert(connection, existing, nameof(options.TransmissionAuditMqttPublish), options.TransmissionAuditMqttPublish.ToString()); await Upsert(connection, existing, nameof(options.TxEnabled), options.TxEnabled.ToString()); + await Upsert(connection, existing, nameof(options.MeshCoreEnabled), options.MeshCoreEnabled.ToString()); + await Upsert(connection, existing, nameof(options.MeshCorePort), options.MeshCorePort); + await Upsert(connection, existing, nameof(options.MeshCoreRegion), options.MeshCoreRegion); + await Upsert(connection, existing, nameof(options.MeshCoreTxPowerDbm), options.MeshCoreTxPowerDbm.ToString()); + await Upsert(connection, existing, nameof(options.MeshCoreChannelIndex), options.MeshCoreChannelIndex.ToString()); + await Upsert(connection, existing, nameof(options.MeshCoreChannelName), options.MeshCoreChannelName); + await Upsert(connection, existing, nameof(options.MeshCoreChannelPsk), options.MeshCoreChannelPsk); + await Upsert(connection, existing, nameof(options.MeshCoreNodeName), options.MeshCoreNodeName); + await Upsert(connection, existing, nameof(options.MeshCoreAirtimeBudgetSecondsPerHour), options.MeshCoreAirtimeBudgetSecondsPerHour.ToString(CultureInfo.InvariantCulture)); + 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()); Reload(); } @@ -173,6 +186,18 @@ private static SystemOptions Parse(Dictionary r) TransmissionAuditRetentionDays = TryGetInt(r, nameof(SystemOptions.TransmissionAuditRetentionDays), 90, min: 0), TransmissionAuditMqttPublish = TryGetBool(r, nameof(SystemOptions.TransmissionAuditMqttPublish), false), TxEnabled = TryGetBool(r, nameof(SystemOptions.TxEnabled), true), + MeshCoreEnabled = TryGetBool(r, nameof(SystemOptions.MeshCoreEnabled), false), + MeshCorePort = TryGet(r, nameof(SystemOptions.MeshCorePort), "/dev/ttyUSB0"), + MeshCoreRegion = TryGet(r, nameof(SystemOptions.MeshCoreRegion), "uk-test"), + MeshCoreTxPowerDbm = TryGetInt(r, nameof(SystemOptions.MeshCoreTxPowerDbm), 8, min: 0, max: 30), + MeshCoreChannelIndex = TryGetInt(r, nameof(SystemOptions.MeshCoreChannelIndex), 1, min: 0, max: 255), + MeshCoreChannelName = TryGet(r, nameof(SystemOptions.MeshCoreChannelName), "dapps"), + MeshCoreChannelPsk = TryGet(r, nameof(SystemOptions.MeshCoreChannelPsk), "dapps-default-channel"), + MeshCoreNodeName = TryGet(r, nameof(SystemOptions.MeshCoreNodeName), "DAPPS"), + MeshCoreAirtimeBudgetSecondsPerHour = TryGetDouble(r, nameof(SystemOptions.MeshCoreAirtimeBudgetSecondsPerHour), 30, min: 0), + 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), }; } @@ -193,6 +218,18 @@ private static int TryGetInt(Dictionary r, string key, int fallb private static bool TryGetBool(Dictionary r, string key, bool fallback) => r.TryGetValue(key, out var s) && bool.TryParse(s, out var v) ? v : fallback; + private static double TryGetDouble(Dictionary r, string key, double fallback, double? min = null, double? max = null) + { + if (r.TryGetValue(key, out var s) + && double.TryParse(s, NumberStyles.Any, CultureInfo.InvariantCulture, out var v)) + { + if (min is { } lo && v < lo) return fallback; + if (max is { } hi && v > hi) return fallback; + return v; + } + return fallback; + } + private sealed class Subscription : IDisposable { private readonly SystemOptionsStore store; diff --git a/src/dapps/dapps.meshcore.soak/Program.cs b/src/dapps/dapps.meshcore.soak/Program.cs index 862b487..c6e6c5b 100644 --- a/src/dapps/dapps.meshcore.soak/Program.cs +++ b/src/dapps/dapps.meshcore.soak/Program.cs @@ -32,6 +32,8 @@ NodeName = self, AirtimeBudgetSecPerHour = a.GetDouble("budget", 120), Compress = !a.Has("no-compress"), + CongestionBackoffFraction = a.GetDouble("congestion", 0.5), + LbtGuardMs = a.GetInt("lbt", 400), AppName = "dapps-soak", }; @@ -59,7 +61,7 @@ var inboundTask = Task.Run(() => inbound.RunAsync(cts.Token)); -long sent = 0, accepted = 0, throttled = 0, failed = 0; +long sent = 0, accepted = 0, throttled = 0, backedOff = 0, failed = 0; string[] samples = [ "73 de " + self, "QSL 73 GL", "GM all, nice signal this morning, 599 here", @@ -83,6 +85,7 @@ Interlocked.Increment(ref sent); if (r.Accepted) Interlocked.Increment(ref accepted); else if (r.Error?.Contains("budget") == true) { Interlocked.Increment(ref throttled); log.LogWarning("TX seq={0} throttled: {1}", seq, r.Error); } + else if (r.Error?.Contains("congested") == true) { Interlocked.Increment(ref backedOff); log.LogWarning("TX seq={0} backoff: {1}", seq, r.Error); } else { Interlocked.Increment(ref failed); log.LogWarning("TX seq={0} failed: {1}", seq, r.Error); } } catch (Exception ex) { Interlocked.Increment(ref failed); log.LogWarning("TX seq={0} exception: {1}", seq, ex.Message); } @@ -110,8 +113,9 @@ var (recv, maxSeq, distinct) = inbox.Snapshot(); double lossPct = maxSeq >= 0 ? 100.0 * (1.0 - (double)distinct / (maxSeq + 1)) : 0; log.LogInformation("================= SOAK SUMMARY ({0}) =================", self); -log.LogInformation("TX: offered={0} accepted={1} throttled={2} failed={3}", sent, accepted, throttled, failed); +log.LogInformation("TX: offered={0} accepted={1} throttled={2} backoff={3} failed={4}", sent, accepted, throttled, backedOff, failed); log.LogInformation("RX: delivered={0} distinctSeq={1} maxSeqFromPeer={2} loss={3:0.0}%", recv, distinct, maxSeq, lossPct); +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); return 0; diff --git a/src/dapps/dapps.meshcore/ChannelMonitor.cs b/src/dapps/dapps.meshcore/ChannelMonitor.cs new file mode 100644 index 0000000..fb46376 --- /dev/null +++ b/src/dapps/dapps.meshcore/ChannelMonitor.cs @@ -0,0 +1,64 @@ +namespace dapps.meshcore; + +/// +/// Estimates how busy the shared LoRa channel is from the radio's LOG_RX_DATA +/// (0x88) "packet heard" events (#157). Every packet the radio overhears — our +/// peers' floods, other same-preset traffic — is recorded with its estimated +/// airtime; occupancy is the busy fraction over a trailing window. The bearer +/// uses this to be a *dynamically* good citizen: listen-before-talk and back off +/// when the channel is congested, on top of the static airtime budget. +/// +public sealed class ChannelMonitor +{ + private readonly RegionPreset _region; + private readonly TimeSpan _window; + private readonly Queue<(DateTime when, double airMs)> _heard = new(); + private readonly object _lock = new(); + private double _sumMs; + + public DateTime LastHeardUtc { get; private set; } = DateTime.MinValue; + public long HeardCount { get; private set; } + + public ChannelMonitor(RegionPreset region, TimeSpan? window = null) + { + _region = region; + _window = window ?? TimeSpan.FromSeconds(60); + } + + /// Record a packet overheard on the channel. + /// is the logged on-air length; its airtime is estimated for the active preset. + public void RecordHeard(int rawLen, DateTime now) + { + var airMs = LoRaAirtime.Ms(Math.Max(rawLen, 1), _region.Sf, _region.BwKhz * 1000, _region.Cr); + lock (_lock) + { + Prune(now); + _heard.Enqueue((now, airMs)); + _sumMs += airMs; + LastHeardUtc = now; + HeardCount++; + } + } + + /// Busy fraction (0..1) over the trailing window. + public double OccupancyFraction(DateTime now) + { + lock (_lock) + { + Prune(now); + return Math.Min(1.0, _sumMs / _window.TotalMilliseconds); + } + } + + public TimeSpan SinceLastHeard(DateTime now) => + LastHeardUtc == DateTime.MinValue ? TimeSpan.MaxValue : now - LastHeardUtc; + + private void Prune(DateTime now) + { + var cutoff = now - _window; + while (_heard.Count > 0 && _heard.Peek().when < cutoff) + _sumMs -= _heard.Dequeue().airMs; + // Reset exactly when the window empties to avoid float residual drift. + if (_heard.Count == 0 || _sumMs < 0) _sumMs = 0; + } +} diff --git a/src/dapps/dapps.meshcore/MeshCoreBearerOptions.cs b/src/dapps/dapps.meshcore/MeshCoreBearerOptions.cs index 5fd6b22..ffeb9c2 100644 --- a/src/dapps/dapps.meshcore/MeshCoreBearerOptions.cs +++ b/src/dapps/dapps.meshcore/MeshCoreBearerOptions.cs @@ -20,6 +20,15 @@ public sealed class MeshCoreBearerOptions public bool Compress { get; set; } = true; public string AppName { get; set; } = "dapps"; + /// Adaptive congestion backoff (#157): refuse sends when the channel's + /// trailing-window occupancy is at or above this fraction (0..1). 0 disables. + public double CongestionBackoffFraction { get; set; } = 0.5; + + /// Listen-before-talk guard (ms): if a packet was overheard more + /// recently than this, wait out the remainder (plus jitter) before transmitting, + /// to avoid colliding with an in-progress flood. 0 disables. + public int LbtGuardMs { get; set; } = 400; + public RegionPreset ResolveRegion() => Regions.Find(Region) ?? throw new ArgumentException($"unknown MeshCore region '{Region}'"); diff --git a/src/dapps/dapps.meshcore/MeshCoreClient.cs b/src/dapps/dapps.meshcore/MeshCoreClient.cs index 05daa6a..6c51b13 100644 --- a/src/dapps/dapps.meshcore/MeshCoreClient.cs +++ b/src/dapps/dapps.meshcore/MeshCoreClient.cs @@ -44,6 +44,7 @@ public sealed class MeshCoreClient : IAsyncDisposable // push codes (device → host, async) public const byte PUSH_SEND_CONFIRMED = 0x82; public const byte PUSH_MSG_WAITING = 0x83; + public const byte PUSH_LOG_RX_DATA = 0x88; // a packet was heard on the channel private const byte FrameToRadio = 0x3C; private const byte FrameFromRadio = 0x3E; @@ -64,6 +65,11 @@ public sealed class MeshCoreClient : IAsyncDisposable /// Raised when the device signals queued inbound messages (0x83). public event Action? MessageWaiting; + /// Raised for every packet the radio overhears on the channel + /// (LOG_RX_DATA 0x88) — args are (logged length, snr*4). Feeds channel- + /// occupancy estimation (#157). + public event Action? PacketHeard; + public MeshCoreClient(string portName, int baud = 115200) { _port = new SerialPort(portName, baud, Parity.None, 8, StopBits.One) @@ -114,7 +120,18 @@ private async Task ReadLoopAsync(CancellationToken ct) private void HandlePush(byte code, byte[] payload) { - if (code == PUSH_MSG_WAITING) MessageWaiting?.Invoke(); + switch (code) + { + case PUSH_MSG_WAITING: + MessageWaiting?.Invoke(); + break; + case PUSH_LOG_RX_DATA: + // payload is the logged RX record; first byte after the opcode is + // SNR in the RX-frame family. Length is a proxy for on-air size. + var snr = payload.Length > 1 ? unchecked((sbyte)payload[1]) : (sbyte)0; + PacketHeard?.Invoke(payload.Length, snr); + break; + } } /// Send a command frame and await the next synchronous response whose @@ -256,12 +273,20 @@ public async Task DrainAsync(CancellationToken ct) TimeSpan.FromMilliseconds(1500), ct); } catch (TimeoutException) { break; } - switch (resp[0]) + if (resp[0] == RSP_NO_MORE_MESSAGES) return new InboundBatch(texts, data); + try + { + switch (resp[0]) + { + case RSP_CHANNEL_MSG_RECV: texts.Add(ChannelMessage.ParseLegacy(resp)); break; + case RSP_CHANNEL_MSG_RECV_V3: texts.Add(ChannelMessage.ParseV3(resp)); break; + case RSP_CHANNEL_DATA_RECV: data.Add(ChannelData.ParseRecv(resp)); break; + } + } + catch (Exception ex) when (ex is InvalidDataException or IndexOutOfRangeException or ArgumentException) { - case RSP_NO_MORE_MESSAGES: return new InboundBatch(texts, data); - case RSP_CHANNEL_MSG_RECV: texts.Add(ChannelMessage.ParseLegacy(resp)); break; - case RSP_CHANNEL_MSG_RECV_V3: texts.Add(ChannelMessage.ParseV3(resp)); break; - case RSP_CHANNEL_DATA_RECV: data.Add(ChannelData.ParseRecv(resp)); break; + // Skip one malformed inbound frame; keep draining the rest of the queue. + if (Trace) Console.Error.WriteLine($"drain: skipping malformed 0x{resp[0]:X2} frame: {ex.Message}"); } } return new InboundBatch(texts, data); diff --git a/src/dapps/dapps.meshcore/MeshCoreCompanionBackhaul.cs b/src/dapps/dapps.meshcore/MeshCoreCompanionBackhaul.cs index 42d8d67..7028560 100644 --- a/src/dapps/dapps.meshcore/MeshCoreCompanionBackhaul.cs +++ b/src/dapps/dapps.meshcore/MeshCoreCompanionBackhaul.cs @@ -28,6 +28,8 @@ public sealed class MeshCoreCompanionBackhaul : IDappsBackhaul private readonly RegionPreset _region; private readonly ILogger _log; private readonly MeshCoreChannelTransport _tx = new(); + private readonly ChannelMonitor _monitor; + private readonly double _congestionThreshold; private readonly Dictionary _recent = new(); private readonly object _recentLock = new(); @@ -40,8 +42,16 @@ public MeshCoreCompanionBackhaul( _log = log; _txGate = txGate ?? AlwaysOpenTxGate.Instance; _region = opts.ResolveRegion(); + _monitor = new ChannelMonitor(_region); + link.PacketHeard += (len, _) => _monitor.RecordHeard(len, DateTime.UtcNow); + // Per-node jitter (±15%) on the congestion threshold so two contending + // nodes don't back off in lockstep — fairer channel sharing (#157). + _congestionThreshold = Math.Clamp(opts.CongestionBackoffFraction * (0.85 + 0.30 * Random.Shared.NextDouble()), 0, 1); } + /// Trailing-window channel occupancy (0..1), for observability. + public double Occupancy => _monitor.OccupancyFraction(DateTime.UtcNow); + /// Handle routes that carry a MeshCore channel hint (positive /// bearer selection — no reliance on AGW's "no UDP endpoint" catch-all). public bool CanHandle(BackhaulRoute route) => !string.IsNullOrEmpty(route.MeshCoreChannel); @@ -57,6 +67,11 @@ public async Task SendAsync( if (AlreadyBroadcast(message.Id)) return BackhaulSendResult.Ok(); + // Adaptive congestion backoff (#157): don't pile onto a busy shared channel. + var occ = _monitor.OccupancyFraction(DateTime.UtcNow); + if (_opts.CongestionBackoffFraction > 0 && occ >= _congestionThreshold) + return BackhaulSendResult.Fail($"channel congested {occ:P0} (>= {_congestionThreshold:P0}); backing off"); + var stamped = message with { LinkSourceCallsign = localCallsign }; var mode = _opts.Compress ? DappsCompression.Mode.ZstdDict : DappsCompression.Mode.None; var frames = _tx.ToFrames(stamped, mode); @@ -67,20 +82,36 @@ public async Task SendAsync( if (!_budget.TryReserve(airMs, DateTime.UtcNow, out var reason)) 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) { return BackhaulSendResult.Fail($"meshcore send failed: {ex.Message}"); } - if (!sent) return BackhaulSendResult.Fail($"meshcore link not ready ({_link.State})"); + 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})"); } 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}%", - message.Id, frames.Count, message.Destination, localCallsign, _budget.DutyPercent(DateTime.UtcNow)); + _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); return BackhaulSendResult.Ok(); } + /// Listen-before-talk: if a packet was overheard within the guard, + /// wait out the remainder plus a little jitter before transmitting. + private async Task ListenBeforeTalkAsync(CancellationToken ct) + { + if (_opts.LbtGuardMs <= 0) return; + var since = _monitor.SinceLastHeard(DateTime.UtcNow); + var guard = TimeSpan.FromMilliseconds(_opts.LbtGuardMs); + if (since < guard) + { + var wait = guard - since + TimeSpan.FromMilliseconds(Random.Shared.Next(0, 150)); + try { await Task.Delay(wait, ct); } catch { } + } + } + private bool AlreadyBroadcast(string id) { lock (_recentLock) diff --git a/src/dapps/dapps.meshcore/MeshCoreFrames.cs b/src/dapps/dapps.meshcore/MeshCoreFrames.cs index 3247a7c..ef87834 100644 --- a/src/dapps/dapps.meshcore/MeshCoreFrames.cs +++ b/src/dapps/dapps.meshcore/MeshCoreFrames.cs @@ -12,6 +12,7 @@ public sealed record SelfInfo( public static SelfInfo Parse(byte[] p) { + if (p.Length < 58) throw new InvalidDataException($"SELF_INFO frame too short ({p.Length} bytes)"); var pub = p[4..36]; double freq = BinaryPrimitives.ReadUInt32LittleEndian(p.AsSpan(48, 4)) / 1000.0; double bw = BinaryPrimitives.ReadUInt32LittleEndian(p.AsSpan(52, 4)) / 1000.0; @@ -28,6 +29,7 @@ public sealed record ChannelInfo(byte Index, string Name, byte[] Secret) public static ChannelInfo Parse(byte[] p) { + if (p.Length < 50) throw new InvalidDataException($"CHANNEL_INFO frame too short ({p.Length} bytes)"); byte idx = p[1]; // The firmware returns a null-terminated name in a 32-byte field whose // tail is uninitialised; trim at the first null. @@ -47,6 +49,7 @@ public sealed record ChannelMessage( public static ChannelMessage ParseV3(byte[] p) { + if (p.Length < 11) throw new InvalidDataException($"CHANNEL_MSG_RECV_V3 frame too short ({p.Length} bytes)"); sbyte snr = unchecked((sbyte)p[1]); byte ch = p[4], pathLen = p[5], txtType = p[6]; uint ts = BinaryPrimitives.ReadUInt32LittleEndian(p.AsSpan(7, 4)); @@ -56,6 +59,7 @@ public static ChannelMessage ParseV3(byte[] p) public static ChannelMessage ParseLegacy(byte[] p) { + if (p.Length < 8) throw new InvalidDataException($"CHANNEL_MSG_RECV frame too short ({p.Length} bytes)"); byte ch = p[1], pathLen = p[2], txtType = p[3]; uint ts = BinaryPrimitives.ReadUInt32LittleEndian(p.AsSpan(4, 4)); string text = p.Length > 8 ? Encoding.UTF8.GetString(p, 8, p.Length - 8) : ""; @@ -71,6 +75,7 @@ public sealed record ChannelData(sbyte Snr, byte ChannelIndex, byte PathLen, ush public static ChannelData ParseRecv(byte[] p) { + if (p.Length < 9) throw new InvalidDataException($"CHANNEL_DATA_RECV frame too short ({p.Length} bytes)"); sbyte snr = unchecked((sbyte)p[1]); byte ch = p[4], pathLen = p[5]; ushort dataType = BinaryPrimitives.ReadUInt16LittleEndian(p.AsSpan(6, 2)); diff --git a/src/dapps/dapps.meshcore/MeshCoreInbound.cs b/src/dapps/dapps.meshcore/MeshCoreInbound.cs index 833b718..023e05a 100644 --- a/src/dapps/dapps.meshcore/MeshCoreInbound.cs +++ b/src/dapps/dapps.meshcore/MeshCoreInbound.cs @@ -41,7 +41,16 @@ public async Task RunAsync(CancellationToken ct) while (!ct.IsCancellationRequested) { // Wake on the MSG_WAITING push, or poll every 800ms as a safety net. - await Task.WhenAny(_wake.WaitAsync(ct), Task.Delay(800, ct)); + // Cancel the loser so we don't leak an orphaned semaphore waiter each + // iteration (which would also steal future MSG_WAITING releases). + using (var linked = CancellationTokenSource.CreateLinkedTokenSource(ct)) + { + var wake = _wake.WaitAsync(linked.Token); + var poll = Task.Delay(800, linked.Token); + await Task.WhenAny(wake, poll); + linked.Cancel(); + try { await Task.WhenAll(wake, poll); } catch { /* loser cancelled */ } + } if (ct.IsCancellationRequested) break; var batch = await _link.DrainAsync(ct); diff --git a/src/dapps/dapps.meshcore/MeshCoreLink.cs b/src/dapps/dapps.meshcore/MeshCoreLink.cs index 20f4dc6..24a8f57 100644 --- a/src/dapps/dapps.meshcore/MeshCoreLink.cs +++ b/src/dapps/dapps.meshcore/MeshCoreLink.cs @@ -36,6 +36,7 @@ public enum LinkState { Down, Healthy, Resetting, Failed } public int ResetCount { get; private set; } public SelfInfo? Self { get; private set; } public event Action? MessageWaiting; + public event Action? PacketHeard; public MeshCoreLink(MeshCoreBearerOptions opts, ILogger log) { @@ -56,13 +57,26 @@ private async Task OpenAndConfigureAsync(CancellationToken ct) { var client = new MeshCoreClient(_opts.SerialPort); client.MessageWaiting += () => MessageWaiting?.Invoke(); - client.Open(); - await client.AppStartAsync(_opts.AppName, ct); - await client.SetRadioParamsAsync(_region.FreqMhz, _region.BwKhz, _region.Sf, _region.Cr, ct); - await client.SetTxPowerAsync(Math.Min(_opts.TxPowerDbm, _region.MaxPowerDbm), ct); - await client.SetNameAsync(_opts.NodeName, ct); - await client.SetChannelAsync(_opts.ChannelIndex, _opts.ChannelName, _psk, ct); - var self = await client.AppStartAsync(_opts.AppName, ct); + client.PacketHeard += (len, snr) => PacketHeard?.Invoke(len, snr); + + SelfInfo self; + try + { + client.Open(); + await client.AppStartAsync(_opts.AppName, ct); + await client.SetRadioParamsAsync(_region.FreqMhz, _region.BwKhz, _region.Sf, _region.Cr, ct); + await client.SetTxPowerAsync(Math.Min(_opts.TxPowerDbm, _region.MaxPowerDbm), ct); + await client.SetNameAsync(_opts.NodeName, ct); + await client.SetChannelAsync(_opts.ChannelIndex, _opts.ChannelName, _psk, ct); + self = await client.AppStartAsync(_opts.AppName, ct); + } + catch + { + // Configuration failed after the port was opened / read loop started: + // dispose the local client so we don't leak the port + read-loop task. + try { await client.DisposeAsync(); } catch { } + throw; + } _client = client; Self = self; @@ -150,8 +164,17 @@ public async Task SendDataAsync(byte[] payload, CancellationToken ct) { var client = _client; if (client is null || State != LinkState.Healthy) return false; - await client.SendChannelDataAsync(_opts.ChannelIndex, payload, MeshCoreClient.DATA_TYPE_DEV, ct); - return true; + try + { + await client.SendChannelDataAsync(_opts.ChannelIndex, payload, MeshCoreClient.DATA_TYPE_DEV, ct); + return true; + } + catch (ObjectDisposedException) + { + // A concurrent recovery disposed the client between our State check and + // the write; treat as a soft failure so the caller retries. + return false; + } } /// Drain queued inbound messages, or null if the link is unavailable. diff --git a/src/dapps/dapps.meshcore/README.md b/src/dapps/dapps.meshcore/README.md index 7c67c05..c286abf 100644 --- a/src/dapps/dapps.meshcore/README.md +++ b/src/dapps/dapps.meshcore/README.md @@ -17,6 +17,10 @@ evidence this is built on. - **Airtime governor** (`TxBudget`) — a hard, self-enforced trailing-hour airtime budget (default 30 s/hr ≈ 0.83 % duty). Sends over budget are refused (back-pressure), so DAPPS can't burst the shared channel. The good-citizen control for riding a shared preset. +- **Adaptive airtime** (`ChannelMonitor`, #157) — estimates channel occupancy from the radio's + `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. - **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. @@ -45,6 +49,8 @@ Configure via `DAPPS_MESHCORE_*` env vars (or the `systemoptions` table): | `DAPPS_MESHCORE_NODE_NAME` | `DAPPS` | radio advert name | | `DAPPS_MESHCORE_AIRTIME_BUDGET_SECONDS_PER_HOUR` | `30` | governor budget | | `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) | 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 2b88aa6..b0c87d7 100644 --- a/src/dapps/dapps.meshcore/TxBudget.cs +++ b/src/dapps/dapps.meshcore/TxBudget.cs @@ -12,7 +12,7 @@ public sealed class TxBudget public const double DefaultSecondsPerHour = 30; // ≈0.83% duty private readonly double _budgetMs; - private readonly Queue<(DateTime when, double ms)> _window = new(); + private readonly List<(DateTime when, double ms)> _window = new(); private readonly object _lock = new(); private double _sumMs; @@ -42,18 +42,37 @@ public bool TryReserve(double airtimeMs, DateTime now, out string reason) reason = $"airtime budget exceeded: used {_sumMs / 1000:0.0}s + {airtimeMs / 1000:0.00}s > {_budgetMs / 1000:0.0}s/hr"; return false; } - _window.Enqueue((now, airtimeMs)); + _window.Add((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() + { + lock (_lock) + { + if (_window.Count > 0) + { + _sumMs -= _window[^1].ms; + _window.RemoveAt(_window.Count - 1); + } + if (_window.Count == 0 || _sumMs < 0) _sumMs = 0; + } + } + private void Prune(DateTime now) { var cutoff = now.AddHours(-1); - while (_window.Count > 0 && _window.Peek().when < cutoff) - _sumMs -= _window.Dequeue().ms; - if (_sumMs < 0) _sumMs = 0; + // Entries are appended in time order, so the stale ones are a prefix. + var drop = 0; + while (drop < _window.Count && _window[drop].when < cutoff) + _sumMs -= _window[drop++].ms; + if (drop > 0) _window.RemoveRange(0, drop); + if (_window.Count == 0 || _sumMs < 0) _sumMs = 0; } }