feat(mollifier): trigger burst smoothing — Phase 1 (monitoring)

[d-cs]

Summary

• Introduce the Mollifier: a Redis-backed buffer for trigger() API calls during traffic spikes, with a per-env trip evaluator and a drainer ack-loop.
• Phase 1 is dual-write monitoring — every mollified trigger is buffered to Redis AND continues to engine.trigger. No customer-facing behaviour change.
• Telemetry events: mollifier.would_mollify, mollifier.buffered, mollifier.drained, plus the mollifier.decisions counter.
• Gated behind a feature flag (default off).

Test plan

• pnpm run test --filter @trigger.dev/redis-worker
• pnpm run test --filter webapp -- mollifier
• Manual: with flag off, no behaviour change vs main
• Manual: with flag on + threshold lowered, observe mollifier.buffered + mollifier.drained log pairs with matching runId

[changeset-bot]

🦋 Changeset detected

Latest commit: e26f6ed

The changes in this PR will be included in the next version bump.

Not sure what this means? Click here to learn what changesets are.

Click here if you're a maintainer who wants to add another changeset to this PR

[coderabbitai]

Note

Reviews paused

It looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the reviews.auto_review.auto_pause_after_reviewed_commits setting.

Use the following commands to manage reviews:

• @coderabbitai resume to resume automatic reviews.
• @coderabbitai review to trigger a single review.

Use the checkboxes below for quick actions:

• ▶️ Resume reviews
• 🔍 Trigger review

Walkthrough

This PR implements the first two phases of a trigger burst-smoothing system ("mollifier"): it adds a Redis-backed MollifierBuffer and MollifierDrainer, Zod schemas and payload (de)serialization, a real trip evaluator and mollifier gate wired into RunEngineTriggerTaskService, OpenTelemetry metrics, worker startup drainer wiring, environment configuration and feature flag, package re-exports, and comprehensive tests (unit, integration, and fuzz) validating buffer, drainer, gate, and evaluator behavior while keeping fail-open semantics and deferring full activation to later phases.

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~60 minutes

🚥 Pre-merge checks | ✅ 3 | ❌ 2

❌ Failed checks (2 warnings)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 9.52% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.
Description check	⚠️ Warning	The PR description is incomplete and does not follow the required template structure. Missing required sections and structured information.	Add missing template sections: fill in issue reference, complete testing details, and add screenshots if applicable. Structure should match the provided template with all checkboxes and sections.

✅ Passed checks (3 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title clearly and specifically summarizes the main change: introducing the Mollifier feature for trigger burst smoothing with Phase 1 focusing on monitoring/dual-write behavior.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch mollifier-phase-2

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[d-cs]

Code review

Found 2 issues:

1. resolveOrgFlag resolves the global feature flag, not a per-org flag. flag() is called without overrides, so it reads only the global FeatureFlag row — other call sites in the webapp (e.g. canAccessAi/canAccessQuery) explicitly pass the org's featureFlags JSON as overrides. With the current code, the variable named orgFlagEnabled and the call-site comment "if the org has the mollifier feature flag enabled" don't match behaviour: the moment the global flag flips on, every org is mollified — there's no per-org rollout.

trigger.dev/apps/webapp/app/v3/mollifier/mollifierGate.server.ts

Lines 83 to 88 in 452ebda

	isMollifierEnabled: () => env.MOLLIFIER_ENABLED === "1",
	isShadowModeOn: () => env.MOLLIFIER_SHADOW_MODE === "1",
	resolveOrgFlag: () =>
	flag({ key: FEATURE_FLAG.mollifierEnabled, defaultValue: false }),
	evaluator: defaultEvaluator,
	logShadow: (inputs, decision) =>

2. evaluateGate is awaited outside the try block in the trigger hot path. evaluateGate itself has no internal try/catch around resolveOrgFlag() (a DB feature-flag read) or evaluator() (a Redis Lua call), so any transient failure once MOLLIFIER_ENABLED=1 is flipped will throw past the call site and fail the customer's trigger — turning a monitoring-only Phase 1 into a fleet-wide trigger outage on a Redis/DB blip. Suggest wrapping the gate call in a try/catch that falls back to pass_through.

trigger.dev/apps/webapp/app/runEngine/services/triggerTask.server.ts

Lines 338 to 346 in 452ebda

	};
	const mollifierOutcome = await this.evaluateGate({
	envId: environment.id,
	orgId: environment.organizationId,
	taskId,
	});
	try {

🤖 Generated with Claude Code

_{- If this code review was useful, please react with 👍. Otherwise, react with 👎.}

[d-cs]

Nitpicks (test-mock removal) also addressed in 98c1520:

• apps/webapp/test/mollifierGate.test.ts: dropped vi.fn. makeDeps now returns real closure-based dependencies with plain counter/array spies; the 16-row truth table and decision-log shape tests are unchanged.
• apps/webapp/test/mollifierTripEvaluator.test.ts: removed the vi.fn/cast fakeBuffer doubles. Each test now uses redisTest from @internal/testcontainers to spin up Redis and exercise a real MollifierBuffer. The trip case uses threshold: 2 + three real evaluateTrip calls to deterministically trip via the production Lua window; the fail-open case closes the buffer up front so the next evaluateTrip hits Connection is closed. — a real failure mode rather than a stub.

[d-cs]

Code review (follow-up)

Four additional issues from the earlier scan, posting now since the gate-naming fix landed:

1. Webapp CLAUDE.md says "Always use findFirst instead of findUnique" — new test introduces a findUnique. Trivial swap.

trigger.dev/apps/webapp/test/engine/triggerTask.test.ts

Lines 1343 to 1347 in 98c1520

	expect(result).toBeDefined();
	expect(result?.run.friendlyId).toBeDefined();
	const pgRun = await prisma.taskRun.findUnique({ where: { id: result!.run.id } });
	expect(pgRun).not.toBeNull();
	expect(pgRun!.friendlyId).toBe(result!.run.friendlyId);

2. Drainer is started inside init() unconditionally — workerQueue.initialize() two lines above is gated on WORKER_ENABLED === "true", but getMollifierDrainer() is not. Once MOLLIFIER_ENABLED=1 flips on, every webapp replica (including API-only ones) will spin up a polling loop + Redis connection and race for the same buffer. Consider gating on WORKER_ENABLED (or a dedicated MOLLIFIER_DRAINER_ENABLED).

trigger.dev/apps/webapp/app/services/worker.server.ts

Lines 129 to 152 in 98c1520

	if (env.WORKER_ENABLED === "true") {
	await workerQueue.initialize();
	}
	try {
	const drainer = getMollifierDrainer();
	if (drainer) {
	// The drainer owns a polling loop and a Redis client; let it drain
	// in-flight pops on shutdown rather than tearing the process down
	// mid-handler. Idempotent — `drainer.stop()` short-circuits if already
	// stopped, so registering on both signals is safe.
	const stopDrainer = () => {
	drainer.stop().catch((error) => {
	logger.error("Failed to stop mollifier drainer", { error });
	});
	};
	process.once("SIGTERM", stopDrainer);
	process.once("SIGINT", stopDrainer);
	}
	} catch (error) {
	logger.error("Failed to initialise mollifier drainer", { error });
	}
	}

