diff --git a/packages/durably/src/durably.ts b/packages/durably/src/durably.ts index 8b58edb..63d3212 100644 --- a/packages/durably/src/durably.ts +++ b/packages/durably/src/durably.ts @@ -1,5 +1,5 @@ import type { Dialect } from 'kysely' -import { Kysely } from 'kysely' +import { Kysely, sql } from 'kysely' import { monotonicFactory } from 'ulidx' import type { z } from 'zod' import type { JobDefinition } from './define-job' @@ -134,6 +134,7 @@ function parseDuration(value: string): number { } const PURGE_INTERVAL_MS = 60_000 +const CHECKPOINT_INTERVAL_MS = 60_000 /** Real wall-clock delegating to globalThis timers. */ const realClock: RuntimeClock = { @@ -462,6 +463,8 @@ interface DurablyState< preserveSteps: boolean migrating: Promise | null migrated: boolean + walCheckpointSupported: boolean + probeWalCheckpoint: (() => Promise) | null leaseMs: number leaseRenewIntervalMs: number pollingIntervalMs: number @@ -885,8 +888,10 @@ function createDurablyInstance< } state.migrating = runMigrations(db) - .then(() => { + .then(async () => { state.migrated = true + await state.probeWalCheckpoint?.() + state.probeWalCheckpoint = null }) .finally(() => { state.migrating = null @@ -971,20 +976,44 @@ export function createDurably< const eventEmitter = createEventEmitter() const jobRegistry = createJobRegistry() let lastPurgeAt = 0 + let lastCheckpointAt = 0 const runIdleMaintenance = async (): Promise => { try { - const now = new Date().toISOString() - await storage.releaseExpiredLeases(now) + const nowMs = Date.now() + await storage.releaseExpiredLeases(new Date(nowMs).toISOString()) if (config.retainRunsMs !== null) { - const purgeNow = Date.now() - if (purgeNow - lastPurgeAt >= PURGE_INTERVAL_MS) { - lastPurgeAt = purgeNow - const cutoff = new Date(purgeNow - config.retainRunsMs).toISOString() + if (nowMs - lastPurgeAt >= PURGE_INTERVAL_MS) { + lastPurgeAt = nowMs + const cutoff = new Date(nowMs - config.retainRunsMs).toISOString() await storage.purgeRuns({ olderThan: cutoff, limit: 100 }) } } + + if (state.walCheckpointSupported) { + if (nowMs - lastCheckpointAt >= CHECKPOINT_INTERVAL_MS) { + lastCheckpointAt = nowMs + try { + const result = await sql`PRAGMA wal_checkpoint(TRUNCATE)`.execute( + db, + ) + const row = result.rows[0] as + | { busy: number; log: number; checkpointed: number } + | undefined + if (row?.busy !== 0) { + // Retry sooner on next idle cycle (but not immediately) + lastCheckpointAt = nowMs - CHECKPOINT_INTERVAL_MS / 2 + } + } catch (checkpointError) { + eventEmitter.emit({ + type: 'worker:error', + error: getErrorMessage(checkpointError), + context: 'wal-checkpoint', + }) + } + } + } } catch (error) { eventEmitter.emit({ type: 'worker:error', @@ -1021,6 +1050,8 @@ export function createDurably< preserveSteps: config.preserveSteps, migrating: null, migrated: false, + walCheckpointSupported: false, + probeWalCheckpoint: null, leaseMs: config.leaseMs, leaseRenewIntervalMs: config.leaseRenewIntervalMs, pollingIntervalMs: config.pollingIntervalMs, @@ -1029,6 +1060,23 @@ export function createDurably< runIdleMaintenance, } + if (backend === 'generic' && !isBrowserLikeEnvironment()) { + state.probeWalCheckpoint = async () => { + try { + const result = await sql`PRAGMA wal_checkpoint(PASSIVE)`.execute(db) + const row = result.rows[0] as + | { busy: number; log: number; checkpointed: number } + | undefined + // log === -1 means WAL is not active (e.g. libSQL local uses DELETE mode) + if (row && row.log !== -1) { + state.walCheckpointSupported = true + } + } catch { + // Remote backends (e.g. Turso) reject checkpoint pragmas + } + } + } + const instance = createDurablyInstance, TLabels>( state, {},