fix(core): bound revision cleanup work

.changeset/fix-cron-revision-row-reads.md

@@ -0,0 +1,5 @@
+---
+"emdash": patch
+---
+
+Fixes scheduled maintenance reading the entire revision history on every cron invocation.

packages/core/src/after.ts

@@ -16,6 +16,9 @@
  * `after()`, which they don't during type-checking.
  */
 
+import { trackDeferredTask } from "./deferred-tasks.js";
+import { getRequestContext } from "./request-context.js";
+
 export type WaitUntilFn = (promise: Promise<unknown>) => void;
 
 // Resolves to the host's waitUntil if the adapter provided one, or
@@ -45,11 +48,13 @@ waitUntilReady.catch(() => {});
  * that care about errors should handle them inside `fn`.
  */
 export function after(fn: () => void | Promise<void>): void {
-	const promise = Promise.resolve()
+	const task = Promise.resolve()
 		.then(fn)
 		.catch((error) => {
 			console.error("[emdash] deferred task failed:", error);
 		});
+	const requestTask = getRequestContext()?.deferredTasks?.track(task) ?? task;
+	const promise = trackDeferredTask(requestTask);
 
 	// Defer the lifetime-extender handoff to the microtask that resolves
 	// waitUntilReady. On workerd this is effectively instant (the virtual

packages/core/src/api/handlers/revision.ts

@@ -4,6 +4,7 @@
 
 import type { Kysely } from "kysely";
 
+import { after } from "../../after.js";
 import { ContentRepository } from "../../database/repositories/content.js";
 import { RevisionRepository, type Revision } from "../../database/repositories/revision.js";
 import { withTransaction } from "../../database/transaction.js";
@@ -115,7 +116,7 @@ export async function handleRevisionRestore(
 		// Atomically update content and create a new revision to record the restore.
 		// If either operation fails, neither is committed (on engines that support
 		// transactions; on D1, withTransaction falls back to sequential execution).
-		const item = await withTransaction(db, async (trx) => {
+		const { item, queuedRevisionId } = await withTransaction(db, async (trx) => {
 			const trxContentRepo = new ContentRepository(trx);
 			const trxRevisionRepo = new RevisionRepository(trx);
 
@@ -124,19 +125,32 @@ export async function handleRevisionRestore(
 				slug: typeof _slug === "string" ? _slug : undefined,
 			});
 
-			await trxRevisionRepo.create({
+			const queuedRevision = await trxRevisionRepo.create({
 				collection: revision.collection,
 				entryId: revision.entryId,
 				data: revision.data,
 				authorId: callerUserId,
 			});
 
-			return updated;
+			return { item: updated, queuedRevisionId: queuedRevision.id };
 		});
 
-		// Fire-and-forget: prune old revisions to prevent unbounded growth
 		const pruneRepo = new RevisionRepository(db);
-		void pruneRepo.pruneOldRevisions(revision.collection, revision.entryId, 50).catch(() => {});
+		after(async () => {
+			try {
+				await pruneRepo.pruneQueuedEntry(
+					revision.collection,
+					revision.entryId,
+					queuedRevisionId,
+					50,
+				);
+			} catch (error) {
+				console.error(
+					`[revisions] Failed to prune revisions for ${revision.collection}/${revision.entryId}:`,
+					error,
+				);
+			}
+		});
 
 		return {
 			success: true,

packages/core/src/astro/middleware.ts

@@ -63,7 +63,7 @@ import { createInitLock, type InitLock, initWithLock } from "../utils/init-lock.
 import type { EmDashConfig } from "./integration/runtime.js";
 import {
 	ASTRO_COOKIES_SYMBOL,
-	deferScopedCloseUntilSettled,
+	coordinateScopedDbLifecycle,
 	finishScoped,
 } from "./middleware/scoped-db.js";
 import { wrapBodyForStreamMetrics } from "./middleware/stream-end-metrics.js";
@@ -366,26 +366,33 @@ async function runOutsideRequest<T>(
 		// outside a request. Any close-less scope created above is discarded.
 		return fn(runtime);
 	}
+	const { closed, deferredTasks, lifecycle } = coordinateScopedDbLifecycle(scoped);
 
 	const parent = getRequestContext();
 	const ctx = parent
-		? { ...parent, db: scoped.db }
-		: { editMode: false, db: scoped.db, metrics: createRequestMetrics(performance.now()) };
+		? { ...parent, db: scoped.db, deferredTasks }
+		: {
+				editMode: false,
+				db: scoped.db,
+				metrics: createRequestMetrics(performance.now()),
+				deferredTasks,
+			};
 	try {
 		return await runWithContext(ctx, () => fn(runtime));
 	} finally {
 		// Guard both so a throw in teardown can't mask the callback result or
-		// skip close() and leak the connection. Mirrors closeSafely() in scoped-db.ts.
+		// skip lifecycle settlement and leak the connection.
 		try {
-			scoped.commit();
+			lifecycle.commit();
 		} catch (error) {
 			console.error("[emdash] event-scoped db commit failed:", error);
 		}
 		try {
-			scoped.close();
+			lifecycle.close?.();
 		} catch (error) {
 			console.error("[emdash] event-scoped db close failed:", error);
 		}
+		await closed;
 	}
 }
 
@@ -687,10 +694,11 @@ export const onRequest = defineMiddleware(async (context, next) => {
 					return wrapBodyForStreamMetrics(finalizeResponse(response, timings));
 				};
 				if (anonScoped) {
+					const { deferredTasks, lifecycle } = coordinateScopedDbLifecycle(anonScoped);
 					const parent = getRequestContext();
 					const ctx = parent
-						? { ...parent, db: anonScoped.db }
-						: { editMode: false, db: anonScoped.db, metrics };
+						? { ...parent, db: anonScoped.db, deferredTasks }
+						: { editMode: false, db: anonScoped.db, metrics, deferredTasks };
 					// Eagerly warm site-global layout data (menus, widget areas,
 					// taxonomy terms, settings) concurrently so the layout's
 					// per-component reads overlap into ~one wall-clock round trip and
@@ -710,13 +718,11 @@ export const onRequest = defineMiddleware(async (context, next) => {
 						.trim()
 						.startsWith("text/html");
 					return runWithContext(ctx, async () => {
-						const scopedLifecycle = acceptsHtml
-							? deferScopedCloseUntilSettled(anonScoped, prefetchLayoutData(), after)
-							: anonScoped;
+						if (acceptsHtml) after(() => prefetchLayoutData());
 						// commit() persists per-request state (e.g. the D1 bookmark cookie)
 						// before the response is returned, even if render throws; close()
-						// (connection teardown) is deferred to stream-end. See finishScoped.
-						return finishScoped(scopedLifecycle, runAnon);
+						// waits for stream-end and request-owned deferred work. See finishScoped.
+						return finishScoped(lifecycle, runAnon);
 					});
 				}
 				return runAnon();
@@ -903,15 +909,16 @@ export const onRequest = defineMiddleware(async (context, next) => {
 			};
 
 			if (scoped) {
+				const { deferredTasks, lifecycle } = coordinateScopedDbLifecycle(scoped);
 				const parent = getRequestContext();
 				const ctx = parent
-					? { ...parent, db: scoped.db }
-					: { editMode: false, db: scoped.db, metrics };
+					? { ...parent, db: scoped.db, deferredTasks }
+					: { editMode: false, db: scoped.db, metrics, deferredTasks };
 				return runWithContext(ctx, () =>
 					// commit() persists per-request state (e.g. the D1 bookmark cookie)
 					// before the response returns, even if render throws; close()
-					// (connection teardown) is deferred to stream-end. See finishScoped.
-					finishScoped(scoped, renderAndFinalize),
+					// waits for stream-end and request-owned deferred work. See finishScoped.
+					finishScoped(lifecycle, renderAndFinalize),
 				);
 			}
 

packages/core/src/astro/middleware/scoped-db.ts

@@ -6,6 +6,9 @@
  * request-scoped db adapter's lifecycle around the response.
  */
 
+import { createDeferredTaskTracker } from "../../deferred-tasks.js";
+import type { DeferredTaskTracker } from "../../deferred-tasks.js";
+
 /**
  * Astro attaches AstroCookies to outgoing responses via a well-known global
  * symbol. Cloning a Response (`new Response(body, init)`) drops non-header
@@ -21,36 +24,27 @@ interface ScopedDbLifecycle {
 	close?: () => void;
 }
 
-type DeferredTaskScheduler = (task: () => void | Promise<void>) => void;
-
-/**
- * Keep teardown alive while both the response and request-owned background
- * work finish. The returned close callback only signals response completion;
- * the adapter's real close runs afterward inside the deferred task.
- */
-export function deferScopedCloseUntilSettled(
-	scoped: ScopedDbLifecycle,
-	pending: Promise<unknown>,
-	defer: DeferredTaskScheduler,
-): ScopedDbLifecycle {
-	if (!scoped.close) {
-		defer(async () => {
-			await pending;
-		});
-		return scoped;
-	}
+/** Hold the real adapter close behind both response and deferred-task completion. */
+export function coordinateScopedDbLifecycle(scoped: ScopedDbLifecycle): {
+	closed?: Promise<void>;
+	deferredTasks?: DeferredTaskTracker;
+	lifecycle: ScopedDbLifecycle;
+} {
+	if (!scoped.close) return { lifecycle: scoped };
 
-	let settleResponse!: () => void;
-	const responseSettled = new Promise<void>((resolve) => {
-		settleResponse = resolve;
-	});
 	const close = scoped.close;
-	defer(async () => {
-		await Promise.allSettled([pending, responseSettled]);
-		close();
+	const deferredTasks = createDeferredTaskTracker(() => {
+		try {
+			close();
+		} catch (error) {
+			console.error("[emdash] request-scoped db close failed:", error);
+		}
 	});
-
-	return { commit: scoped.commit, close: settleResponse };
+	return {
+		closed: deferredTasks.settled,
+		deferredTasks,
+		lifecycle: { commit: scoped.commit, close: deferredTasks.settle },
+	};
 }
 
 /**
@@ -99,9 +93,9 @@ export function wrapResponseForScopedClose(response: Response, close: () => void
  * Run the request body under a request-scoped db, then settle its lifecycle:
  * `commit()` runs before the response is returned (so per-request state like a
  * D1 bookmark cookie is persisted in the headers, even if render throws), while
- * `close()` (if any) is deferred to stream-end so a connection-backed adapter
- * isn't torn down mid-render. On error the connection is closed immediately
- * before rethrowing so it never leaks.
+ * `close()` (if any) is deferred to lifecycle settlement so a
+ * connection-backed adapter isn't torn down mid-render or mid-task. On error
+ * the lifecycle is settled before rethrowing so it never leaks.
  *
  * On the error path both `commit()` and `close()` are defended: a throw from
  * either is logged and swallowed so it can't replace the propagating render

packages/core/src/cleanup.ts

@@ -10,7 +10,7 @@
  */
 
 import { createKyselyAdapter, type AuthTables } from "@emdash-cms/auth/adapters/kysely";
-import { sql, type Kysely } from "kysely";
+import type { Kysely } from "kysely";
 
 import { cleanupExpiredChallenges } from "./auth/challenge-store.js";
 import { MediaRepository } from "./database/repositories/media.js";
@@ -36,17 +36,14 @@ export interface CleanupResult {
 	mediaUsageOrphanOccurrences: number;
 }
 
-/** Max revisions to keep per entry during periodic pruning */
 const REVISION_KEEP_COUNT = 50;
-
-/** Only prune entries that exceed this threshold */
-const REVISION_PRUNE_THRESHOLD = REVISION_KEEP_COUNT;
+const REVISION_PRUNE_BATCH_SIZE = 10;
 
 /**
  * Run all system cleanup tasks.
  *
- * Safe to call frequently -- each task is a single DELETE with a WHERE clause,
- * so repeated calls with nothing to clean are cheap (no-op queries).
+ * Safe to call frequently -- each subsystem tolerates repeated calls, and
+ * repeated calls with nothing to clean are cheap.
  *
  * @param db - The database instance
  * @param storage - Optional storage backend for deleting orphaned files.
@@ -136,9 +133,8 @@ export async function runSystemCleanup(
 		console.error("[cleanup] Failed to clean media upload attempts:", error);
 	}
 
-	// 5. Revision pruning -- trim entries with excessive revision counts
 	try {
-		result.revisionsPruned = await pruneExcessiveRevisions(db);
+		result.revisionsPruned = await pruneQueuedRevisions(db);
 	} catch (error) {
 		console.error("[cleanup] Failed to prune revisions:", error);
 	}
@@ -155,31 +151,24 @@ export async function runSystemCleanup(
 	return result;
 }
 
-/**
- * Find entries with more than REVISION_PRUNE_THRESHOLD revisions and prune
- * them down to REVISION_KEEP_COUNT.
- */
-async function pruneExcessiveRevisions(db: Kysely<Database>): Promise<number> {
-	const entries = await sql<{ collection: string; entry_id: string }>`
-		SELECT collection, entry_id
-		FROM revisions
-		GROUP BY collection, entry_id
-		HAVING COUNT(*) > ${REVISION_PRUNE_THRESHOLD}
-	`.execute(db);
-
-	if (entries.rows.length === 0) return 0;
-
+async function pruneQueuedRevisions(db: Kysely<Database>): Promise<number> {
+	const queued = await db
+		.selectFrom("_emdash_revision_prune_queue")
+		.selectAll()
+		.orderBy("revision_id")
+		.limit(REVISION_PRUNE_BATCH_SIZE)
+		.execute();
 	const revisionRepo = new RevisionRepository(db);
 	let totalPruned = 0;
 
-	for (const row of entries.rows) {
+	for (const row of queued) {
 		try {
-			const pruned = await revisionRepo.pruneOldRevisions(
+			totalPruned += await revisionRepo.pruneQueuedEntry(
 				row.collection,
 				row.entry_id,
+				row.revision_id,
 				REVISION_KEEP_COUNT,
 			);
-			totalPruned += pruned;
 		} catch (error) {
 			console.error(
 				`[cleanup] Failed to prune revisions for ${row.collection}/${row.entry_id}:`,

packages/core/src/database/migrations/059_revision_prune_queue.ts

@@ -0,0 +1,36 @@
+import { sql, type Kysely } from "kysely";
+
+const REVISION_KEEP_COUNT = 50;
+
+export async function up(db: Kysely<unknown>): Promise<void> {
+	await db.schema
+		.createTable("_emdash_revision_prune_queue")
+		.ifNotExists()
+		.addColumn("collection", "text", (column) => column.notNull())
+		.addColumn("entry_id", "text", (column) => column.notNull())
+		.addColumn("revision_id", "text", (column) => column.notNull())
+		.addPrimaryKeyConstraint("revision_prune_queue_pk", ["collection", "entry_id"])
+		.execute();
+
+	await db.schema
+		.createIndex("idx_revision_prune_queue_revision_id")
+		.ifNotExists()
+		.on("_emdash_revision_prune_queue")
+		.column("revision_id")
+		.execute();
+
+	await sql`
+		INSERT INTO _emdash_revision_prune_queue (collection, entry_id, revision_id)
+		SELECT collection, entry_id, MAX(id)
+		FROM revisions
+		WHERE true
+		GROUP BY collection, entry_id
+		HAVING COUNT(*) > ${REVISION_KEEP_COUNT}
+		ON CONFLICT (collection, entry_id)
+		DO UPDATE SET revision_id = excluded.revision_id
+	`.execute(db);
+}
+
+export async function down(db: Kysely<unknown>): Promise<void> {
+	await db.schema.dropTable("_emdash_revision_prune_queue").ifExists().execute();
+}

packages/core/src/database/migrations/runner.ts

@@ -61,6 +61,7 @@ import * as m055 from "./055_content_translation_group_locale_index.js";
 import * as m056 from "./056_taxonomy_term_sort_order.js";
 import * as m057 from "./057_collection_hidden.js";
 import * as m058 from "./058_collection_sort_order.js";
+import * as m059 from "./059_revision_prune_queue.js";
 
 const MIGRATIONS: Readonly<Record<string, Migration>> = Object.freeze({
 	"001_initial": m001,
@@ -120,6 +121,7 @@ const MIGRATIONS: Readonly<Record<string, Migration>> = Object.freeze({
 	"056_taxonomy_term_sort_order": m056,
 	"057_collection_hidden": m057,
 	"058_collection_sort_order": m058,
+	"059_revision_prune_queue": m059,
 });
 
 /** Total number of registered migrations. Exported for use in tests. */