test/loop-durable-schedule.test.js

Rama main · réplica de lectura del repositorio original.

var test = require("node:test");
var assert = require("node:assert/strict");
var fs = require("fs");
var os = require("os");
var path = require("path");
var attachScheduler = require("../lib/durable-scheduler").attachDurableScheduler;
var createRegistry = require("../lib/scheduler").createLoopRegistry;
var attachSchedule = require("../lib/loop-durable-schedule").attachLoopDurableSchedule;

var roots = [];
var jobIds = 0;
test.after(function () { for (var i = 0; i < roots.length; i++) fs.rmSync(roots[i], { recursive: true, force: true }); });
function root() { var value = fs.mkdtempSync(path.join(os.tmpdir(), "clay-loop-durable-")); roots.push(value); return value; }
function turn() { return new Promise(function (resolve) { setImmediate(resolve); }); }
function engine(dir, now) {
  return attachScheduler({ storageDir: dir, namespace: "test", now: function () { return now.value; }, makeId: function () { jobIds++; return "job-" + jobIds; }, setTimer: function () { return { unref: function () {} }; }, clearTimer: function () {} });
}
function dueRecord(registry, id, dueAt, ownerId) {
  var record = registry.register({ id: id, name: id, task: "run", cron: "* * * * *", enabled: true, ownerId: ownerId || null });
  registry.updateRecord(id, { nextRunAt: dueAt });
  return record;
}

test("one shared engine dispatches durable occurrences for multiple projects and owners", async function () {
  var dir = root(); var now = { value: 1000 }; var scheduler = engine(dir, now); var calls = [];
  var first = createRegistry({ cwd: dir, registryPath: path.join(dir, "first.jsonl"), onTrigger: function (record) { calls.push("first:" + record.ownerId); } }); first.load(); dueRecord(first, "one", 1000, "owner-a");
  var second = createRegistry({ cwd: dir, registryPath: path.join(dir, "second.jsonl"), onTrigger: function (record) { calls.push("second:" + record.ownerId); } }); second.load(); dueRecord(second, "two", 1000, "owner-b");
  var a = attachSchedule({ scheduler: scheduler, registry: first, projectId: "first", isMultiUser: true, authorize: function () { return true; } });
  var b = attachSchedule({ scheduler: scheduler, registry: second, projectId: "second", isMultiUser: true, authorize: function () { return true; } });
  scheduler.start(); scheduler.tick(); await turn(); await turn();
  assert.deepEqual(calls.sort(), ["first:owner-a", "second:owner-b"]);
  assert.equal(scheduler.listJobs().filter(function (job) { return job.execution && job.execution.receipt; }).length, 2);
  a.shutdown(); b.shutdown(); await scheduler.shutdown();
});

test("metadata edits preserve a queued occurrence while disable and delete cancel it", function () {
  var dir = root(); var now = { value: 1000 }; var scheduler = engine(dir, now);
  var registry = createRegistry({ cwd: dir, registryPath: path.join(dir, "registry.jsonl") }); registry.load(); dueRecord(registry, "stable", 5000);
  var service = attachSchedule({ scheduler: scheduler, registry: registry, projectId: "project", isMultiUser: false });
  var original = scheduler.listJobs().filter(function (job) { return job.state === "queued"; })[0];
  registry.update("stable", { name: "Renamed" }); service.sync();
  var afterEdit = scheduler.listJobs().filter(function (job) { return job.state === "queued"; })[0];
  assert.equal(afterEdit.id, original.id); assert.equal(afterEdit.runAt, 5000);
  registry.toggleEnabled("stable"); service.sync(); assert.equal(scheduler.getJob(original.id).state, "cancelled");
  registry.toggleEnabled("stable"); service.sync(); var replacement = scheduler.listJobs().filter(function (job) { return job.state === "queued"; })[0];
  registry.remove("stable"); service.sync(); assert.equal(scheduler.getJob(replacement.id).state, "cancelled");
  service.shutdown(); scheduler.shutdown();
});

test("ownerless multi-user records fail closed with an explicit setup marker", function () {
  var dir = root(); var now = { value: 1000 }; var scheduler = engine(dir, now);
  var registry = createRegistry({ cwd: dir, registryPath: path.join(dir, "registry.jsonl") }); registry.load(); dueRecord(registry, "ownerless", 1000);
  var service = attachSchedule({ scheduler: scheduler, registry: registry, projectId: "project", isMultiUser: true, authorize: function () { return true; } });
  var record = registry.getById("ownerless");
  assert.equal(record.enabled, false); assert.equal(record.needsOwner, true); assert.equal(record.setupRequired, "owner");
  assert.equal(scheduler.listJobs().filter(function (job) { return job.state === "queued"; }).length, 0);
  service.shutdown(); scheduler.shutdown();
});

