Fix RocksDB @expiresAt eviction sweeps

DESIGN.md

@@ -53,6 +53,8 @@ The mitigations live in three places:
 
 When adding a new commit-handler early-return path: reset `write.skipped = false` at the top of the handler if you don't already, then set `write.skipped = true` immediately before the `return`. Decide first whether the audit log will reference the blob (via `auditRecordToStore`) — if it does, leave `skipped` unset. `cleanupOrphans` is the periodic safety net; don't rely on it for transactional correctness.
 
+RocksDB eviction has the inverse ordering constraint: removing a record that already references a blob must not unlink the file until the raw transaction commits. A transactional `store.remove()` is synchronous because it only stages the delete, so `removeEntry()` cannot infer durability from its return value on this path. `TableResource.evict()` therefore carries the blob-bearing record through the commit and calls `deleteBlobsInObject()` only from the commit-success handler. Shared eviction batches must continue excluding blob-bearing records because one commit covers many staged removals and can conflict as a unit.
+
 **Source-unavailable blobs must not abort the commit.** `startPreCommitBlobsForRecord().complete()` awaits each blob's `saving` promise; a rejection there propagates up and aborts the record's apply (the replication subscription loop catches and logs it as `error in subscription handler`). For a blob the replication source can no longer provide — evicted/expired at the origin, the receiver having flagged the rejection `sourceBlobUnavailable` (harper-pro#403) — that abort permanently wedged a replication copy stream on an expiration cache table whose TTL-evicted blobs are gone everywhere: every orphaned record's apply re-threw, the copy never advanced, and backpressure pinned at ~100%. `complete()` therefore tolerates a `sourceBlobUnavailable` rejection (`isSourceBlobUnavailable`): the record commits with a diverged blob reference, left for proactive backfill (harper-pro#388). Local/transient save faults stay unmarked and still reject, so the write aborts and a reconnect retries it — no silent loss. This is the apply/commit-side complement to the replication receiver's resume-cursor advance (harper-pro#403/#405), which handles the durability-watermark side of the same missing blob.
 
 ## Over-time transactions are aborted, not force-committed (`DatabaseTransaction`/`LMDBTransaction`)

dataLayer/harperBridge/ResourceBridge.ts

@@ -186,7 +186,6 @@ export class ResourceBridge extends BridgeMethods {
 
 	async dropSchema(dropSchemaObj) {
 		await dropDatabase(dropSchemaObj.schema);
-		signalling.signalSchemaChange(new SchemaEventMsg(process.pid, OPERATIONS_ENUM.DROP_SCHEMA, dropSchemaObj.schema));
 	}
 
 	async updateRecords(updateObj) {

integrationTests/database/expiration-drop-quiesce.test.ts

@@ -0,0 +1,135 @@
+import { after, before, suite, test } from 'node:test';
+import { strictEqual } from 'node:assert';
+import { existsSync, mkdtempSync, readFileSync, writeFileSync } from 'node:fs';
+import { tmpdir } from 'node:os';
+import { join, resolve } from 'node:path';
+import { setTimeout as delay } from 'node:timers/promises';
+import { setupHarperWithFixture, teardownHarper, type ContextWithHarper } from '@harperfast/integration-testing';
+import request from 'supertest';
+import { createApiClient } from '../apiTests/utils/client.mjs';
+
+const FIXTURE_PATH = resolve(import.meta.dirname, 'expiration-drop-quiesce');
+const CONTROL_DIRECTORY = mkdtempSync(join(tmpdir(), 'expiration-drop-quiesce-'));
+const skipSuite = process.platform === 'win32'; // Windows cannot expose two HTTP workers on one port.
+
+async function waitFor(predicate: () => boolean, message: string) {
+	const deadline = Date.now() + 30_000;
+	while (Date.now() < deadline) {
+		if (predicate()) return;
+		await delay(20);
+	}
+	throw new Error(message);
+}
+
+suite('cross-worker expiration cleanup quiesces destructive DDL', { skip: skipSuite }, (ctx: ContextWithHarper) => {
+	let client: ReturnType<typeof createApiClient>;
+	let workerIds: number[];
+
+	before(async () => {
+		await setupHarperWithFixture(ctx, FIXTURE_PATH, {
+			config: { threads: { count: 2 }, logging: { level: 'error' } },
+			env: { EXPIRATION_QUIESCE_CONTROL: CONTROL_DIRECTORY },
+		} as any);
+		client = createApiClient(ctx.harper);
+		let ready = false;
+		const deadline = Date.now() + 120_000;
+		while (Date.now() < deadline) {
+			try {
+				const response = await request(client.restURL)
+					.post('/QuiesceControl/')
+					.set(client.headers)
+					.timeout(2_000)
+					.send({ action: 'probe' });
+				if (response.status !== 404) {
+					ready = true;
+					break;
+				}
+			} catch {
+				// Workers are still loading the pre-installed component.
+			}
+			await delay(250);
+		}
+		strictEqual(ready, true, 'QuiesceControl resource did not become ready');
+		const probe = await postControl({ action: 'probe' });
+		workerIds = probe.body.workerIds;
+		strictEqual(workerIds.length, 2, 'the fixture must expose both HTTP workers for deterministic pinning');
+	});
+
+	after(async () => {
+		await teardownHarper(ctx);
+	});
+
+	async function postControl(body: Record<string, unknown>) {
+		return request(client.restURL).post('/QuiesceControl/').set(client.headers).timeout(120_000).send(body);
+	}
+
+	async function postControlOnWorker(body: Record<string, unknown>, targetThreadId: number) {
+		const response = await postControl({ ...body, targetThreadId });
+		strictEqual(response.body?.threadId, targetThreadId, `control action did not execute on worker ${targetThreadId}`);
+		return response;
+	}
+
+	async function proveDropWaitsForWorker(kind: 'indexed' | 'primary') {
+		const indexed = kind === 'indexed';
+		const database = indexed ? 'quiesce_indexed' : 'quiesce_primary';
+		const table = indexed ? 'IndexedExpiry' : 'PrimaryExpiry';
+		const runId = `${kind}-${Date.now()}`;
+		const [sweepWorkerId, ddlWorkerId] = workerIds;
+		await postControl({ action: 'seed', kind, database, table, id: runId }).then((response) =>
+			strictEqual(response.status, 200)
+		);
+		const sweepRunId = `${runId}-sweep`;
+		const dropRunId = `${runId}-drop`;
+		const started = join(CONTROL_DIRECTORY, `${sweepRunId}.started`);
+		const release = join(CONTROL_DIRECTORY, `${runId}.release`);
+		const sweep = postControlOnWorker(
+			{ action: 'sweep', kind, database, table, runId: sweepRunId, releaseRunId: runId },
+			sweepWorkerId
+		);
+		let dropSettled = false;
+		const dropStarted = join(CONTROL_DIRECTORY, `${dropRunId}.started`);
+		let drop: ReturnType<typeof postControlOnWorker> | undefined;
+		let dropped;
+		let primaryError: unknown;
+		const cleanupErrors: unknown[] = [];
+		try {
+			await waitFor(() => existsSync(started), `${kind} sweep did not reach its blocked commit`);
+			strictEqual(JSON.parse(readFileSync(started, 'utf8')).threadId, sweepWorkerId);
+			drop = postControlOnWorker({ action: 'drop', database, table, runId: dropRunId }, ddlWorkerId).then(
+				(response) => {
+					dropSettled = true;
+					return response;
+				}
+			);
+			await waitFor(() => existsSync(dropStarted), `${kind} drop did not start on its pinned worker`);
+			strictEqual(JSON.parse(readFileSync(dropStarted, 'utf8')).threadId, ddlWorkerId);
+			strictEqual(sweepWorkerId === ddlWorkerId, false, 'DDL and cleanup must execute on different workers');
+			await delay(100);
+			strictEqual(dropSettled, false, 'physical drop must wait for the blocked cleanup worker');
+		} catch (error) {
+			primaryError = error;
+		} finally {
+			try {
+				writeFileSync(release, 'release');
+			} catch (error) {
+				cleanupErrors.push(error);
+			}
+			const completions = drop ? [drop, sweep] : [sweep];
+			const results = await Promise.allSettled(completions);
+			if (drop && results[0].status === 'fulfilled') dropped = results[0].value;
+			for (const result of results) if (result.status === 'rejected') cleanupErrors.push(result.reason);
+		}
+		if (primaryError) throw primaryError;
+		if (cleanupErrors.length) throw new AggregateError(cleanupErrors, `${kind} quiescence test cleanup failed`);
+		strictEqual(dropped.status, 200);
+		strictEqual(dropped.body.threadId, ddlWorkerId);
+	}
+
+	test('drop_table waits for a remote indexed expiration sweep', async () => {
+		await proveDropWaitsForWorker('indexed');
+	});
+
+	test('drop_table waits for a remote primary cleanup scan', async () => {
+		await proveDropWaitsForWorker('primary');
+	});
+});

integrationTests/database/expiration-drop-quiesce/config.yaml

@@ -0,0 +1,5 @@
+graphqlSchema:
+  files: '*.graphql'
+jsResource:
+  files: resources.js
+rest: true

integrationTests/database/expiration-drop-quiesce/resources.js

@@ -0,0 +1,105 @@
+import { existsSync, renameSync, writeFileSync } from 'node:fs';
+import { join } from 'node:path';
+import { setTimeout as delay } from 'node:timers/promises';
+import { threadId } from 'node:worker_threads';
+
+const controlDirectory = process.env.EXPIRATION_QUIESCE_CONTROL;
+const CONTROL_REQUEST = 'expiration-quiesce-control-request';
+const CONTROL_RESPONSE = 'expiration-quiesce-control-response';
+const pendingRequests = new Map();
+let nextRequestId = 1;
+
+function publishWorkerMarker(path) {
+	const temporaryPath = `${path}.${threadId}.tmp`;
+	writeFileSync(temporaryPath, JSON.stringify({ threadId }));
+	renameSync(temporaryPath, path);
+}
+
+async function performControl(body) {
+	const Table = databases[body.database]?.[body.table];
+	if (!Table) return { available: false, threadId };
+	if (body.action === 'seed') {
+		const expiresAt = Date.now() - 10_000;
+		if (body.kind === 'indexed') await Table.put(body.id, { id: body.id, expiresAt });
+		else await Table.put(body.id, { id: body.id, value: 'expired' }, { expiresAt });
+		return { seeded: true, threadId };
+	}
+	const started = join(controlDirectory, `${body.runId}.started`);
+	const release = join(controlDirectory, `${body.releaseRunId ?? body.runId}.release`);
+	if (body.action === 'drop') {
+		publishWorkerMarker(started);
+		await Table.dropTable();
+		return { completed: true, threadId };
+	}
+	const hooks = {
+		beforeBatchCommit: async () => {
+			publishWorkerMarker(started);
+			while (!existsSync(release)) await delay(20);
+		},
+	};
+	if (body.kind === 'indexed') await Table.runRecordExpirationSweepForTests(hooks);
+	else await Table.runPrimaryCleanupScanForTests(hooks);
+	return { completed: true, threadId };
+}
+
+threads.onMessageByType(CONTROL_REQUEST, async (message) => {
+	if (message.targetThreadId !== threadId) return;
+	try {
+		const result = await performControl(message.body);
+		threads.sendToThread(message.originThreadId, { type: CONTROL_RESPONSE, requestId: message.requestId, result });
+	} catch (error) {
+		threads.sendToThread(message.originThreadId, {
+			type: CONTROL_RESPONSE,
+			requestId: message.requestId,
+			error: error?.stack ?? error?.message ?? String(error),
+		});
+	}
+});
+
+threads.onMessageByType(CONTROL_RESPONSE, (message) => {
+	const pending = pendingRequests.get(message.requestId);
+	if (!pending) return;
+	pendingRequests.delete(message.requestId);
+	clearTimeout(pending.timer);
+	if (message.error) pending.reject(new Error(message.error));
+	else pending.resolve(message.result);
+});
+
+function performControlOnWorker(body, targetThreadId) {
+	if (targetThreadId === threadId) return performControl(body);
+	return new Promise((resolve, reject) => {
+		const requestId = `${threadId}:${nextRequestId++}`;
+		const timer = setTimeout(() => {
+			pendingRequests.delete(requestId);
+			reject(new Error(`Timed out dispatching expiration quiesce control to worker ${targetThreadId}`));
+		}, 120_000);
+		timer.unref();
+		pendingRequests.set(requestId, { resolve, reject, timer });
+		if (
+			!threads.sendToThread(targetThreadId, {
+				type: CONTROL_REQUEST,
+				requestId,
+				originThreadId: threadId,
+				targetThreadId,
+				body,
+			})
+		) {
+			clearTimeout(timer);
+			pendingRequests.delete(requestId);
+			reject(new Error(`Worker ${targetThreadId} is not reachable`));
+		}
+	});
+}
+
+export class QuiesceControl extends Resource {
+	static loadAsInstance = false;
+
+	async post(_query, body) {
+		if (body.action === 'probe') {
+			const workerIds = [threadId, ...threads.map((port) => port.threadId)].filter((id) => id > 0);
+			return { ready: true, threadId, workerIds: [...new Set(workerIds)] };
+		}
+		if (body.targetThreadId !== undefined) return performControlOnWorker(body, body.targetThreadId);
+		return performControl(body);
+	}
+}

integrationTests/database/expiration-drop-quiesce/schema.graphql

@@ -0,0 +1,9 @@
+type IndexedExpiry @table(database: "quiesce_indexed") @export {
+	id: ID! @primaryKey
+	expiresAt: Float @expiresAt
+}
+
+type PrimaryExpiry @table(database: "quiesce_primary", expiration: 1, eviction: 1) @export {
+	id: ID! @primaryKey
+	value: String
+}

resources/DESIGN.md

@@ -105,6 +105,36 @@ One giant `makeTable()` factory that returns a `TableResource extends Resource`
 
 ---
 
+## Record expiration sweeps
+
+An `@expiresAt` attribute is a public timestamp field backed by the record encoder's expiration metadata.
+Its index stores one canonical, non-negative epoch-millisecond number derived from that metadata, regardless of
+whether the API value was a number, numeric string, ISO string, or `Date`. Writes, index rebuilds, index searches,
+and scan filters apply the same normalization. `expirationIndexVersion` participates in the normal resumable
+schema reindex flow so stores created before canonicalization are rebuilt without a separate open-time scan.
+Rows created before expiration metadata was stored fall back to their public field during that rebuild and while
+sweeping; an explicit no-expiration metadata sentinel never falls back. This preserves upgrade behavior without
+making every open scan the table.
+
+The expiration index represents the effective stored expiration, not an ordinary field-value index. An explicit
+`options.expiresAt` or `context.expiresAt` override can therefore differ from the serialized field while remaining
+authoritative for index searches and reclamation. Source/cache fills do not infer TTL from the returned field; the
+source must set `sourceContext.expiresAt`, with table expiration providing the fallback.
+
+RocksDB `@expiresAt` sweeps walk one bounded composite-index range at a time with a fixed cutoff and an
+owned `(expiresAt, primaryKey)` cursor. A sweep never holds an iterator snapshot across an `await`, and it
+continues until every index entry at or before that cutoff has been considered. This keeps memory and native
+iterator lifetime bounded without leaving a permanent backlog when more than one chunk expires together.
+
+Dangling-index cleanup reads and writes the primary key in the same RocksDB transaction as the index removal.
+The primary write is a conflict guard against concurrent resurrection: absent keys receive a transactional
+remove, while retained audit tombstones are rewritten byte-for-byte so their version and retention semantics do
+not change. An `ERR_BUSY` abort leaves the newly written record and index authoritative for the next sweep.
+
+Table cleanup is a join point for both the primary reclamation scan and the `@expiresAt` index sweep. Database
+close and drop wait for that join before closing or destroying stores. A timeout fails closed: handles and
+registries remain live and cleanup scheduling is resumed, so active work is never raced by teardown.
+
 ## Path routing & parameterised routes
 
 `Resources.ts` is the registry that maps URL paths to `Resource` classes. Resources are registered (`jsResource.ts`) from a component's exports: