diff --git a/backend/src/api/transfer.ts b/backend/src/api/transfer.ts index 718c09d1..7e7fed42 100644 --- a/backend/src/api/transfer.ts +++ b/backend/src/api/transfer.ts @@ -115,6 +115,95 @@ export function initTransferHandlers(networks: Networks, dataServer: DataServer, const activeDownloaders = new Map(); setActiveDownloadersRef(activeDownloaders); + // LISHs whose download was suspended because their last joined lishnet was left, + // mapped to the lishnets they were bound to. Their DB enabled flag stays on (see + // onNetworkLeft), so onNetworkJoined resumes them — but only when a BOUND lishnet + // re-joins, never rebinding to an unrelated one. An empty bound set means "no known + // binding" (startup with no joined lishnet, where the fresh downloader would bind + // to whatever is enabled at resume time) → resume on any join. Cleared when the + // user explicitly enables/disables the download so a rejoin never overrides a + // deliberate user action. + const networkSuspended = new Map>(); + + // When a lishnet is left, stop any download bound EXCLUSIVELY to it: a + // downloader keeps running as long as at least one of its networks is still + // joined (multi-network downloads can still source chunks elsewhere). Only + // when none of its networks remain joined is there no peer source left, so we + // disable it (leaving DB/enabled flags untouched — a re-join can resume it). + networks.onNetworkLeft = (networkID: string) => { + for (const [lishID, dl] of activeDownloaders) { + const ids = dl.getNetworkIDs?.() ?? []; + if (!ids.includes(networkID)) continue; + if (ids.some(id => networks.isJoined(id))) { + // Another joined lishnet can still source this download — keep it + // running but stop using the network we just left, otherwise the + // downloader keeps broadcasting WANTs and probing peers on a topic + // we are no longer part of. + dl.removeNetwork?.(networkID); + continue; + } + // Drop the runtime enabled flag (no DB persist) so `lishs.list` reports the + // download as stopped and restartDownloadIfEnabled cannot silently revive it + // while no usable lishnet is joined. The DB flag stays untouched, so an app + // restart with the lishnet re-joined resumes the download. + const wasEnabled = downloadEnabledLishs.has(lishID); + downloadEnabledLishs.delete(lishID); + // Cancel any pending error-recovery timer for this LISH — otherwise + // ErrorRecovery, holding the captured downloadWasEnabled=true, could + // re-enable the download once the IO condition clears even though the + // user just stopped it by leaving the network. + recovery.stop(lishID); + if (wasEnabled) { + // Persisted download — retain the disabled downloader and remember it as + // suspended-by-leave so onNetworkJoined can resume it after rejoin. + console.log(`[Transfer] ${lishID.slice(0, 8)}: last joined lishnet left, disabling download`); + dl.disable(); + // Bind resume to the download's ORIGINAL lishnets (not the current set, + // which removeNetwork may have shrunk) so only a re-join of a lishnet this + // download actually belongs to resumes it. + networkSuspended.set(lishID, new Set(dl.getOriginalNetworkIDs?.() ?? dl.getNetworkIDs?.() ?? [])); + } else { + // Transient download (from the `download` handler, never enabled/persisted) + // has no resume claim — destroy it and drop it from the map instead of + // leaking a disabled downloader with a dangling download() promise and + // registered network handlers (a fresh start of the same LISH would + // otherwise overwrite the map entry without disposing this one). + console.log(`[Transfer] ${lishID.slice(0, 8)}: last joined lishnet left, dropping transient download`); + dl.destroy().catch(err => console.error(`[Transfer] ${lishID.slice(0, 8)}: destroy on leave failed:`, err?.message ?? err)); + activeDownloaders.delete(lishID); + } + // dl.disable() alone emits nothing over WS — tell the FE the download + // stopped. + broadcast?.('transfer.download:disabled', { lishID }); + } + }; + + // When a previously-left lishnet is re-joined in-process, resume downloads that + // were suspended because it was their last joined network. Their DB enabled flag + // was intentionally left on (see onNetworkLeft), so re-enabling here restores the + // pre-leave state without waiting for an app restart. Only downloads still bound + // to the re-joined network and still suspended are resumed. + networks.onNetworkJoined = (networkID: string) => { + // Re-attach the rejoined network to still-running multi-network downloaders + // that dropped it when it was left (no-op if never bound to it or already active). + for (const dl of activeDownloaders.values()) dl.addNetwork?.(networkID); + // Resume a suspended download only when a lishnet it is BOUND to re-joins (an + // empty bound set = no known binding → resume on any join). Drop the suspension + // ONLY once the resume actually succeeds — a transient failure (busy verifying, + // still no joined lishnet) must be retried on the next join. + for (const [lishID, bound] of [...networkSuspended]) { + if (bound.size > 0 && !bound.has(networkID)) continue; + enableDownload({ lishID }) + .then(r => { + if (r.success) { + networkSuspended.delete(lishID); + console.log(`[Transfer] ${lishID.slice(0, 8)}: lishnet re-joined, download resumed`); + } + }) + .catch(err => console.error(`[Transfer] resume-on-rejoin ${lishID.slice(0, 8)} failed:`, err?.message ?? err)); + } + }; + // Error recovery: auto-retry when IO conditions clear const recovery = new ErrorRecovery({ attemptRecover: async (lishID, downloadWasEnabled, uploadWasEnabled): Promise => { @@ -181,6 +270,7 @@ export function initTransferHandlers(networks: Networks, dataServer: DataServer, function disableDownload(p: { lishID: string }): { success: boolean } { assert(p, ['lishID']); + networkSuspended.delete(p.lishID); recovery.stop(p.lishID); downloadEnabledLishs.delete(p.lishID); persistDownloadEnabled?.(p.lishID, false); @@ -202,6 +292,16 @@ export function initTransferHandlers(networks: Networks, dataServer: DataServer, persistDownloadEnabled?.(p.lishID, true); const dl = activeDownloaders.get(p.lishID); if (dl) { + // A download suspended by leaving its last lishnet is retained (disabled) for + // resume on rejoin. Enabling it before a bound lishnet is re-joined would make + // it broadcast WANTs / probe on a topic we already left — keep it suspended. + const boundIDs = dl.getOriginalNetworkIDs?.() ?? dl.getNetworkIDs?.() ?? []; + if (networkSuspended.has(p.lishID) && boundIDs.length > 0 && !boundIDs.some((id: string) => networks.isJoined(id))) { + // Drop the runtime enabled flag we just set; the DB flag (persisted above) + // stays true so a later rejoin of a bound lishnet resumes the download. + downloadEnabledLishs.delete(p.lishID); + return { success: false }; + } // If downloader is in error state, destroy it and create a fresh one if (dl.getError()) { console.debug(`[Transfer] ${p.lishID.slice(0, 8)}: destroying error-state downloader, will create fresh`); @@ -268,8 +368,16 @@ export function initTransferHandlers(networks: Networks, dataServer: DataServer, const network = networks.getRunningNetwork(); const joinedNetworks = networks.getEnabled().map(n => n.networkID); if (joinedNetworks.length === 0) { + // No lishnet is joined to source this download. Keep the DB enabled flag + // ON — clearing it would permanently forget the user's intent so a later + // rejoin could never resume. Drop only the in-memory active flag and mark + // it suspended so onNetworkJoined resumes it once a lishnet is (re-)joined. downloadEnabledLishs.delete(p.lishID); - persistDownloadEnabled?.(p.lishID, false); + // No active downloader and no joined lishnet → no known network binding + // (the fresh downloader would bind to getEnabled(), empty here). Store the + // empty bound set so onNetworkJoined resumes on any join, since the DB has + // no per-download network to restrict to. + networkSuspended.set(p.lishID, new Set(joinedNetworks)); return { success: false }; } const downloadDir = lish.directory ?? join(dataDir, 'downloads', Date.now().toString()); @@ -361,7 +469,11 @@ export function initTransferHandlers(networks: Networks, dataServer: DataServer, const enabled = getEnabledUploads(); // Active downloads — report the allocation phase distinctly so the UI can show // "allocating" after a reconnect instead of falling back to idle (no peers yet). + // A downloader disabled by leaving its last lishnet stays in the map (retained + // for resume on rejoin) but is stopped — skip it so the LISH is not reported as + // still downloading after transfer.download:disabled. for (const [lishID, dl] of activeDownloaders) { + if (dl.isDisabled?.()) continue; const type = dl.isAllocating?.() ? 'allocating' : 'downloading'; transfers.push({ lishID, type, peers: dl.getPeerCount?.() ?? 0, bytesPerSecond: 0 }); } @@ -491,6 +603,7 @@ export function initTransferHandlers(networks: Networks, dataServer: DataServer, } activeDownloaders.clear(); downloadEnabledLishs.clear(); + networkSuspended.clear(); clearAllUploads(); recovery.stopAll(); } diff --git a/backend/src/lishnet/lishnets.ts b/backend/src/lishnet/lishnets.ts index e57a0a77..2062f70b 100644 --- a/backend/src/lishnet/lishnets.ts +++ b/backend/src/lishnet/lishnets.ts @@ -21,6 +21,12 @@ export class Networks { private _onPeerCountChange: ((counts: { networkID: string; count: number }[]) => void) | null = null; // Callback for bootstrap status changes private _onBootstrapStatusChange: ((networkID: string, status: BootstrapStatus) => void) | null = null; + // Callback fired after a lishnet is left (topic unsubscribed). Lets higher + // layers (e.g. transfer) stop downloads bound exclusively to that lishnet. + private _onNetworkLeft: ((networkID: string) => void) | null = null; + // Callback fired after a lishnet is (re-)joined in-process. Lets higher layers + // resume downloads that were suspended when this lishnet was previously left. + private _onNetworkJoined: ((networkID: string) => void) | null = null; constructor(db: Database, dataDir: string, dataServer: DataServer, settings: Settings) { this.db = db; @@ -50,6 +56,27 @@ export class Networks { this._onBootstrapStatusChange = cb; } + /** + * Set a callback fired right after a lishnet is left (its topic has been + * unsubscribed and removed from {@link joinedNetworks}). The callback runs + * synchronously from {@link leaveNetwork}; consumers should not assume any + * particular peer/connection state beyond "this lishnet is no longer joined". + */ + set onNetworkLeft(cb: ((networkID: string) => void) | null) { + this._onNetworkLeft = cb; + } + + /** + * Set a callback fired right after a lishnet is (re-)joined via {@link joinNetwork} + * (its topic subscribed and added to {@link joinedNetworks}). Lets higher layers + * (e.g. transfer) resume downloads that were suspended when the lishnet was + * previously left. NOT fired for the initial startup join — startup has its own + * auto-resume path. + */ + set onNetworkJoined(cb: ((networkID: string) => void) | null) { + this._onNetworkJoined = cb; + } + init(): void { console.log('✓ Networks initialized'); } @@ -120,24 +147,113 @@ export class Networks { // ordering — the process-level error handlers in app.ts are the safety net. this.network.subscribeTopic(id); this.joinedNetworks.add(id); + // Rejoin is an explicit "I want peers back" — lift the redial suppression for + // THIS lishnet's left peers (bootstrap and content) so maintenance and discovery + // may reconnect them. Scoped per-network: still-left lishnets stay suppressed. + this.network.clearRedialSuppressionForNetwork(id); const net = this.get(id); if (net && net.bootstrapPeers.length > 0) await this.network.addBootstrapPeers(net.bootstrapPeers, id, 'configured'); console.log(`✓ Joined lishnet: ${net?.name ?? id}`); + + // Notify higher layers (e.g. transfer) so downloads suspended when this + // lishnet was last left can resume now that it is joined again. + this._onNetworkJoined?.(id); } /** * Leave a lishnet (unsubscribe from its topic). */ + /** Peer IDs (the /p2p/ component) of a list of bootstrap multiaddr strings. */ + private static bootstrapPeerIDsOf(bootstrapPeers: string[]): string[] { + const ids: string[] = []; + for (const addr of bootstrapPeers) { + // Relayed multiaddrs (.../p2p//p2p-circuit/p2p/) carry two + // /p2p components; the bootstrap peer identity is the FINAL one (the target), + // not the relay. Match all and take the last. + const matches = [...addr.matchAll(/\/p2p\/([^/]+)/g)]; + const last = matches[matches.length - 1]; + if (last) ids.push(last[1]!); + } + return ids; + } + + /** Configured-bootstrap peer IDs of a single network. */ + private configuredBootstrapPeerIDsOf(networkID: string): Set { + return new Set(Networks.bootstrapPeerIDsOf(this.get(networkID)?.bootstrapPeers ?? [])); + } + + /** Configured-bootstrap peer IDs of every joined network except `exceptID`. */ + private configuredBootstrapPeerIDsElsewhere(exceptID: string): Set { + const out = new Set(); + for (const nid of this.joinedNetworks) { + if (nid === exceptID) continue; + for (const pid of Networks.bootstrapPeerIDsOf(this.get(nid)?.bootstrapPeers ?? [])) out.add(pid); + } + return out; + } + private async leaveNetwork(id: string): Promise { if (!this.joinedNetworks.has(id)) return; + // Snapshot the topic subscribers BEFORE unsubscribing — unsubscribeTopic + // tears the topic out of pubsub, after which getTopicPeers(id) returns []. + // Union with recently-seen members (TTL) so a content peer that is momentarily + // disconnected at leave time — but still holds a peerStore entry — is also + // suppressed; a live-subscriber-only snapshot would miss it and maintenance + // could redial it back after we left. + const leftPeers = new Set(this.network.getTopicPeers(id)); + for (const pid of this.network.getRecentTopicMembers(id)) leftPeers.add(pid); + this.network.unsubscribeTopic(id); this.joinedNetworks.delete(id); + // Subscribers of any OTHER joined lishnet must stay connected (shared + // infrastructure). Compute this set BEFORE the bootstrap cleanup so that loop + // can skip them too — a bootstrap of the left net that also subscribes another + // joined net would otherwise be hung up here. + const stillJoinedPeers = new Set(); + for (const otherID of this.joinedNetworks) { + for (const pid of this.network.getTopicPeers(otherID)) stillJoinedPeers.add(pid); + } + + // Drop the exemption AND actively disconnect every configured bootstrap peer + // exclusive to the left lishnet — including ones offline at leave time. Such + // a peer never appears in leftPeers (the topic-subscriber snapshot), so the + // content-peer loop below would miss it: its keep-alive tag would survive and + // redial maintenance / ReconnectQueue would reconnect it within ~30s. Keep it + // if it still subscribes another joined lishnet, or if it is an active circuit + // relay we depend on. disconnectPeer is a safe no-op hangUp for an unconnected + // peer and always strips keep-alive + suppresses redial. + const stillConfigured = this.configuredBootstrapPeerIDsElsewhere(id); + for (const pid of this.configuredBootstrapPeerIDsOf(id)) { + if (stillConfigured.has(pid)) continue; + this.network.pruneConfiguredBootstrapPeer(pid); + if (stillJoinedPeers.has(pid)) continue; + if (this.network.isBootstrapOrRelayPeer(pid)) continue; + await this.network.disconnectPeer(pid, id); + } + + // Disconnect peers that belonged exclusively to the lishnet we just left. + // A peer is kept connected if it is still a subscriber of any OTHER joined + // lishnet, or if it is a bootstrap/relay peer (shared infrastructure other + // networks depend on). Everything else is a plain content peer with no + // remaining reason to stay connected, so hang it up via the single + // Network.disconnectPeer entry point (which also clears the keep-alive tag + // so ReconnectQueue does not immediately re-dial it). + for (const pid of leftPeers) { + if (stillJoinedPeers.has(pid)) continue; + if (this.network.isBootstrapOrRelayPeer(pid)) continue; + await this.network.disconnectPeer(pid, id); + } + const net = this.get(id); console.log(`✓ Left lishnet: ${net?.name ?? id}`); + + // Notify higher layers (e.g. transfer) so downloads bound exclusively to + // this lishnet can be stopped. + this._onNetworkLeft?.(id); } /** @@ -308,6 +424,15 @@ export class Networks { const existing = this.get(id); if (!existing) return null; const cleaned = bootstrapPeers.filter(p => typeof p === 'string' && p.trim().length > 0); + // Drop the bootstrap-exemption for peer IDs removed from this network's + // config, unless still configured for another joined network. Prevents a + // removed bootstrap entry from lingering as infrastructure that a later + // leave-network would refuse to disconnect. + const nextIDs = new Set(Networks.bootstrapPeerIDsOf(cleaned)); + const elsewhere = this.configuredBootstrapPeerIDsElsewhere(id); + for (const pid of Networks.bootstrapPeerIDsOf(existing.bootstrapPeers)) { + if (!nextIDs.has(pid) && !elsewhere.has(pid)) this.network.pruneConfiguredBootstrapPeer(pid); + } const next: LISHNetworkConfig = { ...existing, bootstrapPeers: cleaned }; updateLISHnet(this.db, next); this.network.pruneBootstrapStatus(id, cleaned); diff --git a/backend/src/protocol/downloader.ts b/backend/src/protocol/downloader.ts index 922dde56..f6fc49db 100644 --- a/backend/src/protocol/downloader.ts +++ b/backend/src/protocol/downloader.ts @@ -57,7 +57,11 @@ export class Downloader { private readonly dataServer: DataServer; private network: Network; private readonly downloadDir: string; - private readonly networkIDs: string[]; + private networkIDs: string[]; + // Immutable snapshot of the networks this download was created with. removeNetwork + // mutates networkIDs when a lishnet is left; addNetwork consults this to re-attach + // only networks the download was originally bound to when they are re-joined. + private readonly originalNetworkIDs: string[]; private lishID!: LISHid; private state: State = 'added'; private workMutex = new Mutex(); @@ -77,6 +81,8 @@ export class Downloader { */ private peerDiscoveryTimer: ReturnType | undefined; private retryTimer: ReturnType | undefined; + // Disposer for the network `peer:disconnect` subscription; called in destroy(). + private peerDisconnectDisposer: (() => void) | undefined; private needsManifest = false; private disabled = false; private destroyed = false; @@ -108,6 +114,48 @@ export class Downloader { return this.lishID; } + /** + * The lishnet network IDs this download is bound to (the networks across + * which it searches for and dials peers). Returned as a defensive copy so + * callers cannot mutate the downloader's internal list. + */ + getNetworkIDs(): string[] { + return [...this.networkIDs]; + } + + /** + * The lishnets this download was originally created with, before any + * {@link removeNetwork} shrank the active set. Used to decide which lishnet + * re-joins may resume a suspended download. Defensive copy. + */ + getOriginalNetworkIDs(): string[] { + return [...this.originalNetworkIDs]; + } + + /** + * Stop sourcing this download from a lishnet the node just left. Removes the + * network from the set so subsequent WANT broadcasts and topic-peer probes no + * longer reach the left lishnet; peers exclusive to it are hung up separately + * by the leave path. No-op if it was not one of this download's networks or if + * it is the only one left (the caller disables the whole download in that case). + */ + removeNetwork(networkID: string): void { + if (this.networkIDs.length <= 1 || !this.networkIDs.includes(networkID)) return; + this.networkIDs = this.networkIDs.filter(id => id !== networkID); + } + + /** + * Re-attach a lishnet dropped by {@link removeNetwork} when it was left, now that + * it is joined again — so WANT broadcasts and topic probes reach it once more on + * the next discovery cycle. No-op if the download was never bound to it (not in + * the original set) or it is already active. + */ + addNetwork(networkID: string): void { + if (!this.originalNetworkIDs.includes(networkID)) return; + if (this.networkIDs.includes(networkID)) return; + this.networkIDs = [...this.networkIDs, networkID]; + } + /** * Central state mutation. Validates the requested transition against ALLOWED_TRANSITIONS * and logs a warning (and bails) if the transition is not allowed. @@ -169,7 +217,7 @@ export class Downloader { this.errorDetail = detail; this.clearRetryTimer(); this.clearPeerDiscoveryTimer(); - if (this.lishID) unregisterHaveAnnouncementHandler(this.lishID); + this.disposeNetworkHandlers(); // Fire-and-forget close — stream may already be reset/aborted, which is benign. // Log at trace so real bugs (e.g. TypeError) can still be spotted in debug logs. this.peerManager.closeAll('setError'); @@ -257,6 +305,19 @@ export class Downloader { return this.disabled; } + /** + * Idempotently release network-level subscriptions: the peer:disconnect handler + * and the unicast HAVE announcement handler. Called from every terminal path + * (downloaded / error / destroy); a second call is a no-op. Without this a + * completed/errored downloader dropped from the transfer map (never destroyed) + * would pin itself and its peer/handler closures for the process lifetime. + */ + private disposeNetworkHandlers(): void { + this.peerDisconnectDisposer?.(); + this.peerDisconnectDisposer = undefined; + if (this.lishID) unregisterHaveAnnouncementHandler(this.lishID); + } + async destroy(): Promise { console.debug(`[DL] destroy ${this.lishID.slice(0, 8)}, state=${this.state}, peers=${this.peerManager.size()}`); this.disabled = true; @@ -264,7 +325,7 @@ export class Downloader { this.abortController.abort(); this.clearRetryTimer(); this.clearPeerDiscoveryTimer(); - if (this.lishID) unregisterHaveAnnouncementHandler(this.lishID); + this.disposeNetworkHandlers(); await this.peerManager.closeAllAwait('destroy'); // Notify frontend to reset peers/speed immediately const total = this.dataServer.getAllChunkCount(this.lishID) || 0; @@ -288,11 +349,37 @@ export class Downloader { }); } + /** + * Subscribe to network-wide peer disconnects so a vanished peer is removed + * from our per-LISH peer manager immediately, rather than lingering until the + * next probe/dial fails. Idempotent — a stale subscription is disposed first. + * The disposer is released in {@link destroy}. + */ + private registerPeerDisconnectHandler(): void { + this.peerDisconnectDisposer?.(); + this.peerDisconnectDisposer = this.network.onPeerDisconnect(peerID => this.dropPeer(peerID)); + } + + /** + * Remove a peer from this download's peer manager because the underlying + * libp2p connection dropped. Plain 'disconnect' disposition (not a punitive + * drop/ban) — the peer may reconnect and be re-discovered normally. No-op if + * the peer is not currently a member. + */ + dropPeer(peerID: string): void { + if (this.destroyed) return; + if (!this.peerManager.has(peerID)) return; + trace(`[DL] ${this.lishID?.slice(0, 8) ?? '?'}: dropping disconnected peer ${peerID.slice(0, 12)}`); + this.peerManager.remove(peerID, 'disconnect'); + } + constructor(downloadDir: string, network: Network, dataServer: DataServer, networkIDs: string | string[]) { this.downloadDir = downloadDir; this.network = network; this.dataServer = dataServer; - this.networkIDs = Array.isArray(networkIDs) ? networkIDs : [networkIDs]; + const ids = Array.isArray(networkIDs) ? [...networkIDs] : [networkIDs]; + this.networkIDs = ids; + this.originalNetworkIDs = [...ids]; this.fileAllocator = new FileAllocator(downloadDir); } @@ -307,6 +394,7 @@ export class Downloader { console.log(`[DL] Loading LISH: ${this.lish.name} (${this.lishID.slice(0, 8)}), ${this.dataServer.getMissingChunks(this.lishID).length} chunks to download`); this.missingChunks = this.dataServer.getMissingChunks(this.lishID); this.registerAnnouncementHandler(); + this.registerPeerDisconnectHandler(); this.transitionTo('initialized', 'init() done'); } @@ -326,6 +414,7 @@ export class Downloader { console.log(`[DL] Loading LISH: ${this.lish.name} (${this.lishID.slice(0, 8)}), awaiting manifest from peer`); } this.registerAnnouncementHandler(); + this.registerPeerDisconnectHandler(); this.transitionTo('initialized', 'initFromManifest() done'); } @@ -370,6 +459,10 @@ export class Downloader { this.downloadReject = reject; }); } + // Terminal success: release network subscriptions. The transfer layer drops a + // completed downloader from its map without calling destroy(), so this is the + // only place the handlers get torn down on the happy path. + this.disposeNetworkHandlers(); } async doWork(): Promise { diff --git a/backend/src/protocol/lish-handlers.ts b/backend/src/protocol/lish-handlers.ts index 10dcb75c..61bf8a4f 100644 --- a/backend/src/protocol/lish-handlers.ts +++ b/backend/src/protocol/lish-handlers.ts @@ -52,6 +52,12 @@ export class LISHServingHandlers { /** Handle a `want` pubsub message from a remote peer requesting chunk metadata. */ async handleWant(data: WantMessage, networkID: string, fromPeerID?: string): Promise { + // TODO(out of scope): enforce per-network ACL here — only answer a + // WANT if `data.lishID` is actually shared into `networkID`. Today we + // answer based solely on global upload-enabled state, so a peer on ANY + // joined lishnet can pull any LISH we upload. Implementing this needs a + // LISH↔networkIDs mapping in the DB which does not yet exist; until then + // the topic membership is the only (coarse) access boundary. if (!fromPeerID) { trace(`[NET] want ignored: no verified sender peerID`); return; @@ -121,6 +127,10 @@ export class LISHServingHandlers { * (same query can hit the same node from several peering paths). */ async handleSearchLishs(data: SearchLishsMessage, networkID: string, fromPeerID?: string): Promise { + // TODO(out of scope): scope search results to LISHs actually shared + // into `networkID` (hence the explicit `void networkID` below — the param + // is received but not yet used for filtering). Blocked on the same missing + // LISH↔networkIDs DB mapping as handleWant's ACL TODO. void networkID; if (!fromPeerID) { trace(`[NET] searchLishs ignored: no verified sender peerID`); diff --git a/backend/src/protocol/lish-protocol.ts b/backend/src/protocol/lish-protocol.ts index a940b6d2..217b4db6 100644 --- a/backend/src/protocol/lish-protocol.ts +++ b/backend/src/protocol/lish-protocol.ts @@ -237,6 +237,11 @@ export class LISHClient { } // Request a single chunk (can be called multiple times on same stream) + // TODO(out of scope): thread an AbortSignal through requestChunk so a + // download disabled mid-flight (e.g. by leaving its last lishnet) can cancel + // an in-progress chunk read immediately instead of waiting for the stream + // read to complete or time out. Requires plumbing the downloader's + // abortController.signal down through ChunkDownloader → LISHClient. async requestChunk(lishID: LISHid, chunkID: ChunkID): Promise { // Bail early if stream is already closed/aborted — treat as transient (peer unreachable), // not as a reason to permanently ban the peer. @@ -399,7 +404,21 @@ export function clearAllUploads(): void { const IO_ERROR_THRESHOLD = 3; // consecutive I/O errors before auto-disabling upload -export async function handleLISHProtocol(stream: Stream, dataServer: DataServer, remotePeerID?: string, connectionType?: ConnectionType): Promise { +/** + * Serve-gate predicate for the unicast LISH protocol. Returns true when a served + * response (LISH list / manifest / chunk) must be withheld because we do not share + * a joined lishnet with the remote peer. Fail CLOSED: when the gate is active + * (`sharesNetworkWith` provided) but the stream could not be mapped to a peer id + * (`remotePeerID` absent), block exactly like a not-shared peer — a bare or + * unauthenticated transport connection must never be served. Mirrors handleWant, + * which drops a WANT that has no verified sender peerID. + */ +export function serveGateBlocks(sharesNetworkWith: ((peerID: string) => boolean) | undefined, remotePeerID: string | undefined): boolean { + if (!sharesNetworkWith) return false; + return !remotePeerID || !sharesNetworkWith(remotePeerID); +} + +export async function handleLISHProtocol(stream: Stream, dataServer: DataServer, remotePeerID?: string, connectionType?: ConnectionType, sharesNetworkWith?: (peerID: string) => boolean, canListShares?: (peerID: string) => boolean): Promise { const servedLishIDs = new Set(); const ioErrorCounts = new Map(); // per-LISH consecutive I/O error counter const remotePeer = remotePeerID?.slice(0, 12) ?? 'unknown'; @@ -438,6 +457,17 @@ export async function handleLISHProtocol(stream: Stream, dataServer: DataServer, } if (request.type === 'getLishs') { + // Serve the shared-LISH LISTING under the softer list-gate (canListShares): + // unlike the strict data gate it does not require a synced gossipsub + // SUBSCRIBE, so the unicast search fallback reaches freshly-connected peers. + // Still fail-closed on an unknown peer id and still refuses peers we left. + // Falls back to the strict gate when no list-gate was supplied. + if (serveGateBlocks(canListShares ?? sharesNetworkWith, remotePeerID)) { + trace(`[PROTO] getLishs from ${remotePeer} refused: no shared joined lishnet`); + const gated: LISHGetLishsResponse = { type: 'getLishs-result', lishs: [] }; + sendLengthPrefixed(stream, codecEncode(gated)); + continue; + } // Return list of all shared (upload_enabled) LISHs — id and name only. // Newest first — matches the order shown locally in "Download and Sharing". const allLishs = dataServer.list(); @@ -460,8 +490,9 @@ export async function handleLISHProtocol(stream: Stream, dataServer: DataServer, }; sendLengthPrefixed(stream, codecEncode(response)); } else if (request.type === 'getLish') { - // Only return manifest for LISHs with upload enabled - if (!isUploadAdvertisable(request.lishID)) { + // Only return manifest for LISHs with upload enabled — and only to + // peers we share a joined lishnet with (same gate as getLishs). + if (serveGateBlocks(sharesNetworkWith, remotePeerID) || !isUploadAdvertisable(request.lishID)) { const response: LISHGetLishResponse = { error: ErrorCodes.PEER_LISH_NOT_SHARED }; sendLengthPrefixed(stream, codecEncode(response)); } else { @@ -481,6 +512,16 @@ export async function handleLISHProtocol(stream: Stream, dataServer: DataServer, } else if (request.type === 'getChunk' || request.type === undefined) { // Chunk request (type may be omitted for legacy compatibility) const chunkReq = request as LISHGetChunkRequest; + // Stop serving chunks to peers we no longer share a joined + // lishnet with. Without this a downloader re-dials our stored + // address after we left its network and the transfer silently + // continues over the fresh transport connection. + if (serveGateBlocks(sharesNetworkWith, remotePeerID)) { + trace(`[PROTO] getChunk from ${remotePeer} refused: no shared joined lishnet`); + const gatedResponse: LISHGetChunkResponse = { error: ErrorCodes.PEER_LISH_NOT_SHARED }; + sendLengthPrefixed(stream, codecEncode(gatedResponse)); + continue; + } if (!uploadEnabled.has(chunkReq.lishID)) { const blockedResponse: LISHGetChunkResponse = { error: ErrorCodes.PEER_LISH_NOT_SHARED }; sendLengthPrefixed(stream, codecEncode(blockedResponse)); diff --git a/backend/src/protocol/network.ts b/backend/src/protocol/network.ts index a8bcabac..ea6ea8f0 100644 --- a/backend/src/protocol/network.ts +++ b/backend/src/protocol/network.ts @@ -119,6 +119,14 @@ export class Network { */ private readonly seenSearchIDs = new Map(); private bootstrapPeerIDs: Set = new Set(); + /** + * Peer IDs whose bootstrap entries came from explicit network config + * ('configured' origin — startup config or a manual bootstrap edit). Kept + * separate from bootstrapPeerIDs, which also collects peer-announce + * discoveries: those are plain content peers and must remain + * disconnectable by lishnet leave (isBootstrapOrRelayPeer). + */ + private configuredBootstrapPeerIDs: Set = new Set(); private dcutrPeers: Set = new Set(); private bootstrapMultiaddrs: any[] = []; @@ -175,6 +183,12 @@ export class Network { /** Per-instance dedup set for PX ingress log keys; owned here, passed into gossipsub-patches deps. */ private readonly pxIngressLogKeys = new Set(); + /** + * Handlers subscribed via {@link onPeerDisconnect}. Held at Network level + * (not bound to a node instance) so subscriptions survive node restarts. + */ + private readonly peerDisconnectHandlers = new Set<(peerID: string) => void>(); + /** Handles incoming LISH-serving pubsub messages (want, searchLishs). */ private readonly lishHandlers: LISHServingHandlers; @@ -189,6 +203,20 @@ export class Network { */ private readonly redialBackoff = new Map(); + /** + * Peers deliberately hung up by {@link disconnectPeer} (leave-network), keyed by + * the lishnet they were left with. Redial maintenance / discovery must NOT + * proactively re-dial them — otherwise a peer we just left is re-connected within + * one status tick (~30s), defeating the leave. Per-network so rejoining lishnet A + * lifts only A's peers (not still-left B's — {@link clearRedialSuppressionForNetwork}). + * A peer observed reconnecting by any path is lifted from ALL sets + * ({@link clearRedialSuppressionForPeer}). Bounded by peers of currently-left + * lishnets; drained on rejoin/reconnect and cleared in stop(). NOT pruned against + * the peerStore: leave purges the peer from the store, which would drop the + * suppression and let mDNS rediscovery reconnect within a tick. + */ + private readonly redialSuppressedByNet = new Map>(); + // Tracked libp2p/pubsub event listeners for clean removal in stop(). // Each entry captures the exact handler reference so removeEventListener can unhook it. private listeners: Array<{ target: EventTarget; event: string; handler: (evt: any) => void }> = []; @@ -268,6 +296,23 @@ export class Network { }; } + /** + * Subscribe to peer disconnects for the duration of the returned disposer. + * The handler receives the disconnected peer's ID as a string. + * + * Handlers live at Network level, NOT on the current libp2p node: the + * permanent `peer:disconnect` listener installed by {@link start} fans out + * to this set, so subscriptions survive a node restart (identity + * import/regenerate) that would otherwise silently drop them together with + * the old node's listeners. Used by downloads to drop a vanished peer from + * their per-LISH peer manager immediately, instead of waiting for the next + * failed dial/probe to notice the dead connection. + */ + onPeerDisconnect(handler: (peerID: string) => void): () => void { + this.peerDisconnectHandlers.add(handler); + return () => this.peerDisconnectHandlers.delete(handler); + } + /** * Schedule a debounced check of peer counts for all subscribed topics. */ @@ -354,6 +399,8 @@ export class Network { myPeerID: privateKey.publicKey.toString(), }); this.bootstrapPeerIDs = bootstrapPeerIDs; + // Config-time bootstrap entries are by definition 'configured'. + this.configuredBootstrapPeerIDs = new Set(bootstrapPeerIDs); this.bootstrapMultiaddrs = bootstrapMultiaddrs; console.log('Creating libp2p node...'); @@ -431,7 +478,14 @@ export class Network { } } const connType = remotePeerID ? classifyConnectionFn(remotePeerID, isRelay, this.dcutrPeers) : 'DIRECT'; - await handleLISHProtocol(stream, this.dataServer, remotePeerID, connType); + await handleLISHProtocol( + stream, + this.dataServer, + remotePeerID, + connType, + pid => this.sharesJoinedTopicWith(pid), + pid => this.canListSharesTo(pid) + ); } catch (err: any) { trace(`[NET] LISH handler error: ${err?.message ?? err}`); } @@ -478,6 +532,10 @@ export class Network { // Skip if already connected (autoDial in v2 is unreliable; we dial actively // for mDNS/bootstrap discoveries to ensure local peers form a mesh quickly). if (peerID === this.node!.peerId.toString()) return; + // A peer we deliberately left (leave-network) must not be re-tagged or + // re-dialed by discovery (mDNS/identify/PX) — that would beat the disconnect. + // Suppression lifts on a legitimate inbound reconnect or on network rejoin. + if (this.isRedialSuppressed(peerID)) return; // Stamp `keep-alive-fleet` on every discovered peer, regardless of how they // surfaced (mDNS, bootstrap, autonat, identify, peer-announce). libp2p // ReconnectQueue only acts on peers with a tag whose key starts with @@ -526,6 +584,12 @@ export class Network { }); console.debug(' Tagged as KEEP_ALIVE (bootstrap peer)'); } + // A mere reconnect does NOT lift leave-network suppression: a peer we left can + // dial us back (its own keep-alive/mDNS) without rejoining a shared topic, and + // clearing here would remove the only marker canListSharesTo uses to refuse it. + // Suppression lifts only on an explicit rejoin (clearRedialSuppressionForNetwork) + // or once the peer is verifiably back on a joined topic (genuine mesh reconnect). + if (this.sharesJoinedTopicWith(peerID)) this.clearRedialSuppressionForPeer(peerID); this.schedulePeerCountCheck(); } catch (err: any) { trace(`[NET] peer:connect handler error: ${err?.message ?? err}`); @@ -561,6 +625,14 @@ export class Network { if (topic.startsWith(LISH_TOPIC_PREFIX)) this.lastMeshChange.set(topic, now); } } + // Fan out to Network-level subscribers (see onPeerDisconnect). + for (const h of this.peerDisconnectHandlers) { + try { + h(peerID); + } catch (err: any) { + trace(`[NET] peer-disconnect subscriber error: ${err?.message ?? err}`); + } + } this.schedulePeerCountCheck(); }); @@ -682,6 +754,40 @@ export class Network { // churn at N≈100 without flooding logs or burning CPU on per-second probes. } + /** + * Flat view over the per-network sets: whether a peer was deliberately hung up by + * leave-network (via disconnectPeer) for ANY left lishnet, so no maintenance path + * (redial loop, zero-connection recovery, promote, discovery) re-dials it. + */ + private isRedialSuppressed(peerID: string): boolean { + for (const set of this.redialSuppressedByNet.values()) if (set.has(peerID)) return true; + return false; + } + + /** Record a peer as left with a specific lishnet so maintenance won't re-dial it. */ + private addRedialSuppression(networkID: string, peerID: string): void { + let set = this.redialSuppressedByNet.get(networkID); + if (!set) { + set = new Set(); + this.redialSuppressedByNet.set(networkID, set); + } + set.add(peerID); + } + + /** + * Lift suppression for one lishnet's peers — called on (re)join of that lishnet. + * Scoped: rejoining A does not unblock still-left B's peers (nor lift the + * canListSharesTo browse-privacy protecting B). + */ + clearRedialSuppressionForNetwork(networkID: string): void { + this.redialSuppressedByNet.delete(networkID); + } + + /** Lift suppression for one peer across ALL left lishnets — a legitimate reconnect. */ + private clearRedialSuppressionForPeer(peerID: string): void { + for (const set of this.redialSuppressedByNet.values()) set.delete(peerID); + } + private async runRedialMaintenance(connectedPeers: any[], allPeers: any[]): Promise { // Dial known peers not currently connected (maintains relay connections to NATed peers) const connectedSet = new Set(connectedPeers.map(p => p.toString())); @@ -692,11 +798,19 @@ export class Network { const candidates: Array<{ peer: any; pid: string; addrSummary: string; failCount: number }> = []; let skippedBackoff = 0; let skippedNoReachable = 0; + let skippedSuppressed = 0; const localCidrs = getLocalCidrs(now); for (const peer of allPeers) { const pid = peer.id.toString(); if (connectedSet.has(pid)) { this.redialBackoff.delete(pid); // clear on observed connection + if (this.sharesJoinedTopicWith(pid)) this.clearRedialSuppressionForPeer(pid); // back on a shared topic → resume + continue; + } + // Skip peers we deliberately left (leave-network) so maintenance does not + // silently re-dial them; cleared above once they reconnect on their own. + if (this.isRedialSuppressed(pid)) { + skippedSuppressed++; continue; } const bo = this.redialBackoff.get(pid); @@ -765,10 +879,14 @@ export class Network { }; const workers = Array.from({ length: Math.min(CONCURRENCY, candidates.length) }, () => worker()); await Promise.all(workers); - if (candidates.length > 0 || skippedBackoff > 0 || skippedNoReachable > 0) { - console.debug(` Re-dial: ${redialSuccess}/${candidates.length} succeeded (${skippedBackoff} skipped by backoff, ${skippedNoReachable} skipped no-reachable-addrs)`); + if (candidates.length > 0 || skippedBackoff > 0 || skippedNoReachable > 0 || skippedSuppressed > 0) { + console.debug(` Re-dial: ${redialSuccess}/${candidates.length} succeeded (${skippedBackoff} skipped by backoff, ${skippedNoReachable} skipped no-reachable-addrs, ${skippedSuppressed} skipped left-peer)`); } - // Prune backoff entries for peers that are no longer in peerStore to prevent unbounded growth. + // Prune backoff entries for peers no longer in peerStore to prevent unbounded growth. + // Suppression is NOT pruned this way: leave-network purges the peer from the + // peerStore, so pruning against it would drop the suppression and let mDNS + // rediscovery reconnect the left peer within a tick. Suppression is instead + // bounded by clear-on-rejoin / clear-on-reconnect / stop(). const storeSet = new Set(allPeers.map(p => p.id.toString())); for (const pid of this.redialBackoff.keys()) if (!storeSet.has(pid)) this.redialBackoff.delete(pid); } @@ -795,6 +913,9 @@ export class Network { console.log(` [NET-CHURN] bootstrap stats net=${networkID.slice(0, 8)}: ${parts}`); } for (const ma of this.bootstrapMultiaddrs) { + const p2pComponents = ma.getComponents().filter((c: { code: number; value?: string }) => c.code === 421); + const pid: string | undefined = p2pComponents.length > 0 ? p2pComponents[p2pComponents.length - 1].value : undefined; + if (pid && this.isRedialSuppressed(pid)) continue; // deliberately left — don't resurrect it here const maStr = ma?.toString?.() ?? String(ma); try { console.log(` → Dialing ${maStr}`); @@ -837,6 +958,7 @@ export class Network { for (const peer of allPeers) { const pid = peer.id.toString(); if (pid === myID) continue; + if (this.isRedialSuppressed(pid)) continue; // deliberately left — don't promote it back to bootstrap if (this.bootstrapPeerIDs.has(pid)) continue; if (peer.addresses.length === 0) continue; const addr = peer.addresses[0]!; @@ -858,6 +980,10 @@ export class Network { for (const peer of allPeers) { const pid = peer.id.toString(); if (pid === myID) continue; + // A left-network peer that lingers/reappears in the peerStore must not be + // added to the direct set either — its fast reconnect cadence would undo the + // leave-network disconnect (same guard as the bootstrap promotion above). + if (this.isRedialSuppressed(pid)) continue; if (!gossipsub.direct.has(pid)) { gossipsub.direct.add(pid); added++; @@ -917,7 +1043,20 @@ export class Network { trace(`[NET] addBootstrapPeers skip non-routable: ${peer}`); continue; } - const peerID = ma.getComponents().find(c => c.code === 421)?.value ?? null; + // A relayed bootstrap multiaddr (.../p2p//p2p-circuit/p2p/) + // carries two /p2p components; the peer we actually connect to — and must + // exempt from leave-network disconnect — is the FINAL one (the target), not + // the relay. Take the last /p2p component, never the first. + const p2pComponents = ma.getComponents().filter(c => c.code === 421); + const peerID = (p2pComponents.length > 0 ? p2pComponents[p2pComponents.length - 1]!.value : null) ?? null; + if (peerID && origin === 'configured') { + this.configuredBootstrapPeerIDs.add(peerID); + // A re-configured bootstrap peer means its network was (re-)joined — it + // is no longer "left", so lift any redial suppression left by a prior + // leaveNetwork, otherwise maintenance would skip it forever if this one + // explicit dial fails or the connection drops before the next tick. + this.clearRedialSuppressionForPeer(peerID); + } const alreadyKnown = !!peerID && this.bootstrapPeerIDs.has(peerID); if (peerID && !alreadyKnown) { this.bootstrapPeerIDs.add(peerID); @@ -1019,6 +1158,157 @@ export class Network { } } + /** + * True if the peer is one we must never voluntarily disconnect because it + * provides infrastructure rather than being a plain content peer: an + * explicitly configured bootstrap peer, or a relay some of our circuit + * connections are routed THROUGH (dropping it would also kill transit for + * any NAT'd peers reachable only via that relay). + * + * Peer-announce-discovered bootstrap entries and peers merely REACHED over + * a relay are plain content peers — hanging those up touches only their own + * connection, so lishnet leave may disconnect them. + * + * Used by lishnet leave to decide which topic peers are safe to hang up — + * leaving an empty lishnet must not tear down shared bootstrap/relay links + * that other still-joined lishnets depend on. + */ + /** + * Drop a peer from the configured-bootstrap exemption set. Called by the + * lishnet layer when a bootstrap entry is removed from config or belongs only + * to a lishnet being left, so `isBootstrapOrRelayPeer` stops treating a peer + * that is no longer configured (nor shared with another joined network) as + * infrastructure that leave-network must keep connected. + */ + pruneConfiguredBootstrapPeer(peerID: string): void { + this.configuredBootstrapPeerIDs.delete(peerID); + } + + isBootstrapOrRelayPeer(peerID: string): boolean { + if (this.configuredBootstrapPeerIDs.has(peerID)) return true; + if (!this.node) return false; + try { + // A relay's ID is the hop right before /p2p-circuit in a circuit address: + // /ip4/../tcp/../p2p//p2p-circuit/p2p/ + for (const c of this.node.getConnections()) { + if (!Circuit.matches(c.remoteAddr)) continue; + const relayPrefix = c.remoteAddr.toString().split('/p2p-circuit')[0]!; + if (relayPrefix.endsWith(`/p2p/${peerID}`)) return true; + } + return false; + } catch { + return false; + } + } + + /** Recently-seen subscribers of a lishnet's topic (TTL union, not just the live snapshot). */ + getRecentTopicMembers(networkID: string): string[] { + return this.peerAnnounce.getRecentMembers(lishTopic(networkID)); + } + + /** + * True if we currently share at least one joined lishnet topic with the + * given peer — i.e. some lish topic WE are subscribed to lists the peer + * among its subscribers. + * + * Coarse serve-gate for unicast LISH discovery: a peer we no longer + * share any lishnet with must not be able to browse or search our shared + * LISHs just because a transport connection exists (e.g. the peer's + * keep-alive re-dialed us right after we left its network). + */ + sharesJoinedTopicWith(peerID: string): boolean { + if (!this.pubsub) return false; + for (const topic of this.pubsub.getTopics()) { + if (!topic.startsWith(LISH_TOPIC_PREFIX)) continue; + try { + if (this.pubsub.getSubscribers(topic).some((p: any) => p.toString() === peerID)) return true; + } catch { + // topic may be tearing down — treat as not shared + } + } + return false; + } + + /** + * Softer gate for the low-sensitivity shared-LISH LISTING (getLishs) only — + * data requests (getLish/getChunk) stay on the strict {@link sharesJoinedTopicWith} + * fail-closed gate. {@link sharesJoinedTopicWith} relies on gossipsub's subscriber + * view, which lags for a freshly-connected peer whose SUBSCRIBE has not propagated + * yet — the exact window the unicast search fallback targets, so the listing must + * not be withheld there. Serve the listing to any peer over an authenticated stream + * while we are in at least one lishnet, EXCEPT one we deliberately left (still in + * redial suppression) — that preserves the leave-network browse privacy. + */ + canListSharesTo(peerID: string): boolean { + if (this.isRedialSuppressed(peerID)) return false; + if (!this.pubsub) return false; + // Infrastructure peers (active relay / bootstrap) are kept connected across a + // leave without being redial-suppressed, so the softer gate alone would let a + // relay of a network we just left browse our shares. Require such peers to + // currently share a joined topic. Ordinary content peers still get the soft + // gate — that is the freshly-connected-before-SUBSCRIBE window the search + // fallback depends on. + if (this.isBootstrapOrRelayPeer(peerID) && !this.sharesJoinedTopicWith(peerID)) return false; + return this.pubsub.getTopics().some((t: string) => t.startsWith(LISH_TOPIC_PREFIX)); + } + + /** + * Gracefully disconnect from a single peer and stop libp2p from immediately + * re-dialing it. This is the ONLY place that should call `node.hangUp()` so + * the accompanying ReconnectQueue cleanup (removing the `keep-alive-fleet` + * tag) is never forgotten — without dropping that tag, `peer:discovery` / + * ReconnectQueue would re-dial the peer within seconds and the disconnect + * would be pointless. + * + * Also forgets the peerStore entry (via {@link purgeStalePeer}): in-memory redial + * suppression is lost on restart, but the persisted peerStore is not, so a leave + * followed by a restart before rejoin would otherwise let redial maintenance dial + * the left peer straight back. The sole caller (leaveNetwork) only passes peers + * with no remaining reason to stay, and rejoin re-acquires the entry via + * bootstrap/discovery. Best-effort: failures are logged at trace, never thrown. + * + * `networkID` is the lishnet the peer is being left with — the peer is suppressed + * under it so rejoining that lishnet lifts exactly its peers. + */ + async disconnectPeer(peerID: string, networkID: string): Promise { + if (!this.node) return; + let pid: PeerID; + try { + pid = peerIDFromString(peerID); + } catch (err: any) { + trace(`[NET] disconnectPeer: invalid peerID ${peerID.slice(0, 16)}: ${err?.message ?? err}`); + return; + } + // Remove the keep-alive tags FIRST so the imminent hangUp does not race + // the ReconnectQueue back into a re-dial. Both tags matter: the custom + // 'keep-alive-fleet' tag (peer-announce intake) and the native KEEP_ALIVE + // tag (stamped by addBootstrapPeers on every successfully dialed entry, + // including discovered ones) — libp2p itself re-dials any peer carrying a + // keep-alive tag, which would silently undo this disconnect. Passing + // undefined as the tag value removes it (per @libp2p/interface PeerStore + // merge semantics). + try { + await this.node.peerStore.merge(pid, { + tags: { 'keep-alive-fleet': undefined, [KEEP_ALIVE]: undefined }, + }); + } catch (err: any) { + trace(`[NET] disconnectPeer: tag removal failed for ${peerID.slice(0, 16)}: ${err?.message ?? err}`); + } + try { + await this.node.hangUp(pid); + trace(`[NET] disconnectPeer: hung up ${peerID.slice(0, 16)}`); + } catch (err: any) { + trace(`[NET] disconnectPeer: hangUp failed for ${peerID.slice(0, 16)}: ${err?.message ?? err}`); + } + // Keep redial maintenance from re-dialing this just-left peer on the next + // status tick. Keyed by the left lishnet so rejoin lifts exactly its peers; + // cleared automatically once it reconnects legitimately. + this.addRedialSuppression(networkID, peerID); + // Forget the persisted peerStore entry so the disconnect survives a restart — + // suppression is in-memory only, but the peerStore is on disk. + await this.purgeStalePeer(peerID, 'left-network exclusive peer'); + } + /** Snapshot of all per-network bootstrap statuses. */ getAllBootstrapStatuses(): BootstrapStatus[] { return this.bootstrapTracker.getAllStatuses(); @@ -1416,6 +1706,7 @@ export class Network { this._lastPeerCounts.clear(); this._lastScores.clear(); this.redialBackoff.clear(); + this.redialSuppressedByNet.clear(); this.pxIngressLogKeys.clear(); if (this.node) { await this.node.stop(); diff --git a/backend/src/protocol/peer-announce.ts b/backend/src/protocol/peer-announce.ts index 1d98348d..686dc49c 100644 --- a/backend/src/protocol/peer-announce.ts +++ b/backend/src/protocol/peer-announce.ts @@ -98,6 +98,21 @@ export class PeerAnnounceManager { this.deps = deps; } + /** + * Recently-seen subscribers of a topic (union of getSubscribers over the last + * {@link PEER_ANNOUNCE_MEMBER_TTL_MS}), so a same-network peer that is momentarily + * disconnected at query time is still reported. Used by leave-network to suppress + * offline content peers that a live subscriber snapshot would miss. + */ + getRecentMembers(topic: string): string[] { + const members = this.topicMembers.get(topic); + if (!members) return []; + const now = Date.now(); + const out: string[] = []; + for (const [pid, seen] of members) if (now - seen <= PEER_ANNOUNCE_MEMBER_TTL_MS) out.push(pid); + return out; + } + /** Start the periodic emitter. Safe to call only once per start/stop cycle. */ start(): void { this.stopped = false; diff --git a/backend/tests/e2e/transfer-integration.test.ts b/backend/tests/e2e/transfer-integration.test.ts index e311fcc2..df55d04e 100644 --- a/backend/tests/e2e/transfer-integration.test.ts +++ b/backend/tests/e2e/transfer-integration.test.ts @@ -109,6 +109,13 @@ class MockNetwork { subscribedTopics: Array<{ topic: string; handler: (data: Record) => void }> = []; broadcastMessages: Array<{ topic: string; data: Record }> = []; dialResults = new Map(); + /** Registered peer-disconnect handlers — mirrors Network.onPeerDisconnect. */ + readonly peerDisconnectHandlers: Set<(peerID: string) => void> = new Set(); + + onPeerDisconnect(handler: (peerID: string) => void): () => void { + this.peerDisconnectHandlers.add(handler); + return () => this.peerDisconnectHandlers.delete(handler); + } async subscribe(topic: string, handler: (data: Record) => void): Promise { this.subscribedTopics.push({ topic, handler }); diff --git a/backend/tests/unit/helpers/mock-network.ts b/backend/tests/unit/helpers/mock-network.ts index 571d279c..dcd7834f 100644 --- a/backend/tests/unit/helpers/mock-network.ts +++ b/backend/tests/unit/helpers/mock-network.ts @@ -31,6 +31,20 @@ export class MockNetwork { return []; } + /** Registered peer-disconnect handlers — invoke {@link emitPeerDisconnect} to simulate a peer dropping. */ + readonly peerDisconnectHandlers: Set<(peerID: string) => void> = new Set(); + + /** Mirrors Network.onPeerDisconnect: registers a handler and returns a disposer. */ + onPeerDisconnect(handler: (peerID: string) => void): () => void { + this.peerDisconnectHandlers.add(handler); + return () => this.peerDisconnectHandlers.delete(handler); + } + + /** Test helper: simulate a peer disconnect, invoking every registered handler. */ + emitPeerDisconnect(peerID: string): void { + for (const handler of this.peerDisconnectHandlers) handler(peerID); + } + isRunning(): boolean { return false; } diff --git a/backend/tests/unit/lishnet/leave-network.test.ts b/backend/tests/unit/lishnet/leave-network.test.ts new file mode 100644 index 00000000..705616ec --- /dev/null +++ b/backend/tests/unit/lishnet/leave-network.test.ts @@ -0,0 +1,250 @@ +import { describe, it, expect, beforeEach } from 'bun:test'; +import { Networks } from '../../../src/lishnet/lishnets.ts'; + +/** + * Unit tests for Networks.leaveNetwork peer-disconnect behaviour: + * on leaving a lishnet, peers that belonged exclusively to it are hung up, + * while peers shared with another joined lishnet and bootstrap/relay peers + * stay connected. Uses a bare instance (Object.create) so no real libp2p + * node or database is needed. + */ + +interface MockNet { + topicPeers: Map; + recentMembers: Map; + unsubscribed: string[]; + subscribed: string[]; + disconnected: string[]; + bootstrapOrRelay: Set; + prunedBootstrap: string[]; + getTopicPeers(id: string): string[]; + getRecentTopicMembers(id: string): string[]; + unsubscribeTopic(id: string): void; + subscribeTopic(id: string): void; + isBootstrapOrRelayPeer(pid: string): boolean; + disconnectPeer(pid: string, networkID: string): Promise; + pruneConfiguredBootstrapPeer(pid: string): void; + clearRedialSuppressionForNetwork(networkID: string): void; + suppressionClearedFor: string[]; +} + +function makeMockNet(): MockNet { + return { + topicPeers: new Map(), + recentMembers: new Map(), + unsubscribed: [], + subscribed: [], + disconnected: [], + bootstrapOrRelay: new Set(), + prunedBootstrap: [], + suppressionClearedFor: [], + getTopicPeers(id) { + return this.topicPeers.get(id) ?? []; + }, + getRecentTopicMembers(id) { + return this.recentMembers.get(id) ?? []; + }, + unsubscribeTopic(id) { + this.unsubscribed.push(id); + // Mirror real pubsub: after unsubscribe the topic reports no peers. + this.topicPeers.delete(id); + }, + subscribeTopic(id) { + this.subscribed.push(id); + }, + isBootstrapOrRelayPeer(pid) { + return this.bootstrapOrRelay.has(pid); + }, + async disconnectPeer(pid) { + this.disconnected.push(pid); + }, + pruneConfiguredBootstrapPeer(pid) { + this.prunedBootstrap.push(pid); + }, + clearRedialSuppressionForNetwork(networkID) { + this.suppressionClearedFor.push(networkID); + }, + }; +} + +// bootstrapPeers per network id, exposed to the class via `get`. +function makeNetworks(net: MockNet, joined: string[], configs: Record = {}): Networks { + const networks = Object.create(Networks.prototype) as Networks; + (networks as any).network = net; + (networks as any).joinedNetworks = new Set(joined); + (networks as any)._onNetworkLeft = null; + (networks as any)._onNetworkJoined = null; + (networks as any).get = (id: string) => (configs[id] ? { networkID: id, bootstrapPeers: configs[id] } : undefined); + return networks; +} + +const leave = (networks: Networks, id: string): Promise => (networks as any).leaveNetwork(id); +const join = (networks: Networks, id: string): Promise => (networks as any).joinNetwork(id); + +describe('Networks.leaveNetwork — exclusive peer disconnect', () => { + let net: MockNet; + + beforeEach(() => { + net = makeMockNet(); + }); + + it('disconnects peers that were only in the left lishnet, keeps shared ones', async () => { + net.topicPeers.set('net-a', ['p-only-a', 'p-shared']); + net.topicPeers.set('net-b', ['p-shared']); + const networks = makeNetworks(net, ['net-a', 'net-b']); + await leave(networks, 'net-a'); + expect(net.disconnected).toEqual(['p-only-a']); + }); + + it('disconnects a recently-seen content peer offline at leave time', async () => { + // Not a live subscriber right now, but seen within TTL and still in the peerStore — + // must be suppressed too, or maintenance would redial it after the leave. + net.topicPeers.set('net-a', ['p-live']); + net.recentMembers.set('net-a', ['p-live', 'p-offline']); + const networks = makeNetworks(net, ['net-a']); + await leave(networks, 'net-a'); + expect(net.disconnected.sort()).toEqual(['p-live', 'p-offline']); + }); + + it('keeps bootstrap/relay peers even when exclusive to the left lishnet', async () => { + net.topicPeers.set('net-a', ['p-bootstrap', 'p-plain']); + net.bootstrapOrRelay.add('p-bootstrap'); + const networks = makeNetworks(net, ['net-a']); + await leave(networks, 'net-a'); + expect(net.disconnected).toEqual(['p-plain']); + }); + + it('snapshots topic peers before unsubscribing', async () => { + net.topicPeers.set('net-a', ['p1', 'p2']); + const networks = makeNetworks(net, ['net-a']); + await leave(networks, 'net-a'); + // The mock wipes the topic on unsubscribe — a post-unsubscribe read would + // have seen [] and disconnected nobody. + expect(net.unsubscribed).toEqual(['net-a']); + expect(net.disconnected).toEqual(['p1', 'p2']); + }); + + it('is a no-op for a lishnet that is not joined', async () => { + net.topicPeers.set('net-a', ['p1']); + const networks = makeNetworks(net, []); + let leftFired = 0; + networks.onNetworkLeft = () => leftFired++; + await leave(networks, 'net-a'); + expect(net.unsubscribed).toEqual([]); + expect(net.disconnected).toEqual([]); + expect(leftFired).toBe(0); + }); + + it('fires onNetworkLeft with the left lishnet id and un-joins it', async () => { + net.topicPeers.set('net-a', []); + const networks = makeNetworks(net, ['net-a']); + const leftIDs: string[] = []; + networks.onNetworkLeft = id => leftIDs.push(id); + await leave(networks, 'net-a'); + expect(leftIDs).toEqual(['net-a']); + expect((networks as any).joinedNetworks.has('net-a')).toBe(false); + }); + + it('prunes bootstrap exemption for peers configured only for the left lishnet', async () => { + net.topicPeers.set('net-a', []); + const networks = makeNetworks(net, ['net-a', 'net-b'], { + 'net-a': ['/ip4/192.0.2.1/tcp/9090/p2p/pOnlyA', '/ip4/192.0.2.2/tcp/9090/p2p/pShared'], + 'net-b': ['/ip4/192.0.2.3/tcp/9090/p2p/pShared'], + }); + await leave(networks, 'net-a'); + // pOnlyA is bootstrap only for the left network → exemption pruned. + // pShared is still bootstrap for the joined net-b → exemption kept. + expect(net.prunedBootstrap).toEqual(['pOnlyA']); + }); + + it('prunes the final /p2p target id for a relayed bootstrap multiaddr, not the relay', async () => { + net.topicPeers.set('net-a', []); + const networks = makeNetworks(net, ['net-a'], { + // Relayed entry: /p2p//p2p-circuit/p2p/. The bootstrap + // identity is the target (final /p2p), never the relay. + 'net-a': ['/ip4/192.0.2.10/tcp/9090/p2p/pRelay/p2p-circuit/p2p/pTarget'], + }); + await leave(networks, 'net-a'); + expect(net.prunedBootstrap).toEqual(['pTarget']); + }); + + it('disconnects an offline configured bootstrap peer of the left lishnet', async () => { + // pBootA is configured bootstrap for net-a only and is NOT a current topic + // subscriber (offline at leave time). It must still be disconnected so its + // keep-alive tag is stripped and redial maintenance cannot reconnect it. + net.topicPeers.set('net-a', []); + const networks = makeNetworks(net, ['net-a'], { + 'net-a': ['/ip4/192.0.2.5/tcp/9090/p2p/pBootA'], + }); + await leave(networks, 'net-a'); + expect(net.disconnected).toEqual(['pBootA']); + expect(net.prunedBootstrap).toEqual(['pBootA']); + }); + + it('keeps an offline bootstrap peer still configured for another joined lishnet', async () => { + net.topicPeers.set('net-a', []); + const networks = makeNetworks(net, ['net-a', 'net-b'], { + 'net-a': ['/ip4/192.0.2.5/tcp/9090/p2p/pShared'], + 'net-b': ['/ip4/192.0.2.6/tcp/9090/p2p/pShared'], + }); + await leave(networks, 'net-a'); + expect(net.disconnected).toEqual([]); + expect(net.prunedBootstrap).toEqual([]); + }); + + it('keeps a left-lishnet bootstrap peer that still subscribes another joined lishnet', async () => { + net.topicPeers.set('net-a', []); + net.topicPeers.set('net-b', ['pBootA']); // pBootA is a live subscriber of net-b + const networks = makeNetworks(net, ['net-a', 'net-b'], { + 'net-a': ['/ip4/192.0.2.5/tcp/9090/p2p/pBootA'], + }); + await leave(networks, 'net-a'); + expect(net.disconnected).toEqual([]); // kept — subscriber of joined net-b + expect(net.prunedBootstrap).toEqual(['pBootA']); // exemption still pruned + }); + + it('keeps a left-lishnet bootstrap peer that is still an active circuit relay', async () => { + net.topicPeers.set('net-a', []); + net.bootstrapOrRelay.add('pRelayNode'); // still relaying another connection + const networks = makeNetworks(net, ['net-a'], { + 'net-a': ['/ip4/192.0.2.7/tcp/9090/p2p/pRelayNode'], + }); + await leave(networks, 'net-a'); + expect(net.disconnected).toEqual([]); + // Exemption is still pruned — the relay status alone keeps it connected. + expect(net.prunedBootstrap).toEqual(['pRelayNode']); + }); +}); + +describe('Networks.joinNetwork — onNetworkJoined notification', () => { + let net: MockNet; + + beforeEach(() => { + net = makeMockNet(); + }); + + it('fires onNetworkJoined with the joined lishnet id and joins it', async () => { + const networks = makeNetworks(net, []); + const joinedIDs: string[] = []; + networks.onNetworkJoined = id => joinedIDs.push(id); + await join(networks, 'net-a'); + expect(net.subscribed).toEqual(['net-a']); + expect(joinedIDs).toEqual(['net-a']); + expect((networks as any).joinedNetworks.has('net-a')).toBe(true); + }); + + it('does not fire onNetworkJoined for a lishnet already joined', async () => { + const networks = makeNetworks(net, ['net-a']); + let fired = 0; + networks.onNetworkJoined = () => fired++; + await join(networks, 'net-a'); + expect(fired).toBe(0); + expect(net.subscribed).toEqual([]); + }); + + it('lifts redial suppression for the joined lishnet so its left peers can reconnect', async () => { + const networks = makeNetworks(net, []); + await join(networks, 'net-a'); + expect(net.suppressionClearedFor).toEqual(['net-a']); + }); +}); diff --git a/backend/tests/unit/protocol/downloader-remove-network.test.ts b/backend/tests/unit/protocol/downloader-remove-network.test.ts new file mode 100644 index 00000000..df9553bd --- /dev/null +++ b/backend/tests/unit/protocol/downloader-remove-network.test.ts @@ -0,0 +1,68 @@ +import { describe, it, expect } from 'bun:test'; +import { Downloader } from '../../../src/protocol/downloader.ts'; + +/** + * Unit tests for Downloader.removeNetwork: leaving one lishnet of a multi-network + * download must drop that network from the set (so WANT broadcasts / topic probes + * stop reaching it), while never emptying the set — the caller disables the whole + * download when the last network is left. + */ + +function makeDownloader(networkIDs: string[]): Downloader { + const dl = Object.create(Downloader.prototype) as Downloader; + (dl as any).networkIDs = [...networkIDs]; + (dl as any).originalNetworkIDs = [...networkIDs]; + return dl; +} + +describe('Downloader.removeNetwork', () => { + it('removes one network from a multi-network download', () => { + const dl = makeDownloader(['net-a', 'net-b']); + dl.removeNetwork('net-a'); + expect(dl.getNetworkIDs()).toEqual(['net-b']); + }); + + it('is a no-op when the network is the only one left', () => { + const dl = makeDownloader(['net-b']); + dl.removeNetwork('net-b'); + expect(dl.getNetworkIDs()).toEqual(['net-b']); + }); + + it('is a no-op for a network the download is not bound to', () => { + const dl = makeDownloader(['net-a', 'net-b']); + dl.removeNetwork('net-c'); + expect(dl.getNetworkIDs()).toEqual(['net-a', 'net-b']); + }); +}); + +describe('Downloader.addNetwork', () => { + it('re-adds a network that was previously removed', () => { + const dl = makeDownloader(['net-a', 'net-b']); + dl.removeNetwork('net-a'); + expect(dl.getNetworkIDs()).toEqual(['net-b']); + dl.addNetwork('net-a'); + expect(dl.getNetworkIDs()).toEqual(['net-b', 'net-a']); + }); + + it('is a no-op for a network never in the original set', () => { + const dl = makeDownloader(['net-a', 'net-b']); + dl.addNetwork('net-c'); + expect(dl.getNetworkIDs()).toEqual(['net-a', 'net-b']); + }); + + it('is a no-op when the network is already active', () => { + const dl = makeDownloader(['net-a', 'net-b']); + dl.addNetwork('net-a'); + expect(dl.getNetworkIDs()).toEqual(['net-a', 'net-b']); + }); +}); + +describe('Downloader.getOriginalNetworkIDs', () => { + it('stays the full original set even after removeNetwork shrinks the active set', () => { + const dl = makeDownloader(['net-a', 'net-b']); + dl.removeNetwork('net-a'); + expect(dl.getNetworkIDs()).toEqual(['net-b']); + // Resume-on-rejoin binds to this, so leaving+rejoining net-a can still resume. + expect(dl.getOriginalNetworkIDs()).toEqual(['net-a', 'net-b']); + }); +}); diff --git a/backend/tests/unit/protocol/downloader.test.ts b/backend/tests/unit/protocol/downloader.test.ts index 420c207e..1484986d 100644 --- a/backend/tests/unit/protocol/downloader.test.ts +++ b/backend/tests/unit/protocol/downloader.test.ts @@ -1295,3 +1295,77 @@ describe('Downloader — inline ENOSPC retry', () => { expect(pc.writeResolvers.length).toBe(0); }); }); + +// --------------------------------------------------------------------------- +// Network peer:disconnect handling +// --------------------------------------------------------------------------- + +describe('Downloader – network peer:disconnect handling', () => { + type PeerManagerView = { + tryAdd: (peerID: string, client: unknown, connectionType: 'DIRECT' | 'RELAY' | 'DCUtR') => boolean; + has: (peerID: string) => boolean; + isDropped: (peerID: string) => boolean; + isBanned: (peerID: string) => boolean; + canDial: (peerID: string) => boolean; + }; + + let net: MockNetwork; + let downloader: Downloader; + + const pm = (): PeerManagerView => priv(downloader)['peerManager'] as PeerManagerView; + + beforeEach(async () => { + net = new MockNetwork(); + const ds = new MockDataServer(); + ds.missingChunks = []; + downloader = new Downloader('/tmp/dl', net as never, ds as never, 'net-001'); + await downloader.initFromManifest(makeLISH()); + }); + + afterEach(async () => { + await downloader.destroy(); + }); + + it('initFromManifest subscribes exactly one peer:disconnect handler', () => { + expect(net.peerDisconnectHandlers.size).toBe(1); + }); + + it('a disconnected peer is removed from the peer manager', () => { + pm().tryAdd('peer-gone', new MockLISHClient() as never, 'DIRECT'); + pm().tryAdd('peer-stays', new MockLISHClient() as never, 'DIRECT'); + net.emitPeerDisconnect('peer-gone'); + expect(pm().has('peer-gone')).toBe(false); + expect(pm().has('peer-stays')).toBe(true); + }); + + it('disconnect removal is plain — peer is neither dropped nor banned and may re-dial', () => { + pm().tryAdd('peer-flap', new MockLISHClient() as never, 'DIRECT'); + net.emitPeerDisconnect('peer-flap'); + expect(pm().isDropped('peer-flap')).toBe(false); + expect(pm().isBanned('peer-flap')).toBe(false); + expect(pm().canDial('peer-flap')).toBe(true); + }); + + it('disconnect of a peer not in the peer manager is a no-op', () => { + pm().tryAdd('peer-a', new MockLISHClient() as never, 'DIRECT'); + expect(() => net.emitPeerDisconnect('peer-unknown')).not.toThrow(); + expect(pm().has('peer-a')).toBe(true); + }); + + it('destroy() disposes the peer:disconnect subscription', async () => { + await downloader.destroy(); + expect(net.peerDisconnectHandlers.size).toBe(0); + }); + + it('successful completion disposes the peer:disconnect subscription', async () => { + expect(net.peerDisconnectHandlers.size).toBe(1); + // needsManifest path parks download() on its internal completion promise + // without touching the filesystem; resolve it to simulate a finished download. + const done = downloader.download(); + while (!priv(downloader)['downloadResolve']) await new Promise(r => setTimeout(r, 0)); + priv(downloader)['state'] = 'downloaded'; + (priv(downloader)['downloadResolve'] as () => void)(); + await done; + expect(net.peerDisconnectHandlers.size).toBe(0); + }); +}); diff --git a/backend/tests/unit/protocol/infra-peer-classification.test.ts b/backend/tests/unit/protocol/infra-peer-classification.test.ts new file mode 100644 index 00000000..28261a96 --- /dev/null +++ b/backend/tests/unit/protocol/infra-peer-classification.test.ts @@ -0,0 +1,59 @@ +import { describe, it, expect, beforeAll } from 'bun:test'; +import { multiaddr } from '@multiformats/multiaddr'; +import { generateKeyPair } from '@libp2p/crypto/keys'; +import { peerIdFromPrivateKey } from '@libp2p/peer-id'; +import { Network } from '../../../src/protocol/network.ts'; + +/** + * isBootstrapOrRelayPeer decides which peers lishnet leave may hang up. + * Exempt: explicitly configured bootstrap peers and relays our circuit + * connections are routed through. NOT exempt: peer-announce-discovered + * bootstrap entries and NAT'd peers merely reached via a relay. + */ + +let configuredID: string; +let discoveredID: string; +let relayID: string; +let natID: string; + +async function newPeerID(): Promise { + return peerIdFromPrivateKey(await generateKeyPair('Ed25519')).toString(); +} + +/** Bare Network with only the state isBootstrapOrRelayPeer reads. */ +function bareNetwork(conns: Array<{ remoteAddr: ReturnType }>): Network { + const net = Object.create(Network.prototype) as any; + net.configuredBootstrapPeerIDs = new Set([configuredID]); + net.bootstrapPeerIDs = new Set([configuredID, discoveredID]); + net.node = { getConnections: () => conns }; + return net as Network; +} + +describe('Network.isBootstrapOrRelayPeer — leave disconnect exemptions', () => { + beforeAll(async () => { + [configuredID, discoveredID, relayID, natID] = await Promise.all([newPeerID(), newPeerID(), newPeerID(), newPeerID()]); + }); + + it('exempts an explicitly configured bootstrap peer', () => { + expect(bareNetwork([]).isBootstrapOrRelayPeer(configuredID)).toBe(true); + }); + + it('does NOT exempt a peer-announce-discovered bootstrap entry', () => { + expect(bareNetwork([]).isBootstrapOrRelayPeer(discoveredID)).toBe(false); + }); + + it('exempts the relay a circuit connection is routed through', () => { + const circuit = { remoteAddr: multiaddr(`/ip4/198.51.100.7/tcp/4001/p2p/${relayID}/p2p-circuit/p2p/${natID}`) }; + expect(bareNetwork([circuit]).isBootstrapOrRelayPeer(relayID)).toBe(true); + }); + + it('does NOT exempt a NAT peer merely reached via a relay', () => { + const circuit = { remoteAddr: multiaddr(`/ip4/198.51.100.7/tcp/4001/p2p/${relayID}/p2p-circuit/p2p/${natID}`) }; + expect(bareNetwork([circuit]).isBootstrapOrRelayPeer(natID)).toBe(false); + }); + + it('does NOT exempt a plain directly-connected content peer', () => { + const direct = { remoteAddr: multiaddr(`/ip4/198.51.100.8/tcp/4001/p2p/${natID}`) }; + expect(bareNetwork([direct]).isBootstrapOrRelayPeer(natID)).toBe(false); + }); +}); diff --git a/backend/tests/unit/protocol/network-disconnect.test.ts b/backend/tests/unit/protocol/network-disconnect.test.ts new file mode 100644 index 00000000..a5bf81a5 --- /dev/null +++ b/backend/tests/unit/protocol/network-disconnect.test.ts @@ -0,0 +1,238 @@ +import { describe, it, expect } from 'bun:test'; +import { KEEP_ALIVE } from '@libp2p/interface'; +import { multiaddr } from '@multiformats/multiaddr'; +import { Network } from '../../../src/protocol/network.ts'; + +/** + * Unit tests for Network.disconnectPeer tag hygiene: hanging up a peer must + * remove BOTH keep-alive tags — the custom 'keep-alive-fleet' tag and the + * native libp2p KEEP_ALIVE tag. Leaving either behind makes libp2p re-dial + * the peer right after the hangUp, silently undoing the disconnect. + */ + +const PEER_ID = '12D3KooWPvH1oQjQZS8TtucG4NsW2PsnW87jwMAiRLKgrNGS17fo'; +const NET = 'net-a'; + +function makeNetwork() { + const merges: Array<{ tags: Record }> = []; + const hungUp: string[] = []; + const deleted: string[] = []; + const network = Object.create(Network.prototype) as Network; + (network as any).redialSuppressedByNet = new Map>(); + (network as any).bootstrapPeerIDs = new Set(); + (network as any).node = { + getConnections: () => [], + peerStore: { + async merge(_pid: unknown, patch: { tags: Record }): Promise { + merges.push(patch); + }, + async delete(pid: { toString(): string }): Promise { + deleted.push(pid.toString()); + }, + }, + async hangUp(pid: { toString(): string }): Promise { + hungUp.push(pid.toString()); + }, + }; + const suppressed = (pid: string): boolean => (network as any).isRedialSuppressed(pid); + return { network, merges, hungUp, deleted, suppressed }; +} + +describe('Network.disconnectPeer — keep-alive tag removal', () => { + it('clears both keep-alive-fleet and native KEEP_ALIVE tags before hanging up', async () => { + const { network, merges, hungUp } = makeNetwork(); + await network.disconnectPeer(PEER_ID, NET); + expect(merges.length).toBe(1); + const tags = merges[0]!.tags; + expect(Object.keys(tags)).toContain('keep-alive-fleet'); + expect(Object.keys(tags)).toContain(KEEP_ALIVE); + expect(tags['keep-alive-fleet']).toBeUndefined(); + expect(tags[KEEP_ALIVE]).toBeUndefined(); + expect(hungUp).toEqual([PEER_ID]); + }); + + it('still hangs up when tag removal fails', async () => { + const { network, hungUp } = makeNetwork(); + (network as any).node.peerStore.merge = async (): Promise => { + throw new Error('merge failed'); + }; + await network.disconnectPeer(PEER_ID, NET); + expect(hungUp).toEqual([PEER_ID]); + }); + + it('is a no-op for an invalid peer id', async () => { + const { network, merges, hungUp } = makeNetwork(); + await network.disconnectPeer('not-a-peer-id', NET); + expect(merges).toEqual([]); + expect(hungUp).toEqual([]); + }); + + it('suppresses the hung-up peer from redial maintenance', async () => { + const { network, suppressed } = makeNetwork(); + await network.disconnectPeer(PEER_ID, NET); + expect(suppressed(PEER_ID)).toBe(true); + }); + + it('forgets the peerStore entry so the disconnect survives a restart', async () => { + const { network, deleted } = makeNetwork(); + await network.disconnectPeer(PEER_ID, NET); + expect(deleted).toEqual([PEER_ID]); + }); +}); + +/** + * Per-network suppression: rejoining one lishnet must lift only ITS left peers, + * a legitimate reconnect lifts a peer from all lishnets. + */ +describe('Network per-network redial suppression', () => { + function bareNetwork() { + const network = Object.create(Network.prototype) as Network; + (network as any).redialSuppressedByNet = new Map>(); + return network; + } + + it('rejoin of one lishnet lifts only its suppressed peers', () => { + const net = bareNetwork(); + (net as any).addRedialSuppression('net-a', 'pA'); + (net as any).addRedialSuppression('net-b', 'pB'); + expect((net as any).isRedialSuppressed('pA')).toBe(true); + expect((net as any).isRedialSuppressed('pB')).toBe(true); + net.clearRedialSuppressionForNetwork('net-a'); + expect((net as any).isRedialSuppressed('pA')).toBe(false); + expect((net as any).isRedialSuppressed('pB')).toBe(true); // still-left net-b unaffected + }); + + it('observed reconnect lifts the peer from every lishnet', () => { + const net = bareNetwork(); + (net as any).addRedialSuppression('net-a', 'pX'); + (net as any).addRedialSuppression('net-b', 'pX'); + (net as any).clearRedialSuppressionForPeer('pX'); + expect((net as any).isRedialSuppressed('pX')).toBe(false); + }); +}); + +/** + * runRedialMaintenance must not re-dial peers that leave-network just hung up, + * and must drop that suppression the moment the peer is observed connected again. + */ +describe('Network.runRedialMaintenance — leave-peer suppression', () => { + function bareNetwork(suppressed: string[], sharedTopicPeers: string[] = []) { + const dialed: string[] = []; + const network = Object.create(Network.prototype) as Network; + (network as any).redialBackoff = new Map(); + (network as any).redialSuppressedByNet = new Map([['net-x', new Set(suppressed)]]); + // A reconnected peer's suppression is lifted only if it currently shares a joined + // topic — model that via a pubsub whose subscribers list the "back on topic" peers. + (network as any).pubsub = { + getTopics: () => ['lish/net-x'], + getSubscribers: () => sharedTopicPeers.map(p => ({ toString: () => p })), + }; + (network as any).node = { + async dial(id: { toString(): string }): Promise { + dialed.push(id.toString()); + }, + getConnections: () => [], + }; + return { network, dialed }; + } + + const run = (network: Network, connected: any[], all: any[]): Promise => (network as any).runRedialMaintenance(connected, all); + + it('does not re-dial a peer suppressed by leave-network', async () => { + const { network, dialed } = bareNetwork(['pLeft']); + const peer = { id: { toString: () => 'pLeft' }, addresses: [{ multiaddr: multiaddr('/ip4/203.0.113.5/tcp/9090') }] }; + await run(network, [], [peer]); + expect(dialed).toEqual([]); + expect((network as any).isRedialSuppressed('pLeft')).toBe(true); + }); + + it('clears suppression when a reconnected peer is back on a shared topic', async () => { + const { network, dialed } = bareNetwork(['pBack'], ['pBack']); + const peer = { id: { toString: () => 'pBack' } }; + await run(network, [{ toString: () => 'pBack' }], [peer]); + expect(dialed).toEqual([]); + expect((network as any).isRedialSuppressed('pBack')).toBe(false); + }); + + it('keeps suppression for a reconnected peer NOT back on a shared topic', async () => { + // A left peer dialing us back (keep-alive/mDNS) without rejoining a shared topic + // must stay suppressed — otherwise canListSharesTo would serve it our catalog. + const { network } = bareNetwork(['pBack'], []); + const peer = { id: { toString: () => 'pBack' } }; + await run(network, [{ toString: () => 'pBack' }], [peer]); + expect((network as any).isRedialSuppressed('pBack')).toBe(true); + }); +}); + +/** + * Zero-connection recovery dials bootstrapMultiaddrs when the node has no + * connections. It must skip peers leave-network deliberately hung up, or a left + * bootstrap comes straight back the moment connections briefly hit zero. + */ +describe('Network.runZeroConnectionRecovery — leave-peer suppression', () => { + function bareNetwork(suppressed: string[], bootstrapMaStrs: string[]) { + const dialed: string[] = []; + const network = Object.create(Network.prototype) as Network; + (network as any).redialSuppressedByNet = new Map([['net-x', new Set(suppressed)]]); + (network as any).bootstrapMultiaddrs = bootstrapMaStrs.map(s => multiaddr(s)); + (network as any).recentDisconnects = []; + (network as any).bootstrapTracker = { entries: () => [] }; + (network as any).node = { + async dial(ma: { toString(): string }): Promise { + dialed.push(ma.toString()); + }, + }; + return { network, dialed }; + } + + const run = (network: Network, connected: any[]): Promise => (network as any).runZeroConnectionRecovery(connected); + + it('does not dial a bootstrap peer suppressed by leave-network', async () => { + const ma = `/ip4/192.0.2.1/tcp/9090/p2p/${PEER_ID}`; + const { network, dialed } = bareNetwork([PEER_ID], [ma]); + await run(network, []); + expect(dialed).toEqual([]); + }); + + it('still dials a non-suppressed bootstrap peer', async () => { + const ma = `/ip4/192.0.2.1/tcp/9090/p2p/${PEER_ID}`; + const { network, dialed } = bareNetwork([], [ma]); + await run(network, []); + expect(dialed).toEqual([multiaddr(ma).toString()]); + }); +}); + +/** + * Re-configuring a bootstrap peer (network re-join) must lift any redial + * suppression left by a prior leaveNetwork — otherwise maintenance skips it + * forever if the single explicit join-dial fails or drops before the next tick. + */ +describe('Network.addBootstrapPeers — rejoin clears suppression', () => { + function bareNetwork(suppressed: string[]) { + const network = Object.create(Network.prototype) as Network; + (network as any).redialSuppressedByNet = new Map([['net-a', new Set(suppressed)]]); + (network as any).configuredBootstrapPeerIDs = new Set(); + (network as any).bootstrapPeerIDs = new Set(); + (network as any).bootstrapMultiaddrs = []; + (network as any).bootstrapTracker = { markPending() {}, recordOutcome() {} }; + (network as any).node = { + peerId: { toString: () => 'selfID' }, + getConnections: () => [], + async dial(): Promise {}, + peerStore: { async merge(): Promise {} }, + }; + return network; + } + + it('lifts suppression for a re-configured bootstrap peer', async () => { + const network = bareNetwork([PEER_ID]); + await (network as any).addBootstrapPeers([`/ip4/192.0.2.1/tcp/9090/p2p/${PEER_ID}`], 'net-a', 'configured'); + expect((network as any).isRedialSuppressed(PEER_ID)).toBe(false); + }); + + it('does not lift suppression for a discovered (non-configured) re-add', async () => { + const network = bareNetwork([PEER_ID]); + await (network as any).addBootstrapPeers([`/ip4/192.0.2.1/tcp/9090/p2p/${PEER_ID}`], 'net-a', 'discovered'); + expect((network as any).isRedialSuppressed(PEER_ID)).toBe(true); + }); +}); diff --git a/backend/tests/unit/protocol/search-visibility.test.ts b/backend/tests/unit/protocol/search-visibility.test.ts index 9ec6b9ba..8d89bcf1 100644 --- a/backend/tests/unit/protocol/search-visibility.test.ts +++ b/backend/tests/unit/protocol/search-visibility.test.ts @@ -47,7 +47,12 @@ describe('LISH search visibility', () => { // exact substring that would break on every new predicate. const getLishsBlock = LISH_PROTOCOL_TS.slice(LISH_PROTOCOL_TS.indexOf("request.type === 'getLishs'"), LISH_PROTOCOL_TS.indexOf("request.type === 'getLish'")); expect(getLishsBlock).toContain('isUploadAdvertisable(l.id)'); - expect(LISH_PROTOCOL_TS).toContain('if (!isUploadAdvertisable(request.lishID))'); + // getLish guard: still rejects non-advertisable LISHs; the shared-lishnet + // gate sits in front of it within the same condition. + expect(LISH_PROTOCOL_TS).toContain('!isUploadAdvertisable(request.lishID)'); + // The lishnet serve-gate must protect both unicast discovery request types. + expect(getLishsBlock).toContain('sharesNetworkWith'); + expect(LISH_PROTOCOL_TS.slice(LISH_PROTOCOL_TS.indexOf("request.type === 'getLish'"))).toContain('sharesNetworkWith'); }); it('marks queued verification as busy before broadcasting pending-verification', () => { diff --git a/backend/tests/unit/protocol/serve-gate.test.ts b/backend/tests/unit/protocol/serve-gate.test.ts new file mode 100644 index 00000000..580f0474 --- /dev/null +++ b/backend/tests/unit/protocol/serve-gate.test.ts @@ -0,0 +1,75 @@ +import { describe, it, expect } from 'bun:test'; +import { serveGateBlocks } from '../../../src/protocol/lish-protocol.ts'; +import { Network } from '../../../src/protocol/network.ts'; +import { lishTopic } from '../../../src/protocol/constants.ts'; + +/** + * The unicast LISH serve-gate must fail CLOSED: a stream that could not be mapped + * to a peer id (remotePeerID absent) is refused exactly like a not-shared peer, + * never served. Only when no gate is configured (sharesNetworkWith undefined) is + * everything allowed. + */ +describe('serveGateBlocks', () => { + it('does not block when no gate is configured', () => { + expect(serveGateBlocks(undefined, 'peer-a')).toBe(false); + expect(serveGateBlocks(undefined, undefined)).toBe(false); + }); + + it('does not block a peer we share a joined lishnet with', () => { + expect(serveGateBlocks(() => true, 'peer-a')).toBe(false); + }); + + it('blocks a peer we do not share a joined lishnet with', () => { + expect(serveGateBlocks(() => false, 'peer-a')).toBe(true); + }); + + it('blocks (fail closed) when the remote peer id is unknown', () => { + expect(serveGateBlocks(() => true, undefined)).toBe(true); + expect(serveGateBlocks(() => true, '')).toBe(true); + }); +}); + +/** + * canListSharesTo is the softer LISTING gate: allow any peer while we hold a joined + * lishnet topic (so unicast search works before SUBSCRIBE syncs), except peers we + * deliberately left (still redial-suppressed) and except when we hold no lishnet. + */ +describe('Network.canListSharesTo', () => { + function bareNetwork(suppressed: string[], topics: string[], infra: string[] = [], subscribers: string[] = []) { + const network = Object.create(Network.prototype) as Network; + (network as any).redialSuppressedByNet = new Map([['net-x', new Set(suppressed)]]); + (network as any).pubsub = { + getTopics: () => topics, + getSubscribers: () => subscribers.map(p => ({ toString: () => p })), + }; + (network as any).isBootstrapOrRelayPeer = (pid: string) => infra.includes(pid); + return network; + } + + it('allows a fresh peer while we hold a joined lishnet topic (subscribe may lag)', () => { + const net = bareNetwork([], [lishTopic('net-a')]); + expect((net as any).canListSharesTo('peer-a')).toBe(true); + }); + + it('refuses a peer we deliberately left (still suppressed)', () => { + const net = bareNetwork(['peer-left'], [lishTopic('net-a')]); + expect((net as any).canListSharesTo('peer-left')).toBe(false); + }); + + it('refuses when we hold no lishnet topic', () => { + const net = bareNetwork([], []); + expect((net as any).canListSharesTo('peer-a')).toBe(false); + }); + + it('refuses a kept infrastructure peer that no longer shares a joined topic', () => { + // A relay/bootstrap of a left network is kept connected but must not browse our + // shares unless it currently shares a joined topic. + const net = bareNetwork([], [lishTopic('net-a')], ['relay-x'], []); + expect((net as any).canListSharesTo('relay-x')).toBe(false); + }); + + it('allows an infrastructure peer that still shares a joined topic', () => { + const net = bareNetwork([], [lishTopic('net-a')], ['relay-x'], ['relay-x']); + expect((net as any).canListSharesTo('relay-x')).toBe(true); + }); +});