The task boundary: schemaTask and per-org queue

Right now, clicking Export invoices in the inspector throws an error. By the end of this lesson it fires a real run — validated, scoped to one org, deduped per day — and returns to the user instantly.

Returning instantly shapes everything else. Generating a CSV of every invoice an org owns can take minutes, and a web request can’t wait. So the click doesn’t do the export; it requests one. It writes a queued row, hands the work to a durable Trigger.dev task, and returns the page before the worker picks up the job. This boundary is what the rest of the chapter hangs off. The progress bar sits at 0/0 until the next lesson adds the real page count.

Your mission

The boundary has two sides: a Server Action that fires the task and returns, and a durable task that does the expensive work later. You only read the task; it’s already shipped. Open trigger/export-invoices.ts and confirm the four things that make it one: the queue declared once at module scope with queue({ name: 'export', concurrencyLimit: 1 }), the schemaTask whose schema is a z.strictObject over the two payload ids, the queue: exportQueue binding, and retry: { maxAttempts: 3 }. The task has no session, so the org and user ride in the payload, which schemaTask validates at the trigger edge — before the body runs, before a single retry is spent. That body is a placeholder for now: it sets pagesDone: 0 and returns { ok: true }, and the real work — pagination, email, the audit write — arrives in the next two lessons.

You write the action, src/lib/exports/start.ts. Its payload ids are z.string().min(1), not z.uuid(): the seed and Better Auth assign base62 ids like org_acme, which a uuid schema rejects. A concurrencyKey: organizationId at the trigger call, on that single queue at concurrencyLimit: 1, serializes runs within an org and parallelizes them across orgs. You fire with tasks.trigger, not triggerAndWait, which would block the request past its maxDuration. A same-day duplicate collapses under a 24-hour idempotency key built from (orgId, userId, dayBucket()). And authedAction pins the action to the member role, settling who may export before any work begins.

Firing the export inserts one queued exports row, fires export-invoices with the payload { organizationId, requestedBy }, and stamps the row with the returned runId.

tested

A second trigger the same day for the same org and user returns the first run’s runId and leaves a single exports row.

tested

A malformed payload — empty organizationId or an unexpected extra key — fails at the schema boundary before the body runs.

tested

A caller the member gate refuses is turned away before any row is written or task is fired.

tested

Clicking Export switches the inspector’s run panel to the new runId and the run reaches completed in the dashboard.

untested

Exports across different orgs run in parallel, each in its own concurrencyKey lane on the shared export queue.

untested

You can explain how a per-org lane at concurrencyLimit: 1 serializes two distinct same-org runs, even though the daily key collapses two same-org clicks into one.

untested

Coding time

Read the shipped boundary in trigger/export-invoices.ts, then implement src/lib/exports/start.ts against the brief and the tests. When you have a green run, or you’re stuck, open the walkthrough.

Reference solution and walkthrough

The shipped boundary

Read the boundary you’re triggering. Each part of the placeholder — the queue, the schema, the binding, the retry — carries a decision the chapter leans on.

import { metadata, queue, schemaTask } from '@trigger.dev/sdk/v3';
import { z } from 'zod';

// The per-org back-pressure lane, declared ONCE at module scope (the v4-native
// shape). `concurrencyLimit: 1` serializes runs within a lane; the per-org split
// comes from `concurrencyKey: organizationId` passed at the trigger call in
// startExport, NOT from a dynamically-named queue (the v3 shape v4 rejects).
export const exportQueue = queue({ name: 'export', concurrencyLimit: 1 });

// The durable parent task. `schemaTask` validates the strict payload at the trigger
// edge — never inside the body. organizationId/requestedBy ride in the payload
// because a task has no request context (no requireOrgUser); tenancy is re-derived
// from organizationId via tenantDb inside the run.
export const exportInvoices = schemaTask({
  id: 'export-invoices',
  schema: z.strictObject({
    organizationId: z.string().min(1),
    requestedBy: z.string().min(1),
  }),
  queue: exportQueue,
  retry: { maxAttempts: 3 },
  run: async (_payload) => {
    metadata.set('pagesDone', 0);
    return { ok: true };
  },
});

