diff options
| -rw-r--r-- | lib/lightning.ts | 28 | ||||
| -rw-r--r-- | lib/ln.ts | 56 |
2 files changed, 74 insertions, 10 deletions
diff --git a/lib/lightning.ts b/lib/lightning.ts index b07db055..609fad41 100644 --- a/lib/lightning.ts +++ b/lib/lightning.ts @@ -26,6 +26,26 @@ let waitStartedAt = Date.now(); // out instead of re-arming, so we never run two listeners concurrently. let listenerEpoch = 0; +// Recover the invoice listener the instant its long-poll socket dies (e.g. a cl +// restart) instead of waiting up to LISTENER_STALL_MS for the stall watchdog. +// waitanyinvoice is a NO_TIMEOUT long-poll, so a dead socket leaves it blocked +// forever with no error — this onDrop signal is the only prompt trigger. Bump the +// epoch so the (possibly forever-blocked) old invocation bows out, mark inactive, +// and re-arm. If cl is still down, ensure()'s backoff + the retry loop below wait +// it out. Debounced so a burst of socket errors recycles once. +let recycleArmed = false; +(lnListen as any).onDrop?.(() => { + if (!listenerActive || recycleArmed) return; + recycleArmed = true; + listenerEpoch++; + listenerActive = false; + warn("lightning listener: listen socket dropped — recycling immediately"); + setTimeout(() => { + recycleArmed = false; + listenForLightning(); + }, 100); +}); + export async function listenForLightning() { if (listenerActive) { warn("lightning listener: already active, skipping duplicate call"); @@ -141,8 +161,16 @@ export async function listenForLightning() { `error=${errorMsg}` ); + // A cl restart makes ensure() throw LightningUnavailableError until the socket + // returns — expected, self-heals, and restarting the app can't fix a down cl. + // So don't count it toward the process-exit escalation; just keep retrying and + // resume within LISTENER_RETRY_DELAY of cl coming back. Only genuine listener + // errors climb toward a container restart. if (e instanceof LightningUnavailableError) { err("lightning listener: RPC socket unavailable"); + warn(`lightning listener: cl unavailable, retrying in ${LISTENER_RETRY_DELAY / 1000}s`); + setTimeout(listenForLightning, LISTENER_RETRY_DELAY); + return; } listenerRetries++; @@ -99,6 +99,32 @@ function lightningProxy(rpcPath: string) { let backoff = 250; const maxBackoff = 5000; + const deathCbs = new Set<(reason?: string) => void>(); + + // Fully tear down a dead client. Just doing `client = null` orphans the old + // LightningClient: the @asoltys/clightning-client library keeps its internal + // reconnect() loop running on the (dead) socket forever — spamming + // `connect ENOENT …/lightning-rpc` and leaking a socket + timer on EVERY cl + // restart. So stop its reconnect loop, destroy the socket, and drop listeners. + // Then notify subscribers (the invoice listener) so they can recycle at once + // rather than waiting on the 5-minute stall watchdog. `notify=false` for a + // deliberate reset() where the caller already owns the re-arm. + function dropClient(reason?: string, notify = true) { + const dead = client; + client = null; + if (dead) { + try { + if (dead.reconnectTimeout) { clearTimeout(dead.reconnectTimeout); dead.reconnectTimeout = null; } + dead.reconnect = () => {}; // neuter any pending 'error'/'end' handler + dead.rl?.close?.(); + dead.client?.removeAllListeners?.(); + dead.client?.destroy?.(); + dead.removeAllListeners?.(); + } catch (_) {} + } + if (notify) for (const cb of deathCbs) { try { cb(reason); } catch (_) {} } + } + function ensure() { if (client) return client; @@ -116,10 +142,13 @@ function lightningProxy(rpcPath: string) { } try { - client = new LightningClient(rpcPath); - // Prevent unhandled 'error' events from crashing the process - client.on("error", (e: any) => { - if (isSocketDied(e)) client = null; + const c = new LightningClient(rpcPath); + client = c; + // Prevent unhandled 'error' events from crashing the process. Guard on + // `c === client` so a stale client's late error can't tear down a newer, + // healthy one that already replaced it. + c.on("error", (e: any) => { + if (c === client && isSocketDied(e)) dropClient(e?.code ?? e?.errno); }); backoff = 250; nextTryAt = 0; @@ -151,15 +180,22 @@ function lightningProxy(rpcPath: string) { if (prop === "reset") { return () => { const had = !!client; - try { - client?.removeAllListeners?.(); - client?.destroy?.(); - } catch (_) {} - client = null; + dropClient("reset", false); // caller (watchdog) owns the re-arm return had; }; } + // Subscribe to socket-death so a consumer (the invoice listener) can + // recover the instant the connection dies, instead of polling/waiting. + // Returns an unsubscribe fn. Handled before the generic RPC dispatch so + // "onDrop" is never sent to cl as a method. + if (prop === "onDrop") { + return (cb: (reason?: string) => void) => { + deathCbs.add(cb); + return () => deathCbs.delete(cb); + }; + } + return async (...args: any[]) => { const c = ensure(); const v = c[prop as any]; @@ -186,7 +222,7 @@ function lightningProxy(rpcPath: string) { } catch (e: any) { // ETIMEDOUT (from RpcTimeoutError) is included in isSocketDied, so a // hung RPC drops the client and the next call reconnects. - if (isSocketDied(e)) client = null; + if (isSocketDied(e)) dropClient(e?.code ?? e?.errno); throw e; } }; |
