lib/reticulum/src/node/transport/announces.zig
daab053ee43316e1809a84551d573ddd1e5bf3d2
1 const std = @import("std");
2 const alloc_phase = @import("alloc_phase");
3 const carrier = @import("../../carrier/root.zig");
4 const destination = @import("../../destination/root.zig");
5
6 pub const Seconds = u64;
7
8 /// Two rebroadcasts heard from neighbors finish a pending entry, the count
9 /// Reticulum@1.5.0 RNS/Transport.py:129 fixes.
10 pub const local_rebroadcasts_max: u8 = 2;
11 /// One retry follows the first rebroadcast, the allowance Reticulum@1.5.0
12 /// RNS/Transport.py:120 sets.
13 pub const retries_max: u8 = 1;
14 /// The whole second this port waits before a first rebroadcast, standing in for
15 /// the wait of up to half a second at Reticulum@1.5.0 RNS/Transport.py:2254.
16 pub const first_delay: Seconds = 1;
17 /// The whole second this port waits before a retry, standing in for the 5.5
18 /// second wait at Reticulum@1.5.0 RNS/Transport.py:751.
19 pub const retry_delay: Seconds = 6;
20
21 /// The values that describe one new pending rebroadcast.
22 pub const Fields = struct {
23 destination: [16]u8,
24 due: Seconds,
25 timestamp: Seconds,
26 retries: u8,
27 hops: u8,
28 block_rebroadcasts: bool,
29 attached: ?carrier.Index,
30 context_flag: u1,
31 payload: []const u8,
32 };
33
34 /// One pending rebroadcast, matching the record at Reticulum@1.5.0
35 /// RNS/Transport.py:4061-4069.
36 pub const Record = struct {
37 destination: [16]u8,
38 due: Seconds,
39 timestamp: Seconds,
40 retries: u8,
41 hops: u8,
42 local_rebroadcasts: u8,
43 block_rebroadcasts: bool,
44 attached: ?carrier.Index,
45 context_flag: u1,
46 payload_len: u16,
47 payload_bytes: [destination.announce.payload_bytes_max]u8,
48
49 /// Builds a record carrying its own copy of the announce payload, which
50 /// runs to 465 bytes at most.
51 pub fn init(fields: Fields) Record {
52 std.debug.assert(fields.payload.len <= destination.announce.payload_bytes_max);
53 std.debug.assert(fields.retries <= retries_max);
54 var record = Record{
55 .destination = fields.destination,
56 .due = fields.due,
57 .timestamp = fields.timestamp,
58 .retries = fields.retries,
59 .hops = fields.hops,
60 .local_rebroadcasts = 0,
61 .block_rebroadcasts = fields.block_rebroadcasts,
62 .attached = fields.attached,
63 .context_flag = fields.context_flag,
64 .payload_len = @intCast(fields.payload.len),
65 .payload_bytes = @splat(0),
66 };
67 @memcpy(record.payload_bytes[0..fields.payload.len], fields.payload);
68 return record;
69 }
70
71 pub fn payload(self: *const Record) []const u8 {
72 std.debug.assert(self.payload_len <= self.payload_bytes.len);
73 return self.payload_bytes[0..self.payload_len];
74 }
75 };
76
77 /// One destination with the rebroadcast queued for it and any second record
78 /// waiting its turn.
79 pub const Entry = struct {
80 record: Record,
81 held: ?Record,
82
83 /// Puts the entry's current record behind a new one, following
84 /// Reticulum@1.5.0 RNS/Transport.py:3438-3444.
85 pub fn hold(self: *Entry, record: Record) void {
86 std.debug.assert(std.mem.eql(u8, &self.record.destination, &record.destination));
87 self.held = self.record;
88 self.record = record;
89 }
90
91 /// Returns a held record to the front and gives it a due second later than
92 /// `now`, following Reticulum@1.5.0 RNS/Transport.py:791-794. The call
93 /// reports whether a held record was there.
94 pub fn release(self: *Entry, now: Seconds) bool {
95 const held = self.held orelse return false;
96 std.debug.assert(std.mem.eql(u8, &self.record.destination, &held.destination));
97 self.record = held;
98 self.record.due = @max(held.due, now +| 1);
99 self.held = null;
100 return true;
101 }
102 };
103
104 const TableLimits = struct {
105 announces_max: usize,
106 };
107
108 const TableCapacity = struct {
109 announces_max: usize,
110 storage_bytes: usize,
111
112 pub const DeriveError = error{ InvalidLimit, CapacityOverflow };
113
114 pub fn derive(limits: TableLimits) DeriveError!TableCapacity {
115 if (limits.announces_max == 0) return error.InvalidLimit;
116 const storage_bytes = alloc_phase.capacity.mul(
117 usize,
118 limits.announces_max,
119 @sizeOf(Entry),
120 ) catch return error.CapacityOverflow;
121 return .{ .announces_max = limits.announces_max, .storage_bytes = storage_bytes };
122 }
123 };
124
125 /// Pending announce rebroadcasts, matching Reticulum@1.5.0
126 /// RNS/Transport.py:2279-2289.
127 pub const Table = struct {
128 phase: alloc_phase.capacity.Phase,
129 capacity: Capacity,
130 storage: Storage,
131 entries: []Entry,
132 len: usize = 0,
133
134 pub const storage_alignment: usize = 8;
135 pub const Storage = []align(storage_alignment) u8;
136 pub const Limits: type = TableLimits;
137 pub const Capacity: type = TableCapacity;
138 pub const InitError = Capacity.DeriveError || error{StorageLengthMismatch};
139 pub const Exhaustion = error{Full};
140 pub const work_limits: alloc_phase.capacity.WorkLimits = .{
141 .transition_steps_max = 65_536,
142 .cleanup_steps_per_call_max = 0,
143 .cleanup_calls_at_capacity_max = 0,
144 };
145 pub const claim: alloc_phase.capacity.Declaration = .{
146 .source = .{
147 .id = "reticulum.announces",
148 .kind = .phase_static,
149 .limit_source = .caller,
150 .storage = .{
151 .covered = &.{.{
152 .id = "caller_announce_table",
153 .lifetime = .transferred,
154 .detail = "caller storage for pending and held announce rebroadcasts",
155 }},
156 .excluded = &.{
157 "borrowed announce payload inputs",
158 "rebroadcast frames in effect storage",
159 },
160 },
161 .capacity = .{
162 .inputs = &.{alloc_phase.capacity.bindInput(
163 Limits,
164 "announces_max",
165 "announces_max",
166 )},
167 .type_selectors = &.{alloc_phase.capacity.bindType(Entry, "announce")},
168 .nodes = &.{
169 .{ .input = 0 },
170 .{ .scale = .{
171 .node = 0,
172 .coefficient = .{ .size_of_concrete_type = 0 },
173 } },
174 },
175 .assertions = &.{.{
176 .scope = .closure_total,
177 .measure = .retained,
178 .relation = .exact,
179 .expression = 1,
180 }},
181 },
182 .overload = .{
183 .kind = .reject_before_mutation,
184 .detail = "a new destination replaces the oldest rebroadcast entry, else rejects",
185 },
186 .risks = .{
187 .transitive = .{
188 .status = .excluded,
189 .detail = "announce operations call no allocating owner",
190 },
191 .foreign = .{
192 .status = .excluded,
193 .detail = "announce storage crosses no foreign boundary",
194 },
195 },
196 .work = .{ .equation = "operations scan at most announces_max entries" },
197 .obligations = &.{
198 .{ .key = "reticulum_announces_capacity", .role = .capacity_model },
199 .{ .key = "reticulum_announces_overload", .role = .overload },
200 .{ .key = "reticulum_announces_work", .role = .work_bound },
201 },
202 },
203 .bindings = .{
204 .owner = @This(),
205 .seal = .{
206 .family = alloc_phase.capacity.selector(@This().activate),
207 .premise = .{ .class = .checked_semantic_fact, .authority = .checker },
208 },
209 .teardown = .{
210 .family = alloc_phase.capacity.selector(@This().deinit),
211 .premise = .{ .class = .checked_semantic_fact, .authority = .checker },
212 },
213 },
214 };
215
216 pub fn init(storage: Storage, limits: Limits) InitError!Table {
217 const capacity = try Capacity.derive(limits);
218 if (storage.len != capacity.storage_bytes) return error.StorageLengthMismatch;
219 return .{
220 .phase = .initialization,
221 .capacity = capacity,
222 .storage = storage,
223 .entries = std.mem.bytesAsSlice(Entry, storage),
224 };
225 }
226
227 pub fn activate(self: *Table) void {
228 std.debug.assert(self.phase == .initialization);
229 std.debug.assert(self.len == 0);
230 self.phase = .steady;
231 }
232
233 pub fn find(self: *Table, hash: [16]u8) ?*Entry {
234 std.debug.assert(self.phase == .steady);
235 const index = self.indexOf(hash) orelse return null;
236 return &self.entries[index];
237 }
238
239 /// Answers whether `insert` would take a record for this destination.
240 pub fn admits(self: *const Table, hash: [16]u8) bool {
241 std.debug.assert(self.phase == .steady);
242 if (self.indexOf(hash) != null) return true;
243 if (self.len < self.capacity.announces_max) return true;
244 return self.oldestRebroadcast() != null;
245 }
246
247 /// Writes the entry for one destination, following Reticulum@1.5.0
248 /// RNS/Transport.py:2279-2289. A destination already in the table keeps its
249 /// slot and the record held behind it. With every place taken, the one
250 /// given up is the longest-standing entry the node has already sent at
251 /// least once. A full table with no such entry returns `error.Full` and
252 /// leaves every entry as it was.
253 pub fn insert(self: *Table, record: Record) Exhaustion!*Entry {
254 std.debug.assert(self.phase == .steady);
255 if (self.indexOf(record.destination)) |index| {
256 self.entries[index].record = record;
257 return &self.entries[index];
258 }
259 if (self.len < self.capacity.announces_max) {
260 self.entries[self.len] = .{ .record = record, .held = null };
261 self.len += 1;
262 return &self.entries[self.len - 1];
263 }
264 const index = self.oldestRebroadcast() orelse return error.Full;
265 self.entries[index] = .{ .record = record, .held = null };
266 return &self.entries[index];
267 }
268
269 /// Takes the entry at one position out, leaving the rest in the order they
270 /// were written.
271 pub fn removeAt(self: *Table, index: usize) void {
272 std.debug.assert(self.phase == .steady);
273 std.debug.assert(index < self.len);
274 std.mem.copyForwards(
275 Entry,
276 self.entries[index .. self.len - 1],
277 self.entries[index + 1 .. self.len],
278 );
279 self.len -= 1;
280 }
281
282 pub fn remove(self: *Table, hash: [16]u8) bool {
283 const index = self.indexOf(hash) orelse return false;
284 self.removeAt(index);
285 return true;
286 }
287
288 /// Returns the earliest due second among the pending records. An empty
289 /// table gives null.
290 pub fn earliestDue(self: *const Table) ?Seconds {
291 std.debug.assert(self.phase == .steady);
292 var earliest: ?Seconds = null;
293 for (self.entries[0..self.len]) |*entry| {
294 const due = entry.record.due;
295 if (earliest) |selected| {
296 if (due >= selected) continue;
297 }
298 earliest = due;
299 }
300 return earliest;
301 }
302
303 pub fn count(self: *const Table) usize {
304 std.debug.assert(self.phase == .steady);
305 return self.len;
306 }
307
308 pub fn deinit(self: *Table) Storage {
309 std.debug.assert(self.phase == .steady);
310 self.phase = .teardown;
311 const storage = self.storage;
312 self.* = undefined;
313 return storage;
314 }
315
316 fn indexOf(self: *const Table, hash: [16]u8) ?usize {
317 std.debug.assert(self.len <= self.capacity.announces_max);
318 for (self.entries[0..self.len], 0..) |*entry, index| {
319 if (std.mem.eql(u8, &entry.record.destination, &hash)) return index;
320 }
321 return null;
322 }
323
324 fn oldestRebroadcast(self: *const Table) ?usize {
325 var oldest: ?usize = null;
326 for (self.entries[0..self.len], 0..) |*entry, index| {
327 if (entry.record.retries == 0) continue;
328 if (oldest) |selected| {
329 if (entry.record.timestamp >= self.entries[selected].record.timestamp) continue;
330 }
331 oldest = index;
332 }
333 return oldest;
334 }
335 };
336
337 comptime {
338 alloc_phase.capacity.requireProvisionedRejectingOwnerShape(Table);
339 }
340
341 fn pending(byte: u8, timestamp: Seconds, retries: u8) Record {
342 return Record.init(.{
343 .destination = @splat(byte),
344 .due = timestamp + first_delay,
345 .timestamp = timestamp,
346 .retries = retries,
347 .hops = 1,
348 .block_rebroadcasts = false,
349 .attached = null,
350 .context_flag = 0,
351 .payload = "announce payload",
352 });
353 }
354
355 test "announces admit maximum and reject a new destination before any rebroadcast" {
356 comptime {
357 @stardustClaim(alloc_phase.capacity.witness(
358 Table,
359 "reticulum_announces_capacity",
360 ), null, null, null, null, null, null);
361 @stardustClaim(alloc_phase.capacity.witness(
362 Table,
363 "reticulum_announces_overload",
364 ), null, null, null, null, null, null);
365 @stardustClaim(alloc_phase.capacity.witness(
366 Table,
367 "reticulum_announces_work",
368 ), null, null, null, null, null, null);
369 }
370 const capacity = comptime TableCapacity.derive(.{ .announces_max = 3 }) catch unreachable;
371 var bytes: [capacity.storage_bytes]u8 align(Table.storage_alignment) = undefined;
372 var table = try Table.init(&bytes, .{ .announces_max = 3 });
373 table.activate();
374 defer _ = table.deinit();
375 for (1..4) |value| _ = try table.insert(pending(@intCast(value), 100, 0));
376 try std.testing.expectEqual(@as(usize, 3), table.count());
377 try std.testing.expect(!table.admits(@splat(4)));
378 try std.testing.expectError(error.Full, table.insert(pending(4, 101, 0)));
379 try std.testing.expectEqual(@as(usize, 3), table.count());
380 try std.testing.expect(table.admits(@splat(2)));
381 for (1..4) |value| try std.testing.expect(table.find(@splat(@intCast(value))) != null);
382 }
383
384 test "announces replace the oldest rebroadcast entry at maximum plus one" {
385 const capacity = comptime TableCapacity.derive(.{ .announces_max = 3 }) catch unreachable;
386 var bytes: [capacity.storage_bytes]u8 align(Table.storage_alignment) = undefined;
387 var table = try Table.init(&bytes, .{ .announces_max = 3 });
388 table.activate();
389 defer _ = table.deinit();
390 _ = try table.insert(pending(1, 10, 1));
391 _ = try table.insert(pending(2, 5, 0));
392 _ = try table.insert(pending(3, 7, 1));
393 try std.testing.expect(table.admits(@splat(4)));
394 _ = try table.insert(pending(4, 20, 0));
395 try std.testing.expectEqual(@as(usize, 3), table.count());
396 try std.testing.expect(table.find(@splat(3)) == null);
397 for ([_]u8{ 1, 2, 4 }) |byte| try std.testing.expect(table.find(@splat(byte)) != null);
398 try std.testing.expectEqual(@as(Seconds, 6), table.earliestDue().?);
399 }
400
401 test "Reticulum@1.5.0 RNS/Transport.py:791-794 releases a held record for the next sweep" {
402 const capacity = comptime TableCapacity.derive(.{ .announces_max = 1 }) catch unreachable;
403 var bytes: [capacity.storage_bytes]u8 align(Table.storage_alignment) = undefined;
404 var table = try Table.init(&bytes, .{ .announces_max = 1 });
405 table.activate();
406 defer _ = table.deinit();
407 const entry = try table.insert(pending(9, 10, 0));
408 var response = pending(9, 12, 1);
409 response.block_rebroadcasts = true;
410 entry.hold(response);
411 try std.testing.expect(entry.record.block_rebroadcasts);
412 _ = try table.insert(response);
413 try std.testing.expect(table.find(@splat(9)).?.held != null);
414 try std.testing.expect(entry.release(40));
415 try std.testing.expect(!entry.record.block_rebroadcasts);
416 try std.testing.expectEqual(@as(Seconds, 41), entry.record.due);
417 try std.testing.expect(entry.held == null);
418 try std.testing.expect(!entry.release(41));
419 try std.testing.expect(table.remove(@splat(9)));
420 try std.testing.expectEqual(@as(usize, 0), table.count());
421 }