The queue is declared once, at module scope, in code. concurrencyLimit: 1 lets at most one run execute per lane at a time; the concurrencyKey the action passes at trigger time is what splits the queue into per-tenant lanes.

import { metadata, queue, schemaTask } from '@trigger.dev/sdk/v3';
import { z } from 'zod';

// The per-org back-pressure lane, declared ONCE at module scope (the v4-native
// shape). `concurrencyLimit: 1` serializes runs within a lane; the per-org split
// comes from `concurrencyKey: organizationId` passed at the trigger call in
// startExport, NOT from a dynamically-named queue (the v3 shape v4 rejects).
export const exportQueue = queue({ name: 'export', concurrencyLimit: 1 });

// The durable parent task. `schemaTask` validates the strict payload at the trigger
// edge — never inside the body. organizationId/requestedBy ride in the payload
// because a task has no request context (no requireOrgUser); tenancy is re-derived
// from organizationId via tenantDb inside the run.
export const exportInvoices = schemaTask({
  id: 'export-invoices',
  schema: z.strictObject({
    organizationId: z.string().min(1),
    requestedBy: z.string().min(1),
  }),
  queue: exportQueue,
  retry: { maxAttempts: 3 },
  run: async (_payload) => {
    metadata.set('pagesDone', 0);
    return { ok: true };
  },
});

This string is the task’s durable identity. Trigger.dev keys on the string, not the exported symbol, so a redeploy or a crashed run resumes against the same definition. Your action fires this exact string; rename the symbol freely, but the id is the contract. The v4 primitives were taught in Defining and triggering Trigger.dev tasks.

import { metadata, queue, schemaTask } from '@trigger.dev/sdk/v3';
import { z } from 'zod';

// The per-org back-pressure lane, declared ONCE at module scope (the v4-native
// shape). `concurrencyLimit: 1` serializes runs within a lane; the per-org split
// comes from `concurrencyKey: organizationId` passed at the trigger call in
// startExport, NOT from a dynamically-named queue (the v3 shape v4 rejects).
export const exportQueue = queue({ name: 'export', concurrencyLimit: 1 });

// The durable parent task. `schemaTask` validates the strict payload at the trigger
// edge — never inside the body. organizationId/requestedBy ride in the payload
// because a task has no request context (no requireOrgUser); tenancy is re-derived
// from organizationId via tenantDb inside the run.
export const exportInvoices = schemaTask({
  id: 'export-invoices',
  schema: z.strictObject({
    organizationId: z.string().min(1),
    requestedBy: z.string().min(1),
  }),
  queue: exportQueue,
  retry: { maxAttempts: 3 },
  run: async (_payload) => {
    metadata.set('pagesDone', 0);
    return { ok: true };
  },
});

schemaTask parses the payload at the trigger edge, before the body and before any retry. z.strictObject rejects an unexpected key; .min(1) rejects an empty id. The ids are z.string().min(1), not z.uuid(), because the seed’s base62 ids (org_acme, user_alice) would never pass a uuid schema. Match the schema to the ids you produce.

import { metadata, queue, schemaTask } from '@trigger.dev/sdk/v3';
import { z } from 'zod';

// The per-org back-pressure lane, declared ONCE at module scope (the v4-native
// shape). `concurrencyLimit: 1` serializes runs within a lane; the per-org split
// comes from `concurrencyKey: organizationId` passed at the trigger call in
// startExport, NOT from a dynamically-named queue (the v3 shape v4 rejects).
export const exportQueue = queue({ name: 'export', concurrencyLimit: 1 });

// The durable parent task. `schemaTask` validates the strict payload at the trigger
// edge — never inside the body. organizationId/requestedBy ride in the payload
// because a task has no request context (no requireOrgUser); tenancy is re-derived
// from organizationId via tenantDb inside the run.
export const exportInvoices = schemaTask({
  id: 'export-invoices',
  schema: z.strictObject({
    organizationId: z.string().min(1),
    requestedBy: z.string().min(1),
  }),
  queue: exportQueue,
  retry: { maxAttempts: 3 },
  run: async (_payload) => {
    metadata.set('pagesDone', 0);
    return { ok: true };
  },
});

Binding the task to the predeclared queue puts every run on the shared export lane. retry re-runs the body up to three times on a transient failure — the durability you exploit next lesson.

