/** * `EngineTimers` — runes-free runtime timer scheduler. * * Race-safety contract (see DESIGN_TIMR.md §0.7 + §12.8): * * Every native callback captures the tuple (entryId, key, version) * when armed. Before mutating state — both before and after running * the user's task — `getLiveEntry(id, key, version)` must agree the * entry is still the one the callback expects. Any mismatch makes * the callback a silent no-op. * * This guards against four classes of race that `clearTimeout` + * `AbortSignal` alone do not cover: * * • timeout fires while cancel/replace/reschedule runs in the same * event-loop tick * • async task resolves after the entry was cancelled or replaced * • interval next tick scheduled after dispose() * • ack-style timeout firing after the reply already arrived * * `AbortSignal` is cooperative — useful for in-flight `fetch` etc. — * but is *not* the primary guard. */ import { createSystemTimerClock } from './clock.ts'; import { disposedErrorMessage, duplicateKeyErrorMessage, ENGINE_METHOD_CANCEL, ENGINE_METHOD_CANCEL_ALL, ENGINE_METHOD_SCHEDULE, ENGINE_METHOD_SCHEDULE_AT, inactiveTimerErrorMessage, invalidDelayErrorMessage, invalidKeyErrorMessage, LOG_MSG_SCHEDULE_AT_PAST, LOGGER_CATEGORY, DEFAULT_TIMER_SCOPE_SEPARATOR, TIMER_EVENT_CANCELLED, TIMER_EVENT_COMPLETED, TIMER_EVENT_DISPOSED, TIMER_EVENT_FAILED, TIMER_EVENT_RUNNING, TIMER_EVENT_SCHEDULED, TIMER_KIND_INTERVAL, TIMER_KIND_TIMEOUT, TIMER_STATUS_CANCELLED, TIMER_STATUS_COMPLETED, TIMER_STATUS_FAILED, TIMER_STATUS_PENDING, TIMER_STATUS_RUNNING, listenerThrewMessage, taskFailedMessage } from './consts.ts'; import { TimrDisposedError, TimrDuplicateKeyError, TimrInactiveTimerError, TimrInvalidDelayError, TimrInvalidKeyError } from './errors.ts'; import type { EngineTimers, EngineTimersOptions, TimerClock, TimerEntrySnapshot, TimerEvent, TimerHandle, TimerIntervalOptions, TimerKind, TimerListener, TimerLogger, TimerNativeHandle, TimerOptions, TimerStatus, TimerTask, TimerTaskContext } from './types.ts'; interface InternalTimerEntry { id: number; key: string; version: number; kind: TimerKind; status: TimerStatus; task: TimerTask; controller: AbortController; native: TimerNativeHandle | null; scheduledAt: number; dueAt: number; delayMs: number; intervalMs: number | null; awaitTask: boolean; maxRuns: number | undefined; runCount: number; lastFiredAt: number | null; lastCompletedAt: number | null; lastErrorAt: number | null; error: unknown | null; removeOnComplete: boolean; meta?: Readonly>; } export function createEngineTimers(options: EngineTimersOptions = {}): EngineTimers { const clock: TimerClock = options.clock ?? createSystemTimerClock(); const logger: TimerLogger | undefined = options.logger; const entries = new Map(); const listeners = new Set(); let disposed = false; let nextId = 1; // ── Validation ────────────────────────────────────────────────────────── function ensureLive(method: string): void { if (disposed) throw new TimrDisposedError(disposedErrorMessage(method)); } function validateKey(key: unknown): void { if (typeof key !== 'string' || key.length === 0) { throw new TimrInvalidKeyError(invalidKeyErrorMessage(key)); } } function validateDelay(delayMs: unknown): void { if (typeof delayMs !== 'number' || !Number.isFinite(delayMs) || delayMs < 0) { throw new TimrInvalidDelayError(invalidDelayErrorMessage(delayMs)); } } // ── Snapshot + scope ──────────────────────────────────────────────────── function scopeOf(key: string): string { const idx = key.lastIndexOf(DEFAULT_TIMER_SCOPE_SEPARATOR); return idx === -1 ? key : key.slice(0, idx); } function snapshotOf(entry: InternalTimerEntry): TimerEntrySnapshot { return { key: entry.key, scope: scopeOf(entry.key), kind: entry.kind, status: entry.status, scheduledAt: entry.scheduledAt, dueAt: entry.dueAt, delayMs: entry.delayMs, runCount: entry.runCount, lastFiredAt: entry.lastFiredAt, lastCompletedAt: entry.lastCompletedAt, lastErrorAt: entry.lastErrorAt, error: entry.error, meta: entry.meta }; } // ── Listener fan-out ──────────────────────────────────────────────────── function emit(event: TimerEvent): void { for (const listener of listeners) { try { listener(event); } catch (err) { logger?.error?.(LOGGER_CATEGORY, listenerThrewMessage(event.type), { error: err }); } } } // ── Entry lifecycle ───────────────────────────────────────────────────── /** * Look up an entry only if the captured tuple still matches. This is * the core of the race-safety contract — every callback uses it. */ function getLiveEntry(id: number, key: string, version: number): InternalTimerEntry | null { const current = entries.get(key); if (current === undefined) return null; if (current.id !== id) return null; if (current.version !== version) return null; if (current.controller.signal.aborted) return null; return current; } /** * Arm the native timeout for the given entry. Captures the tuple in * the closure — `runEntry` will reject the callback if the entry has * been cancelled, replaced or rescheduled in the meantime. */ function armEntry(entry: InternalTimerEntry, delayMs: number): void { entry.version += 1; const id = entry.id; const key = entry.key; const version = entry.version; entry.native = clock.setTimeout(() => { void runEntry(id, key, version); }, delayMs); } function makeContext(entry: InternalTimerEntry, firedAt: number): TimerTaskContext { return { key: entry.key, kind: entry.kind, scheduledAt: entry.scheduledAt, dueAt: entry.dueAt, firedAt, driftMs: firedAt - entry.dueAt, runCount: entry.runCount, signal: entry.controller.signal }; } async function runEntry(id: number, key: string, version: number): Promise { const entry = getLiveEntry(id, key, version); if (entry === null) return; if (entry.status !== TIMER_STATUS_PENDING) return; const firedAt = clock.now(); entry.status = TIMER_STATUS_RUNNING; entry.lastFiredAt = firedAt; entry.runCount += 1; entry.native = null; const ctx = makeContext(entry, firedAt); emit({ type: TIMER_EVENT_RUNNING, entry: snapshotOf(entry) }); const isInterval = entry.kind === TIMER_KIND_INTERVAL; const awaitTask = !isInterval || entry.awaitTask; // Branch A — interval with awaitTask=false: arm the next tick BEFORE // running so we honour cadence regardless of how long the task // takes. Stale-callback guard still applies to next tick. if (isInterval && !awaitTask && !hasReachedMaxRuns(entry)) { scheduleNextIntervalTick(entry); } try { const maybe = entry.task(ctx); if (awaitTask && maybe instanceof Promise) { await maybe; } finishEntry(id, key, /* error */ null); } catch (err) { finishEntry(id, key, err); } } /** * Finalise the just-completed task. Identified by `(id, key)` only — * NOT `version`, because in `awaitTask: false` mode the next tick's * `armEntry()` already incremented `version` before the previous * task settled. The version guard is for native-callback rejection, * not for in-flight finalisation; here we only care that the entry * is still the same logical instance and not cancelled/replaced. */ function finishEntry(id: number, key: string, error: unknown | null): void { const entry = entries.get(key); if (entry === undefined) return; // cancelled / replaced (deleted) if (entry.id !== id) return; // replaced (new id) if (entry.status === TIMER_STATUS_CANCELLED) return; const now = clock.now(); const isInterval = entry.kind === TIMER_KIND_INTERVAL; if (error === null) { entry.lastCompletedAt = now; entry.error = null; emit({ type: TIMER_EVENT_COMPLETED, entry: snapshotOf(entry) }); } else { entry.lastErrorAt = now; entry.error = error; logger?.error?.(LOGGER_CATEGORY, taskFailedMessage(entry.key), { error }); emit({ type: TIMER_EVENT_FAILED, entry: snapshotOf(entry), error }); } if (isInterval) { // `entry.status` may already be TIMER_STATUS_PENDING if branch A // already armed the next tick; otherwise schedule it now. if (entry.status === TIMER_STATUS_RUNNING) { if (hasReachedMaxRuns(entry)) { entry.status = error === null ? TIMER_STATUS_COMPLETED : TIMER_STATUS_FAILED; entries.delete(entry.key); return; } scheduleNextIntervalTick(entry); } return; } // One-shot terminal state. entry.status = error === null ? TIMER_STATUS_COMPLETED : TIMER_STATUS_FAILED; if (entry.removeOnComplete) entries.delete(entry.key); } function scheduleNextIntervalTick(entry: InternalTimerEntry): void { if (entry.intervalMs === null) return; if (hasReachedMaxRuns(entry)) { entry.status = TIMER_STATUS_COMPLETED; entries.delete(entry.key); return; } const now = clock.now(); entry.scheduledAt = now; entry.delayMs = entry.intervalMs; entry.dueAt = now + entry.intervalMs; entry.status = TIMER_STATUS_PENDING; armEntry(entry, entry.intervalMs); } // ── Cancellation ──────────────────────────────────────────────────────── function cancelEntry(entry: InternalTimerEntry): void { entry.version += 1; entry.status = TIMER_STATUS_CANCELLED; if (entry.native !== null) { clock.clearTimeout(entry.native); entry.native = null; } entry.controller.abort(); entries.delete(entry.key); emit({ type: TIMER_EVENT_CANCELLED, entry: snapshotOf(entry) }); } // ── Public scheduling ─────────────────────────────────────────────────── function hasReachedMaxRuns(entry: InternalTimerEntry): boolean { return entry.maxRuns !== undefined && entry.runCount >= entry.maxRuns; } function scheduleCommon( key: string, delayMs: number, task: TimerTask, options: TimerOptions | undefined, kind: TimerKind, intervalMs: number | null, awaitTask: boolean, maxRuns: number | undefined ): TimerHandle { ensureLive(ENGINE_METHOD_SCHEDULE); validateKey(key); validateDelay(delayMs); const replace = options?.replace === true; const existing = entries.get(key); if (existing !== undefined) { if (!replace) throw new TimrDuplicateKeyError(duplicateKeyErrorMessage(key), key); cancelEntry(existing); } const now = clock.now(); const entry: InternalTimerEntry = { id: nextId++, key, version: 0, kind, status: TIMER_STATUS_PENDING, task, controller: new AbortController(), native: null, scheduledAt: now, dueAt: now + delayMs, delayMs, intervalMs, awaitTask, maxRuns, runCount: 0, lastFiredAt: null, lastCompletedAt: null, lastErrorAt: null, error: null, removeOnComplete: kind === TIMER_KIND_INTERVAL ? false : options?.removeOnComplete !== false, meta: options?.meta }; entries.set(key, entry); armEntry(entry, delayMs); emit({ type: TIMER_EVENT_SCHEDULED, entry: snapshotOf(entry) }); return makeHandle(entry); } function makeHandle(entry: InternalTimerEntry): TimerHandle { const id = entry.id; const key = entry.key; return { get key() { return key; }, get active() { const live = entries.get(key); return live !== undefined && live.id === id; }, cancel() { const live = entries.get(key); if (live === undefined || live.id !== id) return false; cancelEntry(live); return true; }, reschedule(delayMs: number) { const live = entries.get(key); if (live === undefined || live.id !== id) { throw new TimrInactiveTimerError(inactiveTimerErrorMessage(key), key); } if (live.status === TIMER_STATUS_RUNNING) { throw new TimrInactiveTimerError(inactiveTimerErrorMessage(key), key); } validateDelay(delayMs); if (live.native !== null) { clock.clearTimeout(live.native); live.native = null; } const now = clock.now(); live.scheduledAt = now; live.delayMs = delayMs; live.dueAt = now + delayMs; live.status = TIMER_STATUS_PENDING; armEntry(live, delayMs); emit({ type: TIMER_EVENT_SCHEDULED, entry: snapshotOf(live) }); } }; } // ── Public API ────────────────────────────────────────────────────────── function isInScope(key: string, scope: string): boolean { return key === scope || key.startsWith(`${scope}${DEFAULT_TIMER_SCOPE_SEPARATOR}`); } const engine: EngineTimers = { get size() { return entries.size; }, get disposed() { return disposed; }, schedule(key, delayMs, task, opts) { return scheduleCommon(key, delayMs, task, opts, TIMER_KIND_TIMEOUT, null, true, undefined); }, scheduleAt(key, dueAt, task, opts) { ensureLive(ENGINE_METHOD_SCHEDULE_AT); const now = clock.now(); let delayMs = dueAt - now; if (delayMs < 0) { logger?.debug?.(LOGGER_CATEGORY, LOG_MSG_SCHEDULE_AT_PAST, { key, dueAt, now }); delayMs = 0; } return scheduleCommon(key, delayMs, task, opts, TIMER_KIND_TIMEOUT, null, true, undefined); }, interval(key, everyMs, task, opts) { const intervalOpts = (opts ?? {}) as TimerIntervalOptions; const awaitTask = intervalOpts.awaitTask !== false; return scheduleCommon( key, everyMs, task, intervalOpts, TIMER_KIND_INTERVAL, everyMs, awaitTask, intervalOpts.maxRuns ); }, cancel(key) { ensureLive(ENGINE_METHOD_CANCEL); const entry = entries.get(key); if (entry === undefined) return false; cancelEntry(entry); return true; }, cancelAll(scope) { ensureLive(ENGINE_METHOD_CANCEL_ALL); let count = 0; // Snapshot keys to avoid mutating during iteration. const keys: string[] = []; for (const k of entries.keys()) { if (scope === undefined || isInScope(k, scope)) keys.push(k); } for (const k of keys) { const entry = entries.get(k); if (entry === undefined) continue; cancelEntry(entry); count += 1; } return count; }, has(key) { return entries.has(key); }, keys() { return Array.from(entries.keys()); }, entries() { const out: TimerEntrySnapshot[] = []; for (const entry of entries.values()) out.push(snapshotOf(entry)); return out; }, entry(key) { const e = entries.get(key); return e === undefined ? null : snapshotOf(e); }, onChange(listener) { if (disposed) return () => {}; listeners.add(listener); return () => { listeners.delete(listener); }; }, dispose() { if (disposed) return; disposed = true; // Cancel all without re-triggering ensureLive. const all = Array.from(entries.values()); for (const entry of all) cancelEntry(entry); emit({ type: TIMER_EVENT_DISPOSED }); listeners.clear(); } }; return engine; }