FN-5975: extend archive FTS maintenance coverage
Add archive-database FTS maintenance and document the expanded compaction policy. - add archive FTS maintenance helpers for optimize, rebuild, size measurement, and row counts - extend self-healing maintenance to compact and rebuild archived_tasks_fts on a slower archive-specific cadence - cover archive FTS maintenance with new core and engine tests and update architecture/storage docs Files changed: docs/architecture.md | 2 +- docs/storage.md | 12 +- .../__tests__/archive-db-fts-maintenance.test.ts | 221 +++++++++++++++++++++ packages/core/src/archive-db.ts | 61 +++++- packages/core/src/db.ts | 21 +- packages/core/src/store.ts | 20 ++ .../src/__tests__/fts-maintenance-archive.test.ts | 207 +++++++++++++++++++ packages/engine/src/self-healing.ts | 80 ++++++++ 8 files changed, 608 insertions(+), 16 deletions(-) Fusion-Task-Id: FN-5975 Fusion-Task-Lineage: 13645c8c-3126-4d1b-af23-7cab1a8bb276
This commit is contained in:
gsxdsm
207
packages/engine/src/__tests__/fts-maintenance-archive.test.ts
Normal file
207
packages/engine/src/__tests__/fts-maintenance-archive.test.ts
Normal file
@@ -0,0 +1,207 @@
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import { EventEmitter } from "node:events";
|
||||
import { mkdtempSync, rmSync } from "node:fs";
|
||||
import { rm } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
|
||||
import { TaskStore, type Settings, type TaskStore as TaskStoreType } from "@fusion/core";
|
||||
|
||||
import { SelfHealingManager } from "../self-healing.js";
|
||||
|
||||
function createMockStore(overrides: Record<string, unknown> = {}): TaskStoreType & EventEmitter {
|
||||
const emitter = new EventEmitter();
|
||||
return Object.assign(emitter, {
|
||||
getSettings: vi.fn().mockResolvedValue({
|
||||
maintenanceIntervalMs: 0,
|
||||
globalPause: false,
|
||||
enginePaused: false,
|
||||
} as unknown as Settings),
|
||||
listTasks: vi.fn().mockResolvedValue([]),
|
||||
recordRunAuditEvent: vi.fn().mockResolvedValue(undefined),
|
||||
fts5Available: true,
|
||||
archiveFts5Available: false,
|
||||
getFtsIndexBytes: vi.fn().mockReturnValueOnce(4096).mockReturnValueOnce(2048),
|
||||
getTaskRowCount: vi.fn().mockReturnValue(4),
|
||||
optimizeFts5: vi.fn().mockReturnValue(true),
|
||||
getDatabase: vi.fn().mockReturnValue({ rebuildFts5Index: vi.fn().mockReturnValue(true) }),
|
||||
getArchiveFtsIndexBytes: vi.fn(),
|
||||
getArchivedRowCount: vi.fn(),
|
||||
optimizeArchiveFts5: vi.fn(),
|
||||
rebuildArchiveFts5Index: vi.fn(),
|
||||
...overrides,
|
||||
}) as unknown as TaskStoreType & EventEmitter;
|
||||
}
|
||||
|
||||
function makeTmpDir(prefix: string): string {
|
||||
return mkdtempSync(join(tmpdir(), prefix));
|
||||
}
|
||||
|
||||
const createdDirs = new Set<string>();
|
||||
|
||||
function trackDir(path: string): string {
|
||||
createdDirs.add(path);
|
||||
return path;
|
||||
}
|
||||
|
||||
async function createStore(options?: { disableFts5?: boolean; inMemoryDb?: boolean }) {
|
||||
const prevEnv = process.env.FUSION_DISABLE_FTS5;
|
||||
if (options?.disableFts5) {
|
||||
process.env.FUSION_DISABLE_FTS5 = "1";
|
||||
} else if (prevEnv === "1") {
|
||||
delete process.env.FUSION_DISABLE_FTS5;
|
||||
}
|
||||
|
||||
const rootDir = trackDir(makeTmpDir("kb-engine-archive-fts-root-"));
|
||||
const globalDir = trackDir(makeTmpDir("kb-engine-archive-fts-global-"));
|
||||
const store = new TaskStore(rootDir, globalDir, { inMemoryDb: options?.inMemoryDb === true });
|
||||
await store.init();
|
||||
const manager = new SelfHealingManager(store, { rootDir });
|
||||
|
||||
return {
|
||||
rootDir,
|
||||
globalDir,
|
||||
store,
|
||||
manager,
|
||||
restoreEnv() {
|
||||
if (prevEnv === undefined) {
|
||||
delete process.env.FUSION_DISABLE_FTS5;
|
||||
} else {
|
||||
process.env.FUSION_DISABLE_FTS5 = prevEnv;
|
||||
}
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
async function cleanupStore(context: Awaited<ReturnType<typeof createStore>> | undefined) {
|
||||
if (!context) return;
|
||||
context.manager.stop();
|
||||
context.store.close();
|
||||
context.restoreEnv();
|
||||
await rm(context.rootDir, { recursive: true, force: true });
|
||||
await rm(context.globalDir, { recursive: true, force: true });
|
||||
createdDirs.delete(context.rootDir);
|
||||
createdDirs.delete(context.globalDir);
|
||||
}
|
||||
|
||||
afterEach(async () => {
|
||||
vi.restoreAllMocks();
|
||||
for (const dir of Array.from(createdDirs)) {
|
||||
try {
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
} catch {
|
||||
rmSync(dir, { recursive: true, force: true });
|
||||
} finally {
|
||||
createdDirs.delete(dir);
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
describe("SelfHealingManager archive FTS maintenance", () => {
|
||||
it("skips the archive branch without disturbing live maintenance when archive FTS is unavailable", async () => {
|
||||
const store = createMockStore();
|
||||
const manager = new SelfHealingManager(store, { rootDir: "/tmp/test-project" });
|
||||
(manager as any).maintenanceTickCounter = 1;
|
||||
|
||||
await (manager as any).maintainTaskFts();
|
||||
|
||||
expect(store.optimizeFts5).toHaveBeenCalledWith("merge");
|
||||
expect(store.getArchiveFtsIndexBytes).not.toHaveBeenCalled();
|
||||
expect(store.optimizeArchiveFts5).not.toHaveBeenCalled();
|
||||
expect(store.recordRunAuditEvent).toHaveBeenCalledTimes(1);
|
||||
expect(store.recordRunAuditEvent).toHaveBeenCalledWith(expect.objectContaining({
|
||||
mutationType: "task:fts-maintenance",
|
||||
target: "tasks_fts",
|
||||
}));
|
||||
});
|
||||
|
||||
it("compacts a real disk-backed archive index and preserves archive search results", async () => {
|
||||
let ctx: Awaited<ReturnType<typeof createStore>> | undefined;
|
||||
try {
|
||||
ctx = await createStore();
|
||||
const { store, manager } = ctx;
|
||||
const archiveDb = (store as any).archiveDb;
|
||||
if (!archiveDb.fts5Available) {
|
||||
expect(store.archiveFts5Available).toBe(false);
|
||||
return;
|
||||
}
|
||||
|
||||
const archivedTask = await store.createTask({
|
||||
title: "archive maintenance seed",
|
||||
description: "archive-maintenance-needle",
|
||||
column: "done",
|
||||
});
|
||||
await store.archiveTask(archivedTask.id);
|
||||
|
||||
const archivedEntry = await store.findInArchive(archivedTask.id);
|
||||
expect(archivedEntry).toBeDefined();
|
||||
const seedEntry = archivedEntry!;
|
||||
|
||||
const payload = "alpha ".repeat(1600);
|
||||
for (let i = 0; i < 240; i++) {
|
||||
archiveDb.upsert({
|
||||
...seedEntry,
|
||||
title: `archive-maintenance-seed-${i}`,
|
||||
description: `${payload}archive-maintenance-needle marker-${i}`,
|
||||
comments: [{ id: `c-${i}`, text: `${payload}comment-${i}`, author: "tester", createdAt: new Date(1717372800000 + i * 1000).toISOString() }],
|
||||
archivedAt: new Date(1717372800000 + i * 1000).toISOString(),
|
||||
updatedAt: new Date(1717372800000 + i * 1000).toISOString(),
|
||||
});
|
||||
}
|
||||
|
||||
const grownBytes = store.getArchiveFtsIndexBytes();
|
||||
expect(grownBytes).not.toBeNull();
|
||||
expect(grownBytes!).toBeGreaterThan(512 * 1024);
|
||||
|
||||
const beforeResults = await store.searchTasks("archive-maintenance-needle");
|
||||
expect(beforeResults.map((task) => task.id)).toContain(archivedTask.id);
|
||||
|
||||
(manager as any).maintenanceTickCounter = 24;
|
||||
await (manager as any).maintainTaskFts();
|
||||
|
||||
const compactedBytes = store.getArchiveFtsIndexBytes();
|
||||
expect(compactedBytes).not.toBeNull();
|
||||
expect(compactedBytes!).toBeLessThan(grownBytes!);
|
||||
expect(compactedBytes!).toBeLessThan(store.getArchivedRowCount() * 512 * 1024);
|
||||
|
||||
const afterResults = await store.searchTasks("archive-maintenance-needle");
|
||||
expect(afterResults.map((task) => task.id)).toContain(archivedTask.id);
|
||||
|
||||
const auditEvents = store.getRunAuditEvents({ mutationType: "task:fts-maintenance", limit: 20 })
|
||||
.filter((event) => event.target === "archived_tasks_fts");
|
||||
expect(auditEvents.length).toBeGreaterThan(0);
|
||||
expect(auditEvents.at(-1)).toEqual(expect.objectContaining({
|
||||
mutationType: "task:fts-maintenance",
|
||||
target: "archived_tasks_fts",
|
||||
metadata: expect.objectContaining({
|
||||
rowCount: 1,
|
||||
}),
|
||||
}));
|
||||
} finally {
|
||||
await cleanupStore(ctx);
|
||||
}
|
||||
});
|
||||
|
||||
it("keeps archive fallback search working when FTS5 is disabled", async () => {
|
||||
let ctx: Awaited<ReturnType<typeof createStore>> | undefined;
|
||||
try {
|
||||
ctx = await createStore({ disableFts5: true });
|
||||
const { store, manager } = ctx;
|
||||
|
||||
const archivedTask = await store.createTask({
|
||||
title: "archive fallback target",
|
||||
description: "archive-fallback-needle",
|
||||
column: "done",
|
||||
});
|
||||
await store.archiveTask(archivedTask.id);
|
||||
|
||||
await expect((manager as any).maintainTaskFts()).resolves.toBeUndefined();
|
||||
expect(store.archiveFts5Available).toBe(false);
|
||||
|
||||
const results = await store.searchTasks("archive-fallback-needle");
|
||||
expect(results.map((task) => task.id)).toContain(archivedTask.id);
|
||||
} finally {
|
||||
await cleanupStore(ctx);
|
||||
}
|
||||
});
|
||||
});
|
||||
@@ -78,6 +78,13 @@ const FTS_MAINTENANCE_OPTIMIZE_CADENCE_TICKS = 4;
|
||||
// bounded so sustained text churn heals before segment growth becomes material.
|
||||
const FTS_REBUILD_THRESHOLD_BYTES = 32 * 1024 * 1024;
|
||||
const FTS_REBUILD_BYTES_PER_TASK = 1 * 1024 * 1024;
|
||||
// The archive index is mostly append-only, so maintenance can run much less
|
||||
// often than the live task index. We still cap total growth because archive
|
||||
// rows retain full title/description/comments payloads for the project's life.
|
||||
const ARCHIVE_FTS_MAINTENANCE_MERGE_CADENCE_TICKS = 8;
|
||||
const ARCHIVE_FTS_MAINTENANCE_OPTIMIZE_CADENCE_TICKS = 24;
|
||||
const ARCHIVE_FTS_REBUILD_THRESHOLD_BYTES = 64 * 1024 * 1024;
|
||||
const ARCHIVE_FTS_REBUILD_BYTES_PER_TASK = 512 * 1024;
|
||||
export const STALE_ACTIVE_BRANCH_EXECUTION_GRACE_MS = 10 * 60_000;
|
||||
export const COMPLETION_HANDOFF_LIMBO_GRACE_MS = 5 * 60_000;
|
||||
export const MAX_COMPLETION_HANDOFF_LIMBO_RECOVERIES = 3;
|
||||
@@ -8820,6 +8827,17 @@ export class SelfHealingManager {
|
||||
}
|
||||
|
||||
private async maintainTaskFts(): Promise<void> {
|
||||
await this.maintainLiveTaskFts();
|
||||
|
||||
try {
|
||||
await this.maintainArchiveTaskFts();
|
||||
} catch (err: unknown) {
|
||||
const errorMessage = err instanceof Error ? err.message : String(err);
|
||||
log.error(`Archive FTS maintenance failed: ${errorMessage}`);
|
||||
}
|
||||
}
|
||||
|
||||
private async maintainLiveTaskFts(): Promise<void> {
|
||||
if (!this.store.fts5Available) {
|
||||
log.log('Maintenance batch 1 step "fts-maintenance" skipped — FTS5 unavailable');
|
||||
return;
|
||||
@@ -8881,6 +8899,68 @@ export class SelfHealingManager {
|
||||
}
|
||||
}
|
||||
|
||||
private async maintainArchiveTaskFts(): Promise<void> {
|
||||
if (!this.store.archiveFts5Available) {
|
||||
log.log('Maintenance batch 1 step "fts-maintenance" archive skipped — FTS5 unavailable');
|
||||
return;
|
||||
}
|
||||
|
||||
const bytesBefore = this.store.getArchiveFtsIndexBytes();
|
||||
if (bytesBefore === null) {
|
||||
log.log('Maintenance batch 1 step "fts-maintenance" archive skipped — FTS shadow tables unavailable');
|
||||
return;
|
||||
}
|
||||
|
||||
const rowCount = this.store.getArchivedRowCount();
|
||||
const relativeThresholdBytes = rowCount > 0 ? rowCount * ARCHIVE_FTS_REBUILD_BYTES_PER_TASK : null;
|
||||
const shouldRebuild = bytesBefore >= ARCHIVE_FTS_REBUILD_THRESHOLD_BYTES
|
||||
|| (relativeThresholdBytes !== null && bytesBefore > relativeThresholdBytes);
|
||||
const shouldOptimize = !shouldRebuild
|
||||
&& ARCHIVE_FTS_MAINTENANCE_OPTIMIZE_CADENCE_TICKS > 0
|
||||
&& this.maintenanceTickCounter % ARCHIVE_FTS_MAINTENANCE_OPTIMIZE_CADENCE_TICKS === 0;
|
||||
const mode = shouldRebuild ? "rebuild" : shouldOptimize ? "optimize" : "merge";
|
||||
|
||||
if (mode === "merge"
|
||||
&& ARCHIVE_FTS_MAINTENANCE_MERGE_CADENCE_TICKS > 1
|
||||
&& this.maintenanceTickCounter % ARCHIVE_FTS_MAINTENANCE_MERGE_CADENCE_TICKS !== 0) {
|
||||
log.log('Maintenance batch 1 step "fts-maintenance" archive skipped — merge cadence not due');
|
||||
return;
|
||||
}
|
||||
|
||||
let rebuilt = false;
|
||||
if (mode === "rebuild") {
|
||||
rebuilt = this.store.rebuildArchiveFts5Index();
|
||||
} else {
|
||||
this.store.optimizeArchiveFts5(mode);
|
||||
}
|
||||
|
||||
const bytesAfter = this.store.getArchiveFtsIndexBytes();
|
||||
log.log(`Maintenance batch 1 step "fts-maintenance" archive ${mode}: ${bytesBefore} → ${bytesAfter ?? "unknown"} bytes (archived=${rowCount})`);
|
||||
|
||||
try {
|
||||
await createRunAuditor(this.store, {
|
||||
runId: generateSyntheticRunId("self-heal-fts-maintenance", "archived_tasks_fts"),
|
||||
agentId: "self-healing",
|
||||
phase: "maintenance-fts",
|
||||
}).database({
|
||||
type: "task:fts-maintenance" as DatabaseMutationType,
|
||||
target: "archived_tasks_fts",
|
||||
metadata: {
|
||||
mode,
|
||||
bytesBefore,
|
||||
bytesAfter,
|
||||
rowCount,
|
||||
rebuilt,
|
||||
absoluteThresholdBytes: ARCHIVE_FTS_REBUILD_THRESHOLD_BYTES,
|
||||
relativeThresholdBytes,
|
||||
},
|
||||
});
|
||||
} catch (err: unknown) {
|
||||
const errorMessage = err instanceof Error ? err.message : String(err);
|
||||
log.warn(`Failed to write archived task:fts-maintenance run-audit event: ${errorMessage}`);
|
||||
}
|
||||
}
|
||||
|
||||
/** Run a best-effort passive WAL checkpoint without forcing live writers to truncate. */
|
||||
private checkpointWal(): void {
|
||||
try {
|
||||
|
||||
Reference in New Issue
Block a user