import { metadata, queue, schemaTask } from '@trigger.dev/sdk/v3';
import { z } from 'zod';

// The per-org back-pressure lane, declared ONCE at module scope (the v4-native
// shape). `concurrencyLimit: 1` serializes runs within a lane; the per-org split
// comes from `concurrencyKey: organizationId` passed at the trigger call in
// startExport, NOT from a dynamically-named queue (the v3 shape v4 rejects).
export const exportQueue = queue({ name: 'export', concurrencyLimit: 1 });

// The durable parent task. `schemaTask` validates the strict payload at the trigger
// edge — never inside the body. organizationId/requestedBy ride in the payload
// because a task has no request context (no requireOrgUser); tenancy is re-derived
// from organizationId via tenantDb inside the run.
export const exportInvoices = schemaTask({
  id: 'export-invoices',
  schema: z.strictObject({
    organizationId: z.string().min(1),
    requestedBy: z.string().min(1),
  }),
  queue: exportQueue,
  retry: { maxAttempts: 3 },
  run: async (_payload) => {
    metadata.set('pagesDone', 0);
    return { ok: true };
  },
});

The run body is a placeholder: set a zero progress value, return. Keeping it empty lets the run complete the moment you wire the trigger, proving the boundary works before any real work exists.

← Previous 1 / 1 Next →

Note

The import reads @trigger.dev/sdk/v3 even though this project runs Trigger.dev v4: the v4 package kept /v3 as its entry path, so every import in this chapter uses it.

Writing startExport

Now the file you own. startExport is a Server Action wrapped by authedAction, which resolves the session, enforces the role, parses the (empty) input schema, and hands your body an AuthedCtx with orgId, user, and an org-scoped db. Inside, the shape is: write the row, fire the task, stamp the row, return.

'use server';

import { idempotencyKeys, tasks } from '@trigger.dev/sdk/v3';
import { eq } from 'drizzle-orm';
import { revalidatePath } from 'next/cache';
import { z } from 'zod';

import { exports } from '@/db/schema';
import { authedAction } from '@/lib/auth/authed-action';
import { dayBucket } from '@/lib/exports/day-bucket';
import { err, ok, type Result } from '@/lib/result';

import type { exportInvoices } from '../../../trigger/export-invoices';

export const startExport = authedAction(
  'member',
  z.strictObject({}),
  async (_input, ctx): Promise<Result<{ runId: string }>> => {
    const bucket = dayBucket();

    const inserted = await ctx.db
      .insert(exports)
      .values({
        requestedBy: ctx.user.id,
        status: 'queued',
        dayBucket: bucket,
        runId: null,
      })
      .returning({ id: exports.id });
    const row = inserted[0];
    if (!row) {
      return err('internal', 'Could not record the export request.');
    }

    const handle = await tasks.trigger<typeof exportInvoices>(
      'export-invoices',
      { organizationId: ctx.orgId, requestedBy: ctx.user.id },
      {
        concurrencyKey: ctx.orgId,
        idempotencyKey: await idempotencyKeys.create(
          [ctx.orgId, ctx.user.id, bucket],
          { scope: 'global' },
        ),
        idempotencyKeyTTL: '24h',
        tags: [`org:${ctx.orgId}`],
      },
    );

    await ctx.db
      .update(exports)
      .set({ runId: handle.id })
      .where(eq(exports.id, row.id));

    revalidatePath('/inspector');
    return ok({ runId: handle.id });
  },
);

authedAction('member', ...) answers who may export: it rejects any caller below member before your body runs. The empty z.strictObject({}) takes no form input; the org and user come from the session.

'use server';

import { idempotencyKeys, tasks } from '@trigger.dev/sdk/v3';
import { eq } from 'drizzle-orm';
import { revalidatePath } from 'next/cache';
import { z } from 'zod';

import { exports } from '@/db/schema';
import { authedAction } from '@/lib/auth/authed-action';
import { dayBucket } from '@/lib/exports/day-bucket';
import { err, ok, type Result } from '@/lib/result';

import type { exportInvoices } from '../../../trigger/export-invoices';