test("busy legacy work stays durably queued and dispatches once eligible", async function () {
  var dir = root(); var now = { value: 1000 }; var scheduler = engine(dir, now); var busy = true; var calls = 0;
  var registry = createRegistry({ cwd: dir, registryPath: path.join(dir, "registry.jsonl"), onTrigger: function () { calls++; } }); registry.load(); dueRecord(registry, "busy", 1000);
  var service = attachSchedule({ scheduler: scheduler, registry: registry, projectId: "project", isMultiUser: false, canRun: function () { return !busy; } });
  scheduler.start(); scheduler.tick(); await turn(); assert.equal(calls, 0); assert.equal(scheduler.listJobs()[0].state, "queued");
  busy = false; scheduler.tick(); await turn(); await turn(); assert.equal(calls, 1); assert.equal(scheduler.listJobs()[0].state, "completed");
  service.shutdown(); scheduler.shutdown();
});

test("a biweekly schedule rule advances from its anchor and persists start counts", async function (t) {
  var originalNow = Date.now;
  var dueAt = new Date(2028, 0, 5, 9, 0, 0).getTime();
  var now = { value: dueAt + 1000 };
  Date.now = function () { return now.value; };
  t.after(function () { Date.now = originalNow; });
  var dir = root(); var scheduler = engine(dir, now); var calls = 0;
  var registry = createRegistry({ cwd: dir, registryPath: path.join(dir, "registry.jsonl"), onTrigger: function () { calls++; } }); registry.load();
  var rule = { version: 1, timezone: "server-local", anchorDate: "2028-01-03", time: "09:00", recurrence: { every: 2, unit: "week", weekdays: [1, 3] } };
  var record = registry.register({ id: "biweekly", name: "Biweekly", task: "run", scheduleRule: rule, date: rule.anchorDate, time: rule.time, enabled: true });
  registry.updateRecord(record.id, { nextRunAt: dueAt });
  var service = attachSchedule({ scheduler: scheduler, registry: registry, projectId: "project", isMultiUser: false });
  scheduler.start(); scheduler.tick(); await turn(); await turn();
  assert.equal(calls, 1); assert.equal(record.scheduleStartCount, 1);
  assert.equal(record.nextRunAt, new Date(2028, 0, 17, 9, 0, 0).getTime());
  service.shutdown(); scheduler.shutdown();
});

test("a persisted queued occurrence survives engine and registry restart", async function () {
  var dir = root(); var now = { value: 1000 }; var firstEngine = engine(dir, now); var registryPath = path.join(dir, "registry.jsonl");
  var firstRegistry = createRegistry({ cwd: dir, registryPath: registryPath }); firstRegistry.load(); dueRecord(firstRegistry, "restart", 1000);
  var firstService = attachSchedule({ scheduler: firstEngine, registry: firstRegistry, projectId: "project", isMultiUser: false }); firstService.shutdown(); await firstEngine.shutdown();
  var secondEngine = engine(dir, now); var calls = 0; var secondRegistry = createRegistry({ cwd: dir, registryPath: registryPath, onTrigger: function () { calls++; } }); secondRegistry.load();
  var secondService = attachSchedule({ scheduler: secondEngine, registry: secondRegistry, projectId: "project", isMultiUser: false }); secondEngine.start(); secondEngine.tick(); await turn(); await turn();
  assert.equal(calls, 1); assert.equal(secondEngine.listJobs()[0].state, "completed");
  secondService.shutdown(); secondEngine.shutdown();
});

test("a running occurrence ignores harmless revision changes without a duplicate idempotency conflict", async function () {
  var dir = root(); var now = { value: 1000 }; var scheduler = engine(dir, now); var calls = 0;
  var registry = createRegistry({ cwd: dir, registryPath: path.join(dir, "registry.jsonl"), onTrigger: function () { calls++; } }); registry.load(); dueRecord(registry, "running-edit", 1000);
  var service = attachSchedule({ scheduler: scheduler, registry: registry, projectId: "project", isMultiUser: false });
  scheduler.start();
  assert.equal(scheduler.listJobs()[0].state, "running");
  assert.doesNotThrow(function () { registry.updateRecord("running-edit", { name: "Renamed while claimed" }); });
  assert.equal(scheduler.listJobs().filter(function (job) { return job.payload.occurrenceAt === 1000; }).length, 1);
  await turn(); await turn(); assert.equal(calls, 1);
  service.shutdown(); scheduler.shutdown();
});

