lib/reticulum/src/node/transport/discoveries.zig
daab053ee43316e1809a84551d573ddd1e5bf3d2
1 const std = @import("std");
2 const alloc_phase = @import("alloc_phase");
3 const carrier = @import("../../carrier/root.zig");
4
5 pub const Seconds = u64;
6
7 /// Fifteen seconds, the wait Reticulum@1.5.0 RNS/Transport.py:131,1786,3482
8 /// gives a discovered path, so a caller learns how long a waiting question
9 /// stays worth answering.
10 pub const timeout: Seconds = 15;
11
12 /// The set of carriers that wait for one discovered path. The set covers every
13 /// carrier index the node can name.
14 pub const Requesters = std.bit_set.ArrayBitSet(u64, 256);
15
16 /// One destination that requesting carriers wait on.
17 pub const Entry = struct {
18 destination: [16]u8,
19 expires: Seconds,
20 requesters: Requesters,
21 engaged: bool,
22 };
23
24 /// Reports whether an entry still counts. This port keeps an entry through the
25 /// second it was created plus fifteen and expires it at plus sixteen.
26 fn live(entry: *const Entry, now: Seconds) bool {
27 return now <= entry.expires;
28 }
29
30 comptime {
31 std.debug.assert(Requesters.bit_length == @as(usize, std.math.maxInt(carrier.Index)) + 1);
32 }
33
34 const TableLimits = struct {
35 discoveries_max: usize,
36 };
37
38 const TableCapacity = struct {
39 discoveries_max: usize,
40 storage_bytes: usize,
41
42 pub const DeriveError = error{ InvalidLimit, CapacityOverflow };
43
44 pub fn derive(limits: TableLimits) DeriveError!TableCapacity {
45 if (limits.discoveries_max == 0) return error.InvalidLimit;
46 const storage_bytes = alloc_phase.capacity.mul(
47 usize,
48 limits.discoveries_max,
49 @sizeOf(Entry),
50 ) catch return error.CapacityOverflow;
51 return .{ .discoveries_max = limits.discoveries_max, .storage_bytes = storage_bytes };
52 }
53 };
54
55 /// Waiting discovery requests, matching Reticulum@1.5.0
56 /// RNS/Transport.py:1777-1792,3478-3497.
57 pub const Table = struct {
58 phase: alloc_phase.capacity.Phase,
59 capacity: Capacity,
60 storage: Storage,
61 entries: []Entry,
62 len: usize = 0,
63
64 pub const storage_alignment: usize = 8;
65 pub const Storage = []align(storage_alignment) u8;
66 pub const Limits: type = TableLimits;
67 pub const Capacity: type = TableCapacity;
68 pub const InitError = Capacity.DeriveError || error{StorageLengthMismatch};
69 pub const Exhaustion = error{Full};
70 pub const work_limits: alloc_phase.capacity.WorkLimits = .{
71 .transition_steps_max = 65_536,
72 .cleanup_steps_per_call_max = 0,
73 .cleanup_calls_at_capacity_max = 0,
74 };
75 pub const claim: alloc_phase.capacity.Declaration = .{
76 .source = .{
77 .id = "reticulum.discoveries",
78 .kind = .phase_static,
79 .limit_source = .caller,
80 .storage = .{
81 .covered = &.{.{
82 .id = "caller_discovery_table",
83 .lifetime = .transferred,
84 .detail = "caller storage for waiting discovery requests and requesters",
85 }},
86 .excluded = &.{"forwarded request and answer frames in effect storage"},
87 },
88 .capacity = .{
89 .inputs = &.{alloc_phase.capacity.bindInput(
90 Limits,
91 "discoveries_max",
92 "discoveries_max",
93 )},
94 .type_selectors = &.{alloc_phase.capacity.bindType(Entry, "discovery")},
95 .nodes = &.{
96 .{ .input = 0 },
97 .{ .scale = .{
98 .node = 0,
99 .coefficient = .{ .size_of_concrete_type = 0 },
100 } },
101 },
102 .assertions = &.{.{
103 .scope = .closure_total,
104 .measure = .retained,
105 .relation = .exact,
106 .expression = 1,
107 }},
108 },
109 .overload = .{
110 .kind = .reject_before_mutation,
111 .detail = "a new destination replaces an expired entry, else rejects",
112 },
113 .risks = .{
114 .transitive = .{
115 .status = .excluded,
116 .detail = "discovery operations call no allocating owner",
117 },
118 .foreign = .{
119 .status = .excluded,
120 .detail = "discovery storage crosses no foreign boundary",
121 },
122 },
123 .work = .{ .equation = "operations scan at most discoveries_max entries" },
124 .obligations = &.{
125 .{ .key = "reticulum_discoveries_capacity", .role = .capacity_model },
126 .{ .key = "reticulum_discoveries_overload", .role = .overload },
127 .{ .key = "reticulum_discoveries_work", .role = .work_bound },
128 },
129 },
130 .bindings = .{
131 .owner = @This(),
132 .seal = .{
133 .family = alloc_phase.capacity.selector(@This().activate),
134 .premise = .{ .class = .checked_semantic_fact, .authority = .checker },
135 },
136 .teardown = .{
137 .family = alloc_phase.capacity.selector(@This().deinit),
138 .premise = .{ .class = .checked_semantic_fact, .authority = .checker },
139 },
140 },
141 };
142
143 pub fn init(storage: Storage, limits: Limits) InitError!Table {
144 const capacity = try Capacity.derive(limits);
145 if (storage.len != capacity.storage_bytes) return error.StorageLengthMismatch;
146 return .{
147 .phase = .initialization,
148 .capacity = capacity,
149 .storage = storage,
150 .entries = std.mem.bytesAsSlice(Entry, storage),
151 };
152 }
153
154 pub fn activate(self: *Table) void {
155 std.debug.assert(self.phase == .initialization);
156 std.debug.assert(self.len == 0);
157 self.phase = .steady;
158 }
159
160 /// Returns the live entry for one destination, so a caller checks whether a
161 /// question for this destination is already outstanding before it sends
162 /// another.
163 pub fn find(self: *Table, destination: [16]u8, now: Seconds) ?*Entry {
164 std.debug.assert(self.phase == .steady);
165 const index = self.indexOf(destination) orelse return null;
166 const entry = &self.entries[index];
167 if (!live(entry, now)) return null;
168 return entry;
169 }
170
171 /// Answers whether a request about this destination would find room, so a
172 /// caller asks before it reserves effects because the insert that follows
173 /// can refuse.
174 pub fn admits(self: *const Table, destination: [16]u8, now: Seconds) bool {
175 std.debug.assert(self.phase == .steady);
176 if (self.indexOf(destination) != null) return true;
177 if (self.len < self.capacity.discoveries_max) return true;
178 return self.expiredIndex(now) != null;
179 }
180
181 /// Adds a requesting carrier to the entry for one destination, following
182 /// Reticulum@1.5.0 RNS/Transport.py:1779-1792, so the answer reaches all of
183 /// them together. A new entry starts unengaged, which marks a question this
184 /// node has yet to pass on. A full table with nothing expired to replace
185 /// returns `error.Full`.
186 pub fn batch(
187 self: *Table,
188 destination: [16]u8,
189 requester: carrier.Index,
190 now: Seconds,
191 ) Exhaustion!void {
192 std.debug.assert(self.phase == .steady);
193 if (self.find(destination, now)) |entry| {
194 entry.requesters.set(requester);
195 return;
196 }
197 const target = try self.slot(destination, now);
198 target.* = .{
199 .destination = destination,
200 .expires = now +| timeout,
201 .requesters = Requesters.empty,
202 .engaged = false,
203 };
204 target.requesters.set(requester);
205 }
206
207 /// Marks the entry for one destination engaged and keeps the carriers
208 /// already waiting, following Reticulum@1.5.0 RNS/Transport.py:3478-3497.
209 /// The entry's wait restarts at the second of the call.
210 pub fn engage(
211 self: *Table,
212 destination: [16]u8,
213 requester: carrier.Index,
214 now: Seconds,
215 ) Exhaustion!void {
216 std.debug.assert(self.phase == .steady);
217 var requesters = Requesters.empty;
218 if (self.find(destination, now)) |entry| requesters = entry.requesters;
219 requesters.set(requester);
220 const target = try self.slot(destination, now);
221 target.* = .{
222 .destination = destination,
223 .expires = now +| timeout,
224 .requesters = requesters,
225 .engaged = true,
226 };
227 }
228
229 /// Removes the entry that an announce answers and returns it, following
230 /// Reticulum@1.5.0 RNS/Transport.py:2350-2352, so a caller sends the answer
231 /// to each carrier the entry holds. An entry past its wait is removed and
232 /// reported as null.
233 pub fn take(self: *Table, destination: [16]u8, now: Seconds) ?Entry {
234 std.debug.assert(self.phase == .steady);
235 const index = self.indexOf(destination) orelse return null;
236 const entry = self.entries[index];
237 self.len -= 1;
238 self.entries[index] = self.entries[self.len];
239 if (!live(&entry, now)) return null;
240 return entry;
241 }
242
243 pub fn count(self: *const Table) usize {
244 std.debug.assert(self.phase == .steady);
245 return self.len;
246 }
247
248 pub fn deinit(self: *Table) Storage {
249 std.debug.assert(self.phase == .steady);
250 self.phase = .teardown;
251 const storage = self.storage;
252 self.* = undefined;
253 return storage;
254 }
255
256 fn indexOf(self: *const Table, destination: [16]u8) ?usize {
257 std.debug.assert(self.len <= self.capacity.discoveries_max);
258 for (self.entries[0..self.len], 0..) |*entry, index| {
259 if (std.mem.eql(u8, &entry.destination, &destination)) return index;
260 }
261 return null;
262 }
263
264 fn expiredIndex(self: *const Table, now: Seconds) ?usize {
265 for (self.entries[0..self.len], 0..) |*entry, index| {
266 if (!live(entry, now)) return index;
267 }
268 return null;
269 }
270
271 fn slot(self: *Table, destination: [16]u8, now: Seconds) Exhaustion!*Entry {
272 if (self.indexOf(destination)) |index| return &self.entries[index];
273 if (self.len < self.capacity.discoveries_max) {
274 self.len += 1;
275 return &self.entries[self.len - 1];
276 }
277 const index = self.expiredIndex(now) orelse return error.Full;
278 return &self.entries[index];
279 }
280 };
281
282 comptime {
283 alloc_phase.capacity.requireProvisionedRejectingOwnerShape(Table);
284 }
285
286 test "discoveries admit maximum and reject a new destination at maximum plus one" {
287 comptime {
288 @stardustClaim(alloc_phase.capacity.witness(
289 Table,
290 "reticulum_discoveries_capacity",
291 ), null, null, null, null, null, null);
292 @stardustClaim(alloc_phase.capacity.witness(
293 Table,
294 "reticulum_discoveries_overload",
295 ), null, null, null, null, null, null);
296 @stardustClaim(alloc_phase.capacity.witness(
297 Table,
298 "reticulum_discoveries_work",
299 ), null, null, null, null, null, null);
300 }
301 const limits = TableLimits{ .discoveries_max = 2 };
302 const capacity = comptime TableCapacity.derive(limits) catch unreachable;
303 var bytes: [capacity.storage_bytes]u8 align(Table.storage_alignment) = undefined;
304 var table = try Table.init(&bytes, limits);
305 table.activate();
306 defer _ = table.deinit();
307 try table.batch(@splat(1), 0, 100);
308 try table.batch(@splat(2), 1, 100);
309 try std.testing.expect(!table.admits(@splat(3), 100));
310 try std.testing.expectError(error.Full, table.batch(@splat(3), 0, 100));
311 try std.testing.expectEqual(@as(usize, 2), table.count());
312 try table.batch(@splat(1), 2, 101);
313 const entry = table.find(@splat(1), 101) orelse return error.TestUnexpectedResult;
314 try std.testing.expect(entry.requesters.isSet(0) and entry.requesters.isSet(2));
315 try std.testing.expect(!entry.engaged);
316 }
317
318 test "Reticulum@1.5.0 RNS/Transport.py:975-984 keeps a discovery through plus 15, not plus 16" {
319 const limits = TableLimits{ .discoveries_max = 1 };
320 const capacity = comptime TableCapacity.derive(limits) catch unreachable;
321 var bytes: [capacity.storage_bytes]u8 align(Table.storage_alignment) = undefined;
322 var table = try Table.init(&bytes, limits);
323 table.activate();
324 defer _ = table.deinit();
325 try table.batch(@splat(7), 0, 1_000);
326 try std.testing.expect(table.find(@splat(7), 1_000 + timeout) != null);
327 try std.testing.expect(table.find(@splat(7), 1_000 + timeout + 1) == null);
328 try std.testing.expect(table.admits(@splat(8), 1_000 + timeout + 1));
329 try table.batch(@splat(8), 1, 1_000 + timeout + 1);
330 try std.testing.expectEqual(@as(usize, 1), table.count());
331 try std.testing.expect(table.take(@splat(7), 1_000 + timeout + 1) == null);
332 }
333
334 test "discoveries engage over batched requesters and cover every carrier" {
335 const limits = TableLimits{ .discoveries_max = 1 };
336 const capacity = comptime TableCapacity.derive(limits) catch unreachable;
337 var bytes: [capacity.storage_bytes]u8 align(Table.storage_alignment) = undefined;
338 var table = try Table.init(&bytes, limits);
339 table.activate();
340 defer _ = table.deinit();
341 try table.batch(@splat(5), 0, 10);
342 try table.engage(@splat(5), 255, 12);
343 const engaged = table.find(@splat(5), 12) orelse return error.TestUnexpectedResult;
344 try std.testing.expect(engaged.engaged);
345 try std.testing.expectEqual(@as(Seconds, 12 + timeout), engaged.expires);
346 try std.testing.expect(engaged.requesters.isSet(0) and engaged.requesters.isSet(255));
347 for (0..256) |index| try table.batch(@splat(5), @intCast(index), 20);
348 const waiting = table.take(@splat(5), 20) orelse return error.TestUnexpectedResult;
349 try std.testing.expectEqual(@as(usize, 256), waiting.requesters.count());
350 try std.testing.expectEqual(@as(usize, 0), table.count());
351 }