export const startExport = authedAction(
  'member',
  z.strictObject({}),
  async (_input, ctx): Promise<Result<{ runId: string }>> => {
    const bucket = dayBucket();

    const inserted = await ctx.db
      .insert(exports)
      .values({
        requestedBy: ctx.user.id,
        status: 'queued',
        dayBucket: bucket,
        runId: null,
      })
      .returning({ id: exports.id });
    const row = inserted[0];
    if (!row) {
      return err('internal', 'Could not record the export request.');
    }

    const handle = await tasks.trigger<typeof exportInvoices>(
      'export-invoices',
      { organizationId: ctx.orgId, requestedBy: ctx.user.id },
      {
        concurrencyKey: ctx.orgId,
        idempotencyKey: await idempotencyKeys.create(
          [ctx.orgId, ctx.user.id, bucket],
          { scope: 'global' },
        ),
        idempotencyKeyTTL: '24h',
        tags: [`org:${ctx.orgId}`],
      },
    );

    await ctx.db
      .update(exports)
      .set({ runId: handle.id })
      .where(eq(exports.id, row.id));

    revalidatePath('/inspector');
    return ok({ runId: handle.id });
  },
);

Write the queued row before the trigger fires, for two reasons: the daily idempotency key needs a row to dedup against, and you want a durable record even if the trigger call fails. ctx.db is the org-scoped handle, so organizationId is stamped for you — that’s why it’s absent from .values.

'use server';

import { idempotencyKeys, tasks } from '@trigger.dev/sdk/v3';
import { eq } from 'drizzle-orm';
import { revalidatePath } from 'next/cache';
import { z } from 'zod';

import { exports } from '@/db/schema';
import { authedAction } from '@/lib/auth/authed-action';
import { dayBucket } from '@/lib/exports/day-bucket';
import { err, ok, type Result } from '@/lib/result';

import type { exportInvoices } from '../../../trigger/export-invoices';

export const startExport = authedAction(
  'member',
  z.strictObject({}),
  async (_input, ctx): Promise<Result<{ runId: string }>> => {
    const bucket = dayBucket();

    const inserted = await ctx.db
      .insert(exports)
      .values({
        requestedBy: ctx.user.id,
        status: 'queued',
        dayBucket: bucket,
        runId: null,
      })
      .returning({ id: exports.id });
    const row = inserted[0];
    if (!row) {
      return err('internal', 'Could not record the export request.');
    }

    const handle = await tasks.trigger<typeof exportInvoices>(
      'export-invoices',
      { organizationId: ctx.orgId, requestedBy: ctx.user.id },
      {
        concurrencyKey: ctx.orgId,
        idempotencyKey: await idempotencyKeys.create(
          [ctx.orgId, ctx.user.id, bucket],
          { scope: 'global' },
        ),
        idempotencyKeyTTL: '24h',
        tags: [`org:${ctx.orgId}`],
      },
    );

    await ctx.db
      .update(exports)
      .set({ runId: handle.id })
      .where(eq(exports.id, row.id));

    revalidatePath('/inspector');
    return ok({ runId: handle.id });
  },
);

tasks.trigger returns the moment the run is enqueued. The <typeof exportInvoices> type argument checks the payload against the task’s schema, so a mistyped key is a compile error. The ids ride in the payload because the task has no session to read them from.

'use server';

import { idempotencyKeys, tasks } from '@trigger.dev/sdk/v3';
import { eq } from 'drizzle-orm';
import { revalidatePath } from 'next/cache';
import { z } from 'zod';

import { exports } from '@/db/schema';
import { authedAction } from '@/lib/auth/authed-action';
import { dayBucket } from '@/lib/exports/day-bucket';
import { err, ok, type Result } from '@/lib/result';

import type { exportInvoices } from '../../../trigger/export-invoices';

