fix(run-store): count records in the cycle-key size metric

The metric is on the wp:<cycleSeq> key, but it measured only the order field.
records dominates that key once populated, so a 20KB cycle key was reported as
7 bytes and the high-water log could never fire on the field that actually
grows.
This commit is contained in:
Daniel Sutton
2026-08-24 10:58:39 +01:00
parent b430abdc10
commit cbcab0a190
2 changed files with 65 additions and 5 deletions
@@ -1254,6 +1254,53 @@ describe("hash tag and keyPrefix", () => {
});
describe("observability", () => {
redisTest("cycle-key bytes cover records, not just order", async ({ redisOptions }) => {
// The plan puts a metric on the wp:<cycleSeq> KEY size. records dominates that key once
// populated, so measuring order alone understates it by orders of magnitude.
const calls: Array<[string, number]> = [];
const metrics = {
recordAppend: () => {},
recordEntryBytes: () => {},
recordCycleKeyBytes: (b: number) => calls.push(["cycleBytes", b]),
recordCycleCount: () => {},
recordSkippedNoKeyspace: () => {},
recordCycleMismatch: () => {},
recordLatency: () => {},
};
const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 60_000, metrics });
try {
const records: CompletedWaitpointRecord[] = [
{
id: "w_a",
friendlyId: "waitpoint_a",
type: "MANUAL",
completedAt: "2026-01-01T00:00:00.000Z",
outputType: "application/json",
outputIsError: false,
output: { inline: "y".repeat(20_000) },
},
];
const orderJson = JSON.stringify(["w_a"]);
const recordsJson = JSON.stringify(records);
await store.append({
entry: entry({ id: "snap_1" }),
kind: "birth",
isTerminal: false,
cycle: { kind: "new", completedWaitpoints: [{ id: "w_a", index: 0 }], records },
});
expect(calls).toEqual([
[
"cycleBytes",
Buffer.byteLength(orderJson, "utf8") + Buffer.byteLength(recordsJson, "utf8"),
],
]);
} finally {
await store.quit();
}
});
redisTest("records sizes and outcomes without ever rejecting", async ({ redisOptions }) => {
const calls: unknown[][] = [];
const metrics = {
@@ -301,11 +301,17 @@ export class RedisSnapshotStore {
args.expectedCur !== undefined ? "1" : "0"
)) as string[];
return this.#interpretAppend(reply, raw, orderJson, args.entry.runId);
return this.#interpretAppend(reply, raw, orderJson, records, args.entry.runId);
});
}
#interpretAppend(reply: string[], raw: string, orderJson: string, runId: string): AppendResult {
#interpretAppend(
reply: string[],
raw: string,
orderJson: string,
records: string,
runId: string
): AppendResult {
if (reply[0] === SKIPPED) {
this.metrics?.recordSkippedNoKeyspace();
this.metrics?.recordAppend("skippedNoKeyspace", "none");
@@ -326,7 +332,7 @@ export class RedisSnapshotStore {
if (cycleMismatch) {
this.metrics?.recordCycleMismatch();
}
this.#observeSizes(raw, orderJson, cycleSeq, runId);
this.#observeSizes(raw, orderJson, records, cycleSeq, runId);
this.metrics?.recordAppend("written", ttl);
return {
outcome: "written",
@@ -337,14 +343,21 @@ export class RedisSnapshotStore {
};
}
#observeSizes(raw: string, orderJson: string, cycleSeq: number, runId: string): void {
#observeSizes(
raw: string,
orderJson: string,
records: string,
cycleSeq: number,
runId: string
): void {
const entryBytes = Buffer.byteLength(raw, "utf8");
this.metrics?.recordEntryBytes(entryBytes);
if (this.highWater.entryBytes !== undefined && entryBytes > this.highWater.entryBytes) {
this.logger.warn("RedisSnapshotStore entry above high-water mark", { runId, entryBytes });
}
if (orderJson !== "") {
const cycleBytes = Buffer.byteLength(orderJson, "utf8");
// The whole wp:<cycleSeq> key, not just its order field: records dominates it once populated.
const cycleBytes = Buffer.byteLength(orderJson, "utf8") + Buffer.byteLength(records, "utf8");
this.metrics?.recordCycleKeyBytes(cycleBytes);
if (this.highWater.cycleKeyBytes !== undefined && cycleBytes > this.highWater.cycleKeyBytes) {
this.logger.warn("RedisSnapshotStore cycle key above high-water mark", {