diigo-cache: the caches take their clock as a constructor argument
Ocap discipline: no ambient time in logic. PagedCache reads time and
waits only through a Clock { now, sleep }, a default constructor
argument (the live clock; the runtime constructs DOs with ctx and env
only). The fill job's timestamps and alarms, the etch time, and the
diigo-call pacing all go through it.
Tests interpose through the constructor: tests/test-worker.ts (now the
tests' main module) adds FixedClockCache, bound as CLOCKED_DO, and a
test pins startedAt, updatedAt, and the alarm time to the fixed clock
(an ambient new Date() fails it).
Generated-by: Claude Opus 5.5
This commit is contained in:
1 parent
7d2100bf77
commit
b6823867ff
4 files changed
+76
-17
No files matched your search
+35
-12
@@ -84,9 +84,19 @@ function disposeStub(stub: unknown): void {
|
||||
if (typeof dispose === "function") dispose.call(stub);
|
||||
}
|
||||
|
||||
function sleep(ms: number): Promise<void> {
|
||||
return new Promise((r) => setTimeout(r, ms));
|
||||
}
|
||||
/** the clock authority a cache uses: reading time, and waiting. Passed
|
||||
* in (ocap discipline), so tests can interpose a fixed clock. */
|
||||
export type Clock = {
|
||||
/** epoch milliseconds */
|
||||
now(): number;
|
||||
sleep(ms: number): Promise<void>;
|
||||
};
|
||||
|
||||
/** the live clock: the one place this module touches time */
|
||||
const liveClock: Clock = {
|
||||
now: () => Date.now(),
|
||||
sleep: (ms) => new Promise((r) => setTimeout(r, ms)),
|
||||
};
|
||||
|
||||
/** a cache's wiring: the webapp instance it asks on a miss, plus (private
|
||||
* cache only) the public companion's id, recorded once published */
|
||||
@@ -110,6 +120,19 @@ export type CacheEnv = {
|
||||
abstract class PagedCache extends DurableObject<CacheEnv> {
|
||||
#step = 0;
|
||||
#pause?: Promise<void>;
|
||||
#clock: Clock;
|
||||
|
||||
/** the runtime constructs DOs with (ctx, env); `clock` defaults to the
|
||||
* live one, and a subclass can pass another */
|
||||
constructor(ctx: DurableObjectState, env: CacheEnv, clock: Clock = liveClock) {
|
||||
super(ctx, env);
|
||||
this.#clock = clock;
|
||||
}
|
||||
|
||||
/** now, as an ISO timestamp */
|
||||
#iso(): string {
|
||||
return new Date(this.#clock.now()).toISOString();
|
||||
}
|
||||
|
||||
/** one-shot wiring. Idempotent; different values are refused (mint a
|
||||
* new cache instead). */
|
||||
@@ -146,7 +169,7 @@ abstract class PagedCache extends DurableObject<CacheEnv> {
|
||||
if (current?.state === "running") return current;
|
||||
const step = Math.trunc(opts.pagesPerStep ?? 20);
|
||||
if (!(step >= 1 && step <= 50)) throw new Error("pagesPerStep: expected 1–50");
|
||||
const now = new Date().toISOString();
|
||||
const now = this.#iso();
|
||||
const status: FillStatus = {
|
||||
state: "running",
|
||||
nextPage: 1,
|
||||
@@ -156,7 +179,7 @@ abstract class PagedCache extends DurableObject<CacheEnv> {
|
||||
updatedAt: now,
|
||||
};
|
||||
await this.ctx.storage.put({ fill: status, fillStep: step });
|
||||
await this.ctx.storage.setAlarm(Date.now() + FILL_STEP_DELAY_MS);
|
||||
await this.ctx.storage.setAlarm(this.#clock.now() + FILL_STEP_DELAY_MS);
|
||||
console.log({ event: "fill", kind: this.kind, state: status.state, pagesPerStep: step });
|
||||
return status;
|
||||
}
|
||||
@@ -173,7 +196,7 @@ abstract class PagedCache extends DurableObject<CacheEnv> {
|
||||
const current = await this.fillStatus();
|
||||
if (!current) throw new Error("no fill job to stop");
|
||||
if (current.state !== "running") return current;
|
||||
const stopped: FillStatus = { ...current, state: "stopped", updatedAt: new Date().toISOString() };
|
||||
const stopped: FillStatus = { ...current, state: "stopped", updatedAt: this.#iso() };
|
||||
await this.ctx.storage.put("fill", stopped);
|
||||
await this.ctx.storage.deleteAlarm();
|
||||
console.log({ event: "fill", kind: this.kind, state: "stopped", nextPage: stopped.nextPage, pages: stopped.pages });
|
||||
@@ -197,10 +220,10 @@ abstract class PagedCache extends DurableObject<CacheEnv> {
|
||||
nextPage: status.nextPage + 1,
|
||||
pages: Math.max(1, Math.ceil(page.total / PAGE_SIZE)),
|
||||
fetched: status.fetched + (page.cached ? 0 : 1),
|
||||
updatedAt: new Date().toISOString(),
|
||||
updatedAt: this.#iso(),
|
||||
};
|
||||
} catch (err) {
|
||||
status = { ...status, state: "failed", error: String(err), updatedAt: new Date().toISOString() };
|
||||
status = { ...status, state: "failed", error: String(err), updatedAt: this.#iso() };
|
||||
console.error({ event: "fill", kind: this.kind, state: "failed", page: status.nextPage, error: status.error });
|
||||
}
|
||||
// a stop that landed mid-step wins
|
||||
@@ -209,11 +232,11 @@ abstract class PagedCache extends DurableObject<CacheEnv> {
|
||||
}
|
||||
if (!status) return;
|
||||
if (status.state === "running" && status.pages !== null && status.nextPage > status.pages) {
|
||||
status = { ...status, state: "done", updatedAt: new Date().toISOString() };
|
||||
status = { ...status, state: "done", updatedAt: this.#iso() };
|
||||
}
|
||||
await this.ctx.storage.put("fill", status);
|
||||
console.log({ event: "fill", kind: this.kind, state: status.state, nextPage: status.nextPage, pages: status.pages, fetched: status.fetched });
|
||||
if (status.state === "running") await this.ctx.storage.setAlarm(Date.now() + FILL_STEP_DELAY_MS);
|
||||
if (status.state === "running") await this.ctx.storage.setAlarm(this.#clock.now() + FILL_STEP_DELAY_MS);
|
||||
}
|
||||
|
||||
/** diigo's page count, if any cached page recorded `total` */
|
||||
@@ -244,7 +267,7 @@ abstract class PagedCache extends DurableObject<CacheEnv> {
|
||||
if (this.#pause) await this.#pause;
|
||||
} while (before !== this.#step);
|
||||
this.#step += 1;
|
||||
this.#pause = sleep(MS_PER_CALL);
|
||||
this.#pause = this.#clock.sleep(MS_PER_CALL);
|
||||
|
||||
const hit = await this.#pageFromStorage(pageNum, count);
|
||||
if (hit) {
|
||||
@@ -288,7 +311,7 @@ abstract class PagedCache extends DurableObject<CacheEnv> {
|
||||
}
|
||||
this.ctx.storage.sql.exec(
|
||||
"insert into pages (page_num, count, total, etched_at) values (?, ?, ?, ?) on conflict(page_num) do update set count = excluded.count, total = excluded.total, etched_at = excluded.etched_at",
|
||||
pageNum, count, total, new Date().toISOString(),
|
||||
pageNum, count, total, this.#iso(),
|
||||
);
|
||||
});
|
||||
return { total, items: rows.map((r) => r.json), cached: false };
|
||||
|
||||
Reference in new issue
Block a user