From e8b705b63f93dc348e0d8a3e425b817deca338c0 Mon Sep 17 00:00:00 2001 From: Theodore Li Date: Tue, 7 Jul 2026 15:05:35 -0700 Subject: [PATCH 1/8] fix(tables): server-authoritative run badge, tail SSE from latest, harden Stop-all --- .../table/[tableId]/dispatches/route.test.ts | 125 +++++++++++ .../api/table/[tableId]/dispatches/route.ts | 12 +- .../table/[tableId]/events/stream/route.ts | 14 +- .../run-status-control/run-status-control.tsx | 17 +- .../table-grid/cells/cell-render.tsx | 78 ++++--- .../components/table-grid/table-grid.tsx | 37 +++- .../[tableId]/hooks/use-table-event-stream.ts | 203 ++++++++++-------- .../[workspaceId]/tables/[tableId]/table.tsx | 3 + apps/sim/hooks/queries/tables.ts | 107 ++++----- apps/sim/lib/api/contracts/tables.test.ts | 24 +++ apps/sim/lib/api/contracts/tables.ts | 21 +- apps/sim/lib/table/dispatcher.ts | 111 ++++------ apps/sim/lib/table/events.test.ts | 78 +++++++ apps/sim/lib/table/events.ts | 31 +++ apps/sim/lib/table/workflow-columns.ts | 171 +++++++++------ 15 files changed, 708 insertions(+), 324 deletions(-) create mode 100644 apps/sim/app/api/table/[tableId]/dispatches/route.test.ts create mode 100644 apps/sim/lib/api/contracts/tables.test.ts create mode 100644 apps/sim/lib/table/events.test.ts diff --git a/apps/sim/app/api/table/[tableId]/dispatches/route.test.ts b/apps/sim/app/api/table/[tableId]/dispatches/route.test.ts new file mode 100644 index 00000000000..994b77749db --- /dev/null +++ b/apps/sim/app/api/table/[tableId]/dispatches/route.test.ts @@ -0,0 +1,125 @@ +/** + * @vitest-environment node + */ +import { hybridAuthMockFns } from '@sim/testing' +import { NextRequest } from 'next/server' +import { beforeEach, describe, expect, it, vi } from 'vitest' +import type { TableDefinition } from '@/lib/table' + +const { mockCheckAccess, mockListActiveDispatches, mockCountRunningCells } = vi.hoisted(() => ({ + mockCheckAccess: vi.fn(), + mockListActiveDispatches: vi.fn(), + mockCountRunningCells: vi.fn(), +})) + +vi.mock('@/lib/table/dispatcher', () => ({ + listActiveDispatches: mockListActiveDispatches, + countRunningCells: mockCountRunningCells, +})) +vi.mock('@/app/api/table/utils', async () => { + const { NextResponse } = await import('next/server') + return { + checkAccess: mockCheckAccess, + accessError: (result: { status: number }) => + NextResponse.json({ error: 'denied' }, { status: result.status }), + } +}) + +import { GET } from '@/app/api/table/[tableId]/dispatches/route' + +function buildTable(overrides: Partial = {}): TableDefinition { + return { + id: 'tbl_1', + name: 'People', + description: null, + schema: { columns: [] }, + metadata: null, + rowCount: 0, + maxRows: 1_000_000, + workspaceId: 'workspace-1', + createdBy: 'user-1', + archivedAt: null, + createdAt: new Date(), + updatedAt: new Date(), + ...overrides, + } +} + +function makeRequest(tableId = 'tbl_1') { + const req = new NextRequest(`http://localhost:3000/api/table/${tableId}/dispatches`) + return GET(req, { params: Promise.resolve({ tableId }) }) +} + +function buildDispatchRow(overrides: Record = {}) { + return { + id: 'dispatch-1', + tableId: 'tbl_1', + workspaceId: 'workspace-1', + requestId: 'req-1', + mode: 'all', + scope: { groupIds: ['group-1'] }, + status: 'dispatching', + cursor: 4, + limit: null, + isManualRun: true, + processedCount: 5, + ...overrides, + } +} + +describe('GET /api/table/[tableId]/dispatches', () => { + beforeEach(() => { + vi.clearAllMocks() + hybridAuthMockFns.mockCheckSessionOrInternalAuth.mockResolvedValue({ + success: true, + userId: 'user-1', + authType: 'session', + }) + mockCheckAccess.mockResolvedValue({ ok: true, table: buildTable() }) + mockListActiveDispatches.mockResolvedValue([]) + mockCountRunningCells.mockResolvedValue({}) + }) + + it('returns dispatches and the per-row running map, without a total field', async () => { + mockListActiveDispatches.mockResolvedValue([buildDispatchRow()]) + mockCountRunningCells.mockResolvedValue({ 'row-1': 2, 'row-2': 1 }) + + const response = await makeRequest() + const data = await response.json() + + expect(response.status).toBe(200) + expect(data.data.dispatches).toEqual([ + { + id: 'dispatch-1', + status: 'dispatching', + mode: 'all', + isManualRun: true, + cursor: 4, + scope: { groupIds: ['group-1'] }, + }, + ]) + expect(data.data.runningByRowId).toEqual({ 'row-1': 2, 'row-2': 1 }) + expect(data.data).not.toHaveProperty('runningCellCount') + }) + + it('includes unclaimed pre-stamps only while a dispatch is active', async () => { + mockListActiveDispatches.mockResolvedValue([buildDispatchRow()]) + await makeRequest() + expect(mockCountRunningCells).toHaveBeenCalledWith('tbl_1', { + includeUnclaimedPreStamps: true, + }) + + mockListActiveDispatches.mockResolvedValue([]) + await makeRequest() + expect(mockCountRunningCells).toHaveBeenLastCalledWith('tbl_1', { + includeUnclaimedPreStamps: false, + }) + }) + + it('returns 401 when unauthenticated', async () => { + hybridAuthMockFns.mockCheckSessionOrInternalAuth.mockResolvedValue({ success: false }) + const response = await makeRequest() + expect(response.status).toBe(401) + expect(mockListActiveDispatches).not.toHaveBeenCalled() + }) +}) diff --git a/apps/sim/app/api/table/[tableId]/dispatches/route.ts b/apps/sim/app/api/table/[tableId]/dispatches/route.ts index 25b6f871649..dfc6b2a3bab 100644 --- a/apps/sim/app/api/table/[tableId]/dispatches/route.ts +++ b/apps/sim/app/api/table/[tableId]/dispatches/route.ts @@ -5,7 +5,7 @@ import { parseRequest } from '@/lib/api/server' import { checkSessionOrInternalAuth } from '@/lib/auth/hybrid' import { generateRequestId } from '@/lib/core/utils/request' import { withRouteHandler } from '@/lib/core/utils/with-route-handler' -import { countActiveRunCells, listActiveDispatches } from '@/lib/table/dispatcher' +import { countRunningCells, listActiveDispatches } from '@/lib/table/dispatcher' import { accessError, checkAccess } from '@/app/api/table/utils' const logger = createLogger('TableDispatchesAPI') @@ -38,7 +38,12 @@ export const GET = withRouteHandler(async (request: NextRequest, { params }: Rou if (!result.ok) return accessError(result, requestId, tableId) const rows = await listActiveDispatches(tableId) - const running = await countActiveRunCells(tableId, rows) + // Unclaimed `pending` pre-stamps are real queued work while a dispatch is + // active; with none active they're abandoned orphans that would pin the + // "X running" badge above zero forever. + const runningByRowId = await countRunningCells(tableId, { + includeUnclaimedPreStamps: rows.length > 0, + }) const dispatches: ActiveDispatch[] = rows.map((r) => ({ id: r.id, status: r.status as 'pending' | 'dispatching', @@ -53,8 +58,7 @@ export const GET = withRouteHandler(async (request: NextRequest, { params }: Rou success: true, data: { dispatches, - runningCellCount: running.total, - runningByRowId: running.byRowId, + runningByRowId, }, }) } catch (error) { diff --git a/apps/sim/app/api/table/[tableId]/events/stream/route.ts b/apps/sim/app/api/table/[tableId]/events/stream/route.ts index d938c80c3ec..d38000ed5b3 100644 --- a/apps/sim/app/api/table/[tableId]/events/stream/route.ts +++ b/apps/sim/app/api/table/[tableId]/events/stream/route.ts @@ -8,7 +8,11 @@ import { checkSessionOrInternalAuth } from '@/lib/auth/hybrid' import { generateRequestId } from '@/lib/core/utils/request' import { SSE_HEADERS } from '@/lib/core/utils/sse' import { withRouteHandler } from '@/lib/core/utils/with-route-handler' -import { readTableEventsSince, type TableEventEntry } from '@/lib/table/events' +import { + getLatestTableEventId, + readTableEventsSince, + type TableEventEntry, +} from '@/lib/table/events' import { accessError, checkAccess } from '@/app/api/table/utils' const logger = createLogger('TableEventStreamAPI') @@ -26,10 +30,12 @@ interface RouteContext { /** GET /api/table/[tableId]/events/stream?from= * - * SSE stream of cell-state transitions. Replay-on-reconnect via `from`. + * SSE stream of cell-state transitions. Replay-on-reconnect via `from`; + * absent `from` tails from the latest event id (fresh mount — the client has + * just fetched current state, so replaying history would rewind it). * Pruning (buffer cap exceeded or TTL expired) sends a `pruned` event and * closes; the client responds with a full row-query refetch and reconnects - * from the new earliest. */ + * tailing from latest. */ export const GET = withRouteHandler(async (req: NextRequest, context: RouteContext) => { const requestId = generateRequestId() const parsed = await parseRequest(tableEventStreamContract, req, context) @@ -52,7 +58,7 @@ export const GET = withRouteHandler(async (req: NextRequest, context: RouteConte const stream = new ReadableStream({ async start(controller) { - let lastEventId = fromEventId + let lastEventId = fromEventId ?? (await getLatestTableEventId(tableId)) const deadline = Date.now() + MAX_STREAM_DURATION_MS let nextHeartbeatAt = Date.now() + HEARTBEAT_INTERVAL_MS diff --git a/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/run-status-control/run-status-control.tsx b/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/run-status-control/run-status-control.tsx index 3ef2db128c7..9a834d80e57 100644 --- a/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/run-status-control/run-status-control.tsx +++ b/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/run-status-control/run-status-control.tsx @@ -6,6 +6,10 @@ import { Loader, Square } from '@sim/emcn/icons' interface RunStatusControlProps { running: number + /** No cell has been claimed by a worker yet — everything counted is queued + * or pending, so labeling it "running" would be dishonest. Renders + * "Queueing" without a count instead. */ + queueing: boolean onStopAll: () => void isStopping: boolean } @@ -17,6 +21,7 @@ interface RunStatusControlProps { */ export const RunStatusControl = memo(function RunStatusControl({ running, + queueing, onStopAll, isStopping, }: RunStatusControlProps) { @@ -24,8 +29,14 @@ export const RunStatusControl = memo(function RunStatusControl({
- {running} - running + {queueing ? ( + Queueing + ) : ( + <> + {running} + running + + )}
) diff --git a/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/table-grid/cells/cell-render.tsx b/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/table-grid/cells/cell-render.tsx index b5bdccd21db..6772f467558 100644 --- a/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/table-grid/cells/cell-render.tsx +++ b/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/table-grid/cells/cell-render.tsx @@ -15,7 +15,7 @@ export type CellRenderKind = | { kind: 'value'; text: string } | { kind: 'block-error' } | { kind: 'running' } - | { kind: 'pending-upstream' } + | { kind: 'pending-upstream'; paused: boolean } | { kind: 'queued' } | { kind: 'cancelled' } | { kind: 'error' } @@ -93,9 +93,9 @@ export function resolveCellRender({ exec?.status === 'pending' && typeof exec.jobId === 'string' && exec.jobId.startsWith('paused-') - if (isPaused) return { kind: 'pending-upstream' } + if (isPaused) return { kind: 'pending-upstream', paused: true } if (exec?.status === 'queued' || exec?.status === 'pending') return { kind: 'queued' } - return { kind: 'pending-upstream' } + return { kind: 'pending-upstream', paused: false } } // Waiting wins over a stale terminal status — show the actionable state. @@ -259,48 +259,55 @@ export function CellRender({ kind, isEditing }: CellRenderProps): React.ReactEle case 'running': return ( - + + + ) case 'pending-upstream': return ( - + + + ) case 'cancelled': return ( - + + + ) case 'queued': return ( - - Queued - + + + Queued + + ) case 'waiting': return ( - - - - - Waiting - - - - - Waiting on {kind.labels.map((l) => `"${l}"`).join(', ')} - - + `"${l}"`).join(', ')}`}> + + Waiting + + ) @@ -391,18 +398,22 @@ export function CellRender({ kind, isEditing }: CellRenderProps): React.ReactEle case 'not-found': return ( - - Not found - + + + Not found + + ) case 'no-output': return ( - - No output - + + + No output + + ) @@ -421,6 +432,19 @@ function Wrap({ isEditing, children }: { isEditing: boolean; children: React.Rea return
{children}
} +/** Hover explanation for a status badge. The `` trigger is required — + * Badge/StatusBadge don't forward refs, so the tooltip anchors to a wrapper. */ +function BadgeTooltip({ tip, children }: { tip: string; children: React.ReactNode }) { + return ( + + + {children} + + {tip} + + ) +} + const TYPEWRITER_MS_PER_CHAR = 15 /** diff --git a/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/table-grid/table-grid.tsx b/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/table-grid/table-grid.tsx index 7521df83ecb..0390fa622a0 100644 --- a/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/table-grid/table-grid.tsx +++ b/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/table-grid/table-grid.tsx @@ -89,6 +89,11 @@ export interface SelectionSnapshot { /** Total running/queued workflow runs across ALL rows. Drives the page-header * RunStatusControl ("N running, Stop all"). */ totalRunning: number + /** Whether any LOADED cell has actually been claimed by a worker + * (`status === 'running'`). False while the in-flight set is all + * queued/pending stamps — the header control then reads "Queueing" + * instead of labeling queued work as running. */ + hasRunningCell: boolean /** Whether any dispatch is active (pending/dispatching). Keeps the RunStatusControl * + Stop-all visible during a run even when the per-row count momentarily reads 0 * (e.g. the first window of an auto-fired/capped dispatch before cells stamp). */ @@ -397,13 +402,25 @@ export function TableGrid({ const { data: tableRunState } = useTableRunState(tableId) const activeDispatches = tableRunState?.dispatches const runningByRowId = tableRunState?.runningByRowId ?? EMPTY_RUNNING_BY_ROW - // Actual in-flight cell count = sum of the live per-row map (kept current by - // applyCell's SSE deltas, and the same source the per-row gutter uses). The - // dispatch-scope `runningCellCount` over-counts already-completed groups on - // rows still inside a dispatch's scope — e.g. a cascade where 3 of 4 columns - // finished would read "4 running" instead of "1". + // In-flight cell count = sum of the server-derived per-row map (refetched on + // a throttle as cell SSE events arrive, stamped optimistically on run-click + // — the same source the per-row gutter uses). const totalRunning = Object.values(runningByRowId).reduce((sum, n) => sum + n, 0) const hasActiveDispatch = (activeDispatches?.length ?? 0) > 0 + // Loaded-rows scan (early-exit) — the SSE `running` claim events keep this + // current. Loaded rows are an honest proxy for the whole table: the + // dispatcher's active window is what's claimable, and a fresh Run-all has + // nothing running anywhere yet. + const hasRunningCell = useMemo(() => { + for (const row of rows) { + const executions = row.executions + if (!executions) continue + for (const exec of Object.values(executions)) { + if (exec?.status === 'running') return true + } + } + return false + }, [rows]) // True "select all" total: the filter-scoped COUNT(*) when a filter is active, else the whole // table. Drives the delete-confirm count and the action-bar cell count. @@ -3279,10 +3296,9 @@ export function TableGrid({ }, [rowSelection, rows]) // `runningByRowId` + `totalRunning` come from `useTableRunState` above — - // backend-bootstrapped via `countRunningCells` and kept live by - // `applyCell`'s SSE-driven delta. Counts only cells whose worker has - // actually claimed the cell (`status === 'running'`), ignoring optimistic - // queued/pending stamps. + // server-derived via `countRunningCells` (queued/running/pending), refetched + // on a throttle as cell SSE events arrive, plus optimistic stamps on + // run-click. // Context-menu wrappers: act on `contextMenuRowIds`, then close the menu. // Mirror the action bar's Play / Refresh split: Play fills empty/failed, @@ -3520,6 +3536,7 @@ export function TableGrid({ sameStats && prev.runningInActionBarSelection === runningInActionBarSelection && prev.totalRunning === totalRunning && + prev.hasRunningCell === hasRunningCell && prev.hasActiveDispatch === hasActiveDispatch && prev.hasWorkflowColumns === hasWorkflowColumns && prev.actionBarRowIds.length === actionBarRowIds.length && @@ -3531,6 +3548,7 @@ export function TableGrid({ actionBarRowIds, runningInActionBarSelection, totalRunning, + hasRunningCell, hasActiveDispatch, hasWorkflowColumns, selectedRunScope, @@ -3543,6 +3561,7 @@ export function TableGrid({ actionBarRowIds, runningInActionBarSelection, totalRunning, + hasRunningCell, hasActiveDispatch, hasWorkflowColumns, selectedRunScope, diff --git a/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/hooks/use-table-event-stream.ts b/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/hooks/use-table-event-stream.ts index 45200624ff8..fe8ffb559a6 100644 --- a/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/hooks/use-table-event-stream.ts +++ b/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/hooks/use-table-event-stream.ts @@ -3,10 +3,10 @@ import { useEffect, useRef } from 'react' import { toast } from '@sim/emcn' import { createLogger } from '@sim/logger' +import { backoffWithJitter } from '@sim/utils/retry' import { useQueryClient } from '@tanstack/react-query' import type { ActiveDispatch } from '@/lib/api/contracts/tables' import type { RowData, RowExecutionMetadata, RowExecutions, TableDefinition } from '@/lib/table' -import { isExecInFlight } from '@/lib/table/deps' import type { TableEvent, TableEventEntry } from '@/lib/table/events' import { consumeInitiatedExport, @@ -22,30 +22,10 @@ interface PrunedEvent { earliestEventId: number | null } -const RECONNECT_BACKOFF_MS = [500, 1_000, 2_000, 5_000, 10_000] -const POINTER_PREFIX = 'table-event-stream-pointer:' -const DISPATCH_INVALIDATE_DEBOUNCE_MS = 250 - -function loadPointer(tableId: string): number { - if (typeof window === 'undefined') return 0 - try { - const raw = window.sessionStorage.getItem(`${POINTER_PREFIX}${tableId}`) - if (!raw) return 0 - const parsed = Number.parseInt(raw, 10) - return Number.isFinite(parsed) && parsed >= 0 ? parsed : 0 - } catch { - return 0 - } -} - -function savePointer(tableId: string, eventId: number): void { - if (typeof window === 'undefined') return - try { - window.sessionStorage.setItem(`${POINTER_PREFIX}${tableId}`, String(eventId)) - } catch { - // sessionStorage can throw under quota / private mode — ignore. - } -} +const RECONNECT_BACKOFF_BASE_MS = 500 +const RECONNECT_BACKOFF_MAX_MS = 10_000 +const RUN_STATE_REFETCH_THROTTLE_MS = 1_000 +const ROWS_INVALIDATE_DEBOUNCE_MS = 250 interface UseTableEventStreamArgs { tableId: string | undefined @@ -60,10 +40,13 @@ interface UseTableEventStreamArgs { * Subscribes to the table's SSE event stream and patches the React Query * cache as cell-state events arrive. * - * Reconnect-resume: on transport error, reconnects with `from=` set to the - * last seen `eventId`; server replays missed events from the Redis-backed - * buffer. If the gap exceeds buffer retention (server emits `pruned`), the - * hook full-refetches the row queries and resumes from the new earliest. + * Fresh mount tails from the latest event — the rows + run-state queries + * fetch current state from the DB, so replaying buffered history would only + * rewind fresh cells through stale intermediate states (queued → running → + * completed churn). Reconnect-resume: on transport error, reconnects with + * `from=` set to the last seen `eventId`; server replays missed events from + * the Redis-backed buffer. If the gap exceeds buffer retention (server emits + * `pruned`), the hook full-refetches and resumes tailing from latest. */ export function useTableEventStream({ tableId, @@ -83,20 +66,71 @@ export function useTableEventStream({ let cancelled = false let eventSource: EventSource | null = null let reconnectTimer: ReturnType | null = null - // Resume from the last seen eventId persisted in sessionStorage. Survives - // tab refresh; if the buffer has rolled past this id the server replies - // `pruned` and we full-refetch + restart from the new earliest. - let lastEventId = loadPointer(tableId) + // `null` = no cursor yet: connect without `from` and tail from latest. + // Advanced in memory per event; within-session reconnects resume from it. + let lastEventId: number | null = null let reconnectAttempt = 0 - // Trailing-edge debounce coalesces window-completion bursts. - let dispatchInvalidateTimer: ReturnType | null = null + // Leading + trailing throttle for run-state refetches. Cell/dispatch SSE + // events arrive in bursts (the server flushes its buffer every 500ms): the + // leading edge keeps the badge stepping promptly on sporadic completions; + // the trailing timer coalesces a burst into one refetch per interval. A + // debounce would starve here — sustained bursts reset it indefinitely. + let runStateInvalidateTimer: ReturnType | null = null + let lastRunStateInvalidateAt = 0 + let runStateFetchInFlight = false + let runStateDirtyDuringFetch = false + const invalidateRunState = async (): Promise => { + lastRunStateInvalidateAt = Date.now() + // cancelRefetch: false — the default (true) cancels an in-flight refetch + // and restarts it. When the run-state fetch is slower than the throttle + // interval (a busy run congests the server), that livelocks: every + // interval kills the previous fetch before it can land and the badge + // freezes on the last value that ever resolved. Instead, let an + // in-flight fetch complete (slightly stale counts land), remember that + // events arrived meanwhile, and run one follow-up afterwards — without + // the follow-up, a run's final events deduping into a stale fetch would + // freeze the badge non-zero forever. + if (runStateFetchInFlight) { + runStateDirtyDuringFetch = true + return + } + runStateFetchInFlight = true + try { + await queryClient.invalidateQueries( + { queryKey: tableKeys.activeDispatches(tableId) }, + { cancelRefetch: false } + ) + } finally { + runStateFetchInFlight = false + if (runStateDirtyDuringFetch) { + runStateDirtyDuringFetch = false + scheduleDispatchInvalidate() + } + } + } const scheduleDispatchInvalidate = (): void => { - if (dispatchInvalidateTimer !== null) clearTimeout(dispatchInvalidateTimer) - dispatchInvalidateTimer = setTimeout(() => { - dispatchInvalidateTimer = null - void queryClient.invalidateQueries({ queryKey: tableKeys.activeDispatches(tableId) }) - }, DISPATCH_INVALIDATE_DEBOUNCE_MS) + if (cancelled || runStateInvalidateTimer !== null) return + const elapsed = Date.now() - lastRunStateInvalidateAt + if (elapsed >= RUN_STATE_REFETCH_THROTTLE_MS) { + void invalidateRunState() + return + } + runStateInvalidateTimer = setTimeout(() => { + runStateInvalidateTimer = null + void invalidateRunState() + }, RUN_STATE_REFETCH_THROTTLE_MS - elapsed) + } + /** Urgent resync (usage-limit halt, prune recovery) — skips the throttle. + * Default cancelRefetch here: a fetch started before the halt is stale by + * definition, so kill it and read fresh. One-shot, so no churn risk. */ + const invalidateDispatchesNow = (): void => { + if (runStateInvalidateTimer !== null) { + clearTimeout(runStateInvalidateTimer) + runStateInvalidateTimer = null + } + lastRunStateInvalidateAt = Date.now() + void queryClient.invalidateQueries({ queryKey: tableKeys.activeDispatches(tableId) }) } // Live-fill: import progress ticks arrive every N rows; coalesce the row @@ -107,25 +141,7 @@ export function useTableEventStream({ jobInvalidateTimer = setTimeout(() => { jobInvalidateTimer = null void queryClient.invalidateQueries({ queryKey: tableKeys.rowsRoot(tableId) }) - }, DISPATCH_INVALIDATE_DEBOUNCE_MS) - } - - // Keeps the per-row gutter (`runningByRowId`) live between dispatch events. - // `runningCellCount` (the "X running" badge) is NOT touched here — it's the - // server's dispatch-scope count, seeded optimistically on click and - // re-synced by `applyDispatch` on every window, so live matches reload. - const updateRunningByRow = (rowId: string, wasInFlight: boolean, isInFlight: boolean): void => { - if (wasInFlight === isInFlight) return - const delta = isInFlight ? 1 : -1 - queryClient.setQueryData(tableKeys.activeDispatches(tableId), (prev) => { - if (!prev) return prev - const prevForRow = prev.runningByRowId[rowId] ?? 0 - const nextForRow = Math.max(0, prevForRow + delta) - const nextByRow = { ...prev.runningByRowId } - if (nextForRow === 0) delete nextByRow[rowId] - else nextByRow[rowId] = nextForRow - return { ...prev, runningByRowId: nextByRow } - }) + }, ROWS_INVALIDATE_DEBOUNCE_MS) } const applyCell = (event: Extract): void => { @@ -140,18 +156,12 @@ export function useTableEventStream({ runningBlockIds, blockErrors, } = event - let wasInFlight: boolean | null = null void snapshotAndMutateRows( queryClient, tableId, (row) => { if (row.id !== rowId) return null const prevExec = row.executions?.[groupId] - // In-flight = queued | running | pending. Server's countRunningCells - // counts all three (the gutter Run/Stop button reads this map and - // needs Stop visible during queued too, else clicking Play would - // re-enqueue a cell that's already queued). - if (wasInFlight === null) wasInFlight = isExecInFlight(prevExec) const nextExec: RowExecutionMetadata = { status, executionId: executionId ?? null, @@ -171,13 +181,11 @@ export function useTableEventStream({ }, { cancelInFlight: false } ) - if (wasInFlight === null) { - // Row outside the loaded page slice — can't compute the delta locally. - // Refetch the run-state snapshot from the server. Cheap and rare. - scheduleDispatchInvalidate() - } else { - updateRunningByRow(rowId, wasInFlight, isExecInFlight({ status } as RowExecutionMetadata)) - } + // `runningByRowId` (the "X running" badge + per-row gutter) is + // server-derived: refetch the snapshot on the throttle instead of + // maintaining client-side ±1 deltas, which drift on unloaded rows, + // replays, and races with optimistic stamps. + scheduleDispatchInvalidate() } const applyDispatch = (event: Extract): void => { @@ -185,10 +193,9 @@ export function useTableEventStream({ queryClient.setQueryData(tableKeys.activeDispatches(tableId), (prev) => { // SSE may arrive before the initial fetch lands. Seed an empty // run-state so the dispatch isn't dropped; counters are reconciled - // by the subsequent fetch / per-cell SSE events. + // by the subsequent fetch. const base: TableRunState = prev ?? { dispatches: [], - runningCellCount: 0, runningByRowId: {}, } const list = base.dispatches @@ -225,9 +232,9 @@ export function useTableEventStream({ return { ...base, dispatches: merged } }) // The dispatcher emits this once per window (after the window's cells - // finish + the cursor advances) and on completion. Re-sync the - // dispatch-scope `runningCellCount` from the server so the badge steps - // down per window and matches a reload exactly. + // finish + the cursor advances) and on completion. Re-sync + // `runningByRowId` from the server so the badge steps down per window + // and matches a reload exactly. scheduleDispatchInvalidate() } @@ -304,9 +311,11 @@ export function useTableEventStream({ } // Blocked cells are left `queued` in the DB with no terminal cell event, // so `runningByRowId` would otherwise stay non-zero (stale "X running"). - // Re-sync the server counts, and refetch rows so cells whose pre-stamps - // the server cleared drop their "Queued" state. - scheduleDispatchInvalidate() + // Re-sync the server counts immediately (the user is being told they're + // over limit — the badge must not linger behind the throttle), and + // refetch rows so cells whose pre-stamps the server cleared drop their + // "Queued" state. + invalidateDispatchesNow() void queryClient.invalidateQueries({ queryKey: tableKeys.rowsRoot(tableId) }) onUsageLimitReachedRef.current?.({ dispatchId: event.dispatchId, message: event.message }) } @@ -314,9 +323,10 @@ export function useTableEventStream({ const handlePrune = (payload: PrunedEvent): void => { logger.info('Table event buffer pruned — full refetch', { tableId, ...payload }) void queryClient.invalidateQueries({ queryKey: tableKeys.rowsRoot(tableId) }) - scheduleDispatchInvalidate() - lastEventId = typeof payload.earliestEventId === 'number' ? payload.earliestEventId : 0 - savePointer(tableId, lastEventId) + invalidateDispatchesNow() + // Tail from latest after the refetch — replaying the surviving buffer + // over freshly-refetched rows would rewind them through stale states. + lastEventId = null // Close proactively so the server's close doesn't fire onerror and route // through the backoff path. Reconnect immediately from the new cursor. eventSource?.close() @@ -327,9 +337,11 @@ export function useTableEventStream({ const scheduleReconnect = (): void => { if (cancelled) return - const idx = Math.min(reconnectAttempt, RECONNECT_BACKOFF_MS.length - 1) - const delay = RECONNECT_BACKOFF_MS[idx] reconnectAttempt++ + const delay = backoffWithJitter(reconnectAttempt, null, { + baseMs: RECONNECT_BACKOFF_BASE_MS, + maxMs: RECONNECT_BACKOFF_MAX_MS, + }) reconnectTimer = setTimeout(() => { reconnectTimer = null connect() @@ -338,7 +350,11 @@ export function useTableEventStream({ const connect = (): void => { if (cancelled) return - const url = `/api/table/${tableId}/events/stream?from=${lastEventId}` + // No cursor → tail from latest (server-side); otherwise replay-resume. + const url = + lastEventId === null + ? `/api/table/${tableId}/events/stream` + : `/api/table/${tableId}/events/stream?from=${lastEventId}` try { eventSource = new EventSource(url) } catch (err) { @@ -354,9 +370,8 @@ export function useTableEventStream({ eventSource.onmessage = (msg: MessageEvent) => { try { const entry = JSON.parse(msg.data) as TableEventEntry - if (entry.eventId <= lastEventId) return + if (lastEventId !== null && entry.eventId <= lastEventId) return lastEventId = entry.eventId - savePointer(tableId, lastEventId) if (entry.event?.kind === 'cell') applyCell(entry.event) else if (entry.event?.kind === 'dispatch') applyDispatch(entry.event) else if (entry.event?.kind === 'job') applyJob(entry.event) @@ -388,12 +403,22 @@ export function useTableEventStream({ } } + // In-SPA remount over a warm cache (table A → B → back to A within + // staleTime): the tail starts at "latest", so transitions that fired while + // unmounted were neither refetched (cache still fresh) nor replayed. + // Reconcile once. Cold mounts have no cached run-state → skip, the + // queries are already fetching. + if (queryClient.getQueryState(tableKeys.activeDispatches(tableId))?.data !== undefined) { + void queryClient.invalidateQueries({ queryKey: tableKeys.rowsRoot(tableId) }) + void invalidateRunState() + } + connect() return () => { cancelled = true if (reconnectTimer !== null) clearTimeout(reconnectTimer) - if (dispatchInvalidateTimer !== null) clearTimeout(dispatchInvalidateTimer) + if (runStateInvalidateTimer !== null) clearTimeout(runStateInvalidateTimer) if (jobInvalidateTimer !== null) clearTimeout(jobInvalidateTimer) eventSource?.close() eventSource = null diff --git a/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/table.tsx b/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/table.tsx index 199ec81ed82..0f7485636bb 100644 --- a/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/table.tsx +++ b/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/table.tsx @@ -167,6 +167,7 @@ export function Table({ actionBarRowIds: [], runningInActionBarSelection: 0, totalRunning: 0, + hasRunningCell: false, hasActiveDispatch: false, hasWorkflowColumns: false, selectedRunScope: null, @@ -657,6 +658,7 @@ export function Table({ {selection.totalRunning > 0 || selection.hasActiveDispatch ? ( @@ -686,6 +688,7 @@ export function Table({ embedded && (selection.totalRunning > 0 || selection.hasActiveDispatch) ? ( diff --git a/apps/sim/hooks/queries/tables.ts b/apps/sim/hooks/queries/tables.ts index e9e10f908dd..0c48191d3a7 100644 --- a/apps/sim/hooks/queries/tables.ts +++ b/apps/sim/hooks/queries/tables.ts @@ -271,7 +271,6 @@ export function getTableDetailQueryOptions(workspaceId: string, tableId: string) export interface TableRunState { dispatches: ActiveDispatch[] - runningCellCount: number runningByRowId: Record } @@ -282,7 +281,6 @@ async function fetchTableRunState(tableId: string, signal?: AbortSignal): Promis }) return { dispatches: response.data.dispatches, - runningCellCount: response.data.runningCellCount, runningByRowId: response.data.runningByRowId, } } @@ -335,46 +333,28 @@ function countNewlyInFlight(before: RowExecutions, after: RowExecutions): number return n } -/** The table's maintained, unfiltered `rowCount` from the detail cache (or - * `null` when the detail hasn't loaded). This is the right scope for a Run-all - * estimate: the dispatcher runs every row regardless of the active view - * filter, whereas the rows query's `totalCount` is filter-scoped. */ -function readTableRowCount( - queryClient: ReturnType, - tableId: string -): number | null { - const def = queryClient.getQueryData(tableKeys.detail(tableId)) - return typeof def?.rowCount === 'number' ? def.rowCount : null -} - /** Optimistically reflect a run on the "X running" badge + per-row gutter Stop - * instantly (the optimistic stamp eats the dispatcher's `pending` SSE, so - * `applyCell` never bumps the count, and the server's dispatch-scope count - * isn't live until the first window). `stampedByRow` drives the per-row gutter - * (loaded rows only); `cellCountDelta` is the badge delta — pass the full run - * scope (rows × groups) for Run-all so it matches the server, or omit to use - * the stamped total. Returns the prior snapshot for rollback. */ -function bumpRunState( + * instantly, ahead of the dispatcher's real pending stamps and the next + * server snapshot refetch. Cancels any in-flight run-state fetch first — a + * fetch started before the click would otherwise resolve after this write + * and clobber the bump back to the pre-run snapshot. Returns the prior + * snapshot for rollback. */ +async function bumpRunState( queryClient: ReturnType, tableId: string, - stampedByRow: Record, - cellCountDelta?: number -): { snapshot: TableRunState | undefined } | null { + stampedByRow: Record +): Promise<{ snapshot: TableRunState | undefined } | null> { const stampedTotal = Object.values(stampedByRow).reduce((s, n) => s + n, 0) - const countDelta = cellCountDelta ?? stampedTotal - if (countDelta === 0 && stampedTotal === 0) return null + if (stampedTotal === 0) return null + await queryClient.cancelQueries({ queryKey: tableKeys.activeDispatches(tableId) }) const snapshot = queryClient.getQueryData(tableKeys.activeDispatches(tableId)) queryClient.setQueryData(tableKeys.activeDispatches(tableId), (prev) => { - const base = prev ?? { dispatches: [], runningCellCount: 0, runningByRowId: {} } + const base = prev ?? { dispatches: [], runningByRowId: {} } const nextByRow = { ...base.runningByRowId } for (const [rid, n] of Object.entries(stampedByRow)) { nextByRow[rid] = (nextByRow[rid] ?? 0) + n } - return { - ...base, - runningCellCount: base.runningCellCount + countDelta, - runningByRowId: nextByRow, - } + return { ...base, runningByRowId: nextByRow } }) return { snapshot } } @@ -647,7 +627,7 @@ export function useCreateTableRow({ workspaceId, tableId }: RowMutationContext) }, }) }, - onSuccess: (response) => { + onSuccess: async (response) => { const row = response.data.row if (!row) return @@ -660,7 +640,7 @@ export function useCreateTableRow({ workspaceId, tableId }: RowMutationContext) // the "X running" badge + gutter Stop show immediately (the row had no // prior executions, so the stamped set is the full delta). const stampedCount = countNewlyInFlight({}, stamped.executions ?? {}) - if (stampedCount > 0) bumpRunState(queryClient, tableId, { [row.id]: stampedCount }) + if (stampedCount > 0) await bumpRunState(queryClient, tableId, { [row.id]: stampedCount }) // `reconcileCreatedRow` only patches the default-order view. Filtered / // column-sorted rows queries can't be reconciled from that heuristic @@ -904,7 +884,7 @@ export function useUpdateTableRow({ workspaceId, tableId }: RowMutationContext) } }) - const bumped = bumpRunState(queryClient, tableId, stampedByRow) + const bumped = await bumpRunState(queryClient, tableId, stampedByRow) return { previousQueries, runStateSnapshot: bumped?.snapshot, @@ -995,7 +975,7 @@ export function useBatchUpdateTableRows({ workspaceId, tableId }: RowMutationCon } }) - const bumped = bumpRunState(queryClient, tableId, stampedByRow) + const bumped = await bumpRunState(queryClient, tableId, stampedByRow) return { previousQueries, runStateSnapshot: bumped?.snapshot, @@ -1319,6 +1299,7 @@ export function useCancelTableRuns({ workspaceId, tableId }: RowMutationContext) }) ) : undefined + const touchedRowIds = new Set() const snapshots = await snapshotAndMutateRows( queryClient, tableId, @@ -1350,21 +1331,50 @@ export function useCancelTableRuns({ workspaceId, tableId }: RowMutationContext) } rowTouched = true } - return rowTouched ? { ...r, executions: nextExecutions } : null + if (!rowTouched) return null + touchedRowIds.add(r.id) + return { ...r, executions: nextExecutions } }, { onlyKey } ) - return { snapshots } + + // Zero the badge + per-row gutter for the stopped rows immediately. + // Cancel any in-flight run-state fetch first — one started before the + // server processed the cancel would resolve with stale non-zero counts + // and resurrect the badge until onSettled's refetch lands. + await queryClient.cancelQueries({ queryKey: tableKeys.activeDispatches(tableId) }) + const runStateSnapshot = queryClient.getQueryData( + tableKeys.activeDispatches(tableId) + ) + queryClient.setQueryData(tableKeys.activeDispatches(tableId), (prev) => { + if (!prev) return prev + const nextByRow: Record = {} + for (const [rid, n] of Object.entries(prev.runningByRowId)) { + if (scope === 'all' && !filter) { + // Table-wide stop: everything not explicitly excluded is cancelled, + // including rows outside the loaded page slice. + if (!excludedRowIds?.has(rid)) continue + } else if (touchedRowIds.has(rid)) { + continue + } + nextByRow[rid] = n + } + return { ...prev, runningByRowId: nextByRow } + }) + return { snapshots, runStateSnapshot } }, - onError: (_err, _variables, context) => { + onError: (error, _variables, context) => { if (context?.snapshots) restoreCachedWorkflowCells(queryClient, context.snapshots) + queryClient.setQueryData(tableKeys.activeDispatches(tableId), context?.runStateSnapshot) + // A failed Stop must be loud — the optimistic clear above made the run + // look stopped, and silently reverting reads as "the cancel didn't work". + toast.error(`Failed to stop runs: ${error.message}`, { duration: 5000 }) }, onSettled: () => { queryClient.invalidateQueries({ queryKey: tableKeys.rowsRoot(tableId) }) - // Refetch the run-state snapshot — server re-derives runningCellCount + - // runningByRowId from the freshly-updated sidecar via countRunningCells. - // Without this, the counter and row gutter button stay stale until the - // user refetches manually. + // Refetch the run-state snapshot — server re-derives runningByRowId from + // the freshly-updated sidecar via countRunningCells. Reconciles the + // optimistic clear above with whatever the cancel actually stopped. queryClient.invalidateQueries({ queryKey: tableKeys.activeDispatches(tableId) }) }, }) @@ -2062,14 +2072,7 @@ export function useRunColumn({ workspaceId, tableId }: RowMutationContext) { return { ...r, data: nextData, executions: next } }) - // Badge counts the whole run scope (rows × groups), matching the server's - // dispatch-scope count — not just the loaded rows we could stamp. For - // Run-all that's the table's totalCount; for a scoped run, the rowIds. - const scopeRowCount = targetRowIds - ? targetRowIds.size - : (readTableRowCount(queryClient, tableId) ?? Object.keys(stampedByRow).length) - const cellCountDelta = scopeRowCount * targetGroupIds.size - const bumped = bumpRunState(queryClient, tableId, stampedByRow, cellCountDelta) + const bumped = await bumpRunState(queryClient, tableId, stampedByRow) return { snapshots, runStateSnapshot: bumped?.snapshot, didBumpRunState: bumped !== null } }, onError: (_err, _variables, context) => { @@ -2092,7 +2095,7 @@ export function useRunColumn({ workspaceId, tableId }: RowMutationContext) { return } queryClient.setQueryData(tableKeys.activeDispatches(tableId), (prev) => { - const base = prev ?? { dispatches: [], runningCellCount: 0, runningByRowId: {} } + const base = prev ?? { dispatches: [], runningByRowId: {} } if (base.dispatches.some((d) => d.id === dispatchId)) return base const dispatch: ActiveDispatch = { id: dispatchId, diff --git a/apps/sim/lib/api/contracts/tables.test.ts b/apps/sim/lib/api/contracts/tables.test.ts new file mode 100644 index 00000000000..0f368363d3b --- /dev/null +++ b/apps/sim/lib/api/contracts/tables.test.ts @@ -0,0 +1,24 @@ +/** + * @vitest-environment node + */ +import { describe, expect, it } from 'vitest' +import { tableEventStreamQuerySchema } from '@/lib/api/contracts/tables' + +describe('tableEventStreamQuerySchema', () => { + it('parses an explicit cursor', () => { + expect(tableEventStreamQuerySchema.parse({ from: '7' })).toEqual({ from: 7 }) + }) + + it('keeps 0 as an explicit replay-from-start cursor', () => { + expect(tableEventStreamQuerySchema.parse({ from: '0' })).toEqual({ from: 0 }) + }) + + it('yields undefined when absent — the tail-from-latest signal', () => { + expect(tableEventStreamQuerySchema.parse({})).toEqual({ from: undefined }) + }) + + it('yields undefined for invalid values instead of coercing to a full replay', () => { + expect(tableEventStreamQuerySchema.parse({ from: 'abc' })).toEqual({ from: undefined }) + expect(tableEventStreamQuerySchema.parse({ from: '-4' })).toEqual({ from: undefined }) + }) +}) diff --git a/apps/sim/lib/api/contracts/tables.ts b/apps/sim/lib/api/contracts/tables.ts index b9d0512b0c0..97daf70146a 100644 --- a/apps/sim/lib/api/contracts/tables.ts +++ b/apps/sim/lib/api/contracts/tables.ts @@ -1334,12 +1334,11 @@ export const listActiveDispatchesContract = defineRouteContract({ schema: successResponseSchema( z.object({ dispatches: z.array(activeDispatchSchema), - /** Total cells across the table whose `status === 'running'`. The - * client maintains this incrementally via cell SSE events; this - * field is the bootstrap snapshot on mount. */ - runningCellCount: z.number().int().nonnegative(), - /** Map rowId → number of running cells on that row. Drives the - * per-row badge next to the Stop button. */ + /** Map rowId → number of in-flight (queued/running/pending) cells on + * that row. Sums to the "X running" badge and drives the per-row + * gutter Run/Stop button. Server-authoritative: refetched on a + * throttle as cell SSE events arrive, plus optimistic stamps on + * run-click. */ runningByRowId: z.record(z.string(), z.number().int().positive()), }) ), @@ -1349,11 +1348,15 @@ export const listActiveDispatchesContract = defineRouteContract({ export type ActiveDispatch = z.output export const tableEventStreamQuerySchema = z.object({ + /** Replay cursor: events with `eventId > from` are replayed on connect. + * `0` replays the whole buffer (prune recovery). Absent → the server tails + * from the latest event id — a fresh mount has just fetched current state + * from the DB, so replaying history would only rewind it. */ from: z.preprocess((value) => { - if (typeof value !== 'string') return 0 + if (typeof value !== 'string') return undefined const parsed = Number.parseInt(value, 10) - return Number.isFinite(parsed) && parsed >= 0 ? parsed : 0 - }, z.number().int().min(0)), + return Number.isFinite(parsed) && parsed >= 0 ? parsed : undefined + }, z.number().int().min(0).optional()), }) export const tableEventStreamContract = defineRouteContract({ diff --git a/apps/sim/lib/table/dispatcher.ts b/apps/sim/lib/table/dispatcher.ts index 449bc531914..dc13a1ee3c0 100644 --- a/apps/sim/lib/table/dispatcher.ts +++ b/apps/sim/lib/table/dispatcher.ts @@ -207,15 +207,10 @@ export async function insertDispatch(input: { return id } -/** Read every dispatch on a table whose status is still `pending` or - * `dispatching`. Drives the client-side "about to run" overlay: rows in an - * active dispatch's scope ahead of its cursor are rendered as queued even - * before the dispatcher has reached them, so refresh during a long Run-all - * doesn't lose the queued indicators. */ -/** Counts in-flight cells (queued / running / pending) across the entire - * table — the authoritative source for the "X running" badge and the per-row - * gutter Run/Stop button. All three statuses are user-cancellable, so the - * gutter must surface Stop whenever any of them are present (else clicking +/** Counts in-flight cells (queued / running / pending) per row across the + * entire table — the authoritative source for the "X running" badge and the + * per-row gutter Run/Stop button. All three statuses are user-cancellable, so + * the gutter must surface Stop whenever any of them are present (else clicking * Play during the queued window would re-run an already-queued cell). * * Excludes orphan pre-stamps — `pending` rows with no `executionId` — which @@ -230,7 +225,7 @@ export async function insertDispatch(input: { export async function countRunningCells( tableId: string, opts?: { includeUnclaimedPreStamps?: boolean } -): Promise<{ total: number; byRowId: Record }> { +): Promise> { // `pending` + null-executionId rows are unclaimed pre-stamps. With an active // dispatch they're real queued work (include); with none they're abandoned // orphans that would pin the badge above zero forever (exclude). @@ -251,62 +246,18 @@ export async function countRunningCells( ) ) .groupBy(tableRowExecutions.rowId) - let total = 0 const byRowId: Record = {} for (const r of rows) { - if (r.runningCount > 0) { - byRowId[r.rowId] = r.runningCount - total += r.runningCount - } + if (r.runningCount > 0) byRowId[r.rowId] = r.runningCount } - return { total, byRowId } -} - -/** Authoritative "cells queued or running" count for the table, derived from - * active dispatches so it survives reload and matches the live count. For each - * active dispatch every row in scope ahead of the cursor still has to run each - * targeted group, so remaining work = (rows ahead of cursor) × |groupIds|. - * Exact for Run-all; an upper bound for incomplete/new (rows the eligibility - * filter later skips are still counted). Falls back to the sidecar in-flight - * count when no dispatch is active (orphan stragglers). `byRowId` stays - * sidecar-based — the client overlay renders queued rows ahead of the cursor. */ -export async function countActiveRunCells( - tableId: string, - dispatches?: DispatchRow[] -): Promise<{ total: number; byRowId: Record }> { - const active = dispatches ?? (await listActiveDispatches(tableId)) - if (active.length === 0) return countRunningCells(tableId) - - const countRowsAhead = async (d: DispatchRow): Promise => { - const groupCount = d.scope.groupIds.length - if (groupCount === 0) return 0 - const filters = [eq(userTableRows.tableId, tableId), gt(userTableRows.position, d.cursor)] - if (d.scope.rowIds && d.scope.rowIds.length > 0) { - filters.push(inArray(userTableRows.id, d.scope.rowIds)) - } - const [row] = await db - .select({ rowsAhead: sql`count(*)::int` }) - .from(userTableRows) - .where(and(...filters)) - let rowsAhead = row?.rowsAhead ?? 0 - // A `rows` cap means at most `max - processed` more rows will run, even if - // many more sit ahead of the cursor — clamp so the badge doesn't over-count. - if (d.limit?.type === 'rows') { - rowsAhead = Math.min(rowsAhead, Math.max(0, d.limit.max - d.processedCount)) - } - return rowsAhead * groupCount - } - - // Include pre-stamps so `byRowId` matches the live SSE count (which counts - // `pending`); otherwise the badge flickers 20→0 on each refetch. - const [sidecar, perDispatch] = await Promise.all([ - countRunningCells(tableId, { includeUnclaimedPreStamps: true }), - Promise.all(active.map(countRowsAhead)), - ]) - const total = perDispatch.reduce((sum, n) => sum + n, 0) - return { total, byRowId: sidecar.byRowId } + return byRowId } +/** Read every dispatch on a table whose status is still `pending` or + * `dispatching`. Drives the client-side "about to run" overlay: rows in an + * active dispatch's scope ahead of its cursor are rendered as queued even + * before the dispatcher has reached them, so refresh during a long Run-all + * doesn't lose the queued indicators. */ export async function listActiveDispatches(tableId: string): Promise { const rows = await db .select() @@ -714,6 +665,34 @@ export async function markDispatchComplete(dispatchId: string): Promise { .where(eq(tableRunDispatches.id, dispatchId)) } +/** Cancel one dispatch by id (if still active) and emit the terminal SSE so + * the client overlay clears. Used when `runWorkflowColumn` fails between + * inserting its dispatch row and firing the dispatcher — without this the + * orphaned `pending` row would pin the "about to run" overlay forever. */ +export async function cancelDispatchById(dispatchId: string): Promise { + const [row] = await db + .update(tableRunDispatches) + .set({ status: 'cancelled', cancelledAt: new Date() }) + .where( + and( + eq(tableRunDispatches.id, dispatchId), + inArray(tableRunDispatches.status, [...ACTIVE_DISPATCH_STATUSES]) + ) + ) + .returning() + if (!row) return + await appendTableEvent({ + kind: 'dispatch', + tableId: row.tableId, + dispatchId: row.id, + status: 'cancelled', + scope: row.scope as DispatchScope, + cursor: row.cursor, + mode: row.mode as DispatchMode, + isManualRun: row.isManualRun, + }) +} + /** Complete a dispatch only if it's still active, returning whether THIS call * performed the transition. Lets concurrent cells that all hit a hard stop * (e.g. usage limit) elect a single owner — only the winner emits the @@ -752,12 +731,15 @@ export async function markDispatchCancelled(dispatchId: string): Promise { * is that exact filter (a filtered "select all" Stop must not halt * whole-table or differently-filtered runs). Pass `spareExcludedRowIds` * (select-all-minus-deselections Stop) to spare row-scoped dispatches whose - * rows are ALL deselected — that work wasn't in the stopped selection. */ + * rows are ALL deselected — that work wasn't in the stopped selection. Pass + * `spareDispatchId` when the caller is a manual run cancelling *prior* work: + * its own dispatch row is already inserted (so a concurrent Stop-all has + * something to cancel) and must not cancel itself. */ export async function markActiveDispatchesCancelled( tableId: string, - scopeFilter?: Filter, - spareExcludedRowIds?: string[] + opts?: { scopeFilter?: Filter; spareExcludedRowIds?: string[]; spareDispatchId?: string } ): Promise { + const { scopeFilter, spareExcludedRowIds, spareDispatchId } = opts ?? {} const cancelled = await db .update(tableRunDispatches) .set({ status: 'cancelled', cancelledAt: new Date() }) @@ -765,6 +747,7 @@ export async function markActiveDispatchesCancelled( and( eq(tableRunDispatches.tableId, tableId), inArray(tableRunDispatches.status, [...ACTIVE_DISPATCH_STATUSES]), + spareDispatchId ? ne(tableRunDispatches.id, spareDispatchId) : undefined, scopeFilter ? sql`${tableRunDispatches.scope}->'filter' = ${JSON.stringify(scopeFilter)}::jsonb` : undefined, diff --git a/apps/sim/lib/table/events.test.ts b/apps/sim/lib/table/events.test.ts new file mode 100644 index 00000000000..d3bdb6deca3 --- /dev/null +++ b/apps/sim/lib/table/events.test.ts @@ -0,0 +1,78 @@ +/** + * @vitest-environment node + */ +import { beforeEach, describe, expect, it, vi } from 'vitest' + +vi.mock('@/lib/core/config/redis', () => ({ + getRedisClient: () => null, +})) + +vi.mock('@/lib/core/config/env', () => ({ + env: { REDIS_URL: undefined }, +})) + +import type { TableEvent } from '@/lib/table/events' +import { appendTableEvent, getLatestTableEventId, readTableEventsSince } from '@/lib/table/events' + +/** Module-level memory buffer can't be reset without vi.resetModules — use a + * unique tableId per test to avoid cross-test bleed. */ +let seq = 0 +function uniqueTableId(): string { + seq++ + return `table-events-test-${seq}` +} + +function cellEvent(tableId: string): TableEvent { + return { + kind: 'cell', + tableId, + rowId: 'row-1', + groupId: 'group-1', + status: 'running', + } +} + +describe('getLatestTableEventId (memory buffer)', () => { + beforeEach(() => { + vi.clearAllMocks() + }) + + it('returns 0 for a table with no events, without allocating a stream', async () => { + const tableId = uniqueTableId() + expect(await getLatestTableEventId(tableId)).toBe(0) + // A pure read must not have created a buffer: appending afterwards still + // starts the sequence at 1. + const entry = await appendTableEvent(cellEvent(tableId)) + expect(entry?.eventId).toBe(1) + }) + + it('returns the latest assigned eventId after appends', async () => { + const tableId = uniqueTableId() + await appendTableEvent(cellEvent(tableId)) + const second = await appendTableEvent(cellEvent(tableId)) + expect(second?.eventId).toBe(2) + expect(await getLatestTableEventId(tableId)).toBe(2) + }) + + it('tailing from the latest id yields no replayed events', async () => { + const tableId = uniqueTableId() + await appendTableEvent(cellEvent(tableId)) + await appendTableEvent(cellEvent(tableId)) + const latest = await getLatestTableEventId(tableId) + const result = await readTableEventsSince(tableId, latest) + expect(result).toEqual({ status: 'ok', events: [] }) + }) + + it('a subsequent append is visible to a reader tailing from the prior latest', async () => { + const tableId = uniqueTableId() + await appendTableEvent(cellEvent(tableId)) + const latest = await getLatestTableEventId(tableId) + await appendTableEvent(cellEvent(tableId)) + const result = await readTableEventsSince(tableId, latest) + expect(result.status).toBe('ok') + if (result.status === 'ok') { + expect(result.events).toHaveLength(1) + expect(result.events[0].eventId).toBe(latest + 1) + } + }) +}) diff --git a/apps/sim/lib/table/events.ts b/apps/sim/lib/table/events.ts index 8b8dc6da93c..fee8f313687 100644 --- a/apps/sim/lib/table/events.ts +++ b/apps/sim/lib/table/events.ts @@ -276,6 +276,37 @@ export async function appendTableEvent(event: TableEvent): Promise { + const redis = getRedisClient() + if (!redis) { + if (canUseMemoryBuffer()) { + // Pure read — getMemoryStream() would allocate a stream as a side effect. + const stream = memoryTableStreams.get(tableId) + return stream ? stream.nextEventId - 1 : 0 + } + return 0 + } + try { + const raw = await redis.get(getSeqKey(tableId)) + if (!raw) return 0 + const parsed = Number.parseInt(raw, 10) + return Number.isFinite(parsed) && parsed > 0 ? parsed : 0 + } catch (error) { + logger.warn('getLatestTableEventId failed', { tableId, error: toError(error).message }) + return 0 + } +} + /** * Read events for a table where eventId > afterEventId. Returns 'pruned' if * the caller has fallen off the back of the buffer (TTL expired or cap rolled diff --git a/apps/sim/lib/table/workflow-columns.ts b/apps/sim/lib/table/workflow-columns.ts index c60a04efb48..50bffd4db51 100644 --- a/apps/sim/lib/table/workflow-columns.ts +++ b/apps/sim/lib/table/workflow-columns.ts @@ -365,7 +365,14 @@ export const TABLE_CONCURRENCY_LIMIT = 20 export async function cancelWorkflowGroupRuns( tableId: string, rowId?: string, - options?: { groupIds?: string[]; filter?: Filter; excludeRowIds?: string[] } + options?: { + groupIds?: string[] + filter?: Filter + excludeRowIds?: string[] + /** Set by a manual run cancelling prior work: its own already-inserted + * dispatch must survive the table-wide dispatch cancel. */ + spareDispatchId?: string + } ): Promise { const { getTableById } = await import('@/lib/table/service') const { updateRow } = await import('@/lib/table/rows/service') @@ -387,7 +394,15 @@ export async function cancelWorkflowGroupRuns( // (its own run); whole-table or differently-scoped dispatches keep running — // their cells cancelled below are skipped via `cancelledAt > requestedAt`. if (!rowId) { - await markActiveDispatchesCancelled(tableId, options?.filter, options?.excludeRowIds) + const cancelledDispatches = await markActiveDispatchesCancelled(tableId, { + scopeFilter: options?.filter, + spareExcludedRowIds: options?.excludeRowIds, + spareDispatchId: options?.spareDispatchId, + }) + logger.info( + `cancelWorkflowGroupRuns: cancelled ${cancelledDispatches.length} active dispatch(es) for table ${tableId}`, + { dispatchIds: cancelledDispatches.map((d) => d.id) } + ) } const allGroups = table.schema.workflowGroups ?? [] @@ -678,67 +693,22 @@ export async function runWorkflowColumn(opts: { if (targetGroups.length === 0) return { dispatchId: null } const targetGroupIds = targetGroups.map((g) => g.id) - const { bulkClearWorkflowGroupCells, insertDispatch, runDispatcherToCompletion } = await import( - './dispatcher' - ) - - // For manual runs (Run all rows / Run column / Refresh-row / Refresh-cell), - // cancel any prior active dispatches AND in-flight cells in scope before - // clearing. Without this: - // - Two dispatcher loops would walk overlapping rows and burn duplicate work. - // - mode:'all' bulk-clear deletes in-flight sidecar rows without aborting - // workers — those would keep writing into the wiped state. - // Scope: table-wide cancel when rowIds is empty (also cancels active - // dispatches via markActiveDispatchesCancelled), per-row cancel otherwise - // (no dispatch cancel — other rows' dispatches keep running). Dep-edit - // cascade in `updateRow` already cancels its own scope before calling, - // so the duplicate work here is a cheap no-op for that caller. - // Auto-fire (`mode:'new'`) is harmless overlap-wise — the NOT EXISTS - // filter excludes already-attempted rows. - const cancelPriorRuns = isManualRun && (mode === 'all' || mode === 'incomplete') - if (cancelPriorRuns) { - if (!rowIds || rowIds.length === 0) { - // Filtered runs cancel only their own scope — a table-wide cancel here - // would stop unrelated work on rows outside the filter (or on deselected rows). - await cancelWorkflowGroupRuns(tableId, undefined, { - groupIds: targetGroupIds, - filter, - excludeRowIds, - }) - } else { - // Per-row cancel — sequential so we don't fan out N parallel - // markActiveDispatchesCancelled calls (it's a no-op when rowId is set, - // but each call still touches the DB). - for (const rowId of rowIds) { - await cancelWorkflowGroupRuns(tableId, rowId, { groupIds: targetGroupIds }) - } - } - } - - // Wipe targeted output cols + executions[gid] before any cells fire so the - // user sees the column flip to empty/Pending instantly. Skipped for capped - // runs: the eager clear can't know which N rows the dispatcher will pick - // (they depend on per-row eligibility as it walks positions), so wiping all - // rows in scope would blank far more than we re-run. `mode: 'all'` re-runs - // completed cells without the clear anyway — the clear is only for instant - // feedback, which the capped rows still get via the dispatcher's pre-stamp. - // Skip the eager clear for a filtered run: `bulkClearWorkflowGroupCells` keys by `rowIds`, and a - // filtered scope has none — clearing table-wide would blank rows that don't match the filter. The - // dispatcher's per-row pre-stamp still provides instant Pending feedback as it walks. - if (!limit && !filter) { - await bulkClearWorkflowGroupCells({ - tableId, - groups: targetGroups.map((g) => ({ id: g.id, outputs: g.outputs })), - rowIds, - excludeRowIds, - mode, - }) - } - - // Always insert a `table_run_dispatches` row. The dispatcher state machine - // is the single source of truth for cursor advancement, SSE emission, and - // cancel — backend (trigger.dev SaaS vs in-process) only affects how each - // window's cells get executed. + const { + bulkClearWorkflowGroupCells, + cancelDispatchById, + insertDispatch, + runDispatcherToCompletion, + } = await import('./dispatcher') + + // Always insert a `table_run_dispatches` row, and insert it FIRST — before + // the prior-run cancel and the bulk clear below, which can take seconds on + // a large table. The client shows its Stop control optimistically from the + // moment the user clicks Run, so a Stop-all arriving during that prep work + // must find a dispatch row to cancel; inserted-after ordering made an early + // Stop-all a silent no-op and the run proceeded anyway. The dispatcher + // state machine is the single source of truth for cursor advancement, SSE + // emission, and cancel — backend (trigger.dev SaaS vs in-process) only + // affects how each window's cells get executed. const dispatchId = await insertDispatch({ tableId, workspaceId, @@ -757,6 +727,81 @@ export async function runWorkflowColumn(opts: { triggeredByUserId, }) + try { + // For manual runs (Run all rows / Run column / Refresh-row / Refresh-cell), + // cancel any prior active dispatches AND in-flight cells in scope before + // clearing. Without this: + // - Two dispatcher loops would walk overlapping rows and burn duplicate work. + // - mode:'all' bulk-clear deletes in-flight sidecar rows without aborting + // workers — those would keep writing into the wiped state. + // Scope: table-wide cancel when rowIds is empty (also cancels active + // dispatches via markActiveDispatchesCancelled, sparing the one just + // inserted above), per-row cancel otherwise (no dispatch cancel — other + // rows' dispatches keep running). Dep-edit cascade in `updateRow` already + // cancels its own scope before calling, so the duplicate work here is a + // cheap no-op for that caller. Auto-fire (`mode:'new'`) is harmless + // overlap-wise — the NOT EXISTS filter excludes already-attempted rows. + const cancelPriorRuns = isManualRun && (mode === 'all' || mode === 'incomplete') + if (cancelPriorRuns) { + if (!rowIds || rowIds.length === 0) { + // Filtered runs cancel only their own scope — a table-wide cancel here + // would stop unrelated work on rows outside the filter (or on deselected rows). + await cancelWorkflowGroupRuns(tableId, undefined, { + groupIds: targetGroupIds, + filter, + excludeRowIds, + spareDispatchId: dispatchId, + }) + } else { + // Per-row cancel — sequential so we don't fan out N parallel + // markActiveDispatchesCancelled calls (it's a no-op when rowId is set, + // but each call still touches the DB). + for (const rowId of rowIds) { + await cancelWorkflowGroupRuns(tableId, rowId, { groupIds: targetGroupIds }) + } + } + } + + // Wipe targeted output cols + executions[gid] before any cells fire so the + // user sees the column flip to empty/Pending instantly. Skipped for capped + // runs: the eager clear can't know which N rows the dispatcher will pick + // (they depend on per-row eligibility as it walks positions), so wiping all + // rows in scope would blank far more than we re-run. `mode: 'all'` re-runs + // completed cells without the clear anyway — the clear is only for instant + // feedback, which the capped rows still get via the dispatcher's pre-stamp. + // Skip the eager clear for a filtered run: `bulkClearWorkflowGroupCells` keys by `rowIds`, and a + // filtered scope has none — clearing table-wide would blank rows that don't match the filter. The + // dispatcher's per-row pre-stamp still provides instant Pending feedback as it walks. + if (!limit && !filter) { + await bulkClearWorkflowGroupCells({ + tableId, + groups: targetGroups.map((g) => ({ id: g.id, outputs: g.outputs })), + rowIds, + excludeRowIds, + mode, + }) + } + } catch (err) { + // Prep failed after the dispatch row was inserted — cancel it so an + // orphaned `pending` dispatch can't pin the client's "about to run" + // overlay, then fail the request. + await cancelDispatchById(dispatchId) + throw err + } + + // A Stop-all can land during the prep above; its dispatch cancel is the + // authoritative stop. Don't fire the dispatcher loop for a dead dispatch — + // it would exit on its first status read, but the trigger.dev path would + // still spin up a task for nothing. + const { readDispatch } = await import('./dispatcher') + const current = await readDispatch(dispatchId) + if (!current || current.status === 'cancelled' || current.status === 'complete') { + logger.info( + `[Cascade] [${requestId}] dispatch ${dispatchId} cancelled during prep — not firing` + ) + return { dispatchId } + } + logger.info( `[Cascade] [${requestId}] dispatch ${dispatchId} table=${tableId} groups=[${targetGroupIds.join(',')}] rows=${rowIds ? `[${rowIds.join(',')}]` : 'all'} mode=${mode}` ) From ed3b78a74998db1bb9c521fd60aa0518548127b3 Mon Sep 17 00:00:00 2001 From: Theodore Li Date: Tue, 7 Jul 2026 15:12:28 -0700 Subject: [PATCH 2/8] fix(tables): return null dispatchId when Stop-all cancels a run during prep --- apps/sim/lib/table/workflow-columns.ts | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/apps/sim/lib/table/workflow-columns.ts b/apps/sim/lib/table/workflow-columns.ts index 50bffd4db51..ffefda28c07 100644 --- a/apps/sim/lib/table/workflow-columns.ts +++ b/apps/sim/lib/table/workflow-columns.ts @@ -792,14 +792,17 @@ export async function runWorkflowColumn(opts: { // A Stop-all can land during the prep above; its dispatch cancel is the // authoritative stop. Don't fire the dispatcher loop for a dead dispatch — // it would exit on its first status read, but the trigger.dev path would - // still spin up a task for nothing. + // still spin up a task for nothing. Return a null dispatchId: the client + // seeds a returned id into its active-dispatch overlay, which would + // resurrect the Run/Stop UI the cancelled SSE event already cleared; null + // takes its "no dispatch created" path and rolls the optimistic bump back. const { readDispatch } = await import('./dispatcher') const current = await readDispatch(dispatchId) if (!current || current.status === 'cancelled' || current.status === 'complete') { logger.info( `[Cascade] [${requestId}] dispatch ${dispatchId} cancelled during prep — not firing` ) - return { dispatchId } + return { dispatchId: null } } logger.info( From ab40f49075d347b90f60b001dd7891c2af73d730 Mon Sep 17 00:00:00 2001 From: Theodore Li Date: Tue, 7 Jul 2026 15:17:24 -0700 Subject: [PATCH 3/8] improvement(tables): co-locate dispatcher imports, stamp throttle clock only on fetch start --- .../tables/[tableId]/hooks/use-table-event-stream.ts | 4 +++- apps/sim/lib/table/workflow-columns.ts | 2 +- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/hooks/use-table-event-stream.ts b/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/hooks/use-table-event-stream.ts index fe8ffb559a6..f15b22f83d4 100644 --- a/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/hooks/use-table-event-stream.ts +++ b/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/hooks/use-table-event-stream.ts @@ -81,7 +81,6 @@ export function useTableEventStream({ let runStateFetchInFlight = false let runStateDirtyDuringFetch = false const invalidateRunState = async (): Promise => { - lastRunStateInvalidateAt = Date.now() // cancelRefetch: false — the default (true) cancels an in-flight refetch // and restarts it. When the run-state fetch is slower than the throttle // interval (a busy run congests the server), that livelocks: every @@ -95,6 +94,9 @@ export function useTableEventStream({ runStateDirtyDuringFetch = true return } + // Stamped only when a fetch actually starts — the coalesced path above + // must not reset the throttle clock, or it delays the follow-up. + lastRunStateInvalidateAt = Date.now() runStateFetchInFlight = true try { await queryClient.invalidateQueries( diff --git a/apps/sim/lib/table/workflow-columns.ts b/apps/sim/lib/table/workflow-columns.ts index ffefda28c07..b3538c7f036 100644 --- a/apps/sim/lib/table/workflow-columns.ts +++ b/apps/sim/lib/table/workflow-columns.ts @@ -697,6 +697,7 @@ export async function runWorkflowColumn(opts: { bulkClearWorkflowGroupCells, cancelDispatchById, insertDispatch, + readDispatch, runDispatcherToCompletion, } = await import('./dispatcher') @@ -796,7 +797,6 @@ export async function runWorkflowColumn(opts: { // seeds a returned id into its active-dispatch overlay, which would // resurrect the Run/Stop UI the cancelled SSE event already cleared; null // takes its "no dispatch created" path and rolls the optimistic bump back. - const { readDispatch } = await import('./dispatcher') const current = await readDispatch(dispatchId) if (!current || current.status === 'cancelled' || current.status === 'complete') { logger.info( From f4a4bd315f4b1ab4090c13d4878e49e06c2a2616 Mon Sep 17 00:00:00 2001 From: Theodore Li Date: Tue, 7 Jul 2026 15:24:49 -0700 Subject: [PATCH 4/8] fix(tables): table-wide hasRunning signal for Queueing label, fail fast on seq-read errors --- .../table/[tableId]/dispatches/route.test.ts | 8 ++++++-- .../api/table/[tableId]/dispatches/route.ts | 3 ++- .../table/[tableId]/events/stream/route.ts | 8 +++++++- .../components/table-grid/table-grid.tsx | 12 ++++++----- .../[tableId]/hooks/use-table-event-stream.ts | 1 + apps/sim/hooks/queries/tables.ts | 7 +++++-- apps/sim/lib/api/contracts/tables.ts | 5 +++++ apps/sim/lib/table/dispatcher.ts | 10 ++++++++-- apps/sim/lib/table/events.ts | 20 ++++++++----------- 9 files changed, 49 insertions(+), 25 deletions(-) diff --git a/apps/sim/app/api/table/[tableId]/dispatches/route.test.ts b/apps/sim/app/api/table/[tableId]/dispatches/route.test.ts index 994b77749db..3acddbbe325 100644 --- a/apps/sim/app/api/table/[tableId]/dispatches/route.test.ts +++ b/apps/sim/app/api/table/[tableId]/dispatches/route.test.ts @@ -77,12 +77,15 @@ describe('GET /api/table/[tableId]/dispatches', () => { }) mockCheckAccess.mockResolvedValue({ ok: true, table: buildTable() }) mockListActiveDispatches.mockResolvedValue([]) - mockCountRunningCells.mockResolvedValue({}) + mockCountRunningCells.mockResolvedValue({ byRowId: {}, hasRunning: false }) }) it('returns dispatches and the per-row running map, without a total field', async () => { mockListActiveDispatches.mockResolvedValue([buildDispatchRow()]) - mockCountRunningCells.mockResolvedValue({ 'row-1': 2, 'row-2': 1 }) + mockCountRunningCells.mockResolvedValue({ + byRowId: { 'row-1': 2, 'row-2': 1 }, + hasRunning: true, + }) const response = await makeRequest() const data = await response.json() @@ -99,6 +102,7 @@ describe('GET /api/table/[tableId]/dispatches', () => { }, ]) expect(data.data.runningByRowId).toEqual({ 'row-1': 2, 'row-2': 1 }) + expect(data.data.hasRunning).toBe(true) expect(data.data).not.toHaveProperty('runningCellCount') }) diff --git a/apps/sim/app/api/table/[tableId]/dispatches/route.ts b/apps/sim/app/api/table/[tableId]/dispatches/route.ts index dfc6b2a3bab..a39d3409402 100644 --- a/apps/sim/app/api/table/[tableId]/dispatches/route.ts +++ b/apps/sim/app/api/table/[tableId]/dispatches/route.ts @@ -41,7 +41,7 @@ export const GET = withRouteHandler(async (request: NextRequest, { params }: Rou // Unclaimed `pending` pre-stamps are real queued work while a dispatch is // active; with none active they're abandoned orphans that would pin the // "X running" badge above zero forever. - const runningByRowId = await countRunningCells(tableId, { + const { byRowId: runningByRowId, hasRunning } = await countRunningCells(tableId, { includeUnclaimedPreStamps: rows.length > 0, }) const dispatches: ActiveDispatch[] = rows.map((r) => ({ @@ -59,6 +59,7 @@ export const GET = withRouteHandler(async (request: NextRequest, { params }: Rou data: { dispatches, runningByRowId, + hasRunning, }, }) } catch (error) { diff --git a/apps/sim/app/api/table/[tableId]/events/stream/route.ts b/apps/sim/app/api/table/[tableId]/events/stream/route.ts index d38000ed5b3..2e9cff2eb37 100644 --- a/apps/sim/app/api/table/[tableId]/events/stream/route.ts +++ b/apps/sim/app/api/table/[tableId]/events/stream/route.ts @@ -58,7 +58,7 @@ export const GET = withRouteHandler(async (req: NextRequest, context: RouteConte const stream = new ReadableStream({ async start(controller) { - let lastEventId = fromEventId ?? (await getLatestTableEventId(tableId)) + let lastEventId = fromEventId ?? 0 const deadline = Date.now() + MAX_STREAM_DURATION_MS let nextHeartbeatAt = Date.now() + HEARTBEAT_INTERVAL_MS @@ -98,6 +98,12 @@ export const GET = withRouteHandler(async (req: NextRequest, context: RouteConte } try { + // No replay cursor → tail from the latest event id. Resolved inside + // the try so a Redis failure errors the stream (client reconnects + // with backoff) rather than silently replaying the whole buffer. + if (fromEventId === undefined) { + lastEventId = await getLatestTableEventId(tableId) + } // Initial replay from buffer. const initial = await readTableEventsSince(tableId, lastEventId) if (initial.status === 'pruned') { diff --git a/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/table-grid/table-grid.tsx b/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/table-grid/table-grid.tsx index 0390fa622a0..7ce7926c9b7 100644 --- a/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/table-grid/table-grid.tsx +++ b/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/table-grid/table-grid.tsx @@ -407,11 +407,12 @@ export function TableGrid({ // — the same source the per-row gutter uses). const totalRunning = Object.values(runningByRowId).reduce((sum, n) => sum + n, 0) const hasActiveDispatch = (activeDispatches?.length ?? 0) > 0 - // Loaded-rows scan (early-exit) — the SSE `running` claim events keep this - // current. Loaded rows are an honest proxy for the whole table: the - // dispatcher's active window is what's claimable, and a fresh Run-all has - // nothing running anywhere yet. - const hasRunningCell = useMemo(() => { + // Claimed-cell signal for the "Queueing" vs "N running" label. Two sources + // OR'd: the loaded-rows scan flips instantly via SSE claim events, and the + // server's table-wide flag covers runs whose active window has scrolled + // past the loaded pages (where the scan alone would read "Queueing" for the + // rest of a long run). + const hasRunningLoaded = useMemo(() => { for (const row of rows) { const executions = row.executions if (!executions) continue @@ -421,6 +422,7 @@ export function TableGrid({ } return false }, [rows]) + const hasRunningCell = hasRunningLoaded || (tableRunState?.hasRunning ?? false) // True "select all" total: the filter-scoped COUNT(*) when a filter is active, else the whole // table. Drives the delete-confirm count and the action-bar cell count. diff --git a/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/hooks/use-table-event-stream.ts b/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/hooks/use-table-event-stream.ts index f15b22f83d4..95c47809b65 100644 --- a/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/hooks/use-table-event-stream.ts +++ b/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/hooks/use-table-event-stream.ts @@ -199,6 +199,7 @@ export function useTableEventStream({ const base: TableRunState = prev ?? { dispatches: [], runningByRowId: {}, + hasRunning: false, } const list = base.dispatches // Terminal states drop the dispatch from the overlay; client renders diff --git a/apps/sim/hooks/queries/tables.ts b/apps/sim/hooks/queries/tables.ts index 0c48191d3a7..9828a8e8264 100644 --- a/apps/sim/hooks/queries/tables.ts +++ b/apps/sim/hooks/queries/tables.ts @@ -272,6 +272,8 @@ export function getTableDetailQueryOptions(workspaceId: string, tableId: string) export interface TableRunState { dispatches: ActiveDispatch[] runningByRowId: Record + /** Any in-flight cell claimed by a worker, table-wide. */ + hasRunning: boolean } async function fetchTableRunState(tableId: string, signal?: AbortSignal): Promise { @@ -282,6 +284,7 @@ async function fetchTableRunState(tableId: string, signal?: AbortSignal): Promis return { dispatches: response.data.dispatches, runningByRowId: response.data.runningByRowId, + hasRunning: response.data.hasRunning, } } @@ -349,7 +352,7 @@ async function bumpRunState( await queryClient.cancelQueries({ queryKey: tableKeys.activeDispatches(tableId) }) const snapshot = queryClient.getQueryData(tableKeys.activeDispatches(tableId)) queryClient.setQueryData(tableKeys.activeDispatches(tableId), (prev) => { - const base = prev ?? { dispatches: [], runningByRowId: {} } + const base = prev ?? { dispatches: [], runningByRowId: {}, hasRunning: false } const nextByRow = { ...base.runningByRowId } for (const [rid, n] of Object.entries(stampedByRow)) { nextByRow[rid] = (nextByRow[rid] ?? 0) + n @@ -2095,7 +2098,7 @@ export function useRunColumn({ workspaceId, tableId }: RowMutationContext) { return } queryClient.setQueryData(tableKeys.activeDispatches(tableId), (prev) => { - const base = prev ?? { dispatches: [], runningByRowId: {} } + const base = prev ?? { dispatches: [], runningByRowId: {}, hasRunning: false } if (base.dispatches.some((d) => d.id === dispatchId)) return base const dispatch: ActiveDispatch = { id: dispatchId, diff --git a/apps/sim/lib/api/contracts/tables.ts b/apps/sim/lib/api/contracts/tables.ts index 97daf70146a..da9bc6b0214 100644 --- a/apps/sim/lib/api/contracts/tables.ts +++ b/apps/sim/lib/api/contracts/tables.ts @@ -1340,6 +1340,11 @@ export const listActiveDispatchesContract = defineRouteContract({ * throttle as cell SSE events arrive, plus optimistic stamps on * run-click. */ runningByRowId: z.record(z.string(), z.number().int().positive()), + /** Whether any in-flight cell is actually claimed by a worker + * (`status === 'running'`) — table-wide, unlike the client's + * loaded-rows view. Drives the header's "Queueing" vs "N running" + * label once the run's active window scrolls past the loaded rows. */ + hasRunning: z.boolean(), }) ), }, diff --git a/apps/sim/lib/table/dispatcher.ts b/apps/sim/lib/table/dispatcher.ts index dc13a1ee3c0..9bba4d27dd5 100644 --- a/apps/sim/lib/table/dispatcher.ts +++ b/apps/sim/lib/table/dispatcher.ts @@ -225,7 +225,7 @@ export async function insertDispatch(input: { export async function countRunningCells( tableId: string, opts?: { includeUnclaimedPreStamps?: boolean } -): Promise> { +): Promise<{ byRowId: Record; hasRunning: boolean }> { // `pending` + null-executionId rows are unclaimed pre-stamps. With an active // dispatch they're real queued work (include); with none they're abandoned // orphans that would pin the badge above zero forever (exclude). @@ -234,6 +234,10 @@ export async function countRunningCells( .select({ rowId: tableRowExecutions.rowId, runningCount: sql`count(*)::int`, + // Cells actually claimed by a worker — drives the header's + // "Queueing" vs "N running" label table-wide (the client can only see + // claims on loaded rows; a long run's active window scrolls past them). + claimedCount: sql`count(*) FILTER (WHERE ${tableRowExecutions.status} = 'running')::int`, }) .from(tableRowExecutions) .where( @@ -247,10 +251,12 @@ export async function countRunningCells( ) .groupBy(tableRowExecutions.rowId) const byRowId: Record = {} + let hasRunning = false for (const r of rows) { if (r.runningCount > 0) byRowId[r.rowId] = r.runningCount + if (r.claimedCount > 0) hasRunning = true } - return byRowId + return { byRowId, hasRunning } } /** Read every dispatch on a table whose status is still `pending` or diff --git a/apps/sim/lib/table/events.ts b/apps/sim/lib/table/events.ts index fee8f313687..7d1c8135fe0 100644 --- a/apps/sim/lib/table/events.ts +++ b/apps/sim/lib/table/events.ts @@ -282,9 +282,10 @@ export async function appendTableEvent(event: TableEvent): Promise { const redis = getRedisClient() @@ -296,15 +297,10 @@ export async function getLatestTableEventId(tableId: string): Promise { } return 0 } - try { - const raw = await redis.get(getSeqKey(tableId)) - if (!raw) return 0 - const parsed = Number.parseInt(raw, 10) - return Number.isFinite(parsed) && parsed > 0 ? parsed : 0 - } catch (error) { - logger.warn('getLatestTableEventId failed', { tableId, error: toError(error).message }) - return 0 - } + const raw = await redis.get(getSeqKey(tableId)) + if (!raw) return 0 + const parsed = Number.parseInt(raw, 10) + return Number.isFinite(parsed) && parsed > 0 ? parsed : 0 } /** From 64c85d1bdb498b59b4a9a3599036ea23efb0799b Mon Sep 17 00:00:00 2001 From: Theodore Li Date: Tue, 7 Jul 2026 15:29:09 -0700 Subject: [PATCH 5/8] fix(tables): clear hasRunning on table-wide stop, refetch rows when a run resolves without a dispatch --- apps/sim/hooks/queries/tables.ts | 16 ++++++++++++++-- 1 file changed, 14 insertions(+), 2 deletions(-) diff --git a/apps/sim/hooks/queries/tables.ts b/apps/sim/hooks/queries/tables.ts index 9828a8e8264..8f63b5c1777 100644 --- a/apps/sim/hooks/queries/tables.ts +++ b/apps/sim/hooks/queries/tables.ts @@ -1362,7 +1362,12 @@ export function useCancelTableRuns({ workspaceId, tableId }: RowMutationContext) } nextByRow[rid] = n } - return { ...prev, runningByRowId: nextByRow } + // An unexcluded table-wide stop cancels every claim, so the stale + // table-wide flag must drop with it (else the header reads "0 + // running" until onSettled refetches). Scoped stops leave other rows' + // claims running — keep it. + const hasRunning = scope === 'all' && !filter && !excludedRowIds ? false : prev.hasRunning + return { ...prev, runningByRowId: nextByRow, hasRunning } }) return { snapshots, runStateSnapshot } }, @@ -2091,10 +2096,17 @@ export function useRunColumn({ workspaceId, tableId }: RowMutationContext) { // optimistic counter to the server's still-zero count. const dispatchId = data?.data?.dispatchId if (!dispatchId) { - // No dispatch created → no SSE to reconcile the bump; roll it back. + // No dispatch created (empty scope, or a Stop-all cancelled the run + // during server prep) → no SSE will reconcile the optimistic state. + // Roll back the counter bump, and refetch rows so the optimistic + // pending stamps drop — restoring the onMutate snapshot instead could + // clobber SSE events applied since (other runs, cascades). if (context?.didBumpRunState) { queryClient.setQueryData(tableKeys.activeDispatches(tableId), context.runStateSnapshot) } + if (context?.snapshots) { + void queryClient.invalidateQueries({ queryKey: tableKeys.rowsRoot(tableId) }) + } return } queryClient.setQueryData(tableKeys.activeDispatches(tableId), (prev) => { From 188967c86c668999ce4d72fa22914eb9c38020a4 Mon Sep 17 00:00:00 2001 From: Theodore Li Date: Tue, 7 Jul 2026 15:31:45 -0700 Subject: [PATCH 6/8] fix(tables): don't let dispatch-cancel cleanup mask the original prep failure --- apps/sim/lib/table/workflow-columns.ts | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/apps/sim/lib/table/workflow-columns.ts b/apps/sim/lib/table/workflow-columns.ts index b3538c7f036..3cd45940a0f 100644 --- a/apps/sim/lib/table/workflow-columns.ts +++ b/apps/sim/lib/table/workflow-columns.ts @@ -785,8 +785,16 @@ export async function runWorkflowColumn(opts: { } catch (err) { // Prep failed after the dispatch row was inserted — cancel it so an // orphaned `pending` dispatch can't pin the client's "about to run" - // overlay, then fail the request. - await cancelDispatchById(dispatchId) + // overlay, then fail the request with the ORIGINAL error. The cleanup is + // best-effort: its own failure must not mask the prep failure. + try { + await cancelDispatchById(dispatchId) + } catch (cleanupErr) { + logger.error(`[Cascade] [${requestId}] failed to cancel dispatch after prep failure`, { + dispatchId, + error: toError(cleanupErr).message, + }) + } throw err } From 0c7e829d938adecaec6eba98a38732cf7ff393b6 Mon Sep 17 00:00:00 2001 From: Theodore Li Date: Tue, 7 Jul 2026 15:36:09 -0700 Subject: [PATCH 7/8] fix(tables): reconcile null-dispatch runs via invalidation, not stale snapshot restore --- apps/sim/hooks/queries/tables.ts | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/apps/sim/hooks/queries/tables.ts b/apps/sim/hooks/queries/tables.ts index 8f63b5c1777..e24ba44f9e0 100644 --- a/apps/sim/hooks/queries/tables.ts +++ b/apps/sim/hooks/queries/tables.ts @@ -2098,11 +2098,11 @@ export function useRunColumn({ workspaceId, tableId }: RowMutationContext) { if (!dispatchId) { // No dispatch created (empty scope, or a Stop-all cancelled the run // during server prep) → no SSE will reconcile the optimistic state. - // Roll back the counter bump, and refetch rows so the optimistic - // pending stamps drop — restoring the onMutate snapshot instead could - // clobber SSE events applied since (other runs, cascades). + // Refetch both caches rather than restoring the onMutate snapshots — + // either restore could clobber fresher state applied since (SSE + // events, throttled refetches, a concurrent Stop-all's clear). if (context?.didBumpRunState) { - queryClient.setQueryData(tableKeys.activeDispatches(tableId), context.runStateSnapshot) + void queryClient.invalidateQueries({ queryKey: tableKeys.activeDispatches(tableId) }) } if (context?.snapshots) { void queryClient.invalidateQueries({ queryKey: tableKeys.rowsRoot(tableId) }) From a5a39a86831628773d0923a0cc4223c66680df5e Mon Sep 17 00:00:00 2001 From: Theodore Li Date: Tue, 7 Jul 2026 15:41:18 -0700 Subject: [PATCH 8/8] fix(tables): widen warm-cache remount check to either query cache --- .../tables/[tableId]/hooks/use-table-event-stream.ts | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/hooks/use-table-event-stream.ts b/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/hooks/use-table-event-stream.ts index 95c47809b65..65416e113d6 100644 --- a/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/hooks/use-table-event-stream.ts +++ b/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/hooks/use-table-event-stream.ts @@ -409,9 +409,15 @@ export function useTableEventStream({ // In-SPA remount over a warm cache (table A → B → back to A within // staleTime): the tail starts at "latest", so transitions that fired while // unmounted were neither refetched (cache still fresh) nor replayed. - // Reconcile once. Cold mounts have no cached run-state → skip, the - // queries are already fetching. - if (queryClient.getQueryState(tableKeys.activeDispatches(tableId))?.data !== undefined) { + // Reconcile once — either cache being warm is enough (they can evict + // independently). Cold mounts have neither → skip, the queries are + // already fetching. + const hasWarmRunState = + queryClient.getQueryState(tableKeys.activeDispatches(tableId))?.data !== undefined + const hasWarmRows = queryClient + .getQueriesData({ queryKey: tableKeys.rowsRoot(tableId) }) + .some(([, data]) => data !== undefined) + if (hasWarmRunState || hasWarmRows) { void queryClient.invalidateQueries({ queryKey: tableKeys.rowsRoot(tableId) }) void invalidateRunState() }