diff --git a/docs/app-developers/reference.md b/docs/app-developers/reference.md index fb5bb0c..c3be6d2 100644 --- a/docs/app-developers/reference.md +++ b/docs/app-developers/reference.md @@ -203,6 +203,9 @@ DAPPS does not impose a hard payload-size limit at the app interface - submit an | Read source callsign | `dapps-source` user property | `sourceCallsign` field | | Read residual TTL | `dapps-ttl` user property (absent if no TTL) | `ttl` field (`null` if no TTL) | | Set TTL on submit | `dapps-ttl` user property | `ttl` field | +| Opt into ordered delivery | `dapps-stream` user property | `streamId` field | +| Set ordering gap policy | `dapps-stream-gap-timeout` user property | `streamGapTimeoutSeconds` field | +| Read delivered stream / seq | `dapps-stream`, `dapps-stream-seq` properties | (not surfaced on REST inbound today) | ## DAPPSv1 wire format (summary) @@ -218,7 +221,7 @@ Then a back-and-forth of one-line commands and responses: | Command | Direction | Meaning | |-------------------------------------------------|------------------|------------------------------------------------------| -| `ihave id=<7hex> dst= sz= ttl= [src=] [mid= frag=/]` | sender → receiver | "I have this message; do you want it?" | +| `ihave id=<7hex> dst= sz= ttl= [src=] [mid= frag=/] [sid= sn= gt=]` | sender → receiver | "I have this message; do you want it?" | | `send` | receiver → sender | "Yes, send it." | | `?` | receiver → sender | "Already have it / don't recognise this command." | | `data ` | sender → receiver | The payload, exactly `` bytes. | @@ -228,10 +231,66 @@ Then a back-and-forth of one-line commands and responses: | `end` | reply | End of `peers` response. | | `rev [,...]` | either | "Send me anything you're holding for these callsigns." | -Headers on `ihave` are forward-compatible - receivers ignore unknown ones. New optional fields (e.g. `src=` for source tracking, `mid=` + `frag=N/M` for multi-part) ride the existing `DAPPSv1>` prompt. Breaking changes bump the prompt to `DAPPSv2>`. +Headers on `ihave` are forward-compatible - receivers ignore unknown ones. New optional fields (e.g. `src=` for source tracking, `mid=` + `frag=N/M` for multi-part, `sid=`/`sn=`/`gt=` for opt-in ordering) ride the existing `DAPPSv1>` prompt. Breaking changes bump the prompt to `DAPPSv2>`. The full wire spec lives in the [main repository README](https://github.com/M0LTE/dapps/blob/master/README.md#on-air-protocol). +## Message ordering (opt-in) + +DAPPS doesn't order messages by default. Each submission is independent; under retries and routing reconvergence the receiver can see them in any order. For most apps this is correct: idempotent or content-addressed work doesn't care. + +When an app does care - chat transcripts, telemetry sequences, change-log streams - opt-in ordering is available. Setting `streamId` on a submission tags the message as part of a per-sender ordered stream; the daemon mints a monotonic sequence number and the receiving daemon delivers messages on that stream in submit order. + +### Opting in + +REST: + +```bash +curl -sS -X POST http://localhost:5086/AppApi/outbound \ + -H 'content-type: application/json' \ + -d '{ + "app": "chat", + "destCallsign": "M0LTE", + "payload": "aGVsbG8=", + "streamId": "c1", + "streamGapTimeoutSeconds": 600 + }' +``` + +MQTT: + +``` +publish dapps/out/chat/M0LTE + user-property dapps-stream=c1 + user-property dapps-stream-gap-timeout=600 + payload +``` + +`streamId` is sender-scoped: two senders can pick the same id without colliding because the receiver keys its cursor on `(originator-callsign, streamId)`. Pick something short (it travels on every wire frame for that stream). + +`streamGapTimeoutSeconds` chooses the policy when a message is missing: + +- **`0` (default, "strict")**: stall forever waiting for the missing seq. Later messages park until the gap fills. Use when you'd rather wait than skip. +- **`>0` ("timeout")**: stall for that many seconds, then skip past the gap and deliver waiting messages. Use when stale data is worse than missing data. + +### What the receiver sees + +Inbound messages tagged with a stream show two extra MQTT user properties: + +- `dapps-stream` - the stream id the sender chose. +- `dapps-stream-seq` - the seq within that stream, ascending. + +Apps that don't care can ignore them; apps that opted in can use them to detect stream id changes (the sender rotated to a fresh stream after a reset) or to assert seq monotonicity for their own bookkeeping. + +### Tradeoffs + +- **Latency cost**. One missing message stalls the whole stream until it arrives or the timeout fires. On lossy radio links this is real - opt-in is the right default. +- **Sender resets**. The sender persists its counter to disk; a fresh install / wiped database starts back at `sn=1`. Re-using the same `streamId` after a reset will cause receivers to drop the new messages as `stream-stale` (their cursor is well past `sn=1`). Mitigate by appending a short epoch suffix to the stream id when you reset (e.g. `chat:tom.2`). +- **End-to-end semantics**. Ordering is enforced at the receiving daemon, not at intermediate forwarders. Hops can reorder, retry, and flood freely - the trio rides the envelope verbatim. +- **Forward compatibility**. A daemon that doesn't understand `sid`/`sn`/`gt` ignores the keys and delivers each message immediately. Apps subscribed to a partially-ordering-aware mesh see ordered delivery only between aware nodes. + +The dashboard's `/Streams` page surfaces both sender-side counters and receiver-side cursors plus pending row counts; a stalled stream shows up as a non-empty pending column. + ## See also - [Concepts](concepts.md) - the mental model. diff --git a/src/dapps/dapps.client/Backhaul/BackhaulMessage.cs b/src/dapps/dapps.client/Backhaul/BackhaulMessage.cs index 3820169..1a02b4c 100644 --- a/src/dapps/dapps.client/Backhaul/BackhaulMessage.cs +++ b/src/dapps/dapps.client/Backhaul/BackhaulMessage.cs @@ -26,7 +26,10 @@ public sealed record BackhaulMessage( IReadOnlyList? TraversedHops = null, string? MasterId = null, int? FragmentIndex = null, - int? FragmentTotal = null); + int? FragmentTotal = null, + string? StreamId = null, + uint? StreamSeq = null, + uint? StreamGapTimeoutSeconds = null); // LinkSourceCallsign: the *immediate sender's* callsign, distinct from // Originator (the F1 end-to-end source). Carried on bearers that don't @@ -72,3 +75,13 @@ public sealed record BackhaulMessage( // `mid=…` + `frag=N/M` headers); the receiver's IHaveValidator // rejects any mismatched-presence combination. FragmentTotal ≥ 2; // single-fragment messages just omit all three fields. +// +// StreamId / StreamSeq / StreamGapTimeoutSeconds (opt-in ordering): +// when StreamSeq is set the message is part of a per-sender ordered +// stream identified by StreamId. The receiver delivers messages on +// each (sender-callsign, StreamId) cursor in monotonically-increasing +// StreamSeq order; gaps stall until the missing seq arrives or +// StreamGapTimeoutSeconds elapses (gt=0 = strict, never skip). +// Wire form: `sid=`, `sn=`, `gt=` keys on the ihave line; codec flag +// bit 9 on datagram bearers. All three are required together when +// any one is set; intermediate forwarders preserve them verbatim. diff --git a/src/dapps/dapps.client/Backhaul/Dappsv1SessionBackhaul.cs b/src/dapps/dapps.client/Backhaul/Dappsv1SessionBackhaul.cs index ddbcc59..0201d22 100644 --- a/src/dapps/dapps.client/Backhaul/Dappsv1SessionBackhaul.cs +++ b/src/dapps/dapps.client/Backhaul/Dappsv1SessionBackhaul.cs @@ -81,7 +81,10 @@ public async Task SendAsync( originator: message.Originator, masterId: message.MasterId, fragmentIndex: message.FragmentIndex, - fragmentTotal: message.FragmentTotal)) + fragmentTotal: message.FragmentTotal, + streamId: message.StreamId, + streamSeq: message.StreamSeq, + streamGapTimeoutSeconds: message.StreamGapTimeoutSeconds)) { return BackhaulSendResult.Fail($"offer rejected for {message.Id}"); } @@ -112,7 +115,10 @@ public async Task SendAsync( Originator: polled.Originator, MasterId: polled.MasterId, FragmentIndex: polled.FragmentIndex, - FragmentTotal: polled.FragmentTotal); + FragmentTotal: polled.FragmentTotal, + StreamId: polled.StreamId, + StreamSeq: polled.StreamSeq, + StreamGapTimeoutSeconds: polled.StreamGapTimeoutSeconds); await opportunisticInbox.DeliverAsync(inbound, route.Callsign, ct); } } diff --git a/src/dapps/dapps.client/Backhaul/Datagram/BackhaulMessageCodec.cs b/src/dapps/dapps.client/Backhaul/Datagram/BackhaulMessageCodec.cs index f1f330d..2b8fcd5 100644 --- a/src/dapps/dapps.client/Backhaul/Datagram/BackhaulMessageCodec.cs +++ b/src/dapps/dapps.client/Backhaul/Datagram/BackhaulMessageCodec.cs @@ -23,7 +23,8 @@ namespace dapps.client.Backhaul.Datagram; /// bit3=originator, bit4=link-source, /// bit5=flood-hops-remaining, /// bit6=source-route, bit7=traversed-hops, -/// bit8=fragment (F2 multi-part) +/// bit8=fragment (F2 multi-part), +/// bit9=stream (opt-in ordering) /// [7] id (UTF-8 ASCII, 7-char hex from DappsMessage.ComputeHash) /// [8] salt (only when flags bit0) /// [4] ttl seconds (only when flags bit1) @@ -43,6 +44,10 @@ namespace dapps.client.Backhaul.Datagram; /// [7] master id (only when flags bit8; ASCII) /// [2] fragment index (only when flags bit8; UInt16, 1-based) /// [2] fragment total (only when flags bit8; UInt16) +/// [1] stream id len (only when flags bit9; max 255 bytes) +/// [S] stream id (only when flags bit9; UTF-8) +/// [4] stream seq (only when flags bit9; UInt32 LE) +/// [4] stream gap timeout (only when flags bit9; UInt32 LE seconds, 0=strict) /// [2] headers count (only when flags bit2) /// per header: /// [2] key len, [K] key (UTF-8), [2] value len, [V] value (UTF-8) @@ -56,7 +61,7 @@ public static class BackhaulMessageCodec /// Version this encoder writes AND the only version the /// decoder accepts. Bump on any wire-format change so a mismatched /// peer fails fast instead of silently misreading flag bits. - public const byte Version = 6; + public const byte Version = 7; public const int IdLength = 7; [Flags] @@ -72,6 +77,7 @@ private enum Flags : ushort HasSourceRoute = 1 << 6, HasTraversedHops = 1 << 7, HasFragment = 1 << 8, + HasStream = 1 << 9, } public static byte[] Encode(BackhaulMessage message) @@ -102,6 +108,28 @@ public static byte[] Encode(BackhaulMessage message) throw new ArgumentException("fragment index/total without master id", nameof(message)); } + // Opt-in ordering trio: present together or absent together. The + // receiver enforces this in IHaveValidator too, but a relay that + // forwards a partial set would silently drop ordering for the + // downstream hop, which is worse than failing fast here. + var hasStream = !string.IsNullOrEmpty(message.StreamId) + || message.StreamSeq.HasValue + || message.StreamGapTimeoutSeconds.HasValue; + if (hasStream + && (string.IsNullOrEmpty(message.StreamId) + || !message.StreamSeq.HasValue + || !message.StreamGapTimeoutSeconds.HasValue)) + { + throw new ArgumentException( + "stream id/seq/gap-timeout must all be set together (opt-in ordering) or all be absent", + nameof(message)); + } + var streamIdBytes = hasStream ? Encoding.UTF8.GetBytes(message.StreamId!) : []; + if (hasStream && streamIdBytes.Length > byte.MaxValue) + { + throw new ArgumentException("stream id exceeds 255 bytes", nameof(message)); + } + var idBytes = Encoding.ASCII.GetBytes(message.Id); var dstBytes = Encoding.UTF8.GetBytes(message.Destination); var origBytes = string.IsNullOrEmpty(message.Originator) @@ -133,6 +161,7 @@ public static byte[] Encode(BackhaulMessage message) if (sourceRouteBytes.Length > 0) flags |= Flags.HasSourceRoute; if (traversedBytes.Length > 0) flags |= Flags.HasTraversedHops; if (hasFragment) flags |= Flags.HasFragment; + if (hasStream) flags |= Flags.HasStream; var size = 1 + 2 + IdLength + (message.Salt.HasValue ? 8 : 0) @@ -144,6 +173,7 @@ public static byte[] Encode(BackhaulMessage message) + sourceRouteBytes.Length + traversedBytes.Length + (hasFragment ? IdLength + 2 + 2 : 0) + + (hasStream ? 1 + streamIdBytes.Length + 4 + 4 : 0) + headerBytes.Length + 4 + message.Payload.Length; @@ -216,6 +246,17 @@ public static byte[] Encode(BackhaulMessage message) offset += 2; } + if (hasStream) + { + buffer[offset++] = (byte)streamIdBytes.Length; + streamIdBytes.CopyTo(buffer.AsSpan(offset)); + offset += streamIdBytes.Length; + BinaryPrimitives.WriteUInt32LittleEndian(buffer.AsSpan(offset, 4), message.StreamSeq!.Value); + offset += 4; + BinaryPrimitives.WriteUInt32LittleEndian(buffer.AsSpan(offset, 4), message.StreamGapTimeoutSeconds!.Value); + offset += 4; + } + headerBytes.CopyTo(buffer.AsSpan(offset)); offset += headerBytes.Length; @@ -313,6 +354,20 @@ public static BackhaulMessage Decode(ReadOnlySpan buffer) offset += 2; } + string? streamId = null; + uint? streamSeq = null; + uint? streamGapTimeout = null; + if ((flags & Flags.HasStream) != 0) + { + var sidLen = buffer[offset++]; + streamId = Encoding.UTF8.GetString(buffer.Slice(offset, sidLen)); + offset += sidLen; + streamSeq = BinaryPrimitives.ReadUInt32LittleEndian(buffer.Slice(offset, 4)); + offset += 4; + streamGapTimeout = BinaryPrimitives.ReadUInt32LittleEndian(buffer.Slice(offset, 4)); + offset += 4; + } + IReadOnlyDictionary? headers = null; if ((flags & Flags.HasHeaders) != 0) { @@ -338,7 +393,7 @@ public static BackhaulMessage Decode(ReadOnlySpan buffer) offset += 4; var payload = buffer.Slice(offset, (int)payloadLen).ToArray(); - return new BackhaulMessage(id, destination, salt, ttl, payload, headers, originator, linkSource, floodHops, sourceRoute, traversedHops, masterId, fragmentIndex, fragmentTotal); + return new BackhaulMessage(id, destination, salt, ttl, payload, headers, originator, linkSource, floodHops, sourceRoute, traversedHops, masterId, fragmentIndex, fragmentTotal, streamId, streamSeq, streamGapTimeout); } private static byte[] EncodeHeaders(IReadOnlyDictionary headers) diff --git a/src/dapps/dapps.client/DappsProtocolClient.cs b/src/dapps/dapps.client/DappsProtocolClient.cs index 0d1faf3..81b47e7 100644 --- a/src/dapps/dapps.client/DappsProtocolClient.cs +++ b/src/dapps/dapps.client/DappsProtocolClient.cs @@ -87,7 +87,10 @@ public async Task OfferMessageAsync( string? originator = null, string? masterId = null, int? fragmentIndex = null, - int? fragmentTotal = null) + int? fragmentTotal = null, + string? streamId = null, + uint? streamSeq = null, + uint? streamGapTimeoutSeconds = null) { if (format != DappsMessage.MessageFormat.Plain) { @@ -132,6 +135,21 @@ public async Task OfferMessageAsync( { sb.Append($" mid={masterId} frag={fragmentIndex}/{fragmentTotal}"); } + // Opt-in ordering keys. All three travel together; the receiver's + // IHaveValidator rejects a partial set. Belt-and-braces sender- + // side: catch the malformed envelope before it reaches the wire. + var hasStream = !string.IsNullOrEmpty(streamId) + && streamSeq.HasValue && streamGapTimeoutSeconds.HasValue; + if (!hasStream + && (!string.IsNullOrEmpty(streamId) || streamSeq.HasValue || streamGapTimeoutSeconds.HasValue)) + { + throw new ArgumentException( + "streamId, streamSeq, streamGapTimeoutSeconds must all be set together (opt-in ordering) or all be null"); + } + if (hasStream) + { + sb.Append($" sid={streamId} sn={streamSeq} gt={streamGapTimeoutSeconds}"); + } sb.Append('\n'); await stream.WriteAsync(Encoding.UTF8.GetBytes(sb.ToString()), ct); @@ -244,7 +262,10 @@ public sealed record PolledMessage( string? Originator, string? MasterId, int? FragmentIndex, - int? FragmentTotal); + int? FragmentTotal, + string? StreamId, + uint? StreamSeq, + uint? StreamGapTimeoutSeconds); /// /// Plan F3 - reverse forwarding from the client side. Send @@ -340,7 +361,10 @@ public async IAsyncEnumerable PollAsync( Originator: parsed.Originator, MasterId: parsed.MasterId, FragmentIndex: parsed.FragmentIndex, - FragmentTotal: parsed.FragmentTotal); + FragmentTotal: parsed.FragmentTotal, + StreamId: parsed.StreamId, + StreamSeq: parsed.StreamSeq, + StreamGapTimeoutSeconds: parsed.StreamGapTimeoutSeconds); } } @@ -372,7 +396,7 @@ private async Task ReadExactlyAsync(byte[] buffer, CancellationToken ct) /// minimum fields aren't present. private static (bool Ok, ParsedOffer? Offer) TryParseOffer(string line) { - // line: "ihave len=N fmt=p dst=… [s=…] [ttl=…] [src=…] [mid=… frag=N/M] …" + // line: "ihave len=N fmt=p dst=… [s=…] [ttl=…] [src=…] [mid=… frag=N/M] [sid=… sn=… gt=…] …" var parts = line.Split(' '); if (parts.Length < 4 || parts[0] != "ihave") return (false, null); var id = parts[1]; @@ -384,6 +408,9 @@ private static (bool Ok, ParsedOffer? Offer) TryParseOffer(string line) string? masterId = null; int? fragIndex = null; int? fragTotal = null; + string? streamId = null; + uint? streamSeq = null; + uint? streamGapTimeout = null; for (var i = 2; i < parts.Length; i++) { var kv = parts[i]; @@ -409,15 +436,19 @@ private static (bool Ok, ParsedOffer? Offer) TryParseOffer(string line) fragIndex = fn; fragTotal = fm; } break; + case "sid": streamId = value; break; + case "sn": if (uint.TryParse(value, out var snv)) streamSeq = snv; break; + case "gt": if (uint.TryParse(value, out var gtv)) streamGapTimeout = gtv; break; } } - if (destination is null || len is null) return (false, new ParsedOffer(id, "", 0, null, null, null, null, null, null)); - return (true, new ParsedOffer(id, destination, len.Value, salt, ttl, originator, masterId, fragIndex, fragTotal)); + if (destination is null || len is null) return (false, new ParsedOffer(id, "", 0, null, null, null, null, null, null, null, null, null)); + return (true, new ParsedOffer(id, destination, len.Value, salt, ttl, originator, masterId, fragIndex, fragTotal, streamId, streamSeq, streamGapTimeout)); } private sealed record ParsedOffer( string Id, string Destination, int Length, long? Salt, int? Ttl, - string? Originator, string? MasterId, int? FragmentIndex, int? FragmentTotal); + string? Originator, string? MasterId, int? FragmentIndex, int? FragmentTotal, + string? StreamId, uint? StreamSeq, uint? StreamGapTimeoutSeconds); /// /// Reads a line terminated by \n, \r, or \r\n. diff --git a/src/dapps/dapps.client/IHaveCommand.cs b/src/dapps/dapps.client/IHaveCommand.cs index fa7421a..3ea7f3f 100644 --- a/src/dapps/dapps.client/IHaveCommand.cs +++ b/src/dapps/dapps.client/IHaveCommand.cs @@ -13,6 +13,14 @@ public class IHaveCommand /// public string? Originator { get; init; } + /// Opt-in ordering: stream id (per sender), monotonic + /// sequence, and gap timeout in seconds (0 = strict, >0 = skip + /// gap after N seconds). All three travel together or all three + /// are absent. + public string? StreamId { get; init; } + public uint? StreamSeq { get; init; } + public uint? StreamGapTimeoutSeconds { get; init; } + public static string Checksum(string ihave) => Crc16CcittFalse.ComputeHex(Encoding.UTF8.GetBytes(ihave)); @@ -24,6 +32,10 @@ public override string ToString() { sb.Append($" src={Originator}"); } + if (!string.IsNullOrEmpty(StreamId) && StreamSeq.HasValue && StreamGapTimeoutSeconds.HasValue) + { + sb.Append($" sid={StreamId} sn={StreamSeq} gt={StreamGapTimeoutSeconds}"); + } if (Message.Kvps.Count > 0) { sb.Append($" {string.Join(" ", Message.Kvps.Select(kvp => $"{kvp.Key}={kvp.Value}"))}"); diff --git a/src/dapps/dapps.core.tests/StreamOrderingTests.cs b/src/dapps/dapps.core.tests/StreamOrderingTests.cs new file mode 100644 index 0000000..6e9876f --- /dev/null +++ b/src/dapps/dapps.core.tests/StreamOrderingTests.cs @@ -0,0 +1,447 @@ +using System.Net; +using System.Net.Sockets; +using AwesomeAssertions; +using dapps.client; +using dapps.client.Backhaul; +using dapps.client.Backhaul.Datagram; +using dapps.core.Models; +using dapps.core.Routing; +using dapps.core.Services; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using Microsoft.Extensions.Time.Testing; +using MQTTnet; +using MQTTnet.Client; +using MQTTnet.Server; +using SQLite; + +namespace dapps.core.tests; + +/// +/// Opt-in message ordering: end-to-end coverage of the +/// sid=/sn=/gt= protocol additions and the +/// receiver-side reorder buffer + gap sweeper. +/// +/// Wire-level tests round-trip the new fields through both the binary +/// codec and the IHave text validator. Inbox tests drive +/// with synthetic +/// out-of-order deliveries against a real MQTT broker so we can +/// assert the actual delivery order observed by an app. +/// +[Collection(SqliteOverridePathCollection.Name)] +public sealed class StreamOrderingTests : IAsyncLifetime +{ + private string dbPath = null!; + private int brokerPort; + private FakeTimeProvider clock = null!; + private Database database = null!; + private MqttServer mqttServer = null!; + private MqttBrokerService broker = null!; + private DatabaseAndMqttInbox inbox = null!; + + public async ValueTask InitializeAsync() + { + brokerPort = PickFreeTcpPort(); + dbPath = Path.Combine(Path.GetTempPath(), $"dapps-stream-test-{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(); + } + + clock = new FakeTimeProvider(new DateTimeOffset(2026, 5, 5, 12, 0, 0, TimeSpan.Zero)); + var optionsMonitor = new TestOptionsMonitor(new SystemOptions + { + Callsign = "N0SELF", + MqttPort = brokerPort, + FragmentThresholdBytes = 0, + }); + database = new Database(NullLogger.Instance, optionsMonitor, clock); + var tokens = new AppTokenStore(NullLogger.Instance); + mqttServer = new MqttFactory().CreateMqttServer(new MqttServerOptionsBuilder() + .WithDefaultEndpoint().WithDefaultEndpointPort(brokerPort).Build()); + await mqttServer.StartAsync(); + broker = new MqttBrokerService(NullLogger.Instance, optionsMonitor, database, tokens, mqttServer); + var routingContext = new DatabaseRoutingContext(database, optionsMonitor); + var routingAlgorithm = new StaticRoutingAlgorithm(NullLogger.Instance); + inbox = new DatabaseAndMqttInbox(database, broker, new InboundEventBus(), + optionsMonitor, routingAlgorithm, routingContext, clock, + NullLogger.Instance); + + await broker.StartAsync(CancellationToken.None); + } + + public async ValueTask DisposeAsync() + { + await broker.StopAsync(CancellationToken.None); + await mqttServer.StopAsync(); + mqttServer.Dispose(); + DbInfo.OverridePath = null; + try { File.Delete(dbPath); } catch { /* ignore */ } + } + + // ── Wire format: codec ───────────────────────────────────────── + + [Fact] + public void Codec_RoundTrip_StreamFields_PreservesAllThree() + { + var input = new BackhaulMessage( + Id: "stream01", + Destination: "chat@N0DEST", + Salt: 1L, + Ttl: 600, + Payload: "ordered-msg"u8.ToArray(), + StreamId: "c1", + StreamSeq: 42u, + StreamGapTimeoutSeconds: 600u); + + var decoded = BackhaulMessageCodec.Decode(BackhaulMessageCodec.Encode(input with { Id = "stream0" })); + + decoded.StreamId.Should().Be("c1"); + decoded.StreamSeq.Should().Be(42u); + decoded.StreamGapTimeoutSeconds.Should().Be(600u); + } + + [Fact] + public void Codec_RoundTrip_StrictMode_Gt0Preserved() + { + var input = new BackhaulMessage( + Id: "stream0", + Destination: "chat@N0DEST", + Salt: null, + Ttl: null, + Payload: "x"u8.ToArray(), + StreamId: "s", + StreamSeq: 1u, + StreamGapTimeoutSeconds: 0u); + + var decoded = BackhaulMessageCodec.Decode(BackhaulMessageCodec.Encode(input)); + + decoded.StreamGapTimeoutSeconds.Should().Be(0u, "gt=0 is the strict marker and must round-trip distinctly from absent"); + decoded.StreamId.Should().Be("s"); + } + + [Fact] + public void Codec_PartialStreamTrio_Throws() + { + var bad = new BackhaulMessage( + Id: "stream0", + Destination: "chat@N0DEST", + Salt: null, + Ttl: null, + Payload: "x"u8.ToArray(), + StreamId: "s", + StreamSeq: 1u); + var act = () => BackhaulMessageCodec.Encode(bad); + act.Should().Throw().WithMessage("*stream*"); + } + + // ── Wire format: ihave parser ────────────────────────────────── + + [Fact] + public void Validator_ParsesAllThreeStreamKeys() + { + var line = "ihave 1234567 len=5 fmt=p s=1 sid=chat sn=42 gt=600 dst=app@N0DEST"; + var result = IHaveValidator.Validate(line); + result.IsValid.Should().BeTrue(); + result.Offer!.StreamId.Should().Be("chat"); + result.Offer.StreamSeq.Should().Be(42u); + result.Offer.StreamGapTimeoutSeconds.Should().Be(600u); + } + + [Fact] + public void Validator_RejectsPartialStreamSet() + { + var line = "ihave 1234567 len=5 fmt=p s=1 sid=chat sn=42 dst=app@N0DEST"; + var result = IHaveValidator.Validate(line); + result.IsValid.Should().BeFalse(); + result.Error.Should().Contain("sid=, sn=, gt="); + } + + [Fact] + public void Validator_RejectsNonNumericSn() + { + var line = "ihave 1234567 len=5 fmt=p s=1 sid=chat sn=foo gt=600 dst=app@N0DEST"; + var result = IHaveValidator.Validate(line); + result.IsValid.Should().BeFalse(); + result.Error.Should().Contain("sn="); + } + + [Fact] + public void IHaveCommand_EmitsAllThreeWhenSet() + { + var cmd = new IHaveCommand + { + Message = new DappsMessage + { + Payload = "hello"u8.ToArray(), + Destination = "chat@N0DEST", + Salt = 1L, + }, + StreamId = "c1", + StreamSeq = 7u, + StreamGapTimeoutSeconds = 0u, + }; + var line = cmd.ToString(); + line.Should().Contain("sid=c1"); + line.Should().Contain("sn=7"); + line.Should().Contain("gt=0"); + } + + // ── Inbound: out-of-order delivery ───────────────────────────── + + [Fact] + public async Task Inbox_OutOfOrderArrival_DeliversInOrderToMqtt() + { + // Three messages on the same (sender, stream) arrive sn=3, sn=2, sn=1. + // The inbox must hold 3 and 2 until 1 lands, then drain all three + // in 1, 2, 3 order onto the MQTT topic. + var ct = TestContext.Current.CancellationToken; + var client = await ConnectClient(); + var received = new List<(string Id, byte[] Payload, uint? Sn)>(); + var allThree = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + client.ApplicationMessageReceivedAsync += async e => + { + var snProp = e.ApplicationMessage.UserProperties? + .FirstOrDefault(p => p.Name == "dapps-stream-seq"); + uint? sn = snProp is not null && uint.TryParse(snProp.Value, out var s) ? s : null; + var idProp = e.ApplicationMessage.UserProperties!.Single(p => p.Name == "dapps-id"); + received.Add((idProp.Value, e.ApplicationMessage.PayloadSegment.ToArray(), sn)); + if (received.Count == 3) allThree.TrySetResult(); + await Task.CompletedTask; + }; + await client.SubscribeAsync("dapps/in/chat", + MQTTnet.Protocol.MqttQualityOfServiceLevel.AtLeastOnce, ct); + + var bm3 = MakeStreamMessage("ord0003", "G0FROM", "c1", 3, 0, "msg-three"); + var bm2 = MakeStreamMessage("ord0002", "G0FROM", "c1", 2, 0, "msg-two"); + var bm1 = MakeStreamMessage("ord0001", "G0FROM", "c1", 1, 0, "msg-one"); + + await inbox.DeliverAsync(bm3, "G0HOP", ct); + await inbox.DeliverAsync(bm2, "G0HOP", ct); + // Nothing on MQTT yet. + await Task.Delay(150, ct); + received.Should().BeEmpty("ordered messages park until the missing prior arrives"); + + await inbox.DeliverAsync(bm1, "G0HOP", ct); + await allThree.Task.WaitAsync(TimeSpan.FromSeconds(5), ct); + + received.Select(r => r.Sn).Should().Equal(1u, 2u, 3u); + received.Select(r => System.Text.Encoding.UTF8.GetString(r.Payload)) + .Should().Equal("msg-one", "msg-two", "msg-three"); + + // Recv state cursor advanced past the run. + var state = await database.GetStreamRecvStateAsync("N0SELF", "G0FROM", "c1"); + state!.NextExpectedSeq.Should().Be(4u); + state.GapDeadline.Should().Be(DateTime.MinValue); + + await client.DisconnectAsync(cancellationToken: ct); + } + + [Fact] + public async Task Inbox_StrictMode_NeverSkipsGap() + { + // gt=0: a missing prior stalls indefinitely. Even after a long + // wall-clock advance, the sweeper must not skip. + var ct = TestContext.Current.CancellationToken; + var client = await ConnectClient(); + var receivedCount = 0; + client.ApplicationMessageReceivedAsync += async e => + { + Interlocked.Increment(ref receivedCount); + await Task.CompletedTask; + }; + await client.SubscribeAsync("dapps/in/chat", + MQTTnet.Protocol.MqttQualityOfServiceLevel.AtLeastOnce, ct); + + var bm2 = MakeStreamMessage("strict02", "G0FROM", "s1", 2, 0, "two"); + await inbox.DeliverAsync(bm2, "G0HOP", ct); + + // Run the sweeper logic manually after a huge clock advance: + // strict mode means GapDeadline stays MinValue, so nothing + // qualifies as stale. + clock.Advance(TimeSpan.FromDays(1)); + var stale = await database.GetStaleStreamGapsAsync(clock.GetUtcNow().UtcDateTime); + stale.Should().BeEmpty("strict-mode (gt=0) parked rows never set GapDeadline; the sweeper has nothing to do"); + + await Task.Delay(150, ct); + receivedCount.Should().Be(0, "strict mode keeps the row parked until the missing prior arrives"); + + await client.DisconnectAsync(cancellationToken: ct); + } + + [Fact] + public async Task Inbox_TimeoutMode_SweeperSkipsGapAndDrains() + { + // gt=600: sn=2 parks with deadline = arrival+600s. After 600s + // elapse without sn=1 arriving, the sweeper advances the cursor + // past the gap and delivers sn=2. + var ct = TestContext.Current.CancellationToken; + var client = await ConnectClient(); + var received = new List(); + var firstDelivery = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + client.ApplicationMessageReceivedAsync += async e => + { + var snProp = e.ApplicationMessage.UserProperties? + .FirstOrDefault(p => p.Name == "dapps-stream-seq"); + uint? sn = snProp is not null && uint.TryParse(snProp.Value, out var s) ? s : null; + received.Add(sn); + firstDelivery.TrySetResult(); + await Task.CompletedTask; + }; + await client.SubscribeAsync("dapps/in/chat", + MQTTnet.Protocol.MqttQualityOfServiceLevel.AtLeastOnce, ct); + + var bm2 = MakeStreamMessage("to000002", "G0FROM", "t1", 2, 600, "after-gap"); + await inbox.DeliverAsync(bm2, "G0HOP", ct); + + // Cursor should be 1, message parked, deadline set 600s out. + var state = await database.GetStreamRecvStateAsync("N0SELF", "G0FROM", "t1"); + state!.NextExpectedSeq.Should().Be(1u); + state.GapDeadline.Should().NotBe(DateTime.MinValue); + + // Advance clock past deadline + manually invoke the sweeper. + clock.Advance(TimeSpan.FromSeconds(601)); + var stale = await database.GetStaleStreamGapsAsync(clock.GetUtcNow().UtcDateTime); + stale.Should().HaveCount(1); + await inbox.SkipGapAsync(stale.Single(), ct); + + await firstDelivery.Task.WaitAsync(TimeSpan.FromSeconds(5), ct); + received.Single().Should().Be(2u, "the sweeper skipped sn=1 and delivered sn=2"); + + var after = await database.GetStreamRecvStateAsync("N0SELF", "G0FROM", "t1"); + after!.NextExpectedSeq.Should().Be(3u); + after.GapDeadline.Should().Be(DateTime.MinValue); + + await client.DisconnectAsync(cancellationToken: ct); + } + + [Fact] + public async Task Inbox_StaleSeqAfterCursor_DroppedAsStreamStale() + { + // After sn=1 delivers and the cursor advances to 2, a re-arrival + // of sn=1 is stale and must be dropped (not re-delivered to MQTT). + var ct = TestContext.Current.CancellationToken; + var client = await ConnectClient(); + var deliveryCount = 0; + client.ApplicationMessageReceivedAsync += async e => + { + Interlocked.Increment(ref deliveryCount); + await Task.CompletedTask; + }; + await client.SubscribeAsync("dapps/in/chat", + MQTTnet.Protocol.MqttQualityOfServiceLevel.AtLeastOnce, ct); + + await inbox.DeliverAsync(MakeStreamMessage("stale001", "G0FROM", "x1", 1, 0, "first"), "G0HOP", ct); + await Task.Delay(150, ct); + deliveryCount.Should().Be(1); + + await inbox.DeliverAsync(MakeStreamMessage("stale002", "G0FROM", "x1", 1, 0, "duplicate-first"), "G0HOP", ct); + await Task.Delay(150, ct); + + deliveryCount.Should().Be(1, "stale re-arrivals of an already-delivered seq must not double-deliver"); + var dropped = await database.GetRecentDroppedMessages(); + dropped.Should().Contain(d => d.Id == "stale002" && d.Reason == "stream-stale"); + + await client.DisconnectAsync(cancellationToken: ct); + } + + // ── Sender-side counter persistence ──────────────────────────── + + [Fact] + public async Task Sender_CounterPersistsAcrossSubmissions() + { + await database.SubmitOutboundMessage("chat", "G0DEST", "one"u8.ToArray(), + streamId: "c1"); + await database.SubmitOutboundMessage("chat", "G0DEST", "two"u8.ToArray(), + streamId: "c1"); + await database.SubmitOutboundMessage("chat", "G0DEST", "three"u8.ToArray(), + streamId: "c1"); + + var states = await database.GetStreamSendStatesAsync(); + states.Should().ContainSingle(); + states[0].NextSeq.Should().Be(4u, "after three submissions the next-to-mint seq is 4"); + + var rows = (await database.GetRecentMessages(10)).OrderBy(r => r.StreamSeq).ToList(); + rows.Select(r => r.StreamSeq).Should().Equal(1u, 2u, 3u); + rows.Should().AllSatisfy(r => r.StreamId.Should().Be("c1")); + rows.Should().AllSatisfy(r => r.StreamGapTimeoutSeconds.Should().Be(0u)); + } + + [Fact] + public async Task Sender_DifferentStreamIds_GetIndependentCounters() + { + await database.SubmitOutboundMessage("chat", "G0DEST", "a"u8.ToArray(), streamId: "c1"); + await database.SubmitOutboundMessage("chat", "G0DEST", "b"u8.ToArray(), streamId: "c2"); + await database.SubmitOutboundMessage("chat", "G0DEST", "c"u8.ToArray(), streamId: "c1"); + + var rows = await database.GetRecentMessages(10); + rows.Where(r => r.StreamId == "c1").Select(r => r.StreamSeq).OrderBy(s => s).Should().Equal(1u, 2u); + rows.Where(r => r.StreamId == "c2").Select(r => r.StreamSeq).OrderBy(s => s).Should().Equal(1u); + } + + [Fact] + public async Task Sender_NoStreamId_NoCounterMutationOrStreamFields() + { + await database.SubmitOutboundMessage("chat", "G0DEST", "plain"u8.ToArray()); + + var states = await database.GetStreamSendStatesAsync(); + states.Should().BeEmpty(); + + var rows = await database.GetRecentMessages(10); + rows.Single().StreamId.Should().BeNull(); + rows.Single().StreamSeq.Should().BeNull(); + rows.Single().StreamGapTimeoutSeconds.Should().BeNull(); + } + + // ── helpers ──────────────────────────────────────────────────── + + private static BackhaulMessage MakeStreamMessage( + string id, string originator, string sid, uint sn, uint gt, string body) + => new( + Id: id, + Destination: "chat@N0SELF", + Salt: 1L, + Ttl: 600, + Payload: System.Text.Encoding.UTF8.GetBytes(body), + Originator: originator, + StreamId: sid, + StreamSeq: sn, + StreamGapTimeoutSeconds: gt); + + private async Task ConnectClient() + { + var client = new MqttFactory().CreateMqttClient(); + var opts = new MqttClientOptionsBuilder() + .WithProtocolVersion(MQTTnet.Formatter.MqttProtocolVersion.V500) + .WithTcpServer("127.0.0.1", brokerPort) + .WithClientId("test-" + Guid.NewGuid().ToString("N")[..6]) + .WithCleanSession(true) + .Build(); + await client.ConnectAsync(opts); + return client; + } + + private static int PickFreeTcpPort() + { + var l = new TcpListener(IPAddress.Loopback, 0); + l.Start(); + var port = ((IPEndPoint)l.LocalEndpoint).Port; + l.Stop(); + return port; + } + + private sealed class TestOptionsMonitor(T value) : IOptionsMonitor + { + public T CurrentValue { get; } = value; + public T Get(string? name) => CurrentValue; + public IDisposable? OnChange(Action listener) => null; + } +} diff --git a/src/dapps/dapps.core/Controllers/AppApiController.cs b/src/dapps/dapps.core/Controllers/AppApiController.cs index 721cb2e..eef3a7e 100644 --- a/src/dapps/dapps.core/Controllers/AppApiController.cs +++ b/src/dapps/dapps.core/Controllers/AppApiController.cs @@ -27,9 +27,22 @@ public async Task> SubmitOutbound([FromBody] Outb if (string.IsNullOrWhiteSpace(request.DestCallsign)) return BadRequest("DestCallsign is required"); if (request.Payload is null || request.Payload.Length == 0) return BadRequest("Payload is required"); if (request.Ttl is { } ttl && ttl <= 0) return BadRequest("Ttl must be a positive integer (seconds)"); + // Stream id constraints: non-empty when supplied (to keep the + // wire form parseable), capped at 255 bytes (datagram codec + // length-prefix is one byte). Spaces forbidden because the + // text wire form is space-delimited. + if (request.StreamId is { } sid) + { + if (string.IsNullOrWhiteSpace(sid)) return BadRequest("StreamId, when supplied, must be non-empty"); + if (sid.Contains(' ') || sid.Contains('=')) return BadRequest("StreamId must not contain spaces or '='"); + if (System.Text.Encoding.UTF8.GetByteCount(sid) > 255) return BadRequest("StreamId exceeds 255 bytes"); + } if (!HttpContext.IsAuthorisedForApp(request.App)) return Forbid(); - var id = await database.SubmitOutboundMessage(request.App, request.DestCallsign, request.Payload, request.Ttl); + var id = await database.SubmitOutboundMessage( + request.App, request.DestCallsign, request.Payload, request.Ttl, + streamId: request.StreamId, + streamGapTimeoutSeconds: request.StreamGapTimeoutSeconds); return Ok(new OutboundResponse(id)); } @@ -69,8 +82,24 @@ public async Task Ack(string app, string id) /// outgoing ihave as ttl=N. Null = no expiry; the /// message stays in the queue until forwarded or manually deleted. /// Apps that care about cleanup should set a value. +/// +/// opts the message into per-sender ordered +/// delivery: messages with the same StreamId destined for the same +/// recipient deliver in submit order at the receiver, gated by the +/// receiver's reorder cursor. +/// chooses the gap policy: 0 (default) = strict (stall forever for +/// the missing seq); >0 = skip the gap after that many seconds. +/// Both fields ride on the wire as the sid=, sn=, +/// gt= ihave keys; the daemon mints sn automatically +/// from the persisted send-state counter. /// -public sealed record OutboundRequest(string App, string DestCallsign, byte[] Payload, int? Ttl = null); +public sealed record OutboundRequest( + string App, + string DestCallsign, + byte[] Payload, + int? Ttl = null, + string? StreamId = null, + uint? StreamGapTimeoutSeconds = null); public sealed record OutboundResponse(string Id); diff --git a/src/dapps/dapps.core/Models/DbDroppedMessage.cs b/src/dapps/dapps.core/Models/DbDroppedMessage.cs index c8ce0a5..e4d57fc 100644 --- a/src/dapps/dapps.core/Models/DbDroppedMessage.cs +++ b/src/dapps/dapps.core/Models/DbDroppedMessage.cs @@ -31,7 +31,15 @@ public class DbDroppedMessage public DateTime DroppedAt { get; init; } = DateTime.UtcNow; /// Short tag explaining the drop. Currently used: - /// ttl-expired. Stable across releases - kept short for + /// ttl-expired, stalled-awaiting-prior, + /// gap-skipped. Stable across releases - kept short for /// dashboard display. public string Reason { get; init; } = ""; + + /// Opt-in ordering: stream id, seq, and gap-timeout + /// preserved from the live row so the dropped-messages dashboard + /// can show why a stream stalled. Null on non-stream rows. + public string? StreamId { get; init; } + public uint? StreamSeq { get; init; } + public uint? StreamGapTimeoutSeconds { get; init; } } diff --git a/src/dapps/dapps.core/Models/DbMessage.cs b/src/dapps/dapps.core/Models/DbMessage.cs index 83fce11..16e7e8e 100644 --- a/src/dapps/dapps.core/Models/DbMessage.cs +++ b/src/dapps/dapps.core/Models/DbMessage.cs @@ -99,4 +99,25 @@ public class DbMessage /// F2 multi-part: total fragment count for the master id. /// Null when not fragmented; ≥ 2 when set. public int? FragmentTotal { get; init; } + + /// Opt-in ordering stream id. Null when this message + /// isn't part of an ordered stream. The receiver's reorder buffer + /// keys on (originator-callsign, StreamId). + public string? StreamId { get; init; } + + /// Opt-in ordering: monotonic seq within + /// (originator, StreamId). Null when not ordered. + public uint? StreamSeq { get; init; } + + /// Opt-in ordering: gap timeout in seconds at originator + /// time. 0 = strict, >0 = skip gap after that many seconds. + /// Carried verbatim across re-forwards so the destination sees the + /// originator's policy regardless of intermediate hops. + public uint? StreamGapTimeoutSeconds { get; init; } + + /// True when the inbox parked this row awaiting an + /// earlier StreamSeq. Set on arrival when seq > expected; + /// cleared as the cursor advances and the row drains to MQTT. + /// Always false for non-ordered messages. + public bool PendingInOrder { get; set; } } diff --git a/src/dapps/dapps.core/Models/DbOffer.cs b/src/dapps/dapps.core/Models/DbOffer.cs index 1257a20..bad39fb 100644 --- a/src/dapps/dapps.core/Models/DbOffer.cs +++ b/src/dapps/dapps.core/Models/DbOffer.cs @@ -46,4 +46,21 @@ internal sealed class DbOffer /// is also non-null. Always ≥ 2 when set /// (single-part messages skip the fragment headers entirely). public int? FragmentTotal { get; init; } + + /// Opt-in ordering: sid= from the ihave line, + /// or null when the message isn't part of an ordered stream. + /// Carried forward into the DbMessage row when the payload + /// arrives so the inbox's reorder buffer keys correctly. + public string? StreamId { get; init; } + + /// Opt-in ordering: sn= monotonic seq within + /// (sender, StreamId). Null when not ordered; non-null only when + /// is also non-null. + public uint? StreamSeq { get; init; } + + /// Opt-in ordering: gt= gap timeout in seconds. + /// 0 = strict (stall forever for the missing seq); >0 = skip + /// the gap after that many seconds and emit gap-skipped. Null when + /// not ordered. + public uint? StreamGapTimeoutSeconds { get; init; } } diff --git a/src/dapps/dapps.core/Models/DbStreamRecvState.cs b/src/dapps/dapps.core/Models/DbStreamRecvState.cs new file mode 100644 index 0000000..de875f7 --- /dev/null +++ b/src/dapps/dapps.core/Models/DbStreamRecvState.cs @@ -0,0 +1,48 @@ +using SQLite; + +namespace dapps.core.Models; + +/// +/// Receiver-side cursor for opt-in message ordering. One row per +/// (LocalCallsign, SenderCallsign, StreamId) tracks the next seq the +/// inbox expects to deliver. Messages with seq > cursor park as +/// ; the cursor advances and +/// drains pending rows when the gap fills (or, in timeout mode, when +/// the gap deadline elapses). +/// +/// SenderCallsign is the F1 originator callsign (not the link source) - +/// ordering is end-to-end at the originator's intent, irrespective of +/// the intermediate forwarding path. +/// +[Table("streamrecvstate")] +public sealed class DbStreamRecvState +{ + /// Composite key {LocalCallsign}|{SenderCallsign}|{StreamId}. + /// Built by on insert. + [PrimaryKey, NotNull] + public string Key { get; set; } = ""; + + public string LocalCallsign { get; set; } = ""; + public string SenderCallsign { get; set; } = ""; + public string StreamId { get; set; } = ""; + + /// The next seq the inbox will deliver. New streams start + /// at 1; the first arrival with sn=1 delivers immediately and + /// advances to 2. An arrival with sn > this value parks until + /// the gap fills. + public uint NextExpectedSeq { get; set; } = 1; + + public DateTime LastReceivedAt { get; set; } = DateTime.UtcNow; + + /// UTC instant after which a still-open gap is allowed to + /// be skipped (timeout mode). DateTime.MinValue means "no gap + /// active" - either the cursor is up to date OR all parked rows + /// were originated with gt=0 (strict). The sweeper computes the + /// deadline from each parked message's StreamGapTimeoutSeconds and + /// the LastReceivedAt of the row that caused the cursor to fall + /// behind. + public DateTime GapDeadline { get; set; } = DateTime.MinValue; + + public static string MakeKey(string localCallsign, string senderCallsign, string streamId) + => $"{localCallsign}|{senderCallsign}|{streamId}"; +} diff --git a/src/dapps/dapps.core/Models/DbStreamSendState.cs b/src/dapps/dapps.core/Models/DbStreamSendState.cs new file mode 100644 index 0000000..b135bba --- /dev/null +++ b/src/dapps/dapps.core/Models/DbStreamSendState.cs @@ -0,0 +1,38 @@ +using SQLite; + +namespace dapps.core.Models; + +/// +/// Sender-side counter for opt-in message ordering. One row per +/// (LocalCallsign, RemoteCallsign, StreamId) - the daemon mints +/// monotonically increasing seq numbers from this row when stamping +/// outbound messages. Persisted so a reboot doesn't reset the counter +/// mid-stream and cause receiver-side dup drops or seq collisions. +/// +/// StreamId is sender-scoped (per the protocol design): the receiver +/// keys its cursor on (originator-callsign, StreamId), so two senders +/// can pick the same StreamId without colliding. +/// +[Table("streamsendstate")] +public sealed class DbStreamSendState +{ + /// Composite key {LocalCallsign}|{RemoteCallsign}|{StreamId}. + /// Built by on insert. + [PrimaryKey, NotNull] + public string Key { get; set; } = ""; + + public string LocalCallsign { get; set; } = ""; + public string RemoteCallsign { get; set; } = ""; + public string StreamId { get; set; } = ""; + + /// The next seq to mint. After taking it, the daemon + /// updates this row to NextSeq + 1 in the same transaction + /// as persisting the outbound message, so a crash between the + /// two leaves the counter coherent with what's on disk. + public uint NextSeq { get; set; } = 1; + + public DateTime UpdatedAt { get; set; } = DateTime.UtcNow; + + public static string MakeKey(string localCallsign, string remoteCallsign, string streamId) + => $"{localCallsign}|{remoteCallsign}|{streamId}"; +} diff --git a/src/dapps/dapps.core/Pages/IHave.cshtml b/src/dapps/dapps.core/Pages/IHave.cshtml index a1ee84d..fee6673 100644 --- a/src/dapps/dapps.core/Pages/IHave.cshtml +++ b/src/dapps/dapps.core/Pages/IHave.cshtml @@ -14,6 +14,7 @@ dashboard inbound + streams audit @if (User.Identity?.IsAuthenticated == true) { diff --git a/src/dapps/dapps.core/Pages/Inbound.cshtml b/src/dapps/dapps.core/Pages/Inbound.cshtml index ca9b19a..60a5350 100644 --- a/src/dapps/dapps.core/Pages/Inbound.cshtml +++ b/src/dapps/dapps.core/Pages/Inbound.cshtml @@ -14,6 +14,7 @@ dashboard ihave + streams audit @if (User.Identity?.IsAuthenticated == true) { diff --git a/src/dapps/dapps.core/Pages/Index.cshtml b/src/dapps/dapps.core/Pages/Index.cshtml index 20e07e7..1960766 100644 --- a/src/dapps/dapps.core/Pages/Index.cshtml +++ b/src/dapps/dapps.core/Pages/Index.cshtml @@ -14,6 +14,7 @@ inbound ihave + streams audit @if (User.Identity?.IsAuthenticated == true) { diff --git a/src/dapps/dapps.core/Pages/Streams.cshtml b/src/dapps/dapps.core/Pages/Streams.cshtml new file mode 100644 index 0000000..44efec2 --- /dev/null +++ b/src/dapps/dapps.core/Pages/Streams.cshtml @@ -0,0 +1,129 @@ +@page +@model StreamsModel +@{ + ViewData["Title"] = $"DAPPS - streams (@{Model.Options.Callsign})"; + var now = DateTime.UtcNow; +} + +
+ + + DAPPS + @Model.Options.Callsign + - streams + + + dashboard + inbound + ihave + audit + @if (User.Identity?.IsAuthenticated == true) + { + sign out + } +
+ +
+