test("an interrupted ledger occurrence is consumed before its successor is enqueued", async function () {
  var dir = root(); var now = { value: 1000 }; var registryPath = path.join(dir, "registry.jsonl"); var calls = 0;
  var firstEngine = engine(dir, now); var firstRegistry = createRegistry({ cwd: dir, registryPath: registryPath, onTrigger: function () { calls++; } }); firstRegistry.load(); dueRecord(firstRegistry, "interrupted", 1000);
  var firstService = attachSchedule({ scheduler: firstEngine, registry: firstRegistry, projectId: "project", isMultiUser: false }); firstEngine.start();
  assert.equal(firstEngine.listJobs()[0].state, "running");
  await firstEngine.shutdown(); firstService.shutdown();
  var secondEngine = engine(dir, now); var secondRegistry = createRegistry({ cwd: dir, registryPath: registryPath, onTrigger: function () { calls++; } }); secondRegistry.load();
  var secondService = attachSchedule({ scheduler: secondEngine, registry: secondRegistry, projectId: "project", isMultiUser: false });
  assert.equal(calls, 0); assert.equal(secondEngine.listJobs()[0].state, "interrupted");
  assert.notEqual(secondRegistry.getById("interrupted").nextRunAt, 1000);
  assert.equal(secondEngine.listJobs().filter(function (job) { return job.state === "queued"; }).length, 1);
  secondService.shutdown(); secondEngine.shutdown();
});

test("restart after registry advance but before engine settlement keeps only the persisted successor", async function () {
  var dir = root(); var now = { value: 1000 }; var registryPath = path.join(dir, "registry.jsonl"); var calls = 0; var firstEngine = engine(dir, now);
  var firstRegistry = createRegistry({
    cwd: dir, registryPath: registryPath,
    onTrigger: function () { calls++; firstEngine.shutdown(); },
  });
  firstRegistry.load(); dueRecord(firstRegistry, "advance-crash", 1000);
  var firstService = attachSchedule({ scheduler: firstEngine, registry: firstRegistry, projectId: "project", isMultiUser: false });
  firstEngine.start(); await turn(); await turn();
  var advancedAt = firstRegistry.getById("advance-crash").nextRunAt;
  assert.equal(calls, 1); assert.notEqual(advancedAt, 1000); assert.equal(firstEngine.listJobs()[0].state, "interrupted");
  firstService.shutdown();

  var secondEngine = engine(dir, now); var secondRegistry = createRegistry({ cwd: dir, registryPath: registryPath, onTrigger: function () { calls++; } }); secondRegistry.load();
  var secondService = attachSchedule({ scheduler: secondEngine, registry: secondRegistry, projectId: "project", isMultiUser: false });
  var queued = secondEngine.listJobs().filter(function (job) { return job.state === "queued"; });
  assert.equal(queued.length, 1); assert.equal(queued[0].payload.occurrenceAt, advancedAt);
  assert.equal(secondEngine.listJobs().filter(function (job) { return job.payload.occurrenceAt === 1000; }).length, 1);
  secondService.shutdown(); secondEngine.shutdown();
});

test("durable sync applies the overdue rule cutoff before enqueueing", function (t) {
  var originalNow = Date.now; var dir = root(); var registryPath = path.join(dir, "registry.jsonl");
  var now = { value: new Date(2099, 0, 5, 8, 59, 0).getTime() };
  Date.now = function () { return now.value; };
  t.after(function () { Date.now = originalNow; });
  var scheduler = engine(dir, now); var calls = 0;
  var registry = createRegistry({ cwd: dir, registryPath: registryPath, onTrigger: function () { calls++; } }); registry.load();
  var rule = { version: 1, timezone: "server-local", anchorDate: "2099-01-05", time: "09:00", recurrence: { every: 1, unit: "day" }, recurrenceEnd: { type: "until", date: "2099-01-05" }, interval: { every: 30, unit: "minute", end: { type: "until", time: "09:45" } } };
  var record = registry.register({ id: "overdue-cutoff", name: "Overdue", task: "run", source: "schedule", scheduleRule: rule, date: rule.anchorDate, time: rule.time, enabled: true });
  now.value = new Date(2099, 0, 5, 10, 0, 0).getTime();
  var service = attachSchedule({ scheduler: scheduler, registry: registry, projectId: "project", isMultiUser: false });
  assert.equal(calls, 0); assert.equal(record.enabled, false); assert.equal(record.nextRunAt, null);
  assert.equal(scheduler.listJobs().filter(function (job) { return job.state === "queued"; }).length, 0);
  service.shutdown(); scheduler.shutdown();
});

