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: 5255c47

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 0 new potential issues.

View 14 additional findings in Devin Review.

[devin-ai-integration]

Devin Review found 1 new potential issue.

View 18 additional findings in Devin Review.

[devin-ai-integration]

🔴 Hard-coded env.TRIGGER_MOLLIFIER_ENABLED check bypasses DI'd evaluateGate, making mollifier integration tests unreachable

The short-circuit at triggerTask.server.ts:347 reads the global env singleton directly (env.TRIGGER_MOLLIFIER_ENABLED === "1") instead of going through the DI'd evaluateGate. When this env var is not set (defaults to "0" per env.server.ts:1057), mollifierOutcome is unconditionally null, so the injected gate is never called and the buffer-write branch at line 377 is unreachable.

This means three mollifier integration tests will fail:

• "mollify action triggers dual-write" (line 1328: expect(buffer.accepted).toHaveLength(1) gets 0)
• "engine.trigger throwing AFTER buffer.accept" (line 1491: same assertion)
• "debounce match produces an orphan buffer entry" (line 1730: same assertion)

The .env file does not exist in CI (confirmed: apps/webapp/.env is a symlink to ../../.env which is absent), and .env.example contains no TRIGGER_MOLLIFIER vars, so dotenv silently no-ops and the Zod default of "0" takes effect.

The DI pattern is correctly wired at construction (lines 91–105) but defeated by the env read at the call site

The constructor accepts optional evaluateGate / getMollifierBuffer hooks and stores them as instance fields. But the call site gates on the module-level env singleton before reaching those hooks. The fix is to make the enabled check part of the DI surface — e.g. check this.evaluateGate !== defaultEvaluateGate to skip the env guard when a custom gate is injected, or make the env check itself injectable.

Prompt for agents

The issue is in RunEngineTriggerTaskService.call() in apps/webapp/app/runEngine/services/triggerTask.server.ts at lines 346-355. The code reads env.TRIGGER_MOLLIFIER_ENABLED directly from the global env singleton to short-circuit the mollifier gate evaluation. This bypasses the DI'd evaluateGate that was injected via the constructor (lines 91-105), making the mollifier integration tests in test/engine/triggerTask.test.ts unreachable because the env var defaults to '0' in test environments.

The production intent is correct: skip the gate when the feature is globally off. The problem is that the env check is not part of the DI surface.

Possible approaches:
1. Add an isMollifierEnabled callback to the constructor options (like evaluateGate and getMollifierBuffer), defaulting to () => env.TRIGGER_MOLLIFIER_ENABLED === '1'. Tests can override it to return true.
2. Move the env check inside evaluateGate itself (it already has isMollifierEnabled in GateDependencies), and remove the outer check. The DI'd evaluateGate in tests already returns a fixed outcome, so the inner check is irrelevant. The production evaluateGate already checks isMollifierEnabled internally, so the outer check is redundant.
3. Skip the env guard when a non-default evaluateGate is injected (detect by comparing to the default import).

Approach 2 is simplest: just remove the ternary at lines 346-355 and always call this.evaluateGate. When TRIGGER_MOLLIFIER_ENABLED is '0' in production, the default evaluateGate's isMollifierEnabled() check returns false and it returns pass_through with an OTel counter increment — but the comment at line 340-345 explicitly says the short-circuit exists to avoid that counter increment. If that overhead is acceptable, approach 2 works. Otherwise approach 1 is the cleanest DI fix.

---

Was this helpful? React with 👍 or 👎 to provide feedback.