feat(convex): on-demand run synchronization to the integration

Add on-demand synchronization to `@trigger.dev/convex` with
`trigger.refresh(ctx, componentRunId)`.

Applications can use Trigger Realtime for responsive progress, then
request an earlier authoritative status check when Realtime reports
completion. The component keeps its existing polling and watchdog as
recovery when no browser is connected.

- Refresh requests are scheduled asynchronously and coalesced per run.
- Refresh-driven status checks start at least five seconds after the
previous status check. Hints during an in-flight request queue one
follow-up without discarding that response.
- Pending submissions, paused synchronization, and completed records are
no-ops. Refresh does not trigger another task, reset retry failures, or
resume paused work.
- Existing terminal-state handling, transactional completion callbacks,
output budgets, generation fencing, and request/credential cleanup are
reused.
- Optional internal coordination fields require no record migration and
are excluded from the public run shape.
- README guidance includes an ownership-checked mutation, the
Realtime/polling split, and the warning that Realtime column exclusions
do not prevent a run-read token from retrieving payloads through REST.

No Trigger server changes, persistent subscriptions, new dependencies,
or token-issuing API are added.

Mono-RevId: 91a3884a75fe0972740ee42cd6094a82ae036c63

Author: carderne

.changeset/convex-refresh.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/convex": patch
+---
+
+Add `trigger.refresh(ctx, id)` to request an earlier status check for a Convex-tracked run, so applications using Trigger Realtime can publish results without waiting for the next scheduled poll. Requests are coalesced and rate-limited, use the existing completion callback, and retain polling as a recovery fallback.

packages/convex/README.md

@@ -124,6 +124,7 @@ The ID returned by `trigger.trigger` identifies the **component record**, not th
 - `trigger(ctx, taskId, payload, options?)`: enqueue work. Calling from a mutation makes the request atomic with your application's writes.
 - `getRun(ctx, id)`: read one record reactively.
 - `listRuns(ctx, paginationOpts)`: paginate records, newest first, up to 100 per page and a bounded read budget.
+- `refresh(ctx, id)`: schedule an earlier status check for an active, submitted run. Returns `null` after scheduling, not the updated run.
 - `resume(ctx, id)`: resume paused synchronization using the current credentials.
 
 Trigger options include tags, delay, queue name, concurrency key, machine, metadata, max attempts, max duration, TTL, and a component-level idempotency key.
