perf: bound quota scans and batch durable resource recovery

.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

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": [

docs/adr/0090-module-data-recovery.md

@@ -36,3 +36,13 @@ durable intent, учёт временных файлов и проверку к
 Начатая работа, потеря владения и другие ошибки не повторяются; после
 исчерпания попыток запись остаётся закрытой. Обычные мутации сохраняют
 прежнее неблокирующее получение блокировки.
+
+Полный пересчёт выполняет независимые `lstat` группами максимум по восемь.
+Блокировка допуска по-прежнему охватывает весь обход; счётчики фиксируются
+только после его успеха. Ошибка или отмена дожидается всех начатых файловых
+операций до выхода из блокировки. Проверки ссылок, полный состав обхода,
+частота сверки и требования к восстановлению грязного журнала не меняются.
+
+Остановка также отменяет периодический обход пишущего каталога. Незавершённая
+reconciliation не перепубликует прежние счётчики как проверенные: durable intent
+остаётся dirty, и следующий допуск требует полного восстановления учёта.

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
+остаются обязательными. Масштабная проверка записи после восстановления
+выполняется отдельно на настоящем ядре и не заменяется маленьким контрактом.

packages/infrastructure/src/module-data-accounting.ts

@@ -248,7 +248,7 @@ export class ModuleDataAccounting {
 
   reconcile(): Promise<void> {
     if (this.options.readOnly) return this.refresh();
-    return this.run(() => this.ledger.rebuild());
+    return this.run(() => this.ledger.rebuild(this.preparation.signal));
   }
 
   close(): Promise<void> {

packages/infrastructure/src/module-data-ledger.ts

@@ -20,12 +20,14 @@ export class ModuleDataLedger {
   owners: string[] = [];
   private readonly touched = new Map<string, Tracked>();
   private first = true;
+  private measured = true;
   constructor(
     private readonly root: string,
     private readonly owner: string,
   ) {}
 
   async rebuild(signal?: AbortSignal): Promise<void> {
+    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<boolean> {
@@ -110,6 +113,10 @@ export class ModuleDataLedger {
   }
 
   async commit(guard: ModuleDataGuard): Promise<void> {
+    // 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;

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<void>[] = [];
+  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<void> => {
     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 };
 }

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,13 +22,15 @@ 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,
     readOnly: boolean,
     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<void> {
-    return restoreResource(this.database, file);
+    return this.restoration.restore(file);
   }
   initialize(): Promise<void> {
     return initializeCatalog(this.database);

packages/infrastructure/src/resource-catalog-db.ts

@@ -52,10 +52,32 @@ export class ResourceCatalogDatabase {
       .toString();
   }
 
-  async run<T>(work: (client: PoolClient) => Promise<T>, write = true): Promise<T> {
+  /** 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<T>(work: (client: PoolClient) => Promise<T>, write = true): Promise<T> {
+    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<T>(work: (client: PoolClient) => Promise<T>, write = true): Promise<T> {
+    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--;
     }
   }
 

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<boolean
 }
 
 /** Additive rebuild only. Missing/corrupt files must abort reconciliation, not become garbage. */
-export async function restoreResource(
-  db: ResourceCatalogDatabase,
-  file: ResourceFile,
-): Promise<void> {
+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<typeof prepareRestore>;
+
+/** 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<void> {
+  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<void> {