import type { Connection } from "@libp2p/interface"; import { type Multiaddr, multiaddr } from "@multiformats/multiaddr"; import { getPublicKeyFromPeerId, ready, sha256Sync, toBase64, } from "@peerbit/crypto"; import { DirectStream, type DirectStreamComponents, type DirectStreamOptions, type PeerStreams, type RustFanoutTree, dontThrowIfDeliveryError, } from "@peerbit/stream"; import { AnyWhere, DataMessage, type FanoutRouteTokenHint, type StreamEvents, } from "@peerbit/stream-interface"; import { AbortError, TimeoutError, delay } from "@peerbit/time"; import { anySignal } from "any-signal"; import { Uint8ArrayList } from "uint8arraylist"; import { normalizeUint8Array, toUint8Array } from "./bytes.js"; import { JOIN_REJECT_LOW_BID, JOIN_REJECT_NO_CAPACITY, JOIN_REJECT_NOT_ATTACHED, JOIN_REJECT_REDIRECT_ADDR_MAX, JOIN_REJECT_REDIRECT_MAX, MSG_DATA, MSG_END, MSG_FETCH_REQ, MSG_IHAVE, MSG_JOIN_ACCEPT, MSG_JOIN_REJECT, MSG_JOIN_REQ, MSG_KICK, MSG_LEAVE, MSG_PARENT_PROBE_REPLY, MSG_PARENT_PROBE_REQ, MSG_PROVIDER_ANNOUNCE, MSG_PROVIDER_NOTIFY, MSG_PROVIDER_QUERY, MSG_PROVIDER_REPLY, MSG_PROVIDER_SUBSCRIBE, MSG_PROVIDER_UNSUBSCRIBE, MSG_PUBLISH_PROXY, MSG_REPAIR_REQ, MSG_ROUTE_QUERY, MSG_ROUTE_REPLY, MSG_TRACKER_ANNOUNCE, MSG_TRACKER_FEEDBACK, MSG_TRACKER_QUERY, MSG_TRACKER_REPLY, MSG_UNICAST, MSG_UNICAST_ACK, PARENT_PROBE_FLAG_ACCEPTING, PARENT_PROBE_FLAG_OVERLOADED, PARENT_PROBE_FLAG_REPAIRING, PARENT_PROBE_FLAG_ROOTED, TRACKER_FEEDBACK_DIAL_FAILED, TRACKER_FEEDBACK_JOINED, TRACKER_FEEDBACK_JOIN_REJECT, TRACKER_FEEDBACK_JOIN_TIMEOUT, UNICAST_ACK_DEFAULT_TIMEOUT_MS, type FanoutWireCodec, type JoinRejectRedirect, type ParentProbeReply, type ProviderCandidate, type ProviderEntry, type TrackerEntry, clampU16, readU32BE, tsFanoutWireCodec, writeU32BE, } from "./fanout-tree-codec.js"; import { evaluateParentUpgradeGate, normalizeParentUpgradePolicy, recordParentUpgradeSkip, type ParentUpgradePolicy, type ParentUpgradeSkipReason, } from "./fanout-tree-parent-upgrade.js"; import { TopicRootControlPlane } from "./topic-root-control-plane.js"; export type FanoutTreeChannelId = { root: string; topic: string; key: Uint8Array; // 32 bytes suffixKey: string; // base64 of key[0..24) }; export type FanoutTreeChannelRole = "root" | "node"; export type FanoutTreeStreamOptions = DirectStreamOptions & { /** * Optional RNG hook used for shuffling/join request ids. * * This is primarily intended for deterministic simulation harnesses. */ random?: () => number; /** * Optional topic-root resolver for consumers (e.g. shared-log) that need to * resolve a root when joining a channel without an explicit root. * * FanoutTree does not currently use this internally, but we expose it as a * shared control-plane hook so applications do not need to hang root * resolution off of a separate pubsub implementation. */ topicRootControlPlane?: TopicRootControlPlane; }; export type FanoutTreeChannelOptions = { role: FanoutTreeChannelRole; // Used to approximate upload capacity in "children per node" terms. // For real enforcement we will likely move to token buckets and dynamic sizing, // but this gets us a bounded topology that we can reason about. msgRate: number; // messages/sec msgSize: number; // bytes uploadLimitBps: number; maxChildren: number; /** * Extra per-child overhead (bytes) to include when estimating upload usage. * * This should include stream/protocol overhead and any framing/signature/etc * that is not part of `msgSize`. */ uploadOverheadBytes?: number; /** * Token bucket burst window for upload shaping. * * Capacity = `uploadLimitBps * (uploadBurstMs/1000)`. */ uploadBurstMs?: number; bidPerByte?: number; allowKick?: boolean; /** * Enable bounded pull-repair (tree-first delivery, parent-based backfill). * * Defaults to `true`. Set to `false` to disable caching/repair logic for the channel. */ repair?: boolean; repairWindowMessages?: number; /** * How far back (in messages) we will attempt to repair missing sequences. * * Defaults to `repairWindowMessages` (no extra pruning). For "live" workloads you * may want to set this lower so repair doesn't waste work on stale data. */ repairMaxBackfillMessages?: number; repairIntervalMs?: number; repairMaxPerReq?: number; /** * If enabled, try to repair missing sequences by querying a small number of * additional peers (not only the current parent). This is a stepping stone * towards Plumtree-style neighbor-assisted repair. */ neighborRepair?: boolean; neighborRepairPeers?: number; /** * Target number of "lazy" mesh peers to keep connected for neighbor-assisted repair. * These peers are discovered via trackers and are not part of the parent/child tree. */ neighborMeshPeers?: number; /** * How often to send an IHAVE-style cache summary to mesh peers. * This is throttled and sent only when the node has seen new data. */ neighborAnnounceIntervalMs?: number; /** * How often to refresh mesh peers via tracker queries. */ neighborMeshRefreshIntervalMs?: number; /** * How long to trust an IHAVE summary from a peer when choosing fetch targets. */ neighborHaveTtlMs?: number; /** * Optional budget (token bucket) for neighbor-assisted repair `FETCH_REQ` control traffic. * * If `<= 0`, no budget is applied. */ neighborRepairBudgetBps?: number; neighborRepairBurstMs?: number; /** * Optional ingress budget (token bucket) for proxy publishes (`MSG_PUBLISH_PROXY`). * * This is enforced per established child link to cap amplification/DoS at the root * and on intermediate relays. Cost is measured in payload bytes. * * If `<= 0`, no budget is applied. */ proxyPublishBudgetBps?: number; proxyPublishBurstMs?: number; /** * Optional ingress budget (token bucket) for relaying unicast control traffic * (`MSG_UNICAST`, `MSG_UNICAST_ACK`). * * This is enforced per established child link. Cost is measured in payload bytes. * * If `<= 0`, no budget is applied. */ unicastBudgetBps?: number; unicastBurstMs?: number; /** * If set (>0), do not forward data that is older than this many milliseconds, * based on the message header timestamp (origin publish time, when forwarding * without re-signing). * * This is primarily intended for "live" workloads where late data is worse * than missing data. */ maxDataAgeMs?: number; /** * Max number of cached route tokens kept per channel for targeted unicast. * * `0` disables route caching. */ routeCacheMaxEntries?: number; /** * TTL for cached route tokens (milliseconds). * * `0` disables expiry. */ routeCacheTtlMs?: number; /** * Best-effort "peer hints" cache size bound (per channel). * * This tracks peer hashes observed on the channel control-plane and is used * for route proxy fanout and neighbor-assisted repair. It is intentionally * bounded so it cannot grow toward channel size at large scale. * * `0` disables peer hint tracking. */ peerHintMaxEntries?: number; /** * TTL for peer hint entries (milliseconds). * * `0` disables expiry. */ peerHintTtlMs?: number; }; export type FanoutTreeJoinOptions = { timeoutMs?: number; retryMs?: number; signal?: AbortSignal; staleAfterMs?: number; /** * How long to wait for a JOIN_ACCEPT/JOIN_REJECT after sending a JOIN_REQ. */ joinReqTimeoutMs?: number; /** * Optional bootstrap nodes used as rendezvous/tracker servers for joining. * If omitted, `FanoutTree.setBootstraps()` (if configured) is used. */ bootstrap?: Array; /** * Max time to wait for a bootstrapped peer to become a `@peerbit/stream` * neighbor (protocol streams established) after dialing. */ bootstrapDialTimeoutMs?: number; /** * Max number of bootstrap peers to dial and keep as tracker candidates while joining. * * Set to `0` to use all provided bootstraps. */ bootstrapMaxPeers?: number; /** * How many candidate parents to request per tracker query. */ trackerCandidates?: number; /** * Shuffle only within the first K ranked candidates to spread load without * destroying the bias towards low-level/high-capacity parents. * * Set to `0` to disable shuffling. */ candidateShuffleTopK?: number; /** * How long to wait for a tracker reply before proceeding. */ trackerQueryTimeoutMs?: number; /** * How often to announce parent capacity to trackers (keep-alive). */ announceIntervalMs?: number; /** * TTL for announcements stored by trackers. * Trackers should treat entries as stale after this. */ announceTtlMs?: number; /** * Min interval between re-dial attempts to bootstrap peers when joining. */ bootstrapEnsureIntervalMs?: number; /** * Min interval between tracker queries while joining. */ trackerQueryIntervalMs?: number; /** * Min interval between parent-improvement checks while already attached. * * Set to `0` to preserve the current stability-first behavior where a healthy * parent is only replaced after disconnect/staleness/kick. */ parentUpgradeIntervalMs?: number; /** * Restrict proactive parent upgrades to leaves (nodes with no children). * * Defaults to `true` so parent improvement can be evaluated without * introducing relay churn higher in the tree. */ parentUpgradeLeafOnly?: boolean; /** * Minimum tree-level improvement required before replacing a healthy parent. * * `1` means only strictly better parents are eligible. Higher values add * hysteresis against marginal topology churn. */ parentUpgradeMinLevelGain?: number; /** * Minimum tree-level improvement required for proactive upgrades directly to * the root. * * Defaults to `3`. Root fanout is the scarce sender-side resource for a topic, * so direct-root upgrades require a stronger benefit than ordinary relay * improvements. */ parentUpgradeRootMinLevelGain?: number; /** * Minimum local branch improvement required for proactive upgrades directly to * the root. * * This is `levelGain * (1 + directChildren)`. It lets a relay-to-root move be * admitted when it improves a branch, even if the relay's own level gain is * smaller than `parentUpgradeRootMinLevelGain`. Defaults to * `parentUpgradeRootMinLevelGain`. */ parentUpgradeRootMinSubtreeGain?: number; /** * Minimum tree-level improvement required for proactive upgrades to non-root * relay parents. * * Defaults to `2` so a late direct root edge can still repair a one-level * detour, while marginal relay-to-relay moves are avoided under load. */ parentUpgradeNonRootMinLevelGain?: number; /** * Minimum advertised free child slots required for a proactive upgrade target. * * Defaults to `8` so a target still has several spare slots after accepting * this child; this avoids stampeding nearly-full parents under constrained * fanout. */ parentUpgradeMinFreeSlots?: number; /** * Minimum advertised free child slots required for a proactive direct-root * upgrade target. * * Defaults to `parentUpgradeMinFreeSlots`. This can be tuned independently * because root spare capacity protects sender-side fanout pressure, while * relay spare capacity protects intermediate branch stability. */ parentUpgradeRootMinFreeSlots?: number; /** * Maximum child-load ratio allowed for a live-probed proactive upgrade target * after accepting this peer. * * Defaults to `0.5`. Set to `0` to disable the ratio check. */ parentUpgradeMaxChildLoadRatio?: number; /** * Maximum child-load ratio allowed for a live-probed direct-root upgrade target * after accepting this peer. * * Defaults to `min(parentUpgradeMaxChildLoadRatio, 0.4)`. Root fanout is the * scarce sender-side resource, so root pressure stays more conservative than * relay parent pressure. Peers maintaining multiple local channels apply an * additional effective cap of `0.2` for direct-root upgrades so concurrent * writer trees do not concentrate on the same root at once. */ parentUpgradeRootMaxChildLoadRatio?: number; /** * Cooldown after a successful proactive upgrade before trying another one. */ parentUpgradeCooldownMs?: number; /** * Initial backoff after a failed probe/shadow parent-upgrade round. * * Failed rounds usually mean stale tracker state or saturated candidate * parents. Backoff is per channel and doubles up to * `parentUpgradeFailedBackoffMaxMs`. */ parentUpgradeFailedBackoffMinMs?: number; /** * Maximum per-channel backoff after repeated failed probe/shadow rounds. */ parentUpgradeFailedBackoffMaxMs?: number; /** * Minimum quiet time since parent attachment/data before a proactive upgrade. * * Defaults to 5s. This avoids voluntarily probing or switching parents during * active delivery or directly after a repair/rejoin event. */ parentUpgradeQuietMs?: number; /** * Minimum quiet time since the last local repair request before a proactive * upgrade check. * * Defaults to `parentUpgradeQuietMs` so upgrade probing stays silent while the * peer is still recovering missing data. */ parentUpgradeRepairQuietMs?: number; /** * Max successful proactive upgrades for one channel peer. Set to `0` for no cap. */ parentUpgradeMaxPerPeer?: number; /** * Skip proactive upgrades while this peer is actively missing data. */ parentUpgradeRepairGuard?: boolean; /** * Skip proactive upgrades until a finite channel end has been received and caught up. */ parentUpgradeDataGuard?: boolean; /** * Parent-upgrade strategy. * * - `direct`: join a better advertised parent directly. * - `probe`: first ask candidate parents for fresh level/capacity/data-state, * then only join a bounded winner. * - `shadow` (default): keep the current parent active, repeatedly probe one * candidate over an observation window, then promote only after it remains * healthy. */ parentUpgradeMode?: "direct" | "probe" | "shadow"; /** * Allow shadow mode to probe a direct root candidate even when tracker * capacity says the root is full. * * Defaults to `true` in shadow mode and `false` otherwise. Shadow mode keeps * this bounded by `parentUpgradeStaleRootProbeProbability` and live capacity * probes, while direct/probe modes avoid stale-root pressure unless explicitly * enabled. */ parentUpgradeVerifyStaleRootCapacity?: boolean; /** * Deterministic per-peer base sampling probability for stale-root verification. * * This applies only when `parentUpgradeVerifyStaleRootCapacity` is enabled and * tracker capacity says the root is full. Defaults to `0.015625` to avoid every * eligible peer probing the same advertised-full root at once. Single-channel * branch peers may use a bounded sample boost, capped at `0.25`; quiet, * completed single-channel leaves may use a `0.5` sample. Settled leaves that * maintain multiple local channels require fresh tracker capacity, recent slow * parent-delivery evidence, and a low-rate deterministic sample at the default * probability; raise this value explicitly to opt into more pressure. */ parentUpgradeStaleRootProbeProbability?: number; /** * How long to wait for a parent probe reply before skipping that candidate. */ parentProbeTimeoutMs?: number; /** * Max candidate parents to probe per upgrade check. */ parentProbeMaxPerRound?: number; /** * Max sequence lag tolerated for a probed parent candidate. * * `0` requires the candidate to be at least as fresh as this peer. */ parentProbeMaxLagMessages?: number; /** * Cooldown for a candidate after a probe proves it cannot safely become * our parent. This bounds repeated probes against stale tracker entries. */ parentProbeRejectCooldownMs?: number; /** * Maximum adaptive cooldown for a candidate after repeated rejected parent * probes. */ parentProbeRejectCooldownMaxMs?: number; /** * Minimum time a shadow parent candidate must stay healthy before promotion. */ parentShadowObserveMs?: number; /** * Minimum successful shadow observations required before promotion. */ parentShadowMinObservations?: number; /** * In shadow mode, keep the old parent attached for this long after a * candidate accepts JOIN before promoting during active data flow. * * Defaults to `5000` in shadow mode and `0` otherwise. Set to `0` to use * immediate post-probe promotion. */ parentShadowDualPathMs?: number; /** * Minimum live messages that must arrive from a shadow-attached candidate * before active-flow promotion is allowed. * * Defaults to `32` in shadow mode and `1` otherwise. */ parentShadowDualPathMinMessages?: number; /** * Max number of join candidates to try per retry "round". * * This prevents a long tail of sequential JOIN_REQ timeouts from blocking * the join loop for tens of seconds under overload. */ joinAttemptsPerRound?: number; /** * Cooldown applied to a candidate parent after dial/join failures. * * This reduces hot-spotting (everyone hammering the same few parents) and * keeps control-plane traffic bounded at large scale. */ candidateCooldownMs?: number; /** * Candidate scoring mode for selecting parent join targets. * * - `ranked-shuffle` (default): rank by (level, freeSlots, bid, source) and * shuffle within `candidateShuffleTopK` to spread load. * - `ranked-strict`: try ranked candidates in order (no shuffle). * - `weighted`: weighted shuffle within `candidateShuffleTopK` using * `candidateScoringWeights` (defaults bias low level + free slots). */ candidateScoringMode?: "ranked-shuffle" | "ranked-strict" | "weighted"; /** * Weights used when `candidateScoringMode="weighted"`. * * Larger values increase the influence of that signal. */ candidateScoringWeights?: { level?: number; freeSlots?: number; connected?: number; bidPerByte?: number; source?: number; }; }; /** * Tracker-backed provider discovery for targeted pulls (blocks/RPC). * * Providers announce under a bounded namespace key, and consumers query for K candidates. * This avoids any search/flood behaviors at large scale. */ export type FanoutProviderCandidate = { hash: string; addrs: Multiaddr[]; }; export type FanoutProviderAnnounceOptions = { ttlMs?: number; announceIntervalMs?: number; bootstrap?: Array; bootstrapDialTimeoutMs?: number; bootstrapMaxPeers?: number; }; export type FanoutProviderQueryOptions = { want?: number; seed?: number; timeoutMs?: number; queryTimeoutMs?: number; cacheTtlMs?: number; signal?: AbortSignal; bootstrap?: Array; bootstrapDialTimeoutMs?: number; bootstrapMaxPeers?: number; }; export type FanoutProviderWatchOptions = { want?: number; signal?: AbortSignal; onProviders: (providers: FanoutProviderCandidate[]) => void; bootstrap?: Array; bootstrapDialTimeoutMs?: number; bootstrapMaxPeers?: number; ttlMs?: number; renewIntervalMs?: number; }; export type FanoutProviderHandle = { close: () => void; }; export type FanoutTreeDataEvent = { topic: string; root: string; seq: number; payload: Uint8Array; from: string; // immediate sender (edge used for forwarding) origin: string; // original sender (signature[0]) if present, else `from` timestamp: bigint; // sender-provided timestamp (DataMessage.header.timestamp) message: DataMessage; // transport-level message (signed by root for proxy publishes) }; export type FanoutTreeUnicastEvent = { topic: string; root: string; route: string[]; payload: Uint8Array; from: string; // immediate sender (edge used for forwarding) origin: string; // original sender (signature[0]) to: string; // final destination hash timestamp: bigint; // sender-provided timestamp (DataMessage.header.timestamp) message: DataMessage; // transport-level message carrying the unicast control frame }; /** * Diagnostic counters used by tests and simulation harnesses. * * These fields are intentionally detailed so failures can explain which guard * fired, but they are not a stable product API. * * @internal */ export type FanoutTreeChannelMetrics = { controlSends: number; controlBytesSent: number; controlReceives: number; controlBytesReceived: number; /** * Control-plane byte breakdown by purpose. * * These include the full encoded control message bytes and are counted per transmission. */ controlBytesSentJoin: number; controlBytesSentRepair: number; controlBytesSentTracker: number; controlBytesReceivedJoin: number; controlBytesReceivedRepair: number; controlBytesReceivedTracker: number; dataSends: number; dataPayloadBytesSent: number; dataReceives: number; dataPayloadBytesReceived: number; staleForwardsDropped: number; dataWriteDrops: number; /** * Proxy publish frames dropped due to local rate limiting. * * This is an abuse-resistance knob: it caps how much a single child can * amplify traffic via the root. */ proxyPublishDrops: number; /** * Unicast frames dropped due to local rate limiting. */ unicastDrops: number; joinReqSent: number; joinReqReceived: number; joinAcceptSent: number; joinAcceptReceived: number; joinRejectSent: number; joinRejectReceived: number; joinPeerResets: number; kickSent: number; kickReceived: number; reparentDisconnect: number; reparentStale: number; reparentKicked: number; reparentUpgrade: number; reparentUpgradeSkipLeaf: number; reparentUpgradeSkipRepair: number; reparentUpgradeSkipData: number; reparentUpgradeSkipCooldown: number; reparentUpgradeSkipQuiet: number; reparentUpgradeSkipBudget: number; reparentUpgradeSkipCandidateLevel: number; reparentUpgradeSkipCandidateSlots: number; reparentUpgradeSkipCandidatePressure: number; reparentUpgradeSkipRootPressure: number; reparentUpgradeSkipProbeNoReply: number; reparentUpgradeSkipProbeNotRooted: number; reparentUpgradeSkipProbeRepair: number; reparentUpgradeSkipProbeLag: number; reparentUpgradeSkipProbeOverloaded: number; reparentUpgradeSkipProbeCooldown: number; parentProbeReqSent: number; parentProbeReqReceived: number; parentProbeReplySent: number; parentProbeReplyReceived: number; parentUpgradeRootReservationCreated: number; parentUpgradeRootReservationConsumed: number; parentUpgradeRootReservationRejected: number; parentUpgradeRootReservationMarginRejected: number; parentUpgradeRootReservationBlocked: number; parentUpgradeRootReservationExpired: number; parentShadowStart: number; parentShadowObserve: number; parentShadowPromote: number; parentShadowReset: number; parentShadowRejectNoReply: number; parentShadowRejectNotRooted: number; parentShadowRejectCapacity: number; parentShadowRejectRepair: number; parentShadowRejectLag: number; parentShadowRejectOverloaded: number; parentShadowRejectLevel: number; endSent: number; endReceived: number; repairReqSent: number; repairReqReceived: number; fetchReqSent: number; fetchReqReceived: number; ihaveSent: number; ihaveReceived: number; trackerAnnounceSent: number; trackerAnnounceReceived: number; trackerQuerySent: number; trackerQueryReceived: number; trackerReplySent: number; trackerReplyReceived: number; trackerFeedbackSent: number; trackerFeedbackReceived: number; cacheHitsServed: number; cacheMissesServed: number; holeFillsFromNeighbor: number; earnings: number; routeCacheHits: number; routeCacheMisses: number; routeCacheExpirations: number; routeCacheEvictions: number; routeProxyQueries: number; routeProxyTimeouts: number; routeProxyFanout: number; routeProxyCoalesced: number; routeProxyRejected: number; }; export interface FanoutTreeEvents extends StreamEvents { "fanout:data": CustomEvent; "fanout:unicast": CustomEvent; "fanout:joined": CustomEvent<{ topic: string; root: string; parent: string }>; "fanout:kicked": CustomEvent<{ topic: string; root: string; from: string }>; "fanout:provider": CustomEvent<{ namespace: string; providers: FanoutProviderCandidate[]; }>; "fanout:peer-unreachable": CustomEvent<{ topic: string; root: string; publicKeyHash: string; }>; } const CONTROL_PRIORITY = 10; const DATA_PRIORITY = 1; const ROUTE_PROXY_TIMEOUT_MS = 10_000; // Backstop bounds for the route-proxy machinery under query storms (#983): // each pending entry retains a live timer closure and a Set of child hashes. const ROUTE_PROXY_MAX_PENDING = 2_048; const ROUTE_PROXY_MAX_WAITERS = 512; const ROUTE_CACHE_MAX_ENTRIES_HARD_CAP = 100_000; const PEER_HINT_MAX_ENTRIES_HARD_CAP = 100_000; // Best-effort address cache for join candidates learned via trackers/redirects. // This must stay small; large values can blow up memory in large-scale sims. const KNOWN_CANDIDATE_ADDRS_MAX_ENTRIES = 256; const PROVIDER_DIRECTORY_MAX_ENTRIES = 16_384; // Best-effort bound on the number of provider namespaces we keep cached locally. // This prevents unbounded memory growth if callers use extremely high-cardinality namespaces // (for example, `cid:` per block). const PROVIDER_DIRECTORY_MAX_NAMESPACES = 4_096; // Best-effort bounds for tracker state (join candidate directory). // Trackers should keep only a bounded set of recent candidates per channel, otherwise // large networks can cause unbounded memory growth on the tracker nodes. const TRACKER_DIRECTORY_MAX_ENTRIES = 16_384; const TRACKER_DIRECTORY_MAX_NAMESPACES = 4_096; const FANOUT_PROTOCOLS = ["/peerbit/fanout-tree/0.5.0"]; const ID_PREFIX = Uint8Array.from([0x46, 0x4f, 0x55, 0x54]); // "FOUT" const OVERLOAD_KICK_STREAK_THRESHOLD = 5; const OVERLOAD_KICK_COOLDOWN_MS = 2_000; const OVERLOAD_KICK_MAX_PER_EVENT = 4; // DATA-plane sends are best-effort (repair provides reliability). Never block on // stream writability; optionally kick persistently failing children. const DATA_WRITE_FAIL_KICK_STREAK_THRESHOLD = 3; const DATA_WRITE_FAIL_KICK_COOLDOWN_MS = 2_000; const DATA_WRITE_FAIL_KICK_MAX_PER_EVENT = 4; const PARENT_REPAIR_DEAD_STREAK_THRESHOLD = 16; const PARENT_REPAIR_DEAD_MIN_LIVENESS_MS = 15_000; const REPAIR_RETRY_MIN_MS = 1_000; const REPAIR_RETRY_INTERVAL_FACTOR = 5; // A stream can remain locally readable/writable while silently dropping every // control frame. Repeated unanswered JOIN requests are end-to-end evidence that // the peer path needs to be rebuilt. const JOIN_TIMEOUT_RESET_STREAK_THRESHOLD = 3; const JOIN_TIMEOUT_RESET_COOLDOWN_MS = 10_000; const JOIN_REJECT_REDIRECT_QUEUE_MAX = 64; // When a relay loses its parent, pause before trying to rejoin so its children can // attach elsewhere (helps avoid disconnected components stabilizing). // Rejoin cooldown after losing a parent while acting as a relay (had children). // // This gives kicked children a window to reattach elsewhere (helps avoid stable // disconnected components). Leaf nodes (no children) rejoin immediately. const RELAY_REJOIN_COOLDOWN_MS = 3_000; const PARENT_UPGRADE_ROOT_RESERVATION_TTL_MS = 5_000; const stableUnitInterval = (input: string) => { let hash = 0x811c9dc5; for (let i = 0; i < input.length; i++) { hash ^= input.charCodeAt(i); hash = Math.imul(hash, 0x01000193); } return (hash >>> 0) / 0x100000000; }; const textEncoder = new TextEncoder(); type RouteCacheEntry = { route: string[]; updatedAt: number; }; type TrackerCandidate = { hash: string; level: number; freeSlots: number; bidPerByte: number; addrs: Multiaddr[]; }; type ParentShadowState = { hash: string; startedAt: number; observations: number; level: number; freeSlots: number; haveToExclusive: number; liveDataMessages: number; liveFirstDataMessages: number; liveParentFirstDataMessages: number; liveCandidateLeadSamples: number; liveCandidateLeadMsTotal: number; liveCandidateFirstAtBySeq: Map; liveParentFirstAtBySeq: Map; liveComparedSeqs: Set; liveMaxSeqSeen: number; liveLastDataAt: number; }; type ParentUpgradeGraceState = { previousParent: string; candidateParent: string; previousLevel: number; previousRouteFromRoot?: string[]; previousLastParentDataAt: number; previousLastParentUpgradeActivityAt: number; previousReceivedAnyParentData: boolean; timer: ReturnType; candidateFirstMessages: number; previousFirstMessages: number; minCandidateFirstMessages: number; minCandidateAdvantageMessages: number; maxPreviousFirstMessages: number; }; type ProviderNamespaceId = { namespace: string; key: Uint8Array; suffixKey: string; }; type ProviderWatchRegistration = { hash: string; want: number; expiresAt: number; }; type ProviderAnnounceState = { id: ProviderNamespaceId; ttlMs: number; announceIntervalMs: number; bootstrapOverride?: Multiaddr[]; bootstrapDialTimeoutMs: number; bootstrapMaxPeers: number; closed: boolean; loop?: Promise; }; type ProviderWatchState = { id: ProviderNamespaceId; want: number; ttlMs: number; renewIntervalMs: number; bootstrapOverride?: Multiaddr[]; bootstrapDialTimeoutMs: number; bootstrapMaxPeers: number; onProviders: (providers: FanoutProviderCandidate[]) => void; signal?: AbortSignal; closed: boolean; trackerPeers: string[]; loop?: Promise; }; type ChildInfo = { bidPerByte: number }; type JoinAttemptResult = { ok: boolean; parentLevel?: number; parentRouteFromRoot?: string[]; rejectReason?: number; timedOut?: boolean; redirects?: Array<{ hash: string; addrs: Multiaddr[] }>; }; type PendingUnicastAck = { expectedOrigin: string; resolve: () => void; reject: (error: unknown) => void; timer: ReturnType; signal?: AbortSignal; onAbort?: () => void; }; type ChannelState = { id: FanoutTreeChannelId; metrics: FanoutTreeChannelMetrics; level: number; isRoot: boolean; closed: boolean; parent?: string; children: Map; /** * Cooldown applied after a relay loses its parent. * * This gives recently-kicked children a chance to attach elsewhere before the * relay races to reclaim scarce parent capacity, which helps avoid * disconnected components stabilizing under an unrooted relay. */ rejoinCooldownUntil: number; parentUpgradeCount: number; parentUpgradeLastAt: number; parentUpgradeBackoffMs: number; parentUpgradeBackoffUntil: number; parentUpgradeRetryAfterSeq: number; parentUpgradeStaleRootProbeRound: number; parentUpgradeTrackerNoCapacityUntil: number; parentProbeRejectBackoffMsByHash: Map; /** * True once this node has successfully joined the channel at least once. * * Join timeouts should only apply to the initial `joinChannel()` await, not to * later re-parenting after disconnects. */ joinedAtLeastOnce: boolean; /** * Source route from the channel root to this node (inclusive). * * This is learned during JOIN via the parent and can be shared out-of-band * to enable economical unicast within the channel (no global membership map). */ routeFromRoot?: string[]; /** * Best-effort route cache keyed by target peer hash. * * Entries are learned on-demand via `ROUTE_QUERY/ROUTE_REPLY` and cached locally. */ routeByPeer: Map; seq: number; // options bidPerByte: number; uploadLimitBps: number; maxChildren: number; effectiveMaxChildren: number; allowKick: boolean; maxDataAgeMs: number; repairEnabled: boolean; repairWindowMessages: number; repairMaxBackfillMessages: number; repairIntervalMs: number; repairMaxPerReq: number; neighborRepair: boolean; neighborRepairPeers: number; neighborMeshPeers: number; neighborAnnounceIntervalMs: number; neighborMeshRefreshIntervalMs: number; neighborHaveTtlMs: number; neighborRepairBudgetBps: number; neighborRepairTokenCapacity: number; neighborRepairTokens: number; neighborRepairLastRefillAt: number; proxyPublishBudgetBps: number; proxyPublishTokenCapacity: number; proxyPublishTokensByPeer: Map< string, { tokens: number; lastRefillAt: number } >; unicastBudgetBps: number; unicastTokenCapacity: number; unicastTokensByPeer: Map; uploadOverheadBytes: number; uploadBurstMs: number; uploadTokenCapacity: number; uploadTokens: number; uploadLastRefillAt: number; droppedForwards: number; overloadStreak: number; lastOverloadKickAt: number; dataWriteFailStreakByChild: Map; lastDataWriteFailKickAt: number; cacheSeqs?: Int32Array; cachePayloads?: Array; nextExpectedSeq: number; missingSeqs: Set; repairRequestedAtBySeq: Map; endSeqExclusive: number; lastRepairSentAt: number; parentRepairUnansweredStreak: number; lastParentDataAt: number; lastParentUpgradeActivityAt: number; receivedAnyParentData: boolean; parentDataLatencySamples: number; parentDataLatencyEwmaMs: number; parentDataLatencyMaxMs: number; channelPeers: Map; knownCandidateAddrs: Map; lazyPeers: Set; haveByPeer: Map< string, { haveFrom: number; haveToExclusive: number; updatedAt: number; requests: number; successes: number; } >; maxSeqSeen: number; lastIHaveSentAt: number; cachedAnnounceTrackerPeers?: string[]; lastIHaveSentMaxSeq: number; pendingJoin: Map< number, { resolve(res: JoinAttemptResult): void; shadowAttach?: boolean } >; pendingTrackerQuery: Map< number, { resolve(entries: TrackerCandidate[]): void } >; pendingParentProbe: Map; parentUpgradeReservationsByHash: Map< string, { token: number; expiresAt: number; minFreeSlots: number } >; parentProbeRejectUntilByHash: Map; parentUpgradeGrace?: ParentUpgradeGraceState; parentShadow?: ParentShadowState; pendingRouteQuery: Map; pendingUnicastAck: Map; pendingRouteProxy: Map< number, { requester: string; downstreamReqId: number; timer: ReturnType; expectedReplies: Set; targetHash: string; direction: "up" | "down"; /** * Additional requesters coalesced onto this in-flight search; all are * answered from its single result. Without this, S concurrent lookups * for the same cold target each launch an independent subtree flood. */ waiters?: Array< | { requester: string; downstreamReqId: number } | { localResolve: (route?: string[]) => void } >; /** * Optional local completion callback (used when the root resolves a route token * for itself by fanning out queries to children). */ localResolve?: (route?: string[]) => void; } >; /** `${direction}:${targetHash}` -> in-flight proxyReqId, for coalescing. */ routeProxyByTarget: Map; bootstrapOverride?: Multiaddr[]; bootstrapDialTimeoutMs: number; bootstrapMaxPeers: number; bootstrapEnsureIntervalMs: number; cachedBootstrapPeers: string[]; lastBootstrapEnsureAt: number; announceIntervalMs: number; announceTtlMs: number; lastAnnouncedAt: number; peerHintMaxEntries: number; peerHintTtlMs: number; routeCacheMaxEntries: number; routeCacheTtlMs: number; announceLoop?: Promise; repairLoop?: Promise; meshLoop?: Promise; joinLoop?: Promise; joinedOnce?: { resolve(): void; reject(err: unknown): void; promise: Promise; }; trackerQueryIntervalMs: number; cachedTrackerCandidates: TrackerCandidate[]; lastTrackerQueryAt: number; }; const createDeferred = (): { resolve(): void; reject(err: unknown): void; promise: Promise; } => { let resolve!: () => void; let reject!: (err: unknown) => void; const promise = new Promise((res, rej) => { resolve = res; reject = rej; }); return { resolve, reject, promise }; }; const createEmptyMetrics = (): FanoutTreeChannelMetrics => ({ controlSends: 0, controlBytesSent: 0, controlReceives: 0, controlBytesReceived: 0, controlBytesSentJoin: 0, controlBytesSentRepair: 0, controlBytesSentTracker: 0, controlBytesReceivedJoin: 0, controlBytesReceivedRepair: 0, controlBytesReceivedTracker: 0, dataSends: 0, dataPayloadBytesSent: 0, dataReceives: 0, dataPayloadBytesReceived: 0, staleForwardsDropped: 0, dataWriteDrops: 0, proxyPublishDrops: 0, unicastDrops: 0, joinReqSent: 0, joinReqReceived: 0, joinAcceptSent: 0, joinAcceptReceived: 0, joinRejectSent: 0, joinRejectReceived: 0, joinPeerResets: 0, kickSent: 0, kickReceived: 0, reparentDisconnect: 0, reparentStale: 0, reparentKicked: 0, reparentUpgrade: 0, reparentUpgradeSkipLeaf: 0, reparentUpgradeSkipRepair: 0, reparentUpgradeSkipData: 0, reparentUpgradeSkipCooldown: 0, reparentUpgradeSkipQuiet: 0, reparentUpgradeSkipBudget: 0, reparentUpgradeSkipCandidateLevel: 0, reparentUpgradeSkipCandidateSlots: 0, reparentUpgradeSkipCandidatePressure: 0, reparentUpgradeSkipRootPressure: 0, reparentUpgradeSkipProbeNoReply: 0, reparentUpgradeSkipProbeNotRooted: 0, reparentUpgradeSkipProbeRepair: 0, reparentUpgradeSkipProbeLag: 0, reparentUpgradeSkipProbeOverloaded: 0, reparentUpgradeSkipProbeCooldown: 0, parentProbeReqSent: 0, parentProbeReqReceived: 0, parentProbeReplySent: 0, parentProbeReplyReceived: 0, parentUpgradeRootReservationCreated: 0, parentUpgradeRootReservationConsumed: 0, parentUpgradeRootReservationRejected: 0, parentUpgradeRootReservationMarginRejected: 0, parentUpgradeRootReservationBlocked: 0, parentUpgradeRootReservationExpired: 0, parentShadowStart: 0, parentShadowObserve: 0, parentShadowPromote: 0, parentShadowReset: 0, parentShadowRejectNoReply: 0, parentShadowRejectNotRooted: 0, parentShadowRejectCapacity: 0, parentShadowRejectRepair: 0, parentShadowRejectLag: 0, parentShadowRejectOverloaded: 0, parentShadowRejectLevel: 0, endSent: 0, endReceived: 0, repairReqSent: 0, repairReqReceived: 0, fetchReqSent: 0, fetchReqReceived: 0, ihaveSent: 0, ihaveReceived: 0, trackerAnnounceSent: 0, trackerAnnounceReceived: 0, trackerQuerySent: 0, trackerQueryReceived: 0, trackerReplySent: 0, trackerReplyReceived: 0, trackerFeedbackSent: 0, trackerFeedbackReceived: 0, cacheHitsServed: 0, cacheMissesServed: 0, holeFillsFromNeighbor: 0, earnings: 0, routeCacheHits: 0, routeCacheMisses: 0, routeCacheExpirations: 0, routeCacheEvictions: 0, routeProxyQueries: 0, routeProxyTimeouts: 0, routeProxyFanout: 0, routeProxyCoalesced: 0, routeProxyRejected: 0, }); const isDataId = (id: Uint8Array) => id.length === 32 && id[0] === ID_PREFIX[0] && id[1] === ID_PREFIX[1] && id[2] === ID_PREFIX[2] && id[3] === ID_PREFIX[3]; export class FanoutTree extends DirectStream { private channelsBySuffixKey = new Map(); private readonly cachedSuffixKey = new WeakMap(); private readonly metricsBySuffixKey = new Map< string, FanoutTreeChannelMetrics >(); private readonly joinTimeoutStreakByPeer = new Map(); private readonly joinResetCooldownUntilByPeer = new Map(); private bootstraps: Multiaddr[] = []; private trackerBySuffixKey = new Map>(); private trackerNamespaceLru = new Map(); private providerBySuffixKey = new Map>(); private providerNamespaceLru = new Map(); private providerNamespaceBySuffixKey = new Map(); private providerWatchersBySuffixKey = new Map< string, Map >(); private providerWatchesBySuffixKey = new Map< string, Set >(); private underlayPeerDisconnectHandler?: (ev: any) => void; private pendingProviderQueryBySuffixKey = new Map< string, Map void }> >(); private providerAnnounceBySuffixKey = new Map< string, ProviderAnnounceState >(); private parentUpgradeShadowInFlightSuffixKey?: string; private readonly defaultUploadOverheadBytes = 128; private readonly random: () => number; private readonly unicastAckNodeTag32: number; private unicastAckSeq = 0; public readonly topicRootControlPlane: TopicRootControlPlane; // Native fanout components of the rust-core DirectStream engine // (`@peerbit/network-rust`). When set, every frame encode/decode of the // `/peerbit/fanout-tree/0.5.0` codec plus the parent-upgrade policy/gate // decisions run natively (byte- and decision-identical, covered by the // golden parity suite); the channel state machine, timers and events // stay host-side. Unset (the default) leaves every code path as-is. private readonly nativeFanout?: RustFanoutTree; private readonly codec: FanoutWireCodec; constructor( components: DirectStreamComponents, opts?: FanoutTreeStreamOptions, ) { const { random, topicRootControlPlane, ...rest } = (opts ?? {}) as FanoutTreeStreamOptions; super(components, FANOUT_PROTOCOLS, { canRelayMessage: false, ...rest, }); this.random = typeof random === "function" ? random : Math.random; this.nativeFanout = this.rustCore?.fanout; this.codec = this.nativeFanout ?? tsFanoutWireCodec; this.unicastAckNodeTag32 = readU32BE( sha256Sync(textEncoder.encode(this.publicKeyHash)), 0, ); this.topicRootControlPlane = topicRootControlPlane ?? new TopicRootControlPlane(); const onPeerDisconnect = (ev: any) => { const peerId = ev?.detail; if (!peerId) return; let peerHash: string; try { peerHash = getPublicKeyFromPeerId(peerId).hashcode(); } catch { return; } this.onPeerDisconnectedFromUnderlay(peerHash); }; this.underlayPeerDisconnectHandler = onPeerDisconnect; this.components.events.addEventListener( "peer:disconnect", onPeerDisconnect as any, ); } public override async stop() { if (this.underlayPeerDisconnectHandler) { this.components.events.removeEventListener( "peer:disconnect", this.underlayPeerDisconnectHandler as any, ); this.underlayPeerDisconnectHandler = undefined; } this.joinTimeoutStreakByPeer.clear(); this.joinResetCooldownUntilByPeer.clear(); return super.stop(); } public setBootstraps(addrs: Array) { this.bootstraps = addrs .map((a) => (typeof a === "string" ? multiaddr(a) : a)) .filter((a) => Boolean(a)); } public addBootstraps(addrs: Array) { const merged = new Map( this.bootstraps.map((addr) => [addr.toString(), addr]), ); for (const addr of addrs) { const multiaddrValue = typeof addr === "string" ? multiaddr(addr) : addr; merged.set(multiaddrValue.toString(), multiaddrValue); } this.bootstraps = [...merged.values()]; } private touchTrackerNamespace(suffixKey: string, now = Date.now()) { if (!suffixKey) return; // LRU touch this.trackerNamespaceLru.delete(suffixKey); this.trackerNamespaceLru.set(suffixKey, now); while (this.trackerNamespaceLru.size > TRACKER_DIRECTORY_MAX_NAMESPACES) { const oldest = this.trackerNamespaceLru.keys().next().value as | string | undefined; if (!oldest) break; this.trackerNamespaceLru.delete(oldest); this.trackerBySuffixKey.delete(oldest); } } private pruneTrackerNamespaceIfEmpty(suffixKey: string) { const byPeer = this.trackerBySuffixKey.get(suffixKey); if (byPeer && byPeer.size === 0) { this.trackerBySuffixKey.delete(suffixKey); this.trackerNamespaceLru.delete(suffixKey); } } private touchProviderNamespace(suffixKey: string, now = Date.now()) { if (!suffixKey) return; // LRU touch this.providerNamespaceLru.delete(suffixKey); this.providerNamespaceLru.set(suffixKey, now); while (this.providerNamespaceLru.size > PROVIDER_DIRECTORY_MAX_NAMESPACES) { const oldest = this.providerNamespaceLru.keys().next().value as | string | undefined; if (!oldest) break; this.providerNamespaceLru.delete(oldest); this.providerBySuffixKey.delete(oldest); this.providerNamespaceBySuffixKey.delete(oldest); this.pendingProviderQueryBySuffixKey.delete(oldest); } } private pruneProviderNamespaceIfEmpty(suffixKey: string) { const byPeer = this.providerBySuffixKey.get(suffixKey); if (byPeer && byPeer.size === 0) { this.providerBySuffixKey.delete(suffixKey); this.providerNamespaceLru.delete(suffixKey); this.pendingProviderQueryBySuffixKey.delete(suffixKey); } } private pruneProviderWatchersIfEmpty(suffixKey: string) { const watchers = this.providerWatchersBySuffixKey.get(suffixKey); if (watchers && watchers.size === 0) { this.providerWatchersBySuffixKey.delete(suffixKey); } } private getProviderCandidatesFromCache( id: ProviderNamespaceId, options?: { now?: number; want?: number }, ): FanoutProviderCandidate[] { const now = options?.now ?? Date.now(); const cached: FanoutProviderCandidate[] = []; const byPeer = this.providerBySuffixKey.get(id.suffixKey); if (byPeer) { for (const [hash, e] of byPeer) { if (e.expiresAt <= now) { byPeer.delete(hash); continue; } if (hash === this.publicKeyHash) continue; const addrs: Multiaddr[] = []; for (const a of e.addrs) { try { addrs.push(multiaddr(a)); } catch { // ignore invalid } } cached.push({ hash, addrs }); if (options?.want && cached.length >= options.want) break; } this.pruneProviderNamespaceIfEmpty(id.suffixKey); } return cached; } /** Decoded provider entries (raw multiaddr bytes) -> candidates. Invalid * multiaddrs are dropped exactly like the historical inline parsing. */ private toProviderCandidates( entries: { hash: string; addrs: Uint8Array[] }[], ): ProviderCandidate[] { const candidates: ProviderCandidate[] = []; for (const entry of entries) { const addrs: Multiaddr[] = []; for (const bytes of entry.addrs) { try { addrs.push(multiaddr(bytes)); } catch { // ignore invalid multiaddrs } } candidates.push({ hash: entry.hash, addrs }); } return candidates; } private rememberProviderCandidates( id: ProviderNamespaceId, candidates: FanoutProviderCandidate[], expiresAt: number, ) { if (candidates.length === 0) return; let cache = this.providerBySuffixKey.get(id.suffixKey); if (!cache) { cache = new Map(); this.providerBySuffixKey.set(id.suffixKey, cache); } for (const c of candidates) { cache.delete(c.hash); cache.set(c.hash, { hash: c.hash, addrs: c.addrs.map((a) => a.bytes), expiresAt, }); } while (cache.size > PROVIDER_DIRECTORY_MAX_ENTRIES) { const oldest = cache.keys().next().value as string | undefined; if (!oldest) break; cache.delete(oldest); } this.touchProviderNamespace(id.suffixKey); } private emitProviderUpdate( id: ProviderNamespaceId, providers: FanoutProviderCandidate[], ) { if (providers.length === 0) return; this.dispatchEvent( new CustomEvent("fanout:provider", { detail: { namespace: id.namespace, providers }, }), ); const localWatches = this.providerWatchesBySuffixKey.get(id.suffixKey); if (!localWatches || localWatches.size === 0) return; for (const watch of localWatches) { if (watch.closed) continue; try { watch.onProviders(providers.slice(0, watch.want)); } catch { // ignore consumer callback failures } } } private async publishProviderWatchUpdate(id: ProviderNamespaceId) { const watchers = this.providerWatchersBySuffixKey.get(id.suffixKey); if (!watchers || watchers.size === 0) return; const now = Date.now(); const cached = this.getProviderCandidatesFromCache(id, { now }); const byPeer = new Map(watchers); for (const [hash, watch] of byPeer) { if (watch.expiresAt <= now) { watchers.delete(hash); continue; } const entries = cached .filter((candidate) => candidate.hash !== hash) .slice(0, Math.max(1, watch.want)) .map((candidate) => ({ hash: candidate.hash, addrs: candidate.addrs.map((a) => a.bytes), expiresAt: now + 60_000, })); void this._sendControl(hash, this.codec.encodeProviderNotify(id.key, entries)).catch( dontThrowIfDeliveryError, ); } this.pruneProviderWatchersIfEmpty(id.suffixKey); } private getProviderNamespaceId(namespace: string): ProviderNamespaceId { const key = sha256Sync(textEncoder.encode(`provider|${namespace}`)); const suffixKey = toBase64(key.subarray(0, 24)); this.providerNamespaceBySuffixKey.set(suffixKey, namespace); return { namespace, key, suffixKey }; } private getProviderNamespaceIdFromKey( key: Uint8Array, suffixKey = toBase64(key.subarray(0, 24)), ): ProviderNamespaceId { return { namespace: this.providerNamespaceBySuffixKey.get(suffixKey) ?? "", key, suffixKey, }; } public provide( namespace: string, options: FanoutProviderAnnounceOptions = {}, ): FanoutProviderHandle { if (!this.started) { throw new Error("FanoutTree must be started before providing"); } const id = this.getProviderNamespaceId(namespace); const ttlMsRaw = Math.max(0, Math.floor(options.ttlMs ?? 120_000)); const ttlMs = Math.min(ttlMsRaw, 120_000); const announceIntervalMsRaw = options.announceIntervalMs ?? Math.max(1_000, Math.floor(ttlMs / 2)); const announceIntervalMs = Math.max(100, Math.floor(announceIntervalMsRaw)); const bootstrapOverride = options.bootstrap && options.bootstrap.length > 0 ? options.bootstrap .map((a) => (typeof a === "string" ? multiaddr(a) : a)) .filter((a) => Boolean(a)) : undefined; const bootstrapDialTimeoutMs = Math.max( 0, Math.floor(options.bootstrapDialTimeoutMs ?? 2_000), ); const bootstrapMaxPeers = Math.max( 0, Math.floor(options.bootstrapMaxPeers ?? 0), ); let state = this.providerAnnounceBySuffixKey.get(id.suffixKey); if (!state) { state = { id, ttlMs, announceIntervalMs, bootstrapOverride, bootstrapDialTimeoutMs, bootstrapMaxPeers, closed: false, }; this.providerAnnounceBySuffixKey.set(id.suffixKey, state); state.loop = this._providerAnnounceLoop(state).catch(() => {}); } else { state.ttlMs = ttlMs; state.announceIntervalMs = announceIntervalMs; state.bootstrapOverride = bootstrapOverride; state.bootstrapDialTimeoutMs = bootstrapDialTimeoutMs; state.bootstrapMaxPeers = bootstrapMaxPeers; } return { close: () => { const current = this.providerAnnounceBySuffixKey.get(id.suffixKey); if (!current) return; current.closed = true; this.providerAnnounceBySuffixKey.delete(id.suffixKey); // Best-effort immediate withdrawal (ttl=0). void this.announceProviderOnce( current, this.closeController.signal, 0, ).catch(() => {}); }, }; } /** * Announce provider presence once (no background loop). * * This is useful for "on-demand" discovery where the caller wants to publish a * short-lived provider hint (e.g. after putting a block) without keeping a * per-namespace timer alive. */ public async announceProvider( namespace: string, options: Omit = {}, ): Promise { if (!this.started) { throw new Error("FanoutTree must be started before providing"); } const id = this.getProviderNamespaceId(namespace); const ttlMsRaw = Math.max(0, Math.floor(options.ttlMs ?? 120_000)); const ttlMs = Math.min(ttlMsRaw, 120_000); const bootstrapOverride = options.bootstrap && options.bootstrap.length > 0 ? options.bootstrap .map((a) => (typeof a === "string" ? multiaddr(a) : a)) .filter((a) => Boolean(a)) : undefined; const bootstrapDialTimeoutMs = Math.max( 0, Math.floor(options.bootstrapDialTimeoutMs ?? 2_000), ); const bootstrapMaxPeers = Math.max( 0, Math.floor(options.bootstrapMaxPeers ?? 0), ); const state: ProviderAnnounceState = { id, ttlMs, // ignored by `announceProviderOnce` announceIntervalMs: 0, bootstrapOverride, bootstrapDialTimeoutMs, bootstrapMaxPeers, closed: false, }; await this.announceProviderOnce(state, this.closeController.signal, ttlMs); } private async _providerAnnounceLoop( state: ProviderAnnounceState, ): Promise { const signal = this.closeController.signal; for (;;) { if (signal.aborted || state.closed) return; try { await this.announceProviderOnce(state, signal); } catch { // ignore } await delay(Math.max(50, state.announceIntervalMs)); } } private async announceProviderOnce( state: ProviderAnnounceState, signal: AbortSignal, ttlOverrideMs?: number, ): Promise { const bootstraps = state.bootstrapOverride ?? this.bootstraps; if (bootstraps.length === 0) return; const peers = await this.ensureBootstrapPeers( bootstraps, state.bootstrapDialTimeoutMs, signal, state.bootstrapMaxPeers, ); if (peers.length === 0) return; const ttlMs = ttlOverrideMs != null ? ttlOverrideMs : state.ttlMs; const addrs = this.getSelfAnnounceAddrs(); const bytes = this.codec.encodeProviderAnnounce(state.id.key, ttlMs, addrs); await this._sendControlMany(peers, bytes); } private nextProviderReqId(suffixKey: string): number { const pending = this.pendingProviderQueryBySuffixKey.get(suffixKey); let reqId = (this.random() * 0xffffffff) >>> 0; while (pending?.has(reqId)) { reqId = (this.random() * 0xffffffff) >>> 0; } return reqId; } public async queryProviderCandidates( namespace: string, options: FanoutProviderQueryOptions = {}, ): Promise { if (!this.started) { throw new Error("FanoutTree must be started before querying providers"); } const want = Math.max(0, Math.floor(options.want ?? 8)); if (want === 0) return []; const id = this.getProviderNamespaceId(namespace); const seed = options.seed != null ? options.seed >>> 0 : 0; const cacheTtlMs = Math.max(0, Math.floor(options.cacheTtlMs ?? 60_000)); const signal = options.signal ? anySignal([this.closeController.signal, options.signal]) : this.closeController.signal; try { const now = Date.now(); this.touchProviderNamespace(id.suffixKey, now); const shuffleInPlace = ( arr: FanoutProviderCandidate[], seed32: number, ) => { if (arr.length <= 1) return; if (seed32 >>> 0 === 0) { for (let i = arr.length - 1; i > 0; i--) { const j = Math.floor(this.random() * (i + 1)); const tmp = arr[i]!; arr[i] = arr[j]!; arr[j] = tmp; } return; } let x = seed32 >>> 0; for (let i = arr.length - 1; i > 0; i--) { // xorshift32 x ^= x << 13; x ^= x >>> 17; x ^= x << 5; const j = (x >>> 0) % (i + 1); const tmp = arr[i]!; arr[i] = arr[j]!; arr[j] = tmp; } }; const orderCandidates = (items: FanoutProviderCandidate[]) => { // Prefer already-connected providers, but still spread load within each group. const connected: FanoutProviderCandidate[] = []; const unconnected: FanoutProviderCandidate[] = []; for (const c of items) { (this.peers.get(c.hash) ? connected : unconnected).push(c); } shuffleInPlace(connected, seed); shuffleInPlace(unconnected, seed ? (seed ^ 0x9e3779b9) >>> 0 : 0); return connected.concat(unconnected).slice(0, want); }; const cached = this.getProviderCandidatesFromCache(id, { now }); // If the cache is warm, avoid tracker queries on hot paths. if (cached.length >= want) { return orderCandidates(cached); } const bootstrapOverride = options.bootstrap && options.bootstrap.length > 0 ? options.bootstrap .map((a) => (typeof a === "string" ? multiaddr(a) : a)) .filter((a) => Boolean(a)) : undefined; const bootstraps = bootstrapOverride ?? this.bootstraps; const dialTimeoutMs = Math.max( 0, Math.floor(options.bootstrapDialTimeoutMs ?? 2_000), ); const bootstrapMaxPeers = Math.max( 0, Math.floor(options.bootstrapMaxPeers ?? 0), ); const trackerPeers = bootstraps.length > 0 ? await this.ensureBootstrapPeers( bootstraps, dialTimeoutMs, signal, bootstrapMaxPeers, ) : []; const perTrackerTimeout = Math.max( 0, Math.floor(options.queryTimeoutMs ?? 1_000), ); const overallTimeout = Math.max( 0, Math.floor(options.timeoutMs ?? 8_000), ); const deadlineAt = overallTimeout > 0 ? Date.now() + overallTimeout : 0; const byReq = this.pendingProviderQueryBySuffixKey.get(id.suffixKey) || new Map< number, { resolve: (providers: FanoutProviderCandidate[]) => void } >(); this.pendingProviderQueryBySuffixKey.set(id.suffixKey, byReq); const results = await Promise.all( trackerPeers.map(async (trackerHash) => { const reqId = this.nextProviderReqId(id.suffixKey); const p = new Promise((resolve) => { byReq.set(reqId, { resolve }); }); void this._sendControl( trackerHash, this.codec.encodeProviderQuery(id.key, reqId, want, seed), ); const remainingMs = deadlineAt > 0 ? Math.max(0, deadlineAt - Date.now()) : perTrackerTimeout; const timeoutMs = deadlineAt > 0 ? Math.min(perTrackerTimeout, remainingMs) : perTrackerTimeout; const res = await Promise.race([ p, delay(timeoutMs, { signal }).then((): null => null), ]); if (res == null) { byReq.delete(reqId); return []; } return res; }), ); const merged: FanoutProviderCandidate[] = [...cached]; for (const r of results) merged.push(...r); const seen = new Set(); const deduped = merged.filter((c) => { if (!c.hash) return false; if (c.hash === this.publicKeyHash) return false; if (seen.has(c.hash)) return false; seen.add(c.hash); return true; }); // Cache (best-effort) to avoid repeated tracker lookups. if (cacheTtlMs > 0) { this.rememberProviderCandidates(id, deduped, Date.now() + cacheTtlMs); } const ordered = orderCandidates(deduped); if (ordered.length > 0) { this.emitProviderUpdate(id, ordered); } return ordered; } finally { (signal as any)?.clear?.(); } } public async queryProviders( namespace: string, options: FanoutProviderQueryOptions = {}, ): Promise { const candidates = await this.queryProviderCandidates(namespace, options); return candidates.map((c) => c.hash); } public watchProviders( namespace: string, options: FanoutProviderWatchOptions, ): FanoutProviderHandle { if (!this.started) { throw new Error("FanoutTree must be started before watching providers"); } const id = this.getProviderNamespaceId(namespace); const ttlMs = Math.max(1_000, Math.floor(options.ttlMs ?? 10_000)); const renewIntervalMs = Math.max( 250, Math.min( ttlMs, Math.floor(options.renewIntervalMs ?? Math.max(500, ttlMs / 2)), ), ); const watch: ProviderWatchState = { id, want: Math.max(1, Math.floor(options.want ?? 8)), ttlMs, renewIntervalMs, bootstrapOverride: options.bootstrap && options.bootstrap.length > 0 ? options.bootstrap .map((a) => (typeof a === "string" ? multiaddr(a) : a)) .filter((a) => Boolean(a)) : undefined, bootstrapDialTimeoutMs: Math.max( 0, Math.floor(options.bootstrapDialTimeoutMs ?? 2_000), ), bootstrapMaxPeers: Math.max( 0, Math.floor(options.bootstrapMaxPeers ?? 0), ), onProviders: options.onProviders, signal: options.signal, closed: false, trackerPeers: [], }; let localWatches = this.providerWatchesBySuffixKey.get(id.suffixKey); if (!localWatches) { localWatches = new Set(); this.providerWatchesBySuffixKey.set(id.suffixKey, localWatches); } localWatches.add(watch); const cached = this.getProviderCandidatesFromCache(id, { want: watch.want, }); if (cached.length > 0) { queueMicrotask(() => { if (watch.closed) return; try { watch.onProviders(cached.slice(0, watch.want)); } catch { // ignore consumer callback failures } }); } const close = () => { if (watch.closed) return; watch.closed = true; localWatches?.delete(watch); if (localWatches && localWatches.size === 0) { this.providerWatchesBySuffixKey.delete(id.suffixKey); } for (const trackerHash of watch.trackerPeers) { void this._sendControl( trackerHash, this.codec.encodeProviderUnsubscribe(id.key), ).catch(dontThrowIfDeliveryError); } watch.trackerPeers = []; }; const onAbort = () => close(); watch.signal?.addEventListener("abort", onAbort, { once: true }); const loopSignal = watch.signal ? anySignal([this.closeController.signal, watch.signal]) : this.closeController.signal; watch.loop = (async () => { for (;;) { if (watch.closed || loopSignal.aborted) return; try { const bootstraps = watch.bootstrapOverride ?? this.bootstraps; const trackerPeers = bootstraps.length > 0 ? await this.ensureBootstrapPeers( bootstraps, watch.bootstrapDialTimeoutMs, loopSignal, watch.bootstrapMaxPeers, ) : []; watch.trackerPeers = trackerPeers; for (const trackerHash of trackerPeers) { void this._sendControl( trackerHash, this.codec.encodeProviderSubscribe(id.key, watch.want, watch.ttlMs), ).catch(dontThrowIfDeliveryError); } } catch (error) { if ( error instanceof AbortError || (error && typeof error === "object" && "name" in error && error.name === "AbortError") ) { return; } } await delay(watch.renewIntervalMs, { signal: loopSignal }).catch( () => {}, ); } })(); void watch.loop.finally(() => { watch.signal?.removeEventListener("abort", onAbort); close(); (loopSignal as any)?.clear?.(); }); return { close }; } public getChannelId(topic: string, root: string): FanoutTreeChannelId { const key = sha256Sync(textEncoder.encode(`fanout-tree|${root}|${topic}`)); const suffixKey = toBase64(key.subarray(0, 24)); return { topic, root, key, suffixKey }; } public openChannel( topic: string, root: string, opts: FanoutTreeChannelOptions, ) { if (!this.started) { throw new Error("FanoutTree must be started before opening channels"); } const id = this.getChannelId(topic, root); let ch = this.channelsBySuffixKey.get(id.suffixKey); if (ch) return ch.id; const bidPerByte = Math.max(0, Math.floor(opts.bidPerByte ?? 0)) >>> 0; const uploadLimitBps = Math.max(0, Math.floor(opts.uploadLimitBps)); const maxChildren = Math.max(0, Math.floor(opts.maxChildren)); const msgRate = Math.max(1, Math.floor(opts.msgRate)); const msgSize = Math.max(1, Math.floor(opts.msgSize)); const uploadOverheadBytes = Math.max( 0, Math.floor(opts.uploadOverheadBytes ?? this.defaultUploadOverheadBytes), ); const perChildBytes = Math.max(1, 1 + msgSize + uploadOverheadBytes); const perChildBps = Math.max(1, Math.floor(msgRate * perChildBytes)); const byBps = uploadLimitBps > 0 ? Math.floor(uploadLimitBps / perChildBps) : 0; const effectiveMaxChildren = Math.max(0, Math.min(maxChildren, byBps)); const maxDataAgeMs = Math.max(0, Math.floor(opts.maxDataAgeMs ?? 0)); const requestedRouteCacheMaxEntries = Math.max( 0, Math.floor( opts.routeCacheMaxEntries ?? (opts.role === "root" ? 2_048 : 512), ), ); const routeCacheMaxEntries = Math.min( ROUTE_CACHE_MAX_ENTRIES_HARD_CAP, requestedRouteCacheMaxEntries, ); const routeCacheTtlMs = Math.max( 0, Math.floor(opts.routeCacheTtlMs ?? 10 * 60_000), ); const requestedPeerHintMaxEntries = Math.max( 0, Math.floor( opts.peerHintMaxEntries ?? (opts.role === "root" ? 4_096 : 1_024), ), ); const peerHintMaxEntries = Math.min( PEER_HINT_MAX_ENTRIES_HARD_CAP, requestedPeerHintMaxEntries, ); const peerHintTtlMs = Math.max( 0, Math.floor(opts.peerHintTtlMs ?? 10 * 60_000), ); const uploadBurstMsDefault = Math.max(1, Math.ceil(1_000 / msgRate)); const uploadBurstMs = Math.max( 0, Math.floor(opts.uploadBurstMs ?? uploadBurstMsDefault), ); const uploadTokenCapacity = uploadLimitBps > 0 && uploadBurstMs > 0 ? Math.max(1, Math.floor((uploadLimitBps * uploadBurstMs) / 1_000)) : 0; const repairEnabled = opts.repair !== false; const repairWindowMessages = Math.max( 0, Math.floor(opts.repairWindowMessages ?? 1024), ); const repairMaxBackfillMessagesRaw = Math.max( 0, Math.floor(opts.repairMaxBackfillMessages ?? repairWindowMessages), ); const repairMaxBackfillMessages = repairEnabled && repairWindowMessages > 0 ? Math.min(repairMaxBackfillMessagesRaw, repairWindowMessages) : 0; const repairIntervalMs = Math.max( 0, Math.floor(opts.repairIntervalMs ?? 200), ); const repairMaxPerReq = Math.max(0, Math.floor(opts.repairMaxPerReq ?? 64)); const neighborRepair = opts.neighborRepair === true; const neighborRepairPeers = Math.max( 0, Math.floor(opts.neighborRepairPeers ?? 2), ); const neighborMeshPeers = Math.max( 0, Math.floor( opts.neighborMeshPeers ?? Math.max(0, neighborRepairPeers * 2), ), ); const neighborAnnounceIntervalMs = Math.max( 0, Math.floor(opts.neighborAnnounceIntervalMs ?? 800), ); const neighborMeshRefreshIntervalMs = Math.max( 0, Math.floor(opts.neighborMeshRefreshIntervalMs ?? 10_000), ); const neighborHaveTtlMs = Math.max( 0, Math.floor(opts.neighborHaveTtlMs ?? 8_000), ); const neighborRepairBudgetBps = Math.max( 0, Math.floor(opts.neighborRepairBudgetBps ?? 0), ); const neighborRepairBurstMs = Math.max( 0, Math.floor(opts.neighborRepairBurstMs ?? 1_000), ); const neighborRepairTokenCapacity = neighborRepairBudgetBps > 0 && neighborRepairBurstMs > 0 ? Math.max( 1, Math.floor( (neighborRepairBudgetBps * neighborRepairBurstMs) / 1_000, ), ) : 0; const proxyPublishBudgetBps = Math.max( 0, Math.floor(opts.proxyPublishBudgetBps ?? perChildBps), ); const proxyPublishBurstMs = Math.max( 0, Math.floor(opts.proxyPublishBurstMs ?? 1_000), ); const proxyPublishTokenCapacity = proxyPublishBudgetBps > 0 && proxyPublishBurstMs > 0 ? Math.max( 1, Math.floor((proxyPublishBudgetBps * proxyPublishBurstMs) / 1_000), ) : 0; const unicastBudgetBps = Math.max( 0, Math.floor(opts.unicastBudgetBps ?? perChildBps), ); const unicastBurstMs = Math.max( 0, Math.floor(opts.unicastBurstMs ?? 1_000), ); const unicastTokenCapacity = unicastBudgetBps > 0 && unicastBurstMs > 0 ? Math.max(1, Math.floor((unicastBudgetBps * unicastBurstMs) / 1_000)) : 0; ch = { id, metrics: this.getMetricsForSuffixKey(id.suffixKey), level: opts.role === "root" ? 0 : Number.POSITIVE_INFINITY, isRoot: opts.role === "root", closed: false, parent: undefined, children: new Map(), rejoinCooldownUntil: 0, parentUpgradeCount: 0, parentUpgradeLastAt: 0, parentUpgradeBackoffMs: 0, parentUpgradeBackoffUntil: 0, parentUpgradeRetryAfterSeq: -1, parentUpgradeStaleRootProbeRound: 0, parentUpgradeTrackerNoCapacityUntil: 0, parentProbeRejectBackoffMsByHash: new Map(), joinedAtLeastOnce: opts.role === "root", routeFromRoot: opts.role === "root" ? [this.publicKeyHash] : undefined, routeByPeer: new Map(), seq: 0, bidPerByte, uploadLimitBps, maxChildren, effectiveMaxChildren, allowKick: opts.allowKick === true, maxDataAgeMs, repairEnabled, repairWindowMessages, repairMaxBackfillMessages, repairIntervalMs, repairMaxPerReq, neighborRepair, neighborRepairPeers, neighborMeshPeers, neighborAnnounceIntervalMs, neighborMeshRefreshIntervalMs, neighborHaveTtlMs, neighborRepairBudgetBps, neighborRepairTokenCapacity, neighborRepairTokens: neighborRepairTokenCapacity, neighborRepairLastRefillAt: Date.now(), proxyPublishBudgetBps, proxyPublishTokenCapacity, proxyPublishTokensByPeer: new Map(), unicastBudgetBps, unicastTokenCapacity, unicastTokensByPeer: new Map(), uploadOverheadBytes, uploadBurstMs, uploadTokenCapacity, uploadTokens: uploadTokenCapacity, uploadLastRefillAt: Date.now(), droppedForwards: 0, overloadStreak: 0, lastOverloadKickAt: 0, dataWriteFailStreakByChild: new Map(), lastDataWriteFailKickAt: 0, cacheSeqs: repairEnabled && repairWindowMessages > 0 ? new Int32Array(repairWindowMessages).fill(-1) : undefined, cachePayloads: repairEnabled && repairWindowMessages > 0 ? new Array(repairWindowMessages) : undefined, nextExpectedSeq: 0, missingSeqs: new Set(), repairRequestedAtBySeq: new Map(), endSeqExclusive: -1, lastRepairSentAt: 0, parentRepairUnansweredStreak: 0, lastParentDataAt: 0, lastParentUpgradeActivityAt: 0, receivedAnyParentData: false, parentDataLatencySamples: 0, parentDataLatencyEwmaMs: 0, parentDataLatencyMaxMs: 0, channelPeers: new Map(), knownCandidateAddrs: new Map(), lazyPeers: new Set(), haveByPeer: new Map(), maxSeqSeen: -1, lastIHaveSentAt: 0, cachedAnnounceTrackerPeers: undefined, lastIHaveSentMaxSeq: -1, pendingJoin: new Map(), pendingTrackerQuery: new Map(), pendingParentProbe: new Map(), parentUpgradeReservationsByHash: new Map(), parentProbeRejectUntilByHash: new Map(), pendingRouteQuery: new Map(), pendingUnicastAck: new Map(), pendingRouteProxy: new Map(), routeProxyByTarget: new Map(), bootstrapOverride: undefined, bootstrapDialTimeoutMs: 10_000, bootstrapMaxPeers: 0, bootstrapEnsureIntervalMs: 5_000, cachedBootstrapPeers: [], lastBootstrapEnsureAt: 0, announceIntervalMs: 2_000, announceTtlMs: 10_000, lastAnnouncedAt: 0, peerHintMaxEntries, peerHintTtlMs, routeCacheMaxEntries, routeCacheTtlMs, trackerQueryIntervalMs: 2_000, cachedTrackerCandidates: [], lastTrackerQueryAt: 0, }; this.channelsBySuffixKey.set(id.suffixKey, ch); const needsAnnounceLoop = ch.effectiveMaxChildren > 0; if (needsAnnounceLoop) { ch.announceLoop = this._announceLoop(ch).catch(() => {}); } if (ch.repairEnabled && !ch.isRoot && ch.repairIntervalMs >= 0) { ch.repairLoop = this._repairLoop(ch).catch(() => {}); } if ( ch.repairEnabled && ch.neighborRepair && !ch.isRoot && ch.neighborMeshPeers > 0 ) { ch.meshLoop = this._meshLoop(ch).catch(() => {}); } return id; } private abortPendingUnicastAcks(ch: ChannelState, error: AbortError) { for (const pending of ch.pendingUnicastAck.values()) { try { clearTimeout(pending.timer); } catch { // ignore } try { if (pending.signal && pending.onAbort) { pending.signal.removeEventListener("abort", pending.onAbort); } } catch { // ignore } try { pending.reject(error); } catch { // ignore } } ch.pendingUnicastAck.clear(); } private resetParentDataLatency(ch: ChannelState) { ch.parentDataLatencySamples = 0; ch.parentDataLatencyEwmaMs = 0; ch.parentDataLatencyMaxMs = 0; } private noteParentDataLatency(ch: ChannelState, timestamp: bigint) { const sentAt = Number(timestamp); if (!Number.isFinite(sentAt) || sentAt <= 0) return; const latencyMs = Date.now() - sentAt; if (!Number.isFinite(latencyMs) || latencyMs < 0 || latencyMs > 60_000) { return; } const samples = Math.min(0xffff, ch.parentDataLatencySamples + 1); ch.parentDataLatencySamples = samples; ch.parentDataLatencyEwmaMs = samples <= 1 ? latencyMs : ch.parentDataLatencyEwmaMs * 0.875 + latencyMs * 0.125; ch.parentDataLatencyMaxMs = Math.max(ch.parentDataLatencyMaxMs, latencyMs); } private clearParentUpgradeGrace( ch: ChannelState, notifyPreviousParent = false, notifyCandidateParent = false, ): Array> { const grace = ch.parentUpgradeGrace; if (!grace) return []; clearTimeout(grace.timer); ch.parentUpgradeGrace = undefined; const sends: Array> = []; if (notifyPreviousParent && grace.previousParent !== ch.parent) { sends.push( this._sendControl(grace.previousParent, this.codec.encodeLeave(ch.id.key)).catch( () => {}, ), ); } if (notifyCandidateParent && grace.candidateParent !== ch.parent) { sends.push( this._sendControl(grace.candidateParent, this.codec.encodeLeave(ch.id.key)).catch( () => {}, ), ); } return sends; } private commitParentUpgradeGrace(ch: ChannelState) { const grace = ch.parentUpgradeGrace; if (!grace) return; clearTimeout(grace.timer); ch.parentUpgradeGrace = undefined; if (ch.closed || ch.parent !== grace.candidateParent) return; void this._sendControl(grace.previousParent, this.codec.encodeLeave(ch.id.key)).catch( () => {}, ); } private rollbackParentUpgradeGrace(ch: ChannelState) { const grace = ch.parentUpgradeGrace; if (!grace) return; clearTimeout(grace.timer); ch.parentUpgradeGrace = undefined; if (ch.closed || ch.parent !== grace.candidateParent) return; ch.parent = grace.previousParent; ch.level = grace.previousLevel; ch.routeFromRoot = grace.previousRouteFromRoot ? [...grace.previousRouteFromRoot] : undefined; ch.lastParentDataAt = grace.previousLastParentDataAt; ch.lastParentUpgradeActivityAt = grace.previousLastParentUpgradeActivityAt; ch.receivedAnyParentData = grace.previousReceivedAnyParentData; this.resetParentDataLatency(ch); this.touchPeerHint(ch, grace.previousParent); void this._sendControl(grace.candidateParent, this.codec.encodeLeave(ch.id.key)).catch( () => {}, ); void this.announceToTrackers(ch, this.closeController.signal).catch( () => {}, ); } private startParentUpgradeGrace( ch: ChannelState, options: Omit & { timeoutMs: number }, ) { if (ch.closed || options.previousParent === options.candidateParent) return; void Promise.all(this.clearParentUpgradeGrace(ch, true, true)).catch( () => {}, ); const timeoutMs = Math.max(0, Math.floor(options.timeoutMs)); if (timeoutMs <= 0) { void this._sendControl( options.previousParent, this.codec.encodeLeave(ch.id.key), ).catch(() => {}); return; } const timer = setTimeout( () => this.rollbackParentUpgradeGrace(ch), timeoutMs, ); ch.parentUpgradeGrace = { previousParent: options.previousParent, candidateParent: options.candidateParent, previousLevel: options.previousLevel, previousRouteFromRoot: options.previousRouteFromRoot ? [...options.previousRouteFromRoot] : undefined, previousLastParentDataAt: options.previousLastParentDataAt, previousLastParentUpgradeActivityAt: options.previousLastParentUpgradeActivityAt, previousReceivedAnyParentData: options.previousReceivedAnyParentData, timer, candidateFirstMessages: options.candidateFirstMessages, previousFirstMessages: options.previousFirstMessages, minCandidateFirstMessages: options.minCandidateFirstMessages, minCandidateAdvantageMessages: options.minCandidateAdvantageMessages, maxPreviousFirstMessages: options.maxPreviousFirstMessages, }; } private detachFromParent(ch: ChannelState) { if (this.parentUpgradeShadowInFlightSuffixKey === ch.id.suffixKey) { this.parentUpgradeShadowInFlightSuffixKey = undefined; } void Promise.all(this.clearParentUpgradeGrace(ch, true, true)).catch( () => {}, ); ch.parent = undefined; ch.level = Number.POSITIVE_INFINITY; ch.routeFromRoot = undefined; ch.routeByPeer.clear(); ch.lastParentDataAt = 0; ch.lastParentUpgradeActivityAt = 0; ch.receivedAnyParentData = false; this.resetParentDataLatency(ch); ch.pendingJoin.clear(); ch.pendingParentProbe.clear(); ch.parentShadow = undefined; ch.parentUpgradeBackoffMs = 0; ch.parentUpgradeBackoffUntil = 0; ch.parentUpgradeRetryAfterSeq = -1; ch.parentUpgradeStaleRootProbeRound = 0; ch.parentUpgradeTrackerNoCapacityUntil = 0; ch.parentProbeRejectUntilByHash.clear(); ch.parentProbeRejectBackoffMsByHash.clear(); ch.pendingRouteQuery.clear(); this.abortPendingUnicastAcks(ch, new AbortError("fanout channel detached")); this.clearRouteProxies(ch); } private onPeerDisconnectedFromUnderlay(peerHash: string) { if (!peerHash) return; this.joinTimeoutStreakByPeer.delete(peerHash); // Detach from a disconnected parent immediately, so children can rejoin. // This is more reliable than polling `getConnections()` because the underlay // can flap/reconnect faster than the join loop cadence. const now = Date.now(); for (const ch of this.channelsBySuffixKey.values()) { if (!ch.parent) continue; if (ch.parent !== peerHash) continue; if (ch.closed) continue; ch.metrics.reparentDisconnect += 1; const hadChildren = ch.children.size > 0; this.detachFromParent(ch); void this.kickChildren(ch).catch(() => {}); if (hadChildren) { ch.rejoinCooldownUntil = Math.max( ch.rejoinCooldownUntil, now + RELAY_REJOIN_COOLDOWN_MS, ); } } // Also prune disconnected children from rooted nodes to free capacity fast. for (const ch of this.channelsBySuffixKey.values()) { if (ch.children.delete(peerHash)) { ch.dataWriteFailStreakByChild.delete(peerHash); this.dispatchEvent( new CustomEvent("fanout:peer-unreachable", { detail: { topic: ch.id.topic, root: ch.id.root, publicKeyHash: peerHash, }, }), ); } } } public async closeChannel( topic: string, root: string, options?: { notifyParent?: boolean; kickChildren?: boolean }, ): Promise { const id = this.getChannelId(topic, root); const ch = this.channelsBySuffixKey.get(id.suffixKey); if (!ch) return; if (ch.closed) return; ch.closed = true; // If a join is in-flight, surface that it won't complete. try { ch.joinedOnce?.reject(new AbortError("fanout channel closed")); } catch { // ignore } ch.joinedOnce = undefined; const notifyParent = options?.notifyParent !== false; const kickChildren = options?.kickChildren !== false; const pendingSends: Array> = []; if (notifyParent && !ch.isRoot && ch.parent) { pendingSends.push( this._sendControl(ch.parent, this.codec.encodeLeave(ch.id.key)).catch(() => {}), ); } pendingSends.push( ...this.clearParentUpgradeGrace(ch, notifyParent && !ch.isRoot, false), ); if (kickChildren && ch.children.size > 0) { pendingSends.push(this.kickChildren(ch).catch(() => {})); } ch.pendingJoin.clear(); ch.pendingTrackerQuery.clear(); ch.pendingParentProbe.clear(); ch.parentProbeRejectUntilByHash.clear(); ch.parentProbeRejectBackoffMsByHash.clear(); ch.parentShadow = undefined; if (this.parentUpgradeShadowInFlightSuffixKey === ch.id.suffixKey) { this.parentUpgradeShadowInFlightSuffixKey = undefined; } ch.pendingRouteQuery.clear(); this.abortPendingUnicastAcks(ch, new AbortError("fanout channel closed")); this.clearRouteProxies(ch); ch.parent = undefined; ch.children.clear(); ch.lazyPeers.clear(); ch.haveByPeer.clear(); ch.knownCandidateAddrs.clear(); ch.channelPeers.clear(); ch.routeFromRoot = undefined; ch.routeByPeer.clear(); this.channelsBySuffixKey.delete(id.suffixKey); this.trackerBySuffixKey.delete(id.suffixKey); await Promise.all(pendingSends); } public getChannelStats( topic: string, root: string, ): | { topic: string; root: string; parent?: string; level: number; children: number; effectiveMaxChildren: number; uploadLimitBps: number; droppedForwards: number; peerHintEntries: number; peerHintMaxEntries: number; routeCacheEntries: number; routeCacheMaxEntries: number; } | undefined { const id = this.getChannelId(topic, root); const ch = this.channelsBySuffixKey.get(id.suffixKey); if (!ch) return; return { topic, root, parent: ch.parent, level: ch.level, children: ch.children.size, effectiveMaxChildren: ch.effectiveMaxChildren, uploadLimitBps: ch.uploadLimitBps, droppedForwards: ch.droppedForwards, peerHintEntries: ch.channelPeers.size, peerHintMaxEntries: ch.peerHintMaxEntries, routeCacheEntries: ch.routeByPeer.size, routeCacheMaxEntries: ch.routeCacheMaxEntries, }; } /** * Returns diagnostic counters for tests and simulation harnesses. * * @internal */ public getChannelMetrics( topic: string, root: string, ): FanoutTreeChannelMetrics { const id = this.getChannelId(topic, root); return this.getMetricsForSuffixKey(id.suffixKey); } public getChannelPeerHashes( topic: string, root: string, options?: { includeSelf?: boolean }, ): string[] { const id = this.getChannelId(topic, root); const ch = this.channelsBySuffixKey.get(id.suffixKey); if (!ch) return []; const includeSelf = options?.includeSelf === true; const peers = new Set(); if (ch.parent) peers.add(ch.parent); for (const child of ch.children.keys()) peers.add(child); for (const peer of ch.channelPeers.keys()) peers.add(peer); if (ch.routeFromRoot) { for (const peer of ch.routeFromRoot) peers.add(peer); } if (includeSelf) { peers.add(this.publicKeyHash); } else { peers.delete(this.publicKeyHash); } return [...peers]; } public async joinChannel( topic: string, root: string, channelOpts: Omit, joinOpts: FanoutTreeJoinOptions = {}, ): Promise { await ready; const id = this.openChannel(topic, root, { ...channelOpts, role: "node" }); const ch = this.channelsBySuffixKey.get(id.suffixKey)!; if (ch.isRoot) return; if (joinOpts.bootstrap) { ch.bootstrapOverride = joinOpts.bootstrap.map((a) => typeof a === "string" ? multiaddr(a) : a, ); } if (joinOpts.bootstrapDialTimeoutMs != null) { ch.bootstrapDialTimeoutMs = Math.max( 0, Math.floor(joinOpts.bootstrapDialTimeoutMs), ); } if (joinOpts.bootstrapMaxPeers != null) { ch.bootstrapMaxPeers = Math.max( 0, Math.floor(joinOpts.bootstrapMaxPeers), ); } if (joinOpts.announceIntervalMs != null) { ch.announceIntervalMs = Math.max( 0, Math.floor(joinOpts.announceIntervalMs), ); } if (joinOpts.announceTtlMs != null) { ch.announceTtlMs = Math.max(0, Math.floor(joinOpts.announceTtlMs)); } if (joinOpts.bootstrapEnsureIntervalMs != null) { ch.bootstrapEnsureIntervalMs = Math.max( 0, Math.floor(joinOpts.bootstrapEnsureIntervalMs), ); } if (joinOpts.trackerQueryIntervalMs != null) { ch.trackerQueryIntervalMs = Math.max( 0, Math.floor(joinOpts.trackerQueryIntervalMs), ); } if (!ch.joinedOnce) ch.joinedOnce = createDeferred(); if (!ch.joinLoop) { const joinedOnce = ch.joinedOnce; const joinLoop = this._joinLoop(ch, joinOpts) .catch((err) => { // Surface join errors to the caller without crashing the process // via an unhandled rejection (joinLoop is not generally awaited). if (ch.joinLoop === joinLoop) ch.joinLoop = undefined; if (ch.joinedOnce === joinedOnce && !ch.joinedAtLeastOnce) { ch.joinedOnce = undefined; } joinedOnce.reject(err); }) .finally(() => { if (ch.joinLoop === joinLoop) ch.joinLoop = undefined; }); ch.joinLoop = joinLoop; } return ch.joinedOnce.promise; } /** * Returns this node's current route token for a channel, if attached. * * The token is a source-route path `[root, ..., self]` and can be shared * out-of-band to allow economical unicast within the channel. */ public getRouteToken(topic: string, root: string): string[] | undefined { const id = this.getChannelId(topic, root); const ch = this.channelsBySuffixKey.get(id.suffixKey); if (!ch?.routeFromRoot || ch.routeFromRoot.length === 0) return undefined; return [...ch.routeFromRoot]; } public getRouteHint( topic: string, root: string, targetHash: string, ): FanoutRouteTokenHint | undefined { const id = this.getChannelId(topic, root); const ch = this.channelsBySuffixKey.get(id.suffixKey); if (!ch) return undefined; if (targetHash === this.publicKeyHash && ch.routeFromRoot?.length) { return { kind: "fanout-token", root, target: targetHash, route: [...ch.routeFromRoot], updatedAt: Date.now(), }; } if (ch.isRoot && ch.children.has(targetHash)) { return { kind: "fanout-token", root, target: targetHash, route: [root, targetHash], updatedAt: Date.now(), }; } // Speculative probe: a miss here says nothing about cache quality (the // caller falls through to resolveRouteToken, which counts its own miss). const route = this.getCachedRoute(ch, targetHash, { recordMiss: false }); if (!route) return undefined; const entry = ch.routeByPeer.get(targetHash); const updatedAt = entry?.updatedAt ?? Date.now(); return { kind: "fanout-token", root, target: targetHash, route, updatedAt, expiresAt: ch.routeCacheTtlMs > 0 ? updatedAt + ch.routeCacheTtlMs : undefined, }; } private cacheRoute(ch: ChannelState, route: string[]) { if (!route?.length) return; if (route[0] !== ch.id.root) return; const target = route[route.length - 1]; if (!target) return; if (target === this.publicKeyHash) return; if (ch.routeCacheMaxEntries <= 0) return; if (ch.isRoot && route.length === 2 && ch.children.has(target)) return; const now = Date.now(); ch.routeByPeer.delete(target); ch.routeByPeer.set(target, { route: [...route], updatedAt: now }); this.pruneRouteCache(ch, now); } private cacheKnownCandidateAddrs( ch: ChannelState, hash: string, addrs: Multiaddr[], ) { if (!hash) return; if (!addrs || addrs.length === 0) return; // We only need a handful of addresses to dial a candidate. const limited = addrs.length > JOIN_REJECT_REDIRECT_ADDR_MAX ? addrs.slice(0, JOIN_REJECT_REDIRECT_ADDR_MAX) : addrs; ch.knownCandidateAddrs.delete(hash); ch.knownCandidateAddrs.set(hash, limited); while (ch.knownCandidateAddrs.size > KNOWN_CANDIDATE_ADDRS_MAX_ENTRIES) { const oldest = ch.knownCandidateAddrs.keys().next().value as | string | undefined; if (!oldest) break; ch.knownCandidateAddrs.delete(oldest); } } private touchPeerHint(ch: ChannelState, peerHash: string, now = Date.now()) { if (!peerHash) return; if (peerHash === this.publicKeyHash) return; if (ch.peerHintMaxEntries <= 0) return; // LRU touch ch.channelPeers.delete(peerHash); ch.channelPeers.set(peerHash, now); this.prunePeerHints(ch, now); } private prunePeerHints(ch: ChannelState, now = Date.now()) { if (ch.channelPeers.size === 0) return; if (ch.peerHintMaxEntries <= 0) { ch.channelPeers.clear(); return; } // `channelPeers` is maintained as an LRU (touch via delete+set), so the oldest // entry is always first. This makes TTL pruning O(expired) rather than O(N). if (ch.peerHintTtlMs > 0) { for (;;) { const oldest = ch.channelPeers.keys().next().value as | string | undefined; if (!oldest) break; const seenAt = ch.channelPeers.get(oldest) ?? 0; if (now - seenAt <= ch.peerHintTtlMs) break; ch.channelPeers.delete(oldest); } } while (ch.channelPeers.size > ch.peerHintMaxEntries) { const oldest = ch.channelPeers.keys().next().value as string | undefined; if (!oldest) break; ch.channelPeers.delete(oldest); } } private pruneRouteCache(ch: ChannelState, now = Date.now()) { if (ch.routeByPeer.size === 0) return; if (ch.routeCacheMaxEntries <= 0) { const removed = ch.routeByPeer.size; ch.routeByPeer.clear(); if (removed > 0) ch.metrics.routeCacheEvictions += removed; return; } // `routeByPeer` is maintained as an LRU (touch via delete+set), so the oldest // entry is always first. This makes TTL pruning O(expired) rather than O(N). if (ch.routeCacheTtlMs > 0) { for (;;) { const oldest = ch.routeByPeer.keys().next().value as string | undefined; if (!oldest) break; const entry = ch.routeByPeer.get(oldest); if (!entry) { ch.routeByPeer.delete(oldest); continue; } if (now - entry.updatedAt <= ch.routeCacheTtlMs) break; ch.routeByPeer.delete(oldest); ch.metrics.routeCacheExpirations += 1; } } while (ch.routeByPeer.size > ch.routeCacheMaxEntries) { const oldest = ch.routeByPeer.keys().next().value as string | undefined; if (!oldest) break; ch.routeByPeer.delete(oldest); ch.metrics.routeCacheEvictions += 1; } } private getCachedRoute( ch: ChannelState, targetHash: string, opts?: { recordMiss?: boolean }, ): string[] | undefined { const recordMiss = opts?.recordMiss !== false; this.pruneRouteCache(ch); const entry = ch.routeByPeer.get(targetHash); if (!entry) { if (recordMiss) ch.metrics.routeCacheMisses += 1; return undefined; } const now = Date.now(); if (ch.routeCacheTtlMs > 0 && now - entry.updatedAt > ch.routeCacheTtlMs) { ch.routeByPeer.delete(targetHash); if (recordMiss) ch.metrics.routeCacheMisses += 1; ch.metrics.routeCacheExpirations += 1; return undefined; } if (!this.isRouteValidForChannel(ch, entry.route)) { ch.routeByPeer.delete(targetHash); if (recordMiss) ch.metrics.routeCacheMisses += 1; return undefined; } // LRU touch ch.routeByPeer.delete(targetHash); ch.routeByPeer.set(targetHash, { route: entry.route, updatedAt: now }); ch.metrics.routeCacheHits += 1; return [...entry.route]; } private isRouteValidForChannel(ch: ChannelState, route?: string[]) { if (!route || route.length === 0) return false; if (route[0] !== ch.id.root) return false; // Root must be able to forward the first downstream hop immediately. if (ch.isRoot && route.length > 1 && !ch.children.has(route[1]!)) return false; return true; } private nextReqId(ch: ChannelState): number { let reqId = (this.random() * 0xffffffff) >>> 0; while ( ch.pendingJoin.has(reqId) || ch.pendingTrackerQuery.has(reqId) || ch.pendingRouteQuery.has(reqId) || ch.pendingRouteProxy.has(reqId) ) { reqId = (this.random() * 0xffffffff) >>> 0; } return reqId; } private completeRouteProxy( ch: ChannelState, proxyReqId: number, route?: string[], ) { const proxy = ch.pendingRouteProxy.get(proxyReqId); if (!proxy) return; ch.pendingRouteProxy.delete(proxyReqId); const targetKey = `${proxy.direction}:${proxy.targetHash}`; if (ch.routeProxyByTarget.get(targetKey) === proxyReqId) { ch.routeProxyByTarget.delete(targetKey); } clearTimeout(proxy.timer); const reply = (requester: string, downstreamReqId: number): void => { void this._sendControl( requester, this.codec.encodeRouteReply(ch.id.key, downstreamReqId, route), ).catch(() => {}); }; if (proxy.localResolve) { proxy.localResolve(route); } else { reply(proxy.requester, proxy.downstreamReqId); } if (proxy.waiters) { for (const w of proxy.waiters) { if ("localResolve" in w) { w.localResolve(route); } else { reply(w.requester, w.downstreamReqId); } } } } /** * Coalesce onto an in-flight same-target/same-direction proxy search, if one * exists. Returns true when the requester was attached as a waiter (or * answered empty on waiter overflow) and no new search should start. */ private coalesceRouteProxy( ch: ChannelState, targetHash: string, direction: "up" | "down", waiter: | { requester: string; downstreamReqId: number } | { localResolve: (route?: string[]) => void }, ): boolean { const targetKey = `${direction}:${targetHash}`; const existingId = ch.routeProxyByTarget.get(targetKey); if (existingId == null) return false; const existing = ch.pendingRouteProxy.get(existingId); if (!existing) { ch.routeProxyByTarget.delete(targetKey); return false; } if ((existing.waiters?.length ?? 0) >= ROUTE_PROXY_MAX_WAITERS) { ch.metrics.routeProxyRejected += 1; if ("localResolve" in waiter) { waiter.localResolve(undefined); } else { void this._sendControl( waiter.requester, this.codec.encodeRouteReply(ch.id.key, waiter.downstreamReqId), ).catch(() => {}); } return true; } (existing.waiters ??= []).push(waiter); ch.metrics.routeProxyCoalesced += 1; return true; } /** * Drop all in-flight route-proxy state for a channel (detach/kick/close). * Local resolvers are settled with `undefined` so root-origin * `resolveRouteToken()` callers never hang; remote requesters time out on * their own deadlines, as before. */ private clearRouteProxies(ch: ChannelState) { for (const pending of ch.pendingRouteProxy.values()) { clearTimeout(pending.timer); pending.localResolve?.(undefined); if (pending.waiters) { for (const w of pending.waiters) { if ("localResolve" in w) w.localResolve(undefined); } } } ch.pendingRouteProxy.clear(); ch.routeProxyByTarget.clear(); } private proxyRouteQuery( ch: ChannelState, requester: string, downstreamReqId: number, targetHash: string, candidates: string[], timeoutMs = ROUTE_PROXY_TIMEOUT_MS, ) { // Direction matters for coalescing correctness: an upstream lookup and the // root's downstream flood for the same target may legitimately transit the // same node concurrently (e.g. the target lives in the requester's own // branch), and merging them would make them wait on each other. const direction: "up" | "down" = candidates.length === 1 && candidates[0] === ch.parent ? "up" : "down"; if ( this.coalesceRouteProxy(ch, targetHash, direction, { requester, downstreamReqId, }) ) { return; } const unique: string[] = []; const seen = new Set(); for (const candidate of candidates) { if (!candidate || seen.has(candidate)) continue; seen.add(candidate); const stream = this.peers.get(candidate); if (!stream || !stream.isWritable) continue; unique.push(candidate); } if (unique.length === 0) { void this._sendControl( requester, this.codec.encodeRouteReply(ch.id.key, downstreamReqId), ).catch(() => {}); return; } if (ch.pendingRouteProxy.size >= ROUTE_PROXY_MAX_PENDING) { ch.metrics.routeProxyRejected += 1; void this._sendControl( requester, this.codec.encodeRouteReply(ch.id.key, downstreamReqId), ).catch(() => {}); return; } ch.metrics.routeProxyQueries += 1; ch.metrics.routeProxyFanout += unique.length; const proxyReqId = this.nextReqId(ch); const timer = setTimeout( () => { ch.metrics.routeProxyTimeouts += 1; this.completeRouteProxy(ch, proxyReqId); }, Math.max(1, timeoutMs), ); ch.pendingRouteProxy.set(proxyReqId, { requester, downstreamReqId, timer, expectedReplies: new Set(unique), targetHash, direction, }); ch.routeProxyByTarget.set(`${direction}:${targetHash}`, proxyReqId); void this._sendControlMany( unique, this.codec.encodeRouteQuery(ch.id.key, proxyReqId, targetHash), ).catch(() => { this.completeRouteProxy(ch, proxyReqId); }); } public async resolveRouteToken( topic: string, root: string, targetHash: string, options?: { timeoutMs?: number; signal?: AbortSignal }, ): Promise { const id = this.getChannelId(topic, root); const ch = this.channelsBySuffixKey.get(id.suffixKey); if (!ch) throw new Error(`Channel not open: ${topic} (${root})`); if (!targetHash) throw new Error("targetHash is required"); const cached = this.getCachedRoute(ch, targetHash); if (cached) { return cached; } if (targetHash === this.publicKeyHash && ch.routeFromRoot) { return [...ch.routeFromRoot]; } if (ch.isRoot) { if (ch.children.has(targetHash)) { const route = [ch.id.root, targetHash]; this.cacheRoute(ch, route); return route; } const candidates = [...ch.children.keys()]; if (candidates.length === 0) return undefined; const timeoutMs = Math.max(1, Math.floor(options?.timeoutMs ?? 3_000)); return await new Promise((resolve, reject) => { let settled = false; let proxyReqId: number | undefined; const onAbort = () => { if (settled) return; settled = true; clearTimeout(backstop); if (proxyReqId != null) { // Tear the search down only if nobody else coalesced onto it; // otherwise let it finish for the waiters (our localResolve // is a no-op once settled). const entry = ch.pendingRouteProxy.get(proxyReqId); if (entry && (!entry.waiters || entry.waiters.length === 0)) { this.completeRouteProxy(ch, proxyReqId); } } reject(new AbortError()); }; const localResolve = (route?: string[]) => { if (settled) return; settled = true; clearTimeout(backstop); if (options?.signal) { options.signal.removeEventListener("abort", onAbort); } if (this.isRouteValidForChannel(ch, route)) { this.cacheRoute(ch, route!); resolve([...route!]); return; } resolve(undefined); }; // Per-caller timeout: when coalesced, the shared in-flight search // may outlive this caller's own deadline. const backstop = setTimeout(() => localResolve(undefined), timeoutMs); if (options?.signal) { options.signal.addEventListener("abort", onAbort, { once: true }); } if (this.coalesceRouteProxy(ch, targetHash, "down", { localResolve })) { return; } const unique: string[] = []; const seen = new Set(); for (const candidate of candidates) { if (!candidate || seen.has(candidate)) continue; seen.add(candidate); const stream = this.peers.get(candidate); if (!stream || !stream.isWritable) continue; unique.push(candidate); } if (unique.length === 0) { localResolve(undefined); return; } if (ch.pendingRouteProxy.size >= ROUTE_PROXY_MAX_PENDING) { ch.metrics.routeProxyRejected += 1; localResolve(undefined); return; } ch.metrics.routeProxyQueries += 1; ch.metrics.routeProxyFanout += unique.length; proxyReqId = this.nextReqId(ch); const timer = setTimeout(() => { ch.metrics.routeProxyTimeouts += 1; this.completeRouteProxy(ch, proxyReqId!); }, timeoutMs); ch.pendingRouteProxy.set(proxyReqId, { requester: this.publicKeyHash, downstreamReqId: 0, timer, expectedReplies: new Set(unique), targetHash, direction: "down", localResolve, }); ch.routeProxyByTarget.set(`down:${targetHash}`, proxyReqId); void this._sendControlMany( unique, this.codec.encodeRouteQuery(ch.id.key, proxyReqId, targetHash), ).catch(() => { this.completeRouteProxy(ch, proxyReqId!); }); }); } if (!ch.parent) { throw new Error("Cannot resolve route while not attached to a parent"); } const reqId = this.nextReqId(ch); const timeoutMs = Math.max(1, Math.floor(options?.timeoutMs ?? 3_000)); return await new Promise((resolve, reject) => { let settled = false; const onAbort = () => { if (settled) return; settled = true; ch.pendingRouteQuery.delete(reqId); reject(new AbortError()); }; const timer = setTimeout(() => { if (settled) return; settled = true; ch.pendingRouteQuery.delete(reqId); resolve(undefined); }, timeoutMs); if (options?.signal) { options.signal.addEventListener("abort", onAbort, { once: true }); } ch.pendingRouteQuery.set(reqId, { resolve: (route?: string[]) => { if (settled) return; settled = true; clearTimeout(timer); if (options?.signal) { options.signal.removeEventListener("abort", onAbort); } ch.pendingRouteQuery.delete(reqId); if (this.isRouteValidForChannel(ch, route)) { this.cacheRoute(ch, route!); resolve([...route!]); return; } resolve(undefined); }, }); void this._sendControl( ch.parent!, this.codec.encodeRouteQuery(ch.id.key, reqId, targetHash), ).catch((error) => { if (settled) return; settled = true; clearTimeout(timer); if (options?.signal) { options.signal.removeEventListener("abort", onAbort); } ch.pendingRouteQuery.delete(reqId); reject(error); }); }); } private nextUnicastAckToken(): bigint { const seq = this.unicastAckSeq >>> 0; this.unicastAckSeq = (this.unicastAckSeq + 1) >>> 0; return (BigInt(this.unicastAckNodeTag32) << 32n) | BigInt(seq); } public async unicastTo( topic: string, root: string, targetHash: string, payload: Uint8Array, options?: { timeoutMs?: number; signal?: AbortSignal }, ) { const route = await this.resolveRouteToken( topic, root, targetHash, options, ); if (!route) { throw new Error(`No route token available for target ${targetHash}`); } return this.unicast(topic, root, route, payload); } public async unicastToAck( topic: string, root: string, targetHash: string, payload: Uint8Array, options?: { timeoutMs?: number; signal?: AbortSignal }, ): Promise { const route = await this.resolveRouteToken( topic, root, targetHash, options, ); if (!route) { throw new Error(`No route token available for target ${targetHash}`); } return this.unicastAck(topic, root, route, targetHash, payload, options); } public async unicastAck( topic: string, root: string, toRoute: string[], targetHash: string, payload: Uint8Array, options?: { timeoutMs?: number; signal?: AbortSignal }, ): Promise { const id = this.getChannelId(topic, root); const ch = this.channelsBySuffixKey.get(id.suffixKey); if (!ch) throw new Error(`Channel not open: ${topic} (${root})`); if (!Array.isArray(toRoute) || toRoute.length === 0) { throw new Error("Invalid unicast route token"); } if (toRoute[0] !== ch.id.root) { throw new Error("Unicast route token root mismatch"); } const target = toRoute[toRoute.length - 1]!; if (targetHash && targetHash !== target) { throw new Error("Unicast route token target mismatch"); } if (target === this.publicKeyHash) { const wire = this.codec.encodeUnicast(ch.id.key, toRoute, payload); const message = await this.createMessage(wire, { mode: new AnyWhere(), priority: CONTROL_PRIORITY, } as any); this.dispatchEvent( new CustomEvent("fanout:unicast", { detail: { topic: ch.id.topic, root: ch.id.root, route: [...toRoute], payload, from: this.publicKeyHash, origin: this.publicKeyHash, to: target, timestamp: message.header.timestamp, message, }, }), ); return; } const timeoutMs = Math.max( 1, Math.floor(options?.timeoutMs ?? UNICAST_ACK_DEFAULT_TIMEOUT_MS), ); const signal = options?.signal ? anySignal([this.closeController.signal, options.signal]) : this.closeController.signal; const ackToken = this.nextUnicastAckToken(); const replyRoute = ch.routeFromRoot; if (!replyRoute || replyRoute.length === 0) { throw new Error("Cannot unicast with ACK without a route token to self"); } if (replyRoute[0] !== ch.id.root) { throw new Error( "Cannot unicast with ACK: self route token root mismatch", ); } const selfHash = this.publicKeyHash; if (replyRoute[replyRoute.length - 1] !== selfHash) { throw new Error( "Cannot unicast with ACK: self route token target mismatch", ); } const data = this.codec.encodeUnicast(ch.id.key, toRoute, payload, { ackToken, replyRoute, }); return await new Promise((resolve, reject) => { let settled = false; let timer: ReturnType | undefined; const cleanup = () => { if (timer != null) { clearTimeout(timer); timer = undefined; } const pending = ch.pendingUnicastAck.get(ackToken); if (pending?.signal && pending.onAbort) { pending.signal.removeEventListener("abort", pending.onAbort); } ch.pendingUnicastAck.delete(ackToken); }; const settleOk = () => { if (settled) return; settled = true; cleanup(); resolve(); }; const settleErr = (error: unknown) => { if (settled) return; settled = true; cleanup(); reject(error); }; const onAbort = () => { settleErr(new AbortError()); }; if (signal.aborted) { onAbort(); return; } signal.addEventListener("abort", onAbort); timer = setTimeout(() => { settleErr(new TimeoutError("Timeout waiting for unicast ACK")); }, timeoutMs); ch.pendingUnicastAck.set(ackToken, { expectedOrigin: targetHash, resolve: settleOk, reject: settleErr, timer, signal, onAbort, }); void (async () => { try { if (ch.isRoot) { const nextHop = toRoute[1]; if (!nextHop) { throw new Error("Unicast route token missing first hop"); } if (!ch.children.has(nextHop)) { throw new Error( "Unicast first hop is not a direct child of the root", ); } await this._sendControl(nextHop, data); return; } if (!ch.parent) { throw new Error("Cannot unicast while not attached to a parent"); } await this._sendControl(ch.parent, data); } catch (error) { settleErr(error); } })(); }); } /** * Economical unicast within an existing fanout channel. * * Any sender can send to `toRoute` by forwarding the message *up* to the root, * and then letting the root forward it *down* the provided route. * * `toRoute` must be a route token `[root, ..., target]`. */ public async unicast( topic: string, root: string, toRoute: string[], payload: Uint8Array, ): Promise { const id = this.getChannelId(topic, root); const ch = this.channelsBySuffixKey.get(id.suffixKey); if (!ch) throw new Error(`Channel not open: ${topic} (${root})`); if (!Array.isArray(toRoute) || toRoute.length === 0) { throw new Error("Invalid unicast route token"); } if (toRoute[0] !== ch.id.root) { throw new Error("Unicast route token root mismatch"); } const data = this.codec.encodeUnicast(ch.id.key, toRoute, payload); if (ch.isRoot) { const target = toRoute[toRoute.length - 1]!; if (target === this.publicKeyHash) { const message = await this.createMessage(data, { mode: new AnyWhere(), priority: CONTROL_PRIORITY, } as any); this.dispatchEvent( new CustomEvent("fanout:unicast", { detail: { topic: ch.id.topic, root: ch.id.root, route: [...toRoute], payload, from: this.publicKeyHash, origin: this.publicKeyHash, to: target, timestamp: message.header.timestamp, message, }, }), ); return; } const nextHop = toRoute[1]; if (!nextHop) { throw new Error("Unicast route token missing first hop"); } if (!ch.children.has(nextHop)) { throw new Error("Unicast first hop is not a direct child of the root"); } await this._sendControl(nextHop, data); return; } if (!ch.parent) { throw new Error("Cannot unicast while not attached to a parent"); } await this._sendControl(ch.parent, data); } public async publishData( topic: string, root: string, payload: Uint8Array, ): Promise { const id = this.getChannelId(topic, root); const ch = this.channelsBySuffixKey.get(id.suffixKey); if (!ch) throw new Error(`Channel not open: ${topic} (${root})`); if (!ch.isRoot) throw new Error("Only the channel root can publish"); const seq = ch.seq++; const message = await this._sendData( ch, [...ch.children.keys()], seq, payload, ); this.dispatchEvent( new CustomEvent("fanout:data", { detail: { topic: ch.id.topic, root: ch.id.root, seq, payload, from: this.publicKeyHash, origin: this.publicKeyHash, timestamp: message.header.timestamp, message, }, }), ); } /** * Publishes payload to all channel members. * * Root publishes directly on the data-plane. Non-root members proxy the publish * upstream to the root, which assigns a sequence number and broadcasts. */ public async publishToChannel( topic: string, root: string, payload: Uint8Array, ): Promise { const id = this.getChannelId(topic, root); const ch = this.channelsBySuffixKey.get(id.suffixKey); if (!ch) throw new Error(`Channel not open: ${topic} (${root})`); if (ch.isRoot) { return this.publishData(topic, root, payload); } if (!ch.parent) { // Channels can temporarily detach/re-parent under churn. Proxy publishes should // wait briefly for the join loop to attach instead of hard-throwing (which // can surface as unhandled rejections in higher layers like pubsub/RPC). await this.waitForChannelAttachment(ch, 10_000); } if (!ch.parent) { throw new Error( `Cannot proxy publish while not attached to a parent (topic=${topic} root=${root} self=${this.publicKeyHash})`, ); } await this._sendControl(ch.parent, this.codec.encodePublishProxy(ch.id.key, payload)); } public async publishToChannelMaybe( topic: string, root: string, payload: Uint8Array, ): Promise { const id = this.getChannelId(topic, root); const ch = this.channelsBySuffixKey.get(id.suffixKey); if (!ch || ch.closed) { return false; } try { await this.publishToChannel(topic, root, payload); return true; } catch (error) { if ( error instanceof Error && error.message.startsWith(`Channel not open: ${topic} (${root})`) ) { return false; } dontThrowIfDeliveryError(error); return false; } } private waitForChannelAttachment( ch: ChannelState, timeoutMs: number, ): Promise { if (ch.isRoot || ch.parent) return Promise.resolve(); const ms = Math.max(0, Math.floor(timeoutMs)); if (ms === 0) return Promise.resolve(); const signal = this.closeController.signal; return new Promise((resolve, reject) => { let done = false; let timer: ReturnType | undefined; const resolveIfAttached = () => { if (!ch.parent) return false; cleanup(); resolve(); return true; }; const cleanup = () => { if (done) return; done = true; try { if (timer) clearTimeout(timer); } catch { // ignore } try { this.removeEventListener("fanout:joined", onJoined as any); } catch { // ignore } try { signal.removeEventListener("abort", onAbort); } catch { // ignore } }; const onAbort = () => { cleanup(); reject(signal.reason ?? new AbortError("fanout stopped")); }; const onJoined = (ev: any) => { const d = ev?.detail as { topic: string; root: string } | undefined; if (!d) return; if (d.topic !== ch.id.topic) return; if (d.root !== ch.id.root) return; resolveIfAttached(); }; if (signal.aborted) return onAbort(); this.addEventListener("fanout:joined", onJoined as any); signal.addEventListener("abort", onAbort, { once: true }); if (resolveIfAttached()) return; timer = setTimeout(() => { if (resolveIfAttached()) return; cleanup(); reject( new TimeoutError( `fanout proxy publish timed out waiting for attachment (topic=${ch.id.topic} root=${ch.id.root} self=${this.publicKeyHash})`, ), ); }, ms); timer.unref?.(); }); } public async publishEnd( topic: string, root: string, lastSeqExclusive: number, ): Promise { const id = this.getChannelId(topic, root); const ch = this.channelsBySuffixKey.get(id.suffixKey); if (!ch) return; if (!ch.isRoot) return; ch.endSeqExclusive = Math.max(ch.endSeqExclusive, lastSeqExclusive); await this._sendControlMany( [...ch.children.keys()], this.codec.encodeEnd(ch.id.key, lastSeqExclusive), ); } private makeDataId(ch: ChannelState, seq: number): Uint8Array { const id = new Uint8Array(32); id.set(ID_PREFIX, 0); writeU32BE(id, 4, seq >>> 0); id.set(ch.id.key.subarray(0, 24), 8); return id; } private getSuffixKeyFromId(id: Uint8Array): string { const cached = this.cachedSuffixKey.get(id); if (cached) return cached; const key = toBase64(id.subarray(8, 32)); this.cachedSuffixKey.set(id, key); return key; } private getMetricsForSuffixKey(suffixKey: string): FanoutTreeChannelMetrics { let m = this.metricsBySuffixKey.get(suffixKey); if (!m) { m = createEmptyMetrics(); this.metricsBySuffixKey.set(suffixKey, m); } return m; } private recordControlSend(bytes: Uint8Array, transmissions: number) { if (transmissions <= 0) return; if (bytes.length < 1 + 24) return; const kind = bytes[0]!; const suffixKey = toBase64(bytes.subarray(1, 25)); const m = this.getMetricsForSuffixKey(suffixKey); m.controlSends += transmissions; const sentBytes = bytes.byteLength * transmissions; m.controlBytesSent += sentBytes; switch (kind) { case MSG_JOIN_REQ: m.joinReqSent += transmissions; m.controlBytesSentJoin += sentBytes; break; case MSG_JOIN_ACCEPT: m.joinAcceptSent += transmissions; m.controlBytesSentJoin += sentBytes; break; case MSG_JOIN_REJECT: m.joinRejectSent += transmissions; m.controlBytesSentJoin += sentBytes; break; case MSG_KICK: m.kickSent += transmissions; m.controlBytesSentJoin += sentBytes; break; case MSG_END: m.endSent += transmissions; m.controlBytesSentRepair += sentBytes; break; case MSG_REPAIR_REQ: m.repairReqSent += transmissions; m.controlBytesSentRepair += sentBytes; break; case MSG_FETCH_REQ: m.fetchReqSent += transmissions; m.controlBytesSentRepair += sentBytes; break; case MSG_IHAVE: m.ihaveSent += transmissions; m.controlBytesSentRepair += sentBytes; break; case MSG_TRACKER_ANNOUNCE: m.trackerAnnounceSent += transmissions; m.controlBytesSentTracker += sentBytes; break; case MSG_TRACKER_QUERY: m.trackerQuerySent += transmissions; m.controlBytesSentTracker += sentBytes; break; case MSG_TRACKER_REPLY: m.trackerReplySent += transmissions; m.controlBytesSentTracker += sentBytes; break; case MSG_TRACKER_FEEDBACK: m.trackerFeedbackSent += transmissions; m.controlBytesSentTracker += sentBytes; break; case MSG_PARENT_PROBE_REQ: m.parentProbeReqSent += transmissions; m.controlBytesSentJoin += sentBytes; break; case MSG_PARENT_PROBE_REPLY: m.parentProbeReplySent += transmissions; m.controlBytesSentJoin += sentBytes; break; case MSG_PROVIDER_ANNOUNCE: case MSG_PROVIDER_QUERY: case MSG_PROVIDER_REPLY: case MSG_PROVIDER_SUBSCRIBE: case MSG_PROVIDER_UNSUBSCRIBE: case MSG_PROVIDER_NOTIFY: m.controlBytesSentTracker += sentBytes; break; default: break; } } private recordControlReceive( suffixKey: string, kind: number, bytesReceived: number, ) { const m = this.getMetricsForSuffixKey(suffixKey); m.controlReceives += 1; m.controlBytesReceived += bytesReceived; switch (kind) { case MSG_JOIN_REQ: m.joinReqReceived += 1; m.controlBytesReceivedJoin += bytesReceived; break; case MSG_JOIN_ACCEPT: m.joinAcceptReceived += 1; m.controlBytesReceivedJoin += bytesReceived; break; case MSG_JOIN_REJECT: m.joinRejectReceived += 1; m.controlBytesReceivedJoin += bytesReceived; break; case MSG_KICK: m.kickReceived += 1; m.controlBytesReceivedJoin += bytesReceived; break; case MSG_END: m.endReceived += 1; m.controlBytesReceivedRepair += bytesReceived; break; case MSG_REPAIR_REQ: m.repairReqReceived += 1; m.controlBytesReceivedRepair += bytesReceived; break; case MSG_FETCH_REQ: m.fetchReqReceived += 1; m.controlBytesReceivedRepair += bytesReceived; break; case MSG_IHAVE: m.ihaveReceived += 1; m.controlBytesReceivedRepair += bytesReceived; break; case MSG_TRACKER_ANNOUNCE: m.trackerAnnounceReceived += 1; m.controlBytesReceivedTracker += bytesReceived; break; case MSG_TRACKER_QUERY: m.trackerQueryReceived += 1; m.controlBytesReceivedTracker += bytesReceived; break; case MSG_TRACKER_REPLY: m.trackerReplyReceived += 1; m.controlBytesReceivedTracker += bytesReceived; break; case MSG_TRACKER_FEEDBACK: m.trackerFeedbackReceived += 1; m.controlBytesReceivedTracker += bytesReceived; break; case MSG_PARENT_PROBE_REQ: m.parentProbeReqReceived += 1; m.controlBytesReceivedJoin += bytesReceived; break; case MSG_PARENT_PROBE_REPLY: m.parentProbeReplyReceived += 1; m.controlBytesReceivedJoin += bytesReceived; break; case MSG_PROVIDER_ANNOUNCE: case MSG_PROVIDER_QUERY: case MSG_PROVIDER_REPLY: case MSG_PROVIDER_SUBSCRIBE: case MSG_PROVIDER_UNSUBSCRIBE: case MSG_PROVIDER_NOTIFY: m.controlBytesReceivedTracker += bytesReceived; break; default: break; } } private markCached(ch: ChannelState, seq: number, payload: Uint8Array) { if (!ch.cacheSeqs || !ch.cachePayloads || ch.cacheSeqs.length === 0) return; const idx = seq % ch.cacheSeqs.length; ch.cacheSeqs[idx] = seq | 0; // Copy to avoid accidental mutation by caller. ch.cachePayloads[idx] = payload.slice(); } private getCached(ch: ChannelState, seq: number): Uint8Array | undefined { if (!ch.cacheSeqs || !ch.cachePayloads || ch.cacheSeqs.length === 0) return; const idx = seq % ch.cacheSeqs.length; if (ch.cacheSeqs[idx] !== (seq | 0)) return; return ch.cachePayloads[idx]; } private noteJoinResponse(peerHash: string) { this.joinTimeoutStreakByPeer.delete(peerHash); } private noteJoinTimeout(ch: ChannelState, peerHash: string) { const streak = (this.joinTimeoutStreakByPeer.get(peerHash) ?? 0) + 1; if (streak < JOIN_TIMEOUT_RESET_STREAK_THRESHOLD) { this.joinTimeoutStreakByPeer.set(peerHash, streak); return; } this.joinTimeoutStreakByPeer.delete(peerHash); const now = Date.now(); for (const [hash, until] of this.joinResetCooldownUntilByPeer) { if (until <= now) this.joinResetCooldownUntilByPeer.delete(hash); } const resetCooldownUntil = this.joinResetCooldownUntilByPeer.get(peerHash) ?? 0; if (resetCooldownUntil > now) return; const stream = this.peers.get(peerHash); if (!stream) return; this.joinResetCooldownUntilByPeer.set( peerHash, now + JOIN_TIMEOUT_RESET_COOLDOWN_MS, ); ch.metrics.joinPeerResets += 1; try { void this.components.connectionManager .closeConnections(stream.peerId) .catch(() => {}); } catch { // Best-effort reset. The join loop keeps retrying other candidates. } } private async _sendControl(to: string, bytes: Uint8Array) { const stream = this.peers.get(to); if (!stream) return; this.recordControlSend(bytes, 1); const message = await this.createMessage(bytes, { mode: new AnyWhere(), priority: CONTROL_PRIORITY, } as any); await this.publishMessageMaybe(this.publicKey, message, [stream]); } private async _sendControlMany(to: string[], bytes: Uint8Array) { if (to.length === 0) return true; const streams = to .map((t) => this.peers.get(t)) .filter((s): s is PeerStreams => Boolean(s)); if (streams.length === 0) return false; this.recordControlSend(bytes, streams.length); const message = await this.createMessage(bytes, { mode: new AnyWhere(), priority: CONTROL_PRIORITY, } as any); return this.publishMessageMaybe(this.publicKey, message, streams); } private refillUploadTokens(ch: ChannelState, now = Date.now()) { if (ch.uploadLimitBps <= 0) return; if (ch.uploadTokenCapacity <= 0) return; const elapsedMs = now - ch.uploadLastRefillAt; if (elapsedMs <= 0) return; ch.uploadLastRefillAt = now; ch.uploadTokens = Math.min( ch.uploadTokenCapacity, ch.uploadTokens + (elapsedMs * ch.uploadLimitBps) / 1_000, ); } private refillNeighborRepairTokens(ch: ChannelState, now = Date.now()) { if (ch.neighborRepairBudgetBps <= 0) return; if (ch.neighborRepairTokenCapacity <= 0) return; const elapsedMs = now - ch.neighborRepairLastRefillAt; if (elapsedMs <= 0) return; ch.neighborRepairLastRefillAt = now; ch.neighborRepairTokens = Math.min( ch.neighborRepairTokenCapacity, ch.neighborRepairTokens + (elapsedMs * ch.neighborRepairBudgetBps) / 1_000, ); } private takeIngressBudget( ch: ChannelState, kind: "proxy-publish" | "unicast", fromHash: string, costBytes: number, now = Date.now(), ): boolean { if (!fromHash) return false; if (costBytes <= 0) return true; const budgetBps = kind === "proxy-publish" ? ch.proxyPublishBudgetBps : ch.unicastBudgetBps; const capacity = kind === "proxy-publish" ? ch.proxyPublishTokenCapacity : ch.unicastTokenCapacity; const byPeer = kind === "proxy-publish" ? ch.proxyPublishTokensByPeer : ch.unicastTokensByPeer; if (budgetBps <= 0 || capacity <= 0) return true; const prev = byPeer.get(fromHash); const entry = prev || ({ tokens: capacity, lastRefillAt: now } as const); const elapsedMs = now - entry.lastRefillAt; let tokens = entry.tokens; if (elapsedMs > 0) { tokens = Math.min(capacity, tokens + (elapsedMs * budgetBps) / 1_000); } if (tokens < costBytes) { // LRU-touch on access to keep the bucket map bounded by active children. byPeer.delete(fromHash); byPeer.set(fromHash, { tokens, lastRefillAt: now }); return false; } tokens -= costBytes; byPeer.delete(fromHash); byPeer.set(fromHash, { tokens, lastRefillAt: now }); return true; } private async _sendData( ch: ChannelState, to: string[], seq: number, payload: Uint8Array, ): Promise { this.markCached(ch, seq, payload); ch.maxSeqSeen = Math.max(ch.maxSeqSeen, seq); const framed = this.codec.encodeData(payload); const message = await this.createMessage(framed, { mode: new AnyWhere(), priority: DATA_PRIORITY, id: this.makeDataId(ch, seq), } as any); if (to.length === 0) { return message; } this.pruneDisconnectedChildren(ch); const candidates = to .map((hash) => ({ hash, stream: this.peers.get(hash) })) .filter((c): c is { hash: string; stream: PeerStreams } => Boolean(c.stream), ); if (candidates.length === 0) return message; let selected = candidates; let dropped = 0; let costPerChild = 0; // Best-effort upload shaping: token bucket enforced at message boundaries. if (ch.uploadLimitBps > 0 && ch.uploadTokenCapacity > 0) { const now = Date.now(); this.refillUploadTokens(ch, now); costPerChild = Math.max(1, framed.byteLength + ch.uploadOverheadBytes); const affordable = Math.floor(ch.uploadTokens / costPerChild); if (affordable <= 0) { selected = []; dropped = candidates.length; } else if (affordable < candidates.length) { selected = [...candidates] .sort((a, b) => { const bidA = ch.children.get(a.hash)?.bidPerByte ?? 0; const bidB = ch.children.get(b.hash)?.bidPerByte ?? 0; return bidB - bidA; }) .slice(0, affordable); dropped = candidates.length - selected.length; } const hasChildCandidate = candidates.some((c) => ch.children.has(c.hash)); if (hasChildCandidate) { if (dropped > 0) { ch.droppedForwards += dropped; ch.overloadStreak += 1; } else { ch.overloadStreak = 0; } } // Reserve tokens up front to keep the cap consistent under concurrent sends. // Any unused reservation is refunded after attempting writes. if (selected.length > 0) { ch.uploadTokens -= selected.length * costPerChild; } if ( dropped > 0 && hasChildCandidate && ch.allowKick && ch.overloadStreak >= OVERLOAD_KICK_STREAK_THRESHOLD && now - ch.lastOverloadKickAt > OVERLOAD_KICK_COOLDOWN_MS ) { const selectedSet = new Set(selected.map((c) => c.hash)); const droppedCandidates = candidates.filter( (c) => !selectedSet.has(c.hash) && ch.children.has(c.hash), ); droppedCandidates.sort((a, b) => { const bidA = ch.children.get(a.hash)?.bidPerByte ?? 0; const bidB = ch.children.get(b.hash)?.bidPerByte ?? 0; if (bidA !== bidB) return bidA - bidB; return a.hash < b.hash ? -1 : a.hash > b.hash ? 1 : 0; }); const kickCount = Math.min( OVERLOAD_KICK_MAX_PER_EVENT, droppedCandidates.length, ); if (kickCount > 0) { const toKick = droppedCandidates .slice(0, kickCount) .map((c) => c.hash); ch.overloadStreak = 0; ch.lastOverloadKickAt = now; void this.kickChildHashes(ch, toKick).catch(() => {}); void this.announceToTrackers(ch, this.closeController.signal).catch( () => {}, ); } } } if (selected.length === 0) return message; // Best-effort DATA-plane: never block on writability. Drop if not writable and rely on repair. const msgBytes = message.bytes(); const bytesToSend = msgBytes; const priority = Number(message.header.priority ?? DATA_PRIORITY); // Keep seen-cache semantics consistent with DirectStream.publishMessage so we // ignore any loops/duplicates that come back to us. try { const msgId = toBase64(bytesToSend.subarray(0, 33)); // discriminator + id const seen = this.seenCache.get(msgId); this.seenCache.add(msgId, seen ? seen + 1 : 1); } catch { // ignore } let successes = 0; let writeDrops = 0; const now = Date.now(); const writeFailKickCandidates: string[] = []; for (const c of selected) { const stream = c.stream; if (!stream.isWritable) { writeDrops += 1; continue; } try { stream.write(bytesToSend, priority); successes += 1; ch.dataWriteFailStreakByChild.delete(c.hash); const bid = ch.children.get(c.hash)?.bidPerByte ?? 0; if (bid > 0) ch.metrics.earnings += payload.byteLength * bid; } catch { writeDrops += 1; if (ch.children.has(c.hash)) { const streak = (ch.dataWriteFailStreakByChild.get(c.hash) ?? 0) + 1; ch.dataWriteFailStreakByChild.set(c.hash, streak); if (streak >= DATA_WRITE_FAIL_KICK_STREAK_THRESHOLD) { writeFailKickCandidates.push(c.hash); } } } } if (successes > 0) { ch.metrics.dataSends += successes; ch.metrics.dataPayloadBytesSent += payload.byteLength * successes; } if (costPerChild > 0 && successes < selected.length) { const refund = (selected.length - successes) * costPerChild; ch.uploadTokens = Math.min( ch.uploadTokenCapacity, ch.uploadTokens + refund, ); } if (writeDrops > 0) ch.metrics.dataWriteDrops += writeDrops; // Kick persistently failing children (best-effort, bounded per event). if ( ch.allowKick && writeFailKickCandidates.length > 0 && now - ch.lastDataWriteFailKickAt > DATA_WRITE_FAIL_KICK_COOLDOWN_MS ) { const unique = [...new Set(writeFailKickCandidates)].filter((h) => ch.children.has(h), ); unique.sort((a, b) => { const bidA = ch.children.get(a)?.bidPerByte ?? 0; const bidB = ch.children.get(b)?.bidPerByte ?? 0; if (bidA !== bidB) return bidA - bidB; return a < b ? -1 : a > b ? 1 : 0; }); const kickCount = Math.min( DATA_WRITE_FAIL_KICK_MAX_PER_EVENT, unique.length, ); if (kickCount > 0) { const toKick = unique.slice(0, kickCount); ch.lastDataWriteFailKickAt = now; void this.kickChildHashes(ch, toKick, { resetPeerConnections: true }).catch( () => {}, ); void this.announceToTrackers(ch, this.closeController.signal).catch( () => {}, ); } } return message; } private async _forwardDataMessage( ch: ChannelState, to: string[], payload: Uint8Array, message: DataMessage, ): Promise { if (to.length === 0) return; this.pruneDisconnectedChildren(ch); const candidates = to .map((hash) => ({ hash, stream: this.peers.get(hash) })) .filter((c): c is { hash: string; stream: PeerStreams } => Boolean(c.stream), ); if (candidates.length === 0) return; if (ch.maxDataAgeMs > 0) { const ts = Number(message.header.timestamp); if (Number.isFinite(ts)) { const ageMs = Date.now() - ts; if (ageMs > ch.maxDataAgeMs) { ch.metrics.staleForwardsDropped += candidates.length; return; } } } let selected = candidates; let dropped = 0; let costPerChild = 0; // Best-effort upload shaping: token bucket enforced at message boundaries. if (ch.uploadLimitBps > 0 && ch.uploadTokenCapacity > 0) { const now = Date.now(); this.refillUploadTokens(ch, now); costPerChild = Math.max( 1, payload.byteLength + 1 + ch.uploadOverheadBytes, ); const affordable = Math.floor(ch.uploadTokens / costPerChild); if (affordable <= 0) { selected = []; dropped = candidates.length; } else if (affordable < candidates.length) { selected = [...candidates] .sort((a, b) => { const bidA = ch.children.get(a.hash)?.bidPerByte ?? 0; const bidB = ch.children.get(b.hash)?.bidPerByte ?? 0; return bidB - bidA; }) .slice(0, affordable); dropped = candidates.length - selected.length; } const hasChildCandidate = candidates.some((c) => ch.children.has(c.hash)); if (hasChildCandidate) { if (dropped > 0) { ch.droppedForwards += dropped; ch.overloadStreak += 1; } else { ch.overloadStreak = 0; } } // Reserve tokens up front to keep the cap consistent under concurrent sends. // Any unused reservation is refunded after attempting writes. if (selected.length > 0) { ch.uploadTokens -= selected.length * costPerChild; } if ( dropped > 0 && hasChildCandidate && ch.allowKick && ch.overloadStreak >= OVERLOAD_KICK_STREAK_THRESHOLD && now - ch.lastOverloadKickAt > OVERLOAD_KICK_COOLDOWN_MS ) { const selectedSet = new Set(selected.map((c) => c.hash)); const droppedCandidates = candidates.filter( (c) => !selectedSet.has(c.hash) && ch.children.has(c.hash), ); droppedCandidates.sort((a, b) => { const bidA = ch.children.get(a.hash)?.bidPerByte ?? 0; const bidB = ch.children.get(b.hash)?.bidPerByte ?? 0; if (bidA !== bidB) return bidA - bidB; return a.hash < b.hash ? -1 : a.hash > b.hash ? 1 : 0; }); const kickCount = Math.min( OVERLOAD_KICK_MAX_PER_EVENT, droppedCandidates.length, ); if (kickCount > 0) { const toKick = droppedCandidates .slice(0, kickCount) .map((c) => c.hash); ch.overloadStreak = 0; ch.lastOverloadKickAt = now; void this.kickChildHashes(ch, toKick).catch(() => {}); void this.announceToTrackers(ch, this.closeController.signal).catch( () => {}, ); } } } if (selected.length === 0) return; const msgBytes = message.bytes(); const bytesToSend = msgBytes; const priority = Number(message.header.priority ?? DATA_PRIORITY); let successes = 0; let writeDrops = 0; const now = Date.now(); const writeFailKickCandidates: string[] = []; for (const c of selected) { const stream = c.stream; if (!stream.isWritable) { writeDrops += 1; continue; } try { stream.write(bytesToSend, priority); successes += 1; ch.dataWriteFailStreakByChild.delete(c.hash); const bid = ch.children.get(c.hash)?.bidPerByte ?? 0; if (bid > 0) ch.metrics.earnings += payload.byteLength * bid; } catch { writeDrops += 1; const streak = (ch.dataWriteFailStreakByChild.get(c.hash) ?? 0) + 1; ch.dataWriteFailStreakByChild.set(c.hash, streak); if (streak >= DATA_WRITE_FAIL_KICK_STREAK_THRESHOLD) { writeFailKickCandidates.push(c.hash); } } } if (successes > 0) { ch.metrics.dataSends += successes; ch.metrics.dataPayloadBytesSent += payload.byteLength * successes; } if (costPerChild > 0 && successes < selected.length) { const refund = (selected.length - successes) * costPerChild; ch.uploadTokens = Math.min( ch.uploadTokenCapacity, ch.uploadTokens + refund, ); } if (writeDrops > 0) ch.metrics.dataWriteDrops += writeDrops; // Kick persistently failing children (best-effort, bounded per event). if ( ch.allowKick && writeFailKickCandidates.length > 0 && now - ch.lastDataWriteFailKickAt > DATA_WRITE_FAIL_KICK_COOLDOWN_MS ) { const unique = [...new Set(writeFailKickCandidates)].filter((h) => ch.children.has(h), ); unique.sort((a, b) => { const bidA = ch.children.get(a)?.bidPerByte ?? 0; const bidB = ch.children.get(b)?.bidPerByte ?? 0; if (bidA !== bidB) return bidA - bidB; return a < b ? -1 : a > b ? 1 : 0; }); const kickCount = Math.min( DATA_WRITE_FAIL_KICK_MAX_PER_EVENT, unique.length, ); if (kickCount > 0) { const toKick = unique.slice(0, kickCount); ch.lastDataWriteFailKickAt = now; void this.kickChildHashes(ch, toKick, { resetPeerConnections: true }).catch( () => {}, ); void this.announceToTrackers(ch, this.closeController.signal).catch( () => {}, ); } } } private noteReceivedSeq(ch: ChannelState, fromHash: string, seq: number) { if (!ch.repairEnabled) return; ch.repairRequestedAtBySeq.delete(seq); if (!ch.parent || fromHash !== ch.parent) { // Neighbor-assisted repair may deliver payload from a peer other than the // current parent; treat it as a "hole fill" only. const wasMissing = ch.missingSeqs.delete(seq); if (wasMissing) { let have = ch.haveByPeer.get(fromHash); if (!have) { have = { haveFrom: 0, haveToExclusive: 0, updatedAt: Date.now(), requests: 0, successes: 0, }; ch.haveByPeer.set(fromHash, have); } have.successes += 1; ch.metrics.holeFillsFromNeighbor += 1; } return; } if (seq >= ch.nextExpectedSeq) { const maxBackfill = Math.max(0, ch.repairMaxBackfillMessages); const start = maxBackfill > 0 ? Math.max(ch.nextExpectedSeq, Math.max(0, seq - maxBackfill)) : ch.nextExpectedSeq; for (let s = start; s < seq; s++) ch.missingSeqs.add(s); ch.nextExpectedSeq = seq + 1; if (maxBackfill > 0) { const minSeq = Math.max(0, ch.nextExpectedSeq - maxBackfill); for (const s of ch.missingSeqs) { if (s < minSeq) ch.missingSeqs.delete(s); } } } ch.missingSeqs.delete(seq); } private noteEnd( ch: ChannelState, fromHash: string, lastSeqExclusive: number, ) { if (!ch.repairEnabled) return; if (!ch.parent || fromHash !== ch.parent) return; if (lastSeqExclusive > ch.nextExpectedSeq) { const maxBackfill = Math.max(0, ch.repairMaxBackfillMessages); const start = maxBackfill > 0 ? Math.max( ch.nextExpectedSeq, Math.max(0, lastSeqExclusive - maxBackfill), ) : ch.nextExpectedSeq; for (let s = start; s < lastSeqExclusive; s++) ch.missingSeqs.add(s); ch.nextExpectedSeq = lastSeqExclusive; if (maxBackfill > 0) { const minSeq = Math.max(0, ch.nextExpectedSeq - maxBackfill); for (const s of ch.missingSeqs) { if (s < minSeq) ch.missingSeqs.delete(s); } } } } private async tickRepair( ch: ChannelState, now = Date.now(), ): Promise { if (!ch.repairEnabled) return false; // Drop missing seqs that are too old to be realistically repaired (bounded window / live mode). if (ch.missingSeqs.size > 0 && ch.repairMaxBackfillMessages > 0) { const minSeq = Math.max( 0, ch.nextExpectedSeq - ch.repairMaxBackfillMessages, ); for (const s of ch.missingSeqs) { if (s < minSeq) ch.missingSeqs.delete(s); } } for (const s of ch.repairRequestedAtBySeq.keys()) { if (!ch.missingSeqs.has(s)) ch.repairRequestedAtBySeq.delete(s); } if (ch.missingSeqs.size === 0) return false; if ( ch.repairIntervalMs > 0 && now - ch.lastRepairSentAt < ch.repairIntervalMs ) { return false; } const repairRetryMs = Math.max( REPAIR_RETRY_MIN_MS, ch.repairIntervalMs * REPAIR_RETRY_INTERVAL_FACTOR, ); const missing = [...ch.missingSeqs] .filter((s) => now - (ch.repairRequestedAtBySeq.get(s) ?? 0) >= repairRetryMs) .sort((a, b) => a - b); const count = Math.min(ch.repairMaxPerReq, missing.length, 255); if (count <= 0) return false; ch.lastRepairSentAt = now; const reqId = (this.random() * 0xffffffff) >>> 0; const slice = missing.slice(0, count); for (const s of slice) ch.repairRequestedAtBySeq.set(s, now); let sent = false; if (ch.parent) { await this._sendControl( ch.parent, this.codec.encodeRepairReq(ch.id.key, reqId, slice), ); ch.parentRepairUnansweredStreak += 1; sent = true; } if (ch.neighborRepair && ch.neighborRepairPeers > 0) { const candidates: string[] = []; for (const h of ch.knownCandidateAddrs.keys()) candidates.push(h); for (const h of ch.channelPeers.keys()) candidates.push(h); for (const h of ch.lazyPeers) candidates.push(h); for (const h of ch.haveByPeer.keys()) candidates.push(h); const unique = [...new Set(candidates)].filter((h) => { if (h === this.publicKeyHash) return false; if (h === ch.parent) return false; if (ch.children.has(h)) return false; return Boolean(this.peers.get(h)); }); const ttl = Math.max(0, ch.neighborHaveTtlMs); const scored = unique.map((peerHash) => { const have = ch.haveByPeer.get(peerHash); const updatedAt = have?.updatedAt ?? 0; const fresh = !have || ttl <= 0 ? true : updatedAt > 0 && now - updatedAt <= ttl; let coverage = 0; if ( have && fresh && have.haveToExclusive > have.haveFrom && have.haveToExclusive > 0 ) { for (const s of slice) { if (s < have.haveFrom) continue; if (s >= have.haveToExclusive) break; coverage += 1; } } const requests = have?.requests ?? 0; const successes = have?.successes ?? 0; const successRate = (successes + 1) / (requests + 2); return { peerHash, coverage, successRate, updatedAt, }; }); const viable = scored.filter((s) => s.coverage > 0); viable.sort((a, b) => { if (a.coverage !== b.coverage) return b.coverage - a.coverage; if (a.successRate !== b.successRate) return b.successRate - a.successRate; if (a.updatedAt !== b.updatedAt) return b.updatedAt - a.updatedAt; const al = ch.lazyPeers.has(a.peerHash) ? 1 : 0; const bl = ch.lazyPeers.has(b.peerHash) ? 1 : 0; if (al !== bl) return bl - al; return a.peerHash < b.peerHash ? -1 : a.peerHash > b.peerHash ? 1 : 0; }); let peers = viable.slice(0, ch.neighborRepairPeers).map((s) => s.peerHash); if (peers.length > 0) { const bytes = this.codec.encodeFetchReq(ch.id.key, reqId, slice); if ( ch.neighborRepairBudgetBps > 0 && ch.neighborRepairTokenCapacity > 0 ) { this.refillNeighborRepairTokens(ch, now); const costPerPeer = Math.max(1, bytes.byteLength); const affordable = Math.floor(ch.neighborRepairTokens / costPerPeer); if (affordable <= 0) { peers = []; } else if (affordable < peers.length) { peers = peers.slice(0, affordable); } ch.neighborRepairTokens -= peers.length * costPerPeer; } if (peers.length > 0) { for (const p of peers) { let have = ch.haveByPeer.get(p); if (!have) { have = { haveFrom: 0, haveToExclusive: 0, updatedAt: 0, requests: 0, successes: 0, }; ch.haveByPeer.set(p, have); } have.requests += 1; } await this._sendControlMany(peers, bytes); sent = true; } } } return sent; } private getBootstrapsForChannel(ch: ChannelState): Multiaddr[] { return ch.bootstrapOverride ?? this.bootstraps; } private getSelfAnnounceAddrs(): Multiaddr[] { const peerIdStr = this.components.peerId.toString(); const addrs = this.components.addressManager.getAddresses(); if (addrs.length === 0) return []; const out: Multiaddr[] = []; for (const a of addrs) { const s = a.toString(); if (s.includes("/p2p/")) { out.push(a); continue; } try { const withPeer = a.encapsulate(`/p2p/${peerIdStr}`); // Some test/mocked peer ids may not be valid `/p2p/` values. // Validate that the multiaddr can be encoded before advertising it. void withPeer.bytes; out.push(withPeer); } catch { out.push(a); } } return out; } private async ensureBootstrapPeers( addrs: Multiaddr[], timeoutMs: number, signal: AbortSignal, maxPeers = 0, ): Promise { if (addrs.length === 0) return []; const max = Math.max(0, Math.floor(maxPeers)); const shuffled = addrs.slice(); for (let i = shuffled.length - 1; i > 0; i--) { const j = Math.floor(this.random() * (i + 1)); const tmp = shuffled[i]!; shuffled[i] = shuffled[j]!; shuffled[j] = tmp; } const target = max > 0 ? Math.min(max, shuffled.length) : shuffled.length; const out: string[] = []; for (const a of shuffled) { if (signal.aborted) break; if (target > 0 && out.length >= target) break; try { const conn = await this.components.connectionManager.openConnection(a); const h = getPublicKeyFromPeerId(conn.remotePeer).hashcode(); await this.waitFor(h, { seek: "present", timeout: timeoutMs, signal }); out.push(h); } catch { // ignore dial failures } } return [...new Set(out)]; } private async announceToTrackers( ch: ChannelState, signal: AbortSignal, ): Promise { const now = Date.now(); if (now - ch.lastAnnouncedAt < 100) return; ch.lastAnnouncedAt = now; // Only announce capacity if we're attached (or root). if (!ch.isRoot && !ch.parent) return; if (ch.effectiveMaxChildren <= 0) return; const bootstraps = this.getBootstrapsForChannel(ch); if (bootstraps.length === 0) return; // Announce to ALL trackers, not a random one per round: with B bootstraps // and per-round random selection a given directory only refreshes an // entry every ~announceIntervalMs * B on average, which starves // directories (and exceeds the announce TTL entirely once B grows). // Use already-connected tracker peers and only fall back to dialing when // none are connected: serial dial timeouts here would otherwise make // every announce tick glacial. let peers = ch.cachedAnnounceTrackerPeers ?? []; peers = peers.filter((h) => Boolean(this.peers.get(h))); if (peers.length === 0) { // Dial only when no tracker is connected at all: dialing every tick // makes the announce loop glacial and starves everything behind it. peers = await this.ensureBootstrapPeers( bootstraps, ch.bootstrapDialTimeoutMs, signal, 0, ); } ch.cachedAnnounceTrackerPeers = peers; if (peers.length === 0) return; this.pruneDisconnectedChildren(ch); const parentUpgradeNoCapacity = ch.isRoot && ch.parentUpgradeTrackerNoCapacityUntil > now; const freeSlots = parentUpgradeNoCapacity ? 0 : Math.max(0, ch.effectiveMaxChildren - ch.children.size); const announceTtlMs = parentUpgradeNoCapacity ? Math.max( 1_000, Math.min( ch.announceTtlMs, ch.parentUpgradeTrackerNoCapacityUntil - now, ), ) : ch.announceTtlMs; const addrs = this.getSelfAnnounceAddrs(); const bytes = this.codec.encodeTrackerAnnounce( ch.id.key, announceTtlMs, Number.isFinite(ch.level) ? ch.level : 0xffff, ch.effectiveMaxChildren, freeSlots, ch.bidPerByte, addrs, ); await this._sendControlMany(peers, bytes); } private async _announceLoop(ch: ChannelState): Promise { const signal = this.closeController.signal; for (;;) { if (signal.aborted || ch.closed) return; try { this.pruneRouteCache(ch); } catch { // ignore } try { // Child watermark heartbeat first: it only writes to already // connected peers (children/mesh) and must not starve behind the // dial-bound tracker announce below. Parents (root included - it // runs no repair/mesh loop) advertise their have-range to children // so a child whose data writes were silently lost can detect the // gap and repair. await this.maybeSendIHave(ch, Date.now()); } catch { // ignore } try { await this.announceToTrackers(ch, signal); } catch { // ignore } const sleepMs = ch.announceIntervalMs > 0 ? ch.announceIntervalMs : 1_000; await delay(sleepMs); } } private async _repairLoop(ch: ChannelState): Promise { const signal = this.closeController.signal; for (;;) { if (signal.aborted || ch.closed) return; try { await this.tickRepair(ch); } catch { // ignore } const activeMs = ch.repairIntervalMs > 0 ? ch.repairIntervalMs : 200; const sleepMs = ch.missingSeqs.size > 0 ? activeMs : Math.max(activeMs, 1_000); await delay(sleepMs); } } private pruneLazyPeers(ch: ChannelState) { for (const h of ch.lazyPeers) { if (h === this.publicKeyHash) { ch.lazyPeers.delete(h); continue; } if (h === ch.parent) { ch.lazyPeers.delete(h); continue; } if (ch.children.has(h)) { ch.lazyPeers.delete(h); continue; } if (!this.peers.get(h)) { ch.lazyPeers.delete(h); continue; } } if (ch.neighborMeshPeers > 0 && ch.lazyPeers.size > ch.neighborMeshPeers) { const peers = [...ch.lazyPeers]; peers.sort((a, b) => { const sa = ch.haveByPeer.get(a); const sb = ch.haveByPeer.get(b); const ar = sa ? (sa.successes + 1) / (sa.requests + 2) : 0; const br = sb ? (sb.successes + 1) / (sb.requests + 2) : 0; if (ar !== br) return br - ar; const au = sa?.updatedAt ?? 0; const bu = sb?.updatedAt ?? 0; if (au !== bu) return bu - au; return a < b ? -1 : a > b ? 1 : 0; }); for (const h of peers.slice(ch.neighborMeshPeers)) ch.lazyPeers.delete(h); } } private pruneDisconnectedChildren(ch: ChannelState) { for (const childHash of ch.children.keys()) { const peer = this.peers.get(childHash); if (!peer) { ch.children.delete(childHash); ch.dataWriteFailStreakByChild.delete(childHash); continue; } let connected = true; try { const conns = this.components.connectionManager.getConnections( peer.peerId, ); connected = conns.length > 0; } catch { connected = peer.isReadable || peer.isWritable; } if (!connected) { ch.children.delete(childHash); ch.dataWriteFailStreakByChild.delete(childHash); } } } private pruneHaveByPeer(ch: ChannelState, now = Date.now()) { const ttl = Math.max(0, ch.neighborHaveTtlMs); for (const [peerHash, have] of ch.haveByPeer) { if (!this.peers.get(peerHash) && !ch.lazyPeers.has(peerHash)) { ch.haveByPeer.delete(peerHash); continue; } // Keep stats longer than TTL, but prune very stale IHAVE ranges. if (ttl > 0 && have.updatedAt > 0 && now - have.updatedAt > ttl * 4) { have.haveFrom = 0; have.haveToExclusive = 0; } } } private getHaveRange( ch: ChannelState, ): { haveFrom: number; haveToExclusive: number } | undefined { if (!ch.repairEnabled) return; if (!ch.cacheSeqs || !ch.cachePayloads || ch.cacheSeqs.length === 0) return; if (ch.maxSeqSeen < 0) return; const haveToExclusive = ch.maxSeqSeen + 1; const maxScan = Math.min(haveToExclusive, ch.cacheSeqs.length); let haveFrom = haveToExclusive; for (let i = 0; i < maxScan; i++) { const seq = haveToExclusive - 1 - i; if (!this.getCached(ch, seq)) break; haveFrom = seq; } return { haveFrom, haveToExclusive }; } private async maybeSendIHave(ch: ChannelState, now = Date.now()): Promise { const meshEnabled = ch.neighborRepair && ch.neighborMeshPeers > 0; const childWatermarkEnabled = ch.repairEnabled && ch.children.size > 0; if (!meshEnabled && !childWatermarkEnabled) { return; } if (ch.neighborAnnounceIntervalMs <= 0) { return; } if (now - ch.lastIHaveSentAt < ch.neighborAnnounceIntervalMs) { return; } const hasNewData = ch.maxSeqSeen > ch.lastIHaveSentMaxSeq; // Children also need a periodic watermark even without new data: a burst // of data writes can be lost on a not-yet-writable stream and a child // that received nothing has no gap signal to trigger repair. A periodic // IHAVE from the parent converts that silent loss into a repairable gap. const heartbeatDue = childWatermarkEnabled && (ch.maxSeqSeen >= 0 || ch.endSeqExclusive > 0) && now - ch.lastIHaveSentAt >= Math.max(ch.neighborAnnounceIntervalMs * 4, 2_000); if (!hasNewData && !heartbeatDue) { return; } const recipients = new Set(); if (meshEnabled) { for (const h of ch.lazyPeers) { if (this.peers.get(h)) recipients.add(h); } } if (childWatermarkEnabled) { for (const h of ch.children.keys()) { if (this.peers.get(h)) recipients.add(h); } } if (recipients.size === 0) { return; } let sent = false; const range = this.getHaveRange(ch); if (range && range.haveToExclusive > range.haveFrom) { await this._sendControlMany( [...recipients], this.codec.encodeIHave(ch.id.key, range.haveFrom, range.haveToExclusive), ); sent = true; } if (childWatermarkEnabled && ch.endSeqExclusive > 0) { const children = [...ch.children.keys()].filter((h) => this.peers.get(h)); if (children.length > 0) { await this._sendControlMany( children, this.codec.encodeEnd(ch.id.key, ch.endSeqExclusive), ); sent = true; } } if (sent) { ch.lastIHaveSentAt = now; ch.lastIHaveSentMaxSeq = ch.maxSeqSeen; } } private async ensureMeshPeers( ch: ChannelState, signal: AbortSignal, ): Promise { if (ch.neighborMeshPeers <= 0) return; if (ch.lazyPeers.size >= ch.neighborMeshPeers) return; const maxAttempts = Math.max(16, ch.neighborMeshPeers * 16); let attempts = 0; const entries = [...ch.knownCandidateAddrs.entries()]; for (let i = entries.length - 1; i > 0; i--) { const j = Math.floor(this.random() * (i + 1)); const tmp = entries[i]!; entries[i] = entries[j]!; entries[j] = tmp; } for (const [hash, addrs] of entries) { if (attempts++ > maxAttempts) break; if (signal.aborted) break; if (ch.lazyPeers.size >= ch.neighborMeshPeers) break; if (hash === this.publicKeyHash) continue; if (hash === ch.parent) continue; if (ch.children.has(hash)) continue; if (ch.lazyPeers.has(hash)) continue; if (this.peers.get(hash)) { ch.lazyPeers.add(hash); continue; } if (!addrs || addrs.length === 0) continue; const ok = await this.ensurePeerConnection( hash, addrs, ch.bootstrapDialTimeoutMs, signal, ); if (ok) ch.lazyPeers.add(hash); } } private async refreshMeshCandidates( ch: ChannelState, signal: AbortSignal, ): Promise { if (ch.neighborMeshPeers <= 0) return; if (ch.lazyPeers.size >= ch.neighborMeshPeers) return; const bootstraps = this.getBootstrapsForChannel(ch); if (bootstraps.length === 0) return; const trackerPeers = await this.ensureBootstrapPeers( bootstraps, ch.bootstrapDialTimeoutMs, signal, ch.bootstrapMaxPeers, ); if (trackerPeers.length === 0) return; const want = Math.max(16, ch.neighborMeshPeers * 4); const candidates = await this.queryTrackers( ch, trackerPeers, want, 1_000, signal, ); for (const c of candidates) { if (c.hash === this.publicKeyHash) continue; if (c.addrs.length === 0) continue; this.cacheKnownCandidateAddrs(ch, c.hash, c.addrs); } } private async _meshLoop(ch: ChannelState): Promise { const signal = this.closeController.signal; let lastRefreshAt = 0; for (;;) { if (signal.aborted || ch.closed) return; const now = Date.now(); try { this.pruneLazyPeers(ch); this.pruneHaveByPeer(ch, now); await this.ensureMeshPeers(ch, signal); const refreshMs = Math.max(0, ch.neighborMeshRefreshIntervalMs); if ( ch.lazyPeers.size < ch.neighborMeshPeers && (refreshMs === 0 || now - lastRefreshAt >= refreshMs) ) { lastRefreshAt = now; await this.refreshMeshCandidates(ch, signal); await this.ensureMeshPeers(ch, signal); } await this.maybeSendIHave(ch, now); } catch { // ignore } const intervals = [ ch.neighborAnnounceIntervalMs, ch.neighborMeshRefreshIntervalMs, ].filter((v) => v > 0); const sleepMs = intervals.length > 0 ? Math.min(...intervals) : 500; await delay(Math.max(50, sleepMs)); } } private async queryTrackers( ch: ChannelState, trackerPeers: string[], want: number, timeoutMs: number, signal: AbortSignal, ): Promise { if (trackerPeers.length === 0) return []; const perTrackerTimeout = Math.max(0, Math.floor(timeoutMs)); const results = await Promise.all( trackerPeers.map(async (trackerHash) => { const reqId = (this.random() * 0xffffffff) >>> 0; const p = new Promise((resolve) => { ch.pendingTrackerQuery.set(reqId, { resolve }); }); void this._sendControl( trackerHash, this.codec.encodeTrackerQuery(ch.id.key, reqId, want), ); const res = await Promise.race([ p, delay(perTrackerTimeout, { signal }).then((): null => null), ]); if (res == null) { ch.pendingTrackerQuery.delete(reqId); return []; } return res; }), ); const merged: TrackerCandidate[] = []; for (const r of results) merged.push(...r); const seen = new Set(); return merged.filter((c) => { if (seen.has(c.hash)) return false; seen.add(c.hash); return true; }); } private async ensurePeerConnection( hash: string, addrs: Multiaddr[], timeoutMs: number, signal: AbortSignal, ): Promise { if (this.peers.get(hash)) return true; for (const a of addrs) { if (signal.aborted) return false; try { await this.components.connectionManager.openConnection(a); await this.waitFor(hash, { seek: "present", timeout: timeoutMs, signal, }); return true; } catch { // ignore and try next } } return false; } private async getPeerAddrsBytes(hash: string): Promise { const stream = this.peers.get(hash); if (!stream) return []; try { const peer: any = await this.components.peerStore.get(stream.peerId); const addresses: any[] = Array.isArray(peer?.addresses) ? peer.addresses : []; const out: Uint8Array[] = []; for (const a of addresses) { const ma: any = a?.multiaddr ?? a; const bytes = normalizeUint8Array(ma?.bytes); if (!bytes) continue; out.push(bytes); if (out.length >= JOIN_REJECT_REDIRECT_ADDR_MAX) break; } return out; } catch { return []; } } private async pickJoinRejectRedirects( ch: ChannelState, excludeHash: string, limit: number, ): Promise { const max = Math.max( 0, Math.min(JOIN_REJECT_REDIRECT_MAX, Math.floor(limit)), ); if (max === 0) return []; const seen = new Set([excludeHash, this.publicKeyHash]); const out: JoinRejectRedirect[] = []; const children = [...ch.children.entries()]; children.sort((a, b) => b[1].bidPerByte - a[1].bidPerByte); for (const [hash] of children) { if (out.length >= max) break; if (seen.has(hash)) continue; seen.add(hash); const addrs = await this.getPeerAddrsBytes(hash); if (addrs.length === 0) continue; out.push({ hash, addrs }); } if (out.length >= max) return out; // Tracker entries this node has observed (mostly useful when acting as a bootstrap). const now = Date.now(); const byPeer = this.trackerBySuffixKey.get(ch.id.suffixKey); if (byPeer) { const entries: TrackerEntry[] = []; for (const e of byPeer.values()) { if (e.expiresAt <= now) continue; if (e.freeSlots <= 0) continue; if (e.addrs.length === 0) continue; if (seen.has(e.hash)) continue; entries.push(e); } entries.sort((a, b) => { if (a.level !== b.level) return a.level - b.level; if (a.freeSlots !== b.freeSlots) return b.freeSlots - a.freeSlots; if (a.bidPerByte !== b.bidPerByte) return b.bidPerByte - a.bidPerByte; return a.hash < b.hash ? -1 : a.hash > b.hash ? 1 : 0; }); for (const e of entries) { if (out.length >= max) break; seen.add(e.hash); out.push({ hash: e.hash, addrs: e.addrs.slice(0, JOIN_REJECT_REDIRECT_ADDR_MAX), }); } } if (out.length >= max) return out; // Fallback: known candidates this node has learned previously (addresses only). for (const [hash, addrs] of ch.knownCandidateAddrs) { if (out.length >= max) break; if (seen.has(hash)) continue; if (!addrs || addrs.length === 0) continue; const bytes: Uint8Array[] = []; for (const a of addrs.slice(0, JOIN_REJECT_REDIRECT_ADDR_MAX)) { try { bytes.push(a.bytes); } catch { // ignore invalid addrs } } if (bytes.length === 0) continue; seen.add(hash); out.push({ hash, addrs: bytes }); } return out; } private async sendJoinReject( ch: ChannelState, toHash: string, reqId: number, reason: number, ): Promise { const redirects = await this.pickJoinRejectRedirects( ch, toHash, JOIN_REJECT_REDIRECT_MAX, ); await this._sendControl( toHash, this.codec.encodeJoinReject(ch.id.key, reqId, reason, redirects), ); } private async sendTrackerFeedback( ch: ChannelState, trackerPeers: string[], candidateHash: string, event: number, reason = 0, ): Promise { if (trackerPeers.length === 0) return; await this._sendControlMany( trackerPeers, this.codec.encodeTrackerFeedback(ch.id.key, candidateHash, event, reason), ); } private pruneParentUpgradeReservations(ch: ChannelState, now = Date.now()) { for (const [hash, reservation] of ch.parentUpgradeReservationsByHash) { if (reservation.expiresAt <= now) { ch.parentUpgradeReservationsByHash.delete(hash); ch.metrics.parentUpgradeRootReservationExpired += 1; continue; } if (ch.children.has(hash)) { ch.parentUpgradeReservationsByHash.delete(hash); } } } private parentUpgradeReservationCount( ch: ChannelState, excludeHash?: string, now = Date.now(), ) { this.pruneParentUpgradeReservations(ch, now); let count = 0; for (const hash of ch.parentUpgradeReservationsByHash.keys()) { if (hash === excludeHash) continue; if (ch.children.has(hash)) continue; count += 1; } return count; } private createParentUpgradeReservation( ch: ChannelState, toHash: string, minFreeSlots = 0, now = Date.now(), ) { if (!ch.isRoot || ch.children.has(toHash)) return 0; const maxChildren = Math.max(0, Math.floor(ch.effectiveMaxChildren)); if (maxChildren <= 0) return 0; const reserved = this.parentUpgradeReservationCount(ch, toHash, now); const requestedMinFreeSlots = Math.max(0, Math.floor(minFreeSlots)); const freeSlots = maxChildren - ch.children.size - reserved; if ( freeSlots <= 0 || (requestedMinFreeSlots > 0 && freeSlots < requestedMinFreeSlots) ) { ch.metrics.parentUpgradeRootReservationBlocked += 1; return 0; } const token = (this.random() * 0xffffffff) >>> 0 || 1; ch.parentUpgradeReservationsByHash.set(toHash, { token, expiresAt: now + PARENT_UPGRADE_ROOT_RESERVATION_TTL_MS, minFreeSlots: requestedMinFreeSlots, }); ch.metrics.parentUpgradeRootReservationCreated += 1; return token; } private consumeParentUpgradeReservation( ch: ChannelState, fromHash: string, token: number, now = Date.now(), ) { if (!ch.isRoot || token <= 0) return false; this.pruneParentUpgradeReservations(ch, now); const reservation = ch.parentUpgradeReservationsByHash.get(fromHash); if (!reservation) return false; if (reservation.expiresAt <= now) { ch.parentUpgradeReservationsByHash.delete(fromHash); ch.metrics.parentUpgradeRootReservationExpired += 1; return false; } if (reservation.token !== token) return false; const requestedMinFreeSlots = Math.max( 0, Math.floor(reservation.minFreeSlots ?? 0), ); if (requestedMinFreeSlots > 0) { const maxChildren = Math.max(0, Math.floor(ch.effectiveMaxChildren)); const freeSlots = Math.max(0, maxChildren - ch.children.size); if (freeSlots < requestedMinFreeSlots) { ch.parentUpgradeReservationsByHash.delete(fromHash); ch.metrics.parentUpgradeRootReservationMarginRejected += 1; return false; } } ch.parentUpgradeReservationsByHash.delete(fromHash); ch.metrics.parentUpgradeRootReservationConsumed += 1; return true; } private isRootedForParentProbe(ch: ChannelState): boolean { if (ch.isRoot) return true; const route = ch.routeFromRoot; return Boolean( ch.parent && Array.isArray(route) && route.length >= 2 && route[0] === ch.id.root && route[route.length - 1] === this.publicKeyHash, ); } private encodeParentProbeReplyForChannel( ch: ChannelState, reqId: number, toHash: string, minFreeSlots = 0, reserveRootCapacity = true, ): Uint8Array { this.pruneDisconnectedChildren(ch); const rooted = this.isRootedForParentProbe(ch); const requestedMinFreeSlots = Math.max(0, Math.floor(minFreeSlots)); const maxChildren = Math.max(0, Math.floor(ch.effectiveMaxChildren)); const reserved = ch.isRoot ? this.parentUpgradeReservationCount(ch, toHash) : 0; const freeSlots = rooted ? Math.max(0, maxChildren - ch.children.size - reserved) : 0; const requesterHasEnoughSlots = freeSlots > 0 && (requestedMinFreeSlots <= 0 || freeSlots >= requestedMinFreeSlots); if ( ch.isRoot && rooted && maxChildren > 0 && !ch.children.has(toHash) && !requesterHasEnoughSlots ) { ch.metrics.parentUpgradeRootReservationBlocked += 1; } const reservationToken = ch.isRoot && rooted && requesterHasEnoughSlots && reserveRootCapacity ? this.createParentUpgradeReservation(ch, toHash, requestedMinFreeSlots) : 0; const repairing = ch.missingSeqs.size > 0; const overloaded = ch.overloadStreak > 0; let flags = 0; if (rooted) flags |= PARENT_PROBE_FLAG_ROOTED; if ( rooted && (reservationToken > 0 || (!ch.isRoot && requesterHasEnoughSlots) || (ch.isRoot && requesterHasEnoughSlots && !reserveRootCapacity)) ) { flags |= PARENT_PROBE_FLAG_ACCEPTING; } if (repairing) flags |= PARENT_PROBE_FLAG_REPAIRING; if (overloaded) flags |= PARENT_PROBE_FLAG_OVERLOADED; return this.codec.encodeParentProbeReply(ch.id.key, reqId, { flags, level: Number.isFinite(ch.level) ? ch.level : 0xffff, maxChildren, freeSlots, // For roots, expose reserved slots as child pressure so concurrent // upgrade probes cannot all pass the load-ratio check from stale state. children: ch.isRoot ? ch.children.size + reserved : ch.children.size, haveToExclusive: ch.maxSeqSeen >= 0 ? ch.maxSeqSeen + 1 : 0, missingSeqs: ch.missingSeqs.size, dataWriteDrops: ch.metrics.dataWriteDrops, droppedForwards: ch.droppedForwards, reservationToken, }); } private async probeParentCandidate( ch: ChannelState, parentHash: string, timeoutMs: number, signal: AbortSignal, minFreeSlots = 0, reserveRootCapacity = true, ): Promise { if (!this.peers.get(parentHash)) return; const reqId = (this.random() * 0xffffffff) >>> 0; const p = new Promise((resolve) => { ch.pendingParentProbe.set(reqId, { resolve }); }); await this._sendControl( parentHash, this.codec.encodeParentProbeReq(ch.id.key, reqId, minFreeSlots, reserveRootCapacity), ); const res = await Promise.race([ p, delay(Math.max(1, timeoutMs), { signal }).then( (): undefined => undefined, ), ]); if (!res) ch.pendingParentProbe.delete(reqId); return res; } private async _joinLoop( ch: ChannelState, joinOpts: FanoutTreeJoinOptions, ): Promise { const retryMs = Math.max(1, Math.floor(joinOpts.retryMs ?? 200)); const timeoutMs = Math.max(0, Math.floor(joinOpts.timeoutMs ?? 60_000)); const staleAfterMs = Math.max(0, Math.floor(joinOpts.staleAfterMs ?? 0)); const joinReqTimeoutMs = Math.max( 0, Math.floor(joinOpts.joinReqTimeoutMs ?? 2_000), ); const bootstrapDialTimeoutMs = Math.max( 0, Math.floor(joinOpts.bootstrapDialTimeoutMs ?? ch.bootstrapDialTimeoutMs), ); const bootstrapMaxPeers = Math.max( 0, Math.floor(joinOpts.bootstrapMaxPeers ?? ch.bootstrapMaxPeers), ); const trackerCandidates = Math.max( 0, Math.floor(joinOpts.trackerCandidates ?? 16), ); const candidateShuffleTopK = Math.max( 0, Math.floor(joinOpts.candidateShuffleTopK ?? 8), ); const trackerQueryTimeoutMs = Math.max( 0, Math.floor(joinOpts.trackerQueryTimeoutMs ?? 1_000), ); const bootstrapEnsureIntervalMs = Math.max( 0, Math.floor( joinOpts.bootstrapEnsureIntervalMs ?? ch.bootstrapEnsureIntervalMs, ), ); const trackerQueryIntervalMs = Math.max( 0, Math.floor(joinOpts.trackerQueryIntervalMs ?? ch.trackerQueryIntervalMs), ); const joinAttemptsPerRound = Math.max( 1, Math.floor(joinOpts.joinAttemptsPerRound ?? 8), ); const candidateCooldownMs = Math.max( 0, Math.floor(joinOpts.candidateCooldownMs ?? 2_000), ); const parentUpgrade: ParentUpgradePolicy = this.nativeFanout ? this.nativeFanout.normalizeParentUpgradePolicy(joinOpts) : normalizeParentUpgradePolicy(joinOpts); const candidateScoringModeRaw = joinOpts.candidateScoringMode ?? "ranked-shuffle"; const candidateScoringMode: | "ranked-shuffle" | "ranked-strict" | "weighted" = candidateScoringModeRaw === "ranked-strict" || candidateScoringModeRaw === "weighted" || candidateScoringModeRaw === "ranked-shuffle" ? candidateScoringModeRaw : "ranked-shuffle"; const candidateScoringWeights = { level: Number(joinOpts.candidateScoringWeights?.level ?? 1), freeSlots: Number(joinOpts.candidateScoringWeights?.freeSlots ?? 0.25), connected: Number(joinOpts.candidateScoringWeights?.connected ?? 0.5), bidPerByte: Number(joinOpts.candidateScoringWeights?.bidPerByte ?? 0), source: Number(joinOpts.candidateScoringWeights?.source ?? 0.25), }; const start = Date.now(); const cooldownUntilByHash = new Map(); const combinedSignal = joinOpts.signal ? anySignal([this.closeController.signal, joinOpts.signal]) : this.closeController.signal; const signal = combinedSignal as AbortSignal & { clear?: () => void }; let nextParentUpgradeCheckAt = 0; let parentUpgradeCheckSeq = 0; let parentUpgradeActiveGuardBackoffMs = 0; const scheduleNextParentUpgradeCheck = ( now: number, first = false, minDelayMs = parentUpgrade.intervalMs, ) => { if (parentUpgrade.intervalMs <= 0) { nextParentUpgradeCheckAt = 0; return; } const factor = first ? stableUnitInterval( `${ch.id.suffixKey}:${this.publicKeyHash}:parent-upgrade-phase`, ) : 0.75 + stableUnitInterval( `${ch.id.suffixKey}:${this.publicKeyHash}:parent-upgrade-interval:${parentUpgradeCheckSeq++}`, ) * 0.5; nextParentUpgradeCheckAt = now + Math.max( 1, Math.floor(parentUpgrade.intervalMs * factor), Math.floor(minDelayMs), ); }; const resetParentUpgradeActiveGuardBackoff = () => { parentUpgradeActiveGuardBackoffMs = 0; }; const parentUpgradeGuardDelayMs = ( reason: ParentUpgradeSkipReason, now: number, ) => { const baseDelayMs = (() => { switch (reason) { case "data": return Math.max(parentUpgrade.quietMs, parentUpgrade.repairQuietMs); case "repair": return parentUpgrade.repairQuietMs; case "quiet": return ch.lastParentUpgradeActivityAt > 0 ? Math.max( 0, parentUpgrade.quietMs - (now - ch.lastParentUpgradeActivityAt), ) : parentUpgrade.quietMs; case "cooldown": return Math.max( 0, ch.parentUpgradeBackoffUntil - now, ch.parentUpgradeLastAt > 0 ? parentUpgrade.cooldownMs - (now - ch.parentUpgradeLastAt) : 0, ); default: return parentUpgrade.intervalMs; } })(); if (reason !== "data" && reason !== "repair") { resetParentUpgradeActiveGuardBackoff(); return baseDelayMs; } parentUpgradeActiveGuardBackoffMs = parentUpgradeActiveGuardBackoffMs > 0 ? Math.min( Math.max(parentUpgrade.failedBackoff.maxMs, baseDelayMs), Math.max(baseDelayMs, parentUpgradeActiveGuardBackoffMs * 2), ) : baseDelayMs; return parentUpgradeActiveGuardBackoffMs; }; try { for (;;) { if (ch.closed) { throw new AbortError("fanout join aborted: channel closed"); } if (signal.aborted) { throw signal.reason ?? new AbortError("fanout join aborted"); } // Parent disappeared? Rejoin. if (ch.parent) { const parentPeer = this.peers.get(ch.parent); let connected = false; if (parentPeer) { try { const conns = this.components.connectionManager.getConnections( parentPeer.peerId, ) as Connection[] | undefined; connected = (conns?.length ?? 0) > 0; } catch { connected = parentPeer.isReadable || parentPeer.isWritable; } } if (!connected) { ch.metrics.reparentDisconnect += 1; const hadChildren = ch.children.size > 0; this.detachFromParent(ch); nextParentUpgradeCheckAt = 0; resetParentUpgradeActiveGuardBackoff(); // If we lose our parent, we are no longer on the rooted tree; detach children so // they can rejoin as well (prevents stable disconnected components). void this.kickChildren(ch).catch(() => {}); if (hadChildren) { ch.rejoinCooldownUntil = Math.max( ch.rejoinCooldownUntil, Date.now() + Math.max(retryMs, RELAY_REJOIN_COOLDOWN_MS), ); } } } if (ch.parent) { const endedAndComplete = ch.endSeqExclusive > 0 && ch.missingSeqs.size === 0 && ch.nextExpectedSeq >= ch.endSeqExclusive; const expectingData = ch.missingSeqs.size > 0 || (ch.repairEnabled && ch.maxSeqSeen >= 0 && ch.endSeqExclusive < 0) || (ch.maxDataAgeMs > 0 && !endedAndComplete); // Repair requests travel child->parent on a direction that works // even when parent->child data writes silently fail. A repair that // does not fill a gap is not sufficient evidence by itself though: // the repair payload can be dropped or the parent cache can miss. // Re-parent only after both repeated unanswered repairs and a // quiet parent liveness window. const parentRepairDeadLivenessMs = Math.max( PARENT_REPAIR_DEAD_MIN_LIVENESS_MS, ch.repairIntervalMs * PARENT_REPAIR_DEAD_STREAK_THRESHOLD, ch.neighborAnnounceIntervalMs * 8, ); const parentRepairNow = Date.now(); const parentLivenessQuietMs = ch.lastParentDataAt > 0 ? parentRepairNow - ch.lastParentDataAt : 0; const parentRepairDead = ch.repairEnabled && ch.parentRepairUnansweredStreak >= PARENT_REPAIR_DEAD_STREAK_THRESHOLD && parentLivenessQuietMs >= parentRepairDeadLivenessMs; if ( parentRepairDead || ( staleAfterMs > 0 && ch.receivedAnyParentData && ch.lastParentDataAt > 0 && Date.now() - ch.lastParentDataAt > staleAfterMs && expectingData ) ) { // Parent is "alive" at the stream layer but we're not receiving // data at the expected rate; detach and try to re-parent. ch.metrics.reparentStale += 1; ch.parent = undefined; ch.parentRepairUnansweredStreak = 0; ch.level = Number.POSITIVE_INFINITY; ch.routeFromRoot = undefined; ch.routeByPeer.clear(); ch.lastParentDataAt = 0; ch.lastParentUpgradeActivityAt = 0; ch.receivedAnyParentData = false; ch.pendingJoin.clear(); ch.pendingParentProbe.clear(); ch.parentShadow = undefined; void Promise.all( this.clearParentUpgradeGrace(ch, true, true), ).catch(() => {}); if (this.parentUpgradeShadowInFlightSuffixKey === ch.id.suffixKey) { this.parentUpgradeShadowInFlightSuffixKey = undefined; } ch.parentUpgradeRetryAfterSeq = -1; ch.parentProbeRejectUntilByHash.clear(); ch.parentProbeRejectBackoffMsByHash.clear(); nextParentUpgradeCheckAt = 0; resetParentUpgradeActiveGuardBackoff(); ch.pendingRouteQuery.clear(); this.clearRouteProxies(ch); void this.kickChildren(ch).catch(() => {}); await delay(retryMs, { signal }); continue; } if (parentUpgrade.intervalMs > 0 && ch.level > 1) { const now = Date.now(); if (nextParentUpgradeCheckAt === 0) { scheduleNextParentUpgradeCheck( now, true, parentUpgrade.dataGuard || parentUpgrade.repairGuard ? Math.max( parentUpgrade.quietMs, parentUpgrade.repairQuietMs, ) : parentUpgrade.intervalMs, ); } const due = now >= nextParentUpgradeCheckAt; if (due) { const gateOptions = { leafOnly: parentUpgrade.leafOnly, repairGuard: parentUpgrade.repairGuard, dataGuard: parentUpgrade.dataGuard, endedAndComplete, maxPerPeer: parentUpgrade.maxPerPeer, cooldownMs: parentUpgrade.cooldownMs, quietMs: parentUpgrade.quietMs, repairQuietMs: parentUpgrade.repairQuietMs, now, }; const gate = this.nativeFanout ? this.nativeFanout.evaluateParentUpgradeGate(ch, gateOptions) : evaluateParentUpgradeGate(ch, gateOptions); if ("reason" in gate) { recordParentUpgradeSkip(ch.metrics, gate.reason); scheduleNextParentUpgradeCheck( now, false, parentUpgradeGuardDelayMs(gate.reason, now), ); } else { resetParentUpgradeActiveGuardBackoff(); const improved = await this.maybeImproveParent(ch, { signal, candidateShuffleTopK, candidateScoringMode, candidateScoringWeights, joinAttemptsPerRound, joinReqTimeoutMs, parentUpgrade, trackerPeers: ch.cachedBootstrapPeers.filter((h) => Boolean(this.peers.get(h)), ), }); scheduleNextParentUpgradeCheck(Date.now()); if (improved) { ch.metrics.reparentUpgrade += 1; ch.parentUpgradeCount += 1; ch.parentUpgradeLastAt = Date.now(); ch.parentUpgradeBackoffMs = 0; ch.parentUpgradeBackoffUntil = 0; ch.parentUpgradeRetryAfterSeq = -1; ch.parentUpgradeStaleRootProbeRound = 0; } } } } if (!ch.joinedOnce) ch.joinedOnce = createDeferred(); ch.joinedOnce.resolve(); ch.joinedAtLeastOnce = true; // Once attached, we don't need a fast retry cadence; keep polling coarse to // avoid excessive timers when simulating many nodes in one process. await delay(Math.max(retryMs, 1_000), { signal }); continue; } // `timeoutMs` is meant to bound the initial `joinChannel()` await, not to // stop re-parenting attempts for long-running nodes. if ( !ch.joinedAtLeastOnce && timeoutMs > 0 && Date.now() - start > timeoutMs ) { const bootstrapsCount = this.getBootstrapsForChannel(ch).length; const rootPeer = this.peers.get(ch.id.root); const rootNeighbor = Boolean( rootPeer && rootPeer.isReadable && rootPeer.isWritable, ); const bootstrapHint = bootstrapsCount === 0 && !rootNeighbor ? " No fanout bootstraps are configured for this channel, and the root is not a direct neighbor. If this peer reached the network via a bootstrap or relay node, initialize it with Peerbit.bootstrap(...) instead of Peerbit.dial(...), or configure FanoutTree.setBootstraps(...) before joining sharded topics." : ""; throw new Error( `fanout join timed out after ${timeoutMs}ms (topic=${ch.id.topic} root=${ch.id.root} self=${this.publicKeyHash} rootNeighbor=${rootNeighbor} peers=${this.peers.size} bootstraps=${bootstrapsCount} joinReqSent=${ch.metrics.joinReqSent} joinAcceptReceived=${ch.metrics.joinAcceptReceived} joinRejectReceived=${ch.metrics.joinRejectReceived} peerResets=${ch.metrics.joinPeerResets}).${bootstrapHint}`, ); } const cooldownMs = ch.rejoinCooldownUntil - Date.now(); if (cooldownMs > 0) { await delay(cooldownMs, { signal }); continue; } const bootstraps = this.getBootstrapsForChannel(ch); let bootstrapPeers: string[] = []; if (bootstraps.length > 0) { const now = Date.now(); const connectedCached = ch.cachedBootstrapPeers.filter((h) => Boolean(this.peers.get(h)), ); const due = ch.lastBootstrapEnsureAt === 0 || bootstrapEnsureIntervalMs === 0 || now - ch.lastBootstrapEnsureAt >= bootstrapEnsureIntervalMs; const haveEnough = bootstrapMaxPeers > 0 ? connectedCached.length >= bootstrapMaxPeers : false; if (due && !haveEnough) { ch.lastBootstrapEnsureAt = now; const peers = await this.ensureBootstrapPeers( bootstraps, bootstrapDialTimeoutMs, signal, bootstrapMaxPeers, ); if (peers.length > 0) ch.cachedBootstrapPeers = peers; } bootstrapPeers = ch.cachedBootstrapPeers.filter((h) => Boolean(this.peers.get(h)), ); } let tracker: TrackerCandidate[] = []; if (bootstrapPeers.length > 0 && trackerCandidates > 0) { const now = Date.now(); const due = ch.lastTrackerQueryAt === 0 || trackerQueryIntervalMs === 0 || now - ch.lastTrackerQueryAt >= trackerQueryIntervalMs; if (due) { ch.lastTrackerQueryAt = now; const res = await this.queryTrackers( ch, bootstrapPeers, trackerCandidates, trackerQueryTimeoutMs, signal, ); if (res.length > 0) ch.cachedTrackerCandidates = res; } tracker = ch.cachedTrackerCandidates; } const candidatesByHash = new Map< string, { hash: string; addrs: Multiaddr[]; level: number; freeSlots: number; bidPerByte: number; source: number; } >(); const upsertCandidate = (c: { hash: string; addrs: Multiaddr[]; level: number; freeSlots: number; bidPerByte: number; source: number; }) => { const prev = candidatesByHash.get(c.hash); if (!prev) { candidatesByHash.set(c.hash, { ...c }); return; } if (prev.addrs.length === 0 && c.addrs.length > 0) prev.addrs = c.addrs; prev.level = Math.min(prev.level, c.level); prev.freeSlots = Math.max(prev.freeSlots, c.freeSlots); prev.bidPerByte = Math.max(prev.bidPerByte, c.bidPerByte); prev.source = Math.min(prev.source, c.source); }; // Fast path: if the designated root is already a direct neighbor, try it first. // Without this, large join storms can repeatedly time out on arbitrary peers // that don't host the channel yet, starving the real root candidate. if (ch.id.root !== this.publicKeyHash && this.peers.has(ch.id.root)) { upsertCandidate({ hash: ch.id.root, addrs: [], level: 0, freeSlots: Number.MAX_SAFE_INTEGER, bidPerByte: 0, source: -1, }); } for (const c of tracker) { if (c.hash === this.publicKeyHash) continue; if (c.freeSlots <= 0) continue; upsertCandidate({ hash: c.hash, addrs: c.addrs, level: c.level, freeSlots: c.freeSlots, bidPerByte: c.bidPerByte, source: 0, }); } // Secondary fallback: cached tracker candidates we've learned previously. for (const [hash, addrs] of ch.knownCandidateAddrs) { if (hash === this.publicKeyHash) continue; if (!addrs || addrs.length === 0) continue; upsertCandidate({ hash, addrs, level: 0xffff, freeSlots: 0, bidPerByte: 0, source: 1, }); } // Fallback: try a bounded set of already-connected peers. Bootstrap // peers are often just trackers, but they can also be the only // connected channel host and may not list themselves in tracker // replies. Keep them as low-priority candidates unless tracker state // already represented them above. const bootstrapPeerSet = new Set(bootstrapPeers); let connectedFallbackAdded = 0; const connectedFallbackMax = 64; for (const h of this.peers.keys()) { if (h === this.publicKeyHash) continue; if (bootstrapPeerSet.has(h) && candidatesByHash.has(h)) continue; upsertCandidate({ hash: h, addrs: [], level: 0xffff, freeSlots: 0, bidPerByte: 0, source: 2, }); connectedFallbackAdded += 1; if (connectedFallbackAdded >= connectedFallbackMax) break; } const now = Date.now(); const allCandidates = [...candidatesByHash.values()] .filter((c) => c.hash !== this.publicKeyHash) .sort((a, b) => { if (a.level !== b.level) return a.level - b.level; if (a.freeSlots !== b.freeSlots) return b.freeSlots - a.freeSlots; if (a.bidPerByte !== b.bidPerByte) return b.bidPerByte - a.bidPerByte; if (a.source !== b.source) return a.source - b.source; return a.hash < b.hash ? -1 : a.hash > b.hash ? 1 : 0; }); const candidates = allCandidates.filter((c) => { const until = cooldownUntilByHash.get(c.hash); return until == null || until <= now; }); // Everything is in cooldown; sleep briefly (bounded) and retry. if (candidates.length === 0) { if (allCandidates.length > 0) { let nextAt = Number.POSITIVE_INFINITY; for (const c of allCandidates) { const until = cooldownUntilByHash.get(c.hash); if (until != null && until > now) nextAt = Math.min(nextAt, until); } const waitMs = nextAt !== Number.POSITIVE_INFINITY ? Math.max(1, nextAt - now) : retryMs; const capMs = Math.max( retryMs, trackerQueryIntervalMs > 0 ? trackerQueryIntervalMs : retryMs, ); await delay(Math.max(1, Math.min(waitMs, capMs)), { signal }); continue; } await delay(retryMs, { signal }); continue; } let ordered = [...candidates]; if (candidateScoringMode === "ranked-shuffle") { // Shuffle to spread load (only among the top ranked candidates). if (candidateShuffleTopK > 0 && ordered.length > 1) { const k = Math.min(candidateShuffleTopK, ordered.length); for (let i = k - 1; i > 0; i--) { const j = Math.floor(this.random() * (i + 1)); const tmp = ordered[i]!; ordered[i] = ordered[j]!; ordered[j] = tmp; } } } else if (candidateScoringMode === "weighted") { const wLevel = Number.isFinite(candidateScoringWeights.level) ? Math.max(0, candidateScoringWeights.level) : 0; const wSlots = Number.isFinite(candidateScoringWeights.freeSlots) ? Math.max(0, candidateScoringWeights.freeSlots) : 0; const wConnected = Number.isFinite(candidateScoringWeights.connected) ? Math.max(0, candidateScoringWeights.connected) : 0; const wBid = Number.isFinite(candidateScoringWeights.bidPerByte) ? Math.max(0, candidateScoringWeights.bidPerByte) : 0; const wSource = Number.isFinite(candidateScoringWeights.source) ? Math.max(0, candidateScoringWeights.source) : 0; const k = candidateShuffleTopK > 0 ? Math.min(candidateShuffleTopK, ordered.length) : 0; if (k > 1) { const head = ordered.slice(0, k); const tail = ordered.slice(k); const weightOf = (c: (typeof head)[number]) => { const level = Math.max(0, Math.floor(c.level)); const freeSlots = Math.max(0, Math.floor(c.freeSlots)); const bidPerByte = Math.max(0, Math.floor(c.bidPerByte)); const source = Math.max(0, Math.floor(c.source)); const connected = Boolean(this.peers.get(c.hash)); let weight = 1; if (wLevel > 0) weight *= 1 / (1 + wLevel * level); if (wSlots > 0) weight *= 1 + wSlots * freeSlots; if (wConnected > 0 && connected) weight *= 1 + wConnected; if (wBid > 0) weight *= 1 + wBid * bidPerByte; if (wSource > 0) weight *= 1 / (1 + wSource * source); return weight; }; const out: typeof head = []; const remaining = [...head]; while (remaining.length > 0) { let sum = 0; const weights: number[] = new Array(remaining.length); for (let i = 0; i < remaining.length; i++) { const w = weightOf(remaining[i]!); const v = Number.isFinite(w) ? Math.max(0, w) : 0; weights[i] = v; sum += v; } if (sum <= 0) { out.push(...remaining); break; } let r = this.random() * sum; let pick = 0; for (; pick < weights.length; pick++) { r -= weights[pick]!; if (r <= 0) break; } if (pick >= remaining.length) pick = remaining.length - 1; out.push(remaining[pick]!); remaining.splice(pick, 1); } ordered = out.concat(tail); } } const queue = ordered; const queued = new Set(queue.map((c) => c.hash)); const dialedNew = new Set(); let attempts = 0; for ( let i = 0; i < queue.length && i < JOIN_REJECT_REDIRECT_QUEUE_MAX; i++ ) { if (signal.aborted) break; if (attempts >= joinAttemptsPerRound) break; const c = queue[i]!; attempts += 1; const wasConnected = Boolean(this.peers.get(c.hash)); let dialOk = wasConnected; if (!dialOk && c.addrs.length > 0) { dialOk = await this.ensurePeerConnection( c.hash, c.addrs, bootstrapDialTimeoutMs, signal, ); } if (!dialOk) { try { await this.sendTrackerFeedback( ch, bootstrapPeers, c.hash, TRACKER_FEEDBACK_DIAL_FAILED, ); } catch { // ignore } if (candidateCooldownMs > 0) { cooldownUntilByHash.set( c.hash, Date.now() + candidateCooldownMs * 5, ); } continue; } if (!wasConnected) { dialedNew.add(c.hash); } const reqId = (this.random() * 0xffffffff) >>> 0; const res = await this.tryJoinOnce( ch, c.hash, reqId, joinReqTimeoutMs, signal, ); if (res.redirects && res.redirects.length > 0) { for (const r of res.redirects) { if (queue.length >= JOIN_REJECT_REDIRECT_QUEUE_MAX) break; if (!r?.hash) continue; if (r.hash === this.publicKeyHash) continue; if (!r.addrs || r.addrs.length === 0) continue; if (queued.has(r.hash)) continue; queued.add(r.hash); this.cacheKnownCandidateAddrs(ch, r.hash, r.addrs); queue.push({ hash: r.hash, addrs: r.addrs, level: 0xffff, freeSlots: 0, bidPerByte: 0, source: 3, }); } } if (res.ok) { try { await this.sendTrackerFeedback( ch, bootstrapPeers, c.hash, TRACKER_FEEDBACK_JOINED, ); } catch { // ignore } cooldownUntilByHash.delete(c.hash); break; } // Connection hygiene: if we dialed this candidate just for joining and it // didn't work out, drop it so large join storms don't accumulate thousands // of idle neighbours (important for large sims and long-running clients). if (!wasConnected && !bootstrapPeers.includes(c.hash)) { const stream = this.peers.get(c.hash); if (stream) { void this.components.connectionManager .closeConnections(stream.peerId) .catch(() => {}); } } if (res.timedOut) { this.noteJoinTimeout(ch, c.hash); try { await this.sendTrackerFeedback( ch, bootstrapPeers, c.hash, TRACKER_FEEDBACK_JOIN_TIMEOUT, ); } catch { // ignore } if (candidateCooldownMs > 0) { cooldownUntilByHash.set( c.hash, Date.now() + candidateCooldownMs * 2, ); } continue; } const rejectReason = res.rejectReason ?? 0; if (candidateCooldownMs > 0) { let factor = 2; if (rejectReason === JOIN_REJECT_LOW_BID) factor = 30; else if ( rejectReason === JOIN_REJECT_NO_CAPACITY || rejectReason === JOIN_REJECT_NOT_ATTACHED ) { factor = 1; } cooldownUntilByHash.set( c.hash, Date.now() + candidateCooldownMs * factor, ); } try { await this.sendTrackerFeedback( ch, bootstrapPeers, c.hash, TRACKER_FEEDBACK_JOIN_REJECT, rejectReason, ); } catch { // ignore } } if (ch.parent) { // Keep only the selected parent + bootstraps. Everything else we dialed is // best-effort and can be re-dialed if needed later. for (const h of dialedNew) { if (h === ch.parent) continue; if (bootstrapPeers.includes(h)) continue; const stream = this.peers.get(h); if (!stream) continue; void this.components.connectionManager .closeConnections(stream.peerId) .catch(() => {}); } continue; } await delay(retryMs, { signal }); } } finally { signal.clear?.(); } } private async maybeImproveParent( ch: ChannelState, options: { signal: AbortSignal; candidateShuffleTopK: number; candidateScoringMode: "ranked-shuffle" | "ranked-strict" | "weighted"; candidateScoringWeights: { level: number; freeSlots: number; connected: number; bidPerByte: number; source: number; }; joinAttemptsPerRound: number; joinReqTimeoutMs: number; parentUpgrade: ParentUpgradePolicy; trackerPeers?: string[]; }, ): Promise { const parentUpgrade = options.parentUpgrade; const currentParent = ch.parent; if (!currentParent || ch.closed || ch.isRoot || ch.level <= 1) return false; if ( parentUpgrade.mode === "shadow" && this.parentUpgradeShadowInFlightSuffixKey != null && this.parentUpgradeShadowInFlightSuffixKey !== ch.id.suffixKey ) { ch.metrics.reparentUpgradeSkipBudget += 1; return false; } const staleRootProbeRound = Math.max( 0, Math.floor(ch.parentUpgradeStaleRootProbeRound || 0), ); ch.parentUpgradeStaleRootProbeRound = Math.min( 0x7fffffff, staleRootProbeRound + 1, ); const upgradeRandom = (scope: string) => stableUnitInterval( `${ch.id.suffixKey}:${this.publicKeyHash}:${currentParent}:${scope}`, ); const clearFailedBackoff = () => { ch.parentUpgradeBackoffMs = 0; ch.parentUpgradeBackoffUntil = 0; }; const clearShadowInFlight = () => { if (this.parentUpgradeShadowInFlightSuffixKey === ch.id.suffixKey) { this.parentUpgradeShadowInFlightSuffixKey = undefined; } }; const resetShadow = () => { if (!ch.parentShadow) return; ch.parentShadow = undefined; clearShadowInFlight(); ch.metrics.parentShadowReset += 1; }; const deferParentUpgradeRetryUntilFreshData = (multiplier = 1) => { const minFreshMessages = Math.max( 1, parentUpgrade.shadow.dualPathMinMessages * Math.max(1, Math.floor(multiplier)), ); ch.parentUpgradeRetryAfterSeq = Math.max( ch.parentUpgradeRetryAfterSeq, ch.maxSeqSeen + minFreshMessages - 1, ); }; const applyFailedBackoff = () => { const minMs = parentUpgrade.failedBackoff.minMs; const maxMs = parentUpgrade.failedBackoff.maxMs; if (minMs <= 0 || maxMs <= 0) { ch.parentUpgradeBackoffMs = 0; ch.parentUpgradeBackoffUntil = 0; return; } const nextMs = ch.parentUpgradeBackoffMs > 0 ? Math.min(maxMs, Math.max(minMs, ch.parentUpgradeBackoffMs * 2)) : minMs; ch.parentUpgradeBackoffMs = nextMs; const jitter = 0.75 + upgradeRandom(`failed-backoff:${nextMs}`) * 0.5; ch.parentUpgradeBackoffUntil = Date.now() + Math.max(1, Math.floor(nextMs * jitter)); }; const isDirectNeighbor = (hash: string) => { const peer = this.peers.get(hash); if (!peer) return false; try { const conns = this.components.connectionManager.getConnections( peer.peerId, ) as Connection[] | undefined; return (conns?.length ?? 0) > 0; } catch { return peer.isReadable || peer.isWritable; } }; const candidatesByHash = new Map< string, { hash: string; addrs: Multiaddr[]; level: number; freeSlots: number; bidPerByte: number; source: number; } >(); const upsertCandidate = (c: { hash: string; addrs: Multiaddr[]; level: number; freeSlots: number; bidPerByte: number; source: number; }) => { const prev = candidatesByHash.get(c.hash); if (!prev) { candidatesByHash.set(c.hash, { ...c }); return; } if (prev.addrs.length === 0 && c.addrs.length > 0) prev.addrs = c.addrs; prev.level = Math.min(prev.level, c.level); prev.freeSlots = Math.max(prev.freeSlots, c.freeSlots); prev.bidPerByte = Math.max(prev.bidPerByte, c.bidPerByte); prev.source = Math.min(prev.source, c.source); }; const nonRootMinLevelGain = Math.max( parentUpgrade.minLevelGain, parentUpgrade.nonRootMinLevelGain, ); const rootMinLevelGain = Math.max( parentUpgrade.minLevelGain, parentUpgrade.rootMinLevelGain, ); const rootMinSubtreeGain = Math.max( parentUpgrade.minLevelGain, parentUpgrade.rootMinSubtreeGain, ); const levelGainFor = (parentLevel: number) => ch.level - (parentLevel + 1); const impactedPeersForRoot = () => Math.max(1, 1 + ch.children.size); const rootSubtreeGainFor = (parentLevel: number) => levelGainFor(parentLevel) * impactedPeersForRoot(); const isEnoughLevelGain = (hash: string, parentLevel: number) => { const levelGain = levelGainFor(parentLevel); if (hash !== ch.id.root) return levelGain >= nonRootMinLevelGain; if (levelGain >= rootMinLevelGain) return true; return rootSubtreeGainFor(parentLevel) >= rootMinSubtreeGain; }; let localChannelCount = 1; if (this.channelsBySuffixKey instanceof Map) { localChannelCount = 0; for (const localCh of this.channelsBySuffixKey.values()) { if (localCh.closed) continue; localChannelCount += 1; if (localChannelCount > 1) break; } } const singleChannelBranch = ch.children.size > 0 && localChannelCount <= 1; const multiChannelBranch = ch.children.size > 0 && localChannelCount > 1; const endedAndComplete = ch.endSeqExclusive > 0 && ch.missingSeqs.size === 0 && ch.nextExpectedSeq >= ch.endSeqExclusive; const singleChannelSettledLeaf = ch.children.size === 0 && localChannelCount <= 1 && endedAndComplete; const multiChannelSettledLeaf = ch.children.size === 0 && localChannelCount > 1 && endedAndComplete; const knownChannelPeerCount = Math.max(1, ch.channelPeers?.size ?? 1); const defaultStaleRootProbeProbability = 0.015625; const configuredStaleRootProbeProbabilityRaw = Number( parentUpgrade.staleRootProbeProbability, ); const staleRootProbeBaseProbability = Number.isFinite( configuredStaleRootProbeProbabilityRaw, ) ? Math.max(0, Math.min(1, configuredStaleRootProbeProbabilityRaw)) : defaultStaleRootProbeProbability; const defaultMultiChannelSettledLeafRootCandidate = parentUpgrade.leafOnly && multiChannelSettledLeaf && staleRootProbeBaseProbability <= defaultStaleRootProbeProbability; const hasDefaultMultiChannelLeafRootSignal = () => { if (!defaultMultiChannelSettledLeafRootCandidate) return true; if (!isEnoughLevelGain(ch.id.root, 0)) return false; if ((ch.parentDataLatencySamples ?? 0) < 8) return false; if ( (ch.parentDataLatencyEwmaMs ?? 0) < 64 && (ch.parentDataLatencyMaxMs ?? 0) < 256 ) { return false; } if (staleRootProbeBaseProbability <= 0) return false; const sampleProbability = Math.min(1, staleRootProbeBaseProbability); return ( stableUnitInterval( `${ch.id.suffixKey}:${this.publicKeyHash}:${ch.id.root}:multi-channel-leaf-root-signal:${ch.maxSeqSeen}`, ) < sampleProbability ); }; const defaultMultiChannelLeafRootSignal = hasDefaultMultiChannelLeafRootSignal(); if ( defaultMultiChannelSettledLeafRootCandidate && defaultMultiChannelLeafRootSignal && ch.id.root !== this.publicKeyHash && isDirectNeighbor(ch.id.root) && (options.trackerPeers?.length ?? 0) > 0 ) { const rootMinFreeSlots = parentUpgrade.rootMinFreeSlots; const cachedRoot = ch.cachedTrackerCandidates.find( (c) => c.hash === ch.id.root, ); if (!cachedRoot || cachedRoot.freeSlots < rootMinFreeSlots) { const refreshed = await this.queryTrackers( ch, options.trackerPeers ?? [], Math.max(16, ch.cachedTrackerCandidates.length), parentUpgrade.probe.timeoutMs, options.signal, ).catch((): TrackerCandidate[] => []); if (refreshed.length > 0) { ch.cachedTrackerCandidates = refreshed; } } } const configuredRejectCooldownMs = Math.max( 0, parentUpgrade.probe.rejectCooldownMs, ); const rejectCooldownMs = localChannelCount > 1 ? Math.max(configuredRejectCooldownMs, 20_000) : configuredRejectCooldownMs; const rejectCooldownMaxMs = Math.max( rejectCooldownMs, parentUpgrade.probe.rejectCooldownMaxMs, ); const clearProbeRejectBackoff = (candidateHash: string) => { ch.parentProbeRejectUntilByHash.delete(candidateHash); ch.parentProbeRejectBackoffMsByHash.delete(candidateHash); }; const rejectProbeCandidate = (candidateHash: string) => { if (rejectCooldownMs <= 0) return; const nextBackoffMs = (ch.parentProbeRejectBackoffMsByHash.get(candidateHash) ?? 0) > 0 ? Math.min( rejectCooldownMaxMs, Math.max( rejectCooldownMs, (ch.parentProbeRejectBackoffMsByHash.get(candidateHash) ?? 0) * 2, ), ) : rejectCooldownMs; ch.parentProbeRejectBackoffMsByHash.set(candidateHash, nextBackoffMs); const jitter = 0.75 + upgradeRandom(`probe-reject:${candidateHash}:${nextBackoffMs}`) * 0.5; ch.parentProbeRejectUntilByHash.set( candidateHash, Date.now() + Math.max(1, Math.floor(nextBackoffMs * jitter)), ); while ( ch.parentProbeRejectBackoffMsByHash.size > KNOWN_CANDIDATE_ADDRS_MAX_ENTRIES ) { const oldest = ch.parentProbeRejectBackoffMsByHash.keys().next() .value as string | undefined; if (!oldest) break; ch.parentProbeRejectBackoffMsByHash.delete(oldest); ch.parentProbeRejectUntilByHash.delete(oldest); } while ( ch.parentProbeRejectUntilByHash.size > KNOWN_CANDIDATE_ADDRS_MAX_ENTRIES ) { const oldest = ch.parentProbeRejectUntilByHash.keys().next().value as | string | undefined; if (!oldest) break; ch.parentProbeRejectUntilByHash.delete(oldest); ch.parentProbeRejectBackoffMsByHash.delete(oldest); } }; const minFreeSlotsFor = (hash: string) => hash === ch.id.root ? parentUpgrade.rootMinFreeSlots : parentUpgrade.minFreeSlots; const refreshCachedTrackerCandidate = ( hash: string, level: number, freeSlots: number, ) => { const cached = ch.cachedTrackerCandidates.find((c) => c.hash === hash); if (!cached) return; if (Number.isFinite(level)) cached.level = Math.max(0, Math.floor(level)); if (Number.isFinite(freeSlots)) { cached.freeSlots = Math.max(0, Math.floor(freeSlots)); } }; const sendProbeTrackerFeedback = ( candidateHash: string, reason: number, ) => { const trackerPeers = options.trackerPeers ?? []; if (trackerPeers.length === 0) return; void this.sendTrackerFeedback( ch, trackerPeers, candidateHash, TRACKER_FEEDBACK_JOIN_REJECT, reason, ).catch(() => {}); }; for (const c of ch.cachedTrackerCandidates) { if (c.hash === this.publicKeyHash) continue; if (!isDirectNeighbor(c.hash)) continue; upsertCandidate({ hash: c.hash, addrs: c.addrs, level: c.level, freeSlots: Number.isFinite(c.freeSlots) ? Math.max(0, Math.floor(c.freeSlots)) : 0, bidPerByte: c.bidPerByte, source: 0, }); } if ( ch.id.root !== this.publicKeyHash && isDirectNeighbor(ch.id.root) && !candidatesByHash.has(ch.id.root) ) { upsertCandidate({ hash: ch.id.root, addrs: [], level: 0, freeSlots: 1, bidPerByte: 0, source: -1, }); } let connectedFallbackAdded = 0; const connectedFallbackMax = 64; for (const h of this.peers.keys()) { if (h === this.publicKeyHash) continue; if (!isDirectNeighbor(h)) continue; upsertCandidate({ hash: h, addrs: [], level: 0xffff, freeSlots: 0, bidPerByte: 0, source: 2, }); connectedFallbackAdded += 1; if (connectedFallbackAdded >= connectedFallbackMax) break; } const orderedCandidates: Array<{ hash: string; addrs: Multiaddr[]; level: number; freeSlots: number; bidPerByte: number; source: number; }> = []; for (const c of candidatesByHash.values()) { if (c.hash === this.publicKeyHash || c.hash === currentParent) continue; const canVerifyRootCapacityLive = parentUpgrade.verifyStaleRootCapacity === true && parentUpgrade.mode === "shadow" && c.hash === ch.id.root; const requiredMinFreeSlots = minFreeSlotsFor(c.hash); if ( c.hash === ch.id.root && defaultMultiChannelSettledLeafRootCandidate && !defaultMultiChannelLeafRootSignal ) { ch.metrics.reparentUpgradeSkipCandidateSlots += 1; continue; } if (canVerifyRootCapacityLive && c.freeSlots < requiredMinFreeSlots) { const staleRootProbeProbability = (() => { if (singleChannelBranch && staleRootProbeBaseProbability > 0) { const highValueBranch = rootSubtreeGainFor(c.level) >= Math.max(rootMinSubtreeGain + 1, 4); const branchCap = highValueBranch ? 0.25 : 0.2; return Math.max( staleRootProbeBaseProbability, Math.min(branchCap, staleRootProbeBaseProbability * 8), ); } if (multiChannelBranch && staleRootProbeBaseProbability > 0) { return Math.max( staleRootProbeBaseProbability, Math.min(0.03125, staleRootProbeBaseProbability * 2), ); } if (singleChannelSettledLeaf && staleRootProbeBaseProbability > 0) { return 0.5; } if (multiChannelSettledLeaf && parentUpgrade.leafOnly) { return staleRootProbeBaseProbability > defaultStaleRootProbeProbability ? Math.max( staleRootProbeBaseProbability, Math.min(0.125, 3 / knownChannelPeerCount), ) : 0; } return Math.min( 1, staleRootProbeBaseProbability * Math.max(1, Math.min(4, staleRootProbeRound + 1)), ); })(); const staleRootProbeScore = stableUnitInterval( `${ch.id.suffixKey}:${this.publicKeyHash}:${c.hash}:stale-root-probe:${staleRootProbeRound}`, ); if (staleRootProbeScore >= staleRootProbeProbability) { ch.metrics.reparentUpgradeSkipCandidateSlots += 1; continue; } } if (!canVerifyRootCapacityLive && c.freeSlots < requiredMinFreeSlots) { ch.metrics.reparentUpgradeSkipCandidateSlots += 1; continue; } if (!isEnoughLevelGain(c.hash, c.level)) { ch.metrics.reparentUpgradeSkipCandidateLevel += 1; continue; } orderedCandidates.push(c); } let ordered = orderedCandidates.sort((a, b) => { if (a.level !== b.level) return a.level - b.level; if (a.freeSlots !== b.freeSlots) return b.freeSlots - a.freeSlots; if (a.bidPerByte !== b.bidPerByte) return b.bidPerByte - a.bidPerByte; if (a.source !== b.source) return a.source - b.source; return a.hash < b.hash ? -1 : a.hash > b.hash ? 1 : 0; }); if (ordered.length === 0) { if (parentUpgrade.mode === "probe" || parentUpgrade.mode === "shadow") { applyFailedBackoff(); } return false; } if (options.candidateScoringMode === "ranked-shuffle") { if (options.candidateShuffleTopK > 0 && ordered.length > 1) { const k = Math.min(options.candidateShuffleTopK, ordered.length); const shuffleScope = ordered.map((c) => c.hash).join(","); for (let i = k - 1; i > 0; i--) { const j = Math.floor( upgradeRandom(`shuffle:${shuffleScope}:${i}`) * (i + 1), ); const tmp = ordered[i]!; ordered[i] = ordered[j]!; ordered[j] = tmp; } } } else if (options.candidateScoringMode === "weighted") { const wLevel = Number.isFinite(options.candidateScoringWeights.level) ? Math.max(0, options.candidateScoringWeights.level) : 0; const wSlots = Number.isFinite(options.candidateScoringWeights.freeSlots) ? Math.max(0, options.candidateScoringWeights.freeSlots) : 0; const wConnected = Number.isFinite( options.candidateScoringWeights.connected, ) ? Math.max(0, options.candidateScoringWeights.connected) : 0; const wBid = Number.isFinite(options.candidateScoringWeights.bidPerByte) ? Math.max(0, options.candidateScoringWeights.bidPerByte) : 0; const wSource = Number.isFinite(options.candidateScoringWeights.source) ? Math.max(0, options.candidateScoringWeights.source) : 0; const k = options.candidateShuffleTopK > 0 ? Math.min(options.candidateShuffleTopK, ordered.length) : 0; const hasWeightedSignal = wLevel > 0 || wSlots > 0 || wConnected > 0 || wBid > 0 || wSource > 0; if (k > 1 && hasWeightedSignal) { const head = ordered.slice(0, k); const tail = ordered.slice(k); const weightOf = (c: (typeof head)[number]) => { const level = Math.max(0, Math.floor(c.level)); const freeSlots = Math.max(0, Math.floor(c.freeSlots)); const bidPerByte = Math.max(0, Math.floor(c.bidPerByte)); const source = Math.max(0, Math.floor(c.source)); const connected = Boolean(this.peers.get(c.hash)); let weight = 1; if (wLevel > 0) weight *= 1 / (1 + wLevel * level); if (wSlots > 0) weight *= 1 + wSlots * freeSlots; if (wConnected > 0 && connected) weight *= 1 + wConnected; if (wBid > 0) weight *= 1 + wBid * bidPerByte; if (wSource > 0) weight *= 1 / (1 + wSource * source); return weight; }; const out: typeof head = []; const remaining = [...head]; while (remaining.length > 0) { let sum = 0; const weights: number[] = new Array(remaining.length); for (let i = 0; i < remaining.length; i++) { const w = weightOf(remaining[i]!); const v = Number.isFinite(w) ? Math.max(0, w) : 0; weights[i] = v; sum += v; } if (sum <= 0) { out.push(...remaining); break; } let r = upgradeRandom( `weighted:${out.length}:${remaining .map((candidate) => candidate.hash) .join(",")}`, ) * sum; let pick = 0; for (; pick < weights.length; pick++) { r -= weights[pick]!; if (r <= 0) break; } if (pick >= remaining.length) pick = remaining.length - 1; out.push(remaining[pick]!); remaining.splice(pick, 1); } ordered = out.concat(tail); } } if (parentUpgrade.mode === "probe" || parentUpgrade.mode === "shadow") { const mode = parentUpgrade.mode; const probeLimit = parentUpgrade.mode === "shadow" ? 1 : parentUpgrade.probe.maxPerRound; const shadowObserveMsForGate = parentUpgrade.shadow.observeMs; const shadowMinObservationsForGate = parentUpgrade.shadow.minObservations; const probeTimeoutMs = parentUpgrade.probe.timeoutMs; const maxLag = parentUpgrade.probe.maxLagMessages; const now = Date.now(); const candidatesToProbe: typeof ordered = []; const shadowCandidate = mode === "shadow" && ch.parentShadow ? ordered.find( (candidate) => candidate.hash === ch.parentShadow?.hash, ) : undefined; if ( mode === "shadow" && shadowCandidate && ch.parentShadow && shadowObserveMsForGate > 0 && now - ch.parentShadow.startedAt < shadowObserveMsForGate ) { return false; } const probeOrder = shadowCandidate ? [ shadowCandidate, ...ordered.filter( (candidate) => candidate.hash !== shadowCandidate.hash, ), ] : ordered; for (const candidate of probeOrder) { const rejectUntil = ch.parentProbeRejectUntilByHash.get(candidate.hash); if (rejectUntil != null && rejectUntil > now) { ch.metrics.reparentUpgradeSkipProbeCooldown += 1; continue; } if (rejectUntil != null) ch.parentProbeRejectUntilByHash.delete(candidate.hash); candidatesToProbe.push(candidate); if (candidatesToProbe.length >= probeLimit) break; } if (candidatesToProbe.length === 0) { applyFailedBackoff(); return false; } const shouldReserveRootCapacityForProbe = ( candidate: (typeof ordered)[number], ) => { if (candidate.hash !== ch.id.root) return true; if (mode !== "shadow") return true; if (endedAndComplete && candidate.level === 0) return true; if (!ch.parentShadow || ch.parentShadow.hash !== candidate.hash) { return ( shadowObserveMsForGate <= 0 && shadowMinObservationsForGate <= 1 ); } return ( now - ch.parentShadow.startedAt >= shadowObserveMsForGate && ch.parentShadow.observations + 1 >= shadowMinObservationsForGate ); }; const rejectShadowCandidate = ( candidateHash: string, reason: | "noReply" | "notRooted" | "capacity" | "repair" | "lag" | "overloaded" | "level", ) => { if (mode !== "shadow") return; switch (reason) { case "noReply": ch.metrics.parentShadowRejectNoReply += 1; break; case "notRooted": ch.metrics.parentShadowRejectNotRooted += 1; break; case "capacity": ch.metrics.parentShadowRejectCapacity += 1; break; case "repair": ch.metrics.parentShadowRejectRepair += 1; break; case "lag": ch.metrics.parentShadowRejectLag += 1; break; case "overloaded": ch.metrics.parentShadowRejectOverloaded += 1; break; case "level": ch.metrics.parentShadowRejectLevel += 1; break; } if (ch.parentShadow?.hash === candidateHash) resetShadow(); deferParentUpgradeRetryUntilFreshData(); }; const probed = await Promise.all( candidatesToProbe.map(async (candidate) => ({ candidate, reply: await this.probeParentCandidate( ch, candidate.hash, probeTimeoutMs, options.signal, minFreeSlotsFor(candidate.hash), shouldReserveRootCapacityForProbe(candidate), ), })), ); type LiveParentCandidate = (typeof ordered)[number] & { haveToExclusive: number; reservationToken: number; }; const liveCandidates: LiveParentCandidate[] = []; for (const { candidate, reply } of probed) { if (!reply) { ch.metrics.reparentUpgradeSkipProbeNoReply += 1; rejectShadowCandidate(candidate.hash, "noReply"); rejectProbeCandidate(candidate.hash); continue; } if (!reply.rooted) { ch.metrics.reparentUpgradeSkipProbeNotRooted += 1; refreshCachedTrackerCandidate(candidate.hash, reply.level, 0); sendProbeTrackerFeedback(candidate.hash, JOIN_REJECT_NOT_ATTACHED); rejectShadowCandidate(candidate.hash, "notRooted"); rejectProbeCandidate(candidate.hash); continue; } const liveLevel = Number.isFinite(reply.level) ? Math.max(0, Math.floor(reply.level)) : candidate.level; const liveFreeSlots = Number.isFinite(reply.freeSlots) ? Math.max(0, Math.floor(reply.freeSlots)) : 0; refreshCachedTrackerCandidate(candidate.hash, liveLevel, liveFreeSlots); const requiredMinFreeSlots = minFreeSlotsFor(candidate.hash); if (!reply.accepting || liveFreeSlots < requiredMinFreeSlots) { ch.metrics.reparentUpgradeSkipCandidateSlots += 1; if (!reply.accepting || liveFreeSlots <= 0) { sendProbeTrackerFeedback(candidate.hash, JOIN_REJECT_NO_CAPACITY); } rejectShadowCandidate(candidate.hash, "capacity"); rejectProbeCandidate(candidate.hash); continue; } const configuredMaxChildLoadRatioRaw = candidate.hash === ch.id.root ? parentUpgrade.rootMaxChildLoadRatio : parentUpgrade.maxChildLoadRatio; const configuredMaxChildLoadRatio = candidate.hash === ch.id.root && localChannelCount > 1 && Number(configuredMaxChildLoadRatioRaw) > 0 ? Math.min(0.2, Number(configuredMaxChildLoadRatioRaw)) : configuredMaxChildLoadRatioRaw; const maxChildLoadRatio = Math.max( 0, Number(configuredMaxChildLoadRatio), ); if ( maxChildLoadRatio > 0 && reply.maxChildren > 0 && (reply.children + 1) / reply.maxChildren > maxChildLoadRatio ) { ch.metrics.reparentUpgradeSkipCandidateSlots += 1; ch.metrics.reparentUpgradeSkipCandidatePressure += 1; if (candidate.hash === ch.id.root) { ch.metrics.reparentUpgradeSkipRootPressure += 1; refreshCachedTrackerCandidate(candidate.hash, liveLevel, 0); sendProbeTrackerFeedback(candidate.hash, JOIN_REJECT_NO_CAPACITY); } rejectShadowCandidate(candidate.hash, "capacity"); rejectProbeCandidate(candidate.hash); continue; } if (reply.repairing || reply.missingSeqs > 0) { ch.metrics.reparentUpgradeSkipProbeRepair += 1; rejectShadowCandidate(candidate.hash, "repair"); rejectProbeCandidate(candidate.hash); continue; } if (reply.overloaded) { ch.metrics.reparentUpgradeSkipProbeOverloaded += 1; rejectShadowCandidate(candidate.hash, "overloaded"); rejectProbeCandidate(candidate.hash); continue; } if ( ch.maxSeqSeen >= 0 && reply.haveToExclusive + maxLag < ch.maxSeqSeen + 1 ) { ch.metrics.reparentUpgradeSkipProbeLag += 1; rejectShadowCandidate(candidate.hash, "lag"); rejectProbeCandidate(candidate.hash); continue; } if (!isEnoughLevelGain(candidate.hash, liveLevel)) { ch.metrics.reparentUpgradeSkipCandidateLevel += 1; rejectShadowCandidate(candidate.hash, "level"); rejectProbeCandidate(candidate.hash); continue; } clearProbeRejectBackoff(candidate.hash); liveCandidates.push({ ...candidate, level: liveLevel, freeSlots: liveFreeSlots, haveToExclusive: reply.haveToExclusive, reservationToken: reply.reservationToken, }); } ordered = liveCandidates.sort((a, b) => { if (a.level !== b.level) return a.level - b.level; if (a.freeSlots !== b.freeSlots) return b.freeSlots - a.freeSlots; if (a.bidPerByte !== b.bidPerByte) return b.bidPerByte - a.bidPerByte; if (a.source !== b.source) return a.source - b.source; return a.hash < b.hash ? -1 : a.hash > b.hash ? 1 : 0; }); if (ordered.length === 0) { applyFailedBackoff(); return false; } clearFailedBackoff(); if (mode === "shadow") { const nowAfterProbe = Date.now(); const observeMs = parentUpgrade.shadow.observeMs; const minObservations = shadowMinObservationsForGate; const requireDataForShadowPromotion = parentUpgrade.shadow.dualPathMs > 0; const selected = ch.parentShadow != null ? (ordered.find( (candidate) => candidate.hash === ch.parentShadow?.hash, ) ?? ordered[0]!) : ordered[0]!; if (!selected) { resetShadow(); return false; } if ( endedAndComplete && selected.hash === ch.id.root && selected.level === 0 && !requireDataForShadowPromotion ) { if (!ch.parentShadow || ch.parentShadow.hash !== selected.hash) { ch.parentShadow = { hash: selected.hash, startedAt: nowAfterProbe - observeMs, observations: minObservations, level: selected.level, freeSlots: selected.freeSlots, haveToExclusive: "haveToExclusive" in selected && typeof selected.haveToExclusive === "number" ? selected.haveToExclusive : 0, liveDataMessages: 0, liveFirstDataMessages: 0, liveParentFirstDataMessages: 0, liveCandidateLeadSamples: 0, liveCandidateLeadMsTotal: 0, liveCandidateFirstAtBySeq: new Map(), liveParentFirstAtBySeq: new Map(), liveComparedSeqs: new Set(), liveMaxSeqSeen: -1, liveLastDataAt: 0, }; this.parentUpgradeShadowInFlightSuffixKey = ch.id.suffixKey; ch.metrics.parentShadowStart += 1; } else { ch.parentShadow.observations += 1; ch.parentShadow.level = selected.level; ch.parentShadow.freeSlots = selected.freeSlots; ch.parentShadow.haveToExclusive = "haveToExclusive" in selected && typeof selected.haveToExclusive === "number" ? selected.haveToExclusive : 0; ch.metrics.parentShadowObserve += 1; } ordered = [selected]; } else { if (!ch.parentShadow || ch.parentShadow.hash !== selected.hash) { ch.parentShadow = { hash: selected.hash, startedAt: nowAfterProbe, observations: 1, level: selected.level, freeSlots: selected.freeSlots, haveToExclusive: "haveToExclusive" in selected && typeof selected.haveToExclusive === "number" ? selected.haveToExclusive : 0, liveDataMessages: 0, liveFirstDataMessages: 0, liveParentFirstDataMessages: 0, liveCandidateLeadSamples: 0, liveCandidateLeadMsTotal: 0, liveCandidateFirstAtBySeq: new Map(), liveParentFirstAtBySeq: new Map(), liveComparedSeqs: new Set(), liveMaxSeqSeen: -1, liveLastDataAt: 0, }; this.parentUpgradeShadowInFlightSuffixKey = ch.id.suffixKey; ch.metrics.parentShadowStart += 1; return false; } ch.parentShadow.observations += 1; ch.parentShadow.level = selected.level; ch.parentShadow.freeSlots = selected.freeSlots; ch.parentShadow.haveToExclusive = "haveToExclusive" in selected && typeof selected.haveToExclusive === "number" ? selected.haveToExclusive : 0; ch.metrics.parentShadowObserve += 1; if ( ch.parentShadow.observations < minObservations || nowAfterProbe - ch.parentShadow.startedAt < observeMs ) { return false; } } ordered = [selected]; } } let attempts = 0; for (const candidate of ordered) { if (options.signal.aborted) break; if (attempts >= options.joinAttemptsPerRound) break; attempts += 1; if (!isDirectNeighbor(candidate.hash)) continue; const previousParent = ch.parent; const previousLevel = ch.level; const previousRouteFromRoot = ch.routeFromRoot ? [...ch.routeFromRoot] : undefined; const previousLastParentDataAt = ch.lastParentDataAt; const previousLastParentUpgradeActivityAt = ch.lastParentUpgradeActivityAt; const previousReceivedAnyParentData = ch.receivedAnyParentData; const shadowDualPathMs = parentUpgrade.shadow.dualPathMs; const shadowDualPathMinMessages = parentUpgrade.shadow.dualPathMinMessages; const shadowDualPathMinAdvantageMessages = Math.max( 1, Math.min(16, Math.ceil(shadowDualPathMinMessages / 2)), ); const shadowDualPathMinLeadSamples = Math.max( 1, Math.min(32, shadowDualPathMinMessages), ); const shadowDualPathMaxParentFirstMessages = Math.max( 0, Math.floor(shadowDualPathMinMessages / 8), ); const shadowDualPathMinLeadMsAvg = shadowDualPathMinMessages <= 1 ? 0 : 128; const useShadowDualPath = parentUpgrade.mode === "shadow" && shadowDualPathMs > 0 && previousParent != null && previousParent !== candidate.hash; const seqAtShadowAttach = ch.maxSeqSeen; const reqId = (this.random() * 0xffffffff) >>> 0; const res = await this.tryJoinOnce( ch, candidate.hash, reqId, options.joinReqTimeoutMs, options.signal, { allowReplace: true, parentUpgradeReservationToken: candidate.hash === ch.id.root && "reservationToken" in candidate && typeof candidate.reservationToken === "number" ? candidate.reservationToken : 0, shadowAttach: useShadowDualPath, }, ); if (res.ok) { if (useShadowDualPath) { const acceptedParentLevel = Number.isFinite(res.parentLevel) ? Math.max(0, Math.floor(res.parentLevel!)) : candidate.level; if (!isEnoughLevelGain(candidate.hash, acceptedParentLevel)) { void this._sendControl( candidate.hash, this.codec.encodeLeave(ch.id.key), ).catch(() => {}); resetShadow(); rejectProbeCandidate(candidate.hash); deferParentUpgradeRetryUntilFreshData(2); applyFailedBackoff(); return false; } const shadow = ch.parentShadow; if (shadow?.hash === candidate.hash) { shadow.liveDataMessages = 0; shadow.liveFirstDataMessages = 0; shadow.liveParentFirstDataMessages = 0; shadow.liveCandidateLeadSamples = 0; shadow.liveCandidateLeadMsTotal = 0; shadow.liveCandidateFirstAtBySeq.clear(); shadow.liveParentFirstAtBySeq.clear(); shadow.liveComparedSeqs.clear(); shadow.liveMaxSeqSeen = seqAtShadowAttach; shadow.liveLastDataAt = 0; } const deadline = Date.now() + shadowDualPathMs; for (;;) { if (options.signal.aborted) break; const currentShadow = ch.parentShadow; const observedFreshCandidateData = currentShadow?.hash === candidate.hash && currentShadow.liveFirstDataMessages >= shadowDualPathMinMessages && currentShadow.liveParentFirstDataMessages <= shadowDualPathMaxParentFirstMessages && currentShadow.liveFirstDataMessages >= currentShadow.liveParentFirstDataMessages + shadowDualPathMinAdvantageMessages && currentShadow.liveCandidateLeadSamples >= shadowDualPathMinLeadSamples && currentShadow.liveCandidateLeadMsTotal >= currentShadow.liveCandidateLeadSamples * shadowDualPathMinLeadMsAvg && currentShadow.liveMaxSeqSeen > seqAtShadowAttach; if (observedFreshCandidateData) { const parentRoute = res.parentRouteFromRoot; if (!parentRoute || parentRoute.length === 0) break; ch.parent = candidate.hash; ch.level = Math.max( 1, Math.floor(res.parentLevel ?? candidate.level) + 1, ); ch.routeFromRoot = [...parentRoute, this.publicKeyHash]; ch.joinedAtLeastOnce = true; ch.lastParentDataAt = currentShadow.liveLastDataAt > 0 ? currentShadow.liveLastDataAt : Date.now(); ch.lastParentUpgradeActivityAt = ch.lastParentDataAt; ch.receivedAnyParentData = true; this.resetParentDataLatency(ch); this.touchPeerHint(ch, candidate.hash); if (previousParent) { this.startParentUpgradeGrace(ch, { previousParent, candidateParent: candidate.hash, previousLevel, previousRouteFromRoot, previousLastParentDataAt, previousLastParentUpgradeActivityAt, previousReceivedAnyParentData, candidateFirstMessages: 0, previousFirstMessages: 0, minCandidateFirstMessages: shadowDualPathMinMessages, minCandidateAdvantageMessages: shadowDualPathMinAdvantageMessages, maxPreviousFirstMessages: shadowDualPathMaxParentFirstMessages, timeoutMs: shadowDualPathMs, }); } ch.metrics.parentShadowPromote += 1; ch.parentShadow = undefined; clearShadowInFlight(); clearFailedBackoff(); void this.announceToTrackers( ch, this.closeController.signal, ).catch(() => {}); return true; } const remainingMs = deadline - Date.now(); if (remainingMs <= 0) break; await delay(Math.min(50, remainingMs), { signal: options.signal, }); } if (ch.parent === previousParent) { ch.level = previousLevel; ch.routeFromRoot = previousRouteFromRoot; ch.lastParentDataAt = previousLastParentDataAt; ch.lastParentUpgradeActivityAt = previousLastParentUpgradeActivityAt; ch.receivedAnyParentData = previousReceivedAnyParentData; } void this._sendControl(candidate.hash, this.codec.encodeLeave(ch.id.key)).catch( () => {}, ); if (ch.parentShadow) { ch.parentShadow = undefined; clearShadowInFlight(); ch.metrics.parentShadowReset += 1; } // A shadow cutover can only be proven by fresh data. If no // candidate-first data arrived, wait for enough newer data before // spending more probes on the same channel. Also cool down this // candidate; accepting JOIN without proving live data is a weak // parent signal, not a reason to probe it repeatedly. rejectProbeCandidate(candidate.hash); deferParentUpgradeRetryUntilFreshData(2); applyFailedBackoff(); return false; } const newParent = ch.parent; if (previousParent && newParent && newParent !== previousParent) { void this._sendControl(previousParent, this.codec.encodeLeave(ch.id.key)).catch( () => {}, ); if (parentUpgrade.mode === "shadow") { ch.metrics.parentShadowPromote += 1; ch.parentShadow = undefined; clearShadowInFlight(); clearFailedBackoff(); } return true; } return false; } if (useShadowDualPath) { if (ch.parent === previousParent) { ch.level = previousLevel; ch.routeFromRoot = previousRouteFromRoot; ch.lastParentDataAt = previousLastParentDataAt; ch.lastParentUpgradeActivityAt = previousLastParentUpgradeActivityAt; ch.receivedAnyParentData = previousReceivedAnyParentData; } if (ch.parentShadow?.hash === candidate.hash) { ch.parentShadow = undefined; clearShadowInFlight(); ch.metrics.parentShadowReset += 1; } rejectProbeCandidate(candidate.hash); deferParentUpgradeRetryUntilFreshData(2); applyFailedBackoff(); return false; } } return false; } private async tryJoinOnce( ch: ChannelState, parentHash: string, reqId: number, timeoutMs: number, signal: AbortSignal, options?: { allowReplace?: boolean; parentUpgradeReservationToken?: number; shadowAttach?: boolean; }, ): Promise { if (ch.parent && options?.allowReplace !== true) return { ok: true }; if (!this.peers.get(parentHash)) return { ok: false, timedOut: true }; const p = new Promise((resolve) => { ch.pendingJoin.set(reqId, { resolve, shadowAttach: options?.shadowAttach === true, }); }); await this._sendControl( parentHash, this.codec.encodeJoinReq( ch.id.key, reqId, ch.bidPerByte, options?.parentUpgradeReservationToken, ), ); const res = await Promise.race([ p, delay(Math.max(1, timeoutMs), { signal }).then( (): JoinAttemptResult => ({ ok: false, timedOut: true, }), ), ]); if (res.timedOut) ch.pendingJoin.delete(reqId); return res; } private async kickChildHashes( ch: ChannelState, children: string[], options?: { resetPeerConnections?: boolean }, ) { const unique = [...new Set(children)].filter((h) => ch.children.has(h)); for (const h of unique) { ch.children.delete(h); ch.dataWriteFailStreakByChild.delete(h); } if (unique.length === 0) return; const resetPeerConnections = options?.resetPeerConnections === true; let kickFailed = false; try { const kickDelivered = await this._sendControlMany( unique, this.codec.encodeKick(ch.id.key), ); kickFailed = !kickDelivered; } catch (error) { kickFailed = true; throw error; } finally { if (resetPeerConnections && kickFailed) { // If KICK cannot be written on the existing stream, close the peer // connection so the topology change becomes observable. for (const h of unique) { const stream = this.peers.get(h); if (!stream) continue; void this.components.connectionManager .closeConnections(stream.peerId) .catch(() => {}); } } } } private async kickChildren(ch: ChannelState) { await this.kickChildHashes(ch, [...ch.children.keys()]); } public async onDataMessage( from: any, peerStream: PeerStreams, message: DataMessage, seenBefore: number, ) { const ignore = this.shouldIgnore(message, seenBefore); const raw = message.data as Uint8ArrayList | Uint8Array | undefined; const data = raw ? toUint8Array(raw) : undefined; if (!data || data.length === 0) return false; const kind = data[0]!; const fromHash = from.hashcode(); // Control-plane + tracker messages carry channelKey (32 bytes) at offset 1. if ( kind === MSG_JOIN_REQ || kind === MSG_JOIN_ACCEPT || kind === MSG_JOIN_REJECT || kind === MSG_KICK || kind === MSG_END || kind === MSG_UNICAST || kind === MSG_UNICAST_ACK || kind === MSG_ROUTE_QUERY || kind === MSG_ROUTE_REPLY || kind === MSG_PUBLISH_PROXY || kind === MSG_LEAVE || kind === MSG_REPAIR_REQ || kind === MSG_FETCH_REQ || kind === MSG_IHAVE || kind === MSG_TRACKER_ANNOUNCE || kind === MSG_TRACKER_QUERY || kind === MSG_TRACKER_REPLY || kind === MSG_TRACKER_FEEDBACK || kind === MSG_PARENT_PROBE_REQ || kind === MSG_PARENT_PROBE_REPLY || kind === MSG_PROVIDER_ANNOUNCE || kind === MSG_PROVIDER_QUERY || kind === MSG_PROVIDER_REPLY || kind === MSG_PROVIDER_SUBSCRIBE || kind === MSG_PROVIDER_UNSUBSCRIBE || kind === MSG_PROVIDER_NOTIFY ) { if (data.length < 1 + 32) return false; const channelKey = data.subarray(1, 33); const suffixKey = toBase64(channelKey.subarray(0, 24)); this.recordControlReceive(suffixKey, kind, data.byteLength); const ch = this.channelsBySuffixKey.get(suffixKey); // DirectStream de-duplicates by message-id, but FanoutTree unicast (and its ACK) // legitimately reflect "up to root, then down" through the same top-level branch. // Allow a duplicate only when it arrives from the parent (downstream reflection). if ( ignore && !( (kind === MSG_UNICAST || kind === MSG_UNICAST_ACK) && ch && fromHash === ch.parent ) ) { return false; } if (!ch && kind === MSG_JOIN_REQ && data.length >= 1 + 32 + 4) { // A joiner probed us for a channel we do not host (connected-peer // fallbacks routinely hit bootstraps and non-subscribers). Failing // fast lets the joiner spend its probe budget elsewhere instead of // burning a full join-request timeout and a long cooldown on us. const reqId = readU32BE(data, 33); void this._sendControl( fromHash, this.codec.encodeJoinReject(channelKey, reqId, JOIN_REJECT_NOT_ATTACHED), ).catch(() => {}); return true; } if (kind === MSG_TRACKER_ANNOUNCE) { const decoded = this.codec.decodeTrackerAnnounce(data); if (!decoded) return false; const { ttlMs, level, freeSlots, bidPerByte, addrs } = decoded; if (addrs.length === 0) { try { const peer: any = await this.components.peerStore.get( peerStream.peerId, ); const addresses: any[] = Array.isArray(peer?.addresses) ? peer.addresses : []; for (const a of addresses) { const ma: any = a?.multiaddr ?? a; const bytes = normalizeUint8Array(ma?.bytes); if (bytes) addrs.push(bytes); if (addrs.length >= 16) break; } } catch { // ignore } } let byPeer = this.trackerBySuffixKey.get(suffixKey); if (!byPeer) { byPeer = new Map(); this.trackerBySuffixKey.set(suffixKey, byPeer); } const now = Date.now(); this.touchTrackerNamespace(suffixKey, now); const ttl = Math.min(ttlMs, 120_000); if (ttl === 0) { byPeer.delete(fromHash); this.pruneTrackerNamespaceIfEmpty(suffixKey); return true; } // LRU by announce freshness, with a hard per-channel cap. byPeer.delete(fromHash); byPeer.set(fromHash, { hash: fromHash, level, freeSlots, bidPerByte, addrs, expiresAt: now + ttl, }); while (byPeer.size > TRACKER_DIRECTORY_MAX_ENTRIES) { const oldest = byPeer.keys().next().value as string | undefined; if (!oldest) break; byPeer.delete(oldest); } return true; } if (kind === MSG_TRACKER_QUERY) { const decoded = this.codec.decodeTrackerQuery(data); if (!decoded) return false; const { reqId, want } = decoded; const now = Date.now(); this.touchTrackerNamespace(suffixKey, now); const byPeer = this.trackerBySuffixKey.get(suffixKey); const entries: TrackerEntry[] = []; if (byPeer) { for (const [hash, e] of byPeer) { if (e.expiresAt <= now) { byPeer.delete(hash); continue; } if (hash === fromHash) continue; if (e.freeSlots <= 0) { continue; } entries.push(e); } this.pruneTrackerNamespaceIfEmpty(suffixKey); } entries.sort((a, b) => { if (a.level !== b.level) return a.level - b.level; if (a.freeSlots !== b.freeSlots) return b.freeSlots - a.freeSlots; if (a.bidPerByte !== b.bidPerByte) return b.bidPerByte - a.bidPerByte; return a.hash < b.hash ? -1 : a.hash > b.hash ? 1 : 0; }); // Returning the deterministic head of the directory sends every // concurrent joiner to the same few candidates while deeper capacity // goes unprobed. Sample the reply from a wider best-first pool so // load spreads across all advertised capacity. const wantCount = clampU16(want); const poolSize = Math.min(entries.length, Math.max(wantCount * 4, wantCount)); const pool = entries.slice(0, poolSize); for (let i = 0; i < Math.min(wantCount, pool.length); i++) { const j = i + Math.floor(this.random() * (pool.length - i)); const tmp = pool[i]!; pool[i] = pool[j]!; pool[j] = tmp; } const picked = pool.slice(0, wantCount); void this._sendControl(fromHash, this.codec.encodeTrackerReply(channelKey, reqId, picked)); return true; } if (kind === MSG_TRACKER_FEEDBACK) { const decoded = this.codec.decodeTrackerFeedback(data); if (!decoded) return false; const { candidateHash, event, reason } = decoded; const byPeer = this.trackerBySuffixKey.get(suffixKey); if (!byPeer) return true; this.touchTrackerNamespace(suffixKey); const entry = byPeer.get(candidateHash); if (!entry) return true; const now = Date.now(); if (entry.expiresAt <= now) { byPeer.delete(candidateHash); this.pruneTrackerNamespaceIfEmpty(suffixKey); return true; } if (event === TRACKER_FEEDBACK_JOINED) { entry.freeSlots = clampU16(Math.max(0, entry.freeSlots - 1)); return true; } if ( event === TRACKER_FEEDBACK_DIAL_FAILED || event === TRACKER_FEEDBACK_JOIN_TIMEOUT ) { byPeer.delete(candidateHash); this.pruneTrackerNamespaceIfEmpty(suffixKey); return true; } if (event === TRACKER_FEEDBACK_JOIN_REJECT) { if ( reason === JOIN_REJECT_NO_CAPACITY || reason === JOIN_REJECT_NOT_ATTACHED ) { entry.freeSlots = 0; // Expire earlier to reduce time-to-recovery if the entry is stale. entry.expiresAt = Math.min(entry.expiresAt, now + 2_000); } return true; } return true; } if (kind === MSG_PROVIDER_ANNOUNCE) { const decoded = this.codec.decodeProviderAnnounce(data); if (!decoded) return false; const providerId = this.getProviderNamespaceIdFromKey( channelKey, suffixKey, ); const { ttlMs, addrs } = decoded; if (addrs.length === 0) { try { const peer: any = await this.components.peerStore.get( peerStream.peerId, ); const addresses: any[] = Array.isArray(peer?.addresses) ? peer.addresses : []; for (const a of addresses) { const ma: any = a?.multiaddr ?? a; const bytes = normalizeUint8Array(ma?.bytes); if (bytes) addrs.push(bytes); if (addrs.length >= 16) break; } } catch { // ignore } } let byPeer = this.providerBySuffixKey.get(suffixKey); if (!byPeer) { byPeer = new Map(); this.providerBySuffixKey.set(suffixKey, byPeer); } const now = Date.now(); this.touchProviderNamespace(suffixKey, now); const ttl = Math.min(ttlMs, 120_000); if (ttl === 0) { byPeer.delete(fromHash); this.pruneProviderNamespaceIfEmpty(suffixKey); this.emitProviderUpdate( providerId, this.getProviderCandidatesFromCache(providerId, { now }), ); await this.publishProviderWatchUpdate(providerId); return true; } // LRU by announce freshness, with a hard per-namespace cap. byPeer.delete(fromHash); byPeer.set(fromHash, { hash: fromHash, addrs, expiresAt: now + ttl, }); while (byPeer.size > PROVIDER_DIRECTORY_MAX_ENTRIES) { const oldest = byPeer.keys().next().value as string | undefined; if (!oldest) break; byPeer.delete(oldest); } this.pruneProviderNamespaceIfEmpty(suffixKey); this.emitProviderUpdate( providerId, this.getProviderCandidatesFromCache(providerId, { now }), ); await this.publishProviderWatchUpdate(providerId); return true; } if (kind === MSG_PROVIDER_QUERY) { const decoded = this.codec.decodeProviderQuery(data); if (!decoded) return false; const { reqId, want, seed } = decoded; const now = Date.now(); this.touchProviderNamespace(suffixKey, now); const byPeer = this.providerBySuffixKey.get(suffixKey); const entries: ProviderEntry[] = []; if (byPeer) { for (const [hash, e] of byPeer) { if (e.expiresAt <= now) { byPeer.delete(hash); continue; } if (hash === fromHash) continue; entries.push(e); } this.pruneProviderNamespaceIfEmpty(suffixKey); } // Shuffle to spread load (deterministic if seed is provided). if (entries.length > 1) { if (seed !== 0) { let x = seed >>> 0; for (let i = entries.length - 1; i > 0; i--) { x ^= x << 13; x ^= x >>> 17; x ^= x << 5; const j = (x >>> 0) % (i + 1); const tmp = entries[i]!; entries[i] = entries[j]!; entries[j] = tmp; } } else { for (let i = entries.length - 1; i > 0; i--) { const j = Math.floor(this.random() * (i + 1)); const tmp = entries[i]!; entries[i] = entries[j]!; entries[j] = tmp; } } } const picked = entries.slice(0, clampU16(want)); void this._sendControl( fromHash, this.codec.encodeProviderReply(channelKey, reqId, picked), ); return true; } if (kind === MSG_PROVIDER_SUBSCRIBE) { const decoded = this.codec.decodeProviderSubscribe(data); if (!decoded) return false; const want = Math.max(1, decoded.want); const ttlMs = Math.min(120_000, Math.max(1_000, decoded.ttlMs)); const now = Date.now(); let watchers = this.providerWatchersBySuffixKey.get(suffixKey); if (!watchers) { watchers = new Map(); this.providerWatchersBySuffixKey.set(suffixKey, watchers); } watchers.set(fromHash, { hash: fromHash, want, expiresAt: now + ttlMs, }); await this.publishProviderWatchUpdate( this.getProviderNamespaceIdFromKey(channelKey, suffixKey), ); return true; } if (kind === MSG_PROVIDER_UNSUBSCRIBE) { const watchers = this.providerWatchersBySuffixKey.get(suffixKey); watchers?.delete(fromHash); this.pruneProviderWatchersIfEmpty(suffixKey); return true; } if (kind === MSG_PROVIDER_REPLY) { const decoded = this.codec.decodeProviderReply(data); if (!decoded) return false; const reqId = decoded.reqId; const pendingByReq = this.pendingProviderQueryBySuffixKey.get(suffixKey); const pending = pendingByReq?.get(reqId); if (!pending) return true; pendingByReq!.delete(reqId); const now = Date.now(); this.touchProviderNamespace(suffixKey, now); const candidates = this.toProviderCandidates(decoded.entries); this.rememberProviderCandidates( this.getProviderNamespaceIdFromKey(channelKey, suffixKey), candidates, now + 60_000, ); this.pruneProviderNamespaceIfEmpty(suffixKey); if (candidates.length > 0) { this.emitProviderUpdate( this.getProviderNamespaceIdFromKey(channelKey, suffixKey), candidates, ); } pending.resolve(candidates); return true; } if (kind === MSG_PROVIDER_NOTIFY) { const decoded = this.codec.decodeProviderNotify(data); if (!decoded) return false; const now = Date.now(); const providers = this.toProviderCandidates(decoded.entries); this.rememberProviderCandidates( this.getProviderNamespaceIdFromKey(channelKey, suffixKey), providers, now + 60_000, ); if (providers.length > 0) { this.emitProviderUpdate( this.getProviderNamespaceIdFromKey(channelKey, suffixKey), providers, ); } return true; } if (kind === MSG_TRACKER_REPLY) { if (!ch) return true; const decoded = this.codec.decodeTrackerReply(data); if (!decoded) return false; const reqId = decoded.reqId; const pending = ch.pendingTrackerQuery.get(reqId); if (pending) { ch.pendingTrackerQuery.delete(reqId); } const candidates: TrackerCandidate[] = []; for (const entry of decoded.entries) { const addrs: Multiaddr[] = []; for (const bytes of entry.addrs) { try { addrs.push(multiaddr(bytes)); } catch { // ignore invalid multiaddrs } } candidates.push({ hash: entry.hash, level: entry.level, freeSlots: entry.freeSlots, bidPerByte: entry.bidPerByte, addrs, }); this.cacheKnownCandidateAddrs(ch, entry.hash, addrs); } if (pending) { pending.resolve(candidates); } else if (candidates.length > 0) { // Late reply: the query already timed out, but the candidate // list is still fresh directory state. Under load the 1s query // timeout is routinely missed while trackers hold plenty of // capacity - dropping these replies starves joiners of // candidates they already paid a round trip for. const merged = new Map(); for (const c of ch.cachedTrackerCandidates) merged.set(c.hash, c); for (const c of candidates) merged.set(c.hash, c); ch.cachedTrackerCandidates = [...merged.values()].slice(-256); } return true; } if (kind === MSG_PARENT_PROBE_REQ) { if (!ch || ch.closed) return true; const decoded = this.codec.decodeParentProbeReq(data); if (!decoded) return false; this.touchPeerHint(ch, fromHash); void this._sendControl( fromHash, this.encodeParentProbeReplyForChannel( ch, decoded.reqId, fromHash, decoded.minFreeSlots, decoded.reserveRootCapacity, ), ).catch(() => {}); return true; } if (kind === MSG_PARENT_PROBE_REPLY) { if (!ch || ch.closed) return true; const decoded = this.codec.decodeParentProbeReply(data, fromHash); if (!decoded) return false; const { reqId, ...reply } = decoded; const pending = ch.pendingParentProbe.get(reqId); if (!pending) return true; ch.pendingParentProbe.delete(reqId); pending.resolve(reply); this.touchPeerHint(ch, fromHash); return true; } if (!ch) return true; if (ch.closed) return true; if (kind === MSG_LEAVE) { // Best-effort: allow a child to explicitly detach to immediately free capacity. if (!ch.children.has(fromHash)) return true; ch.children.delete(fromHash); ch.dataWriteFailStreakByChild.delete(fromHash); ch.channelPeers.delete(fromHash); ch.lazyPeers.delete(fromHash); ch.haveByPeer.delete(fromHash); return true; } if (kind === MSG_ROUTE_QUERY) { const decoded = this.codec.decodeRouteQuery(data); if (!decoded) return false; const reqId = decoded.reqId; if (decoded.targetHash == null) { void this._sendControl( fromHash, this.codec.encodeRouteReply(ch.id.key, reqId), ).catch(() => {}); return true; } const targetHash = decoded.targetHash; const localRoute = targetHash === this.publicKeyHash && ch.routeFromRoot ? ch.routeFromRoot : this.getCachedRoute(ch, targetHash); if (this.isRouteValidForChannel(ch, localRoute)) { void this._sendControl( fromHash, this.codec.encodeRouteReply(ch.id.key, reqId, localRoute), ).catch(() => {}); return true; } if (ch.isRoot) { const rootRoute = ch.children.has(targetHash) ? [ch.id.root, targetHash] : undefined; if (rootRoute) this.cacheRoute(ch, rootRoute); if (rootRoute) { void this._sendControl( fromHash, this.codec.encodeRouteReply(ch.id.key, reqId, rootRoute), ).catch(() => {}); return true; } } const fromParent = !ch.isRoot && ch.parent != null && fromHash === ch.parent; // Child->parent lookups still go upstream first, which keeps cross-branch // route discovery efficient when caches are warm. if (!ch.isRoot && !fromParent && ch.parent) { this.proxyRouteQuery(ch, fromHash, reqId, targetHash, [ch.parent]); return true; } // Cache miss on a parent->child (or root->child) query: // recursively search subtree branches and reply with the first valid route. this.proxyRouteQuery(ch, fromHash, reqId, targetHash, [ ...ch.children.keys(), ]); return true; } if (kind === MSG_ROUTE_REPLY) { const decoded = this.codec.decodeRouteReply(data); if (!decoded) return false; const { reqId, route } = decoded; const parsedRoute = this.isRouteValidForChannel(ch, route) ? route : undefined; if (parsedRoute) { this.cacheRoute(ch, parsedRoute); } const pendingLocal = ch.pendingRouteQuery.get(reqId); if (pendingLocal) { ch.pendingRouteQuery.delete(reqId); pendingLocal.resolve(parsedRoute); return true; } const proxy = ch.pendingRouteProxy.get(reqId); if (!proxy) return true; if (!proxy.expectedReplies.has(fromHash)) return true; proxy.expectedReplies.delete(fromHash); if (parsedRoute) { this.completeRouteProxy(ch, reqId, parsedRoute); return true; } if (proxy.expectedReplies.size === 0) { this.completeRouteProxy(ch, reqId); } return true; } if (kind === MSG_IHAVE) { const decoded = this.codec.decodeIHave(data); if (!decoded) return false; const { haveFrom, haveToExclusive } = decoded; const now = Date.now(); if (ch.parent && fromHash === ch.parent) { ch.lastParentDataAt = now; ch.receivedAnyParentData = true; ch.parentRepairUnansweredStreak = 0; } const prev = ch.haveByPeer.get(fromHash); if (prev) { prev.haveFrom = haveFrom; prev.haveToExclusive = haveToExclusive; prev.updatedAt = now; } else { ch.haveByPeer.set(fromHash, { haveFrom, haveToExclusive, updatedAt: now, requests: 0, successes: 0, }); } this.touchPeerHint(ch, fromHash, now); if (ch.parent && fromHash === ch.parent && ch.maxDataAgeMs === 0) { // A watermark from our parent doubles as gap detection: if data // writes to us were lost (e.g. burst into a not-yet-writable // stream) we may have received nothing at all and would // otherwise never know data exists. Mark the advertised range // missing so the repair loop pulls it. Deadline-oriented live // channels (maxDataAgeMs > 0) skip this: catching up on stale // history would trade deadline latency for completeness. this.noteEnd(ch, fromHash, haveToExclusive); } // Opportunistic reciprocity: if we still have room in our lazy mesh, add // peers that are actively exchanging IHAVE summaries with us. if ( ch.neighborRepair && ch.neighborMeshPeers > 0 && ch.lazyPeers.size < ch.neighborMeshPeers && fromHash !== ch.parent && !ch.children.has(fromHash) && this.peers.get(fromHash) ) { ch.lazyPeers.add(fromHash); } return true; } if (kind === MSG_JOIN_REQ) { const decoded = this.codec.decodeJoinReq(data); if (!decoded) return false; const { reqId, bidPerByte, parentUpgradeReservationToken } = decoded; this.pruneDisconnectedChildren(ch); const hasParentUpgradeReservation = ch.isRoot && parentUpgradeReservationToken > 0; const consumedParentUpgradeReservation = hasParentUpgradeReservation ? this.consumeParentUpgradeReservation( ch, fromHash, parentUpgradeReservationToken, ) : false; if (hasParentUpgradeReservation && !consumedParentUpgradeReservation) { ch.metrics.parentUpgradeRootReservationRejected += 1; void this.sendJoinReject( ch, fromHash, reqId, JOIN_REJECT_NO_CAPACITY, ).catch(() => {}); void this.announceToTrackers(ch, this.closeController.signal).catch( () => {}, ); return true; } // Only accept if we're already attached. if (!ch.isRoot && !ch.parent) { void this.sendJoinReject( ch, fromHash, reqId, JOIN_REJECT_NOT_ATTACHED, ).catch(() => {}); return true; } // Only accept children if we can prove we're on the rooted tree. // Otherwise, disconnected components can "stabilize" by joining via unrooted parents. if (!ch.isRoot) { const route = ch.routeFromRoot; const rooted = Array.isArray(route) && route.length >= 2 && route[0] === ch.id.root && route[route.length - 1] === this.publicKeyHash; if (!rooted) { void this.sendJoinReject( ch, fromHash, reqId, JOIN_REJECT_NOT_ATTACHED, ).catch(() => {}); return true; } } if (ch.effectiveMaxChildren <= 0) { void this.sendJoinReject( ch, fromHash, reqId, JOIN_REJECT_NO_CAPACITY, ).catch(() => {}); return true; } if ( !ch.children.has(fromHash) && (consumedParentUpgradeReservation ? ch.children.size >= ch.effectiveMaxChildren : ch.children.size + this.parentUpgradeReservationCount(ch, fromHash) >= ch.effectiveMaxChildren) ) { const blockedByReservedRootCapacity = ch.isRoot && !consumedParentUpgradeReservation && ch.children.size < ch.effectiveMaxChildren; if (!ch.allowKick || blockedByReservedRootCapacity) { void this.sendJoinReject( ch, fromHash, reqId, JOIN_REJECT_NO_CAPACITY, ).catch(() => {}); void this.announceToTrackers(ch, this.closeController.signal).catch( () => {}, ); return true; } let worstChild: string | undefined; let worstBid = Number.POSITIVE_INFINITY; for (const [childHash, info] of ch.children) { if (info.bidPerByte < worstBid) { worstBid = info.bidPerByte; worstChild = childHash; } } if (worstChild == null || bidPerByte <= worstBid) { void this.sendJoinReject( ch, fromHash, reqId, JOIN_REJECT_LOW_BID, ).catch(() => {}); void this.announceToTrackers(ch, this.closeController.signal).catch( () => {}, ); return true; } void this.kickChildHashes(ch, [worstChild]).catch(() => {}); } ch.children.set(fromHash, { bidPerByte }); if (ch.isRoot && consumedParentUpgradeReservation) { ch.parentUpgradeTrackerNoCapacityUntil = Math.max( ch.parentUpgradeTrackerNoCapacityUntil, Date.now() + 2_000, ); } this.touchPeerHint(ch, fromHash); const joinAccept = this.codec.encodeJoinAccept( ch.id.key, reqId, ch.level, ch.routeFromRoot, this.getHaveRange(ch), ); void (async () => { await this._sendControl(fromHash, joinAccept); if (ch.endSeqExclusive > 0) { await this._sendControl( fromHash, this.codec.encodeEnd(ch.id.key, ch.endSeqExclusive), ); } })().catch(() => {}); void this.announceToTrackers(ch, this.closeController.signal).catch( () => {}, ); return true; } if (kind === MSG_JOIN_ACCEPT || kind === MSG_JOIN_REJECT) { const reqId = this.codec.decodeJoinResponseReqId(data); if (reqId == null) return false; // Any explicit join response proves this path is alive end to end. this.noteJoinResponse(fromHash); const pending = ch.pendingJoin.get(reqId); if (!pending) return true; ch.pendingJoin.delete(reqId); if (kind === MSG_JOIN_ACCEPT) { const decoded = this.codec.decodeJoinAccept(data); if (!decoded) return false; const { parentLevel, parentRouteFromRoot } = decoded; const hasValidParentRoute = parentRouteFromRoot.length > 0 && parentRouteFromRoot[0] === ch.id.root && parentRouteFromRoot[parentRouteFromRoot.length - 1] === fromHash; // Defensive: a JOIN_ACCEPT without a rooted route token (unless the parent is the // actual root) can create stable disconnected components. Treat it as a reject. if (fromHash !== ch.id.root && !hasValidParentRoute) { pending.resolve({ ok: false, rejectReason: JOIN_REJECT_NOT_ATTACHED, }); return true; } const acceptedParentRoute = hasValidParentRoute || fromHash === ch.id.root ? fromHash === ch.id.root && !hasValidParentRoute ? [ch.id.root] : parentRouteFromRoot : undefined; if (pending.shadowAttach) { this.touchPeerHint(ch, fromHash); pending.resolve({ ok: true, parentLevel, parentRouteFromRoot: acceptedParentRoute, }); return true; } const attachedAt = Date.now(); ch.parent = fromHash; ch.level = parentLevel + 1; ch.joinedAtLeastOnce = true; ch.lastParentDataAt = attachedAt; ch.lastParentUpgradeActivityAt = attachedAt; ch.parentRepairUnansweredStreak = 0; this.resetParentDataLatency(ch); // Treat JOIN_ACCEPT as parent liveness for stale re-parenting: // if callers enable `staleAfterMs`, we should be able to detach even // before the first data message arrives (for example, during churn // or when a component is partitioned from the root). ch.receivedAnyParentData = true; // Build/refresh a route token that enables economical unicast. if (hasValidParentRoute) { ch.routeFromRoot = [...parentRouteFromRoot, this.publicKeyHash]; } else if (fromHash === ch.id.root) { // Minimal fallback: parent is the root. ch.routeFromRoot = [ch.id.root, this.publicKeyHash]; } if ( ch.repairEnabled && ch.maxDataAgeMs === 0 && decoded.haveRange != null ) { // Optional appendix: the parent's current have-range. Lets a // freshly attached child learn immediately which sequences // already exist, so lost deliveries surface as missing seqs // (and repair) instead of silent absence. const { haveFrom, haveToExclusive } = decoded.haveRange; if (haveToExclusive > haveFrom) { // Mark only; the repair loop's own cadence picks the gaps // up, giving in-flight live deliveries time to land first // (an immediate pull here races them into duplicates). this.noteEnd(ch, fromHash, haveToExclusive); } } this.touchPeerHint(ch, fromHash); pending.resolve({ ok: true, parentLevel, parentRouteFromRoot: acceptedParentRoute, }); this.dispatchEvent( new CustomEvent("fanout:joined", { detail: { topic: ch.id.topic, root: ch.id.root, parent: fromHash, }, }), ); void this.announceToTrackers(ch, this.closeController.signal).catch( () => {}, ); } else { const decoded = this.codec.decodeJoinReject(data); if (!decoded) return false; const reason = decoded.reason; const redirects: Array<{ hash: string; addrs: Multiaddr[] }> = []; for (const redirect of decoded.redirects) { const addrs: Multiaddr[] = []; for (const bytes of redirect.addrs) { try { addrs.push(multiaddr(bytes)); } catch { // ignore invalid multiaddrs } } if (redirect.hash && addrs.length > 0) { redirects.push({ hash: redirect.hash, addrs }); this.cacheKnownCandidateAddrs(ch, redirect.hash, addrs); } } pending.resolve({ ok: false, rejectReason: reason, redirects }); } return true; } if (kind === MSG_PUBLISH_PROXY) { // Requires an open channel state if (!ch) return true; // Only accept/forward within established tree edges to keep the data-plane economical. const isFromParent = fromHash === ch.parent; const isFromChild = ch.children.has(fromHash); if (!isFromParent && !isFromChild) return true; const payload = data.subarray(33); if (isFromChild) { const ok = this.takeIngressBudget( ch, "proxy-publish", fromHash, payload.byteLength, ); if (!ok) { ch.metrics.proxyPublishDrops += 1; return true; } } this.touchPeerHint(ch, fromHash); if (ch.isRoot) { // Only accept upstream proxy publishes from established children. if (!isFromChild) return true; const seq = ch.seq++; const message = await this._sendData( ch, [...ch.children.keys()], seq, payload, ); this.dispatchEvent( new CustomEvent("fanout:data", { detail: { topic: ch.id.topic, root: ch.id.root, seq, payload, from: this.publicKeyHash, origin: this.publicKeyHash, timestamp: message.header.timestamp, message, }, }), ); return true; } // Upstream forwarding (child -> ... -> root). if (!isFromChild) return true; if (!ch.parent) return true; const up = this.peers.get(ch.parent); if (!up) return true; void up .waitForWrite( message.bytes(), Number(message.header.priority ?? CONTROL_PRIORITY), this.closeController.signal, ) .catch(() => {}); return true; } if (kind === MSG_UNICAST_ACK) { // Requires an open channel state if (!ch) return true; // Only accept/forward within established tree edges. const isFromParent = fromHash === ch.parent; const isFromChild = ch.children.has(fromHash); if (!isFromParent && !isFromChild) return true; const decoded = this.codec.decodeUnicastAck(data); if (!decoded) return false; const ackToken = decoded.ackToken; const route = decoded.route; const target = route.length > 0 ? route[route.length - 1]! : ""; const origin = message.header.signatures?.publicKeys?.[0]?.hashcode?.() ?? fromHash; this.touchPeerHint(ch, fromHash); const settleLocal = () => { const pending = ch.pendingUnicastAck.get(ackToken); if (!pending) return; if (pending.expectedOrigin !== origin) return; pending.resolve(); }; // Root routes downward using the provided token. if (ch.isRoot) { if (route.length === 0 || route[0] !== ch.id.root) return true; if (target === this.publicKeyHash) { settleLocal(); return true; } const nextHop = route[1]; if (!nextHop || !ch.children.has(nextHop)) return true; const stream = this.peers.get(nextHop); if (!stream) return true; void stream .waitForWrite( message.bytes(), Number(message.header.priority ?? CONTROL_PRIORITY), this.closeController.signal, ) .catch(() => {}); return true; } // Downstream routing: parent -> child -> ... -> sender. if (isFromParent) { const selfHash = this.publicKeyHash; const myIndex = route.indexOf(selfHash); if (myIndex < 0) return true; if (myIndex === route.length - 1) { settleLocal(); return true; } const nextHop = route[myIndex + 1]; if (!nextHop || !ch.children.has(nextHop)) return true; const stream = this.peers.get(nextHop); if (!stream) return true; void stream .waitForWrite( message.bytes(), Number(message.header.priority ?? CONTROL_PRIORITY), this.closeController.signal, ) .catch(() => {}); return true; } // Upstream forwarding (child -> ... -> root). Route token is only used by the root. if (!ch.parent) return true; const up = this.peers.get(ch.parent); if (!up) return true; void up .waitForWrite( message.bytes(), Number(message.header.priority ?? CONTROL_PRIORITY), this.closeController.signal, ) .catch(() => {}); return true; } if (kind === MSG_UNICAST) { // Requires an open channel state if (!ch) return true; // Only accept/forward within established tree edges to keep the data-plane economical. const isFromParent = fromHash === ch.parent; const isFromChild = ch.children.has(fromHash); if (!isFromParent && !isFromChild) return true; if (isFromChild) { const ok = this.takeIngressBudget( ch, "unicast", fromHash, data.byteLength, ); if (!ok) { ch.metrics.unicastDrops += 1; return true; } } const decoded = this.codec.decodeUnicast(data); if (!decoded) return false; const ackToken = decoded.ackToken; const route = decoded.route; const replyRoute = decoded.replyRoute; const payload = data.subarray(decoded.payloadOffset); const target = route.length > 0 ? route[route.length - 1]! : ""; const origin = message.header.signatures?.publicKeys?.[0]?.hashcode?.() ?? fromHash; this.touchPeerHint(ch, fromHash); // Root routes downward using the provided token. if (ch.isRoot) { if (route.length === 0 || route[0] !== ch.id.root) return true; if (target === this.publicKeyHash) { this.dispatchEvent( new CustomEvent("fanout:unicast", { detail: { topic: ch.id.topic, root: ch.id.root, route, payload, from: fromHash, origin, to: target, timestamp: message.header.timestamp, message, }, }), ); if (ackToken != null) { const canAck = replyRoute && replyRoute.length > 0 && replyRoute[0] === ch.id.root && replyRoute[replyRoute.length - 1] === origin; if (canAck) { const nextHop = replyRoute![1]; if (nextHop && ch.children.has(nextHop)) { void this._sendControl( nextHop, this.codec.encodeUnicastAck(ch.id.key, ackToken, replyRoute!), ).catch(() => {}); } } } return true; } const nextHop = route[1]; if (!nextHop || !ch.children.has(nextHop)) return true; const stream = this.peers.get(nextHop); if (!stream) return true; // A validated source route is passing through: learn it (and the // reply route) so subsequent unicasts skip route discovery. this.cacheRoute(ch, route); if (replyRoute) this.cacheRoute(ch, replyRoute); void stream .waitForWrite( message.bytes(), Number(message.header.priority ?? CONTROL_PRIORITY), this.closeController.signal, ) .catch(() => {}); return true; } // Downstream routing: parent -> child -> ... -> target. if (isFromParent) { const selfHash = this.publicKeyHash; const myIndex = route.indexOf(selfHash); if (myIndex < 0) return true; if (myIndex === route.length - 1) { this.dispatchEvent( new CustomEvent("fanout:unicast", { detail: { topic: ch.id.topic, root: ch.id.root, route, payload, from: fromHash, origin, to: target, timestamp: message.header.timestamp, message, }, }), ); if (ackToken != null && ch.parent) { const canAck = replyRoute && replyRoute.length > 0 && replyRoute[0] === ch.id.root && replyRoute[replyRoute.length - 1] === origin; if (canAck) { void this._sendControl( ch.parent, this.codec.encodeUnicastAck(ch.id.key, ackToken, replyRoute!), ).catch(() => {}); } } return true; } const nextHop = route[myIndex + 1]; if (!nextHop || !ch.children.has(nextHop)) return true; const stream = this.peers.get(nextHop); if (!stream) return true; this.cacheRoute(ch, route); if (replyRoute) this.cacheRoute(ch, replyRoute); void stream .waitForWrite( message.bytes(), Number(message.header.priority ?? CONTROL_PRIORITY), this.closeController.signal, ) .catch(() => {}); return true; } // Upstream forwarding (child -> ... -> root). Route token is only used by the root. if (!ch.parent) return true; const up = this.peers.get(ch.parent); if (!up) return true; void up .waitForWrite( message.bytes(), Number(message.header.priority ?? CONTROL_PRIORITY), this.closeController.signal, ) .catch(() => {}); return true; } if (kind === MSG_KICK) { ch.metrics.reparentKicked += 1; ch.parent = undefined; ch.level = Number.POSITIVE_INFINITY; ch.routeFromRoot = undefined; ch.routeByPeer.clear(); ch.lastParentDataAt = 0; ch.lastParentUpgradeActivityAt = 0; ch.receivedAnyParentData = false; ch.pendingJoin.clear(); ch.pendingParentProbe.clear(); ch.parentShadow = undefined; void Promise.all(this.clearParentUpgradeGrace(ch, true, true)).catch( () => {}, ); if (this.parentUpgradeShadowInFlightSuffixKey === ch.id.suffixKey) { this.parentUpgradeShadowInFlightSuffixKey = undefined; } ch.pendingRouteQuery.clear(); this.abortPendingUnicastAcks( ch, new AbortError("fanout channel kicked"), ); this.clearRouteProxies(ch); void this.kickChildren(ch).catch(() => {}); this.dispatchEvent( new CustomEvent("fanout:kicked", { detail: { topic: ch.id.topic, root: ch.id.root, from: fromHash }, }), ); return true; } if (kind === MSG_END) { const lastSeqExclusive = this.codec.decodeEnd(data); if (lastSeqExclusive == null) return false; if (ch.parent && fromHash === ch.parent) { ch.lastParentDataAt = Date.now(); ch.receivedAnyParentData = true; ch.parentRepairUnansweredStreak = 0; } ch.endSeqExclusive = Math.max(ch.endSeqExclusive, lastSeqExclusive); this.touchPeerHint(ch, fromHash); this.noteEnd(ch, fromHash, lastSeqExclusive); if (ch.children.size > 0) { void this._sendControlMany( [...ch.children.keys()], this.codec.encodeEnd(ch.id.key, lastSeqExclusive), ); } void this.tickRepair(ch).catch(() => {}); return true; } if (kind === MSG_REPAIR_REQ) { const seqs = this.codec.decodeRepairSeqs(data); if (!seqs) return false; if (!ch.children.has(fromHash)) return true; for (const seq of seqs) { const cached = this.getCached(ch, seq); if (!cached) { ch.metrics.cacheMissesServed += 1; continue; } ch.metrics.cacheHitsServed += 1; void this._sendData(ch, [fromHash], seq, cached); } return true; } if (kind === MSG_FETCH_REQ) { const seqs = this.codec.decodeRepairSeqs(data); if (!seqs) return false; this.touchPeerHint(ch, fromHash); for (const seq of seqs) { const cached = this.getCached(ch, seq); if (!cached) { ch.metrics.cacheMissesServed += 1; continue; } ch.metrics.cacheHitsServed += 1; void this._sendData(ch, [fromHash], seq, cached); } return true; } return true; } // Data-plane (topic identified via id suffix) if (kind === MSG_DATA) { const id = message.id as Uint8Array; if (!(id instanceof Uint8Array) || !isDataId(id)) return false; const suffixKey = this.getSuffixKeyFromId(id); const ch = this.channelsBySuffixKey.get(suffixKey); if (!ch) return false; const seq = readU32BE(id, 4); if (ch.parent && fromHash === ch.parent) { const receivedAt = Date.now(); ch.lastParentDataAt = receivedAt; ch.lastParentUpgradeActivityAt = receivedAt; ch.receivedAnyParentData = true; ch.parentRepairUnansweredStreak = 0; } const hadCachedBeforeData = this.getCached(ch, seq) != null; if (ch.parent && fromHash === ch.parent && !hadCachedBeforeData) { this.noteParentDataLatency(ch, message.header.timestamp); } ch.maxSeqSeen = Math.max(ch.maxSeqSeen, seq); if ( ch.parentUpgradeRetryAfterSeq >= 0 && seq > ch.parentUpgradeRetryAfterSeq ) { ch.parentUpgradeRetryAfterSeq = -1; } const parentShadow = ch.parentShadow; if (parentShadow != null) { const now = Date.now(); const pruneShadowSeqMap = (map: Map) => { while (map.size > 512) { const oldest = map.keys().next().value as number | undefined; if (oldest == null) break; map.delete(oldest); } }; const pruneShadowSeqSet = (set: Set) => { while (set.size > 512) { const oldest = set.keys().next().value as number | undefined; if (oldest == null) break; set.delete(oldest); } }; const rememberShadowArrival = (map: Map) => { if (map.has(seq)) return map.get(seq); map.set(seq, now); pruneShadowSeqMap(map); return now; }; const recordCandidateLead = ( candidateAt: number | undefined, parentAt: number | undefined, ) => { if ( candidateAt == null || parentAt == null || parentShadow.liveComparedSeqs.has(seq) ) { return; } parentShadow.liveComparedSeqs.add(seq); pruneShadowSeqSet(parentShadow.liveComparedSeqs); if (candidateAt < parentAt) { parentShadow.liveCandidateLeadSamples += 1; parentShadow.liveCandidateLeadMsTotal += parentAt - candidateAt; } }; if (parentShadow.hash === fromHash) { const candidateAt = rememberShadowArrival( parentShadow.liveCandidateFirstAtBySeq, ); recordCandidateLead( candidateAt, parentShadow.liveParentFirstAtBySeq.get(seq), ); if (seq > parentShadow.liveMaxSeqSeen) { parentShadow.liveMaxSeqSeen = seq; parentShadow.liveDataMessages += 1; } if (!hadCachedBeforeData) { parentShadow.liveFirstDataMessages += 1; } parentShadow.liveLastDataAt = now; } else if (ch.parent != null && fromHash === ch.parent) { const parentAt = rememberShadowArrival( parentShadow.liveParentFirstAtBySeq, ); recordCandidateLead( parentShadow.liveCandidateFirstAtBySeq.get(seq), parentAt, ); if (!hadCachedBeforeData) { parentShadow.liveParentFirstDataMessages += 1; } } } const parentUpgradeGrace = ch.parentUpgradeGrace; if (parentUpgradeGrace != null && !hadCachedBeforeData) { if (fromHash === parentUpgradeGrace.candidateParent) { parentUpgradeGrace.candidateFirstMessages += 1; } else if (fromHash === parentUpgradeGrace.previousParent) { parentUpgradeGrace.previousFirstMessages += 1; } if ( parentUpgradeGrace.previousFirstMessages > parentUpgradeGrace.maxPreviousFirstMessages ) { this.rollbackParentUpgradeGrace(ch); } else if ( parentUpgradeGrace.candidateFirstMessages >= parentUpgradeGrace.minCandidateFirstMessages && parentUpgradeGrace.candidateFirstMessages >= parentUpgradeGrace.previousFirstMessages + parentUpgradeGrace.minCandidateAdvantageMessages ) { this.commitParentUpgradeGrace(ch); } } this.noteReceivedSeq(ch, fromHash, seq); const payload = (data as Uint8Array).subarray(1); ch.metrics.dataReceives += 1; ch.metrics.dataPayloadBytesReceived += payload.byteLength; this.markCached(ch, seq, payload); this.dispatchEvent( new CustomEvent("fanout:data", { detail: { topic: ch.id.topic, root: ch.id.root, seq, payload, from: fromHash, origin: message.header.signatures?.publicKeys?.[0]?.hashcode?.() ?? fromHash, timestamp: message.header.timestamp, message, }, }), ); if (ch.children.size > 0) { void this._forwardDataMessage( ch, [...ch.children.keys()], payload, message, ); } void this.tickRepair(ch).catch(() => {}); return true; } return false; } }