@@ -134,6 +135,35 @@ The constructor accepts `secretKey` (defaults to `TRIGGER_SECRET_KEY`), `baseUrl
 
 **Authorization belongs in your application's functions.** The component has no tenant or user model. Do not expose arbitrary record IDs or `listRuns` to clients without enforcing ownership. Queries never return stored credentials or request payloads.
 
+## Realtime progress and faster result delivery
+
+Use [Trigger Realtime](https://trigger.dev/docs/realtime/overview) for responsive browser progress, and Convex queries for durable application state. When Realtime reports that a run is complete, call an ownership-checked application mutation that requests a component refresh:
+
+```ts
+// convex/jobs.ts
+export const refresh = mutation({
+  args: { jobId: v.id("jobs") },
+  handler: async (ctx, { jobId }) => {
+    const user = await ctx.auth.getUserIdentity();
+    const job = await ctx.db.get(jobId);
+    if (!user || !job || job.owner !== user.subject) throw new Error("Unauthorized");
+    if (job.runId) await trigger.refresh(ctx, job.runId);
+  },
+});
+```
+
+Pass the component record ID to `refresh`, not the remote Trigger run ID. Your browser subscribes to the latter, available as `run.runId` from `getRun`.
+
+A refresh is only a hint: the component retrieves status and output directly from Trigger and uses the same validation, completion callback, and credential cleanup as normal polling. It never accepts client-supplied status or output. Observe `getRun` or your application query for the persisted result; awaiting `refresh` does not wait for that result.
+
+- Refresh requests are coalesced per run. Status checks requested this way start at least five seconds after the previous status check.
+- A hint received during a check queues one follow-up instead of interrupting or discarding the in-flight response. If that response completes the run, no follow-up is needed.
+- Pending submissions, paused synchronization, and completed records are no-ops. Refresh never submits a task or resumes paused work; use `resume` after repairing a paused synchronization.
+- Call refresh when Realtime observes completion, including an already-completed run after reconnecting. Do not call it on every metadata update to animate progress.
+- Scheduled polling and its watchdog remain enabled. Convex can still finish the run if the browser closes, the subscription disconnects, or the refresh request fails. No persistent subscription inside a Convex action is required.
+
+**Realtime token safety:** authenticate the token request, verify job ownership, and issue a short-lived token scoped to the exact remote run. Keep the root Trigger key server-side. A run-read token can also read the run through REST, including its payload; Realtime `skipColumns` is not a restriction on REST access. Keep worker credentials, storage-upload capabilities, and other secrets out of token-readable payloads, outputs, metadata, and errors. Application mutations must enforce ownership even when component refreshes are coalesced.
+
 ## Execution and reliability
 
 1. The calling mutation writes a component record and schedules submission. If the transaction rolls back, neither survives.

packages/convex/src/client/index.ts

@@ -75,6 +75,14 @@ export class Trigger {
     return ctx.runQuery(this.component.runs.list, { paginationOpts });
   }
 
+  /**
+   * Request an earlier status check without accepting client-supplied run data.
+   * The check is scheduled, coalesced, and rate-limited; observe getRun for its result.
+   */
+  async refresh(ctx: MutationCtx, id: string): Promise<null> {
+    return ctx.runMutation(this.component.runs.refresh, { id });
+  }
+
   async resume(ctx: MutationCtx, id: string): Promise<null> {
     return ctx.runMutation(this.component.runs.resume, { id, config: this.runtimeConfig() });
   }

packages/convex/src/component/runs.test.ts

@@ -6,7 +6,7 @@ import { defineSchema, defineTable } from "convex/server";
 import { v } from "convex/values";
 import { Trigger, vRun, type ComponentApi } from "../client/index.js";
 import schema from "./schema.js";
-import { MAX_INLINE_BYTES } from "./shared.js";
+import { MAX_INLINE_BYTES, MIN_REFRESH_INTERVAL_MS } from "./shared.js";
 
 const modules = {
   "./_generated/component.ts": () => import("./_generated/component.js"),
@@ -63,6 +63,10 @@ const { createTest, api, internal } = app.defineModules({
         return id;
       },
     }),
+    refresh: app.mutation({
+      args: { id: v.string() },
+      handler: async (ctx, { id }) => new Trigger(app.components.trigger).refresh(ctx, id),
+    }),
   },
 });
 
@@ -92,6 +96,8 @@ let remoteStatus: string;
 let remoteIsCompleted: boolean;
 let largeOutput: boolean;
 let submittedOptions: Record<string, unknown>;
+let holdNextPoll: boolean;
+let releasePoll: (() => void) | undefined;
 
 beforeAll(async () => {
   server = createServer(async (req, res) => {
@@ -125,19 +131,26 @@ beforeAll(async () => {
         return;
       }
       const completed = activePolls-- <= 0;
-      res.end(
-        JSON.stringify({
-          id: "run_1",
-          status: completed ? remoteStatus : "EXECUTING",
-          isCompleted: completed && remoteIsCompleted,
-          output: completed
-            ? largeOutput
-              ? "x".repeat(MAX_INLINE_BYTES + 1)
-              : { url: "https://example.com/video.mp4" }
-            : undefined,
-          metadata: { progress: completed ? 100 : 50 },
-        })
-      );
+      const body = JSON.stringify({
+        id: "run_1",
+        status: completed ? remoteStatus : "EXECUTING",
+        isCompleted: completed && remoteIsCompleted,
+        output: completed
+          ? largeOutput
+            ? "x".repeat(MAX_INLINE_BYTES + 1)
+            : { url: "https://example.com/video.mp4" }
+          : undefined,
+        metadata: { progress: completed ? 100 : 50 },
+      });
+      if (holdNextPoll) {
+        holdNextPoll = false;
+        releasePoll = () => {
+          if (!res.writableEnded) res.end(body);
+          releasePoll = undefined;
+        };
+      } else {
+        res.end(body);
+      }
     }
   });
   server.listen(0, "127.0.0.1");
@@ -162,8 +175,13 @@ beforeEach(() => {
   remoteIsCompleted = true;
   largeOutput = false;
   submittedOptions = {};
+  holdNextPoll = false;
+  releasePoll = undefined;
+});
+afterEach(() => {
+  releasePoll?.();
+  vi.useRealTimers();
 });
-afterEach(() => vi.useRealTimers());
 afterAll(
   () =>
     new Promise<void>((resolve, reject) => {
@@ -344,6 +362,135 @@ test("a failing callback rolls back completion and can be retried without duplic
   expect(await t.run((ctx) => ctx.db.query("completions").collect())).toHaveLength(1);
 });
 
+async function runScheduled(t: ReturnType<typeof createTest>, advanceMs = 0): Promise<void> {
+  // Convex timestamps have sub-millisecond precision, so zero-delay jobs may fall just after now.
+  vi.advanceTimersByTime(advanceMs + 1);
+  await t.finishInProgressScheduledFunctions();
+  vi.advanceTimersByTime(1);
+  await t.finishInProgressScheduledFunctions();
+}
+
+// Catches: refresh bursts bypassing the check cooldown or racing the polling callback.
+test("refreshes promptly, bounds repeated checks, and completes once alongside polling", async () => {
+  activePolls = 1;
+  const t = createTest();
+  const id = await t.mutation(api.jobs.request, {});
+  await runScheduled(t);
+  expect(submissions).toBe(1);
+  expect(polls).toBe(0);
+
+  await Promise.all(Array.from({ length: 3 }, () => t.mutation(api.jobs.refresh, { id })));
+  await runScheduled(t);
+  expect(polls).toBe(1);
+  expect(await t.query(app.components.trigger.runs.get, { id })).toMatchObject({
+    status: "EXECUTING",
+    metadata: { progress: 50 },
+  });
+
+  await Promise.all(Array.from({ length: 3 }, () => t.mutation(api.jobs.refresh, { id })));
+  await runScheduled(t, MIN_REFRESH_INTERVAL_MS / 2);
+  expect(polls).toBe(1);
+  await runScheduled(t, MIN_REFRESH_INTERVAL_MS / 2);
+  expect(polls).toBe(2);
+  expect(await t.query(app.components.trigger.runs.get, { id })).toMatchObject({
+    status: "COMPLETED",
+    syncStatus: "complete",
+    output: { url: "https://example.com/video.mp4" },
+  });
+  await t.mutation(api.jobs.refresh, { id });
+  await drain(t);
+  expect(polls).toBe(2);
+  expect(submissions).toBe(1);
+  expect(await t.run((ctx) => ctx.db.query("completions").collect())).toHaveLength(1);
+});
+
+// Catches: refresh fencing out an in-flight observation or losing its completion hint.
+test("coalesces hints during an in-flight request and follows it with another check", async () => {
+  activePolls = 1;
+  const t = createTest();
+  const id = await t.mutation(api.jobs.request, {});
+  await runScheduled(t);
+  holdNextPoll = true;
+  const received = once(server, "request");
+  await t.mutation(api.jobs.refresh, { id });
+  vi.advanceTimersByTime(1);
+  await t.finishInProgressScheduledFunctions();
+  vi.advanceTimersByTime(1);
+  const checking = t.finishInProgressScheduledFunctions();
+  await received;
+  try {
+    await t.mutation(api.jobs.refresh, { id });
+    await t.mutation(api.jobs.refresh, { id });
+    vi.advanceTimersByTime(MIN_REFRESH_INTERVAL_MS);
+    expect(polls).toBe(1);
+    releasePoll!();
+    await checking;
+    expect(await t.query(app.components.trigger.runs.get, { id })).toMatchObject({
+      status: "EXECUTING",
+      syncStatus: "active",
+      metadata: { progress: 50 },
+    });
+    await runScheduled(t);
+    expect(polls).toBe(2);
+    expect(await t.query(app.components.trigger.runs.get, { id })).toMatchObject({
+      status: "COMPLETED",
+      syncStatus: "complete",
+    });
+    expect(await t.run((ctx) => ctx.db.query("completions").collect())).toHaveLength(1);
+  } finally {
+    releasePoll?.();
+    await checking;
+  }
+});
+
+// Catches: a failed status request leaving refresh permanently blocked as in-flight.
+test("allows a refresh after a retryable status error", async () => {
+  pollFailures = 1;
+  const t = createTest();
+  const id = await t.mutation(api.jobs.request, {});
+  await runScheduled(t);
+  await t.mutation(api.jobs.refresh, { id });
+  await runScheduled(t);
+  expect(polls).toBe(1);
+  expect(await t.query(app.components.trigger.runs.get, { id })).toMatchObject({
+    syncStatus: "active",
+  });
+  await t.mutation(api.jobs.refresh, { id });
+  await runScheduled(t, MIN_REFRESH_INTERVAL_MS);
+  expect(polls).toBe(2);
+  expect(await t.query(app.components.trigger.runs.get, { id })).toMatchObject({
+    syncStatus: "complete",
+  });
+});
+
+// Catches: browser hints resubmitting pending work or bypassing a permanent synchronization failure.
+test("does not submit pending work early or resume paused synchronization", async () => {
+  pollFailures = 1;
+  pollFailureStatus = 401;
+  const t = createTest();
+  const id = await t.mutation(api.jobs.request, {});
+  await t.mutation(api.jobs.refresh, { id });
+  expect(submissions).toBe(0);
+  await runScheduled(t);
+  await t.mutation(api.jobs.refresh, { id });
+  await runScheduled(t);
+  expect(await t.query(app.components.trigger.runs.get, { id })).toMatchObject({
+    syncStatus: "paused",
+  });
+  await t.mutation(api.jobs.refresh, { id });
+  await runScheduled(t, MIN_REFRESH_INTERVAL_MS);
+  expect(polls).toBe(1);
+  expect(submissions).toBe(1);
+  await t.mutation(app.components.trigger.runs.resume, {
+    id,
+    config: { secretKey: "test-only-key", baseUrl },
+  });
+  await drain(t);
+  expect(await t.query(app.components.trigger.runs.get, { id })).toMatchObject({
+    syncStatus: "complete",
+  });
+});
+
 test("rejects large inputs before scheduling remote work", async () => {
   const t = createTest();
   await expect(

packages/convex/src/component/runs.ts

@@ -9,6 +9,7 @@ import { internalMutation, internalQuery, mutation, query, type DataModel } from
 import {
   MAX_INLINE_BYTES,
   MAX_PAGE_BYTES,
+  MIN_REFRESH_INTERVAL_MS,
   SUBMISSION_WINDOW_MS,
   vConfig,
   vRun,
@@ -19,7 +20,18 @@ import {
 type RunDoc = DataModel["runs"]["document"];
 
 export function toRun(run: RunDoc): Run {
-  const { _id, _creationTime, generation, checks, failures, callback, ...fields } = run;
+  const {
+    _id,
+    _creationTime,
+    generation,
+    checks,
+    failures,
+    callback,
+    checking,
+    refreshRequested,
+    lastCheckAt,
+    ...fields
+  } = run;
   return { id: _id, createdAt: _creationTime, ...fields };
 }
 
@@ -110,6 +122,36 @@ export const list = query({
   },
 });
 
+export async function scheduleRefresh(
+  ctx: GenericMutationCtx<DataModel>,
+  run: RunDoc
+): Promise<void> {
+  if (!run.refreshRequested) return;
+  const delay =
+    run.lastCheckAt === undefined
+      ? 0
+      : Math.max(0, run.lastCheckAt + MIN_REFRESH_INTERVAL_MS - Date.now());
+  await ctx.scheduler.runAfter(delay, internal.worker.tick, {
+    id: run._id,
+    generation: run.generation,
+  });
+}
+
+export const refresh = mutation({
+  args: { id: v.string() },
+  returns: v.null(),
+  handler: async (ctx, { id }): Promise<null> => {
+    const runId = ctx.db.normalizeId("runs", id);
+    const run = runId ? await ctx.db.get("runs", runId) : null;
+    if (!run) throw new Error("Run not found");
+    if (!run.runId || run.syncStatus !== "active" || run.refreshRequested) return null;
+    await ctx.db.patch(run._id, { refreshRequested: true });
+    // Preserve the in-flight observation; its result will schedule any requested follow-up.
+    if (!run.checking) await scheduleRefresh(ctx, { ...run, refreshRequested: true });
+    return null;
+  },
+});
+
 export const resume = mutation({
   args: { id: v.string(), config: vConfig },
   returns: v.null(),
@@ -136,6 +178,8 @@ export const resume = mutation({
       failures: 0,
       checks: 0,
       generation,
+      checking: undefined,
+      refreshRequested: undefined,
       updatedAt: Date.now(),
     });
     await ctx.scheduler.runAfter(0, internal.worker.tick, { id: run._id, generation });
@@ -168,6 +212,7 @@ export const recordSubmission = internalMutation({
       syncError: undefined,
       failures: 0,
       checks: 0,
+      checking: undefined,
       updatedAt: Date.now(),
     });
     return null;
@@ -195,6 +240,8 @@ export const recordStatus = internalMutation({
       syncStatus: completed ? ("complete" as const) : ("active" as const),
       syncError: undefined,
       failures: 0,
+      checking: undefined,
+      refreshRequested: completed ? undefined : run.refreshRequested,
       updatedAt: Date.now(),
     };
     await ctx.db.patch("runs", id, updates);
@@ -209,6 +256,8 @@ export const recordStatus = internalMutation({
         .withIndex("by_run", (q) => q.eq("run", id))
         .unique();
       if (request) await ctx.db.delete("requests", request._id);
+    } else {
+      await scheduleRefresh(ctx, run);
     }
     return null;
   },
@@ -222,6 +271,8 @@ export async function pause(
   await ctx.db.patch("runs", run._id, {
     syncStatus: "paused",
     syncError: message,
+    checking: undefined,
+    refreshRequested: undefined,
     updatedAt: Date.now(),
   });
 }