export const startExport = authedAction(
  'member',
  z.strictObject({}),
  async (_input, ctx): Promise<Result<{ runId: string }>> => {
    const bucket = dayBucket();

    const inserted = await ctx.db
      .insert(exports)
      .values({
        requestedBy: ctx.user.id,
        status: 'queued',
        dayBucket: bucket,
        runId: null,
      })
      .returning({ id: exports.id });
    const row = inserted[0];
    if (!row) {
      return err('internal', 'Could not record the export request.');
    }

    const handle = await tasks.trigger<typeof exportInvoices>(
      'export-invoices',
      { organizationId: ctx.orgId, requestedBy: ctx.user.id },
      {
        concurrencyKey: ctx.orgId,
        idempotencyKey: await idempotencyKeys.create(
          [ctx.orgId, ctx.user.id, bucket],
          { scope: 'global' },
        ),
        idempotencyKeyTTL: '24h',
        tags: [`org:${ctx.orgId}`],
      },
    );

    await ctx.db
      .update(exports)
      .set({ runId: handle.id })
      .where(eq(exports.id, row.id));

    revalidatePath('/inspector');
    return ok({ runId: handle.id });
  },
);

concurrencyLimit: 1 applies per concurrencyKey, so org_acme’s runs serialize among themselves, org_globex’s among themselves, and the two orgs run in parallel. Only the key varies per call.

'use server';

import { idempotencyKeys, tasks } from '@trigger.dev/sdk/v3';
import { eq } from 'drizzle-orm';
import { revalidatePath } from 'next/cache';
import { z } from 'zod';

import { exports } from '@/db/schema';
import { authedAction } from '@/lib/auth/authed-action';
import { dayBucket } from '@/lib/exports/day-bucket';
import { err, ok, type Result } from '@/lib/result';

import type { exportInvoices } from '../../../trigger/export-invoices';

export const startExport = authedAction(
  'member',
  z.strictObject({}),
  async (_input, ctx): Promise<Result<{ runId: string }>> => {
    const bucket = dayBucket();

    const inserted = await ctx.db
      .insert(exports)
      .values({
        requestedBy: ctx.user.id,
        status: 'queued',
        dayBucket: bucket,
        runId: null,
      })
      .returning({ id: exports.id });
    const row = inserted[0];
    if (!row) {
      return err('internal', 'Could not record the export request.');
    }

    const handle = await tasks.trigger<typeof exportInvoices>(
      'export-invoices',
      { organizationId: ctx.orgId, requestedBy: ctx.user.id },
      {
        concurrencyKey: ctx.orgId,
        idempotencyKey: await idempotencyKeys.create(
          [ctx.orgId, ctx.user.id, bucket],
          { scope: 'global' },
        ),
        idempotencyKeyTTL: '24h',
        tags: [`org:${ctx.orgId}`],
      },
    );

    await ctx.db
      .update(exports)
      .set({ runId: handle.id })
      .where(eq(exports.id, row.id));

    revalidatePath('/inspector');
    return ok({ runId: handle.id });
  },
);

The idempotency key is derived from (org, user, day), so two clicks by the same user, same org, same day produce the same key, and Trigger.dev returns the first run’s handle instead of a new run. scope: 'global' makes the key app-level, not namespaced to a parent run; the 24h TTL lets the next day export again. Idempotency-key scopes were taught in Retries, waits, idempotency.

'use server';

import { idempotencyKeys, tasks } from '@trigger.dev/sdk/v3';
import { eq } from 'drizzle-orm';
import { revalidatePath } from 'next/cache';
import { z } from 'zod';

import { exports } from '@/db/schema';
import { authedAction } from '@/lib/auth/authed-action';
import { dayBucket } from '@/lib/exports/day-bucket';
import { err, ok, type Result } from '@/lib/result';

import type { exportInvoices } from '../../../trigger/export-invoices';

export const startExport = authedAction(
  'member',
  z.strictObject({}),
  async (_input, ctx): Promise<Result<{ runId: string }>> => {
    const bucket = dayBucket();

    const inserted = await ctx.db
      .insert(exports)
      .values({
        requestedBy: ctx.user.id,
        status: 'queued',
        dayBucket: bucket,
        runId: null,
      })
      .returning({ id: exports.id });
    const row = inserted[0];
    if (!row) {
      return err('internal', 'Could not record the export request.');
    }

    const handle = await tasks.trigger<typeof exportInvoices>(
      'export-invoices',
      { organizationId: ctx.orgId, requestedBy: ctx.user.id },
      {
        concurrencyKey: ctx.orgId,
        idempotencyKey: await idempotencyKeys.create(
          [ctx.orgId, ctx.user.id, bucket],
          { scope: 'global' },
        ),
        idempotencyKeyTTL: '24h',
        tags: [`org:${ctx.orgId}`],
      },
    );

    await ctx.db
      .update(exports)
      .set({ runId: handle.id })
      .where(eq(exports.id, row.id));

    revalidatePath('/inspector');
    return ok({ runId: handle.id });
  },
);

