diff --git a/.github/workflows/check.yml b/.github/workflows/check.yml index 4f1af07f..04798c0c 100644 --- a/.github/workflows/check.yml +++ b/.github/workflows/check.yml @@ -71,7 +71,7 @@ jobs: node-version: '24' cache: npm - run: npm ci --ignore-scripts - - run: npm run gate -- quick + - run: npm run gate -- quick module-data-scan - name: Publish gate report to logs and job summary if: ${{ always() }} run: node scripts/ci-report.mjs diff --git a/config/gates.json b/config/gates.json index 34f946ff..7605cf3f 100644 --- a/config/gates.json +++ b/config/gates.json @@ -19,7 +19,8 @@ "deployment-ha", "deployment-ha-power", "deployment-ha-operations", - "deployment-ha-three" + "deployment-ha-three", + "module-data-scan" ], "release": [ "static", @@ -41,7 +42,8 @@ "deployment-ha", "deployment-ha-power", "deployment-ha-operations", - "deployment-ha-three" + "deployment-ha-three", + "module-data-scan" ], "large": ["large-full", "large-multipart"] }, @@ -240,6 +242,17 @@ "entries": ["tests/deployment/ha-stand.mjs"], "args": ["power"], "kind": "scenario" + }, + "module-data-scan": { + "kind": "node-test", + "concurrency": 1, + "timeoutSeconds": 180, + "entries": [ + "tests/module-data-scan.test.mjs", + "tests/module-data-accounting.test.mjs", + "tests/module-data-recovery.test.mjs" + ], + "needs": ["build"] } }, "spawnedHelpers": [ diff --git a/docs/adr/0090-module-data-recovery.md b/docs/adr/0090-module-data-recovery.md index f207249c..56ce8748 100644 --- a/docs/adr/0090-module-data-recovery.md +++ b/docs/adr/0090-module-data-recovery.md @@ -36,3 +36,13 @@ durable intent, учёт временных файлов и проверку к Начатая работа, потеря владения и другие ошибки не повторяются; после исчерпания попыток запись остаётся закрытой. Обычные мутации сохраняют прежнее неблокирующее получение блокировки. + +Полный пересчёт выполняет независимые `lstat` группами максимум по восемь. +Блокировка допуска по-прежнему охватывает весь обход; счётчики фиксируются +только после его успеха. Ошибка или отмена дожидается всех начатых файловых +операций до выхода из блокировки. Проверки ссылок, полный состав обхода, +частота сверки и требования к восстановлению грязного журнала не меняются. + +Остановка также отменяет периодический обход пишущего каталога. Незавершённая +reconciliation не перепубликует прежние счётчики как проверенные: durable intent +остаётся dirty, и следующий допуск требует полного восстановления учёта. diff --git a/docs/adr/0091-bounded-resource-restore-batches.md b/docs/adr/0091-bounded-resource-restore-batches.md new file mode 100644 index 00000000..89eef415 --- /dev/null +++ b/docs/adr/0091-bounded-resource-restore-batches.md @@ -0,0 +1,38 @@ +# 0091. Ограниченные группы восстановления каталога ресурсов + +Статус: реализация и приёмка. Дата: 2026-10-23. + +## Причина + +Восстановление миллионов записей выполняло отдельный durable COMMIT для каждого +файла. На локальном стенде зарегистрированы ожидания WALSync и медленное +продвижение даже после перекрытия независимых файловых чтений. + +## Решение и гарантии + +Публичный порт restore не меняется. Адаптер одного связанного каталога собирает +поступившие за оборот event loop запросы в группу максимум по восемь. Каждое +обращение сразу занимает место в прежнем общем лимите восьми операций БД, +включая время ожидания группы. Другие операции учитываются в том же лимите. +Входные имена, ключи, sources и metadata проверяются и копируются до ожидания. + +Группа исполняется в порядке поступления под прежней advisory xact lock, с +прежними SQL timeout и проверками неизменяемости. Успех всех обещаний наступает +только после COMMIT; synchronous_commit и fsync не отключаются. Конфликт или +ошибка откатывает всю группу и отклоняет все её запросы. Частичный успех не +сообщается; вызывающий может повторить идемпотентное восстановление. + +Запросы restore следует дождаться перед initialize, как и прежде. Общая +готовность по-прежнему требует всех файлов источников. До initialize GC закрыт. +Ни пакетная обработка, ни более быстрый файловый обход не разрешают удаление +и не меняют reader pins, квоты, роли либо namespace. Порядок канонического +представления сохраняется; вызывающие параллельные запросы для разных ключей +не должны нарушать порядок вариантов одного ключа. + +## Приёмка + +PostgreSQL-проверки подтверждают один xmin для группы, общий предел admission, +копирование изменяемого входа, порядок canonical_name и полный rollback при +конфликте. Существующие проверки missing sources, namespace, read-only и GC +остаются обязательными. Масштабная проверка записи после восстановления +выполняется отдельно на настоящем ядре и не заменяется маленьким контрактом. diff --git a/packages/infrastructure/src/module-data-accounting.ts b/packages/infrastructure/src/module-data-accounting.ts index 37bc9708..78317c9c 100644 --- a/packages/infrastructure/src/module-data-accounting.ts +++ b/packages/infrastructure/src/module-data-accounting.ts @@ -248,7 +248,7 @@ export class ModuleDataAccounting { reconcile(): Promise { if (this.options.readOnly) return this.refresh(); - return this.run(() => this.ledger.rebuild()); + return this.run(() => this.ledger.rebuild(this.preparation.signal)); } close(): Promise { diff --git a/packages/infrastructure/src/module-data-ledger.ts b/packages/infrastructure/src/module-data-ledger.ts index 0f3194ba..7209ec99 100644 --- a/packages/infrastructure/src/module-data-ledger.ts +++ b/packages/infrastructure/src/module-data-ledger.ts @@ -20,12 +20,14 @@ export class ModuleDataLedger { owners: string[] = []; private readonly touched = new Map(); private first = true; + private measured = true; constructor( private readonly root: string, private readonly owner: string, ) {} async rebuild(signal?: AbortSignal): Promise { + this.measured = false; const data = await measureTree(join(this.root, 'data'), signal); const cache = await measureTree(join(this.root, 'cache'), signal); const tmp = await measureTree(join(this.root, 'tmp'), signal); @@ -36,6 +38,7 @@ export class ModuleDataLedger { temporaryFiles: tmp.files, }; this.touched.clear(); + this.measured = true; } private async load(): Promise { @@ -110,6 +113,10 @@ export class ModuleDataLedger { } async commit(guard: ModuleDataGuard): Promise { + // A cancelled/failed reconciliation must leave the durable intent dirty. Another host + // must scan again, rather than treating the previous totals as newly verified credit. + if (!this.measured) + throw new ModuleServiceError('unavailable', 'Data accounting reconciliation incomplete'); for (const [path, before] of this.touched) { const after = await entryAt(path); const bytes = (after?.isFile() ? after.size : 0) - before.bytes; diff --git a/packages/infrastructure/src/module-data-paths.ts b/packages/infrastructure/src/module-data-paths.ts index 5084fb02..291d1b5f 100644 --- a/packages/infrastructure/src/module-data-paths.ts +++ b/packages/infrastructure/src/module-data-paths.ts @@ -103,6 +103,13 @@ export async function measureTree( ): Promise<{ bytes: number; files: number }> { let bytes = 0; let files = 0; + const pending: Promise[] = []; + const drain = async () => { + const results = await Promise.allSettled(pending); + pending.length = 0; + for (const result of results) if (result.status === 'rejected') throw result.reason; + signal?.throwIfAborted(); + }; const walk = async (folder: string): Promise => { signal?.throwIfAborted(); let entries; @@ -115,13 +122,30 @@ export async function measureTree( for (const entry of entries) { signal?.throwIfAborted(); const child = join(folder, entry.name); - if (entry.isDirectory()) await walk(child); - else if (entry.isFile()) { - files++; - bytes += (await lstat(child)).size; + if (entry.isDirectory()) { + await drain(); + await walk(child); + } else if (entry.isFile()) { + pending.push( + (async () => { + const measured = await lstat(child); + signal?.throwIfAborted(); + files++; + bytes += measured.size; + })(), + ); + if (pending.length === 8) await drain(); } } + await drain(); }; - await walk(path); + try { + // The caller still holds the quota/admission lock for the complete walk. Only independent + // stats overlap; cancellation/failure drains them before that lock can be released. + await walk(path); + } catch (error) { + await Promise.allSettled(pending); + throw error; + } return { bytes, files }; } diff --git a/packages/infrastructure/src/module-resource-catalog.ts b/packages/infrastructure/src/module-resource-catalog.ts index 3aea640a..d41797dd 100644 --- a/packages/infrastructure/src/module-resource-catalog.ts +++ b/packages/infrastructure/src/module-resource-catalog.ts @@ -9,7 +9,8 @@ import { resourceObject, } from './resource-catalog-db.js'; import { collectResources, type ResourceCleanup } from './resource-catalog-collect.js'; -import { catalogReady, restoreResource, initializeCatalog } from './resource-catalog-rebuild.js'; +import { catalogReady, initializeCatalog } from './resource-catalog-rebuild.js'; +import { ResourceRestoreQueue } from './resource-catalog-restore-queue.js'; interface FileRow { key: string; @@ -21,6 +22,7 @@ interface FileRow { /** Durable references protect every physical generation of a logical resource. No TTL release. */ export class PostgresResourceCatalog implements ResourceCatalog { readonly database: ResourceCatalogDatabase; + private readonly restoration: ResourceRestoreQueue; constructor( pool: Pool, moduleId: string, @@ -28,6 +30,7 @@ export class PostgresResourceCatalog implements ResourceCatalog { private readonly cleanup: ResourceCleanup, ) { this.database = new ResourceCatalogDatabase(pool, moduleId, readOnly); + this.restoration = new ResourceRestoreQueue(this.database); } /** Only this bound port crosses into module code: no pool, namespace or cleanup grant. */ @@ -49,7 +52,7 @@ export class PostgresResourceCatalog implements ResourceCatalog { return catalogReady(this.database); } restore(file: ResourceFile): Promise { - return restoreResource(this.database, file); + return this.restoration.restore(file); } initialize(): Promise { return initializeCatalog(this.database); diff --git a/packages/infrastructure/src/resource-catalog-db.ts b/packages/infrastructure/src/resource-catalog-db.ts index 1ad185dd..6f32d206 100644 --- a/packages/infrastructure/src/resource-catalog-db.ts +++ b/packages/infrastructure/src/resource-catalog-db.ts @@ -52,10 +52,32 @@ export class ResourceCatalogDatabase { .toString(); } - async run(work: (client: PoolClient) => Promise, write = true): Promise { + /** Admission also covers restore requests waiting for a batch, not just open transactions. */ + admit(write = true): () => void { if (write && this.readOnly) throw new ModuleServiceError('read_only'); if (this.pending >= 8) throw new ModuleServiceError('busy'); this.pending++; + let released = false; + return () => { + if (released) return; + released = true; + this.pending--; + }; + } + + async run(work: (client: PoolClient) => Promise, write = true): Promise { + const release = this.admit(write); + try { + return await this.runAdmitted(work, write); + } finally { + release(); + } + } + + /** Infrastructure-only: the caller must retain admission for every grouped request until + * this transaction settles. No namespace, admission or DB object crosses the module port. */ + async runAdmitted(work: (client: PoolClient) => Promise, write = true): Promise { + if (write && this.readOnly) throw new ModuleServiceError('read_only'); let client: PoolClient | undefined; try { client = await this.pool.connect(); @@ -79,7 +101,6 @@ export class ResourceCatalogDatabase { throw new ModuleServiceError('unavailable', 'Resource catalog unavailable'); } finally { client?.release(); - this.pending--; } } diff --git a/packages/infrastructure/src/resource-catalog-rebuild.ts b/packages/infrastructure/src/resource-catalog-rebuild.ts index f1a4e492..0d3f9d92 100644 --- a/packages/infrastructure/src/resource-catalog-rebuild.ts +++ b/packages/infrastructure/src/resource-catalog-rebuild.ts @@ -1,4 +1,5 @@ import { ModuleServiceError } from '@proanima/arkvory-extension-api'; +import type { PoolClient } from 'pg'; import type { ResourceFile } from '@proanima/arkvory-extension-api'; import { resourceKey, @@ -22,10 +23,7 @@ export async function catalogReady(db: ResourceCatalogDatabase): Promise { +export function prepareRestore(file: ResourceFile) { resourceObject(file); resourceKey(file.key); resourceName(file.name); @@ -36,55 +34,65 @@ export async function restoreResource( file.metadata.byteLength > 8192 ) throw new ModuleServiceError('invalid_input'); - const key = file.key, - name = file.name, - metadata = Buffer.from(file.metadata); - await db.run(async (client) => { - const m = db.moduleId; - const ready = await client.query<{ ready: boolean }>( - 'SELECT ready FROM arkvory_module_resource_catalogs WHERE module_id=$1', - [m], - ); - if (ready.rows[0]?.ready) - throw new ModuleServiceError('conflict', 'Catalog already initialized'); - await client.query( - `INSERT INTO arkvory_module_resources(module_id,key,unclaimed_at) + return { key: file.key, name: file.name, metadata: Buffer.from(file.metadata), sources }; +} + +export type PreparedRestore = ReturnType; + +/** Caller holds admission for every item. One commit; disagreement rolls the entire batch back. */ +export async function restoreResources( + db: ResourceCatalogDatabase, + files: readonly PreparedRestore[], +) { + await db.runAdmitted(async (client) => { + for (const file of files) await restoreEntry(client, db.moduleId, file); + }); +} + +async function restoreEntry(client: PoolClient, m: string, file: PreparedRestore): Promise { + const { key, name, metadata, sources } = file; + const ready = await client.query<{ ready: boolean }>( + 'SELECT ready FROM arkvory_module_resource_catalogs WHERE module_id=$1', + [m], + ); + if (ready.rows[0]?.ready) throw new ModuleServiceError('conflict', 'Catalog already initialized'); + await client.query( + `INSERT INTO arkvory_module_resources(module_id,key,unclaimed_at) SELECT $1,unnest($2::text[]),clock_timestamp() ON CONFLICT DO NOTHING`, - [m, [key, ...sources]], - ); - const previous = await client.query<{ key: string; metadata: Buffer; sources: string[] }>( - 'SELECT key,metadata,sources FROM arkvory_module_resource_files WHERE module_id=$1 AND name=$2', - [m, name], - ); - const row = previous.rows[0]; - if ( - row && - (row.key !== key || - !row.metadata.equals(metadata) || - JSON.stringify(row.sources) !== JSON.stringify(sources)) - ) - throw new ModuleServiceError('conflict', 'Resource rebuild disagreement'); - await client.query( - `INSERT INTO arkvory_module_resource_files(module_id,key,name,metadata,sources,ready) + [m, [key, ...sources]], + ); + const previous = await client.query<{ key: string; metadata: Buffer; sources: string[] }>( + 'SELECT key,metadata,sources FROM arkvory_module_resource_files WHERE module_id=$1 AND name=$2', + [m, name], + ); + const row = previous.rows[0]; + if ( + row && + (row.key !== key || + !row.metadata.equals(metadata) || + JSON.stringify(row.sources) !== JSON.stringify(sources)) + ) + throw new ModuleServiceError('conflict', 'Resource rebuild disagreement'); + await client.query( + `INSERT INTO arkvory_module_resource_files(module_id,key,name,metadata,sources,ready) VALUES($1,$2,$3,$4,$5,true) ON CONFLICT DO NOTHING`, - [m, key, name, metadata, sources], - ); - await client.query( - `INSERT INTO arkvory_module_resource_edges(module_id,key,source) + [m, key, name, metadata, sources], + ); + await client.query( + `INSERT INTO arkvory_module_resource_edges(module_id,key,source) SELECT $1,$2,unnest($3::text[]) ON CONFLICT DO NOTHING`, - [m, key, sources], - ); - await client.query( - `UPDATE arkvory_module_resources SET unclaimed_at=NULL + [m, key, sources], + ); + await client.query( + `UPDATE arkvory_module_resources SET unclaimed_at=NULL WHERE module_id=$1 AND key=ANY($2::text[])`, - [m, sources], - ); - await client.query( - `UPDATE arkvory_module_resources SET canonical_name=COALESCE(canonical_name,$3) + [m, sources], + ); + await client.query( + `UPDATE arkvory_module_resources SET canonical_name=COALESCE(canonical_name,$3) WHERE module_id=$1 AND key=$2`, - [m, key, name], - ); - }); + [m, key, name], + ); } export async function initializeCatalog(db: ResourceCatalogDatabase): Promise { diff --git a/packages/infrastructure/src/resource-catalog-restore-queue.ts b/packages/infrastructure/src/resource-catalog-restore-queue.ts new file mode 100644 index 00000000..4d08290e --- /dev/null +++ b/packages/infrastructure/src/resource-catalog-restore-queue.ts @@ -0,0 +1,57 @@ +import type { ResourceFile } from '@proanima/arkvory-extension-api'; +import type { ResourceCatalogDatabase } from './resource-catalog-db.js'; +import { + prepareRestore, + restoreResources, + type PreparedRestore, +} from './resource-catalog-rebuild.js'; + +interface Request { + readonly file: PreparedRestore; + readonly resolve: () => void; + readonly reject: (error: unknown) => void; + readonly release: () => void; +} + +/** At most eight outstanding restores per bound catalog, including the in-flight batch. + * All results wait for the same durable commit; FIFO preserves canonical selection. */ +export class ResourceRestoreQueue { + private readonly queue: Request[] = []; + private scheduled = false; + constructor(private readonly database: ResourceCatalogDatabase) {} + + async restore(file: ResourceFile): Promise { + // Snapshot mutable module input before handing control back to the caller. + const prepared = prepareRestore(file); + const release = this.database.admit(); + return new Promise((resolve, reject) => { + this.queue.push({ file: prepared, resolve, reject, release }); + this.schedule(); + }); + } + + private schedule(): void { + if (this.scheduled) return; + this.scheduled = true; + setImmediate(() => { + void this.flush(); + }); + } + + private async flush(): Promise { + const batch = this.queue.splice(0); + try { + await restoreResources( + this.database, + batch.map((request) => request.file), + ); + for (const request of batch) request.resolve(); + } catch (error) { + for (const request of batch) request.reject(error); + } finally { + for (const request of batch) request.release(); + this.scheduled = false; + if (this.queue.length) this.schedule(); + } + } +} diff --git a/tests/integration/resource-catalog.test.mjs b/tests/integration/resource-catalog.test.mjs index cbd53f45..b67fbfa1 100644 --- a/tests/integration/resource-catalog.test.mjs +++ b/tests/integration/resource-catalog.test.mjs @@ -90,6 +90,50 @@ resourceCatalogContract('PostgreSQL resource catalog', async (t) => { }; }); +test('concurrent restores share one durable transaction and keep the eight-request bound', async (t) => { + const f = await fixture(t); + const requests = Array.from({ length: 8 }, (_, i) => f.catalog.restore(f.record(`batch-${i}`))); + await assert.rejects(f.catalog.restore(f.record('overflow')), { code: 'busy' }); + await assert.rejects(f.catalog.ready(), { code: 'busy' }); + await Promise.all(requests); + const rows = await f.catalog.database.pool.query( + `SELECT count(*)::integer AS files, count(DISTINCT xmin::text)::integer AS transactions + FROM arkvory_module_resource_files WHERE module_id='resource-test'`, + ); + assert.deepEqual(rows.rows[0], { files: 8, transactions: 1 }); + assert.equal(await f.catalog.ready(), false); + assert.equal(await f.catalog.collect(0, 10), 0); +}); + +test('restore batch snapshots input, preserves canonical order and rolls back disagreements', async (t) => { + const f = await fixture(t); + const original = f.record('same', 'first'); + const first = f.catalog.restore(original); + original.metadata.fill(0); + original.key = 'changed'; + original.sources.push('missing'); + await Promise.all([first, f.catalog.restore(f.record('same', 'second'))]); + const row = await f.catalog.find('same'); + assert.equal(row.name, 'resources/same/first'); + assert.deepEqual(Buffer.from(row.metadata), Buffer.from('same')); + const conflicting = { ...f.record('same', 'first'), metadata: Buffer.from('disagreement') }; + const results = await Promise.allSettled([ + f.catalog.restore(f.record('must-roll-back')), + f.catalog.restore(conflicting), + ]); + for (const result of results) { + assert.equal(result.status, 'rejected'); + assert.equal(result.reason.code, 'conflict'); + } + assert.equal(await f.catalog.find('must-roll-back'), null); + assert.equal(await f.catalog.ready(), false); + await assert.rejects(f.create('resource-test', true).restore(f.record('read-only')), { + code: 'read_only', + }); + await f.catalog.restore(f.record('after-failure')); + assert.ok(await f.catalog.find('after-failure'), 'failed batch releases every admission slot'); +}); + test('resource catalog protects shared owners across adapters and permanently fences closed uploads', async (t) => { const f = await fixture(t), a = f.catalog, diff --git a/tests/module-data-scan.test.mjs b/tests/module-data-scan.test.mjs new file mode 100644 index 00000000..097db2ab --- /dev/null +++ b/tests/module-data-scan.test.mjs @@ -0,0 +1,139 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import fs from 'node:fs'; +import { syncBuiltinESMExports } from 'node:module'; +import { randomUUID } from 'node:crypto'; +import { mkdtemp, mkdir, writeFile, readFile } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join, sep } from 'node:path'; +import { setImmediate as turn } from 'node:timers/promises'; +import { openModuleData } from '@proanima/arkvory-infrastructure'; +import { removeTestDirectory } from './helpers.mjs'; + +async function fixture(t, mode) { + const root = await mkdtemp(join(tmpdir(), 'arkvory-quota-scan-')); + t.after(() => removeTestDirectory(root)); + const base = join(root, 'module-data', 'scan'); + const area = join(base, 'data'); + await mkdir(area, { recursive: true }); + for (let i = 0; i < 20; i++) await writeFile(join(area, String(i)), Buffer.alloc(i + 1)); + const entered = Promise.withResolvers(); + const release = Promise.withResolvers(); + const state = { active: 0, peak: 0, block: mode === 'cancel' }; + const original = fs.promises.lstat; + fs.promises.lstat = async (path, ...args) => { + if (typeof path !== 'string' || !path.startsWith(area + sep)) return original(path, ...args); + state.active++; + state.peak = Math.max(state.peak, state.active); + if (state.active === 8 && state.block) entered.resolve(); + try { + if (state.block) await release.promise; + else await turn(); + if (mode === 'fault') throw new Error('scan fault'); + return await original(path, ...args); + } finally { + state.active--; + } + }; + syncBuiltinESMExports(); + t.after(() => { + release.resolve(); + fs.promises.lstat = original; + syncBuiltinESMExports(); + }); + return { + base, + state, + entered, + release, + options: { + dataDirectory: root, + moduleId: 'scan', + quota: { bytes: 210, files: 20 }, + reserve: { bytes: 0, inodes: 0 }, + readOnly: false, + coordinator: { + owner: randomUUID(), + run: async (_id, work) => work({ check() {}, orphan: async () => {} }), + }, + }, + }; +} + +test('quota reconciliation bounds stats and retains exact byte/file admission', async (t) => { + const f = await fixture(t); + const data = await openModuleData(f.options); + t.after(() => data.close()); + assert.ok(f.state.peak > 1 && f.state.peak <= 8); + assert.equal(f.state.active, 0); + assert.deepEqual(data.directory.data.usage(), { + bytes: 210, + files: 20, + quotaBytes: 210, + quotaFiles: 20, + }); + await assert.rejects( + data.directory.data.create('overflow', { durable: false, exclusive: true }), + { code: 'module_quota_exceeded' }, + ); +}); + +test('cancelled periodic accounting leaves dirty intent and forces full quota recovery', async (t) => { + const f = await fixture(t); + const data = await openModuleData(f.options); + const next = await openModuleData({ + ...f.options, + quota: { bytes: 212, files: 21 }, + coordinator: { ...f.options.coordinator, owner: randomUUID() }, + }); + t.after(() => next.close()); + t.after(async () => { + f.release.resolve(); + await data.close(); + }); + await writeFile(join(f.base, 'data', 'late'), Buffer.alloc(2)); + f.state.block = true; + const rejected = assert.rejects(data.reconcile(), { code: 'unavailable' }); + await f.entered.promise; + const closing = data.close(); + f.release.resolve(); + await rejected; + await closing; + assert.equal(f.state.active, 0); + assert.equal(await readFile(join(f.base, '.accounting.dirty'), 'utf8'), '1'); + f.state.block = false; + await assert.rejects( + next.directory.data.create('overflow', { durable: false, exclusive: true }), + { code: 'module_quota_exceeded' }, + ); + assert.equal(next.directory.data.usage().bytes, 212); + assert.equal(next.directory.data.usage().files, 21); +}); + +test('quota scan failure drains stats without committing partial accounting', async (t) => { + const f = await fixture(t, 'fault'); + await assert.rejects(openModuleData(f.options), /scan fault/); + assert.equal(f.state.active, 0); + await assert.rejects(readFile(join(f.base, '.accounting.json')), { code: 'ENOENT' }); +}); + +test('closing recovery waits for cancelled scan stats and publishes no partial credit', async (t) => { + const f = await fixture(t, 'cancel'); + const data = await openModuleData({ ...f.options, deferredAccounting: true }); + t.after(async () => { + f.release.resolve(); + await data.close(); + }); + await f.entered.promise; + let closed = false; + const closing = data.close().then(() => { + closed = true; + }); + await turn(); + assert.equal(closed, false); + f.release.resolve(); + await closing; + assert.equal(f.state.active, 0); + assert.throws(() => data.directory.data.usage()); + await assert.rejects(readFile(join(f.base, '.accounting.json')), { code: 'ENOENT' }); +});