Eric Allam
a3407287c9
## Summary Currently, every triggered run follows a two-step path through Redis: 1. **Enqueue** — A Lua script atomically adds the message to a queue sorted set (ordered by priority-adjusted timestamp) 2. **Dequeue** — A debounced `processQueueForWorkerQueue` job fires ~500ms later, checks concurrency limits, removes the message from the sorted set, and pushes it to a worker queue (Redis list) where workers pick it up via `BLPOP` This means every run pays at least ~500ms of latency between being triggered and being available for a worker to execute, even when the queue is empty and concurrency is wide open. ### What changed The enqueue Lua scripts now atomically decide whether to **skip the queue sorted set entirely** and push directly to the worker queue. This happens inside the same Lua script that handles normal enqueue, so the decision is atomic with respect to concurrency bookkeeping. A run takes the **fast path** when all of these are true: - **Fast path is enabled** for this worker queue (gated per `WorkerInstanceGroup`) - **No available messages** in the queue (`ZRANGEBYSCORE` finds nothing with score ≤ now) — this respects priority ordering and allows fast path even when the queue has future-scored messages (e.g. nacked retries with delay) - **Environment concurrency** has capacity - **Queue concurrency** has capacity (including per-concurrency-key limits for CK queues) When the fast path is taken: - The message is stored and pushed directly to the worker queue (`RPUSH`) - Concurrency slots are claimed (`SADD` to the same sets used by the normal dequeue path) - The `processQueueForWorkerQueue` job is **not scheduled** (no work to do) - TTL sorted set is skipped (the `expireRun` worker job handles TTL independently) When any condition fails, the existing slow path runs unchanged. ### Rollout gating - **Development environments**: Fast path is always enabled - **Production environments**: Gated by a new `enableFastPath` boolean on `WorkerInstanceGroup` (defaults to `false`), allowing region-by-region rollout ### Rolling deploy safety Each process registers its own Lua scripts via `defineCommand` (identified by SHA hash). Old and new processes never share scripts. The Redis data structures are fully compatible in both directions — ack, nack, and release operations work identically regardless of which path a message took. ## Test plan - [x] Fast path taken when queue is empty and concurrency available - [x] Slow path when `enableFastPath` is false - [x] Slow path when queue has available messages (respects priority ordering) - [x] Fast path when queue only has future-scored messages - [x] Slow path when env concurrency is full - [x] Fast-path message can be acknowledged correctly - [x] Fast-path message can be nacked and re-enqueued to the queue sorted set - [x] Run all existing run-queue tests (ack, nack, CK, concurrency sweeper, dequeue) to verify no regressions - [x] Typecheck passes for run-engine and webapp
@trigger.dev/database
This is the internal database package for the Trigger.dev project. It exports a generated prisma client that can be instantiated with a connection string.
How to switch branches when you've done migrations
Sometimes you've applied migrations and then want to switch branches without wiping out your local database.
To do this you can run the following command:
DB_VOLUME=database-data-alt pnpm run docker
This will switch to the alt volume for your local database. This database will be blank if you haven't switched to this volume before, so you'll need to follow the normal steps (in the Contributing guide) to get setup, e.g. apply migrations and seed.
To switch back to the original volume, run the following command:
pnpm run docker
How to add a new index on a large table
- Modify the Prisma.schema with a single index change (no other changes, just one index at a time)
- Create a Prisma migration using
cd internal-packages/database && pnpm run db:migrate:dev:create - Modify the SQL file: add IF NOT EXISTS to it and CONCURRENTLY:
CREATE INDEX CONCURRENTLY IF NOT EXISTS "JobRun_eventId_idx" ON "JobRun" ("eventId");
- Don’t apply the Prisma migration locally yet. This is a good opportunity to test the flow.
- Manually apply the index to your database, by running the index command.
- Then locally run
pnpm run db:migrate:deploy
Before deploying
Run the index creation before deploying
CREATE INDEX CONCURRENTLY IF NOT EXISTS "JobRun_eventId_idx" ON "JobRun" ("eventId");
These commands are useful:
-- creates an index safely, this can take a long time (2 mins maybe)
CREATE INDEX CONCURRENTLY IF NOT EXISTS "JobRun_eventId_idx" ON "JobRun" ("eventId");
-- checks the status of an index
SELECT * FROM pg_stat_progress_create_index WHERE relid = '"JobRun"'::regclass;
-- checks if the index is there
SELECT * FROM pg_indexes WHERE tablename = 'JobRun' AND indexname = 'JobRun_eventId_idx';
Now, when you deploy and prisma runs the migration, it will skip the index creation because it already exists. If you don't do this, there will be pain.