swaits
a0a26a32f1
Upgrade the MeshCore bot to talk the DongLoRa Protocol v2 via donglora 1.0.0. * BotRuntime now holds a donglora.Dongle instead of a raw serial conn. All transmit / receive / ADVERT plumbing goes through the async Dongle surface. * Drop the direct pyserial imports; SerialException handling is now dongle-level (reconnect is handled by donglora.connect()). * Update constants and dispatch to match the Dongle accessors (info, set_config, transmit, start_rx) exposed by the 1.0 client.
654 lines
22 KiB
Python
654 lines
22 KiB
Python
"""RX dispatch: parse incoming frames, drive ACK/PATH/worker/retry feedback."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import random
|
|
import threading
|
|
import time
|
|
from collections.abc import Callable
|
|
|
|
import donglora as dl
|
|
from orac import events, followup, logfmt
|
|
from orac.ai import extract_trigger_query, rate_limit_message
|
|
from orac.constants import (
|
|
ACK_JITTER_MAX_S,
|
|
ACK_JITTER_MIN_S,
|
|
BOT_NAME,
|
|
CHANNEL_FOLLOWUP_WINDOW_S,
|
|
MIN_HASH_SIZE,
|
|
MULTI_ACK_COUNT,
|
|
MULTI_ACK_SPACING_S,
|
|
PATH_RETURN_DELAY_S,
|
|
RATE_LIMIT_DM,
|
|
RATE_LIMIT_GLOBAL,
|
|
RATE_LIMIT_PER_SENDER,
|
|
RATE_LIMIT_REPLY_COOLDOWN,
|
|
REPLY_CACHE_RESEND_WINDOW_S,
|
|
SCREENER_PER_CHANNEL_RATE_S,
|
|
)
|
|
from orac.crypto import my_hash, pubkey_bytes
|
|
from orac.meshcore import (
|
|
build_ack_packet,
|
|
build_grp_channel_packet,
|
|
build_login_response_packet,
|
|
build_multiack_packet,
|
|
build_path_return_packet,
|
|
compute_ack_hash,
|
|
get_route,
|
|
has_pending_dms,
|
|
is_dm_duplicate,
|
|
is_duplicate,
|
|
is_pending_expired,
|
|
learn_route,
|
|
lookup_node_by_hash,
|
|
node_name,
|
|
parse_header_and_path,
|
|
pop_pending_dms,
|
|
queue_pending_dm,
|
|
register_node,
|
|
route_name,
|
|
try_decode_advert,
|
|
try_decrypt_anon_req,
|
|
try_decrypt_dm,
|
|
try_decrypt_grp,
|
|
)
|
|
from orac.reply_state import PendingAckTable, ReplyCache
|
|
from orac.runtime import Metrics, TxItem, TxPriority, TxQueue
|
|
from orac.state import get_channel_history, record_channel_msg, record_heard_position
|
|
from orac.worker import ChannelReplyWork, ChannelScreenWork, DmReplyWork, Worker
|
|
|
|
log = logging.getLogger("orac")
|
|
|
|
|
|
class RateLimiter:
|
|
"""Bot-level rate limiter — per-sender + global timers, thread-safe."""
|
|
|
|
def __init__(self) -> None:
|
|
self._lock = threading.Lock()
|
|
self._per_key: dict[str, float] = {}
|
|
self._global_last = 0.0
|
|
self._reply_last: dict[str, float] = {}
|
|
|
|
def check(self, key: str) -> bool:
|
|
now = time.monotonic()
|
|
with self._lock:
|
|
if now - self._global_last < RATE_LIMIT_GLOBAL:
|
|
return False
|
|
limit = RATE_LIMIT_DM if key.startswith("dm:") else RATE_LIMIT_PER_SENDER
|
|
last = self._per_key.get(key)
|
|
return not (last is not None and now - last < limit)
|
|
|
|
def reply_ok(self, key: str) -> bool:
|
|
now = time.monotonic()
|
|
with self._lock:
|
|
last = self._reply_last.get(key)
|
|
if last is not None and now - last < RATE_LIMIT_REPLY_COOLDOWN:
|
|
return False
|
|
self._reply_last[key] = now
|
|
return True
|
|
|
|
def record(self, key: str) -> None:
|
|
now = time.monotonic()
|
|
with self._lock:
|
|
self._per_key[key] = now
|
|
self._global_last = now
|
|
|
|
|
|
class RxRouter:
|
|
"""Parses and dispatches every RxPacket the IOThread drains.
|
|
|
|
Designed to be cheap and non-blocking — never calls Claude, never sleeps.
|
|
Reply work goes to :class:`Worker`; ACK/PATH transmissions go to
|
|
:class:`TxQueue`; retry cancellation goes to :class:`PendingAckTable` and
|
|
:meth:`TxQueue.cancel_by_ack`.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
tx_queue: TxQueue,
|
|
worker: Worker,
|
|
pending_acks: PendingAckTable,
|
|
reply_cache: ReplyCache,
|
|
rate_limiter: RateLimiter,
|
|
metrics: Metrics,
|
|
) -> None:
|
|
self._tx = tx_queue
|
|
self._worker = worker
|
|
self._pending = pending_acks
|
|
self._cache = reply_cache
|
|
self._rl = rate_limiter
|
|
self._metrics = metrics
|
|
|
|
# -- top-level dispatch ---------------------------------------
|
|
|
|
def handle(self, pkt: dl.RxEvent) -> None:
|
|
payload_bytes = pkt.data
|
|
snr = pkt.snr_db
|
|
|
|
parsed = parse_header_and_path(payload_bytes)
|
|
if parsed is None:
|
|
return
|
|
(
|
|
payload_type,
|
|
route_type,
|
|
_payload_ver,
|
|
_pos,
|
|
raw_payload,
|
|
path_hops,
|
|
hash_size,
|
|
) = parsed
|
|
|
|
# Route-learn from the incoming path
|
|
if raw_payload and len(raw_payload) >= 2:
|
|
src_hash: int | None = None
|
|
if payload_type in (0x01, 0x02, 0x08):
|
|
src_hash = raw_payload[1]
|
|
elif payload_type == 0x04:
|
|
src_hash = raw_payload[0]
|
|
if src_hash is not None:
|
|
learn_route(src_hash, path_hops, hash_size, snr=snr)
|
|
|
|
# Packet-hash dedup (spec Section 16)
|
|
if is_duplicate(payload_type, raw_payload):
|
|
self._metrics.inc("rx_dup")
|
|
return
|
|
|
|
if payload_type == 0x03:
|
|
self._handle_ack(raw_payload)
|
|
elif payload_type == 0x0A:
|
|
self._handle_multipart(raw_payload)
|
|
elif payload_type == 0x04:
|
|
self._handle_advert(raw_payload)
|
|
elif payload_type == 0x05:
|
|
self._handle_grp_txt(raw_payload)
|
|
elif payload_type == 0x07:
|
|
self._handle_anon_req(raw_payload, path_hops, hash_size)
|
|
elif payload_type == 0x02:
|
|
self._handle_dm(raw_payload, route_type, path_hops, hash_size, snr)
|
|
|
|
# -- ACK ------------------------------------------------------
|
|
|
|
def _handle_ack(self, raw_payload: bytes) -> None:
|
|
"""Handle a plain PAYLOAD_TYPE_ACK (0x03): payload is 4-byte CRC."""
|
|
if len(raw_payload) < 4:
|
|
return
|
|
self._consume_ack(raw_payload[:4], source="ack")
|
|
|
|
def _handle_multipart(self, raw_payload: bytes) -> None:
|
|
"""Handle PAYLOAD_TYPE_MULTIPART (0x0A).
|
|
|
|
Upstream uses MULTIPART as a transport-level wrapper. For inner
|
|
type == PAYLOAD_TYPE_ACK (0x03), the layout is
|
|
``[remaining<<4 | 0x03][crc:4]`` (5 bytes). We unwrap it and treat
|
|
the inner CRC identically to a plain ACK — which matches how the
|
|
firmware's receiver handles it (src/Mesh.cpp:270-282).
|
|
"""
|
|
if len(raw_payload) < 5:
|
|
return
|
|
inner_type = raw_payload[0] & 0x0F
|
|
if inner_type != 0x03:
|
|
# MULTIPART wrappers for non-ACK inner types aren't something we
|
|
# currently generate or consume — ignore.
|
|
return
|
|
self._consume_ack(raw_payload[1:5], source="multipart")
|
|
|
|
def _consume_ack(self, ack_crc: bytes, source: str) -> None:
|
|
entry = self._pending.consume(ack_crc)
|
|
if entry is None:
|
|
return
|
|
now = time.monotonic()
|
|
elapsed_ms = int((now - entry.first_sent_at) * 1000)
|
|
self._tx.cancel_by_ack(ack_crc)
|
|
self._metrics.inc("ack_inbound_consumed")
|
|
self._metrics.inc(f"ack_inbound_source_{source}")
|
|
self._metrics.inc(f"reply_confirmed_attempt_{entry.attempt}")
|
|
events.emit(
|
|
"reply.confirmed",
|
|
peer=entry.peer_name,
|
|
peer_pk=entry.peer_pk.hex(),
|
|
attempt=entry.attempt,
|
|
elapsed_ms=elapsed_ms,
|
|
ack_crc=ack_crc.hex(),
|
|
source=source,
|
|
)
|
|
logfmt.ack_ok(entry.peer_name, entry.attempt, elapsed_ms)
|
|
|
|
# -- ADVERT ---------------------------------------------------
|
|
|
|
def _handle_advert(self, raw_payload: bytes) -> None:
|
|
result = try_decode_advert(raw_payload)
|
|
if result is None:
|
|
return
|
|
adv_pubkey, name, node_type, position = result
|
|
if adv_pubkey == pubkey_bytes():
|
|
return # our own advert echo
|
|
register_node(adv_pubkey, name)
|
|
if node_type == 2 and position is not None:
|
|
record_heard_position(position[0], position[1])
|
|
logfmt.net(
|
|
"ADVERT %s (hash=0x%02x pk=%s...)",
|
|
name,
|
|
adv_pubkey[0],
|
|
adv_pubkey.hex()[:16],
|
|
)
|
|
if has_pending_dms(adv_pubkey[0]):
|
|
self._process_pending_dms(adv_pubkey[0], name)
|
|
|
|
# -- GRP_TXT --------------------------------------------------
|
|
|
|
def _handle_grp_txt(self, raw_payload: bytes) -> None:
|
|
result = try_decrypt_grp(raw_payload)
|
|
if result is None:
|
|
return
|
|
channel_name, text = result
|
|
record_channel_msg(channel_name, text)
|
|
|
|
if ": " in text:
|
|
sender, _, body = text.partition(": ")
|
|
logfmt.ch_in(channel_name, sender, body)
|
|
else:
|
|
sender, body = "", text
|
|
logfmt.ch_in(channel_name, "?", body)
|
|
if sender == BOT_NAME:
|
|
return
|
|
|
|
rl_key = f"ch:{channel_name}:{sender}"
|
|
query = extract_trigger_query(body)
|
|
|
|
if query is not None:
|
|
# Explicit trigger or @mention — always process.
|
|
logfmt.net("trigger from %s on %s: %s", sender, channel_name, query)
|
|
self._dispatch_channel_reply(
|
|
channel_name=channel_name,
|
|
sender=sender,
|
|
query=query,
|
|
rl_key=rl_key,
|
|
origin="trigger",
|
|
)
|
|
return
|
|
|
|
# No trigger. Only consider this message as a potential follow-up
|
|
# if Orac has been active in this channel recently. The vast majority
|
|
# of channel chatter hits this early return with zero API cost.
|
|
if not followup.was_recent(channel_name, CHANNEL_FOLLOWUP_WINDOW_S):
|
|
return
|
|
|
|
# Rate-limit screener calls per channel to cap cost on busy channels.
|
|
if not followup.screener_ok(channel_name, SCREENER_PER_CHANNEL_RATE_S):
|
|
self._metrics.inc("screener_skipped_rate")
|
|
return
|
|
|
|
self._metrics.inc("screener_submitted")
|
|
self._worker.submit(
|
|
ChannelScreenWork(
|
|
channel_name=channel_name,
|
|
sender=sender,
|
|
text=body,
|
|
history_snapshot=get_channel_history(channel_name),
|
|
received_at=time.monotonic(),
|
|
rl_key=rl_key,
|
|
)
|
|
)
|
|
|
|
def _dispatch_channel_reply(
|
|
self,
|
|
*,
|
|
channel_name: str,
|
|
sender: str,
|
|
query: str,
|
|
rl_key: str,
|
|
origin: str,
|
|
) -> None:
|
|
"""Run rate-limit gate then enqueue a ChannelReplyWork (or a rate-limit TX)."""
|
|
if not self._rl.check(rl_key):
|
|
log.warning("rate limited: %s on %s (%s)", sender, channel_name, origin)
|
|
if self._rl.reply_ok(f"ch:{channel_name}"):
|
|
msg = rate_limit_message()
|
|
packet = build_grp_channel_packet(channel_name, BOT_NAME, msg)
|
|
if packet is not None:
|
|
self._tx.push(
|
|
TxItem(
|
|
priority=TxPriority.GRP_TXT,
|
|
not_before=time.monotonic(),
|
|
packet=packet,
|
|
label=f"GRP-rate({channel_name})",
|
|
)
|
|
)
|
|
self._rl.record(rl_key)
|
|
return
|
|
|
|
# Mark the channel warm so screener activates for subsequent messages.
|
|
followup.touch(channel_name)
|
|
self._worker.submit(
|
|
ChannelReplyWork(
|
|
channel_name=channel_name,
|
|
sender=sender,
|
|
query=query,
|
|
received_at=time.monotonic(),
|
|
rl_key=rl_key,
|
|
)
|
|
)
|
|
|
|
# -- ANON_REQ -------------------------------------------------
|
|
|
|
def _handle_anon_req(
|
|
self,
|
|
raw_payload: bytes,
|
|
path_hops: list[bytes],
|
|
hash_size: int,
|
|
) -> None:
|
|
result = try_decrypt_anon_req(raw_payload)
|
|
if result is None:
|
|
return
|
|
peer_pubkey, peer_name, _plaintext = result
|
|
learn_route(peer_pubkey[0], path_hops, hash_size)
|
|
logfmt.net("LOGIN from %s -- sending login OK", peer_name)
|
|
packet = build_login_response_packet(peer_pubkey)
|
|
self._tx.push(
|
|
TxItem(
|
|
priority=TxPriority.LOGIN_RESP,
|
|
not_before=time.monotonic(),
|
|
packet=packet,
|
|
label=f"LOGIN_RESP->{peer_name}",
|
|
)
|
|
)
|
|
if has_pending_dms(peer_pubkey[0]):
|
|
self._process_pending_dms(peer_pubkey[0], peer_name)
|
|
|
|
# -- DM -------------------------------------------------------
|
|
|
|
def _handle_dm(
|
|
self,
|
|
raw_payload: bytes,
|
|
route_type: int,
|
|
path_hops: list[bytes],
|
|
hash_size: int,
|
|
snr: float,
|
|
) -> None:
|
|
result = try_decrypt_dm(raw_payload)
|
|
if result is None:
|
|
# Addressed to us but sender unknown: queue for later
|
|
if len(raw_payload) >= 4 and raw_payload[0] == my_hash():
|
|
src_hash = raw_payload[1]
|
|
if not lookup_node_by_hash(src_hash):
|
|
is_first = queue_pending_dm(src_hash, raw_payload)
|
|
if is_first:
|
|
logfmt.net(
|
|
"queued DM from unknown 0x%02x, waiting for their ADVERT",
|
|
src_hash,
|
|
)
|
|
return
|
|
|
|
peer_pubkey, peer_name, dm_text, raw_plaintext = result
|
|
pk_hex = peer_pubkey.hex()
|
|
now = time.monotonic()
|
|
|
|
# 1) Schedule ACK with jitter — always, regardless of hash_size or dedup.
|
|
# Receiving and ACKing is the highest-priority guarantee.
|
|
self._schedule_ack(raw_plaintext, peer_pubkey, now, route_type=route_type)
|
|
|
|
# 2) Schedule PATH return — only if inbound used >=2-byte hashes.
|
|
# We refuse to echo 1-byte paths; if the inbound was 1-byte we skip
|
|
# the optional path hint. The peer still gets ACKed above and will
|
|
# still get our reply below (via flood, since learn_route also
|
|
# rejects 1-byte — see MIN_HASH_SIZE policy).
|
|
if hash_size >= MIN_HASH_SIZE:
|
|
self._schedule_path_return(peer_pubkey, path_hops, hash_size, now)
|
|
|
|
events.emit(
|
|
"dm.received",
|
|
peer=peer_name,
|
|
peer_pk=pk_hex,
|
|
route=route_name(route_type),
|
|
path_hops=len(path_hops),
|
|
snr=snr,
|
|
text_len=len(dm_text),
|
|
)
|
|
|
|
# 3) Text-level dedup
|
|
if is_dm_duplicate(pk_hex, dm_text):
|
|
self._metrics.inc("dm_text_duplicate")
|
|
cached = self._cache.get(pk_hex, dm_text)
|
|
# Only treat the duplicate as loss-recovery (resend cached reply)
|
|
# if it's close in time to the original reply; older retries are
|
|
# assumed to be the peer legitimately re-asking.
|
|
if cached is None or (now - cached.ts) > REPLY_CACHE_RESEND_WINDOW_S:
|
|
if cached is not None:
|
|
self._metrics.inc("cached_resend_skipped_stale")
|
|
return
|
|
# Don't pile on if the retry scheduler is still working the original.
|
|
if self._pending.has_pending_for_peer(peer_pubkey):
|
|
self._metrics.inc("cached_resend_skipped_pending")
|
|
return
|
|
jitter = random.uniform(
|
|
ACK_JITTER_MIN_S + PATH_RETURN_DELAY_S, PATH_RETURN_DELAY_S + 0.3
|
|
)
|
|
self._tx.push(
|
|
TxItem(
|
|
priority=TxPriority.REPLY_DIRECT,
|
|
not_before=now + jitter,
|
|
packet=cached.reply_packet,
|
|
label=f"DM(cached)->{peer_name}",
|
|
expected_ack=cached.expected_ack,
|
|
peer_pk=peer_pubkey,
|
|
attempt=1,
|
|
cached_reply=True,
|
|
)
|
|
)
|
|
self._metrics.inc("reply_cached_resent")
|
|
events.emit(
|
|
"reply.cached_resend",
|
|
peer=peer_name,
|
|
peer_pk=pk_hex,
|
|
ack_crc=cached.expected_ack.hex(),
|
|
age_ms=int((now - cached.ts) * 1000),
|
|
)
|
|
logfmt.net(
|
|
"duplicate DM from %s within %.0fs; re-sent cached reply",
|
|
peer_name,
|
|
now - cached.ts,
|
|
)
|
|
return
|
|
|
|
logfmt.dm_in(peer_name, dm_text, route=route_name(route_type))
|
|
|
|
# 4) Rate limit check — drops silently (ACK already queued)
|
|
rl_key = f"dm:{pk_hex}"
|
|
if not self._rl.check(rl_key):
|
|
log.warning("rate limited DM: %s", peer_name)
|
|
self._metrics.inc("dm_rate_limited")
|
|
events.emit(
|
|
"dm.rate_limited",
|
|
peer=peer_name,
|
|
peer_pk=pk_hex,
|
|
)
|
|
return
|
|
|
|
# 5) Post reply work to worker (non-blocking)
|
|
self._worker.submit(
|
|
DmReplyWork(
|
|
peer_pk=peer_pubkey,
|
|
peer_name=peer_name,
|
|
dm_text=dm_text,
|
|
received_at=now,
|
|
)
|
|
)
|
|
|
|
# -- pending-DM drain (run on IO thread, cheap decrypts only) ---
|
|
|
|
def _process_pending_dms(self, node_hash: int, peer_label: str) -> None:
|
|
pending = pop_pending_dms(node_hash)
|
|
if not pending:
|
|
return
|
|
logfmt.net("processing %d pending DM(s) from %s", len(pending), peer_label)
|
|
for pending_payload, ts in pending:
|
|
if is_pending_expired(ts):
|
|
continue
|
|
result = try_decrypt_dm(pending_payload)
|
|
if result is None:
|
|
continue
|
|
peer_pubkey, peer_name, dm_text, raw_plaintext = result
|
|
pk_hex = peer_pubkey.hex()
|
|
now = time.monotonic()
|
|
|
|
# ACK always
|
|
self._schedule_ack(raw_plaintext, peer_pubkey, now, route_type=1)
|
|
|
|
if is_dm_duplicate(pk_hex, dm_text):
|
|
continue
|
|
|
|
logfmt.dm_in(peer_name, dm_text, route="pending")
|
|
|
|
rl_key = f"dm:{pk_hex}"
|
|
if not self._rl.check(rl_key):
|
|
log.warning("rate limited pending DM: %s", peer_name)
|
|
continue
|
|
|
|
self._worker.submit(
|
|
DmReplyWork(
|
|
peer_pk=peer_pubkey,
|
|
peer_name=peer_name,
|
|
dm_text=dm_text,
|
|
received_at=now,
|
|
)
|
|
)
|
|
|
|
# -- helpers --------------------------------------------------
|
|
|
|
def _schedule_ack(
|
|
self,
|
|
plaintext: bytes,
|
|
peer_pubkey: bytes,
|
|
now: float,
|
|
route_type: int,
|
|
) -> None:
|
|
"""Schedule the ACK(s) for a received DM.
|
|
|
|
Protocol (upstream Mesh::routeDirectRecvAcks + NodePrefs.multi_acks):
|
|
|
|
* When :data:`MULTI_ACK_COUNT` > 0 AND we have a learned direct route
|
|
to the peer: emit N MULTIPART ACKs with ``remaining`` counting down
|
|
to 1, spaced ~300 ms apart, then the final plain ACK 300 ms later.
|
|
* Otherwise: a single plain ACK with small jitter.
|
|
|
|
MULTIPART ACKs on flood paths are NOT protocol-correct and would just
|
|
burn airtime (receivers key MULTIPART handling to direct-route recv).
|
|
"""
|
|
ack_crc = compute_ack_hash(plaintext, peer_pubkey)
|
|
target_name = node_name(peer_pubkey.hex())
|
|
jitter = random.uniform(ACK_JITTER_MIN_S, ACK_JITTER_MAX_S)
|
|
|
|
use_multipart = MULTI_ACK_COUNT > 0 and get_route(peer_pubkey[0]) is not None
|
|
|
|
if not use_multipart:
|
|
packet = build_ack_packet(ack_crc, peer_pubkey[0])
|
|
self._tx.push(
|
|
TxItem(
|
|
priority=TxPriority.ACK,
|
|
not_before=now + jitter,
|
|
packet=packet,
|
|
label=f"ACK->0x{peer_pubkey[0]:02x}",
|
|
)
|
|
)
|
|
self._metrics.inc("ack_sent")
|
|
events.emit(
|
|
"ack.sent",
|
|
target_hash=peer_pubkey[0],
|
|
target=target_name,
|
|
ack_crc=ack_crc.hex(),
|
|
route=route_name(route_type),
|
|
jitter_ms=int(jitter * 1000),
|
|
multipart=False,
|
|
)
|
|
return
|
|
|
|
# MULTIPART chain: N MULTIPART packets (remaining N, N-1, ..., 1),
|
|
# then final plain ACK, each 300 ms + jitter after the previous.
|
|
cumulative = jitter
|
|
for i in range(MULTI_ACK_COUNT):
|
|
remaining = MULTI_ACK_COUNT - i
|
|
multi_packet = build_multiack_packet(ack_crc, remaining, peer_pubkey[0])
|
|
self._tx.push(
|
|
TxItem(
|
|
priority=TxPriority.ACK,
|
|
not_before=now + cumulative,
|
|
packet=multi_packet,
|
|
label=f"MULTIACK->0x{peer_pubkey[0]:02x}(r={remaining})",
|
|
)
|
|
)
|
|
self._metrics.inc("multiack_sent")
|
|
cumulative += MULTI_ACK_SPACING_S
|
|
# Final plain ACK
|
|
final_packet = build_ack_packet(ack_crc, peer_pubkey[0])
|
|
self._tx.push(
|
|
TxItem(
|
|
priority=TxPriority.ACK,
|
|
not_before=now + cumulative,
|
|
packet=final_packet,
|
|
label=f"ACK->0x{peer_pubkey[0]:02x}",
|
|
)
|
|
)
|
|
self._metrics.inc("ack_sent")
|
|
events.emit(
|
|
"ack.sent",
|
|
target_hash=peer_pubkey[0],
|
|
target=target_name,
|
|
ack_crc=ack_crc.hex(),
|
|
route=route_name(route_type),
|
|
jitter_ms=int(jitter * 1000),
|
|
multipart=True,
|
|
multipart_count=MULTI_ACK_COUNT,
|
|
)
|
|
|
|
def _schedule_path_return(
|
|
self,
|
|
peer_pubkey: bytes,
|
|
path_hops: list[bytes],
|
|
hash_size: int,
|
|
now: float,
|
|
) -> None:
|
|
packet = build_path_return_packet(peer_pubkey, path_hops, hash_size)
|
|
self._tx.push(
|
|
TxItem(
|
|
priority=TxPriority.PATH,
|
|
not_before=now + PATH_RETURN_DELAY_S,
|
|
packet=packet,
|
|
label=f"PATH->{node_name(peer_pubkey.hex())}",
|
|
)
|
|
)
|
|
self._metrics.inc("path_sent")
|
|
|
|
|
|
# ── Advert scheduling callback ───────────────────────────────────
|
|
|
|
|
|
def make_advert_pusher(
|
|
build_advert_fn: Callable[[], bytes],
|
|
metrics: Metrics,
|
|
) -> Callable[[TxQueue], None]:
|
|
"""Return a fn that enqueues an ADVERT onto the TxQueue."""
|
|
|
|
def _advert_push(tx: TxQueue) -> None:
|
|
packet = build_advert_fn()
|
|
tx.push(
|
|
TxItem(
|
|
priority=TxPriority.ADVERT,
|
|
not_before=time.monotonic(),
|
|
packet=packet,
|
|
label=f"ADVERT({BOT_NAME})",
|
|
)
|
|
)
|
|
metrics.inc("advert_sent")
|
|
events.emit("advert.sent", bot=BOT_NAME)
|
|
logfmt.net("queued ADVERT as %s (hash=0x%02x)", BOT_NAME, my_hash())
|
|
|
|
return _advert_push
|
|
|
|
|
|
__all__ = [
|
|
"RateLimiter",
|
|
"RxRouter",
|
|
"make_advert_pusher",
|
|
]
|