134 lines
8.9 KiB
TypeScript
134 lines
8.9 KiB
TypeScript
import assert from "node:assert/strict";
|
|
import { access, mkdir, mkdtemp, rm } from "node:fs/promises";
|
|
import os from "node:os";
|
|
import path from "node:path";
|
|
import test from "node:test";
|
|
import { createApp } from "../../src/app.js";
|
|
import { env } from "../../src/config/env.js";
|
|
import { KnowledgeLifecycleReconciler } from "../../src/modules/catalog/reconciler.js";
|
|
import { CatalogRepository } from "../../src/modules/catalog/repository.js";
|
|
import { OcrRetentionService, type OcrRetentionCandidate, type OcrRetentionStore } from "../../src/modules/ocr/retention.js";
|
|
|
|
const activeId = "11111111-1111-4111-8111-111111111111";
|
|
const failedId = "22222222-2222-4222-8222-222222222222";
|
|
const deletingId = "33333333-3333-4333-8333-333333333333";
|
|
|
|
function candidate(versionId: string, state: OcrRetentionCandidate["state"], artifactState: OcrRetentionCandidate["artifactState"] = "present"): OcrRetentionCandidate {
|
|
return { versionId, sourceId: `source:${versionId}`, state, artifactState };
|
|
}
|
|
|
|
test("repository retention selection applies exact TTLs and CAS-protects active versions", async () => {
|
|
const queries: Array<{ sql: string; params?: unknown[] }> = [];
|
|
const pool = { async query(sql: string, params?: unknown[]) {
|
|
queries.push({ sql, params });
|
|
if (sql.includes("SELECT v.version_id")) return { rowCount: 1, rows: [{ version_id: failedId, source_id: "source:failed", state: "failed", artifact_state: "present" }] };
|
|
return { rowCount: 1, rows: [] };
|
|
} };
|
|
const repository = new CatalogRepository(pool as never);
|
|
const now = new Date("2026-09-15T12:00:00.000Z");
|
|
|
|
assert.deepEqual(await repository.listOcrRetentionCandidates(now), [{ ...candidate(failedId, "failed"), sourceId: "source:failed" }]);
|
|
assert.equal(await repository.expireOcrReview("source:review", activeId, now), true);
|
|
assert.equal(await repository.claimOcrRetentionDeletion("source:failed", failedId, "failed", "present"), true);
|
|
assert.equal(await repository.completeOcrRetentionDeletion("source:failed", failedId, "failed"), true);
|
|
|
|
assert.match(queries[0]!.sql, /review_required.*30 days.*failed.*rejected.*7 days.*superseded.*30 days/s);
|
|
assert.match(queries[0]!.sql, /active_version_id IS DISTINCT FROM v\.version_id/);
|
|
assert.deepEqual(queries[0]!.params, [now]);
|
|
assert.match(queries[1]!.sql, /state = 'rejected'.*REVIEW_EXPIRED.*retention_due_at/s);
|
|
assert.match(queries[2]!.sql, /SET artifact_state = 'retention_deleting'.*state = \$3.*artifact_state = \$4/s);
|
|
assert.match(queries[2]!.sql, /active_version_id IS DISTINCT FROM v\.version_id/);
|
|
assert.match(queries[3]!.sql, /artifact_state = 'retention_deleted'.*artifact_state = 'retention_deleting'/s);
|
|
});
|
|
|
|
test("activation cannot race an artifact deletion already claimed by retention", async () => {
|
|
const pool = {
|
|
async query(sql: string) {
|
|
if (sql.includes("maintenance")) return { rowCount: 1, rows: [{ maintenance: false }] };
|
|
if (sql.includes("FROM rag_sources")) return { rowCount: 1, rows: [{ active_version_id: null }] };
|
|
return { rowCount: 1, rows: [{ version_id: failedId, source_id: "source:failed", version_number: 2, previous_version_id: null, state: "ready", tags: [], source_content_hash: "hash", processing_fingerprint: "fingerprint", metadata_hash: "metadata", embedding_provider: "test", embedding_model: "test", embedding_dimensions: 3, expected_document_count: 1, expected_point_count: 1, verified_point_count: 1, qdrant_collection: "rag", artifact_state: "retention_deleting" }] };
|
|
},
|
|
async connect() { return { query: this.query, release() {} }; }
|
|
};
|
|
const repository = new CatalogRepository(pool as never);
|
|
|
|
await assert.rejects(repository.activateVersion("source:failed", failedId, null), (error) => error instanceof Error && (error as { code?: string }).code === "VERSION_ARTIFACTS_UNAVAILABLE");
|
|
});
|
|
|
|
test("retention expires review before removal and preserves active artifacts", async (context) => {
|
|
const root = await mkdtemp(path.join(os.tmpdir(), "rag-retention-"));
|
|
context.after(() => rm(root, { recursive: true, force: true }));
|
|
await Promise.all([activeId, failedId].map((id) => mkdir(path.join(root, id))));
|
|
const calls: string[] = [];
|
|
const store: OcrRetentionStore = {
|
|
async listOcrRetentionCandidates() { return [candidate(activeId, "active"), candidate(failedId, "failed"), candidate(deletingId, "review_required")]; },
|
|
async withVersionTryLock(versionId, handler) { calls.push(`lock:${versionId}`); return handler(); },
|
|
async expireOcrReview(_sourceId, versionId) { calls.push(`expire:${versionId}`); return true; },
|
|
async claimOcrRetentionDeletion(_sourceId, versionId) { calls.push(`claim:${versionId}`); return true; },
|
|
async completeOcrRetentionDeletion(_sourceId, versionId) { calls.push(`complete:${versionId}`); return true; }
|
|
};
|
|
|
|
assert.deepEqual(await new OcrRetentionService(store, root).runOnce(), { expired: 1, deleted: 1, resumed: 0, skipped: 1 });
|
|
await access(path.join(root, activeId));
|
|
await assert.rejects(access(path.join(root, failedId)));
|
|
assert.equal(calls.some((call) => call.includes(activeId)), false);
|
|
assert.equal(calls.includes(`expire:${deletingId}`), true);
|
|
});
|
|
|
|
test("interrupted deletion resumes idempotently without affecting another version", async (context) => {
|
|
const root = await mkdtemp(path.join(os.tmpdir(), "rag-retention-resume-"));
|
|
context.after(() => rm(root, { recursive: true, force: true }));
|
|
await Promise.all([deletingId, activeId].map((id) => mkdir(path.join(root, id))));
|
|
let completed = false;
|
|
let failCompletion = true;
|
|
const store: OcrRetentionStore = {
|
|
async listOcrRetentionCandidates() { return completed ? [] : [candidate(deletingId, "rejected", "retention_deleting")]; },
|
|
async withVersionTryLock(_versionId, handler) { return handler(); }, async expireOcrReview() { return false; },
|
|
async claimOcrRetentionDeletion() { return true; },
|
|
async completeOcrRetentionDeletion() { if (failCompletion) { failCompletion = false; throw new Error("simulated restart"); } completed = true; return true; }
|
|
};
|
|
const retention = new OcrRetentionService(store, root);
|
|
|
|
await assert.rejects(retention.runOnce(), /simulated restart/);
|
|
await assert.rejects(access(path.join(root, deletingId)));
|
|
assert.deepEqual(await retention.runOnce(), { expired: 0, deleted: 0, resumed: 1, skipped: 0 });
|
|
assert.deepEqual(await retention.runOnce(), { expired: 0, deleted: 0, resumed: 0, skipped: 0 });
|
|
await access(path.join(root, activeId));
|
|
});
|
|
|
|
test("OCR flag off keeps native HTTP synchronous and hides candidate surfaces", async (context) => {
|
|
const previous = { enabled: env.ocrIngestEnabled, lifecycle: env.knowledgeLifecycleEnforced, token: env.lifecycleAdminToken };
|
|
Object.assign(env, { ocrIngestEnabled: false, knowledgeLifecycleEnforced: true, lifecycleAdminToken: "retention-token" });
|
|
context.after(() => Object.assign(env, previous));
|
|
let candidateReads = 0;
|
|
const native = { accepted: true, sourceId: "source:native", versionId: "native-1", state: "active" };
|
|
const app = createApp({
|
|
ingestService: { async ingest() { return native; }, async cleanup() { return { deleted: 0 }; } } as never,
|
|
catalog: { async getIngestionStatus() { candidateReads += 1; return {}; } } as never,
|
|
reviewService: { async view() { candidateReads += 1; return {}; }, async approve() { candidateReads += 1; return {} as never; }, async reject() { candidateReads += 1; return {} as never; } },
|
|
indexingService: { async index() { candidateReads += 1; return {} as never; } }, startReconciler: false
|
|
});
|
|
const server = app.listen(0);
|
|
context.after(() => server.close());
|
|
const address = server.address();
|
|
assert.ok(address && typeof address === "object");
|
|
const base = `http://127.0.0.1:${address.port}`;
|
|
|
|
assert.equal((await fetch(`${base}/ingest`, { method: "POST", headers: { "content-type": "application/json" }, body: "{}" })).status, 201);
|
|
for (const [suffix, method] of [["", "GET"], ["/review", "GET"], ["/approve", "POST"], ["/reject", "POST"]] as const) {
|
|
assert.equal((await fetch(`${base}/ingestions/${failedId}${suffix}`, { method, headers: { authorization: "Bearer retention-token", "content-type": "application/json" }, body: method === "POST" ? "{}" : undefined })).status, 404);
|
|
}
|
|
assert.equal(candidateReads, 0);
|
|
});
|
|
|
|
test("reconciler runs retention inside its exclusive global lock", async () => {
|
|
let locked = false;
|
|
const catalog = {
|
|
async withGlobalTryLock(_name: string, handler: () => Promise<unknown>) { locked = true; try { return await handler(); } finally { locked = false; } },
|
|
async resolveActiveVersions() { return []; }, async validateActiveInvariant() { return []; }, async listOrphanedIndexingCandidates() { return []; }
|
|
};
|
|
const retention = { async runOnce() { assert.equal(locked, true); return { expired: 0, deleted: 0, resumed: 0, skipped: 0 }; } };
|
|
const reconciler = new KnowledgeLifecycleReconciler(catalog as never, {} as never, undefined, retention);
|
|
|
|
assert.equal((await reconciler.runOnce()).ocrRetentionDeleted, 0);
|
|
});
|