perf(tables): stop a table write refetching every loaded page in the tab that made it (#6698)

apps/docs/content/docs/en/tables/using-in-workflows.mdx

@@ -125,7 +125,7 @@ After the run, the table holds the enriched rows. The next run queries them agai
 
 **Iterate row by row.** Wrap a Query → process → update cycle in a [Loop block](/workflows/blocks/loop) to handle one row at a time. This runs sequentially, slower than a batch update but useful when each row needs its own multi-step logic. Inside the loop the Agent reads the current row and an Update Row by ID writes its result.
 
-**Paginate large reads.** Query Rows returns at most 1000 rows. When `totalCount` exceeds your **Limit**, increase **Offset** on each pass (0, then 100, then 200) to walk through the whole table, typically inside a Loop.
+**Paginate large reads.** Query Rows returns at most 1000 rows, and a page can also end early once its rows reach the response size budget — so a page may come back shorter than your **Limit** even when more rows match. Advance **Offset** by the `rowCount` you actually received, not by the Limit you asked for, and keep going while `nextCursor` is set. Stop when `nextCursor` is null. Stepping by the Limit instead skips whatever a short page left behind.
 
 ## Inspecting reads and writes
 

apps/sim/app/api/table/[tableId]/rows/[rowId]/route.ts

@@ -3,6 +3,7 @@ import { userTableRows } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
 import { and, eq } from 'drizzle-orm'
 import { type NextRequest, NextResponse } from 'next/server'
+import { readClientId } from '@/lib/api/client-id'
 import {
   deleteTableRowContract,
   getTableQuerySchema,
@@ -14,7 +15,7 @@ import { generateRequestId } from '@/lib/core/utils/request'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 import type { RowData, TableSchema } from '@/lib/table'
 import { updateRow } from '@/lib/table'
-import { signalTableRowsChanged } from '@/lib/table/events'
+import { signalTableRowsChangedByActor } from '@/lib/table/events'
 import { performDeleteTableRow } from '@/lib/table/orchestration'
 import {
   createTableRowsResponse,
@@ -172,7 +173,7 @@ export const PATCH = withRouteHandler(async (request: NextRequest, context: RowR
     )
 
     // Live-collab: tell open viewers the change landed so they refetch.
-    signalTableRowsChanged(tableId)
+    signalTableRowsChangedByActor(tableId, readClientId(request))
     // Only `null` when a `cancellationGuard` is supplied and the SQL guard
     // rejects the write — this route doesn't pass one, so reaching null is a bug.
     if (!updatedRow) throw new Error('updateRow returned null without a cancellationGuard')
@@ -251,7 +252,7 @@ export const DELETE = withRouteHandler(async (request: NextRequest, context: Row
     }
 
     // Live-collab: tell open viewers the change landed so they refetch.
-    signalTableRowsChanged(tableId)
+    signalTableRowsChangedByActor(tableId, readClientId(request))
 
     return NextResponse.json({
       success: true,

apps/sim/app/api/table/[tableId]/rows/route.ts

@@ -1,5 +1,6 @@
 import { createLogger } from '@sim/logger'
 import { type NextRequest, NextResponse } from 'next/server'
+import { readClientId } from '@/lib/api/client-id'
 import {
   type BatchInsertTableRowsBodyInput,
   batchUpdateTableRowsBodySchema,
@@ -26,7 +27,7 @@ import {
   validateRowSize,
 } from '@/lib/table'
 import { TableQueryValidationError } from '@/lib/table/errors'
-import { signalTableRowsChanged } from '@/lib/table/events'
+import { signalTableRowsChanged, signalTableRowsChangedByActor } from '@/lib/table/events'
 import { isTablePredicate, predicateToFilter } from '@/lib/table/query-builder/converters'
 import {
   validatePredicateShape,
@@ -254,7 +255,9 @@ export const POST = withRouteHandler(
         table,
         requestId
       )
-      signalTableRowsChanged(tableId)
+      // Attributed unlike the batch path above: the acting tab's insert deliberately avoids
+      // invalidating the rows root to prevent flicker, which an unattributed echo would undo.
+      signalTableRowsChangedByActor(tableId, readClientId(request))
 
       const responseBody = {
         success: true,

apps/sim/app/workspace/[workspaceId]/tables/[tableId]/hooks/use-table-event-stream.ts

@@ -5,6 +5,7 @@ import { toast } from '@sim/emcn'
 import { createLogger } from '@sim/logger'
 import { backoffWithJitter } from '@sim/utils/retry'
 import { useQueryClient } from '@tanstack/react-query'
+import { getClientFingerprint } from '@/lib/api/client-id'
 import type { ActiveDispatch } from '@/lib/api/contracts/tables'
 import type {
   RowData,
@@ -245,6 +246,30 @@ export function useTableEventStream({
       }, ROWS_INVALIDATE_DEBOUNCE_MS)
     }
 
+    /**
+     * This tab's fingerprint as it appears on a broadcast it caused. Resolved once, asynchronously;
+     * until it lands `applyEdit` simply takes the refetch path, which is the pre-existing behavior.
+     */
+    let ownFingerprint: string | undefined
+    void getClientFingerprint().then((fingerprint) => {
+      ownFingerprint = fingerprint
+    })
+
+    /**
+     * A manual row edit landed. Refetch the rows so the winning last-write value shows live —
+     * unless this tab is the one that made it.
+     *
+     * The signal names its originator only for writes whose mutation hook already applies the
+     * server's answer to every cached rows query, active or not (single-row create, update,
+     * delete). For those the refetch is pure duplication: on a scrolled table it re-fetches every
+     * loaded page, and on delete it races the refetch the hook itself issued. Other tabs see
+     * someone else's fingerprint and refetch normally; an unattributed edit refetches everywhere.
+     */
+    const applyEdit = (event: Extract<TableEvent, { kind: 'edit' }>): void => {
+      if (event.originatorId && event.originatorId === ownFingerprint) return
+      scheduleRowsInvalidate()
+    }
+
     const applyCell = (event: Extract<TableEvent, { kind: 'cell' }>): void => {
       void snapshotAndMutateRows(queryClient, tableId, (row) => applyCellEventToRow(row, event), {
         cancelInFlight: false,
@@ -445,9 +470,7 @@ export function useTableEventStream({
           else if (entry.event?.kind === 'dispatch') applyDispatch(entry.event)
           else if (entry.event?.kind === 'job') applyJob(entry.event)
           else if (entry.event?.kind === 'usageLimitReached') applyUsageLimit(entry.event)
-          // A collaborator's manual edit: refetch rows (debounced) so the winning
-          // last-write value shows live, in this client's own wire format.
-          else if (entry.event?.kind === 'edit') scheduleRowsInvalidate()
+          else if (entry.event?.kind === 'edit') applyEdit(entry.event)
           // A collaborator changed the table structure: mirror the local
           // invalidateTableSchema set — the definition (exact, so rows stay on the
           // debounce), the run-state + enrichment sibling queries under detail (a group

apps/sim/hooks/queries/tables.ts

@@ -1071,6 +1071,18 @@ export function useUpdateTableRow({ workspaceId, tableId }: RowMutationContext)
           updatedAt: serverRow.updatedAt,
         }
       })
+
+      // `patchCachedRows` rewrites values in place, which is the whole answer for the default
+      // view. It cannot be for a filtered or column-sorted one: editing a cell can move a row in
+      // or out of the filter and change its sort position and `totalCount`, none of which a
+      // per-row patch can express. Those views are refetched instead — the same split
+      // `useCreateTableRow` makes, and previously supplied by the broadcast this write no longer
+      // makes the acting tab honor.
+      queryClient.invalidateQueries({
+        queryKey: tableKeys.rowsRoot(tableId),
+        exact: false,
+        predicate: (query) => !isDefaultOrderRowsQuery(query.queryKey),
+      })
     },
     onError: (error, _vars, context) => {
       if (context?.previousQueries) {

apps/sim/lib/api/client-id.test.ts

@@ -0,0 +1,54 @@
+/**
+ * @vitest-environment node
+ */
+import { describe, expect, it } from 'vitest'
+import { CLIENT_ID_HEADER, fingerprintClientId, readClientId } from '@/lib/api/client-id'
+
+describe('readClientId', () => {
+  it('reads the sending tab id off the request', () => {
+    const request = new Request('https://sim.ai/api/table/t1/rows', {
+      headers: { [CLIENT_ID_HEADER]: 'tab-abc' },
+    })
+    expect(readClientId(request)).toBe('tab-abc')
+  })
+
+  /** Absent must read as "unattributed" — the signal then makes every client refetch, as before. */
+  it('is undefined when the caller sent no id', () => {
+    const request = new Request('https://sim.ai/api/table/t1/rows')
+    expect(readClientId(request)).toBeUndefined()
+  })
+
+  /**
+   * The value is caller-controlled and is broadcast to every subscriber of the table, so an
+   * over-long one is dropped rather than fanned out.
+   */
+  it('drops an over-long id instead of broadcasting it', () => {
+    const request = new Request('https://sim.ai/api/table/t1/rows', {
+      headers: { [CLIENT_ID_HEADER]: 'x'.repeat(65) },
+    })
+    expect(readClientId(request)).toBeUndefined()
+  })
+})
+
+/**
+ * Every subscriber of a table sees every broadcast, so what gets published must not be replayable.
+ * If the raw id travelled, a collaborator could read it off the stream, send it as their own
+ * header, and have their write attributed to someone else's tab — which would then suppress a
+ * refetch it genuinely needed and sit on stale rows.
+ */
+describe('fingerprintClientId', () => {
+  it('is stable for the same id, so a tab recognises its own broadcast', async () => {
+    expect(await fingerprintClientId('tab-abc')).toBe(await fingerprintClientId('tab-abc'))
+  })
+
+  it('differs between tabs, so one tab never suppresses on another tab’s write', async () => {
+    expect(await fingerprintClientId('tab-abc')).not.toBe(await fingerprintClientId('tab-xyz'))
+  })
+
+  it('does not reveal the id it was derived from', async () => {
+    const fingerprint = await fingerprintClientId('tab-abc')
+    expect(fingerprint).not.toContain('tab-abc')
+    // SHA-256 hex — knowing this cannot produce the header value that would match it.
+    expect(fingerprint).toMatch(/^[0-9a-f]{64}$/)
+  })
+})

apps/sim/lib/api/client-id.ts

@@ -0,0 +1,77 @@
+import { generateShortId } from '@sim/utils/id'
+
+/**
+ * Header naming the browser tab that sent a request.
+ *
+ * Shared by the client that sets it and the route handlers that read it. An opaque correlation
+ * token, never an authorization input.
+ */
+export const CLIENT_ID_HEADER = 'x-sim-client-id'
+
+/**
+ * Generated ids are {@link generateShortId} length; the ceiling is slack for that, not a format.
+ * Bounded because the value is caller-controlled and gets fanned out to every subscriber of a
+ * table — uncapped, one request could inflate every broadcast payload it triggers.
+ */
+const MAX_CLIENT_ID_LENGTH = 64
+
+let cachedClientId: string | undefined
+
+/**
+ * An id for this browser tab, generated once per page load and not stable across reloads.
+ *
+ * Deliberately per-TAB rather than per-user or per-session: its only consumer compares it against
+ * the originator stamped on a broadcast, so two tabs belonging to the same user must not share one.
+ * A shared id would make the second tab ignore the first tab's edits and silently go stale.
+ *
+ * Returns `undefined` on the server, where there is no tab to identify.
+ */
+export function getClientId(): string | undefined {
+  if (typeof window === 'undefined') return undefined
+  cachedClientId ??= generateShortId()
+  return cachedClientId
+}
+
+/**
+ * The sending tab's id, as seen by a route handler. Absent for server-to-server callers, for any
+ * client that did not send one, and for an over-long value — all read as "unattributed", never as
+ * "not the actor".
+ *
+ * Untrusted, and never safe to broadcast as-is: see {@link fingerprintClientId}.
+ */
+export function readClientId(request: Request): string | undefined {
+  const raw = request.headers.get(CLIENT_ID_HEADER)
+  return raw && raw.length <= MAX_CLIENT_ID_LENGTH ? raw : undefined
+}
+
+/**
+ * One-way digest of a tab id, for naming the originator of a broadcast.
+ *
+ * The raw id must never travel on a broadcast. Every subscriber of a table sees every event, so a
+ * raw id would be observable by any collaborator, who could then replay it as their own
+ * `x-sim-client-id` — their write would be attributed to your tab, your tab would suppress its
+ * refetch, and it would sit on stale rows. Publishing the digest instead means matching it
+ * requires already knowing the id, which only the tab that generated it does.
+ *
+ * Web Crypto rather than `node:crypto` so one implementation serves both sides — the server
+ * stamping the event and the browser recognising its own — with no chance of the two disagreeing.
+ */
+export async function fingerprintClientId(clientId: string): Promise<string> {
+  const digest = await crypto.subtle.digest('SHA-256', new TextEncoder().encode(clientId))
+  return Array.from(new Uint8Array(digest))
+    .map((byte) => byte.toString(16).padStart(2, '0'))
+    .join('')
+}
+
+let cachedFingerprint: string | undefined
+
+/**
+ * This tab's fingerprint, as it appears on a broadcast it caused. `undefined` on the server, and
+ * until the first digest resolves — callers must treat that as "not me" and take the normal path.
+ */
+export async function getClientFingerprint(): Promise<string | undefined> {
+  const clientId = getClientId()
+  if (!clientId) return undefined
+  cachedFingerprint ??= await fingerprintClientId(clientId)
+  return cachedFingerprint
+}

apps/sim/lib/api/client/request.test.ts

@@ -4,6 +4,7 @@
 import { afterEach, describe, expect, it, vi } from 'vitest'
 import { z } from 'zod'
 import { requestJson } from '@/lib/api/client/request'
+import { CLIENT_ID_HEADER } from '@/lib/api/client-id'
 import { listKnowledgeDocumentsContract } from '@/lib/api/contracts/knowledge'
 import { defineRouteContract } from '@/lib/api/contracts/types'
 
@@ -87,3 +88,38 @@ describe('requestJson query serialization', () => {
     expect(url).toContain('tags=a&tags=b')
   })
 })
+
+/**
+ * The tab id rides on every request so a broadcast raised by one can be attributed back to the tab
+ * that caused it. Asserted here rather than on the reader, because the header being *sent* is the
+ * half that silently does nothing if it regresses.
+ */
+describe('requestJson client id header', () => {
+  const contract = defineRouteContract({
+    method: 'GET',
+    path: '/api/test',
+    response: { mode: 'json', schema: z.object({ ok: z.boolean() }) },
+  })
+
+  function sentHeaders(fetchMock: ReturnType<typeof mockFetchReturning>): Record<string, string> {
+    return (fetchMock.mock.calls[0][1] as RequestInit).headers as Record<string, string>
+  }
+
+  it('sends the tab id in the browser', async () => {
+    vi.stubGlobal('window', {})
+    const fetchMock = mockFetchReturning({ ok: true })
+
+    await requestJson(contract, {})
+
+    expect(sentHeaders(fetchMock)[CLIENT_ID_HEADER]).toEqual(expect.any(String))
+  })
+
+  it('omits it on the server, where there is no tab to name', async () => {
+    vi.stubGlobal('window', undefined)
+    const fetchMock = mockFetchReturning({ ok: true })
+
+    await requestJson(contract, {})
+
+    expect(sentHeaders(fetchMock)[CLIENT_ID_HEADER]).toBeUndefined()
+  })
+})

apps/sim/lib/api/client/request.ts

@@ -1,4 +1,5 @@
 import { ApiClientError } from '@/lib/api/client/errors'
+import { CLIENT_ID_HEADER, getClientId } from '@/lib/api/client-id'
 import type {
   AnyApiRouteContract,
   ApiSchema,
@@ -104,6 +105,10 @@ function buildHeaders(headers: unknown, hasBody: boolean): Record<string, string
     output['Content-Type'] = 'application/json'
   }
 
+  /** Set here rather than per call site so every request carries it without a decision to get wrong. */
+  const clientId = getClientId()
+  if (clientId) output[CLIENT_ID_HEADER] = clientId
+
   if (headers && typeof headers === 'object') {
     for (const [key, value] of Object.entries(headers as Record<string, unknown>)) {
       if (typeof value === 'string') output[key] = value

apps/sim/lib/table/events.attribution.test.ts

@@ -0,0 +1,48 @@
+/**
+ * @vitest-environment node
+ */
+import { readdir, readFile } from 'node:fs/promises'
+import { join } from 'node:path'
+import { describe, expect, it } from 'vitest'
+
+/**
+ * `signalTableRowsChangedByActor` lets the acting tab skip its own refetch, which is only sound
+ * where that tab's mutation hook already applies the server's answer to every cached rows query.
+ * That invariant lives in `hooks/queries/tables.ts` — nothing in the type system ties it to the
+ * call site, so a well-meaning fourth call would silently strand that client on stale rows.
+ *
+ * This pins the allowlist. If you are here because it failed: adding a call means proving the
+ * calling route's client hook reconciles locally, then adding it below. Removing one is always safe.
+ */
+const ATTRIBUTED_CALL_SITES = [
+  'app/api/table/[tableId]/rows/route.ts',
+  'app/api/table/[tableId]/rows/[rowId]/route.ts',
+] as const
+
+const APP_ROOT = join(import.meta.dirname, '../..')
+/** Declares the function; matching its own definition would say nothing about call sites. */
+const DECLARING_MODULE = 'lib/table/events.ts'
+
+async function* walk(dir: string): AsyncGenerator<string> {
+  for (const entry of await readdir(dir, { withFileTypes: true })) {
+    if (entry.name === 'node_modules' || entry.name === '.next') continue
+    const full = join(dir, entry.name)
+    if (entry.isDirectory()) yield* walk(full)
+    else if (entry.name.endsWith('.ts') && !entry.name.includes('.test.')) yield full
+  }
+}
+
+describe('signalTableRowsChangedByActor call sites', () => {
+  it('is called only where the acting tab reconciles the write locally', async () => {
+    const callers: string[] = []
+    for await (const file of walk(APP_ROOT)) {
+      const source = await readFile(file, 'utf8')
+      if (!source.includes('signalTableRowsChangedByActor(')) continue
+      const relative = file.slice(APP_ROOT.length + 1)
+      if (relative === DECLARING_MODULE) continue
+      callers.push(relative)
+    }
+
+    expect(callers.sort()).toEqual([...ATTRIBUTED_CALL_SITES].sort())
+  })
+})

apps/sim/lib/table/events.ts

@@ -14,6 +14,7 @@
  * in-memory `lastEventId` no longer matches), so both are intentionally fixed here.
  */
 
+import { fingerprintClientId } from '@/lib/api/client-id'
 import {
   appendEvent,
   type EventLogConfig,
@@ -118,6 +119,14 @@ export type TableEvent =
        *  translation on the wire. */
       kind: 'edit'
       tableId: string
+      /**
+       * One-way digest naming the tab whose request caused this edit, when that tab is known to
+       * reconcile the change locally — so it can skip refetching what it already holds. Digested
+       * rather than raw because every subscriber sees this field; a raw id could be replayed by a
+       * collaborator to make someone else's tab suppress a refetch it needed. Absent means
+       * "unattributed" — every client refetches, which is the pre-existing behavior.
+       */
+      originatorId?: string
     }
   | {
       /** A user changed the table's structure (added/updated/deleted a column, or
@@ -179,22 +188,39 @@ export async function appendTableEvent(event: TableEvent): Promise<TableEventEnt
   })
 }
 
-// The mutating client receives its own signal too (the stream carries no originator id)
-// and self-refetches. Data-correct — signals fire after the write commits, so the refetch
-// returns the committed state, and in-flight edits are protected by the update/delete
-// hooks' cancelQueries. The one caveat is an own row-CREATE on a scrolled, multi-page
-// table, which can briefly reshuffle loaded pages (the create hook otherwise skips that
-// refetch); if that ever proves visible, stamp an originator id so the actor ignores its
-// own signal.
-
 /**
  * Signal collaborators that a user changed row data so they refetch the rows live.
  * Fire-and-forget — a Redis blip must never fail the write that triggered it.
+ *
+ * Unattributed, so every subscriber refetches — including the client that made the write. Correct
+ * for any write whose client hook does not already apply the server's answer locally: bulk and
+ * filter-scoped writes, imports, copilot edits, run dispatch.
  */
 export function signalTableRowsChanged(tableId: string): void {
   void appendTableEvent({ kind: 'edit', tableId })
 }
 
+/**
+ * As {@link signalTableRowsChanged}, but names the tab that caused the write so that tab can skip
+ * its own refetch — which for it is pure duplication: on a scrolled table the broadcast re-fetches
+ * every loaded page, and on delete it races the refetch the hook already issued.
+ *
+ * Use ONLY where the client hook reconciles the change from the mutation's own response across
+ * every cached rows query — the single-row create, update, and delete paths. On a bulk or
+ * filter-scoped write the actor genuinely needs the refetch, and suppressing it would leave that
+ * client showing stale rows. The call sites are pinned by `events.attribution.test.ts`.
+ */
+export function signalTableRowsChangedByActor(tableId: string, clientId: string | undefined): void {
+  if (!clientId) {
+    void appendTableEvent({ kind: 'edit', tableId })
+    return
+  }
+  // Digested, never raw — every subscriber sees this event, and a raw id would be replayable.
+  void fingerprintClientId(clientId).then((originatorId) =>
+    appendTableEvent({ kind: 'edit', tableId, originatorId })
+  )
+}
+
 /**
  * Signal collaborators that a user changed the table structure so they refetch the
  * definition + rows live. Fire-and-forget for the same reason as

apps/sim/tools/table/query_rows.ts

@@ -94,6 +94,7 @@ export const tableQueryRowsTool: ToolConfig<TableRowQueryParams, TableQueryRespo
         totalCount: data.totalCount,
         limit: data.limit,
         offset: data.offset,
+        nextCursor: data.nextCursor ?? null,
       },
     }
   },
@@ -105,5 +106,11 @@ export const tableQueryRowsTool: ToolConfig<TableRowQueryParams, TableQueryRespo
     totalCount: { type: 'number', description: 'Total rows matching filter' },
     limit: { type: 'number', description: 'Limit used in query' },
     offset: { type: 'number', description: 'Offset used in query' },
+    nextCursor: {
+      type: 'string',
+      nullable: true,
+      description:
+        'Non-null when more rows match past this page. A page can end early at the byte budget, so this — not a short rowCount — is what says whether more remain. To page, advance offset by rowCount and stop when this is null.',
+    },
   },
 }

apps/sim/tools/table/types.ts

@@ -106,6 +106,12 @@ export interface TableQueryResponse extends ToolResponse {
     totalCount: number
     limit: number
     offset: number
+    /**
+     * Non-null when more rows match past this page — the only reliable end-of-data signal. A page
+     * can end early at the response byte budget, so `rowCount < limit` does NOT mean the last page,
+     * and advancing an offset by `limit` rather than by `rowCount` skips whatever the cut left out.
+     */
+    nextCursor: string | null
   }
 }