diff --git a/src/dapps/dapps.client/Backhaul/Dappsv1SessionBackhaul.cs b/src/dapps/dapps.client/Backhaul/Dappsv1SessionBackhaul.cs index 61e1e1c..551cb33 100644 --- a/src/dapps/dapps.client/Backhaul/Dappsv1SessionBackhaul.cs +++ b/src/dapps/dapps.client/Backhaul/Dappsv1SessionBackhaul.cs @@ -45,10 +45,20 @@ public Dappsv1SessionBackhaul( /// /// AGW handles any route that does not specify a higher-priority - /// bearer like UDP. Effectively: this is the fallback bearer when - /// only callsign + bearer port are known. + /// bearer like UDP or MeshCore. Effectively: this is the fallback bearer + /// when only callsign + bearer port are known. + /// + /// The MeshCore exclusion matters for passive discovery (#27): a peer + /// heard only over MeshCore produces a route with a MeshCoreChannel and a + /// null UdpEndpoint. If the MeshCore bearer is currently down (disabled, + /// serial link failed, or not yet started), it declines the route - and + /// without this guard AGW would claim it and attempt a doomed connected-mode + /// session (or spurious RF on a gateway node) to a callsign only ever heard + /// over LoRa. Excluding MeshCore routes leaves it Unreachable so the message + /// waits for MeshCore to return rather than mis-routing over the wrong bearer. /// - public bool CanHandle(BackhaulRoute route) => route.UdpEndpoint is null; + public bool CanHandle(BackhaulRoute route) => + route.UdpEndpoint is null && route.MeshCoreChannel is null; public async Task SendAsync( BackhaulMessage message, diff --git a/src/dapps/dapps.core.tests/Dappsv1SessionBackhaulTests.cs b/src/dapps/dapps.core.tests/Dappsv1SessionBackhaulTests.cs index cca3696..c4cff4a 100644 --- a/src/dapps/dapps.core.tests/Dappsv1SessionBackhaulTests.cs +++ b/src/dapps/dapps.core.tests/Dappsv1SessionBackhaulTests.cs @@ -35,6 +35,19 @@ public async Task CanHandle_UdpEndpointSet_False() await Task.CompletedTask; } + [Fact] + public async Task CanHandle_MeshCoreChannelSet_False() + { + var sb = MakeBackhaul([]); + // AGW must NOT claim a route discovered only over MeshCore (#27). If the + // MeshCore bearer is down (disabled / link failed / not yet started) it + // declines the route; AGW claiming it would mis-route a LoRa-only peer over a + // connected-mode session (spurious RF on a gateway node). Leave it Unreachable. + sb.CanHandle(new BackhaulRoute("N0DEST", BearerPort: 0, MeshCoreChannel: "dapps")) + .Should().BeFalse(); + await Task.CompletedTask; + } + [Fact] public async Task SendAsync_HappyPath_ReturnsOkAndWritesIhaveLine() { diff --git a/src/dapps/dapps.core.tests/MeshCoreDiscoveryRoutingTests.cs b/src/dapps/dapps.core.tests/MeshCoreDiscoveryRoutingTests.cs new file mode 100644 index 0000000..7e17789 --- /dev/null +++ b/src/dapps/dapps.core.tests/MeshCoreDiscoveryRoutingTests.cs @@ -0,0 +1,167 @@ +using AwesomeAssertions; +using dapps.client.Discovery; +using dapps.core.Models; +using dapps.core.Routing; +using dapps.core.Services; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; + +namespace dapps.core.tests; + +/// +/// Passive MeshCore discovery (#27): a peer we hear over the MeshCore bearer is +/// recorded as a with Bearer="meshcore" and a +/// non-null . This test pins the +/// routing half of the feature — that turns +/// such a row into a route the MeshCore bearer will claim (its CanHandle +/// keys purely on a non-empty MeshCoreChannel), so outbound to a +/// passively-discovered peer works with no manual neighbour configured. +/// +[Collection(SqliteOverridePathCollection.Name)] +public sealed class MeshCoreDiscoveryRoutingTests : IAsyncLifetime +{ + private string dbPath = null!; + private Database database = null!; + private DatabaseRoutingContext context = null!; + private StaticRoutingAlgorithm algorithm = null!; + + private const string OurCallsign = "G0US-1"; + private const string PeerCallsign = "G0MC-1"; + private const string PeerBaseCallsign = "G0MC"; + private const string Channel = "dapps-soak"; + + public ValueTask InitializeAsync() + { + dbPath = Path.Combine(Path.GetTempPath(), $"dapps-mcdisc-{Guid.NewGuid():N}.db"); + DbInfo.OverridePath = dbPath; + + using (var c = DbInfo.GetConnection()) + { + c.CreateTable(); + c.CreateTable(); + c.CreateTable(); + c.CreateTable(); + c.CreateTable(); + c.CreateTable(); + c.CreateTable(); + c.CreateTable(); + c.CreateTable(); + c.CreateTable(); + c.CreateTable(); + } + + var optionsMonitor = new TestOptionsMonitor(new SystemOptions { Callsign = OurCallsign }); + database = new Database(NullLogger.Instance, optionsMonitor); + context = new DatabaseRoutingContext(database, optionsMonitor); + algorithm = new StaticRoutingAlgorithm(NullLogger.Instance); + return ValueTask.CompletedTask; + } + + public ValueTask DisposeAsync() + { + DbInfo.OverridePath = null; + try { File.Delete(dbPath); } catch { /* ignore */ } + return ValueTask.CompletedTask; + } + + private Task Record(DbDiscoveredPeer peer) => database.UpsertDiscoveredPeer(peer); + + private static DbDiscoveredPeer MeshCorePeer(DateTime lastSeen) => new() + { + Callsign = PeerCallsign, + Bearer = "meshcore", + ChannelKey = Channel, + LinkClass = LinkClass.MeshCore, + CostHint = LinkClassDefaults.CostHint(LinkClass.MeshCore), + Hops = 1, + TtlSeconds = LinkClassDefaults.AdvertisedTtlSeconds(LinkClass.MeshCore), + MeshCoreChannel = Channel, + LastSeen = lastSeen, + }; + + private static DbMessage OutboundTo(string baseCallsign) => new() + { + Id = "0000002", + Destination = $"chat@{baseCallsign}-1", + Payload = "x"u8.ToArray(), + }; + + [Fact] + public async Task FreshMeshCorePeer_RoutesWithChannelHint() + { + await Record(MeshCorePeer(DateTime.UtcNow)); + + var decision = await algorithm.ResolveAsync( + OutboundTo(PeerBaseCallsign), context, TestContext.Current.CancellationToken); + + var nh = decision.Should().BeOfType().Subject; + nh.Route.Callsign.Should().Be(PeerCallsign); + nh.Route.MeshCoreChannel.Should().Be(Channel); + } + + [Fact] + public async Task StalePeer_AgedOut_IsUnreachable() + { + // Older than the MeshCore advertised TTL (10800s) → the freshness + // filter drops it, so there's no route. + await Record(MeshCorePeer(DateTime.UtcNow - TimeSpan.FromSeconds(LinkClassDefaults.AdvertisedTtlSeconds(LinkClass.MeshCore) + 60))); + + var decision = await algorithm.ResolveAsync( + OutboundTo(PeerBaseCallsign), context, TestContext.Current.CancellationToken); + + decision.Should().BeOfType(); + } + + [Fact] + public async Task NonMeshCorePeer_HasNoChannelHint() + { + // A UDP-heard peer must not gain a MeshCore channel hint — otherwise the + // MeshCore bearer's CanHandle (non-empty MeshCoreChannel) would wrongly + // claim a route that should go out over UDP. + await Record(new DbDiscoveredPeer + { + Callsign = PeerCallsign, + Bearer = "udp", + ChannelKey = "239.0.0.1:5000", + LinkClass = LinkClass.LanMulticast, + CostHint = LinkClassDefaults.CostHint(LinkClass.LanMulticast), + Hops = 1, + TtlSeconds = LinkClassDefaults.AdvertisedTtlSeconds(LinkClass.LanMulticast), + UdpEndpoint = "127.0.0.1:5000", + LastSeen = DateTime.UtcNow, + }); + + var decision = await algorithm.ResolveAsync( + OutboundTo(PeerBaseCallsign), context, TestContext.Current.CancellationToken); + + var nh = decision.Should().BeOfType().Subject; + nh.Route.MeshCoreChannel.Should().BeNull(); + nh.Route.UdpEndpoint.Should().Be("127.0.0.1:5000"); + } + + [Fact] + public async Task CheaperMeshCore_PreferredOverIpForSamePeer() + { + // Same peer heard on both a MeshCore channel (cost 3, RF-in-spirit) and a + // UDP channel (cost 8). The router must pick the cheaper MeshCore row. + await Record(new DbDiscoveredPeer + { + Callsign = PeerCallsign, + Bearer = "udp", + ChannelKey = "239.0.0.1:5000", + LinkClass = LinkClass.LanMulticast, + CostHint = LinkClassDefaults.CostHint(LinkClass.LanMulticast), + Hops = 1, + TtlSeconds = LinkClassDefaults.AdvertisedTtlSeconds(LinkClass.LanMulticast), + UdpEndpoint = "127.0.0.1:5000", + LastSeen = DateTime.UtcNow, + }); + await Record(MeshCorePeer(DateTime.UtcNow)); + + var decision = await algorithm.ResolveAsync( + OutboundTo(PeerBaseCallsign), context, TestContext.Current.CancellationToken); + + var nh = decision.Should().BeOfType().Subject; + nh.Route.MeshCoreChannel.Should().Be(Channel); + } +} diff --git a/src/dapps/dapps.core/Models/DbDiscoveredPeer.cs b/src/dapps/dapps.core/Models/DbDiscoveredPeer.cs index afe9e6e..ce4d267 100644 --- a/src/dapps/dapps.core/Models/DbDiscoveredPeer.cs +++ b/src/dapps/dapps.core/Models/DbDiscoveredPeer.cs @@ -59,6 +59,11 @@ public sealed class DbDiscoveredPeer /// (UDP only). public string? UdpEndpoint { get; set; } + /// MeshCore channel name the peer was heard on (MeshCore bearer only), + /// so the router can build a MeshCore route back to a passively-discovered peer + /// without a manual neighbour (#27). Null for other bearers. + public string? MeshCoreChannel { get; set; } + public DateTime LastSeen { get; set; } public static string MakeKey(string callsign, string bearer, string channelKey) diff --git a/src/dapps/dapps.core/Routing/StaticRoutingAlgorithm.cs b/src/dapps/dapps.core/Routing/StaticRoutingAlgorithm.cs index b9f8a74..53bdd16 100644 --- a/src/dapps/dapps.core/Routing/StaticRoutingAlgorithm.cs +++ b/src/dapps/dapps.core/Routing/StaticRoutingAlgorithm.cs @@ -64,7 +64,8 @@ public async Task ResolveAsync(DbMessage message, IRoutingContext return new RouteDecision.NextHop(new BackhaulRoute( freshPeer.Callsign, BearerPort: freshPeer.BearerPort ?? ctx.DefaultBearerPort, - UdpEndpoint: freshPeer.UdpEndpoint)); + UdpEndpoint: freshPeer.UdpEndpoint, + MeshCoreChannel: freshPeer.MeshCoreChannel)); } // 3. Hand-maintained route hint. The fallback for "I know peer diff --git a/src/dapps/dapps.core/Services/MeshCoreBearer.cs b/src/dapps/dapps.core/Services/MeshCoreBearer.cs index 148b64d..8d85ec5 100644 --- a/src/dapps/dapps.core/Services/MeshCoreBearer.cs +++ b/src/dapps/dapps.core/Services/MeshCoreBearer.cs @@ -1,4 +1,5 @@ using dapps.client.Backhaul; +using dapps.client.Discovery; using dapps.client.Tx; using dapps.core.Models; using dapps.meshcore; @@ -21,6 +22,7 @@ public sealed class MeshCoreBearer : IDappsBackhaul, IAsyncDisposable private readonly IOptionsMonitor _sysOpts; private readonly IBackhaulInbox _inbox; private readonly IDappsTxGate _txGate; + private readonly Database _database; private readonly ILoggerFactory _loggerFactory; private readonly ILogger _log; @@ -29,6 +31,13 @@ public sealed class MeshCoreBearer : IDappsBackhaul, IAsyncDisposable private MeshCoreInbound? _inbound; private MeshCoreReliability? _reliability; + // Passive discovery (#27): throttle DB upserts per peer so a chatty channel + // doesn't hammer SQLite — we only need to refresh a peer's freshness, not record + // every frame. Keyed on the source callsign; guarded by its own lock. + private static readonly TimeSpan DiscoveryUpsertThrottle = TimeSpan.FromSeconds(30); + private readonly object _discoveryLock = new(); + private readonly Dictionary _lastDiscoveryUpsert = new(StringComparer.OrdinalIgnoreCase); + public bool Enabled { get; private set; } public MeshCoreLink? Link => _link; public MeshCoreInbound? Inbound => _inbound; @@ -61,11 +70,13 @@ public MeshCoreBearer( IOptionsMonitor sysOpts, IBackhaulInbox inbox, IDappsTxGate txGate, + Database database, ILoggerFactory loggerFactory) { _sysOpts = sysOpts; _inbox = inbox; _txGate = txGate; + _database = database; _loggerFactory = loggerFactory; _log = loggerFactory.CreateLogger(); } @@ -104,15 +115,42 @@ public async Task RunAsync(CancellationToken ct) _link, _inbox, _loggerFactory.CreateLogger(), reliability, sendAck: (ack, c) => backhaul.ResendAsync(ack, opts.LocalCallsign, c), - localCallsign: opts.LocalCallsign); + localCallsign: opts.LocalCallsign, + onPeerHeard: (src, c) => RecordPeerHeardAsync(src, opts, c)); Enabled = true; // Reliability resend loop runs alongside the inbound drain loop. var resendTask = reliability is not null ? Task.Run(() => ResendLoopAsync(reliability, backhaul, ct), ct) : Task.CompletedTask; + // Age out stale discovered peers ourselves. On a MeshCore-only node there's no + // AGW/UDP discovery channel, so DiscoveryService (the usual sweeper) never runs + // its age-out; without this, passively-recorded rows would accumulate for the + // node's lifetime (#27 review). Bearer-agnostic and idempotent, so it's harmless + // to also run on a mixed node where DiscoveryService sweeps too. + var housekeepingTask = Task.Run(() => AgeOutLoopAsync(ct), ct); try { await _inbound.RunAsync(ct); } - finally { try { await resendTask; } catch { /* shutdown */ } } + finally + { + try { await resendTask; } catch { /* shutdown */ } + try { await housekeepingTask; } catch { /* shutdown */ } + } + } + + private async Task AgeOutLoopAsync(CancellationToken ct) + { + while (!ct.IsCancellationRequested) + { + try { await Task.Delay(TimeSpan.FromMinutes(5), ct); } + catch (OperationCanceledException) { break; } + try + { + var aged = await _database.AgeOutDiscoveredPeers(DateTime.UtcNow); + if (aged.Count > 0) + _log.LogInformation("MeshCore: aged out {0} stale discovered peer(s)", aged.Count); + } + catch (Exception ex) { _log.LogDebug("MeshCore: discovered-peer age-out failed: {0}", ex.Message); } + } } private async Task ResendLoopAsync(MeshCoreReliability reliability, MeshCoreCompanionBackhaul backhaul, CancellationToken ct) @@ -138,6 +176,67 @@ private async Task ResendLoopAsync(MeshCoreReliability reliability, MeshCoreComp } } + /// Passive discovery (#27): record a peer we heard over MeshCore as a + /// fresh so the router can send to it without a + /// manual neighbour. Throttled per peer, skips ourselves (a repeater may echo our + /// own frames), and never lets a DB fault break the inbound drain loop. + private async Task RecordPeerHeardAsync(string source, MeshCoreBearerOptions opts, CancellationToken ct) + { + // A repeater re-broadcasting our own frame would otherwise teach us a route to + // ourselves. Compare on the base callsign (SSID-insensitive), like the router. + // Use the LIVE callsign, not the one captured at bearer start: outbound frames + // are stamped with the current callsign, so after a runtime rename an echo of + // our own frame must still be recognised as self. + var localCallsign = _sysOpts.CurrentValue.Callsign ?? ""; + if (string.Equals(source.Split('-')[0], localCallsign.Split('-')[0], StringComparison.OrdinalIgnoreCase)) + return; + + var now = DateTime.UtcNow; + lock (_discoveryLock) + { + if (_lastDiscoveryUpsert.TryGetValue(source, out var last) && now - last < DiscoveryUpsertThrottle) + return; + // Bound the dict: an entry older than the throttle window no longer gates + // anything, so drop stale ones (also caps memory if the channel injects + // many distinct callsigns). Cheap — runs only when we're about to upsert. + if (_lastDiscoveryUpsert.Count > 0) + { + var cutoff = now - DiscoveryUpsertThrottle; + foreach (var k in _lastDiscoveryUpsert.Where(kv => kv.Value < cutoff).Select(kv => kv.Key).ToList()) + _lastDiscoveryUpsert.Remove(k); + } + _lastDiscoveryUpsert[source] = now; + } + + var peer = new DbDiscoveredPeer + { + Callsign = source, + Bearer = "meshcore", + ChannelKey = opts.ChannelName, + ChannelId = 0, + LinkClass = LinkClass.MeshCore, + CostHint = LinkClassDefaults.CostHint(LinkClass.MeshCore), + Hops = 1, + TtlSeconds = LinkClassDefaults.AdvertisedTtlSeconds(LinkClass.MeshCore), + MeshCoreChannel = opts.ChannelName, + LastSeen = now, + }; + try + { + await _database.UpsertDiscoveredPeer(peer); + _log.LogInformation("MeshCore: discovered peer {0} on {1} (cost={2}, ttl={3}s)", + source, opts.ChannelName, peer.CostHint, peer.TtlSeconds); + } + catch (Exception ex) + { + // Keep the throttle stamp: on a persistent DB fault, rolling it back would + // let every subsequent frame from this peer re-attempt (and re-fail) the + // upsert, defeating the 30s rate-limit exactly when the DB is unhealthy. A + // retry still happens on the next frame after the window, which is enough. + _log.LogWarning("MeshCore: failed to record discovered peer {0}: {1}", source, ex.Message); + } + } + public bool CanHandle(BackhaulRoute route) => Enabled && _backhaul is not null && _backhaul.CanHandle(route); diff --git a/src/dapps/dapps.meshcore.soak/Program.cs b/src/dapps/dapps.meshcore.soak/Program.cs index e607152..21412e8 100644 --- a/src/dapps/dapps.meshcore.soak/Program.cs +++ b/src/dapps/dapps.meshcore.soak/Program.cs @@ -65,12 +65,17 @@ double dropPct = a.GetDouble("drop-pct", 0); Func? drop = dropPct > 0 ? (_ => Random.Shared.NextDouble() * 100 < dropPct) : null; +// Passive discovery (#27): count the peers we learn about purely from hearing +// their traffic — the same signal MeshCoreBearer feeds to Database.UpsertDiscoveredPeer +// in the integrated host. Each node should discover the other with no config. +var discovery = new DiscoveryRecorder(self, log); var inbound = new MeshCoreInbound( link, inbox, lf.CreateLogger(), reliability, sendAck: (ack, c) => backhaul.ResendAsync(ack, self, c), localCallsign: self, - dropForTest: drop); + dropForTest: drop, + onPeerHeard: (src, c) => { discovery.Note(src); return Task.CompletedTask; }); var route = new BackhaulRoute(peer, MeshCoreChannel: opts.ChannelName); var inboundTask = Task.Run(() => inbound.RunAsync(cts.Token)); @@ -122,7 +127,11 @@ } catch (Exception ex) { Interlocked.Increment(ref failed); log.LogWarning("TX seq={0} exception: {1}", seq, ex.Message); } seq++; - try { await Task.Delay(TimeSpan.FromSeconds(intervalSec), cts.Token); } catch { break; } + // Jitter the interval (±25%) so two nodes started at the same instant don't + // phase-lock their transmit schedules and collide every cycle (half-duplex → + // mutual deafness). Real DAPPS traffic is event-driven, not on a fixed clock. + var jittered = intervalSec * (0.75 + Random.Shared.NextDouble() * 0.5); + try { await Task.Delay(TimeSpan.FromSeconds(jittered), cts.Token); } catch { break; } } }); @@ -153,10 +162,34 @@ 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); +log.LogInformation("Discovery: {0}", discovery.Summary()); return 0; // ---------------- helpers ---------------- +sealed class DiscoveryRecorder(string self, ILogger log) +{ + private readonly object _l = new(); + private readonly Dictionary _heard = new(StringComparer.OrdinalIgnoreCase); + + public void Note(string source) + { + // Mirror MeshCoreBearer: never learn ourselves (a repeater could echo us). + if (string.Equals(source.Split('-')[0], self.Split('-')[0], StringComparison.OrdinalIgnoreCase)) return; + int count; + lock (_l) { _heard.TryGetValue(source, out count); _heard[source] = ++count; } + if (count == 1) log.LogInformation("DISCOVERED peer {0} (first heard over MeshCore, no config)", source); + } + + public string Summary() + { + lock (_l) + return _heard.Count == 0 + ? "no peers discovered" + : string.Join(", ", _heard.OrderBy(kv => kv.Key).Select(kv => $"{kv.Key}×{kv.Value}")); + } +} + sealed class SoakInbox(string self, ILogger log) : IBackhaulInbox { private readonly object _l = new(); diff --git a/src/dapps/dapps.meshcore/MeshCoreInbound.cs b/src/dapps/dapps.meshcore/MeshCoreInbound.cs index 2d8b002..ccf2783 100644 --- a/src/dapps/dapps.meshcore/MeshCoreInbound.cs +++ b/src/dapps/dapps.meshcore/MeshCoreInbound.cs @@ -27,6 +27,10 @@ public sealed class MeshCoreInbound private readonly Func>? _sendAck; private readonly string _localCallsign; private readonly Func? _dropForTest; + // Passive discovery (#27): fired with the source callsign of every data message + // we hear (addressed to us or not), so the host can record the peer as reachable + // over MeshCore without a manual neighbour. Null disables discovery. + private readonly Func? _onPeerHeard; // 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. @@ -41,7 +45,8 @@ public MeshCoreInbound( MeshCoreReliability? reliability = null, Func>? sendAck = null, string? localCallsign = null, - Func? dropForTest = null) + Func? dropForTest = null, + Func? onPeerHeard = null) { _link = link; _inbox = inbox; @@ -50,6 +55,7 @@ public MeshCoreInbound( _sendAck = sendAck; _localCallsign = localCallsign ?? ""; _dropForTest = dropForTest; + _onPeerHeard = onPeerHeard; _link.MessageWaiting += () => { try { _wake.Release(); } catch { } }; } @@ -102,6 +108,17 @@ public async Task RunAsync(CancellationToken ct) ? msg.LinkSourceCallsign! : UnknownSourceCallsign; + // Passive discovery (#27): we heard this peer one hop away over + // MeshCore. Record it (throttled by the host) so outbound to it can + // route without a manual neighbour. Anonymous senders (sentinel) and + // recorder faults must not break the drain loop. Fires for duplicates + // too, so a peer we keep hearing stays fresh. + if (_onPeerHeard is not null && source != UnknownSourceCallsign) + { + try { await _onPeerHeard(source, ct); } + catch (Exception ex) { _log.LogDebug("MeshCore: discovery record failed for {0}: {1}", source, ex.Message); } + } + // 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.