Opt-in ordered streams

+

+ Per-(remote, stream-id) state for opt-in message ordering. + Set StreamId on outbound submissions to opt in; + the daemon mints monotonic sn= values from the send-side + counters below. Receive-side cursors gate local delivery on the + next-expected seq, with parked rows held until the gap fills or + (when gt > 0) the gap deadline elapses. +

+ +

Send state

+ @if (Model.SendStates.Count == 0) + { +

No outbound streams have been used yet.

+ } + else + { + + + + + + + + + + + + @foreach (var s in Model.SendStates) + { + + + + + + + + } + +
LocalRemoteStream idNext snUpdated
@s.LocalCallsign@s.RemoteCallsign@s.StreamId@s.NextSeq@s.UpdatedAt.ToString("yyyy-MM-dd HH:mm:ss")Z
+ } + +

Receive state

+ @if (Model.RecvStates.Count == 0) + { +

No inbound streams have been seen yet.

+ } + else + { + + + + + + + + + + + + + + @foreach (var row in Model.RecvStates) + { + var stalled = row.PendingCount > 0; + + + + + + + + + + } + +
SenderStream idNext snPendingOldest pendingGap deadlineLast seen
@row.State.SenderCallsign@row.State.StreamId@row.State.NextExpectedSeq@row.PendingCount + @if (row.OldestPendingAt is { } oldest) + { + @($"{(int)(now - oldest).TotalSeconds}s ago") + } + else + { + @:- + } + + @if (row.State.GapDeadline == DateTime.MinValue) + { + - + } + else + { + @row.State.GapDeadline.ToString("yyyy-MM-dd HH:mm:ss") + @:Z + } + @row.State.LastReceivedAt.ToString("yyyy-MM-dd HH:mm:ss")Z
+

+ Highlighted rows have parked messages awaiting an earlier sn. + Strict streams (gt=0) show no gap deadline; in timeout mode the deadline + is when the gap sweeper (1-minute cadence) will skip past the missing seq and drain. +

+ } +
diff --git a/src/dapps/dapps.core/Pages/Streams.cshtml.cs b/src/dapps/dapps.core/Pages/Streams.cshtml.cs new file mode 100644 index 0000000..2631480 --- /dev/null +++ b/src/dapps/dapps.core/Pages/Streams.cshtml.cs @@ -0,0 +1,42 @@ +using dapps.core.Models; +using dapps.core.Services; +using Microsoft.AspNetCore.Mvc.RazorPages; +using Microsoft.Extensions.Options; + +namespace dapps.core.Pages; + +/// +/// Read-only operator view of opt-in ordering state. Two tables: +/// +/// Send - per-(remote, stream) counters minted by local app submissions. +/// Receive - per-(originator, stream) cursors plus the count of currently-parked rows for each, so a stalled stream is obvious at a glance. +/// +/// Refresh-driven; the work involved is too sparse to warrant SSE. +/// +public sealed class StreamsModel( + Database database, + IOptionsMonitor options) : PageModel +{ + public SystemOptions Options { get; private set; } = new(); + public IReadOnlyList SendStates { get; private set; } = []; + public IReadOnlyList RecvStates { get; private set; } = []; + + public async Task OnGetAsync() + { + Options = options.CurrentValue; + SendStates = await database.GetStreamSendStatesAsync(); + var recvs = await database.GetStreamRecvStatesAsync(); + var rows = new List(recvs.Count); + foreach (var r in recvs) + { + var pending = await database.GetPendingInOrderAsync(r.SenderCallsign, r.StreamId); + DateTime? oldestPending = pending.Count == 0 + ? null + : pending.Min(p => p.CreatedAt); + rows.Add(new RecvRow(r, pending.Count, oldestPending)); + } + RecvStates = rows; + } + + public sealed record RecvRow(DbStreamRecvState State, int PendingCount, DateTime? OldestPendingAt); +} diff --git a/src/dapps/dapps.core/Pages/Transmissions.cshtml b/src/dapps/dapps.core/Pages/Transmissions.cshtml index 9584966..ba486dc 100644 --- a/src/dapps/dapps.core/Pages/Transmissions.cshtml +++ b/src/dapps/dapps.core/Pages/Transmissions.cshtml @@ -15,6 +15,7 @@ dashboard inbound ihave + streams @if (User.Identity?.IsAuthenticated == true) { sign out diff --git a/src/dapps/dapps.core/Program.cs b/src/dapps/dapps.core/Program.cs index b41bd3a..4f9a03c 100644 --- a/src/dapps/dapps.core/Program.cs +++ b/src/dapps/dapps.core/Program.cs @@ -130,6 +130,7 @@ builder.Services.AddHostedService(); builder.Services.AddHostedService(); builder.Services.AddHostedService(); +builder.Services.AddHostedService(); builder.Services.AddSingleton(); builder.Services.AddSingleton(); builder.Services.AddSingleton(); @@ -222,7 +223,8 @@ // /Config flip takes effect on the next session). opportunisticInbox: sp.GetRequiredService(), opportunisticEnabled: () => sp.GetRequiredService>().CurrentValue.OpportunisticPollEnabled)); -builder.Services.AddSingleton(); +builder.Services.AddSingleton(); +builder.Services.AddSingleton(sp => sp.GetRequiredService()); builder.Services.AddHostedService(); // B6.1 - connected-mode probe-and-map. NodeProber is stateless and diff --git a/src/dapps/dapps.core/Services/Database.cs b/src/dapps/dapps.core/Services/Database.cs index b5fa1c7..a1039f0 100644 --- a/src/dapps/dapps.core/Services/Database.cs +++ b/src/dapps/dapps.core/Services/Database.cs @@ -65,17 +65,49 @@ await DbInfo.GetAsyncConnection().ExecuteAsync( /// the message persist in the queue indefinitely if it can't be /// forwarded; apps that want guaranteed cleanup should set a value. /// - public async Task SubmitOutboundMessage(string appName, string destCallsign, byte[] payload, int? ttlSeconds = null) + public async Task SubmitOutboundMessage( + string appName, + string destCallsign, + byte[] payload, + int? ttlSeconds = null, + string? streamId = null, + uint? streamGapTimeoutSeconds = null) { var destination = $"{appName}@{destCallsign}"; var ourCall = options.CurrentValue.Callsign; var threshold = options.CurrentValue.FragmentThresholdBytes; + // Opt-in ordering: if a stream id is supplied, mint the next seq + // from DbStreamSendState. The counter and the message persist in + // the same call site; a crash between the two leaves the counter + // coherent with the on-disk message (the message commit happens + // after the counter bump). Gap timeout 0 (= strict) is the + // default when streamId is set without an explicit value. + var hasStream = !string.IsNullOrEmpty(streamId); + if (hasStream && (streamGapTimeoutSeconds is null)) + { + streamGapTimeoutSeconds = 0; + } + uint? streamSeq = null; + if (hasStream) + { + streamSeq = await AllocateStreamSeqAsync(ourCall, destCallsign, streamId!); + } + // F2 multi-part: split payloads above the threshold into N // fragment rows. Each fragment is its own DbMessage with a // shared MasterId and a 1-based FragmentIndex. The forwarder // picks them up individually; the destination's inbox // reassembles. Threshold = 0 disables fragmentation. + // + // Opt-in ordering note: stream metadata is stamped on the + // master id of a fragmented submission (and on the sole row + // of a non-fragmented one). Each fragment carries the full + // stream trio so a fragment-aware-but-stream-naive forwarder + // doesn't accidentally strip ordering. The receiver only + // applies ordering once reassembly produces the assembled + // payload (fragments arriving out of frag-index order are + // independent of the stream cursor). if (threshold > 0 && payload.Length > threshold) { var masterSalt = (long)(timeProvider.GetUtcNow().UtcDateTime - DateTime.UnixEpoch).TotalMilliseconds; @@ -96,7 +128,9 @@ await SaveMessage( fragId, chunk, fragSalt, destination, sourceCallsign: ourCall, "{}", ttl: ttlSeconds, originatorCallsign: ourCall, - masterId: masterId, fragmentIndex: i + 1, fragmentTotal: total); + masterId: masterId, fragmentIndex: i + 1, fragmentTotal: total, + streamId: streamId, streamSeq: streamSeq, + streamGapTimeoutSeconds: streamGapTimeoutSeconds); } return masterId; } @@ -106,7 +140,10 @@ await SaveMessage( // Local submission: we are both the link-source AND the // originator. Recorded explicitly so re-forwards downstream // surface us in the receiver's dapps-origin user property. - await SaveMessage(id, payload, salt, destination, sourceCallsign: ourCall, "{}", ttl: ttlSeconds, originatorCallsign: ourCall); + await SaveMessage(id, payload, salt, destination, sourceCallsign: ourCall, "{}", ttl: ttlSeconds, + originatorCallsign: ourCall, + streamId: streamId, streamSeq: streamSeq, + streamGapTimeoutSeconds: streamGapTimeoutSeconds); return id; } @@ -117,7 +154,7 @@ internal async Task LoadOfferMetadata(string id) return data; } - internal async Task SaveMessage(string id, byte[] buffer, long? salt, string destination, string sourceCallsign, string additionalProperties, int? ttl, string originatorCallsign = "", byte? floodHopsRemaining = null, string? sourceRouteCsv = null, string? traversedHopsCsv = null, string? masterId = null, int? fragmentIndex = null, int? fragmentTotal = null) + internal async Task SaveMessage(string id, byte[] buffer, long? salt, string destination, string sourceCallsign, string additionalProperties, int? ttl, string originatorCallsign = "", byte? floodHopsRemaining = null, string? sourceRouteCsv = null, string? traversedHopsCsv = null, string? masterId = null, int? fragmentIndex = null, int? fragmentTotal = null, string? streamId = null, uint? streamSeq = null, uint? streamGapTimeoutSeconds = null, bool pendingInOrder = false) { var connection = DbInfo.GetAsyncConnection(); @@ -146,6 +183,10 @@ await DbInfo.GetAsyncConnection().InsertAsync(new DbMessage MasterId = masterId, FragmentIndex = fragmentIndex, FragmentTotal = fragmentTotal, + StreamId = streamId, + StreamSeq = streamSeq, + StreamGapTimeoutSeconds = streamGapTimeoutSeconds, + PendingInOrder = pendingInOrder, }); } @@ -175,6 +216,9 @@ await connection.InsertAsync(new DbOffer MasterId = offer.MasterId, FragmentIndex = offer.Fragment?.Index, FragmentTotal = offer.Fragment?.Total, + StreamId = offer.StreamId, + StreamSeq = offer.StreamSeq, + StreamGapTimeoutSeconds = offer.StreamGapTimeoutSeconds, }); logger.LogInformation("Saved metadata for offer {0}", offer.Id); @@ -227,6 +271,9 @@ await c.InsertAsync(new DbDroppedMessage CreatedAt = row.CreatedAt, DroppedAt = timeProvider.GetUtcNow().UtcDateTime, Reason = reason, + StreamId = row.StreamId, + StreamSeq = row.StreamSeq, + StreamGapTimeoutSeconds = row.StreamGapTimeoutSeconds, }); await c.DeleteAsync(id); } @@ -831,4 +878,98 @@ internal async Task RemovePolledNode(string callsign) "delete from polledNodes where callsign=?", callsign); return deleted > 0; } + + // ── Opt-in message ordering: send-side counter ────────────────── + + /// + /// Mint the next sender-side seq for (LocalCallsign, RemoteCallsign, + /// StreamId) and persist the bumped counter. Idempotent only across + /// transactions: each call returns a fresh seq and advances the row. + /// New rows start at 1; existing rows return NextSeq and increment. + /// + internal async Task AllocateStreamSeqAsync(string localCallsign, string remoteCallsign, string streamId) + { + var connection = DbInfo.GetAsyncConnection(); + var key = DbStreamSendState.MakeKey(localCallsign, remoteCallsign, streamId); + var row = await connection.FindAsync(key); + var now = timeProvider.GetUtcNow().UtcDateTime; + if (row is null) + { + await connection.InsertAsync(new DbStreamSendState + { + Key = key, + LocalCallsign = localCallsign, + RemoteCallsign = remoteCallsign, + StreamId = streamId, + NextSeq = 2, + UpdatedAt = now, + }); + return 1; + } + var seq = row.NextSeq; + row.NextSeq = seq + 1; + row.UpdatedAt = now; + await connection.UpdateAsync(row); + return seq; + } + + /// All sender-side stream counters - dashboard listing. + public async Task> GetStreamSendStatesAsync() + => await DbInfo.GetAsyncConnection().QueryAsync( + "select * from streamsendstate order by UpdatedAt desc"); + + // ── Opt-in message ordering: recv-side cursor ─────────────────── + + /// Look up the receive cursor for a (sender, stream) pair. + /// Null when no message has ever been seen for that pair. + internal async Task GetStreamRecvStateAsync(string localCallsign, string senderCallsign, string streamId) + { + var key = DbStreamRecvState.MakeKey(localCallsign, senderCallsign, streamId); + return await DbInfo.GetAsyncConnection().FindAsync(key); + } + + /// Idempotent upsert of the recv cursor. + internal async Task UpsertStreamRecvStateAsync(DbStreamRecvState state) + { + state.Key = DbStreamRecvState.MakeKey(state.LocalCallsign, state.SenderCallsign, state.StreamId); + var connection = DbInfo.GetAsyncConnection(); + var existing = await connection.FindAsync(state.Key); + if (existing is null) await connection.InsertAsync(state); + else await connection.UpdateAsync(state); + } + + /// All recv cursors - dashboard listing. + public async Task> GetStreamRecvStatesAsync() + => await DbInfo.GetAsyncConnection().QueryAsync( + "select * from streamrecvstate order by LastReceivedAt desc"); + + /// Pending (parked-awaiting-prior) ordered messages for a + /// (sender, stream) pair, ordered by StreamSeq ascending. The inbox + /// drains from the front as the cursor advances. + internal async Task> GetPendingInOrderAsync(string senderCallsign, string streamId) + { + var connection = DbInfo.GetAsyncConnection(); + return await connection.QueryAsync( + "select * from messages where PendingInOrder=1 and OriginatorCallsign=? and StreamId=? order by StreamSeq asc", + senderCallsign, streamId); + } + + /// Mark a parked row as no-longer-pending. Used when the + /// inbox drains it to MQTT or the sweeper skips past it. + internal async Task ClearPendingInOrderAsync(string id) + { + await DbInfo.GetAsyncConnection().ExecuteAsync( + "update messages set PendingInOrder=0 where Id=?", id); + } + + /// Recv cursors that have a non-MinValue gap deadline at or + /// before . The sweeper advances these. + internal async Task> GetStaleStreamGapsAsync(DateTime cutoff) + { + var connection = DbInfo.GetAsyncConnection(); + var rows = await connection.QueryAsync( + "select * from streamrecvstate where GapDeadline > 0 and GapDeadline <= ?", + cutoff.Ticks); + return rows; + } } \ No newline at end of file diff --git a/src/dapps/dapps.core/Services/DatabaseAndMqttInbox.cs b/src/dapps/dapps.core/Services/DatabaseAndMqttInbox.cs index 0a4619d..a176cd0 100644 --- a/src/dapps/dapps.core/Services/DatabaseAndMqttInbox.cs +++ b/src/dapps/dapps.core/Services/DatabaseAndMqttInbox.cs @@ -91,6 +91,18 @@ public async Task DeliverAsync( return; } + var isLocal = DestinationParser.IsLocal(message.Destination, options.CurrentValue.Callsign); + + // Opt-in ordering: when the envelope carries sn=, gate local + // delivery on the per-(originator, sid) cursor. Transit messages + // flow through unchanged - intermediate hops re-emit the trio + // verbatim; only the final destination reorders. + if (isLocal && message.StreamSeq.HasValue && !string.IsNullOrEmpty(message.StreamId)) + { + await DeliverOrderedAsync(message, sourceCallsign, originator, headersJson, ct); + return; + } + await database.SaveMessage( message.Id, message.Payload, @@ -125,9 +137,15 @@ await database.SaveMessage( // messages OR fragments-for-elsewhere.) masterId: message.MasterId, fragmentIndex: message.FragmentIndex, - fragmentTotal: message.FragmentTotal); + fragmentTotal: message.FragmentTotal, + // Opt-in ordering: stream trio is preserved on transit rows + // so the forwarder re-emits them on the next hop verbatim. + // (Local-ordered-delivery took the early-return above.) + streamId: message.StreamId, + streamSeq: message.StreamSeq, + streamGapTimeoutSeconds: message.StreamGapTimeoutSeconds); - if (DestinationParser.IsLocal(message.Destination, options.CurrentValue.Callsign)) + if (isLocal) { var dbMessage = new DbMessage { @@ -139,6 +157,9 @@ await database.SaveMessage( OriginatorCallsign = originator, AdditionalProperties = headersJson, Ttl = message.Ttl, + StreamId = message.StreamId, + StreamSeq = message.StreamSeq, + StreamGapTimeoutSeconds = message.StreamGapTimeoutSeconds, }; await mqtt.InjectInboundMessage(dbMessage); } @@ -159,6 +180,218 @@ await database.SaveMessage( Ttl: message.Ttl)); } + /// + /// Local-destination delivery for an opt-in-ordered message. Compares + /// StreamSeq against the persisted cursor for (LocalCallsign, + /// originator, StreamId): + /// + /// + /// seq == expected -> persist, deliver to MQTT, advance cursor, drain consecutive successors + /// seq > expected -> persist with PendingInOrder=true, set GapDeadline (when gt>0) so can skip later + /// seq < expected -> persist + soft-delete with reason "stream-stale" (already delivered or already skipped past) + /// + /// + private async Task DeliverOrderedAsync(BackhaulMessage message, string sourceCallsign, string originator, string headersJson, CancellationToken ct) + { + var localCall = options.CurrentValue.Callsign; + var streamId = message.StreamId!; + var sn = message.StreamSeq!.Value; + // SenderCallsign on the recv-state row keys on the F1 originator + // (end-to-end intent), not the link source. Falling back to the + // link source preserves usable behaviour on legacy peers that + // don't propagate src= - they're just stuck with one cursor per + // physical neighbour for that StreamId. + var streamSender = !string.IsNullOrEmpty(message.Originator) + ? message.Originator + : sourceCallsign; + var now = timeProvider.GetUtcNow().UtcDateTime; + + var recv = await database.GetStreamRecvStateAsync(localCall, streamSender, streamId) + ?? new DbStreamRecvState + { + LocalCallsign = localCall, + SenderCallsign = streamSender, + StreamId = streamId, + NextExpectedSeq = 1, + LastReceivedAt = now, + GapDeadline = DateTime.MinValue, + }; + + if (sn < recv.NextExpectedSeq) + { + // Already delivered, or the sweeper already advanced past + // this seq. Persist for audit, then soft-delete with a + // dedicated reason so the dashboard's dropped panel makes + // it obvious why this was discarded rather than delivered. + await database.SaveMessage( + message.Id, message.Payload, message.Salt, message.Destination, + sourceCallsign, headersJson, message.Ttl, + originatorCallsign: originator, + streamId: streamId, streamSeq: sn, + streamGapTimeoutSeconds: message.StreamGapTimeoutSeconds); + await database.SoftDeleteMessage(message.Id, "stream-stale"); + recv.LastReceivedAt = now; + await database.UpsertStreamRecvStateAsync(recv); + logger.LogInformation( + "Stream {0}|{1}: dropping {2} (sn={3} < expected {4}) as stream-stale", + streamSender, streamId, message.Id, sn, recv.NextExpectedSeq); + events.Publish(new InboundEvent(now, message.Id, sourceCallsign, message.Destination, message.Payload.Length, message.Ttl)); + return; + } + + if (sn > recv.NextExpectedSeq) + { + // Park. Persist with PendingInOrder=true so the row is in + // the messages table (visible in the dashboard, ack'able by + // ID, etc.) but the inbox doesn't push to MQTT until the + // gap fills. GapDeadline is set on first detection of the + // gap; subsequent arrivals don't push it out - the original + // arrival's deadline is the right cap. + await database.SaveMessage( + message.Id, message.Payload, message.Salt, message.Destination, + sourceCallsign, headersJson, message.Ttl, + originatorCallsign: originator, + streamId: streamId, streamSeq: sn, + streamGapTimeoutSeconds: message.StreamGapTimeoutSeconds, + pendingInOrder: true); + recv.LastReceivedAt = now; + if (recv.GapDeadline == DateTime.MinValue + && message.StreamGapTimeoutSeconds is { } gt && gt > 0) + { + recv.GapDeadline = now + TimeSpan.FromSeconds(gt); + } + await database.UpsertStreamRecvStateAsync(recv); + logger.LogInformation( + "Stream {0}|{1}: parking {2} (sn={3}, expected {4}, gt={5})", + streamSender, streamId, message.Id, sn, recv.NextExpectedSeq, + message.StreamGapTimeoutSeconds ?? 0); + events.Publish(new InboundEvent(now, message.Id, sourceCallsign, message.Destination, message.Payload.Length, message.Ttl)); + return; + } + + // sn == NextExpectedSeq: deliver immediately + drain successors. + await database.SaveMessage( + message.Id, message.Payload, message.Salt, message.Destination, + sourceCallsign, headersJson, message.Ttl, + originatorCallsign: originator, + streamId: streamId, streamSeq: sn, + streamGapTimeoutSeconds: message.StreamGapTimeoutSeconds); + + var dbMessage = new DbMessage + { + Id = message.Id, Payload = message.Payload, Salt = message.Salt, + Destination = message.Destination, SourceCallsign = sourceCallsign, + OriginatorCallsign = originator, AdditionalProperties = headersJson, + Ttl = message.Ttl, + StreamId = streamId, StreamSeq = sn, + StreamGapTimeoutSeconds = message.StreamGapTimeoutSeconds, + }; + await mqtt.InjectInboundMessage(dbMessage); + recv.NextExpectedSeq = sn + 1; + recv.LastReceivedAt = now; + + await DrainConsecutivePendingAsync(recv, streamSender, streamId, ct); + events.Publish(new InboundEvent(now, message.Id, sourceCallsign, message.Destination, message.Payload.Length, message.Ttl)); + } + + /// + /// Drain pending rows whose StreamSeq matches the cursor, advancing + /// it as it consumes. Stops at the first gap; recomputes the + /// recv-state's GapDeadline based on the remaining pending head's + /// gt (or clears it when no gap remains). Used by both the + /// in-order arrival path and the gap sweeper. + /// + internal async Task DrainConsecutivePendingAsync( + DbStreamRecvState recv, string streamSender, string streamId, CancellationToken ct) + { + var pending = await database.GetPendingInOrderAsync(streamSender, streamId); + // pending is ordered by StreamSeq asc, so a single forward pass + // either delivers a consecutive run or stops at the first gap. + var byteOffset = 0; + foreach (var p in pending) + { + if (p.StreamSeq is null) continue; + if (p.StreamSeq.Value < recv.NextExpectedSeq) + { + // Stale parked row (the sweeper advanced past it via + // gap-skip and a new arrival didn't catch this up). + await database.SoftDeleteMessage(p.Id, "stream-stale"); + continue; + } + if (p.StreamSeq.Value != recv.NextExpectedSeq) break; + + var inject = new DbMessage + { + Id = p.Id, Payload = p.Payload, Salt = p.Salt, + Destination = p.Destination, SourceCallsign = p.SourceCallsign, + OriginatorCallsign = p.OriginatorCallsign, + AdditionalProperties = p.AdditionalProperties, + Ttl = p.Ttl, CreatedAt = p.CreatedAt, + StreamId = p.StreamId, StreamSeq = p.StreamSeq, + StreamGapTimeoutSeconds = p.StreamGapTimeoutSeconds, + }; + await mqtt.InjectInboundMessage(inject); + await database.ClearPendingInOrderAsync(p.Id); + recv.NextExpectedSeq = p.StreamSeq.Value + 1; + byteOffset += p.Payload.Length; + } + + // Recompute deadline: if anything still pending above the new + // cursor, use the head's stored gt as the new gap budget; else + // clear. + var stillPending = (await database.GetPendingInOrderAsync(streamSender, streamId)) + .FirstOrDefault(p => p.StreamSeq is { } s && s > recv.NextExpectedSeq); + if (stillPending is null) + { + recv.GapDeadline = DateTime.MinValue; + } + else if (stillPending.StreamGapTimeoutSeconds is { } gt && gt > 0) + { + recv.GapDeadline = recv.LastReceivedAt + TimeSpan.FromSeconds(gt); + } + else + { + // gt=0 (strict): the new gap stalls forever. + recv.GapDeadline = DateTime.MinValue; + } + await database.UpsertStreamRecvStateAsync(recv); + if (byteOffset > 0) + { + logger.LogInformation( + "Stream {0}|{1}: drained pending up to sn={2}", + streamSender, streamId, recv.NextExpectedSeq - 1); + } + } + + /// + /// Called by when a recv-state's + /// gap deadline has elapsed. Skips the cursor forward to the + /// smallest pending sn for that stream and drains successors. + /// Logs each skipped seq for operator visibility. + /// + internal async Task SkipGapAsync(DbStreamRecvState recv, CancellationToken ct) + { + var pending = await database.GetPendingInOrderAsync(recv.SenderCallsign, recv.StreamId); + var firstAbove = pending.FirstOrDefault(p => p.StreamSeq is { } s && s >= recv.NextExpectedSeq); + if (firstAbove?.StreamSeq is null) + { + // Nothing pending - clear the deadline. + recv.GapDeadline = DateTime.MinValue; + await database.UpsertStreamRecvStateAsync(recv); + return; + } + var fromSn = recv.NextExpectedSeq; + var toSn = firstAbove.StreamSeq.Value; + for (var k = fromSn; k < toSn; k++) + { + logger.LogWarning( + "Stream {0}|{1}: gap-skipped sn={2} (deadline elapsed)", + recv.SenderCallsign, recv.StreamId, k); + } + recv.NextExpectedSeq = toSn; + await DrainConsecutivePendingAsync(recv, recv.SenderCallsign, recv.StreamId, ct); + } + /// /// Plan F2 - destination-side fragment handling. Stores the /// fragment in , checks whether the full diff --git a/src/dapps/dapps.core/Services/DbStartup.cs b/src/dapps/dapps.core/Services/DbStartup.cs index abc5798..db469ee 100644 --- a/src/dapps/dapps.core/Services/DbStartup.cs +++ b/src/dapps/dapps.core/Services/DbStartup.cs @@ -64,6 +64,8 @@ public static void EnsureSchemaAndSeed(ILogger? logger = null) db.CreateTable(); db.CreateTable(); db.CreateTable(); + db.CreateTable(); + db.CreateTable(); var optionsTable = db.Table().Table.TableName; var options = db.Query($"select * from {optionsTable};"); diff --git a/src/dapps/dapps.core/Services/IHaveValidator.cs b/src/dapps/dapps.core/Services/IHaveValidator.cs index 00cbb3b..52ecddc 100644 --- a/src/dapps/dapps.core/Services/IHaveValidator.cs +++ b/src/dapps/dapps.core/Services/IHaveValidator.cs @@ -21,7 +21,10 @@ public sealed record IHaveOffer( Dictionary AdditionalHeaders, string? Originator = null, // src= - originating callsign (F1), null when sender pre-dates F1 string? MasterId = null, // mid= - F2 multi-part: opaque grouping id, null = not fragmented - FragmentInfo? Fragment = null); // frag=N/M - F2 multi-part fragment index/total, null = not fragmented + FragmentInfo? Fragment = null, // frag=N/M - F2 multi-part fragment index/total, null = not fragmented + string? StreamId = null, // sid= - opt-in ordering stream identifier (per sender) + uint? StreamSeq = null, // sn= - monotonic seq within (sender, sid) + uint? StreamGapTimeoutSeconds = null); // gt= - 0 = strict, >0 = skip gap after N seconds /// F2 multi-part fragment metadata. Index is 1-based; total is /// the number of fragments in the original payload. Both must satisfy @@ -55,7 +58,7 @@ public static class IHaveValidator private const int ChkValueLength = 4; private static readonly HashSet ReservedKeys = new(StringComparer.Ordinal) - { "len", "fmt", "s", "clen", "dst", "chk", "ttl", "src", "mid", "frag" }; + { "len", "fmt", "s", "clen", "dst", "chk", "ttl", "src", "mid", "frag", "sid", "sn", "gt" }; public static OfferValidationResult Validate(string ihaveCommand) { @@ -192,8 +195,40 @@ public static OfferValidationResult Validate(string ihaveCommand) fragment = new FragmentInfo(fragN, fragM); } + // sid= / sn= / gt= - opt-in ordering. All three travel together + // or all three are absent; partial sets are an error since a + // gappy implementation downstream couldn't reconstruct the + // ordering contract. sn / gt are 32-bit unsigned: sn is the + // monotonic seq, gt is the gap timeout in seconds (0 = strict). + string? streamId = null; + uint? streamSeq = null; + uint? streamGapTimeout = null; + var hasSid = kvps.TryGetValue("sid", out var sidVal) && !string.IsNullOrEmpty(sidVal); + var hasSn = kvps.TryGetValue("sn", out var snVal) && !string.IsNullOrEmpty(snVal); + var hasGt = kvps.TryGetValue("gt", out var gtVal) && !string.IsNullOrEmpty(gtVal); + if (hasSid || hasSn || hasGt) + { + if (!(hasSid && hasSn && hasGt)) + { + return OfferValidationResult.Fail(id, + "sid=, sn=, gt= must all be present together (opt-in ordering) or all be absent"); + } + if (!uint.TryParse(snVal, NumberStyles.None, CultureInfo.InvariantCulture, out var snParsed)) + { + return OfferValidationResult.Fail(id, "sn= must be a 32-bit unsigned integer"); + } + if (!uint.TryParse(gtVal, NumberStyles.None, CultureInfo.InvariantCulture, out var gtParsed)) + { + return OfferValidationResult.Fail(id, "gt= must be a 32-bit unsigned integer (0 = strict, >0 = seconds)"); + } + streamId = sidVal!; + streamSeq = snParsed; + streamGapTimeout = gtParsed; + } + return OfferValidationResult.Success(new IHaveOffer( - id, len, fmt, salt, clen, dst, ttl, headers, originator, masterId, fragment)); + id, len, fmt, salt, clen, dst, ttl, headers, originator, masterId, fragment, + streamId, streamSeq, streamGapTimeout)); } /// diff --git a/src/dapps/dapps.core/Services/InboundConnectionHandler.cs b/src/dapps/dapps.core/Services/InboundConnectionHandler.cs index 63a2c13..1a1304d 100644 --- a/src/dapps/dapps.core/Services/InboundConnectionHandler.cs +++ b/src/dapps/dapps.core/Services/InboundConnectionHandler.cs @@ -305,7 +305,10 @@ private async Task HandleRev(Stream stream, string command, CancellationToken ct originator: string.IsNullOrEmpty(msg.OriginatorCallsign) ? null : msg.OriginatorCallsign, masterId: msg.MasterId, fragmentIndex: msg.FragmentIndex, - fragmentTotal: msg.FragmentTotal); + fragmentTotal: msg.FragmentTotal, + streamId: msg.StreamId, + streamSeq: msg.StreamSeq, + streamGapTimeoutSeconds: msg.StreamGapTimeoutSeconds); if (!offered) { logger.LogInformation("rev drain: caller declined {0}", msg.Id); @@ -452,7 +455,10 @@ private async Task HandleData(Stream stream, string id, CancellationToken stoppi Originator: string.IsNullOrEmpty(offer.OriginatorCallsign) ? null : offer.OriginatorCallsign, MasterId: offer.MasterId, FragmentIndex: offer.FragmentIndex, - FragmentTotal: offer.FragmentTotal); + FragmentTotal: offer.FragmentTotal, + StreamId: offer.StreamId, + StreamSeq: offer.StreamSeq, + StreamGapTimeoutSeconds: offer.StreamGapTimeoutSeconds); await inbox.DeliverAsync(backhaulMessage, sourceCallsign, stoppingToken); await database.DeleteOffer(id); diff --git a/src/dapps/dapps.core/Services/MqttBrokerService.cs b/src/dapps/dapps.core/Services/MqttBrokerService.cs index 4f9c768..ffb74aa 100644 --- a/src/dapps/dapps.core/Services/MqttBrokerService.cs +++ b/src/dapps/dapps.core/Services/MqttBrokerService.cs @@ -171,6 +171,20 @@ public async Task InjectInboundMessage(DbMessage message) "dapps-ttl", residual.ToString(System.Globalization.CultureInfo.InvariantCulture)); } } + // Opt-in ordering: surface the stream id + seq the inbox + // delivered this on. Apps that opted into ordering by setting + // dapps-stream on outbound see the same property on inbound, + // so they can correlate cursors and (if they want) reject + // out-of-stream-id deliveries. + if (!string.IsNullOrEmpty(message.StreamId)) + { + builder = builder.WithUserProperty("dapps-stream", message.StreamId); + if (message.StreamSeq is { } sn) + { + builder = builder.WithUserProperty( + "dapps-stream-seq", sn.ToString(System.Globalization.CultureInfo.InvariantCulture)); + } + } var msg = builder.Build(); try @@ -324,11 +338,35 @@ private async Task OnInterceptingPublish(InterceptingPublishEventArgs e) ttl = parsedTtl; } + // Optional opt-in ordering: dapps-stream is the stream id; + // dapps-stream-gap-timeout is the policy in seconds (0 or + // missing = strict). The daemon allocates the seq. + string? streamId = null; + uint? streamGap = null; + var sidProp = e.ApplicationMessage.UserProperties? + .FirstOrDefault(p => string.Equals(p.Name, "dapps-stream", StringComparison.OrdinalIgnoreCase)); + if (sidProp is not null && !string.IsNullOrWhiteSpace(sidProp.Value) + && !sidProp.Value.Contains(' ') && !sidProp.Value.Contains('=') + && Encoding.UTF8.GetByteCount(sidProp.Value) <= 255) + { + streamId = sidProp.Value; + var gtProp = e.ApplicationMessage.UserProperties? + .FirstOrDefault(p => string.Equals(p.Name, "dapps-stream-gap-timeout", StringComparison.OrdinalIgnoreCase)); + if (gtProp is not null + && uint.TryParse(gtProp.Value, System.Globalization.NumberStyles.None, + System.Globalization.CultureInfo.InvariantCulture, out var parsedGap)) + { + streamGap = parsedGap; + } + } + try { - var id = await database.SubmitOutboundMessage(app, dest, e.ApplicationMessage.PayloadSegment.ToArray(), ttl); - logger.LogInformation("MQTT: queued outbound {0} from app {1} to {2} (ttl={3})", - id, app, dest, ttl?.ToString() ?? "none"); + var id = await database.SubmitOutboundMessage( + app, dest, e.ApplicationMessage.PayloadSegment.ToArray(), ttl, + streamId: streamId, streamGapTimeoutSeconds: streamGap); + logger.LogInformation("MQTT: queued outbound {0} from app {1} to {2} (ttl={3} stream={4})", + id, app, dest, ttl?.ToString() ?? "none", streamId ?? "none"); } catch (Exception ex) { diff --git a/src/dapps/dapps.core/Services/NodePoller.cs b/src/dapps/dapps.core/Services/NodePoller.cs index 3808bc1..f9018f5 100644 --- a/src/dapps/dapps.core/Services/NodePoller.cs +++ b/src/dapps/dapps.core/Services/NodePoller.cs @@ -69,7 +69,10 @@ public async Task PollAsync( Originator: polled.Originator, MasterId: polled.MasterId, FragmentIndex: polled.FragmentIndex, - FragmentTotal: polled.FragmentTotal); + FragmentTotal: polled.FragmentTotal, + StreamId: polled.StreamId, + StreamSeq: polled.StreamSeq, + StreamGapTimeoutSeconds: polled.StreamGapTimeoutSeconds); await inbox.DeliverAsync(inbound, remoteCallsign, ct); drained++; } diff --git a/src/dapps/dapps.core/Services/OutboundMessageManager.cs b/src/dapps/dapps.core/Services/OutboundMessageManager.cs index 28d2e39..bf0e6ff 100644 --- a/src/dapps/dapps.core/Services/OutboundMessageManager.cs +++ b/src/dapps/dapps.core/Services/OutboundMessageManager.cs @@ -113,7 +113,14 @@ private async Task DoRunCore(CancellationToken stoppingToken) // verbatim so the message stays groupable across hops. MasterId: message.MasterId, FragmentIndex: message.FragmentIndex, - FragmentTotal: message.FragmentTotal); + FragmentTotal: message.FragmentTotal, + // Opt-in ordering: stream trio is end-to-end at the + // originator's intent; intermediate hops re-emit + // verbatim so the destination sees the originator's + // gap-timeout policy regardless of forwarding path. + StreamId: message.StreamId, + StreamSeq: message.StreamSeq, + StreamGapTimeoutSeconds: message.StreamGapTimeoutSeconds); await ForwardAndObserveAsync(message, nh.Route, bm, optionsValue, stoppingToken); break; @@ -211,7 +218,10 @@ private async Task FloodAndMarkAsync( TraversedHops: flood.TraversedHops, MasterId: message.MasterId, FragmentIndex: message.FragmentIndex, - FragmentTotal: message.FragmentTotal); + FragmentTotal: message.FragmentTotal, + StreamId: message.StreamId, + StreamSeq: message.StreamSeq, + StreamGapTimeoutSeconds: message.StreamGapTimeoutSeconds); var backhaul = backhauls.FirstOrDefault(b => b.CanHandle(route)); if (backhaul is null) continue; diff --git a/src/dapps/dapps.core/Services/StreamGapSweeperService.cs b/src/dapps/dapps.core/Services/StreamGapSweeperService.cs new file mode 100644 index 0000000..dae4015 --- /dev/null +++ b/src/dapps/dapps.core/Services/StreamGapSweeperService.cs @@ -0,0 +1,68 @@ +namespace dapps.core.Services; + +/// +/// Periodically advances per-(sender, stream-id) cursors past elapsed +/// gap deadlines for opt-in-ordered streams in timeout mode (gt>0). +/// Each tick scans streamrecvstate rows whose GapDeadline +/// has passed; for each, asks the inbox to skip the gap and drain +/// successors. +/// +/// Strict streams (gt=0) never set a deadline, so this sweeper ignores +/// them - parked rows stay parked until the missing seq fills in or +/// the regular TTL sweeper drops them via the live message's TTL. +/// +public sealed class StreamGapSweeperService( + Database database, + DatabaseAndMqttInbox inbox, + TimeProvider timeProvider, + ILogger logger) : BackgroundService +{ + /// How often to scan for elapsed deadlines. Same cadence + /// as the TTL sweeper (1 minute). The granularity is fine for + /// packet-radio gap timeouts, which are typically minutes-to-tens- + /// of-minutes - operators expecting sub-second skip behaviour + /// would not be using packet radio in the first place. + public TimeSpan SweepInterval { get; init; } = TimeSpan.FromMinutes(1); + + protected override async Task ExecuteAsync(CancellationToken stoppingToken) + { + using var timer = new PeriodicTimer(SweepInterval, timeProvider); + await SweepOnce(stoppingToken); + try + { + while (await timer.WaitForNextTickAsync(stoppingToken)) + { + await SweepOnce(stoppingToken); + } + } + catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) + { + } + } + + private async Task SweepOnce(CancellationToken ct) + { + var now = timeProvider.GetUtcNow().UtcDateTime; + try + { + var stale = await database.GetStaleStreamGapsAsync(now); + foreach (var recv in stale) + { + try + { + await inbox.SkipGapAsync(recv, ct); + } + catch (Exception ex) + { + logger.LogError(ex, + "Gap-skip failed for stream {0}|{1}", + recv.SenderCallsign, recv.StreamId); + } + } + } + catch (Exception ex) + { + logger.LogError(ex, "stream gap sweeper threw"); + } + } +}