lib/reticulum/src/node/transport/requests.zig

daab053ee43316e1809a84551d573ddd1e5bf3d2

  1 const std = @import("std");
  2 const reticulum = @import("../../root.zig");
  3 const announces = @import("announces.zig");
  4 const retransmit = @import("retransmit.zig");
  5 
  6 const carrier = reticulum.carrier;
  7 const destination = reticulum.destination;
  8 const node = reticulum.node;
  9 const packet = reticulum.packet;
 10 const wire = reticulum.wire;
 11 
 12 const hash_bytes = reticulum.hash.truncated_bytes;
 13 
 14 /// The destination every path request is addressed to so any node recognizes
 15 /// the question without arranging anything in advance, which Reticulum@1.5.0
 16 /// RNS/Transport.py:347-351 names rnstransport.path.request.
 17 pub const destination_hash: [hash_bytes]u8 = hash: {
 18     @setEvalBranchQuota(1_000_000);
 19     break :hash destination.hash(
 20         .{ .name_bytes_max = 64 },
 21         "rnstransport",
 22         &.{ "path", "request" },
 23         null,
 24     ) catch unreachable;
 25 };
 26 
 27 /// Three hashes for sizing a buffer to hold the largest payload a request can
 28 /// take, the room Reticulum@1.5.0 RNS/Transport.py:3220-3224 needs for a
 29 /// destination, a transport identity, and a tag.
 30 pub const payload_bytes_max: usize = 3 * hash_bytes;
 31 
 32 /// One second, the pause Reticulum@1.5.0 RNS/Transport.py:133,3425 leaves
 33 /// before replying from a path the node holds so questions arriving together
 34 /// share one answer.
 35 pub const response_delay: node.Seconds = 1;
 36 
 37 /// One parsed path request, which a caller reads after parsing to see what was
 38 /// asked and who asked. The tag points into the payload bytes the packet
 39 /// arrived in, so it lives exactly as long as they do.
 40 pub const Request = struct {
 41     destination: [hash_bytes]u8,
 42     requestor: ?[hash_bytes]u8,
 43     tag: []const u8,
 44 
 45     /// Hashes the destination and tag bytes into the key the node deduplicates
 46     /// on, so the node recognizes the same request arriving again by another
 47     /// route, following Reticulum@1.5.0 RNS/Transport.py:1755.
 48     pub fn key(self: Request) packet.Hash {
 49         std.debug.assert(self.tag.len >= 1);
 50         std.debug.assert(self.tag.len <= hash_bytes);
 51         var hasher = reticulum.hash.Hasher.init();
 52         hasher.update(&self.destination);
 53         hasher.update(self.tag);
 54         return hasher.finalFull();
 55     }
 56 };
 57 
 58 pub const ParseError = error{ Short, Tagless };
 59 
 60 const Answer = enum {
 61     local,
 62     known_path,
 63     oversized,
 64     silent,
 65     discover,
 66     engaged,
 67     ignore,
 68 
 69     /// Sorts the answers that settle a request from the ones that leave it
 70     /// open, which Reticulum@1.5.0 RNS/Transport.py:3520-3524 uses to decide
 71     /// when the gate goes.
 72     fn answered(self: Answer) bool {
 73         return switch (self) {
 74             .local, .known_path, .oversized, .silent => true,
 75             .discover, .engaged, .ignore => false,
 76         };
 77     }
 78 };
 79 
 80 /// Splits a path request payload so the caller gets the destination, the asker,
 81 /// and the tag, following Reticulum@1.5.0 RNS/Transport.py:1738-1753,3310-3330.
 82 /// A payload shorter than one hash returns `error.Short`, and one of exactly a
 83 /// hash returns `error.Tagless`. A payload past two hashes carries the asker's
 84 /// transport identity, and the tag follows it.
 85 pub fn parse(payload: []const u8) ParseError!Request {
 86     if (payload.len < hash_bytes) return error.Short;
 87     if (payload.len == hash_bytes) return error.Tagless;
 88     const transported = payload.len > 2 * hash_bytes;
 89     const tag_start: usize = if (transported) 2 * hash_bytes else hash_bytes;
 90     const tag_end = @min(payload.len, tag_start + hash_bytes);
 91     std.debug.assert(tag_start < tag_end);
 92     return .{
 93         .destination = payload[0..hash_bytes].*,
 94         .requestor = if (transported) payload[hash_bytes .. 2 * hash_bytes].* else null,
 95         .tag = payload[tag_start..tag_end],
 96     };
 97 }
 98 
 99 /// Answers whether an arriving packet is one of these requests so the receive
