Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
150 changes: 150 additions & 0 deletions packages/test/src/test/trigger/IntervalTrigger.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,8 @@

import type { ITriggerFireContext } from "@workglow/triggers";
import { IntervalTrigger, TriggerConfigurationError } from "@workglow/triggers";
import type { ILogger } from "@workglow/util";
import { getLogger, setLogger } from "@workglow/util";
import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";

import { advanceFakeTimers, flushAsyncWork } from "../helpers/advanceFakeTimers";
Expand All @@ -26,6 +28,38 @@ function createGate(): Gate {
return { promise, open };
}

interface WarnRecord {
readonly message: string;
readonly meta: Record<string, unknown> | undefined;
}

/**
* Installs a logger that records `warn` calls; returns the sink and a restore fn.
* Built from scratch rather than spread from the installed logger, whose methods
* live on a class prototype and would be lost.
*/
function captureWarnings(): { warnings: WarnRecord[]; restore: () => void } {
const warnings: WarnRecord[] = [];
const previous = getLogger();
const noop = (): void => {};
const logger: ILogger = {
debug: noop,
info: noop,
warn: (message: string, meta?: Record<string, unknown>) => {
warnings.push({ message, meta });
},
error: noop,
fatal: noop,
child: () => logger,
time: noop,
timeEnd: noop,
group: noop,
groupEnd: noop,
};
setLogger(logger);
return { warnings, restore: () => setLogger(previous) };
}

describe("IntervalTrigger", () => {
beforeEach(() => {
vi.useFakeTimers();
Expand Down Expand Up @@ -286,6 +320,93 @@ describe("IntervalTrigger", () => {
});
});

describe("skip logging", () => {
test("a contiguous run of skips logs once, then a count when it ends", async () => {
// A wedged handler at a 1s period would otherwise write 3,600 identical
// warnings an hour, forever.
const { warnings, restore } = captureWarnings();
try {
const trigger = new IntervalTrigger({ intervalMs: PERIOD });
const gate = createGate();
const skips: number[] = [];
trigger.on("skip", (scheduledAt) => skips.push(scheduledAt));
trigger.start(async () => {
await gate.promise;
});

await advanceFakeTimers(PERIOD);
await advanceFakeTimers(PERIOD * 4);

// The EVENT still fires per dropped tick; only the log is collapsed.
expect(skips).toHaveLength(4);
expect(warnings).toHaveLength(1);
expect(warnings[0]?.message).toContain("skipped");

gate.open();
await flushAsyncWork();
await advanceFakeTimers(PERIOD);

expect(warnings).toHaveLength(2);
expect(warnings[1]?.meta?.skipped).toBe(4);

await trigger.stop();
} finally {
restore();
}
});

test("an isolated skip logs once and adds no summary", async () => {
// The ordinary intermittent case must not have its log volume doubled.
const { warnings, restore } = captureWarnings();
try {
const trigger = new IntervalTrigger({ intervalMs: PERIOD });
const gate = createGate();
const skips: number[] = [];
trigger.on("skip", (scheduledAt) => skips.push(scheduledAt));
trigger.start(async () => {
await gate.promise;
});

await advanceFakeTimers(PERIOD * 2);
expect(skips).toHaveLength(1);

gate.open();
await flushAsyncWork();
await advanceFakeTimers(PERIOD);

expect(warnings).toHaveLength(1);

await trigger.stop();
} finally {
restore();
}
});

test("stopping mid-skip still reports the skip count", async () => {
const { warnings, restore } = captureWarnings();
try {
const trigger = new IntervalTrigger({ intervalMs: PERIOD });
const gate = createGate();
const skips: number[] = [];
trigger.on("skip", (scheduledAt) => skips.push(scheduledAt));
trigger.start(async () => {
await gate.promise;
});

await advanceFakeTimers(PERIOD * 4);
expect(skips).toHaveLength(3);

const stopping = trigger.stop();
expect(warnings.map((warning) => warning.meta?.skipped)).toContain(3);

gate.open();
await stopping;
} finally {
restore();
}
});
});

test("a period past the host timer ceiling is served in chunks", async () => {
// A delay over 2^31-1 ms overflows a host timer and fires IMMEDIATELY, so
// an unchunked wait would turn a very long period into a hot loop.
Expand Down Expand Up @@ -515,6 +636,35 @@ describe("IntervalTrigger", () => {
await trigger.stop();
});

test("a restart during stop() does not emit stop while the trigger is running", async () => {
// An observer that sees `stop` while `running` is true concludes the
// trigger is dead and stops watching it — while it is actively firing.
const trigger = new IntervalTrigger({ intervalMs: PERIOD });
const gate = createGate();
const runningAtStop: boolean[] = [];
trigger.on("stop", () => runningAtStop.push(trigger.running));

trigger.start(async () => {
await gate.promise;
});
await advanceFakeTimers(PERIOD);

// NOT awaited: the gated gen-1 handler keeps the old run draining.
const stopping = trigger.stop();
trigger.start(() => {});

gate.open();
await stopping;

// Gen 1 drained, but gen 2 owns the trigger.
expect(runningAtStop).toEqual([]);
expect(trigger.running).toBe(true);

// ...and the real stop still reports itself.
await trigger.stop();
expect(runningAtStop).toEqual([false]);
});

test("a restart during stop() does not emit spurious skips (skip policy)", async () => {
// The first generation's handler is still gated when the second starts.
// Overlap state belongs to a RUN, not to the trigger: if the new
Expand Down
Loading
Loading