diff --git a/src/dapps/dapps.core.tests/MeshFabricScenarioTests.cs b/src/dapps/dapps.core.tests/MeshFabricScenarioTests.cs new file mode 100644 index 0000000..a60dc40 --- /dev/null +++ b/src/dapps/dapps.core.tests/MeshFabricScenarioTests.cs @@ -0,0 +1,99 @@ +using AwesomeAssertions; +using dapps.meshcore.sim; +using Xunit; + +namespace dapps.core.tests; + +/// +/// Full-stack scenarios: the REAL MeshCore bearer (transport + reliability ACK/resend + +/// idempotent inbound + passive discovery) running on many nodes over a multi-hop +/// virtual mesh - the coverage two directly-adjacent bench radios can't give. No RF, no +/// loss (so it's fast + deterministic); loss-recovery timing is left to a longer harness +/// run since reliability resends are on a multi-second cadence. +/// +public sealed class MeshFabricScenarioTests +{ + private static async Task WaitUntil(Func cond, TimeSpan timeout) + { + var deadline = DateTime.UtcNow + timeout; + while (DateTime.UtcNow < deadline) + { + if (cond()) return; + await Task.Delay(100); + } + } + + [Fact] + public async Task FullStack_FourRelayHops_BidirectionalDelivery_Discovery_NoDuplicates() + { + // A off R1 ... R4 off C: the two DAPPS nodes are four relay hops apart. + var f = new MeshFabric(); + f.AddRelay("R1"); f.AddRelay("R2"); f.AddRelay("R3"); f.AddRelay("R4"); + f.ConnectChain(["R1", "R2", "R3", "R4"]); + var a = new MeshDappsNode(f, "GB7A-1"); + var c = new MeshDappsNode(f, "GB7C-1"); + f.Connect("GB7A-1", "R1"); + f.Connect("GB7C-1", "R4"); + + using var cts = new CancellationTokenSource(); + var loops = Task.WhenAll(a.RunAsync(cts.Token), c.RunAsync(cts.Token)); + + const int N = 4; + for (var i = 0; i < N; i++) + { + await a.SendAsync("GB7C-1", i, $"A->C #{i}", cts.Token); + await c.SendAsync("GB7A-1", i, $"C->A #{i}", cts.Token); + } + + await WaitUntil(() => a.DistinctSeqs == N && c.DistinctSeqs == N, TimeSpan.FromSeconds(15)); + cts.Cancel(); + try { await loops; } catch { /* cancelled */ } + + c.DistinctSeqs.Should().Be(N, "every message reached C across four relay hops"); + a.DistinctSeqs.Should().Be(N); + c.Delivered.Should().Be(N, "idempotent: flooding + dedup never double-delivers to the app"); + a.Delivered.Should().Be(N); + c.DiscoveredPeers.Should().Contain("GB7A-1", "C learned A purely by hearing its traffic, multi-hop"); + a.DiscoveredPeers.Should().Contain("GB7C-1"); + } + + [Fact] + public async Task FullStack_FanIn_ManySendersOneCollector_AllDeliveredOnce() + { + // Five DAPPS senders hang off a relay backbone; all send to one collector. Proves + // many nodes + concurrent multi-hop floods still deliver each message exactly once. + var f = new MeshFabric(); + var relays = new[] { "R0", "R1", "R2", "R3", "R4", "R5" }; + foreach (var r in relays) f.AddRelay(r); + f.ConnectChain(relays); + + var collector = new MeshDappsNode(f, "GB7COL-1"); + f.Connect("GB7COL-1", "R0"); + + var senders = new List(); + for (var i = 0; i < 5; i++) + { + var s = new MeshDappsNode(f, $"GB7S{i}-1"); + f.Connect(s.Callsign, relays[i + 1]); // each sender off a different relay + senders.Add(s); + } + + using var cts = new CancellationTokenSource(); + var loops = Task.WhenAll(new[] { collector.RunAsync(cts.Token) } + .Concat(senders.Select(s => s.RunAsync(cts.Token)))); + + const int PerSender = 3; + foreach (var s in senders) + for (var i = 0; i < PerSender; i++) + await s.SendAsync("GB7COL-1", i, $"{s.Callsign} #{i}", cts.Token); + + var expected = senders.Count * PerSender; // seq values collide across senders, so count deliveries + await WaitUntil(() => collector.Delivered >= expected, TimeSpan.FromSeconds(20)); + cts.Cancel(); + try { await loops; } catch { /* cancelled */ } + + collector.Delivered.Should().Be(expected, "each of the 15 messages delivered exactly once (no duplicates)"); + collector.DiscoveredPeers.Should().Contain(senders.Select(s => s.Callsign), + "the collector discovered every sender it heard"); + } +} diff --git a/src/dapps/dapps.core.tests/MeshFabricTests.cs b/src/dapps/dapps.core.tests/MeshFabricTests.cs new file mode 100644 index 0000000..9468d98 --- /dev/null +++ b/src/dapps/dapps.core.tests/MeshFabricTests.cs @@ -0,0 +1,198 @@ +using System.Text; +using AwesomeAssertions; +using dapps.client.Backhaul; +using dapps.meshcore; +using dapps.meshcore.sim; +using Xunit; + +namespace dapps.core.tests; + +/// +/// The simulated multi-hop MeshCore fabric (#24 test-strategy follow-up): validates the +/// flood/dedup/hop/scope model, and that the REAL bearer transport carries a +/// BackhaulMessage across several relay hops and decodes it exactly once. Deterministic +/// and fast - no bearer poll loops, no RF. See MeshFabricScenarioTests for the full +/// bearer stack running over the fabric. +/// +public sealed class MeshFabricTests +{ + // How many app-deliveries a leaf's link received (each fabric Deliver enqueues one + // ChannelData; a deduped duplicate never enqueues). + private static int Received(SimulatedMeshCoreLink link) + { + var batch = link.DrainAsync(CancellationToken.None).GetAwaiter().GetResult(); + return batch?.Data.Count ?? 0; + } + + [Fact] + public void Chain_MultiHop_DeliversToFarLeaf() + { + // A(leaf) - R1 - R2 - R3 - B(leaf): a message from A must traverse 3 relay hops. + var f = new MeshFabric(); + var a = f.AddLeaf("A"); + f.AddRelay("R1"); f.AddRelay("R2"); f.AddRelay("R3"); + var b = f.AddLeaf("B"); + f.ConnectChain(["A", "R1", "R2", "R3", "B"]); + + a.SendDataAsync("hello"u8.ToArray(), CancellationToken.None).GetAwaiter().GetResult(); + + Received(b).Should().Be(1, "the far leaf hears the flood exactly once"); + } + + [Fact] + public void Flood_DedupsAcrossMultiplePaths() + { + // Diamond: A -> {R1, R2} -> B. B hears the same packet via both relays but the + // dedup ring must deliver it only once. + var f = new MeshFabric(); + var a = f.AddLeaf("A"); + f.AddRelay("R1"); f.AddRelay("R2"); + var b = f.AddLeaf("B"); + f.Connect("A", "R1"); f.Connect("A", "R2"); + f.Connect("R1", "B"); f.Connect("R2", "B"); + + a.SendDataAsync("x"u8.ToArray(), CancellationToken.None).GetAwaiter().GetResult(); + + Received(b).Should().Be(1, "two paths, but the packet is deduped to a single delivery"); + } + + [Fact] + public void Leaf_DoesNotRelay() + { + // A - B(leaf) - C: B is a LEAF so it must not re-flood; C never hears it. + var f = new MeshFabric(); + var a = f.AddLeaf("A"); + var b = f.AddLeaf("B"); + var c = f.AddLeaf("C"); + f.ConnectChain(["A", "B", "C"]); + + a.SendDataAsync("y"u8.ToArray(), CancellationToken.None).GetAwaiter().GetResult(); + + Received(b).Should().Be(1, "the adjacent leaf hears it"); + Received(c).Should().Be(0, "a leaf does not relay, so the far node is unreachable"); + } + + [Fact] + public void Ring_Dedup_Terminates() + { + // A ring of relays would flood forever without dedup; the ring must terminate and + // each leaf still gets one copy. + var f = new MeshFabric(); + var a = f.AddLeaf("A"); + f.AddRelay("R1"); f.AddRelay("R2"); f.AddRelay("R3"); f.AddRelay("R4"); + var b = f.AddLeaf("B"); + f.Connect("A", "R1"); + f.ConnectChain(["R1", "R2", "R3", "R4", "R1"]); // R1-R2-R3-R4-R1 cycle + f.Connect("R3", "B"); + + a.SendDataAsync("z"u8.ToArray(), CancellationToken.None).GetAwaiter().GetResult(); + + Received(b).Should().Be(1, "delivered once despite the cycle"); + } + + [Fact] + public void ScopedFlood_ContainedByOutOfScopeRelay() + { + // Model B: A(scope=uk) - R1(scope=uk) - R2(scope=other) - B. The scoped flood must + // stop at R2 (wrong scope), so B never hears it. + var f = new MeshFabric(); + var a = f.AddLeaf("A", scope: "uk"); + f.AddRelay("R1", scope: "uk"); + f.AddRelay("R2", scope: "other"); + var b = f.AddLeaf("B"); + f.ConnectChain(["A", "R1", "R2", "B"]); + + a.SendDataAsync("scoped"u8.ToArray(), CancellationToken.None).GetAwaiter().GetResult(); + + Received(b).Should().Be(0, "an out-of-scope relay drops the flood (containment)"); + } + + [Fact] + public void ScopedFlood_CarriedByInScopeRelays() + { + // Control for the containment test: when every relay shares the scope, delivery works. + var f = new MeshFabric(); + var a = f.AddLeaf("A", scope: "uk"); + f.AddRelay("R1", scope: "uk"); + f.AddRelay("R2", scope: "uk"); + var b = f.AddLeaf("B"); + f.ConnectChain(["A", "R1", "R2", "B"]); + + a.SendDataAsync("scoped"u8.ToArray(), CancellationToken.None).GetAwaiter().GetResult(); + + Received(b).Should().Be(1, "in-scope relays carry the flood to the destination"); + } + + [Fact] + public void UnscopedFlood_CrossesRelaysRegardlessOfTheirScope() + { + // An unscoped flood (model A) is carried by any relay, even scoped ones. + var f = new MeshFabric(); + var a = f.AddLeaf("A"); // no scope + f.AddRelay("R1", scope: "uk"); + f.AddRelay("R2", scope: "other"); + var b = f.AddLeaf("B"); + f.ConnectChain(["A", "R1", "R2", "B"]); + + a.SendDataAsync("open"u8.ToArray(), CancellationToken.None).GetAwaiter().GetResult(); + + Received(b).Should().Be(1, "unscoped floods propagate through every relay"); + } + + [Theory] + [InlineData(64, 1)] // 64 relays all forward (R64 receives path_len 63 < 64) -> B hears it + [InlineData(65, 0)] // the 65th relay would receive path_len 64 -> dropped -> B unreachable + public void HopCap_DeliversAtExactly64Relays_DropsBeyond(int relayCount, int expected) + { + // The firmware cap is MAX_PATH_SIZE=64 forwarders (path_len < 64). Pin the exact + // boundary - this is the long-chain edge the fabric exists to exercise. + var f = new MeshFabric(); + f.AddLeaf("A"); + var chain = new List { "A" }; + for (var i = 1; i <= relayCount; i++) { f.AddRelay($"R{i}"); chain.Add($"R{i}"); } + f.AddLeaf("B"); + chain.Add("B"); + f.ConnectChain(chain); + + f.Link("A").SendDataAsync("edge"u8.ToArray(), CancellationToken.None).GetAwaiter().GetResult(); + + Received(f.Link("B")).Should().Be(expected); + } + + [Theory] + [InlineData("73 de M0LTE")] // one fragment + [InlineData("Hello from the DAPPS mailbox, a longer store-and-forward message over the mesh that fragments. 73 de M0LTE GB7ABC-1 599 599 599")] // multi-fragment + public void Transport_MultiHop_DecodesOnceAtDestination(string text) + { + // The REAL bearer transport (encode + compress + fragment + nonce) carried across + // 3 relay hops must reassemble/decode to the original message, exactly once. + var f = new MeshFabric(); + var a = f.AddLeaf("A"); + f.AddRelay("R1"); f.AddRelay("R2"); f.AddRelay("R3"); + var b = f.AddLeaf("B"); + f.ConnectChain(["A", "R1", "R2", "R3", "B"]); + + var msg = new BackhaulMessage( + Id: "abc0001", Destination: "GB7B-1", Salt: 7, Ttl: 3600, + Payload: Encoding.UTF8.GetBytes(text), Originator: "GB7A-1", LinkSourceCallsign: "GB7A-1"); + + // Encode + fragment + nonce at A, send each frame; B reassembles + decodes. + var frames = new MeshCoreChannelTransport().ToFrames(msg, DappsCompression.Mode.ZstdDict); + foreach (var frame in frames) + a.SendDataAsync(frame, CancellationToken.None).GetAwaiter().GetResult(); + + var rx = new MeshCoreChannelTransport(); + var got = new List(); + var batch = b.DrainAsync(CancellationToken.None).GetAwaiter().GetResult(); + batch.Should().NotBeNull("the destination leaf heard the frames"); + foreach (var d in batch!.Data) + { + var r = rx.Ingest(d.Payload, DateTime.UtcNow); + if (r.Kind == MeshCoreChannelTransport.Kind.BackhaulComplete) got.Add(r.Message!); + } + + got.Should().HaveCount(1, "the message reassembles/decodes exactly once"); + Encoding.UTF8.GetString(got[0].Payload).Should().Be(text); + got[0].Id.Should().Be("abc0001"); + } +} diff --git a/src/dapps/dapps.core.tests/dapps.core.tests.csproj b/src/dapps/dapps.core.tests/dapps.core.tests.csproj index 5c1816a..73ff7f1 100644 --- a/src/dapps/dapps.core.tests/dapps.core.tests.csproj +++ b/src/dapps/dapps.core.tests/dapps.core.tests.csproj @@ -21,6 +21,7 @@ + diff --git a/src/dapps/dapps.meshcore.sim/MeshDappsNode.cs b/src/dapps/dapps.meshcore.sim/MeshDappsNode.cs new file mode 100644 index 0000000..8bbd800 --- /dev/null +++ b/src/dapps/dapps.meshcore.sim/MeshDappsNode.cs @@ -0,0 +1,116 @@ +using System.Collections.Concurrent; +using System.Text; +using dapps.client.Backhaul; +using dapps.meshcore; +using Microsoft.Extensions.Logging.Abstractions; + +namespace dapps.meshcore.sim; + +/// +/// A full DAPPS MeshCore node running on a leaf: the REAL +/// + (+ reliability) +/// wired to a , exactly as the host wires them to a +/// serial link. Lets a scenario exercise the actual transport / dedup / reliability / +/// discovery code over a multi-hop virtual mesh. +/// +public sealed class MeshDappsNode +{ + private readonly SimulatedMeshCoreLink _link; + private readonly MeshCoreCompanionBackhaul _backhaul; + private readonly MeshCoreInbound _inbound; + private readonly MeshCoreReliability? _reliability; + private readonly RecordingInbox _inbox; + private readonly string _channel; + private readonly ConcurrentDictionary _discovered = new(StringComparer.OrdinalIgnoreCase); + + public string Callsign { get; } + + /// Distinct (id) messages delivered to this node's app. + public long Delivered => _inbox.Delivered; + /// True if a message with this sequence number was delivered (addressed to us). + public bool GotSeq(int seq) => _inbox.HasSeq(seq); + public int DistinctSeqs => _inbox.DistinctSeqs; + /// Peers this node learned purely by hearing their traffic (passive discovery). + public IReadOnlyCollection DiscoveredPeers => _discovered.Keys.ToList(); + + public MeshDappsNode(MeshFabric fabric, string callsign, string channel = "dapps-sim", bool reliable = true, string scope = "") + { + Callsign = callsign; + _channel = channel; + _link = fabric.AddLeaf(callsign, scope); + var opts = new MeshCoreBearerOptions + { + Region = "uk-test", ChannelName = channel, ChannelIndex = 1, + LocalCallsign = callsign, NodeName = callsign, ReliableDelivery = reliable, + LbtGuardMs = 0, // no radio timing to wait on in the sim + AirtimeBudgetSecPerHour = 3600, // don't let the governor gate the scenario + }; + _reliability = reliable ? new MeshCoreReliability() : null; + _inbox = new RecordingInbox(callsign); + _backhaul = new MeshCoreCompanionBackhaul(_link, opts, new TxBudget(opts.AirtimeBudgetSecPerHour), + NullLogger.Instance, reliability: _reliability); + _inbound = new MeshCoreInbound( + _link, _inbox, NullLogger.Instance, _reliability, + sendAck: (ack, c) => _backhaul.ResendAsync(ack, callsign, c), + localCallsign: callsign, + onPeerHeard: (src, c) => { _discovered.AddOrUpdate(src, 1, (_, n) => n + 1); return Task.CompletedTask; }); + } + + /// Run the inbound drain loop and (if reliable) the resend loop until cancelled. + public async Task RunAsync(CancellationToken ct) + { + var resend = _reliability is not null ? Task.Run(() => ResendLoopAsync(ct), ct) : Task.CompletedTask; + try { await _inbound.RunAsync(ct); } + finally { try { await resend; } catch { } } + } + + /// Originate a sequence-numbered message to . + public Task SendAsync(string peer, int seq, string text, CancellationToken ct) + { + var msg = new BackhaulMessage( + Id: Guid.NewGuid().ToString("N")[..7], Destination: peer, Salt: null, Ttl: 3600, + Payload: Encoding.UTF8.GetBytes(text), Originator: Callsign, LinkSourceCallsign: Callsign, + Headers: new Dictionary { ["seq"] = seq.ToString() }); + return _backhaul.SendAsync(msg, new BackhaulRoute(peer, MeshCoreChannel: _channel), Callsign, ct); + } + + private async Task ResendLoopAsync(CancellationToken ct) + { + while (!ct.IsCancellationRequested) + { + try { await Task.Delay(TimeSpan.FromSeconds(2), ct); } catch { break; } + var now = DateTime.UtcNow; + _reliability!.DropExpired(now); + foreach (var (m, local) in _reliability.DueResends(now)) + { + try { if ((await _backhaul.ResendAsync(m, local, ct)).Accepted) _reliability.MarkResent(m.Id, DateTime.UtcNow); } + catch { } + } + } + } + + private sealed class RecordingInbox(string self) : IBackhaulInbox + { + private readonly object _l = new(); + private readonly HashSet _seqs = new(); + private long _delivered; + + public long Delivered { get { lock (_l) return _delivered; } } + public int DistinctSeqs { get { lock (_l) return _seqs.Count; } } + public bool HasSeq(int seq) { lock (_l) return _seqs.Contains(seq); } + + public Task DeliverAsync(BackhaulMessage message, string sourceCallsign, CancellationToken ct) + { + // Broadcast medium: only count messages actually addressed to us. + if (!string.Equals(message.Destination, self, StringComparison.OrdinalIgnoreCase)) + return Task.CompletedTask; + lock (_l) + { + _delivered++; + if (message.Headers is not null && message.Headers.TryGetValue("seq", out var sv) && int.TryParse(sv, out var s)) + _seqs.Add(s); + } + return Task.CompletedTask; + } + } +} diff --git a/src/dapps/dapps.meshcore.sim/MeshFabric.cs b/src/dapps/dapps.meshcore.sim/MeshFabric.cs new file mode 100644 index 0000000..221d05e --- /dev/null +++ b/src/dapps/dapps.meshcore.sim/MeshFabric.cs @@ -0,0 +1,174 @@ +using System.Security.Cryptography; + +namespace dapps.meshcore.sim; + +/// Node behaviour in the mesh, mirroring MeshCore firmware roles. +public enum MeshRole +{ + /// A Companion node (a DAPPS node): originates + receives channel traffic + /// but does NOT relay floods (client_repeat=0). + Leaf, + + /// A Repeater/Room-Server: relays floods it hears (subject to hop cap + + /// flood-scope), but has no DAPPS app of its own. + Relay, +} + +/// +/// An in-process model of a multi-hop MeshCore mesh. Nodes are connected by +/// (optionally lossy) undirected edges; a datagram broadcast by one node floods +/// hop-by-hop across the graph, mirroring the firmware's verified behaviour: +/// +/// +/// Every node that hears a packet delivers it to its app (broadcast; the DAPPS +/// inbox self-selects the addressee) - relays have no app so only leaves consume. +/// Per-node dedup by packet identity (the 160-entry no-expiry ring): a packet +/// heard again via another path is dropped, not re-delivered or re-flooded. +/// Only nodes re-flood, and only within the 64-hop +/// cap. +/// Flood-scope: a scoped flood is re-flooded only by relays that share the +/// origin's scope, so nodes reachable only through an out-of-scope relay never hear +/// it (deployment model B containment). +/// +/// +/// The bearer's per-frame rolling nonce means two distinct messages never share +/// a packet id (both flood); a single flood arriving by multiple paths has identical +/// bytes (one id) and is deduped - exactly the property we want to test at scale. +/// +public sealed class MeshFabric +{ + private sealed class Node + { + public required string Id; + public required MeshRole Role; + public string Scope = ""; + public SimulatedMeshCoreLink Link = null!; + public readonly List<(string to, double loss)> Neighbours = []; + public readonly HashSet Seen = new(StringComparer.Ordinal); // packet ids (the dedup ring) + } + + /// Firmware flood hop cap. + public const int HopCap = 64; + + private readonly Dictionary _nodes = new(StringComparer.Ordinal); + private readonly object _lock = new(); + private readonly Random _rng; + private readonly sbyte _snr; + + /// Total packet-deliveries the fabric has made (a hop that a node heard a + /// new packet on), and total re-floods - coarse observability for scenarios. + public long Deliveries { get; private set; } + public long Refloods { get; private set; } + + /// Seed for the loss RNG so scenarios are reproducible. + /// Reported SNR in quarter-dB (default 40 = 10 dB). + public MeshFabric(int seed = 1, sbyte snrQuarterDb = 40) + { + _rng = new Random(seed); + _snr = snrQuarterDb; + } + + public SimulatedMeshCoreLink AddLeaf(string id, string scope = "") => Add(id, MeshRole.Leaf, scope); + public SimulatedMeshCoreLink AddRelay(string id, string scope = "") => Add(id, MeshRole.Relay, scope); + + private SimulatedMeshCoreLink Add(string id, MeshRole role, string scope) + { + if (_nodes.ContainsKey(id)) throw new ArgumentException($"duplicate node id '{id}'"); + var node = new Node { Id = id, Role = role, Scope = scope }; + node.Link = new SimulatedMeshCoreLink(this, id); + _nodes[id] = node; + return node.Link; + } + + /// Connect two nodes with an undirected edge. is the + /// per-transmission drop probability on that edge (0 = perfect, 1 = never delivers). + public void Connect(string a, string b, double loss = 0.0) + { + if (a == b) throw new ArgumentException("cannot connect a node to itself"); + var na = Get(a); + var nb = Get(b); + na.Neighbours.Add((b, loss)); + nb.Neighbours.Add((a, loss)); + } + + /// Build a linear chain a0 - a1 - ... - a(n-1) with the given ids and a + /// uniform per-edge loss. Convenience for the common "string of relays" topology. + public void ConnectChain(IReadOnlyList ids, double loss = 0.0) + { + for (var i = 0; i + 1 < ids.Count; i++) Connect(ids[i], ids[i + 1], loss); + } + + public SimulatedMeshCoreLink Link(string id) => Get(id).Link; + + private Node Get(string id) => + _nodes.TryGetValue(id, out var n) ? n : throw new KeyNotFoundException($"unknown node '{id}'"); + + /// Flood a payload originated by across the mesh. + /// Called by . Runs synchronously so a + /// no-loss scenario is fully deterministic. + internal void Flood(string fromNode, byte[] payload) + { + lock (_lock) + { + var origin = Get(fromNode); + var scope = origin.Scope; + var packetId = PacketId(payload); + // The origin doesn't hear its own transmission, but remembers it so a flood + // that loops back is deduped. + origin.Seen.Add(packetId); + + // BFS over relays that re-flood. Track path_len exactly as the firmware does: + // the origin transmits with path_len=0 (it's not a forwarder); a relay that + // RECEIVED path_len P forwards iff P < HopCap (MAX_PATH_SIZE=64), appending + // itself so its transmission carries P+1. So relays R1..R64 forward and R65 + // (which would receive path_len=64) is dropped. A relay re-floods a packet + // at most once (the dedup ring). + var reflood = new Queue<(Node relay, int receivedPathLen)>(); + Transmit(origin, receiversPathLen: 0, packetId, scope, payload, reflood); + while (reflood.Count > 0) + { + var (relay, receivedPathLen) = reflood.Dequeue(); + if (receivedPathLen >= HopCap) continue; // path full - can't forward further + Transmit(relay, receiversPathLen: receivedPathLen + 1, packetId, scope, payload, reflood); + } + } + } + + // One transmission from `tx`: every neighbour may hear it (subject to loss), dedups, + // delivers-if-a-leaf, and re-floods-if-an-in-scope-relay. `receiversPathLen` is the + // path_len value the neighbours receive (the count of forwarders before them). + private void Transmit(Node tx, int receiversPathLen, string packetId, string scope, byte[] payload, Queue<(Node, int)> reflood) + { + foreach (var (toId, loss) in tx.Neighbours) + { + if (loss > 0 && _rng.NextDouble() < loss) continue; // lost on this edge + var nbr = _nodes[toId]; + + // Overhearing is a PHY event: it happens whenever RF arrives, even for a + // duplicate we'll dedup - that's what the occupancy estimate should see. + nbr.Link.Overhear(payload.Length, _snr); + + if (!nbr.Seen.Add(packetId)) continue; // dedup ring: already processed this packet + + // Deliver to the app only for leaves (relays are infrastructure with no app). + if (nbr.Role == MeshRole.Leaf) + { + nbr.Link.Deliver(payload, _snr); + Deliveries++; + } + + // Re-flood only from relays, and only if the flood's scope is carried: unscoped + // floods propagate through any relay; a scoped flood only through relays that + // share the scope (model B containment). + bool scopeCarried = scope.Length == 0 || nbr.Scope == scope; + if (nbr.Role == MeshRole.Relay && scopeCarried) + { + Refloods++; + reflood.Enqueue((nbr, receiversPathLen)); // the path_len this relay received + } + } + } + + private static string PacketId(byte[] payload) => + Convert.ToHexString(SHA256.HashData(payload)); +} diff --git a/src/dapps/dapps.meshcore.sim/README.md b/src/dapps/dapps.meshcore.sim/README.md new file mode 100644 index 0000000..68afd56 --- /dev/null +++ b/src/dapps/dapps.meshcore.sim/README.md @@ -0,0 +1,63 @@ +# dapps.meshcore.sim — simulated multi-hop MeshCore fabric + +An in-process model of a **multi-hop MeshCore mesh** that backs the **real** bearer +classes (`MeshCoreCompanionBackhaul` + `MeshCoreInbound` + transport + reliability) via +the `IMeshCoreLink` seam. It exists to test the things two directly-adjacent bench +radios **can't** reach — multi-hop flood propagation, dedup across paths, flood-scope +containment, reliability over several hops, and passive-discovery convergence — with +**no RF**, deterministically, in CI. + +## Why + +The bench has two Companion radios that hear each other in one hop. The MeshCore design +centre is the opposite: sparse DAPPS nodes with long strings of relays between them. The +firmware is a fixed, source-verified dependency (64-hop cap, 160-entry no-expiry dedup +ring, transport-code scope drop), so what actually needs scale-testing is **our** logic +running on top of that behaviour. This fabric reproduces the verified firmware behaviour +and runs the unchanged bearer stack over arbitrary topologies. + +## Pieces + +- **`MeshFabric`** — nodes (`Leaf` = a DAPPS Companion; `Relay` = a Repeater/Room-Server) + connected by optionally-lossy undirected edges. `Flood` propagates a datagram hop by + hop: every node that hears it delivers to its app (leaves only), dedups by packet id, + and re-floods only if it's an in-scope relay within the hop cap. The bearer's per-frame + rolling nonce means distinct messages never collide (both flood) while one flood + arriving by two paths is a single id (deduped) — the exact property under test. +- **`SimulatedMeshCoreLink : IMeshCoreLink`** — `SendDataAsync` hands the payload to the + fabric; received datagrams queue for `DrainAsync`; `MessageWaiting`/`PacketHeard` fire + as the real link's do. Drop-in for `MeshCoreLink`. +- **`MeshDappsNode`** — a full DAPPS node (the real backhaul + inbound + reliability + + discovery) on a fabric leaf, with a recording inbox. `SendAsync` originates traffic; + `Delivered` / `DistinctSeqs` / `DiscoveredPeers` expose what it received/learned. + +## Example + +```csharp +var f = new MeshFabric(); +f.AddRelay("R1"); f.AddRelay("R2"); f.ConnectChain(["R1", "R2"]); +var a = new MeshDappsNode(f, "GB7A-1"); +var b = new MeshDappsNode(f, "GB7B-1"); +f.Connect("GB7A-1", "R1"); // A ── R1 ── R2 ── B (two relay hops) +f.Connect("GB7B-1", "R2"); + +using var cts = new CancellationTokenSource(); +var loops = Task.WhenAll(a.RunAsync(cts.Token), b.RunAsync(cts.Token)); +await a.SendAsync("GB7B-1", seq: 0, "hello over the mesh", cts.Token); +// ... await delivery, then assert b.Delivered / b.DiscoveredPeers ... +cts.Cancel(); +``` + +See `MeshFabricTests` (fast, deterministic flood/dedup/scope + transport-over-multi-hop) +and `MeshFabricScenarioTests` (the full bearer stack over a relay backbone) in +`dapps.core.tests`. + +## Scope / limits + +- Models the **channel**, not the serial link's liveness — the simulated link is always + healthy (watchdog/recovery is out of scope). +- No-loss runs are fully deterministic and fast; **loss-recovery** exercises reliability + resends on a multi-second cadence, so drive those from a longer harness run rather than + a CI unit test. +- Propagation is instantaneous (no per-hop latency model yet) — correctness (delivery, + dedup, containment) is faithful; fine-grained timing/contention is not. diff --git a/src/dapps/dapps.meshcore.sim/SimulatedMeshCoreLink.cs b/src/dapps/dapps.meshcore.sim/SimulatedMeshCoreLink.cs new file mode 100644 index 0000000..431edc9 --- /dev/null +++ b/src/dapps/dapps.meshcore.sim/SimulatedMeshCoreLink.cs @@ -0,0 +1,64 @@ +using System.Collections.Concurrent; +using dapps.meshcore; + +namespace dapps.meshcore.sim; + +/// +/// An backed by a instead of a +/// serial radio. SendDataAsync hands the payload to the fabric to flood across the +/// virtual mesh; the fabric calls / back on +/// the nodes that hear it. The real bearer classes run on top unchanged. +/// +public sealed class SimulatedMeshCoreLink : IMeshCoreLink +{ + private readonly MeshFabric _fabric; + private readonly ConcurrentQueue _inbound = new(); + + /// Cap the per-node inbound backlog so a busy/relay node can't grow the + /// queue without bound in a long run. Oldest are dropped (as a real radio would). + private const int MaxBacklog = 4096; + + public string NodeId { get; } + + public event Action? MessageWaiting; + public event Action? PacketHeard; + + // The simulated radio is always healthy; watchdog/recovery is out of scope for the + // fabric (it models the channel, not the serial link's liveness). + public MeshCoreLink.LinkState State => MeshCoreLink.LinkState.Healthy; + + /// Total datagrams this node has been given to send (observability). + public long Sent { get; private set; } + + internal SimulatedMeshCoreLink(MeshFabric fabric, string nodeId) + { + _fabric = fabric; + NodeId = nodeId; + } + + public Task SendDataAsync(byte[] payload, CancellationToken ct) + { + Sent++; + _fabric.Flood(NodeId, payload); + return Task.FromResult(true); + } + + public Task DrainAsync(CancellationToken ct) + { + if (_inbound.IsEmpty) return Task.FromResult(null); + var data = new List(); + while (_inbound.TryDequeue(out var d)) data.Add(d); + return Task.FromResult(new MeshCoreClient.InboundBatch(new(), data)); + } + + /// Fabric hook: queue a received datagram for the app and wake the drain loop. + internal void Deliver(byte[] payload, sbyte snr) + { + _inbound.Enqueue(new ChannelData(snr, 0, 0xFF, MeshCoreClient.DATA_TYPE_DEV, payload)); + while (_inbound.Count > MaxBacklog && _inbound.TryDequeue(out _)) { } + MessageWaiting?.Invoke(); + } + + /// Fabric hook: this node overheard a packet on the channel (occupancy). + internal void Overhear(int len, sbyte snr) => PacketHeard?.Invoke(len, snr); +} diff --git a/src/dapps/dapps.meshcore.sim/dapps.meshcore.sim.csproj b/src/dapps/dapps.meshcore.sim/dapps.meshcore.sim.csproj new file mode 100644 index 0000000..8f629ec --- /dev/null +++ b/src/dapps/dapps.meshcore.sim/dapps.meshcore.sim.csproj @@ -0,0 +1,18 @@ + + + + + net8.0 + enable + enable + + + + + + + diff --git a/src/dapps/dapps.meshcore/IMeshCoreLink.cs b/src/dapps/dapps.meshcore/IMeshCoreLink.cs new file mode 100644 index 0000000..b57c921 --- /dev/null +++ b/src/dapps/dapps.meshcore/IMeshCoreLink.cs @@ -0,0 +1,29 @@ +namespace dapps.meshcore; + +/// +/// The data-path surface the bearer needs from a MeshCore radio link: +/// send a channel-data datagram, drain received ones, and observe wake/occupancy +/// signals. is the real serial implementation; +/// a simulated implementation (dapps.meshcore.sim) backs an in-process multi-hop +/// mesh so the real bearer classes (transport, reliability, inbound, governor) +/// can be tested at scale without radios. +/// +public interface IMeshCoreLink +{ + /// Broadcast one channel-data payload as a flood. Returns false if the + /// link isn't ready (the caller refunds airtime and reports "not ready"). + Task SendDataAsync(byte[] payload, CancellationToken ct); + + /// Drain all queued inbound datagrams, or null if none. + Task DrainAsync(CancellationToken ct); + + /// Raised when a datagram becomes available to drain (inbound wake). + event Action? MessageWaiting; + + /// Raised when a packet is overheard on the channel (len, snr) - feeds + /// the occupancy estimate for the adaptive controls. + event Action? PacketHeard; + + /// Current link health (used in diagnostics / send-not-ready messages). + MeshCoreLink.LinkState State { get; } +} diff --git a/src/dapps/dapps.meshcore/MeshCoreCompanionBackhaul.cs b/src/dapps/dapps.meshcore/MeshCoreCompanionBackhaul.cs index 58e1819..4e04457 100644 --- a/src/dapps/dapps.meshcore/MeshCoreCompanionBackhaul.cs +++ b/src/dapps/dapps.meshcore/MeshCoreCompanionBackhaul.cs @@ -21,7 +21,7 @@ public sealed class MeshCoreCompanionBackhaul : IDappsBackhaul private static readonly TimeSpan CoalesceWindow = TimeSpan.FromSeconds(30); private static readonly TimeSpan FramePace = TimeSpan.FromMilliseconds(1000); - private readonly MeshCoreLink _link; + private readonly IMeshCoreLink _link; private readonly MeshCoreBearerOptions _opts; private readonly TxBudget _budget; private readonly IDappsTxGate _txGate; @@ -35,7 +35,7 @@ public sealed class MeshCoreCompanionBackhaul : IDappsBackhaul private readonly object _recentLock = new(); public MeshCoreCompanionBackhaul( - MeshCoreLink link, MeshCoreBearerOptions opts, TxBudget budget, ILogger log, + IMeshCoreLink link, MeshCoreBearerOptions opts, TxBudget budget, ILogger log, IDappsTxGate? txGate = null, MeshCoreReliability? reliability = null) { _link = link; diff --git a/src/dapps/dapps.meshcore/MeshCoreInbound.cs b/src/dapps/dapps.meshcore/MeshCoreInbound.cs index ba75d7f..641f80c 100644 --- a/src/dapps/dapps.meshcore/MeshCoreInbound.cs +++ b/src/dapps/dapps.meshcore/MeshCoreInbound.cs @@ -18,7 +18,7 @@ public sealed class MeshCoreInbound private static readonly TimeSpan ReassemblyTimeout = TimeSpan.FromMinutes(2); private static readonly TimeSpan SweepInterval = TimeSpan.FromMinutes(1); - private readonly MeshCoreLink _link; + private readonly IMeshCoreLink _link; private readonly IBackhaulInbox _inbox; private readonly ILogger _log; private readonly MeshCoreChannelTransport _rx = new(); @@ -41,7 +41,7 @@ public sealed class MeshCoreInbound public long Delivered { get; private set; } public MeshCoreInbound( - MeshCoreLink link, IBackhaulInbox inbox, ILogger log, + IMeshCoreLink link, IBackhaulInbox inbox, ILogger log, MeshCoreReliability? reliability = null, Func>? sendAck = null, string? localCallsign = null, diff --git a/src/dapps/dapps.meshcore/MeshCoreLink.cs b/src/dapps/dapps.meshcore/MeshCoreLink.cs index 0923565..c312801 100644 --- a/src/dapps/dapps.meshcore/MeshCoreLink.cs +++ b/src/dapps/dapps.meshcore/MeshCoreLink.cs @@ -13,7 +13,7 @@ namespace dapps.meshcore; /// callers see the link's current state; during a reset they get a soft failure /// (send returns false, drain returns null) and retry once the link is back. /// -public sealed class MeshCoreLink : IAsyncDisposable +public sealed class MeshCoreLink : IAsyncDisposable, IMeshCoreLink { public enum LinkState { Down, Healthy, Resetting, Failed } diff --git a/src/dapps/dapps.sln b/src/dapps/dapps.sln index de22580..9b709b8 100644 --- a/src/dapps/dapps.sln +++ b/src/dapps/dapps.sln @@ -23,6 +23,8 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "dapps.meshcore", "dapps.mes EndProject Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "dapps.meshcore.soak", "dapps.meshcore.soak\dapps.meshcore.soak.csproj", "{3332047B-7143-461B-9D9C-1FDCBAAD00AD}" EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "dapps.meshcore.sim", "dapps.meshcore.sim\dapps.meshcore.sim.csproj", "{C5A5C8EA-BE48-41D9-9608-D0AEAB2C0948}" +EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU @@ -117,6 +119,18 @@ Global {3332047B-7143-461B-9D9C-1FDCBAAD00AD}.Release|x64.Build.0 = Release|Any CPU {3332047B-7143-461B-9D9C-1FDCBAAD00AD}.Release|x86.ActiveCfg = Release|Any CPU {3332047B-7143-461B-9D9C-1FDCBAAD00AD}.Release|x86.Build.0 = Release|Any CPU + {C5A5C8EA-BE48-41D9-9608-D0AEAB2C0948}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {C5A5C8EA-BE48-41D9-9608-D0AEAB2C0948}.Debug|Any CPU.Build.0 = Debug|Any CPU + {C5A5C8EA-BE48-41D9-9608-D0AEAB2C0948}.Debug|x64.ActiveCfg = Debug|Any CPU + {C5A5C8EA-BE48-41D9-9608-D0AEAB2C0948}.Debug|x64.Build.0 = Debug|Any CPU + {C5A5C8EA-BE48-41D9-9608-D0AEAB2C0948}.Debug|x86.ActiveCfg = Debug|Any CPU + {C5A5C8EA-BE48-41D9-9608-D0AEAB2C0948}.Debug|x86.Build.0 = Debug|Any CPU + {C5A5C8EA-BE48-41D9-9608-D0AEAB2C0948}.Release|Any CPU.ActiveCfg = Release|Any CPU + {C5A5C8EA-BE48-41D9-9608-D0AEAB2C0948}.Release|Any CPU.Build.0 = Release|Any CPU + {C5A5C8EA-BE48-41D9-9608-D0AEAB2C0948}.Release|x64.ActiveCfg = Release|Any CPU + {C5A5C8EA-BE48-41D9-9608-D0AEAB2C0948}.Release|x64.Build.0 = Release|Any CPU + {C5A5C8EA-BE48-41D9-9608-D0AEAB2C0948}.Release|x86.ActiveCfg = Release|Any CPU + {C5A5C8EA-BE48-41D9-9608-D0AEAB2C0948}.Release|x86.Build.0 = Release|Any CPU EndGlobalSection GlobalSection(SolutionProperties) = preSolution HideSolutionNode = FALSE