diff options
| -rw-r--r-- | scripts/audit-lightning-backlog.ts | 76 | ||||
| -rw-r--r-- | scripts/kvrocks-merge-lowmem.conf | 19 | ||||
| -rw-r--r-- | scripts/merge-kvrocks-archives.ts | 321 |
3 files changed, 416 insertions, 0 deletions
diff --git a/scripts/audit-lightning-backlog.ts b/scripts/audit-lightning-backlog.ts new file mode 100644 index 00000000..4c78e295 --- /dev/null +++ b/scripts/audit-lightning-backlog.ts @@ -0,0 +1,76 @@ +import { db } from "$lib/db"; +import ln from "$lib/ln"; +import { getInvoice, getPayment, getUser } from "$lib/utils"; + +const max = Number(process.argv[2] || 10_000); +const stored = Number((await db.get("pay_index")) || 0); +let cursor = stored; +const rows: any[] = []; + +for (let i = 0; i < max; i++) { + let settled: any; + try { + settled = await ln.waitanyinvoice(cursor, 1); + } catch (e: any) { + if (Number(e?.code ?? e?.errno) === 904) break; + throw e; + } + + if (!settled?.pay_index || Number(settled.pay_index) <= cursor) break; + cursor = Number(settled.pay_index); + + const mintOwned = + typeof settled.label === "string" && settled.label.startsWith("lbl"); + const paymentRequest = settled.bolt11 || settled.bolt12; + const invoiceKey = settled.bolt11 || settled.local_offer_id || settled.bolt12; + const invoice = mintOwned ? null : await getInvoice(invoiceKey); + const payment = paymentRequest ? await getPayment(paymentRequest) : null; + const user = invoice?.uid ? await getUser(invoice.uid) : null; + const claim = settled.payment_preimage + ? await db.get(`credited:${settled.payment_preimage}`) + : null; + + rows.push({ + payIndex: cursor, + amountSats: Math.round(Number(settled.amount_received_msat || 0) / 1000), + mintOwned, + invoiceFound: !!invoice, + paymentFound: !!payment, + claimFound: !!claim, + username: user?.username, + autowithdraw: !!user?.autowithdraw, + }); +} + +const coinos = rows.filter((row) => !row.mintOwned); +const actionable = coinos.filter( + (row) => row.invoiceFound && !row.paymentFound && !row.claimFound, +); + +console.log( + JSON.stringify( + { + storedPayIndex: stored, + latestPayIndex: cursor, + backlog: rows.length, + mintOwned: rows.filter((row) => row.mintOwned).length, + coinos: coinos.length, + alreadyProcessed: coinos.filter((row) => row.paymentFound).length, + missingInvoice: coinos.filter((row) => !row.invoiceFound).length, + claimedWithoutPayment: coinos.filter( + (row) => row.claimFound && !row.paymentFound, + ).length, + actionable: actionable.length, + actionableSats: actionable.reduce( + (total, row) => total + row.amountSats, + 0, + ), + autowithdraw: actionable.filter((row) => row.autowithdraw).length, + rows, + }, + null, + 2, + ), +); + +process.exit(0); diff --git a/scripts/kvrocks-merge-lowmem.conf b/scripts/kvrocks-merge-lowmem.conf new file mode 100644 index 00000000..bdfa641a --- /dev/null +++ b/scripts/kvrocks-merge-lowmem.conf @@ -0,0 +1,19 @@ +bind 0.0.0.0 +port 6379 +workers 2 +timeout 0 +tcp-backlog 512 +daemonize no +dir /data +db-name db +log-dir stdout +redis-cursor-compatible no +max-io-mb 100 +rocksdb.compression snappy +rocksdb.block_cache_size 128 +rocksdb.metadata_block_cache_size 128 +rocksdb.subkey_block_cache_size 128 +rocksdb.write_buffer_size 32 +rocksdb.max_background_compactions 1 +rocksdb.max_background_flushes 1 +rocksdb.max_background_jobs 2 diff --git a/scripts/merge-kvrocks-archives.ts b/scripts/merge-kvrocks-archives.ts new file mode 100644 index 00000000..54ea178d --- /dev/null +++ b/scripts/merge-kvrocks-archives.ts @@ -0,0 +1,321 @@ +#!/usr/bin/env bun + +// Overlay one Redis/Kvrocks archive onto a base archive without materializing +// keys in memory. The source wins on collisions. SCAN is intentionally sent as +// a raw command because Kvrocks cursors can exceed JavaScript's safe integer +// range and node-redis' high-level scan helper rounds them. + +import { createClient, type RedisClientType } from "redis"; + +const valueOption = (name: string) => + process.argv + .find((arg) => arg.startsWith(`--${name}=`)) + ?.slice(name.length + 3); +const hasFlag = (name: string) => process.argv.includes(`--${name}`); +const positiveInteger = (name: string, fallback: number) => { + const parsed = Number(valueOption(name) || fallback); + if (!Number.isSafeInteger(parsed) || parsed <= 0) { + throw new Error(`--${name} must be a positive integer`); + } + return parsed; +}; + +const sourceUrl = valueOption("source-url"); +const baseUrl = valueOption("base-url"); +const targetUrl = valueOption("target-url"); +const verifyUrl = valueOption("verify-url"); +const apply = hasFlag("apply"); +const skipEqual = hasFlag("skip-equal"); +const countTypes = hasFlag("count-types"); +const batchSize = positiveInteger("batch", 512); +const concurrency = positiveInteger("concurrency", 32); +const progressEvery = positiveInteger("progress", 100_000); + +if (!sourceUrl || (apply && !targetUrl)) { + console.error( + "usage: bun scripts/merge-kvrocks-archives.ts " + + "--source-url=redis://127.0.0.1:6402 " + + "[--base-url=redis://127.0.0.1:6401] " + + "[--target-url=redis://127.0.0.1:6403 --apply] " + + "[--batch=512] [--concurrency=32]", + ); + process.exit(2); +} +if (apply && (sourceUrl === targetUrl || baseUrl === targetUrl)) { + throw new Error("The target must be distinct from the source and base"); +} + +const makeClient = (url: string) => + createClient({ + url, + socket: { reconnectStrategy: () => false }, + }); +const source = makeClient(sourceUrl); +const base = baseUrl ? makeClient(baseUrl) : null; +const target = targetUrl ? makeClient(targetUrl) : null; +const verifier = verifyUrl ? makeClient(verifyUrl) : null; +const clients = [source, base, target, verifier].filter( + Boolean, +) as RedisClientType[]; + +const concurrentMap = async <T>( + items: T[], + limit: number, + fn: (item: T) => Promise<void>, +) => { + let next = 0; + const workers = Array.from( + { length: Math.min(limit, items.length) }, + async () => { + while (next < items.length) { + const index = next++; + await fn(items[index]); + } + }, + ); + await Promise.all(workers); +}; + +const valuesMatch = async ( + left: RedisClientType, + right: RedisClientType, + key: string, + type: string, +) => { + if ((await right.type(key)) !== type) return false; + if (type === "string") { + const [leftValue, rightValue] = await Promise.all([ + left.sendCommand<Buffer | null>(["GET", key], { returnBuffers: true }), + right.sendCommand<Buffer | null>(["GET", key], { returnBuffers: true }), + ]); + return ( + leftValue !== null && + rightValue !== null && + Buffer.from(leftValue).equals(Buffer.from(rightValue)) + ); + } + if (type === "list") { + const [leftValues, rightValues] = await Promise.all([ + left.sendCommand<Buffer[]>(["LRANGE", key, "0", "-1"], { + returnBuffers: true, + }), + right.sendCommand<Buffer[]>(["LRANGE", key, "0", "-1"], { + returnBuffers: true, + }), + ]); + return ( + leftValues.length === rightValues.length && + leftValues.every((value, index) => + Buffer.from(value).equals(Buffer.from(rightValues[index])), + ) + ); + } + return false; +}; + +let scanned = 0; +let overlaps = 0; +let restored = 0; +let skippedEqual = 0; +let expiredDuringRead = 0; +const typeCounts: Record<string, number> = {}; +let verified = 0; +const verificationMismatches: string[] = []; +let cursor = "0"; +let nextProgress = progressEvery; +const startedAt = Date.now(); + +try { + await Promise.all(clients.map((client) => client.connect())); + + do { + const reply = await source.sendCommand<[string, string[]]>([ + "SCAN", + cursor, + "COUNT", + String(batchSize), + ]); + cursor = String(reply[0]); + const keys = reply[1]; + scanned += keys.length; + + if (countTypes) { + await concurrentMap(keys, concurrency, async (key) => { + const type = await source.type(key); + typeCounts[type] = (typeCounts[type] || 0) + 1; + }); + } + + if (base) { + await concurrentMap(keys, concurrency, async (key) => { + if (await base.exists(key)) overlaps++; + }); + } + + if (verifier) { + await concurrentMap(keys, concurrency, async (key) => { + const [sourceType, verifyType] = await Promise.all([ + source.type(key), + verifier.type(key), + ]); + if (sourceType !== verifyType) { + if (verificationMismatches.length < 20) { + verificationMismatches.push( + `${key}: type ${sourceType} != ${verifyType}`, + ); + } + return; + } + + if (sourceType === "string") { + const [sourceValue, verifyValue] = await Promise.all([ + source.sendCommand<Buffer | null>(["GET", key], { + returnBuffers: true, + }), + verifier.sendCommand<Buffer | null>(["GET", key], { + returnBuffers: true, + }), + ]); + if ( + sourceValue === null || + verifyValue === null || + !Buffer.from(sourceValue).equals(Buffer.from(verifyValue)) + ) { + if (verificationMismatches.length < 20) { + verificationMismatches.push(`${key}: string value differs`); + } + return; + } + } else if (sourceType === "list") { + const [sourceValues, verifyValues] = await Promise.all([ + source.sendCommand<Buffer[]>(["LRANGE", key, "0", "-1"], { + returnBuffers: true, + }), + verifier.sendCommand<Buffer[]>(["LRANGE", key, "0", "-1"], { + returnBuffers: true, + }), + ]); + if ( + sourceValues.length !== verifyValues.length || + sourceValues.some( + (value, index) => + !Buffer.from(value).equals(Buffer.from(verifyValues[index])), + ) + ) { + if (verificationMismatches.length < 20) { + verificationMismatches.push(`${key}: list value differs`); + } + return; + } + } else { + if (verificationMismatches.length < 20) { + verificationMismatches.push( + `${key}: verification unsupported for ${sourceType}`, + ); + } + return; + } + verified++; + }); + } + + if (apply && target) { + await concurrentMap(keys, concurrency, async (key) => { + const [type, ttlRaw] = await Promise.all([ + source.type(key), + source.sendCommand<string>(["PTTL", key]), + ]); + const ttl = Number(ttlRaw); + if (type === "none" || ttl === -2) { + expiredDuringRead++; + return; + } + + if (skipEqual && (await valuesMatch(source, target, key, type))) { + const targetTtl = await target.pTTL(key); + const ttlMatches = + (ttl === -1 && targetTtl === -1) || + (ttl > 0 && targetTtl > 0 && Math.abs(ttl - targetTtl) < 10_000); + if (ttlMatches) { + skippedEqual++; + return; + } + } + + if (type === "string") { + const value = await source.sendCommand<Buffer | null>(["GET", key], { + returnBuffers: true, + }); + if (value === null) { + expiredDuringRead++; + return; + } + await target.sendCommand(["SET", key, value]); + } else if (type === "list") { + await target.del(key); + let offset = 0; + while (true) { + const values = await source.sendCommand<Buffer[]>( + ["LRANGE", key, String(offset), String(offset + batchSize - 1)], + { returnBuffers: true }, + ); + if (!values.length) break; + await target.sendCommand(["RPUSH", key, ...values]); + offset += values.length; + if (values.length < batchSize) break; + } + } else { + throw new Error(`Unsupported source type ${type} for key ${key}`); + } + + if (ttl > 0) await target.pExpire(key, ttl); + restored++; + }); + } + + if (scanned >= nextProgress) { + console.error( + JSON.stringify({ + scanned, + overlaps, + restored, + skippedEqual, + cursor, + elapsedSeconds: Math.round((Date.now() - startedAt) / 1000), + }), + ); + nextProgress += progressEvery; + } + } while (cursor !== "0"); + + console.log( + JSON.stringify( + { + mode: apply ? "apply" : "audit", + sourceUrl, + baseUrl: baseUrl || null, + targetUrl: targetUrl || null, + sourceKeysScanned: scanned, + overlappingBaseKeys: base ? overlaps : null, + sourceOnlyKeys: base ? scanned - overlaps : null, + restored: apply ? restored : null, + skippedEqual: apply ? skippedEqual : null, + expiredDuringRead, + typeCounts: countTypes ? typeCounts : null, + verifyUrl: verifyUrl || null, + verified: verifier ? verified : null, + verificationMismatches: verifier ? verificationMismatches : null, + elapsedSeconds: Math.round((Date.now() - startedAt) / 1000), + }, + null, + 2, + ), + ); + if (verificationMismatches.length) process.exitCode = 1; +} finally { + await Promise.all( + clients.map(async (client) => { + if (client.isOpen) await client.quit(); + }), + ); +} |