test("registry updates roll back when durable job persistence fails", function () {
  var dir = root(); var now = { value: 1000 }; var registryPath = path.join(dir, "registry.jsonl"); var stored = { version: 1, namespace: "test", jobs: [] }; var failWrites = false; var service;
  var store = {
    load: function () { return JSON.parse(JSON.stringify(stored)); },
    save: function (data) { if (failWrites) throw new Error("injected durable write failure"); stored = JSON.parse(JSON.stringify(data)); },
    close: function () {},
  };
  var scheduler = attachScheduler({ store: store, namespace: "test", now: function () { return now.value; }, makeId: function () { jobIds++; return "job-" + jobIds; }, setTimer: function () { return { unref: function () {} }; }, clearTimer: function () {} });
  var calls = 0;
  var registry = createRegistry({ cwd: dir, registryPath: registryPath, onTrigger: function () { calls++; }, onChange: function () { if (service) service.sync(); } }); registry.load();
  var record = dueRecord(registry, "rollback-sync", 5000); service = attachSchedule({ scheduler: scheduler, registry: registry, projectId: "project", isMultiUser: false });
  var oldRevision = record.updatedAt; failWrites = true;
  var result = registry.update(record.id, { name: "Must roll back" });
  var diskRecord = JSON.parse(fs.readFileSync(registryPath, "utf8").trim());
  assert.equal(result, null); assert.equal(record.name, "rollback-sync"); assert.equal(record.updatedAt, oldRevision); assert.equal(diskRecord.name, "rollback-sync");
  assert.match(registry.getLastSaveError(), /durable synchronization failed/); assert.equal(calls, 0);
  service.shutdown(); scheduler.shutdown();
});

test("pause, resume, delete, and recreate at the same due time use new occurrence identities", function () {
  var dir = root(); var now = { value: 1000 }; var scheduler = engine(dir, now); var registryPath = path.join(dir, "registry.jsonl");
  var registry = createRegistry({ cwd: dir, registryPath: registryPath }); registry.load();
  var first = registry.register({ id: "lifecycle", name: "Once", task: "run", source: "schedule", date: "1970-01-01", time: "00:00", enabled: true }); registry.updateRecord(first.id, { nextRunAt: 5000, enabled: true });
  var service = attachSchedule({ scheduler: scheduler, registry: registry, projectId: "project", isMultiUser: false });
  var firstJob = scheduler.listJobs().filter(function (job) { return job.state === "queued"; })[0];
  registry.toggleEnabled(first.id); registry.toggleEnabled(first.id); registry.updateRecord(first.id, { nextRunAt: 5000 }); service.sync();
  var resumed = scheduler.listJobs().filter(function (job) { return job.state === "queued"; })[0];
  assert.notEqual(resumed.idempotencyKey, firstJob.idempotencyKey); assert.equal(scheduler.getJob(firstJob.id).state, "cancelled");
  registry.remove(first.id);
  var recreated = registry.register({ id: "lifecycle", name: "Recreated", task: "run", source: "schedule", date: "1970-01-01", time: "00:00", enabled: true }); registry.updateRecord(recreated.id, { nextRunAt: 5000, enabled: true }); service.sync();
  var latest = scheduler.listJobs().filter(function (job) { return job.state === "queued"; })[0];
  assert.notEqual(latest.idempotencyKey, resumed.idempotencyKey); assert.equal(scheduler.getJob(resumed.id).state, "cancelled");
  service.shutdown(); scheduler.shutdown();
});

test("null nextRunAt never creates a due-zero job and live multi-user mode revokes the local sentinel", function () {
  var dir = root(); var now = { value: 1000 }; var scheduler = engine(dir, now); var multiUser = false;
  var registry = createRegistry({ cwd: dir, registryPath: path.join(dir, "registry.jsonl") }); registry.load(); var record = dueRecord(registry, "mode-change", 5000);
  var service = attachSchedule({ scheduler: scheduler, registry: registry, projectId: "project", isMultiUser: function () { return multiUser; }, authorize: function () { return true; } });
  var localJob = scheduler.listJobs().filter(function (job) { return job.state === "queued"; })[0]; assert.equal(localJob.ownerId, "local-user");
  multiUser = true; service.sync(); assert.equal(scheduler.getJob(localJob.id).state, "cancelled"); assert.equal(record.needsOwner, true); assert.equal(record.nextRunAt, null);
  service.sync(); assert.equal(scheduler.listJobs().some(function (job) { return job.state === "queued" && job.runAt === 0; }), false);
  service.shutdown(); scheduler.shutdown();
});