diff options
| -rw-r--r-- | index.ts | 3 | ||||
| -rw-r--r-- | lib/health.ts | 147 | ||||
| -rw-r--r-- | lib/invoices.ts | 11 | ||||
| -rw-r--r-- | lib/lightning.ts | 145 | ||||
| -rw-r--r-- | lib/ln.ts | 6 | ||||
| -rw-r--r-- | lib/payments.ts | 59 | ||||
| -rw-r--r-- | routes/info.ts | 7 | ||||
| -rw-r--r-- | routes/users.ts | 51 |
8 files changed, 272 insertions, 157 deletions
@@ -2,6 +2,7 @@ import app from "$lib/app"; import { admin, auth, optional } from "$lib/auth"; import { fixBolt12, listenForLightning, replay } from "$lib/lightning"; +import { startHealthCheck } from "$lib/health"; import { getLocations } from "$lib/locations"; import nwc from "$lib/nwc"; import { catchUp, check } from "$lib/payments"; @@ -28,6 +29,7 @@ try { catchUp(); nwc(); check(); + startHealthCheck(); } catch (e) { console.log(e); } @@ -36,6 +38,7 @@ setTimeout(listenForLightning, 2000); setInterval(sendHeartbeat, 2000); app.get("/balances", info.balances); +app.get("/health", info.health); app.post("/email", email.send); app.get("/fx", rates.fx); diff --git a/lib/health.ts b/lib/health.ts new file mode 100644 index 00000000..c2e268c6 --- /dev/null +++ b/lib/health.ts @@ -0,0 +1,147 @@ +import ln, { LightningUnavailableError } from "$lib/ln"; +import { err, l, warn } from "$lib/logging"; + +const HEALTH_CHECK_INTERVAL = 2 * 60 * 1000; // 2 minutes +const RPC_TIMEOUT = 30_000; // 30 seconds - fail fast if RPC hangs +const MAX_CONSECUTIVE_FAILURES = 3; + +let consecutiveFailures = 0; +let lastSuccessTime = Date.now(); +let lastFailureReason = ""; +let checkTimer: ReturnType<typeof setTimeout> | null = null; + +export function getHealthStatus() { + return { + consecutiveFailures, + lastSuccessTime, + lastFailureReason, + healthy: consecutiveFailures < MAX_CONSECUTIVE_FAILURES, + stalledCheck: checkTimer === null && consecutiveFailures > 0, + }; +} + +function withTimeout<T>(promise: Promise<T>, ms: number, label: string): Promise<T> { + return new Promise((resolve, reject) => { + const timer = setTimeout( + () => reject(new Error(`${label} timed out after ${ms}ms`)), + ms, + ); + promise.then( + (v) => { clearTimeout(timer); resolve(v); }, + (e) => { clearTimeout(timer); reject(e); }, + ); + }); +} + +async function checkLightningHealth(): Promise<boolean> { + const startTime = Date.now(); + + try { + l("health check: testing lightning connection..."); + const info = await withTimeout(ln.getinfo(), RPC_TIMEOUT, "getinfo"); + + if (!info?.id) { + throw new Error("getinfo returned invalid response (no node id)"); + } + + const testLabel = `health_check_${Date.now()}`; + const invoice = await withTimeout( + ln.invoice({ + amount_msat: "1000msat", + label: testLabel, + description: "health check", + expiry: 60, + }), + RPC_TIMEOUT, + "invoice", + ); + + if (!invoice?.bolt11) { + throw new Error("invoice creation returned invalid response"); + } + + try { + await withTimeout( + ln.delinvoice({ label: testLabel, status: "unpaid" }), + RPC_TIMEOUT, + "delinvoice", + ); + } catch (e) { + // Ignore cleanup errors + } + + const duration = Date.now() - startTime; + l(`health check: passed in ${duration}ms, node: ${info.id.slice(0, 16)}...`); + + return true; + } catch (e: any) { + const duration = Date.now() - startTime; + const errorCode = e?.code ?? e?.errno ?? "unknown"; + const errorMsg = e?.message ?? String(e); + + err( + `health check: FAILED after ${duration}ms`, + `code=${errorCode}`, + `error=${errorMsg}`, + ); + + if (e instanceof LightningUnavailableError) { + err("health check: lightning RPC socket unavailable"); + } + + lastFailureReason = `${errorCode}: ${errorMsg}`; + return false; + } +} + +export async function runHealthCheck() { + checkTimer = null; + + try { + const healthy = await checkLightningHealth(); + + if (healthy) { + if (consecutiveFailures > 0) { + l( + `health check: recovered after ${consecutiveFailures} consecutive failures`, + ); + } + consecutiveFailures = 0; + lastSuccessTime = Date.now(); + lastFailureReason = ""; + } else { + consecutiveFailures++; + warn( + `health check: failure ${consecutiveFailures}/${MAX_CONSECUTIVE_FAILURES}`, + ); + + if (consecutiveFailures >= MAX_CONSECUTIVE_FAILURES) { + err( + `health check: ${MAX_CONSECUTIVE_FAILURES} consecutive failures, ` + + `last success was ${Math.round((Date.now() - lastSuccessTime) / 1000)}s ago`, + ); + err(`health check: last failure reason: ${lastFailureReason}`); + err("health check: exiting process to trigger container restart"); + + setTimeout(() => { + process.exit(1); + }, 1000); + return; + } + } + } catch (e: any) { + err("health check: unexpected error during health check", e?.message); + consecutiveFailures++; + } + + // Schedule next check + checkTimer = setTimeout(runHealthCheck, HEALTH_CHECK_INTERVAL); +} + +export function startHealthCheck() { + l( + `health check: starting (interval=${HEALTH_CHECK_INTERVAL / 1000}s, timeout=${RPC_TIMEOUT / 1000}s, max_failures=${MAX_CONSECUTIVE_FAILURES})`, + ); + // Run first check after a delay to let the app initialize + checkTimer = setTimeout(runHealthCheck, 30000); +} diff --git a/lib/invoices.ts b/lib/invoices.ts index e2575723..ad0b604e 100644 --- a/lib/invoices.ts +++ b/lib/invoices.ts @@ -61,17 +61,9 @@ export const generate = async ({ invoice, user }) => { let hash; let text; - let path; let paymentHash; - if (account.seed) { - type = "bitcoin"; - const node = rpc({ ...config[type], wallet: aid }); - hash = await node.getNewAddress({ address_type }); - text = bip21(hash, invoice); - - ({ hdkeypath: path } = await node.getAddressInfo(hash)); - } else if (type === PaymentType.lightning) { + if (type === PaymentType.lightning) { let r; if (bolt11) { const { id: nodeid } = await ln.getinfo(); @@ -151,7 +143,6 @@ export const generate = async ({ invoice, user }) => { request_id, memoPrompt, own, - path, prompt, secret, text, diff --git a/lib/lightning.ts b/lib/lightning.ts index f8d0f6a4..1f3b1665 100644 --- a/lib/lightning.ts +++ b/lib/lightning.ts @@ -1,66 +1,127 @@ import { db, g, s } from "$lib/db"; -import ln from "$lib/ln"; -import { err, warn } from "$lib/logging"; +import ln, { lnListen, LightningUnavailableError } from "$lib/ln"; +import { err, l, warn } from "$lib/logging"; import { handleZap } from "$lib/nostr"; import { credit } from "$lib/payments"; import { PaymentType } from "$lib/types"; import { getInvoice, getPayment, getUser } from "$lib/utils"; -export async function listenForLightning() { - const inv = await ln.waitanyinvoice((await g("pay_index")) || 0); - const { - local_offer_id, - bolt11, - bolt12, - description, - pay_index, - payment_hash, - amount_received_msat, - payment_preimage: preimage, - } = inv; +const LISTENER_RETRY_DELAY = 5000; // 5 seconds +const MAX_LISTENER_RETRIES = 10; +let listenerRetries = 0; +let listenerActive = false; - await s("pay_index", pay_index); - setTimeout(listenForLightning); +export async function listenForLightning() { + if (listenerActive) { + warn("lightning listener: already active, skipping duplicate call"); + return; + } - const received = Math.round(amount_received_msat / 1000); + listenerActive = true; try { - if (!preimage) return; + const payIndex = (await g("pay_index")) || 0; + l(`lightning listener: waiting for invoice (pay_index=${payIndex})`); - const invoice = await getInvoice(bolt11 ?? local_offer_id ?? bolt12); - if (!invoice) return warn("received lightning with no invoice", bolt11); + const inv = await lnListen.waitanyinvoice(payIndex); + const { + local_offer_id, + bolt11, + bolt12, + description, + pay_index, + payment_hash, + amount_received_msat, + payment_preimage: preimage, + } = inv; - const p = await getPayment(bolt11 || bolt12); - if (p) return warn("already processed", bolt11 || bolt12); + await s("pay_index", pay_index); - if (invoice?.memo) { - try { - if (JSON.parse(description).kind === 9734) { - const { pubkey } = await getUser(invoice.uid); - handleZap(inv, pubkey); + // Reset retry counter on successful receive + if (listenerRetries > 0) { + l(`lightning listener: recovered after ${listenerRetries} retries`); + } + listenerRetries = 0; + listenerActive = false; + + // Schedule next listen + setTimeout(listenForLightning); + + const received = Math.round(amount_received_msat / 1000); + + try { + if (!preimage) return; + + const invoice = await getInvoice(bolt11 ?? local_offer_id ?? bolt12); + if (!invoice) return warn("received lightning with no invoice", bolt11); + + const p = await getPayment(bolt11 || bolt12); + if (p) return warn("already processed", bolt11 || bolt12); + + if (invoice?.memo) { + try { + if (JSON.parse(description).kind === 9734) { + const { pubkey } = await getUser(invoice.uid); + handleZap(inv, pubkey); + } + } catch (e) { + if (!e.message.includes("Unexpected")) + warn("failed to handle zap", e.message); } - } catch (e) { - if (!e.message.includes("Unexpected")) - warn("failed to handle zap", e.message); } + + await credit({ + hash: bolt11 || bolt12, + amount: received, + memo: invoice.memo, + ref: preimage, + type: bolt12 ? PaymentType.bolt12 : PaymentType.lightning, + payment_hash, + }); + } catch (e) { + console.log(e); + err("problem receiving lightning payment", e.message); } + } catch (e: any) { + listenerActive = false; + const errorCode = e?.code ?? e?.errno ?? "unknown"; + const errorMsg = e?.message ?? String(e); - await credit({ - hash: bolt11 || bolt12, - amount: received, - memo: invoice.memo, - ref: preimage, - type: bolt12 ? PaymentType.bolt12 : PaymentType.lightning, - payment_hash, - }); - } catch (e) { - console.log(e); - err("problem receiving lightning payment", e.message); + err( + `lightning listener: error waiting for invoice`, + `code=${errorCode}`, + `error=${errorMsg}` + ); + + if (e instanceof LightningUnavailableError) { + err("lightning listener: RPC socket unavailable"); + } + + listenerRetries++; + + if (listenerRetries >= MAX_LISTENER_RETRIES) { + err( + `lightning listener: ${MAX_LISTENER_RETRIES} consecutive failures, ` + + `last error: ${errorMsg}` + ); + err("lightning listener: exiting process to trigger container restart"); + + setTimeout(() => { + process.exit(1); + }, 1000); + return; + } + + warn( + `lightning listener: retry ${listenerRetries}/${MAX_LISTENER_RETRIES} ` + + `in ${LISTENER_RETRY_DELAY / 1000}s` + ); + setTimeout(listenForLightning, LISTENER_RETRY_DELAY); } } export async function replay(index) { - const inv = await ln.waitanyinvoice(index - 1); + const inv = await lnListen.waitanyinvoice(index - 1); const { local_offer_id, bolt11, @@ -79,9 +79,13 @@ function lightningProxy(rpcPath: string) { ); } -// exports preserved +// Main proxy for short-lived RPC calls (getinfo, invoice, decode, etc.) const ln = lightningProxy(config.lightning); export default ln; +// Dedicated proxy for long-polling (waitanyinvoice) — uses its own socket +// so it doesn't block the main connection +export const lnListen = lightningProxy(config.lightning); + export const lnb = lightningProxy(config.lightningb); export { LightningUnavailableError }; diff --git a/lib/payments.ts b/lib/payments.ts index 80aee8b7..b0140f52 100644 --- a/lib/payments.ts +++ b/lib/payments.ts @@ -772,65 +772,6 @@ export const catchUp = async () => { setTimeout(catchUp, 10000); }; -export const reconcile = async (account, initial = false) => { - try { - const { descriptors, id, uid, type } = account; - const user = await getUser(uid); - const node = rpc({ ...config[type], wallet: id }); - - let total; - - if (initial) { - const progress = await node.scanTxOutSet("status"); - if (progress) return setTimeout(() => reconcile(account, initial), 1000); - - const { total_amount } = await node.scanTxOutSet("start", descriptors); - total = Math.round(total_amount * SATS); - } else { - total = Math.round((await node.getBalance({ minconf: 1 })) * SATS); - } - - const { balanceAdjustment: memo } = t(user); - - const balance = await g(`balance:${id}`); - - const amount = Math.abs(total - balance); - const hash = v4(); - - if (total > balance) { - const inv = { - memo, - type: PaymentType.reconcile, - hash, - amount, - uid, - aid: id, - }; - await s(`invoice:${hash}`, inv); - await credit({ - hash, - amount, - type: PaymentType.reconcile, - aid: id, - }); - } else if (total < balance) { - await debit({ - aid: id, - amount, - hash: v4(), - memo, - user, - type: PaymentType.reconcile, - }); - } - } catch (e) { - console.log(e); - warn("problem reconciling", e.message, account); - if (e.message.includes("progress")) - return setTimeout(() => reconcile(account, initial), 1000); - } -}; - export const check = async () => { if (process.env.URL.includes("dev")) return; try { diff --git a/routes/info.ts b/routes/info.ts index c6d03e4d..a0ac7693 100644 --- a/routes/info.ts +++ b/routes/info.ts @@ -1,9 +1,16 @@ import { archive } from "$lib/db"; import { db, g } from "$lib/db"; +import { getHealthStatus } from "$lib/health"; import ln from "$lib/ln"; import { getDecodedToken } from "@cashu/cashu-ts"; export default { + async health(_, res) { + const status = getHealthStatus(); + const httpStatus = status.healthy ? 200 : 503; + res.status(httpStatus).send(status); + }, + async balances(_, res) { let total = 0; diff --git a/routes/users.ts b/routes/users.ts index 4f1cb487..64659a01 100644 --- a/routes/users.ts +++ b/routes/users.ts @@ -1,18 +1,17 @@ import { createReadStream } from "node:fs"; -import { appendFile, unlink, writeFile } from "node:fs/promises"; +import { unlink, writeFile } from "node:fs/promises"; import config from "$config"; import { requirePin } from "$lib/auth"; import { db, g, ga, gf, s } from "$lib/db"; import { err, l, warn } from "$lib/logging"; import { mail, templates } from "$lib/mail"; import { getNostrUser, getProfile, serverPubkey2 } from "$lib/nostr"; -import { reconcile } from "$lib/payments"; import register from "$lib/register"; import { emit } from "$lib/sockets"; import upload from "$lib/upload"; import { bail, fail, fields, getUser, pick } from "$lib/utils"; import whitelist from "$lib/whitelist"; -import rpc from "@coinos/rpc"; + import { $ } from "bun"; import got from "got"; import jwt from "jsonwebtoken"; @@ -844,38 +843,11 @@ export default { async createAccount(req, res) { try { - const { fingerprint, pubkey, name, seed, type } = req.body; + const { name, type } = req.body; const { user } = req; - const { id: uid } = user; const id = v4(); - const account = { id, name, seed, type, uid, descriptors: [] }; - - let node = rpc(config[type]); - - await node.createWallet({ - wallet_name: id, - descriptors: true, - disable_private_keys: true, - load_on_startup: true, - }); - - node = rpc({ ...config[type], wallet: id }); - - for (const i of [0, 1]) { - const desc = `wpkh([${fingerprint}]${pubkey}/${i}/*)`; - const { checksum } = await node.getDescriptorInfo(desc); - account.descriptors.push({ - desc: `${desc}#${checksum}`, - range: [0, 100], - next_index: 0, - timestamp: "now", - internal: i === 1, - active: true, - }); - } - - await node.importDescriptors(account.descriptors); + const account = { id, name, type, uid: user.id }; await db .multi() @@ -885,10 +857,6 @@ export default { .lPush(`${user.id}:accounts`, id) .exec(); - reconcile(account, true); - - await appendFile("/bitcoin.conf", `wallet=${id}\n`); - res.send(account); } catch (e) { bail(res, e.message); @@ -914,17 +882,10 @@ export default { try { const { id } = req.body; const { id: uid } = req.user; - const { type } = await g(`account:${id}`); + const account = await g(`account:${id}`); const pos = await db.lPos(`${uid}:accounts`, id); - if (!(type && pos != null)) fail("account not found"); - - try { - const node = rpc({ ...config[type], wallet: id }); - await node.unloadWallet(id); - } catch (e) { - warn("failed to unload wallet", id); - } + if (!(account && pos != null)) fail("account not found"); await db .multi() |