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 }