lib/reticulum/src/node/transport/retransmit.zig
daab053ee43316e1809a84551d573ddd1e5bf3d2
1 const std = @import("std");
2 const reticulum = @import("../../root.zig");
3 const announces = @import("announces.zig");
4
5 const destination = reticulum.destination;
6 const node = reticulum.node;
7 const packet = reticulum.packet;
8 const wire = reticulum.wire;
9
10 /// Describes what becomes of an accepted announce at the rebroadcast queue, so
11 /// the receive path knows whether accepting an announce also queues one to
12 /// send.
13 pub const Plan = enum {
14 none,
15 write,
16 too_large,
17 };
18
19 const Visit = enum {
20 kept,
21 completed,
22 stopped,
23 };
24
25 /// Decides whether a transport node rebroadcasts, following Reticulum@1.5.0
26 /// RNS/Transport.py:2267-2289, so a caller can reserve effects for the
27 /// rebroadcast in the same step before it accepts an announce. An announce the
28 /// path table refused leaves the queue untouched, as does one that reached a
29 /// node carrying no traffic for others. An announce marked a path response
30 /// leaves the queue untouched, because it answers the one carrier that asked.
31 /// An announce at or past the pathfinder maximum leaves the queue untouched. A
32 /// HEADER_1 payload above 465 bytes has no room left under a HEADER_2
33 /// rebroadcast, so the step reports it and queues nothing.
34 pub fn plan(node_owner: *const node.Node, value: wire.Packet, admitted: bool) Plan {
35 std.debug.assert(value.packet_type == .announce);
36 if (!admitted) return .none;
37 if (!node_owner.transport.enabled) return .none;
38 if (value.context == .path_response) return .none;
39 if (value.hops >= wire.pathfinder_hops) return .none;
40 if (value.payload.len > destination.announce.payload_bytes_max) return .too_large;
41 return .write;
42 }
43
44 /// Answers whether a slot stands free for the timer that drives the rebroadcast
45 /// queue, so a caller checks before changing any table because a sweep with
46 /// nowhere to put its timer has to fail the whole step.
47 pub fn canSchedule(node_owner: *const node.Node) bool {
48 return node_owner.timers.canScheduleAfterCancel(.announces, null);
49 }
50
51 /// Completes entries that neighbors rebroadcast, following Reticulum@1.5.0
52 /// RNS/Transport.py:2093-2116, so the node learns its own copy is no longer
53 /// needed when a neighbor carries the same announce. An announce heard at this
54 /// entry's own hop count counts as a neighbor rebroadcast, and two of them
55 /// drop the entry once the node has sent it at least once. An announce heard
56 /// one hop further on means a neighbor already passed this entry along, so the
57 /// step drops the entry when the node has sent it and its next send is still
58 /// ahead.
59 pub fn detect(node_owner: *node.Node, value: wire.Packet, now: node.Seconds) void {
60 if (!node_owner.transport.enabled) return;
61 if (value.transport_id == null) return;
62 const entry = node_owner.transport.announces.find(value.destination) orelse return;
63 std.debug.assert(value.hops >= 1);
64 std.debug.assert(entry.record.hops <= wire.pathfinder_hops);
65 const heard: u16 = value.hops - 1;
66 const record = &entry.record;
67 if (heard == record.hops) {
68 record.local_rebroadcasts +|= 1;
69 if (record.retries > 0 and record.local_rebroadcasts >= announces.local_rebroadcasts_max) {
70 _ = node_owner.transport.announces.remove(value.destination);
71 return;
72 }
73 }
74 const passed_on = heard == @as(u16, record.hops) + 1;
75 if (passed_on and record.retries > 0 and now < record.due) {
76 _ = node_owner.transport.announces.remove(value.destination);
77 }
78 }
79
80 /// Queues the announce with its turn set one second ahead, as Reticulum@1.5.0
81 /// RNS/Transport.py:2248-2289 does, so the step turns the decision into a
82 /// queued rebroadcast and arms the sweep that will send it. A payload too large
83 /// to relay reports `announce_relay_too_large`. A full announce table reports
84 /// `announce_table_full`.
85 pub fn enter(
86 node_owner: *node.Node,
87 value: wire.Packet,
88 decision: Plan,
89 hash: packet.Hash,
90 now: node.Seconds,
91 ) void {
92 switch (decision) {
93 .none => {},
94 .too_large => node.inbound.diagnostic(node_owner, .announce_relay_too_large, hash),
95 .write => {
96 const due = now +| announces.first_delay;
97 _ = node_owner.transport.announces.insert(announces.Record.init(.{
98 .destination = value.destination,
99 .due = due,
100 .timestamp = now,
101 .retries = 0,
102 .hops = value.hops,
103 .block_rebroadcasts = false,
104 .attached = null,
105 .context_flag = value.context_flag,
106 .payload = value.payload,
107 })) catch {
108 node.inbound.diagnostic(node_owner, .announce_table_full, hash);
109 return;
110 };
111 schedule(node_owner, due);
112 },
113 }
114 }
115
116 /// Moves the timer only when the second given falls sooner than the one already
117 /// set, because entries share one timer.
118 pub fn schedule(node_owner: *node.Node, at: node.Seconds) void {
119 if (node_owner.timers.scheduledAt(.announces)) |scheduled| {
120 if (scheduled <= at) return;
121 }
122 node_owner.timers.schedule(.announces, at) catch unreachable;
123 node_owner.effects.push(.{ .schedule_timer = .{
124 .id = .announces,
125 .at = at,
126 } }) catch unreachable;
127 }
128
129 /// Sends due rebroadcasts by ascending hops when the timer fires, following
130 /// Reticulum@1.5.0 RNS/Transport.py:737-799,1239-1240. A pass that would
131 /// overrun the room for effects breaks off and sets the timer at the second it
132 /// broke off on. An empty table cancels the timer and returns. A destination
133 /// whose identity the node no longer recalls is dropped from the queue.
134 pub fn sweep(node_owner: *node.Node, now: node.Seconds) node.StepError!void {
135 if (node_owner.transport.announces.count() == 0) {
136 _ = node_owner.timers.cancel(.announces);
137 return;
138 }
139 const widest = node.outbound.frameCount(node_owner, .all);
140 try node.outbound.reserve(node_owner, widest + 1, widest);
141 _ = node_owner.timers.cancel(.announces);
142 var hops: u16 = 0;
143 while (hops <= wire.pathfinder_hops) : (hops += 1) {
144 var index: usize = 0;
145 while (index < node_owner.transport.announces.len) {
146 const entry = &node_owner.transport.announces.entries[index];
147 if (entry.record.hops != hops or entry.record.due > now) {
148 index += 1;
149 continue;
150 }
151 switch (try visit(node_owner, entry, now)) {
152 .kept => index += 1,
153 .completed => node_owner.transport.announces.removeAt(index),
154 .stopped => {
155 schedule(node_owner, now);
156 return;
157 },
158 }
159 }
160 }
161 if (node_owner.transport.announces.earliestDue()) |due| schedule(node_owner, due);
162 }
163
164 /// Answers whether an entry has reached the rebroadcast or retry limit,
165 /// following Reticulum@1.5.0 RNS/Transport.py:743-748, so an announce stops
166 /// circulating.
167 fn completed(record: *const announces.Record) bool {
168 if (record.retries > 0 and record.retries >= announces.local_rebroadcasts_max) return true;
169 return record.retries > announces.retries_max;
170 }
171
172 fn visit(node_owner: *node.Node, entry: *announces.Entry, now: node.Seconds) node.StepError!Visit {
173 const record = &entry.record;
174 std.debug.assert(record.retries <= announces.local_rebroadcasts_max);
175 std.debug.assert(record.due <= now);
176 if (completed(record)) return .completed;
177 if (node_owner.known_identities.recall(record.destination) == null) return .completed;
178 const fanout: node.outbound.Fanout = if (record.attached) |index| .{ .one = index } else .all;
179 const routed = record.attached != null or
180 node_owner.transport.paths.find(record.destination, now) != null;
181 const carriers = if (routed) node.outbound.frameCount(node_owner, fanout) else 0;
182 node.outbound.reserve(node_owner, carriers + 1, carriers) catch return .stopped;
183 if (carriers > 0) {
184 const frame = encode(node_owner, record);
185 _ = node_owner.duplicate_hashes.insert(wire.hash.full(frame) catch unreachable);
186 try node.outbound.transmit(node_owner, fanout, frame);
187 }
188 record.due = now +| announces.retry_delay;
189 record.retries += 1;
190 _ = entry.release(now);
191 return .kept;
192 }
193
194 /// Assembles the outgoing packet under a HEADER_2 header from the payload the
195 /// queue kept, as Reticulum@1.5.0 RNS/Transport.py:756-779 does, carrying this
196 /// node's own transport identity so the next node learns who passed it on. A
197 /// record that blocks further rebroadcasts is marked a path response.
198 fn encode(node_owner: *node.Node, record: *const announces.Record) []const u8 {
199 const transport_hash = node_owner.transport.identity_hash orelse unreachable;
200 std.debug.assert(record.hops < wire.pathfinder_hops);
201 return wire.encode(.{
202 .ifac = 0,
203 .header = .two,
204 .context_flag = record.context_flag,
205 .transport = .transport,
206 .destination_type = .single,
207 .packet_type = .announce,
208 .hops = record.hops,
209 .transport_id = transport_hash,
210 .destination = record.destination,
211 .context = if (record.block_rebroadcasts) .path_response else .none,
212 .payload = record.payload(),
213 }, node_owner.scratch) catch unreachable;
214 }