mirror of
https://github.com/simstudioai/sim.git
synced 2026-09-24 15:45:35 +08:00
fix(table): dispatcher cold-start, live run counter, smooth typewriter (#4675)
* fix(table): cut dispatcher cold-start by lazy-loading heavy import chains The trigger.dev table-run-dispatcher spent ~6s in module init before its first batchTriggerAndWait — it imports lib/table/service for getTableById, which eagerly imported lib/table/trigger → @/lib/webhooks/processor → webhook-execution + executor, dragging the entire workflow-execution stack into the dispatcher container even though it never fires a trigger. - trigger.ts lazy-imports the webhook processor + polling utils inside fireTableTrigger (the only consumer), so importing service no longer pulls the executor. - buildEnqueueItems only imports the cell job (for the inline `runner`) on the database backend; the trigger.dev backend triggers by task id and ignores runner. * fix(table): run counter + gutter Stop update instantly on Run The "X running" badge, per-row gutter Stop button, and runningByRowId map stayed at zero after clicking Run until a manual refetch. useRunColumn optimistically stamped cells pending in the rows cache but never bumped the activeDispatches counter — so when the dispatcher's real pending SSE arrived, applyCell saw the cell was already in-flight (wasInFlight === isInFlight) and skipped the counter delta. The optimistic stamp ate the transition. - onMutate now bumps runningCellCount / runningByRowId by the cells it stamps, snapshotting prior run-state for rollback on error. - onSuccess seeds the dispatch into the overlay list from the response instead of invalidating activeDispatches (a refetch would reset the optimistic counter to the server's still-zero count before the dispatcher stamps). * fix(table): drive cell typewriter with rAF so concurrent reveals stay smooth The character-by-character reveal used a per-cell setInterval. When many cells reveal at once (a Run-all completing in waves), the independent interval callbacks fire at uncoordinated times and each forces its own render + layout/paint — O(cells) reflows over an un-virtualized grid, so it degrades as more cells fill. Switch to requestAnimationFrame: all cells' callbacks run before one paint, so React batches them into a single render + paint per frame regardless of cell count. Reveal length is derived from elapsed time, so a dropped frame catches up instead of slowing the animation. * fix(table): roll back optimistic run counter when no dispatch is created useRunColumn.onSuccess returned early on a null dispatchId (no matching groups / eligible rows) without undoing the onMutate counter bump — and no SSE would arrive to correct it, leaving the counter permanently inflated. Restore the pre-mutation run-state on that path, mirroring onError. * chore(table): tighten inline comments on dispatcher cold-start fixes
This commit is contained in:
+14
-7
@@ -296,6 +296,10 @@ const TYPEWRITER_MS_PER_CHAR = 15
|
||||
* value statically — animation fires only for subsequent updates, which in
|
||||
* practice means SSE-driven workflow completions arriving via
|
||||
* `useTableEventStream → applyCell()`.
|
||||
*
|
||||
* rAF-driven (not `setInterval`) so concurrent reveals batch into one
|
||||
* render/paint per frame instead of O(cells) uncoordinated reflows; reveal
|
||||
* length is elapsed-time based so dropped frames catch up rather than slow.
|
||||
*/
|
||||
function useTypewriter(text: string | null): string | null {
|
||||
const [revealed, setRevealed] = useState<string | null>(text)
|
||||
@@ -317,14 +321,17 @@ function useTypewriter(text: string | null): string | null {
|
||||
return
|
||||
}
|
||||
|
||||
const full = text
|
||||
const start = performance.now()
|
||||
let raf = 0
|
||||
const tick = (now: number) => {
|
||||
const chars = Math.min(full.length, Math.floor((now - start) / TYPEWRITER_MS_PER_CHAR))
|
||||
setRevealed(full.slice(0, chars))
|
||||
if (chars < full.length) raf = requestAnimationFrame(tick)
|
||||
}
|
||||
setRevealed('')
|
||||
let i = 0
|
||||
const id = window.setInterval(() => {
|
||||
i++
|
||||
setRevealed(text.slice(0, i))
|
||||
if (i >= text.length) window.clearInterval(id)
|
||||
}, TYPEWRITER_MS_PER_CHAR)
|
||||
return () => window.clearInterval(id)
|
||||
raf = requestAnimationFrame(tick)
|
||||
return () => cancelAnimationFrame(raf)
|
||||
}, [text])
|
||||
|
||||
return revealed
|
||||
|
||||
@@ -1336,10 +1336,12 @@ export function useRunColumn({ workspaceId, tableId }: RowMutationContext) {
|
||||
queryClient.getQueryData<TableDefinition>(tableKeys.detail(tableId))?.schema
|
||||
.workflowGroups ?? []
|
||||
const groupsById = new Map(groups.map((g) => [g.id, g]))
|
||||
// Tally cells stamped per row to bump the run-state counter in lockstep.
|
||||
const stampedByRow: Record<string, number> = {}
|
||||
const snapshots = await snapshotAndMutateRows(queryClient, tableId, (r) => {
|
||||
if (targetRowIds && !targetRowIds.has(r.id)) return null
|
||||
const executions = r.executions ?? {}
|
||||
let changed = false
|
||||
let stamped = 0
|
||||
const next: RowExecutions = { ...executions }
|
||||
const nextData = { ...r.data }
|
||||
for (const groupId of targetGroupIds) {
|
||||
@@ -1367,20 +1369,72 @@ export function useRunColumn({ workspaceId, tableId }: RowMutationContext) {
|
||||
if (o.columnName in nextData) nextData[o.columnName] = null
|
||||
}
|
||||
}
|
||||
changed = true
|
||||
stamped++
|
||||
}
|
||||
if (!changed) return null
|
||||
if (stamped === 0) return null
|
||||
stampedByRow[r.id] = stamped
|
||||
return { ...r, data: nextData, executions: next }
|
||||
})
|
||||
return { snapshots }
|
||||
|
||||
// Bump the counter to match the stamped cells. Without it the "X running"
|
||||
// badge + gutter Stop stay at zero until a refetch: the optimistic stamp
|
||||
// already marks the cell in-flight, so the dispatcher's `pending` SSE
|
||||
// sees no `wasInFlight` transition and never bumps the counter.
|
||||
const runStateSnapshot = queryClient.getQueryData<TableRunState>(
|
||||
tableKeys.activeDispatches(tableId)
|
||||
)
|
||||
const totalStamped = Object.values(stampedByRow).reduce((s, n) => s + n, 0)
|
||||
if (totalStamped > 0) {
|
||||
queryClient.setQueryData<TableRunState>(tableKeys.activeDispatches(tableId), (prev) => {
|
||||
const base = prev ?? { dispatches: [], runningCellCount: 0, runningByRowId: {} }
|
||||
const nextByRow = { ...base.runningByRowId }
|
||||
for (const [rid, n] of Object.entries(stampedByRow)) {
|
||||
nextByRow[rid] = (nextByRow[rid] ?? 0) + n
|
||||
}
|
||||
return {
|
||||
...base,
|
||||
runningCellCount: base.runningCellCount + totalStamped,
|
||||
runningByRowId: nextByRow,
|
||||
}
|
||||
})
|
||||
}
|
||||
return { snapshots, runStateSnapshot, didBumpRunState: totalStamped > 0 }
|
||||
},
|
||||
onError: (_err, _variables, context) => {
|
||||
if (context?.snapshots) restoreCachedWorkflowCells(queryClient, context.snapshots)
|
||||
// Roll back the optimistic counter bump (snapshot may be undefined).
|
||||
if (context?.didBumpRunState) {
|
||||
queryClient.setQueryData(tableKeys.activeDispatches(tableId), context.runStateSnapshot)
|
||||
}
|
||||
},
|
||||
onSuccess: () => {
|
||||
// Seed the active-dispatch overlay immediately (insertDispatch ran
|
||||
// server-side before responding); rows cache stays owned by SSE.
|
||||
void queryClient.invalidateQueries({ queryKey: tableKeys.activeDispatches(tableId) })
|
||||
onSuccess: (data, { groupIds, runMode = 'all', rowIds }, context) => {
|
||||
// Seed the dispatch into the overlay (drives resolveCellExec for
|
||||
// ahead-of-cursor rows) from the response — refetching would reset the
|
||||
// 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.
|
||||
if (context?.didBumpRunState) {
|
||||
queryClient.setQueryData(tableKeys.activeDispatches(tableId), context.runStateSnapshot)
|
||||
}
|
||||
return
|
||||
}
|
||||
queryClient.setQueryData<TableRunState>(tableKeys.activeDispatches(tableId), (prev) => {
|
||||
const base = prev ?? { dispatches: [], runningCellCount: 0, runningByRowId: {} }
|
||||
if (base.dispatches.some((d) => d.id === dispatchId)) return base
|
||||
const dispatch: ActiveDispatch = {
|
||||
id: dispatchId,
|
||||
status: 'pending',
|
||||
mode: runMode,
|
||||
isManualRun: true,
|
||||
cursor: -1,
|
||||
scope: {
|
||||
groupIds,
|
||||
...(rowIds && rowIds.length > 0 ? { rowIds } : {}),
|
||||
},
|
||||
}
|
||||
return { ...base, dispatches: [...base.dispatches, dispatch] }
|
||||
})
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
@@ -9,8 +9,6 @@
|
||||
import { createLogger } from '@sim/logger'
|
||||
import { generateShortId } from '@sim/utils/id'
|
||||
import type { RowData, TableRow, TableSchema } from '@/lib/table/types'
|
||||
import { fetchActiveWebhooks } from '@/lib/webhooks/polling/utils'
|
||||
import { processPolledWebhookEvent } from '@/lib/webhooks/processor'
|
||||
|
||||
const logger = createLogger('TableTrigger')
|
||||
|
||||
@@ -57,6 +55,10 @@ export async function fireTableTrigger(
|
||||
requestId: string
|
||||
): Promise<void> {
|
||||
try {
|
||||
// Lazy: the webhook utils/processor pull in the executor + blocks stack.
|
||||
// Eager imports would force every `lib/table/service` consumer (e.g. the
|
||||
// dispatcher) to pay that cold-start even when no trigger fires.
|
||||
const { fetchActiveWebhooks } = await import('@/lib/webhooks/polling/utils')
|
||||
const webhooks = await fetchActiveWebhooks('table')
|
||||
if (webhooks.length === 0) return
|
||||
|
||||
@@ -74,6 +76,8 @@ export async function fireTableTrigger(
|
||||
|
||||
if (matching.length === 0) return
|
||||
|
||||
const { processPolledWebhookEvent } = await import('@/lib/webhooks/processor')
|
||||
|
||||
logger.info(
|
||||
`[${requestId}] Firing ${matching.length} trigger(s) for ${rows.length} ${eventType} event(s) in table ${tableId}`
|
||||
)
|
||||
|
||||
@@ -203,11 +203,18 @@ export function buildPendingRuns(
|
||||
/** Build the per-cell `{payload, options}` items for `queue.batchEnqueue` /
|
||||
* `queue.batchEnqueueAndWait`. Hydrates trigger.dev tags, concurrency keys,
|
||||
* the inline runner, and the cancel key the inline backend uses to map a
|
||||
* Stop click to the in-flight cell's AbortController. */
|
||||
* Stop click to the in-flight cell's AbortController.
|
||||
*
|
||||
* `runner` is only used by the database backend; trigger.dev triggers by task
|
||||
* id. The cell-job import pulls in the executor + blocks stack, so skip it on
|
||||
* trigger.dev to avoid a multi-second dispatcher cold-start. */
|
||||
export async function buildEnqueueItems(
|
||||
pendingRuns: WorkflowGroupCellPayload[]
|
||||
): Promise<Array<{ payload: WorkflowGroupCellPayload; options: EnqueueOptions }>> {
|
||||
const { executeWorkflowGroupCellJob } = await import('@/background/workflow-column-execution')
|
||||
const runner = isTriggerDevEnabled
|
||||
? undefined
|
||||
: ((await import('@/background/workflow-column-execution'))
|
||||
.executeWorkflowGroupCellJob as EnqueueOptions['runner'])
|
||||
return pendingRuns.map((runOpts) => ({
|
||||
payload: runOpts,
|
||||
options: {
|
||||
@@ -225,7 +232,7 @@ export async function buildEnqueueItems(
|
||||
concurrencyKey: runOpts.tableId,
|
||||
concurrencyLimit: TABLE_CONCURRENCY_LIMIT,
|
||||
tags: cellTagsFor(runOpts),
|
||||
runner: executeWorkflowGroupCellJob as EnqueueOptions['runner'],
|
||||
...(runner ? { runner } : {}),
|
||||
cancelKey: cellCancelKey(runOpts.tableId, runOpts.rowId, runOpts.groupId),
|
||||
},
|
||||
}))
|
||||
|
||||
Reference in New Issue
Block a user