fix(core,cli): retain idempotency key metadata beyond 1000 keys per run (#4094)
Fixes #4046 ## Problem When a single run creates more than 1000 idempotency keys (e.g. a large batch trigger where each item calls `idempotencyKeys.create()`), the original key and scope metadata is silently dropped for all but the most recent 1000 keys. `idempotencyKeys.create()` returns a plain 64-char hash and stores the `{ key, scope }` mapping in an in-process catalog keyed by that hash. That catalog was a fixed-size **LRU capped at 1000 entries**. Once a run creates more than 1000 keys, the earliest mappings are evicted, so when the SDK later looks them up to attach `idempotencyKeyOptions` to the trigger call, it finds nothing and sends `undefined`. The affected runs then: - report `ctx.run.idempotencyKey` as the raw hash instead of the user-provided key - have no `idempotencyKeyScope` - show empty `idempotency_key` / `idempotency_key_scope` in the dashboard and analytics Deduplication still works (the hash is intact); only the human-readable metadata is lost, which makes the failure silent and hard to notice. ## Fix - Replace the LRU catalog with an unbounded in-memory catalog, so every key created within a run keeps its metadata regardless of how many are created. - Clear the catalog at each run boundary via `resetExecutionEnvironment()` (both dev and managed workers), matching how every other per-run manager is reset. Deployed workers reuse one process across many runs (warm starts), so this bounds memory to a single run's keys instead of accumulating across runs — which is the reason the size cap existed in the first place. ## Tests - New public-API test creates 3000 keys and asserts all of them (including the first) retain their key/scope — this fails on `main` and passes with the fix. - New test for the in-memory catalog covers store/retrieve/overwrite, large-N retention (no eviction), and `clear()`. - New test asserts the catalog is emptied after a run-boundary reset. - Replaces the previous LRU catalog + its eviction tests. Verified: `@trigger.dev/core` and `trigger.dev` both build; all idempotency tests pass. Changeset added (patch).
This commit is contained in:
Matt Aitken
@@ -35,6 +35,7 @@ import {
|
||||
realtimeStreams,
|
||||
inputStreams,
|
||||
sessionStreams,
|
||||
resetIdempotencyKeyCatalog,
|
||||
} from "@trigger.dev/core/v3";
|
||||
import { TriggerTracer } from "@trigger.dev/core/v3/tracer";
|
||||
import {
|
||||
@@ -378,6 +379,7 @@ function resetExecutionEnvironment() {
|
||||
taskContext.disable();
|
||||
standardTraceContextManager.reset();
|
||||
standardHeartbeatsManager.reset();
|
||||
resetIdempotencyKeyCatalog();
|
||||
|
||||
// Wait for all streams to finish before completing the run
|
||||
waitUntil.register({
|
||||
|
||||
@@ -34,6 +34,7 @@ import {
|
||||
realtimeStreams,
|
||||
inputStreams,
|
||||
sessionStreams,
|
||||
resetIdempotencyKeyCatalog,
|
||||
} from "@trigger.dev/core/v3";
|
||||
import { TriggerTracer } from "@trigger.dev/core/v3/tracer";
|
||||
import {
|
||||
@@ -350,6 +351,7 @@ function resetExecutionEnvironment() {
|
||||
taskContext.disable();
|
||||
standardTraceContextManager.reset();
|
||||
standardHeartbeatsManager.reset();
|
||||
resetIdempotencyKeyCatalog();
|
||||
|
||||
// Wait for all streams to finish before completing the run
|
||||
waitUntil.register({
|
||||
|
||||
@@ -8,4 +8,5 @@ export type IdempotencyKeyOptions = {
|
||||
export interface IdempotencyKeyCatalog {
|
||||
registerKeyOptions(hash: string, options: IdempotencyKeyOptions): void;
|
||||
getKeyOptions(hash: string): IdempotencyKeyOptions | undefined;
|
||||
clear(): void;
|
||||
}
|
||||
|
||||
@@ -0,0 +1,56 @@
|
||||
import { describe, it, expect } from "vitest";
|
||||
import { InMemoryIdempotencyKeyCatalog } from "./inMemoryIdempotencyKeyCatalog.js";
|
||||
|
||||
describe("InMemoryIdempotencyKeyCatalog", () => {
|
||||
it("stores and retrieves options", () => {
|
||||
const catalog = new InMemoryIdempotencyKeyCatalog();
|
||||
const options = { key: "my-key", scope: "global" as const };
|
||||
|
||||
catalog.registerKeyOptions("hash1", options);
|
||||
|
||||
expect(catalog.getKeyOptions("hash1")).toEqual(options);
|
||||
});
|
||||
|
||||
it("returns undefined for non-existent keys", () => {
|
||||
const catalog = new InMemoryIdempotencyKeyCatalog();
|
||||
|
||||
expect(catalog.getKeyOptions("non-existent")).toBeUndefined();
|
||||
});
|
||||
|
||||
it("updates options when registering the same hash twice", () => {
|
||||
const catalog = new InMemoryIdempotencyKeyCatalog();
|
||||
|
||||
catalog.registerKeyOptions("hash1", { key: "key1", scope: "global" });
|
||||
catalog.registerKeyOptions("hash1", { key: "key1-updated", scope: "run" });
|
||||
|
||||
expect(catalog.getKeyOptions("hash1")).toEqual({ key: "key1-updated", scope: "run" });
|
||||
});
|
||||
|
||||
it("retains every entry regardless of count (no eviction)", () => {
|
||||
const catalog = new InMemoryIdempotencyKeyCatalog();
|
||||
const count = 5000;
|
||||
|
||||
for (let i = 0; i < count; i++) {
|
||||
catalog.registerKeyOptions(`hash${i}`, { key: `key${i}`, scope: "global" });
|
||||
}
|
||||
|
||||
// The very first entry must still be present — nothing is silently evicted.
|
||||
expect(catalog.getKeyOptions("hash0")).toEqual({ key: "key0", scope: "global" });
|
||||
expect(catalog.getKeyOptions(`hash${count - 1}`)).toEqual({
|
||||
key: `key${count - 1}`,
|
||||
scope: "global",
|
||||
});
|
||||
});
|
||||
|
||||
it("clear() removes all entries", () => {
|
||||
const catalog = new InMemoryIdempotencyKeyCatalog();
|
||||
|
||||
catalog.registerKeyOptions("hash1", { key: "key1", scope: "global" });
|
||||
catalog.registerKeyOptions("hash2", { key: "key2", scope: "run" });
|
||||
|
||||
catalog.clear();
|
||||
|
||||
expect(catalog.getKeyOptions("hash1")).toBeUndefined();
|
||||
expect(catalog.getKeyOptions("hash2")).toBeUndefined();
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,25 @@
|
||||
import type { IdempotencyKeyCatalog, IdempotencyKeyOptions } from "./catalog.js";
|
||||
|
||||
/**
|
||||
* Maps an idempotency-key hash back to the original user-provided key and scope.
|
||||
*
|
||||
* The mapping is held for the lifetime of a single run: the worker clears it at
|
||||
* each run boundary (warm starts reuse the process), so it never accumulates
|
||||
* across runs. Within a run every registered key is retained regardless of how
|
||||
* many are created, so the key/scope metadata is never silently dropped.
|
||||
*/
|
||||
export class InMemoryIdempotencyKeyCatalog implements IdempotencyKeyCatalog {
|
||||
private cache = new Map<string, IdempotencyKeyOptions>();
|
||||
|
||||
registerKeyOptions(hash: string, options: IdempotencyKeyOptions): void {
|
||||
this.cache.set(hash, options);
|
||||
}
|
||||
|
||||
getKeyOptions(hash: string): IdempotencyKeyOptions | undefined {
|
||||
return this.cache.get(hash);
|
||||
}
|
||||
|
||||
clear(): void {
|
||||
this.cache.clear();
|
||||
}
|
||||
}
|
||||
@@ -2,7 +2,7 @@ const API_NAME = "idempotency-key-catalog";
|
||||
|
||||
import { getGlobal, registerGlobal } from "../utils/globals.js";
|
||||
import type { IdempotencyKeyCatalog, IdempotencyKeyOptions } from "./catalog.js";
|
||||
import { LRUIdempotencyKeyCatalog } from "./lruIdempotencyKeyCatalog.js";
|
||||
import { InMemoryIdempotencyKeyCatalog } from "./inMemoryIdempotencyKeyCatalog.js";
|
||||
|
||||
export class IdempotencyKeyCatalogAPI {
|
||||
private static _instance?: IdempotencyKeyCatalogAPI;
|
||||
@@ -24,11 +24,15 @@ export class IdempotencyKeyCatalogAPI {
|
||||
return this.#getCatalog().getKeyOptions(hash);
|
||||
}
|
||||
|
||||
public clear(): void {
|
||||
this.#getCatalog().clear();
|
||||
}
|
||||
|
||||
#getCatalog(): IdempotencyKeyCatalog {
|
||||
let catalog = getGlobal(API_NAME);
|
||||
if (!catalog) {
|
||||
// Auto-initialize with LRU catalog on first access
|
||||
catalog = new LRUIdempotencyKeyCatalog();
|
||||
// Auto-initialize on first access
|
||||
catalog = new InMemoryIdempotencyKeyCatalog();
|
||||
registerGlobal(API_NAME, catalog, true);
|
||||
}
|
||||
return catalog;
|
||||
|
||||
@@ -1,209 +0,0 @@
|
||||
import { describe, it, expect } from "vitest";
|
||||
import { LRUIdempotencyKeyCatalog } from "./lruIdempotencyKeyCatalog.js";
|
||||
|
||||
describe("LRUIdempotencyKeyCatalog", () => {
|
||||
describe("registerKeyOptions and getKeyOptions", () => {
|
||||
it("should store and retrieve options", () => {
|
||||
const catalog = new LRUIdempotencyKeyCatalog();
|
||||
const options = { key: "my-key", scope: "global" as const };
|
||||
|
||||
catalog.registerKeyOptions("hash1", options);
|
||||
|
||||
expect(catalog.getKeyOptions("hash1")).toEqual(options);
|
||||
});
|
||||
|
||||
it("should return undefined for non-existent keys", () => {
|
||||
const catalog = new LRUIdempotencyKeyCatalog();
|
||||
|
||||
expect(catalog.getKeyOptions("non-existent")).toBeUndefined();
|
||||
});
|
||||
|
||||
it("should store multiple keys", () => {
|
||||
const catalog = new LRUIdempotencyKeyCatalog();
|
||||
const options1 = { key: "key1", scope: "global" as const };
|
||||
const options2 = { key: "key2", scope: "run" as const };
|
||||
const options3 = { key: "key3", scope: "attempt" as const };
|
||||
|
||||
catalog.registerKeyOptions("hash1", options1);
|
||||
catalog.registerKeyOptions("hash2", options2);
|
||||
catalog.registerKeyOptions("hash3", options3);
|
||||
|
||||
expect(catalog.getKeyOptions("hash1")).toEqual(options1);
|
||||
expect(catalog.getKeyOptions("hash2")).toEqual(options2);
|
||||
expect(catalog.getKeyOptions("hash3")).toEqual(options3);
|
||||
});
|
||||
|
||||
it("should update options when registering same key twice", () => {
|
||||
const catalog = new LRUIdempotencyKeyCatalog();
|
||||
const options1 = { key: "key1", scope: "global" as const };
|
||||
const options2 = { key: "key1-updated", scope: "run" as const };
|
||||
|
||||
catalog.registerKeyOptions("hash1", options1);
|
||||
catalog.registerKeyOptions("hash1", options2);
|
||||
|
||||
expect(catalog.getKeyOptions("hash1")).toEqual(options2);
|
||||
});
|
||||
});
|
||||
|
||||
describe("LRU eviction", () => {
|
||||
it("should evict oldest entry when over capacity", () => {
|
||||
const catalog = new LRUIdempotencyKeyCatalog(3);
|
||||
|
||||
catalog.registerKeyOptions("hash1", { key: "key1", scope: "global" });
|
||||
catalog.registerKeyOptions("hash2", { key: "key2", scope: "global" });
|
||||
catalog.registerKeyOptions("hash3", { key: "key3", scope: "global" });
|
||||
|
||||
// All three should exist
|
||||
expect(catalog.getKeyOptions("hash1")).toBeDefined();
|
||||
expect(catalog.getKeyOptions("hash2")).toBeDefined();
|
||||
expect(catalog.getKeyOptions("hash3")).toBeDefined();
|
||||
|
||||
// After the gets above, the LRU order from oldest to newest is: hash1, hash2, hash3
|
||||
// (each get moves the key to the most recent position)
|
||||
|
||||
// Add a fourth - hash1 should be evicted (it was accessed first, so it's the oldest)
|
||||
catalog.registerKeyOptions("hash4", { key: "key4", scope: "global" });
|
||||
|
||||
expect(catalog.getKeyOptions("hash1")).toBeUndefined();
|
||||
expect(catalog.getKeyOptions("hash2")).toBeDefined();
|
||||
expect(catalog.getKeyOptions("hash3")).toBeDefined();
|
||||
expect(catalog.getKeyOptions("hash4")).toBeDefined();
|
||||
});
|
||||
|
||||
it("should evict least recently registered entry when capacity exceeded", () => {
|
||||
const catalog = new LRUIdempotencyKeyCatalog(3);
|
||||
|
||||
catalog.registerKeyOptions("hash1", { key: "key1", scope: "global" });
|
||||
catalog.registerKeyOptions("hash2", { key: "key2", scope: "global" });
|
||||
catalog.registerKeyOptions("hash3", { key: "key3", scope: "global" });
|
||||
|
||||
// Adding fourth should evict hash1 (oldest)
|
||||
catalog.registerKeyOptions("hash4", { key: "key4", scope: "global" });
|
||||
|
||||
expect(catalog.getKeyOptions("hash1")).toBeUndefined();
|
||||
expect(catalog.getKeyOptions("hash2")).toBeDefined();
|
||||
expect(catalog.getKeyOptions("hash3")).toBeDefined();
|
||||
expect(catalog.getKeyOptions("hash4")).toBeDefined();
|
||||
});
|
||||
|
||||
it("should evict multiple entries when adding many at once would exceed capacity", () => {
|
||||
const catalog = new LRUIdempotencyKeyCatalog(2);
|
||||
|
||||
catalog.registerKeyOptions("hash1", { key: "key1", scope: "global" });
|
||||
catalog.registerKeyOptions("hash2", { key: "key2", scope: "global" });
|
||||
catalog.registerKeyOptions("hash3", { key: "key3", scope: "global" });
|
||||
catalog.registerKeyOptions("hash4", { key: "key4", scope: "global" });
|
||||
|
||||
// Only hash3 and hash4 should remain
|
||||
expect(catalog.getKeyOptions("hash1")).toBeUndefined();
|
||||
expect(catalog.getKeyOptions("hash2")).toBeUndefined();
|
||||
expect(catalog.getKeyOptions("hash3")).toBeDefined();
|
||||
expect(catalog.getKeyOptions("hash4")).toBeDefined();
|
||||
});
|
||||
|
||||
it("should work with maxSize of 1", () => {
|
||||
const catalog = new LRUIdempotencyKeyCatalog(1);
|
||||
|
||||
catalog.registerKeyOptions("hash1", { key: "key1", scope: "global" });
|
||||
expect(catalog.getKeyOptions("hash1")).toBeDefined();
|
||||
|
||||
catalog.registerKeyOptions("hash2", { key: "key2", scope: "global" });
|
||||
expect(catalog.getKeyOptions("hash1")).toBeUndefined();
|
||||
expect(catalog.getKeyOptions("hash2")).toBeDefined();
|
||||
});
|
||||
});
|
||||
|
||||
describe("LRU ordering", () => {
|
||||
it("should move accessed key to most recent position", () => {
|
||||
const catalog = new LRUIdempotencyKeyCatalog(3);
|
||||
|
||||
catalog.registerKeyOptions("hash1", { key: "key1", scope: "global" });
|
||||
catalog.registerKeyOptions("hash2", { key: "key2", scope: "global" });
|
||||
catalog.registerKeyOptions("hash3", { key: "key3", scope: "global" });
|
||||
|
||||
// Access hash1, moving it to most recent
|
||||
catalog.getKeyOptions("hash1");
|
||||
|
||||
// Add hash4 - should evict hash2 (now the oldest)
|
||||
catalog.registerKeyOptions("hash4", { key: "key4", scope: "global" });
|
||||
|
||||
expect(catalog.getKeyOptions("hash1")).toBeDefined();
|
||||
expect(catalog.getKeyOptions("hash2")).toBeUndefined();
|
||||
expect(catalog.getKeyOptions("hash3")).toBeDefined();
|
||||
expect(catalog.getKeyOptions("hash4")).toBeDefined();
|
||||
});
|
||||
|
||||
it("should move re-registered key to most recent position", () => {
|
||||
const catalog = new LRUIdempotencyKeyCatalog(3);
|
||||
|
||||
catalog.registerKeyOptions("hash1", { key: "key1", scope: "global" });
|
||||
catalog.registerKeyOptions("hash2", { key: "key2", scope: "global" });
|
||||
catalog.registerKeyOptions("hash3", { key: "key3", scope: "global" });
|
||||
|
||||
// Re-register hash1, moving it to most recent
|
||||
catalog.registerKeyOptions("hash1", { key: "key1-updated", scope: "run" });
|
||||
|
||||
// Add hash4 - should evict hash2 (now the oldest)
|
||||
catalog.registerKeyOptions("hash4", { key: "key4", scope: "global" });
|
||||
|
||||
expect(catalog.getKeyOptions("hash1")).toEqual({ key: "key1-updated", scope: "run" });
|
||||
expect(catalog.getKeyOptions("hash2")).toBeUndefined();
|
||||
expect(catalog.getKeyOptions("hash3")).toBeDefined();
|
||||
expect(catalog.getKeyOptions("hash4")).toBeDefined();
|
||||
});
|
||||
|
||||
it("should not affect order when getting non-existent key", () => {
|
||||
const catalog = new LRUIdempotencyKeyCatalog(2);
|
||||
|
||||
catalog.registerKeyOptions("hash1", { key: "key1", scope: "global" });
|
||||
catalog.registerKeyOptions("hash2", { key: "key2", scope: "global" });
|
||||
|
||||
// Try to get non-existent key
|
||||
catalog.getKeyOptions("non-existent");
|
||||
|
||||
// Add hash3 - should still evict hash1 (oldest)
|
||||
catalog.registerKeyOptions("hash3", { key: "key3", scope: "global" });
|
||||
|
||||
expect(catalog.getKeyOptions("hash1")).toBeUndefined();
|
||||
expect(catalog.getKeyOptions("hash2")).toBeDefined();
|
||||
expect(catalog.getKeyOptions("hash3")).toBeDefined();
|
||||
});
|
||||
});
|
||||
|
||||
describe("default maxSize", () => {
|
||||
it("should use default maxSize of 1000", () => {
|
||||
const catalog = new LRUIdempotencyKeyCatalog();
|
||||
|
||||
// Register 1001 entries
|
||||
for (let i = 0; i < 1001; i++) {
|
||||
catalog.registerKeyOptions(`hash${i}`, { key: `key${i}`, scope: "global" });
|
||||
}
|
||||
|
||||
// First entry should be evicted
|
||||
expect(catalog.getKeyOptions("hash0")).toBeUndefined();
|
||||
// Last entry should exist
|
||||
expect(catalog.getKeyOptions("hash1000")).toBeDefined();
|
||||
});
|
||||
});
|
||||
|
||||
describe("edge cases", () => {
|
||||
it("should handle negative maxSize by clamping to 0", () => {
|
||||
const catalog = new LRUIdempotencyKeyCatalog(-5);
|
||||
|
||||
// With maxSize clamped to 0, nothing should be stored
|
||||
catalog.registerKeyOptions("hash1", { key: "key1", scope: "global" });
|
||||
|
||||
// Should be immediately evicted since maxSize is 0
|
||||
expect(catalog.getKeyOptions("hash1")).toBeUndefined();
|
||||
});
|
||||
|
||||
it("should handle maxSize of 0", () => {
|
||||
const catalog = new LRUIdempotencyKeyCatalog(0);
|
||||
|
||||
catalog.registerKeyOptions("hash1", { key: "key1", scope: "global" });
|
||||
|
||||
// Should be immediately evicted since maxSize is 0
|
||||
expect(catalog.getKeyOptions("hash1")).toBeUndefined();
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -1,36 +0,0 @@
|
||||
import type { IdempotencyKeyCatalog, IdempotencyKeyOptions } from "./catalog.js";
|
||||
|
||||
export class LRUIdempotencyKeyCatalog implements IdempotencyKeyCatalog {
|
||||
private cache: Map<string, IdempotencyKeyOptions>;
|
||||
private readonly maxSize: number;
|
||||
|
||||
constructor(maxSize: number = 1_000) {
|
||||
this.cache = new Map();
|
||||
// Clamp to non-negative to prevent infinite loop in eviction
|
||||
this.maxSize = Math.max(0, maxSize);
|
||||
}
|
||||
|
||||
registerKeyOptions(hash: string, options: IdempotencyKeyOptions): void {
|
||||
// Delete and re-add to update position (most recently used)
|
||||
this.cache.delete(hash);
|
||||
this.cache.set(hash, options);
|
||||
|
||||
// Evict oldest entries if over capacity
|
||||
while (this.cache.size > this.maxSize) {
|
||||
const oldest = this.cache.keys().next().value;
|
||||
if (oldest !== undefined) {
|
||||
this.cache.delete(oldest);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
getKeyOptions(hash: string): IdempotencyKeyOptions | undefined {
|
||||
const options = this.cache.get(hash);
|
||||
if (options) {
|
||||
// Move to end (most recently used)
|
||||
this.cache.delete(hash);
|
||||
this.cache.set(hash, options);
|
||||
}
|
||||
return options;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,42 @@
|
||||
import { describe, it, expect } from "vitest";
|
||||
import {
|
||||
createIdempotencyKey,
|
||||
getIdempotencyKeyOptions,
|
||||
resetIdempotencyKeyCatalog,
|
||||
} from "./idempotencyKeys.js";
|
||||
|
||||
describe("idempotencyKeys metadata retention", () => {
|
||||
it("retains key/scope options for every key created in a run, even beyond 1000", async () => {
|
||||
const count = 3000;
|
||||
const keys: string[] = [];
|
||||
|
||||
for (let i = 0; i < count; i++) {
|
||||
const key = await createIdempotencyKey(`item-${i}`, { scope: "global" });
|
||||
keys.push(key);
|
||||
}
|
||||
|
||||
// The very first key created should still resolve its original options.
|
||||
// With a fixed-size LRU catalog (cap 1000), the earliest ~2000 keys are
|
||||
// silently evicted and this returns undefined.
|
||||
const firstOptions = getIdempotencyKeyOptions(keys[0]!);
|
||||
expect(firstOptions).toEqual({ key: "item-0", scope: "global" });
|
||||
|
||||
// Every key should resolve to its own original options.
|
||||
for (let i = 0; i < count; i++) {
|
||||
const options = getIdempotencyKeyOptions(keys[i]!);
|
||||
expect(options, `options missing for key index ${i}`).toEqual({
|
||||
key: `item-${i}`,
|
||||
scope: "global",
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
it("forgets options after the catalog is reset at a run boundary", async () => {
|
||||
const key = await createIdempotencyKey("boundary-key", { scope: "global" });
|
||||
expect(getIdempotencyKeyOptions(key)).toEqual({ key: "boundary-key", scope: "global" });
|
||||
|
||||
resetIdempotencyKeyCatalog();
|
||||
|
||||
expect(getIdempotencyKeyOptions(key)).toBeUndefined();
|
||||
});
|
||||
});
|
||||
@@ -38,6 +38,17 @@ export function getIdempotencyKeyOptions(
|
||||
return undefined;
|
||||
}
|
||||
|
||||
/**
|
||||
* Clears the in-process idempotency key catalog.
|
||||
*
|
||||
* The catalog maps an idempotency-key hash back to its original key and scope so
|
||||
* the SDK can attach that metadata when triggering. The worker calls this at each
|
||||
* run boundary so the mapping does not accumulate across warm-start runs.
|
||||
*/
|
||||
export function resetIdempotencyKeyCatalog(): void {
|
||||
idempotencyKeyCatalog.clear();
|
||||
}
|
||||
|
||||
export function isIdempotencyKey(
|
||||
value: string | string[] | IdempotencyKey
|
||||
): value is IdempotencyKey {
|
||||
|
||||
Reference in New Issue
Block a user