100 /// path hands it over before ordinary delivery, which holds for a data packet
101 /// of the plain kind carrying the well-known destination.
102 pub fn addressed(value: wire.Packet) bool {
103     if (value.packet_type != .data) return false;
104     if (value.destination_type != .plain) return false;
105     return std.mem.eql(u8, &value.destination, &destination_hash);
106 }
107 
108 /// Takes one arriving request through every check before the node commits to an
109 /// answer, as Reticulum@1.5.0 RNS/Transport.py:1734-1797 does. A payload that
110 /// fails to parse reports `path_request_malformed`. A request whose key the
111 /// node already holds reports `path_request_duplicate`. A request for a
112 /// destination the node is itself waiting on adds the asking carrier to the
113 /// discovery and reports `path_request_batched`. The step reserves room for
114 /// every frame the answer may send before it changes any table. The node stores
115 /// the packet hash and the request key after the gate has chosen an answer, so
116 /// a request turned away early leaves neither behind.
117 pub fn receive(
118     node_owner: *node.Node,
119     value: wire.Packet,
120     interface: carrier.Index,
121     hash: packet.Hash,
122     now: node.Seconds,
123 ) node.StepError!void {
124     std.debug.assert(addressed(value));
125     const request = parse(value.payload) catch {
126         try node.outbound.reserve(node_owner, 1, 0);
127         node.inbound.diagnostic(node_owner, .path_request_malformed, hash);
128         return;
129     };
130     const state = &node_owner.transport;
131     const key = request.key();
132     if (state.tags.contains(key)) {
133         try node.outbound.reserve(node_owner, 1, 0);
134         node.inbound.diagnostic(node_owner, .path_request_duplicate, hash);
135         return;
136     }
137     if (state.inflight_requests.find(request.destination, now) != null) {
138         try node.outbound.reserve(node_owner, 1, 0);
139         _ = state.tags.insert(key);
140         state.discoveries.batch(request.destination, interface, now) catch {
141             node.inbound.diagnostic(node_owner, .discovery_table_full, hash);
142             return;
143         };
144         node.inbound.diagnostic(node_owner, .path_request_batched, hash);
145         return;
146     }
147     const answer = plan(node_owner, request, interface, now);
148     const forwards = if (answer == .discover)
149         node.outbound.frameCount(node_owner, .{ .all_except = interface })
150     else
151         0;
152     try node.outbound.reserve(node_owner, forwards + 1, forwards);
153     if (answer == .known_path and !retransmit.canSchedule(node_owner)) return error.TimerFull;
154     _ = state.tags.insert(key);
155     _ = node_owner.duplicate_hashes.insert(hash);
156     if (answer.answered()) {
157         _ = state.inflight_requests.remove(request.destination);
158     } else {
159         state.inflight_requests.insert(request.destination, now);
160     }
161     try respond(node_owner, answer, request, interface, hash, now);
162 }
163 
164 /// Chooses the first answer that matches, following Reticulum@1.5.0
165 /// RNS/Transport.py:3375-3518, so the node decides what to say about a
166 /// destination. A destination this node owns is answered by the node itself
167 /// when it is a single destination, and ignored otherwise. A node that carries
168 /// no traffic for others ignores the request. A known path whose announce
169 /// payload was too large to keep reports `announce_relay_too_large`. A known
170 /// path at or past the pathfinder maximum is left unanswered, as is a path
171 /// whose next hop is the asker. A destination with no path is passed on, unless
172 /// the arriving carrier is configured to ask nothing or a discovery for it has
173 /// already gone out.
174 fn plan(
175     node_owner: *node.Node,
176     request: Request,
177     interface: carrier.Index,
178     now: node.Seconds,
179 ) Answer {
180     if (node_owner.destinations.find(request.destination)) |entry| {
181         return if (entry.kind == .single) .local else .ignore;
182     }
183     const state = &node_owner.transport;
184     if (!state.enabled) return .ignore;
185     if (state.paths.find(request.destination, now)) |path| {
186         if (path.announcePayload() == null) return .oversized;
187         if (path.hops >= wire.pathfinder_hops) return .silent;
188         const requestor = request.requestor orelse return .known_path;
189         if (std.mem.eql(u8, &requestor, &path.next_hop)) return .silent;
190         return .known_path;
191     }
192     if (node_owner.interfaces[interface].discover_paths == 0) return .ignore;
193     const waiting = state.discoveries.find(request.destination, now) orelse return .discover;
194     return if (waiting.engaged) .engaged else .discover;
195 }
196 
197 fn respond(
198     node_owner: *node.Node,
199     answer: Answer,
200     request: Request,
201     interface: carrier.Index,
202     hash: packet.Hash,
203     now: node.Seconds,
204 ) node.StepError!void {
205     switch (answer) {
206         .local => node_owner.effects.push(.{ .path_request = .{
207             .destination = request.destination,
208             .interface = interface,
209         } }) catch unreachable,
210         .known_path => answerKnown(node_owner, request, interface, hash, now),
211         .discover => try discover(node_owner, request, interface, hash, now),
212         .oversized => node.inbound.diagnostic(node_owner, .announce_relay_too_large, hash),
213         .silent, .engaged, .ignore => {},
214     }
215 }
216 
217 /// Queues an answer built from the announce the path kept, marked so the asker
218 /// takes the path and passes it no further, as Reticulum@1.5.0
219 /// RNS/Transport.py:3409-3444 does. The answer goes out on the carrier the
220 /// question arrived on, one second later. A destination with a rebroadcast
221 /// already queued keeps that record behind the answer. A full announce table
222 /// reports `announce_table_full`.
223 fn answerKnown(
224     node_owner: *node.Node,
225     request: Request,
226     interface: carrier.Index,
227     hash: packet.Hash,
228     now: node.Seconds,
229 ) void {
230     const state = &node_owner.transport;
231     const path = state.paths.find(request.destination, now) orelse unreachable;
232     const due = now +| response_delay;
233     const record = announces.Record.init(.{
234         .destination = request.destination,
235         .due = due,
236         .timestamp = now,
237         .retries = announces.retries_max,
238         .hops = path.hops,
239         .block_rebroadcasts = true,
240         .attached = interface,
241         .context_flag = path.context_flag,
242         .payload = path.announcePayload() orelse unreachable,
243     });
244     if (state.announces.find(request.destination)) |entry| {
245         entry.hold(record);
246     } else {
247         _ = state.announces.insert(record) catch {
248             node.inbound.diagnostic(node_owner, .announce_table_full, hash);
249             return;
250         };
251     }
252     retransmit.schedule(node_owner, due);
253 }
254 
255 /// Engages discovery and forwards the same tag so the answer is recognized on
256 /// the way back, following Reticulum@1.5.0 RNS/Transport.py:3478-3508. A full
257 /// discovery table reports `discovery_table_full`. The forwarded request goes
258 /// to every carrier but the one it arrived on.
259 fn discover(
260     node_owner: *node.Node,
261     request: Request,
262     interface: carrier.Index,
263     hash: packet.Hash,
264     now: node.Seconds,
265 ) node.StepError!void {
266     node_owner.transport.discoveries.engage(request.destination, interface, now) catch {
267         node.inbound.diagnostic(node_owner, .discovery_table_full, hash);
268         return;
269     };
270     const fanout: node.outbound.Fanout = .{ .all_except = interface };
271     if (node.outbound.frameCount(node_owner, fanout) == 0) return;
272     const frame = encode(node_owner, request.destination, request.tag);
273     _ = node_owner.duplicate_hashes.insert(wire.hash.full(frame) catch unreachable);
274     try node.outbound.transmit(node_owner, fanout, frame);
275 }
276 
277 /// Puts a request to the network carrying tag bytes the caller chose so an
278 /// application asks for a path to a destination, as Reticulum@1.5.0
279 /// RNS/Transport.py:3219-3250 does. A node with no carrier that sends returns
280 /// `error.NoOutgoingCarrier`.
281 pub fn send(node_owner: *node.Node, value: node.ApplicationPathRequest) node.StepError!void {
282     const fanout: node.outbound.Fanout = if (value.interface) |index| .{ .one = index } else .all;
283     const carriers = node.outbound.frameCount(node_owner, fanout);
284     if (carriers == 0) return error.NoOutgoingCarrier;
285     try node.outbound.reserve(node_owner, carriers, carriers);
286     const frame = encode(node_owner, value.destination, &value.tag);
287     _ = node_owner.duplicate_hashes.insert(wire.hash.full(frame) catch unreachable);
288     try node.outbound.transmit(node_owner, fanout, frame);
289 }
290 
291 /// Twenty seconds, the least time Reticulum@1.5.0 RNS/Transport.py:135 leaves
292 /// between two requests about one destination so queries for a failing
293 /// destination remain spaced. The second of the last request comes from the
294 /// gate itself, because this port writes down nothing else about the requests
295 /// it sends.
296 pub const rediscovery_interval: node.Seconds = 20;
297 
298 fn throttled(node_owner: *const node.Node, requested: [hash_bytes]u8, now: node.Seconds) bool {
299     const gate = node_owner.transport.inflight_requests.find(requested, now) orelse return false;
300     return now -| gate.timestamp < rediscovery_interval;
301 }
302 
303 /// Counts how many frames asking after this destination again would put out at
304 /// this second so a caller reserves effects before calling `rediscover`,
305 /// because the reservation has to happen before anything changes.
306 pub fn rediscoveryFrames(
307     node_owner: *const node.Node,
308     requested: [hash_bytes]u8,
309     now: node.Seconds,
310 ) usize {
311     if (throttled(node_owner, requested, now)) return 0;
312     return node.outbound.frameCount(node_owner, .all);
313 }
314 
315 /// Drops the path to the destination and asks every carrier for a new one,
316 /// following Reticulum@1.5.0 RNS/Transport.py:676-695, so a caller whose link
317 /// failed throws away what the node knew and asks again. The path goes whether
318 /// the request travels or the spacing holds it back, which leaves the
319 /// destination open to a later announce from any distance. The caller reserves
320 /// the frames that `rediscoveryFrames` counts.
321 pub fn rediscover(
322     node_owner: *node.Node,
323     requested: [hash_bytes]u8,
324     tag: *const [hash_bytes]u8,
325     now: node.Seconds,
326 ) node.StepError!void {
327     _ = node_owner.transport.paths.expire(requested, now);
328     if (throttled(node_owner, requested, now)) return;
329     if (node.outbound.frameCount(node_owner, .all) == 0) return;
330     const frame = encode(node_owner, requested, tag);
331     _ = node_owner.duplicate_hashes.insert(wire.hash.full(frame) catch unreachable);
332     node_owner.transport.inflight_requests.insert(requested, now);
333     try node.outbound.transmit(node_owner, .all, frame);
334 }
335 
336 /// Builds a HEADER_1 PLAIN request frame for every path request this node sends
337 /// or passes on, following Reticulum@1.5.0 RNS/Transport.py:3219-3227. A node
338 /// that carries traffic for others puts its transport identity between the
339 /// destination and the tag.
340 fn encode(node_owner: *node.Node, requested: [hash_bytes]u8, tag: []const u8) []const u8 {
341     std.debug.assert(tag.len >= 1);
342     std.debug.assert(tag.len <= hash_bytes);
343     var payload: [payload_bytes_max]u8 = undefined;
344     @memcpy(payload[0..hash_bytes], &requested);
345     var length: usize = hash_bytes;
346     if (node_owner.transport.enabled) {
347         const transport_hash = node_owner.transport.identity_hash orelse unreachable;
348         @memcpy(payload[length..][0..hash_bytes], &transport_hash);
349         length += hash_bytes;
350     }
351     @memcpy(payload[length..][0..tag.len], tag);
352     length += tag.len;
353     return wire.encode(.{
354         .ifac = 0,
355         .header = .one,
356         .context_flag = 0,
357         .transport = .broadcast,
358         .destination_type = .plain,
359         .packet_type = .data,
360         .hops = 0,
361         .transport_id = null,
362         .destination = destination_hash,
363         .context = .none,
364         .payload = payload[0..length],
365     }, node_owner.scratch) catch unreachable;
366 }
367 
368 fn answersWaiting(node_owner: *const node.Node, value: wire.Packet) bool {
369     if (node_owner.transport.identity_hash == null) return false;
370     if (value.hops >= wire.pathfinder_hops) return false;
371     return value.payload.len <= destination.announce.payload_bytes_max;
372 }
373 
374 /// Counts how many frames replying to the carriers queued behind this announce
375 /// would put out so a caller reserves effects before accepting an announce,
376 /// because answering the waiting carriers happens in the same step.
377 pub fn waitingFrames(node_owner: *node.Node, value: wire.Packet, now: node.Seconds) usize {
378     std.debug.assert(value.packet_type == .announce);
379     if (!answersWaiting(node_owner, value)) return 0;
380     const waiting = node_owner.transport.discoveries.find(value.destination, now) orelse return 0;
381     var frames: usize = 0;
382     var requesters = waiting.requesters.iterator(.{});
383     while (requesters.next()) |index| {
384         frames += node.outbound.frameCount(node_owner, .{ .one = @intCast(index) });
385     }
386     return frames;
387 }
388 
389 /// Answers waiting carriers at once so one announce replies to every carrier
390 /// that asked about its destination, following Reticulum@1.5.0
391 /// RNS/Transport.py:2347-2371. The answer carries this node's transport
392 /// identity and is marked a path response. An announce at or past the
393 /// pathfinder maximum, or one whose payload is too large to relay, answers
394 /// nobody.
395 pub fn answerWaiting(
396     node_owner: *node.Node,
397     value: wire.Packet,
398     now: node.Seconds,
399 ) node.StepError!void {
400     std.debug.assert(value.packet_type == .announce);
401     const waiting = node_owner.transport.discoveries.take(value.destination, now) orelse return;
402     if (!answersWaiting(node_owner, value)) return;
403     const frame = wire.encode(.{
404         .ifac = 0,
405         .header = .two,
406         .context_flag = value.context_flag,
407         .transport = .transport,
408         .destination_type = .single,
409         .packet_type = .announce,
410         .hops = value.hops,
411         .transport_id = node_owner.transport.identity_hash,
412         .destination = value.destination,
413         .context = .path_response,
414         .payload = value.payload,
415     }, node_owner.scratch) catch unreachable;
416     _ = node_owner.duplicate_hashes.insert(wire.hash.full(frame) catch unreachable);
417     var requesters = waiting.requesters.iterator(.{});
418     while (requesters.next()) |index| {
419         try node.outbound.transmit(node_owner, .{ .one = @intCast(index) }, frame);
420     }
421 }
422 
423 test "Reticulum@1.5.0 RNS/Transport.py:1738-1753 parses every request payload length class" {
424     var payload: [64]u8 = undefined;
425     for (&payload, 0..) |*byte, index| byte.* = @intCast(index);
426     try std.testing.expectError(error.Short, parse(payload[0..15]));
427     try std.testing.expectError(error.Tagless, parse(payload[0..16]));
428     const one = try parse(payload[0..17]);
429     try std.testing.expect(one.requestor == null);
430     try std.testing.expectEqualSlices(u8, payload[16..17], one.tag);
431     const endpoint = try parse(payload[0..32]);
432     try std.testing.expect(endpoint.requestor == null);
433     try std.testing.expectEqualSlices(u8, payload[16..32], endpoint.tag);
434     const short_tag = try parse(payload[0..33]);
435     try std.testing.expectEqualSlices(u8, payload[16..32], &short_tag.requestor.?);
436     try std.testing.expectEqualSlices(u8, payload[32..33], short_tag.tag);
437     const transported = try parse(payload[0..48]);
438     try std.testing.expectEqualSlices(u8, payload[32..48], transported.tag);
439     const long = try parse(payload[0..64]);
440     try std.testing.expectEqualSlices(u8, payload[0..16], &long.destination);
441     try std.testing.expectEqualSlices(u8, payload[32..48], long.tag);
442 }
443 
444 test "Reticulum@1.5.0 RNS/Transport.py:1755 keys a tag the same across request shapes" {
445     var endpoint: [32]u8 = @splat(0x11);
446     var transported: [48]u8 = @splat(0x11);
447     @memset(transported[16..32], 0x22);
448     const first = (try parse(&endpoint)).key();
449     const second = (try parse(&transported)).key();
450     try std.testing.expectEqualSlices(u8, &first, &second);
451     endpoint[31] = 0x12;
452     const changed = (try parse(&endpoint)).key();
453     try std.testing.expect(!std.mem.eql(u8, &first, &changed));
454 }