diff --git a/src/handlers/container.js b/src/handlers/container.js index ed71a286..c1ef1165 100644 --- a/src/handlers/container.js +++ b/src/handlers/container.js @@ -1,5 +1,5 @@ import * as storage from '../storage/filesystem.js'; -import { initializeQuota, checkQuota, updateQuotaUsage } from '../storage/quota.js'; +import { initializeQuota, reserveQuota, updateQuotaUsage } from '../storage/quota.js'; import { getAllHeaders } from '../ldp/headers.js'; import { isContainer, getEffectiveUrlPath, getPodName } from '../utils/url.js'; import { generateProfile, generatePreferences, generateTypeIndex, serialize } from '../webid/profile.js'; @@ -159,8 +159,10 @@ export async function handlePost(request, reply) { // Check storage quota before writing (skip in public mode - no pod structure) const podName = request.config?.public ? null : getPodName(request); + // Atomically reserve before writing so concurrent creates can't both pass + // the check and overshoot the limit; release the reservation on failure. if (podName) { - const { allowed, error } = await checkQuota(podName, content.length, request.defaultQuota || 0); + const { allowed, error } = await reserveQuota(podName, content.length, request.defaultQuota || 0); if (!allowed) { return reply.code(507).send({ error: 'Insufficient Storage', message: error }); } @@ -168,9 +170,9 @@ export async function handlePost(request, reply) { success = await storage.write(newStoragePath, content); - // Update quota usage after successful write - if (success && podName) { - await updateQuotaUsage(podName, content.length); + // reserveQuota already recorded the usage; release it if the write failed. + if (!success && podName) { + await updateQuotaUsage(podName, -content.length); } } diff --git a/src/handlers/pay.js b/src/handlers/pay.js index 20a80220..943cff1a 100644 --- a/src/handlers/pay.js +++ b/src/handlers/pay.js @@ -27,7 +27,7 @@ import crypto from 'crypto'; import { getNostrPubkey, pubkeyToDidNostr } from '../auth/nostr.js'; -import { readLedger, writeLedger, getBalance, credit, debit } from '../webledger.js'; +import { readLedger, writeLedger, getBalance, credit, creditOnce, debit } from '../webledger.js'; import { verifyMrc20Deposit, verifyMrc20Anchor, jcs, btAddress } from '../mrc20.js'; import { loadTrail, transferToken, buildTransaction, broadcastTx, p2trScript, btDeriveChainedPrivkey } from '../token.js'; import { secp256k1 } from '@noble/curves/secp256k1'; @@ -319,6 +319,7 @@ export function createPayHandler(options = {}) { const utxos = await loadUtxos(); const ledger = await readLedger(); let credited = 0; + let appended = false; for (const chainId of payChains) { const chain = CHAIN_REGISTRY[chainId]; @@ -341,14 +342,29 @@ export function createPayHandler(options = {}) { } } catch { /* best effort */ } const currency = chain.unit; - credit(ledger, didUri, u.value, currency); + // Idempotent credit keyed on the outpoint: the balance and the + // "already counted" marker commit together in the ledger, so a + // crash before saveUtxos below can't double-credit on the next + // balance poll (the scanner re-runs on every GET /pay/.balance). + const depositKey = `${chainId}:${u.txid}:${u.vout}`; + const { credited: didCredit } = creditOnce(ledger, depositKey, didUri, u.value, currency); utxos.push({ txid: u.txid, vout: u.vout, amount: u.value, scriptpubkey, chain: chainId, tweak: didUri, spent: false }); - credited += u.value; + appended = true; + if (didCredit) credited += u.value; } } + // Ledger first: a crash before saveUtxos leaves the credit recorded + // together with its idempotency key, so the rescan is a no-op rather + // than a double-credit. The reverse order would under-credit. if (credited > 0) { await writeLedger(ledger); + } + // Keyed on `appended`, not `credited`: a replayed outpoint credits + // nothing (didCredit === false) but must still land in the cache. + // Otherwise the crash window never heals — the scanner re-queries the + // explorer for that outpoint on every single balance poll. + if (appended) { await saveUtxos(utxos); } } catch { /* scan failure is non-fatal */ } diff --git a/src/handlers/resource.js b/src/handlers/resource.js index 1fdcc216..e487c266 100644 --- a/src/handlers/resource.js +++ b/src/handlers/resource.js @@ -1,5 +1,5 @@ import * as storage from '../storage/filesystem.js'; -import { checkQuota, updateQuotaUsage } from '../storage/quota.js'; +import { reserveQuota, updateQuotaUsage } from '../storage/quota.js'; import { getAllHeaders, getNotFoundHeaders } from '../ldp/headers.js'; import { generateContainerJsonLd, serializeJsonLd } from '../ldp/container.js'; import { isContainer, getContentType, isRdfContentType, getEffectiveUrlPath, safeJsonParse, getPodName } from '../utils/url.js'; @@ -1045,8 +1045,11 @@ export async function handlePut(request, reply) { const oldSize = stats?.size || 0; const sizeDelta = content.length - oldSize; + // Atomically reserve the growth before writing so concurrent writers can't + // both pass the check and overshoot the limit. reserveQuota commits the + // reservation, so we must release it if the write then fails. if (podName && sizeDelta > 0) { - const { allowed, error } = await checkQuota(podName, sizeDelta, request.defaultQuota || 0); + const { allowed, error } = await reserveQuota(podName, sizeDelta, request.defaultQuota || 0); if (!allowed) { return reply.code(507).send({ error: 'Insufficient Storage', message: error }); } @@ -1054,11 +1057,14 @@ export async function handlePut(request, reply) { const success = await storage.write(storagePath, content); if (!success) { + if (podName && sizeDelta > 0) { + await updateQuotaUsage(podName, -sizeDelta); // release the reservation + } return reply.code(500).send({ error: 'Write failed' }); } - // Update quota usage after successful write - if (podName && sizeDelta !== 0) { + // Growth was already recorded by reserveQuota; only a shrink needs recording. + if (podName && sizeDelta < 0) { await updateQuotaUsage(podName, sizeDelta); } diff --git a/src/storage/quota.js b/src/storage/quota.js index 0b3e89a8..62798581 100644 --- a/src/storage/quota.js +++ b/src/storage/quota.js @@ -13,6 +13,35 @@ const QUOTA_FILE = '.quota.json'; // between writeFile and rename) doesn't inflate pod usage. const QUOTA_TMP_PREFIX = `${QUOTA_FILE}.tmp.`; +/** + * Per-pod serialization of read-modify-write on .quota.json. + * + * saveQuota is already atomic (temp file + rename), which stops a concurrent + * reader from seeing a truncated file (#309). But atomicity of a single write + * does not stop a lost update: two writers that both `loadQuota` the same + * `used` value, each add their bytes and `saveQuota`, race to clobber one + * another — and, worse, both can pass a `checkQuota` before either records its + * usage, letting concurrent writes overshoot the limit. Serializing the whole + * load→mutate→save critical section per pod closes both windows. This mirrors + * QuotaPolicy::reserve in the solid-pod-rs parity port, which holds a per-pod + * lock for atomic check-and-commit. + */ +const podLocks = new Map(); + +function withPodLock(podName, fn) { + const prev = podLocks.get(podName) || Promise.resolve(); + // Chain onto the previous holder regardless of how it settled. + const run = prev.then(fn, fn); + // Store a tail that never rejects so the chain can't wedge on an error. + const tail = run.then(() => {}, () => {}); + podLocks.set(podName, tail); + // Opportunistic cleanup so the map doesn't grow unbounded across many pods. + tail.then(() => { + if (podLocks.get(podName) === tail) podLocks.delete(podName); + }); + return run; +} + /** * Get quota file path for a pod */ @@ -168,19 +197,75 @@ export async function checkQuota(podName, additionalBytes, defaultQuota) { } /** - * Update quota usage after a write + * Atomically reserve quota for a pending write: check-and-commit under a + * per-pod lock so concurrent writers cannot both pass the check and then + * overshoot the limit. On success the reserved bytes are already recorded in + * `used`, so callers must NOT also call updateQuotaUsage for the same bytes — + * instead, release the reservation with `updateQuotaUsage(pod, -bytes)` if the + * write subsequently fails. + * + * This is the authoritative enforcement point and the only one the write path + * uses. checkQuota remains exported as a lock-free read-only probe for callers + * that want to inspect headroom without reserving it, but it performs no + * pre-check for PUT/POST any more — do not go looking for one. Mirrors + * QuotaPolicy::reserve in the solid-pod-rs parity port. + * + * @param {string} podName - The pod name + * @param {number} additionalBytes - Bytes to reserve (expected >= 0) + * @param {number} defaultQuota - Default quota limit + * @returns {Promise<{allowed: boolean, quota: object, error?: string}>} + */ +export async function reserveQuota(podName, additionalBytes, defaultQuota) { + return withPodLock(podName, async () => { + let quota = await loadQuota(podName); + + // Initialize limit from the default on first use, preserving any usage + // reconciled from a recovered corrupt/empty file (see checkQuota). + if (quota.limit === 0 && defaultQuota > 0) { + quota = { limit: defaultQuota, used: quota.used }; + } + + // No enforcement (and no tracking) when no limit is in effect — matches + // updateQuotaUsage, which skips uninitialized quotas. + if (quota.limit === 0) { + return { allowed: true, quota }; + } + + const projectedUsage = quota.used + additionalBytes; + + if (additionalBytes > 0 && projectedUsage > quota.limit) { + const usedMB = (quota.used / (1024 * 1024)).toFixed(2); + const limitMB = (quota.limit / (1024 * 1024)).toFixed(2); + return { + allowed: false, + quota, + error: `Storage quota exceeded. Used: ${usedMB}MB / ${limitMB}MB` + }; + } + + // Commit the reservation as part of the same locked critical section. + quota.used = Math.max(0, projectedUsage); + await saveQuota(podName, quota); + return { allowed: true, quota }; + }); +} + +/** + * Update quota usage after a write (or to release a reservation). * @param {string} podName - The pod name * @param {number} bytesChange - Bytes added (positive) or removed (negative) */ export async function updateQuotaUsage(podName, bytesChange) { - const quota = await loadQuota(podName); + return withPodLock(podName, async () => { + const quota = await loadQuota(podName); - // Skip if no quota initialized - if (quota.limit === 0) return quota; + // Skip if no quota initialized + if (quota.limit === 0) return quota; - quota.used = Math.max(0, quota.used + bytesChange); - await saveQuota(podName, quota); - return quota; + quota.used = Math.max(0, quota.used + bytesChange); + await saveQuota(podName, quota); + return quota; + }); } /** diff --git a/src/webledger.js b/src/webledger.js index 8944914e..10136ee1 100644 --- a/src/webledger.js +++ b/src/webledger.js @@ -34,7 +34,11 @@ export function createLedger(options = {}) { defaultCurrency: options.defaultCurrency ?? 'satoshi', created: now, updated: now, - entries: [] + entries: [], + // Idempotency keys for deposits already absorbed into balances. Seeded + // here so a freshly created ledger has the same shape readLedger's + // migration produces for a legacy one. + credited: [] }; } @@ -56,6 +60,11 @@ export async function readLedger(ledgerPath = DEFAULT_PATH) { if (!ledger.defaultCurrency) ledger.defaultCurrency = 'satoshi'; if (!ledger.created) ledger.created = Math.floor(Date.now() / 1000); if (!ledger.entries) ledger.entries = []; + // Idempotency keys for deposits already absorbed into balances. Kept in + // the ledger itself so crediting is crash-safe: the balance and the "this + // deposit was counted" marker commit together in one writeLedger, instead + // of the balance landing in one file and the spent-marker in another. + if (!ledger.credited) ledger.credited = []; return ledger; } catch { return createLedger(); @@ -164,6 +173,34 @@ export function debit(ledger, uri, amount, currency) { return { success: true, balance: newBalance }; } +/** + * Idempotently credit a keyed deposit exactly once. + * + * The dedup key (e.g. `${chain}:${txid}:${vout}` for an on-chain UTXO) is + * recorded in the ledger's own `credited` list, so the balance increment and + * the "already counted" marker are committed together by a single writeLedger. + * A crash after that single write can never re-credit the same deposit — which + * is exactly what happened when the credit and the UTXO-spent record lived in + * two separate files written one after another. + * + * @param {object} ledger - WebLedger object (must be persisted afterwards) + * @param {string} key - Stable idempotency key for this deposit + * @param {string} uri - Agent URI + * @param {number} amount - Amount to add + * @param {string} [currency] - Currency code + * @returns {{credited: boolean, balance: number}} `credited` is false when the + * key was already absorbed (no change made) + */ +export function creditOnce(ledger, key, uri, amount, currency) { + if (!ledger.credited) ledger.credited = []; + if (ledger.credited.includes(key)) { + return { credited: false, balance: getBalance(ledger, uri, currency) }; + } + const balance = credit(ledger, uri, amount, currency); + ledger.credited.push(key); + return { credited: true, balance }; +} + /** * List all entries with non-zero balances * @param {object} ledger - WebLedger object diff --git a/test/quota-reserve.test.js b/test/quota-reserve.test.js new file mode 100644 index 00000000..3ebd1b1d --- /dev/null +++ b/test/quota-reserve.test.js @@ -0,0 +1,85 @@ +/** + * Regression tests for the quota check-and-commit race. + * + * saveQuota is atomic (temp + rename, #309), but check→write→record used to be + * three separate async steps: two concurrent writers could both pass the check + * and then overshoot the limit, and the read-modify-write in updateQuotaUsage + * could lose updates. reserveQuota + a per-pod lock make check-and-commit + * atomic — mirroring QuotaPolicy::reserve in the solid-pod-rs parity port. + */ + +import { describe, it, before, after, beforeEach } from 'node:test'; +import assert from 'node:assert'; +import fs from 'fs-extra'; +import os from 'os'; +import path from 'path'; +import { + initializeQuota, + updateQuotaUsage, + reserveQuota, + loadQuota +} from '../src/storage/quota.js'; + +const POD = 'testpod'; +let TEST_ROOT; +let originalDataRoot; + +describe('quota — atomic reservation (check-and-commit race)', () => { + before(async () => { + originalDataRoot = process.env.DATA_ROOT; + TEST_ROOT = await fs.mkdtemp(path.join(os.tmpdir(), 'jss-quota-reserve-')); + process.env.DATA_ROOT = TEST_ROOT; + await fs.ensureDir(path.join(TEST_ROOT, POD)); + }); + + after(async () => { + if (originalDataRoot === undefined) delete process.env.DATA_ROOT; + else process.env.DATA_ROOT = originalDataRoot; + await fs.remove(TEST_ROOT); + }); + + beforeEach(async () => { + await initializeQuota(POD, 1000); + }); + + it('concurrent reservations never overshoot the limit', async () => { + // 40 writers each try to reserve 100 bytes against a 1000-byte limit. + // Only 10 can be admitted; a non-atomic check would let many more through. + const N = 40; + const results = await Promise.all( + Array.from({ length: N }, () => reserveQuota(POD, 100, 0)) + ); + + const admitted = results.filter((r) => r.allowed).length; + assert.strictEqual(admitted, 10, 'exactly limit/size reservations may be admitted'); + + const final = await loadQuota(POD); + assert.ok(final.used <= final.limit, `used ${final.used} must not exceed limit ${final.limit}`); + assert.strictEqual(final.used, 1000, 'committed usage must equal the admitted reservations'); + }); + + it('concurrent updateQuotaUsage does not lose updates', async () => { + await initializeQuota(POD, 10 * 1024 * 1024); + const N = 50; + const each = 100; + await Promise.all(Array.from({ length: N }, () => updateQuotaUsage(POD, each))); + + const final = await loadQuota(POD); + assert.strictEqual(final.used, N * each, 'every increment must be recorded (no lost update)'); + }); + + it('a released reservation frees the space for a later writer', async () => { + // Fill the quota, release one slot, then a new reservation must succeed. + const filled = await Promise.all( + Array.from({ length: 10 }, () => reserveQuota(POD, 100, 0)) + ); + assert.strictEqual(filled.filter((r) => r.allowed).length, 10); + + const overflow = await reserveQuota(POD, 100, 0); + assert.strictEqual(overflow.allowed, false, 'quota is full'); + + await updateQuotaUsage(POD, -100); // simulate a failed write releasing its reservation + const retry = await reserveQuota(POD, 100, 0); + assert.strictEqual(retry.allowed, true, 'freed space is reusable'); + }); +}); diff --git a/test/webledger-credit-once.test.js b/test/webledger-credit-once.test.js new file mode 100644 index 00000000..e9036982 --- /dev/null +++ b/test/webledger-credit-once.test.js @@ -0,0 +1,106 @@ +/** + * Regression tests for the deposit double-credit window. + * + * The GET /pay/.balance auto-scanner credited the ledger and then, in a + * separate write, recorded the UTXO as seen. A crash between the two writes + * lost the seen-record, and because the scanner re-runs on every balance poll + * the same on-chain UTXO was credited again — minting balance from nothing. + * + * creditOnce records the deposit's idempotency key inside the ledger, so the + * balance and the "already counted" marker commit together. Replaying the same + * deposit — the exact effect of the crash-then-rescan — is now a no-op. This + * mirrors the single atomic state.json commit in the solid-pod-rs parity port. + */ + +import { describe, it } from 'node:test'; +import assert from 'node:assert'; +import fs from 'fs-extra'; +import os from 'os'; +import path from 'path'; +import { createLedger, creditOnce, getBalance, readLedger, LEDGER_PATH } from '../src/webledger.js'; +import * as storage from '../src/storage/filesystem.js'; + +const DID = 'did:nostr:npub1example'; +const KEY = 'tbtc4:abcd1234:0'; + +describe('webledger — creditOnce idempotency', () => { + it('credits once and reports the new balance', () => { + const ledger = createLedger(); + const r = creditOnce(ledger, KEY, DID, 5000, 'tbtc4'); + assert.strictEqual(r.credited, true); + assert.strictEqual(r.balance, 5000); + assert.strictEqual(getBalance(ledger, DID, 'tbtc4'), 5000); + }); + + it('a replayed deposit key is a no-op (no double-credit)', () => { + const ledger = createLedger(); + creditOnce(ledger, KEY, DID, 5000, 'tbtc4'); + + // Simulate the crash-then-rescan: the very same outpoint is seen again. + const replay = creditOnce(ledger, KEY, DID, 5000, 'tbtc4'); + assert.strictEqual(replay.credited, false, 'replay must not credit'); + assert.strictEqual(replay.balance, 5000, 'balance unchanged on replay'); + assert.strictEqual(getBalance(ledger, DID, 'tbtc4'), 5000); + }); + + it('distinct outpoints each credit exactly once', () => { + const ledger = createLedger(); + creditOnce(ledger, 'tbtc4:aaaa:0', DID, 1000, 'tbtc4'); + creditOnce(ledger, 'tbtc4:bbbb:1', DID, 2000, 'tbtc4'); + creditOnce(ledger, 'tbtc4:aaaa:0', DID, 1000, 'tbtc4'); // replay of the first + assert.strictEqual(getBalance(ledger, DID, 'tbtc4'), 3000); + }); + + it('readLedger migrates a legacy ledger to carry a credited array', async () => { + // A ledger written before creditOnce existed has no `credited` field. The + // migration in readLedger must add one, otherwise the idempotency guard + // has nothing to consult for deposits made against pre-existing ledgers. + // Exercised through real storage so the migration itself is under test — + // asserting on a hand-built object would only re-test the test. + const prevRoot = process.env.DATA_ROOT; + const tmpRoot = await fs.mkdtemp(path.join(os.tmpdir(), 'jss-webledger-')); + process.env.DATA_ROOT = tmpRoot; + try { + const legacy = { '@context': 'https://w3id.org/webledgers', type: 'WebLedger', entries: [] }; + await storage.write(LEDGER_PATH, Buffer.from(JSON.stringify(legacy))); + + const migrated = await readLedger(); + // Assert before any creditOnce call: creditOnce self-heals the field, so + // checking after one would pass even if the migration were deleted. + assert.ok(Array.isArray(migrated.credited), + 'readLedger must add a credited array to a legacy ledger'); + assert.strictEqual(migrated.credited.length, 0); + + // And the guard works end-to-end on the migrated ledger. + assert.strictEqual(creditOnce(migrated, KEY, DID, 100, 'tbtc4').credited, true); + assert.strictEqual(creditOnce(migrated, KEY, DID, 100, 'tbtc4').credited, false); + assert.strictEqual(getBalance(migrated, DID, 'tbtc4'), 100); + } finally { + if (prevRoot === undefined) delete process.env.DATA_ROOT; + else process.env.DATA_ROOT = prevRoot; + await fs.remove(tmpRoot); + } + }); + + it('a persisted credited marker survives a write/read round trip', async () => { + const prevRoot = process.env.DATA_ROOT; + const tmpRoot = await fs.mkdtemp(path.join(os.tmpdir(), 'jss-webledger-')); + process.env.DATA_ROOT = tmpRoot; + try { + const ledger = createLedger(); + creditOnce(ledger, KEY, DID, 5000, 'tbtc4'); + await storage.write(LEDGER_PATH, Buffer.from(JSON.stringify(ledger))); + + // Simulate the crash-then-rescan across a process restart: the marker + // must come back from disk so the replay is still a no-op. + const reread = await readLedger(); + assert.ok(reread.credited.includes(KEY), 'credited key must persist'); + assert.strictEqual(creditOnce(reread, KEY, DID, 5000, 'tbtc4').credited, false); + assert.strictEqual(getBalance(reread, DID, 'tbtc4'), 5000); + } finally { + if (prevRoot === undefined) delete process.env.DATA_ROOT; + else process.env.DATA_ROOT = prevRoot; + await fs.remove(tmpRoot); + } + }); +});