tiny.reticulum.node.transport.retransmit

1const std = @import("std");2const reticulum = @import("../../root.zig");3const announces = @import("announces.zig");45const destination = reticulum.destination;6const node = reticulum.node;7const packet = reticulum.packet;8const wire = reticulum.wire;910/// Describes what becomes of an accepted announce at the rebroadcast queue, so11/// the receive path knows whether accepting an announce also queues one to12/// send.13pub const Plan = enum {14    none,15    write,16    too_large,17};1819const Visit = enum {20    kept,21    completed,22    stopped,23};2425/// Decides whether a transport node rebroadcasts, following Reticulum@1.5.026/// RNS/Transport.py:2267-2289, so a caller can reserve effects for the27/// rebroadcast in the same step before it accepts an announce. An announce the28/// path table refused leaves the queue untouched, as does one that reached a29/// node carrying no traffic for others. An announce marked a path response30/// 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. A32/// HEADER_1 payload above 465 bytes has no room left under a HEADER_233/// rebroadcast, so the step reports it and queues nothing.34pub 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}4344/// Answers whether a slot stands free for the timer that drives the rebroadcast45/// queue, so a caller checks before changing any table because a sweep with46/// nowhere to put its timer has to fail the whole step.47pub fn canSchedule(node_owner: *const node.Node) bool {48    return node_owner.timers.canScheduleAfterCancel(.announces, null);49}5051/// Completes entries that neighbors rebroadcast, following Reticulum@1.5.052/// RNS/Transport.py:2093-2116, so the node learns its own copy is no longer53/// needed when a neighbor carries the same announce. An announce heard at this54/// entry's own hop count counts as a neighbor rebroadcast, and two of them55/// drop the entry once the node has sent it at least once. An announce heard56/// one hop further on means a neighbor already passed this entry along, so the57/// step drops the entry when the node has sent it and its next send is still58/// ahead.59pub 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}7980/// Queues the announce with its turn set one second ahead, as Reticulum@1.5.081/// RNS/Transport.py:2248-2289 does, so the step turns the decision into a82/// queued rebroadcast and arms the sweep that will send it. A payload too large83/// to relay reports `announce_relay_too_large`. A full announce table reports84/// `announce_table_full`.85pub 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}115116/// Moves the timer only when the second given falls sooner than the one already117/// set, because entries share one timer.118pub 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}128129/// Sends due rebroadcasts by ascending hops when the timer fires, following130/// Reticulum@1.5.0 RNS/Transport.py:737-799,1239-1240. A pass that would131/// overrun the room for effects breaks off and sets the timer at the second it132/// broke off on. An empty table cancels the timer and returns. A destination133/// whose identity the node no longer recalls is dropped from the queue.134pub 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}163164/// 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 stops166/// circulating.167fn 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}171172fn 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 or180        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}193194/// Assembles the outgoing packet under a HEADER_2 header from the payload the195/// queue kept, as Reticulum@1.5.0 RNS/Transport.py:756-779 does, carrying this196/// node's own transport identity so the next node learns who passed it on. A197/// record that blocks further rebroadcasts is marked a path response.198fn 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}

1pub const retransmit = @import("retransmit.zig");