3. includeTtl: true was added to the delayed and pending-version re-enqueue paths, but the producer-side comment in enqueueSystem.ts still explicitly warns these callers off: "Re-enqueues (waitpoint, checkpoint, delayed, pending version) must not add TTL." Either the comment needs updating to reflect the new design, or the new callers are wrong.

trigger.dev/internal-packages/run-engine/src/engine/systems/enqueueSystem.ts

Lines 88 to 95 in 98c1520

	const timestamp = (run.queueTimestamp ?? run.createdAt).getTime() - run.priorityMs;
	// Include TTL only when explicitly requested (first enqueue from trigger).
	// Re-enqueues (waitpoint, checkpoint, delayed, pending version) must not add TTL.
	let ttlExpiresAt: number | undefined;
	if (includeTtl && run.ttl) {
	const expireAt = parseNaturalLanguageDuration(run.ttl);

trigger.dev/internal-packages/run-engine/src/engine/systems/delayedRunSystem.ts

Lines 172 to 176 in 98c1520

	batchId: run.batchId ?? undefined,
	skipRunLock: true,
	includeTtl: true,
	});

trigger.dev/internal-packages/run-engine/src/engine/systems/pendingVersionSystem.ts

Lines 103 to 107 in 98c1520

	// for a worker version). Arm TTL here so the TTL system can expire it
	// if it sits queued waiting on a concurrency slot.
	includeTtl: true,
	});
	});

4. The comment calls count a "sliding-window counter", but the Lua is INCR + PEXPIRE-on-first — a fixed window. At a window boundary this can briefly admit ~2x threshold before tripping. Additionally PSETEX trippedKey, holdMs rearms the hold TTL on every overage call, so under sustained overload the hold never elapses while traffic stays above threshold. Both may be intentional for Phase 1, but the docstring should match the implementation.

trigger.dev/apps/webapp/app/v3/mollifier/mollifierGate.server.ts

Lines 11 to 16 in 98c1520

	} from "./mollifierTelemetry.server";
	// `count` is the *single-instance* sliding-window counter, not a fleet-wide
	// aggregate. Each webapp instance maintains its own Redis key, so the fleet
	// effective ceiling is `instance_count * threshold`. Phase 2 consumers must
	// not treat `count` as a global rate.

trigger.dev/packages/redis-worker/src/mollifier/buffer.ts

Lines 294 to 312 in 98c1520

	numberOfKeys: 2,
	lua: `
	local rateKey = KEYS[1]
	local trippedKey = KEYS[2]
	local windowMs = tonumber(ARGV[1])
	local threshold = tonumber(ARGV[2])
	local holdMs = tonumber(ARGV[3])
	local count = redis.call('INCR', rateKey)
	if count == 1 then
	redis.call('PEXPIRE', rateKey, windowMs)
	end
	if count > threshold then
	redis.call('PSETEX', trippedKey, holdMs, '1')
	end
	local tripped = redis.call('EXISTS', trippedKey)
	return {count, tripped}

🤖 Generated with Claude Code

_{- If this code review was useful, please react with 👍. Otherwise, react with 👎.}

[devin-ai-integration]

Devin Review found 1 new potential issue.

View 25 additional findings in Devin Review.