The handle carries the run’s id; stamp it onto the queued row and return it. The inspector reads that runId and points its poller at the new run.

← Previous 1 / 1 Next →

The v3-to-v4 queue break. v3 let you name a queue dynamically per tenant; v4 rejects that. A queue is now a declared, named resource, and per-tenant back-pressure is a concurrencyKey on it.

v3 (rejected)

// A new queue name per tenant, limit set at the trigger call — v4 rejects this.
await tasks.trigger('export-invoices', payload, {
  queue: { name: `export-${orgId}`, concurrencyLimit: 1 },
});

The shape v4 rejects. A queue per org at trigger time scaled badly — one queue resource per tenant.

v4 (this project)

// One queue, declared once in code.
export const exportQueue = queue({ name: 'export', concurrencyLimit: 1 });

// Per-org isolation is the key, not the queue name.
await tasks.trigger('export-invoices', payload, {
  concurrencyKey: orgId,
});

The shape the project ships. One named queue, the org id as the concurrencyKey, so each org gets its own serial lane on the shared queue.

Why tasks.trigger, not triggerAndWait. triggerAndWait belongs inside a task body, where blocking on a child is free; in a Server Action it would hang the user’s request past maxDuration. From the action you only fire and return.

The two-step write. Insert, then update with the runId after the trigger returns — two writes around a network call. A trigger failure after the insert leaves a rare orphan queued row that never gets a run. Production hardening would wrap the trigger in a transaction; this project stays at two statements, since the orphan is harmless.

metadata is a module import. The boundary file imports metadata and calls metadata.set(...); it is not a field on the run’s second argument, so destructuring { metadata } off the run params fails the type-checker.

Trigger.dev — schemaTask

trigger.dev

Validating a strict payload at the trigger edge — the boundary you read before writing the action.

Trigger.dev — Concurrency & queues

trigger.dev

concurrencyLimit on a declared queue plus concurrencyKey — the per-org lane this lesson installs.

Trigger.dev — Idempotency

trigger.dev

idempotencyKeys.create with global scope and a TTL — the same-day dedup key.

Trigger.dev — Triggering

trigger.dev

tasks.trigger vs triggerAndWait — why the action fires and returns instead of waiting.

Verify the boundary

Run the lesson’s test suite:

pnpm test:lesson 2

With the Trigger.dev runtime out of process, the suite drives startExport through the real authedAction, fakes only the infrastructure, and parses the shipped payload schema directly. Expect all four requirements green.

pnpm test:lesson 2

 ✓ tests/lessons/Lesson 2.test.ts (11 tests)
   ✓ Requirement 1 — insert → trigger → runId update (3)
   ✓ Requirement 2 — the daily key short-circuits a duplicate (2)
   ✓ Requirement 3 — the schemaTask payload boundary rejects bad input (4)
   ✓ Requirement 4 — the member gate fires nothing when refused (2)

 Test Files  1 passed (1)
      Tests  11 passed (11)

The tests cover only what runs in-process. The rest of the boundary needs a live worker, so run npx trigger.dev@latest dev alongside the app and walk the checklist by hand against the dashboard and browser.

Click Export invoices for the active org. The inspector switches its run panel to the new runId; the dashboard shows one run, status: completed, payload { organizationId, requestedBy }; the progress bar reads 0/0 (page count comes next lesson).

untested

Click Export twice quickly. The second click returns the same runId; there is one exports row and one dashboard run.

untested

From the dashboard’s run-task tool, fire export-invoices with { organizationId: '' } or an extra key. It fails immediately at the Zod parse — the body never runs.

untested

Click Trigger 2 (same org). Both submits share the daily key and collapse to one run. Then reason through how a concurrencyKey lane at limit 1 would serialize two distinct same-org runs.

untested

Switch the acting org with the dev switcher and fire Trigger 3 (cross org). Each org’s run reaches executing in its own lane on the shared queue, in parallel.

untested