Skip to content

Commit 8e18553

Browse files
committed
perf(tables): stop a table write refetching every loaded page in the tab that made it
1 parent 4181912 commit 8e18553

14 files changed

Lines changed: 316 additions & 18 deletions

File tree

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -125,7 +125,7 @@ After the run, the table holds the enriched rows. The next run queries them agai
125125

126126
**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.
127127

128-
**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.
128+
**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.
129129

130130
## Inspecting reads and writes
131131

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

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ import { userTableRows } from '@sim/db/schema'
33
import { createLogger } from '@sim/logger'
44
import { and, eq } from 'drizzle-orm'
55
import { type NextRequest, NextResponse } from 'next/server'
6+
import { readClientId } from '@/lib/api/client-id'
67
import {
78
deleteTableRowContract,
89
getTableQuerySchema,
@@ -14,7 +15,7 @@ import { generateRequestId } from '@/lib/core/utils/request'
1415
import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
1516
import type { RowData, TableSchema } from '@/lib/table'
1617
import { updateRow } from '@/lib/table'
17-
import { signalTableRowsChanged } from '@/lib/table/events'
18+
import { signalTableRowsChangedByActor } from '@/lib/table/events'
1819
import { performDeleteTableRow } from '@/lib/table/orchestration'
1920
import {
2021
createTableRowsResponse,
@@ -172,7 +173,7 @@ export const PATCH = withRouteHandler(async (request: NextRequest, context: RowR
172173
)
173174

174175
// Live-collab: tell open viewers the change landed so they refetch.
175-
signalTableRowsChanged(tableId)
176+
signalTableRowsChangedByActor(tableId, readClientId(request))
176177
// Only `null` when a `cancellationGuard` is supplied and the SQL guard
177178
// rejects the write — this route doesn't pass one, so reaching null is a bug.
178179
if (!updatedRow) throw new Error('updateRow returned null without a cancellationGuard')
@@ -251,7 +252,7 @@ export const DELETE = withRouteHandler(async (request: NextRequest, context: Row
251252
}
252253

253254
// Live-collab: tell open viewers the change landed so they refetch.
254-
signalTableRowsChanged(tableId)
255+
signalTableRowsChangedByActor(tableId, readClientId(request))
255256

256257
return NextResponse.json({
257258
success: true,

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

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
import { createLogger } from '@sim/logger'
22
import { type NextRequest, NextResponse } from 'next/server'
3+
import { readClientId } from '@/lib/api/client-id'
34
import {
45
type BatchInsertTableRowsBodyInput,
56
batchUpdateTableRowsBodySchema,
@@ -26,7 +27,7 @@ import {
2627
validateRowSize,
2728
} from '@/lib/table'
2829
import { TableQueryValidationError } from '@/lib/table/errors'
29-
import { signalTableRowsChanged } from '@/lib/table/events'
30+
import { signalTableRowsChanged, signalTableRowsChangedByActor } from '@/lib/table/events'
3031
import { isTablePredicate, predicateToFilter } from '@/lib/table/query-builder/converters'
3132
import {
3233
validatePredicateShape,
@@ -254,7 +255,9 @@ export const POST = withRouteHandler(
254255
table,
255256
requestId
256257
)
257-
signalTableRowsChanged(tableId)
258+
// Attributed unlike the batch path above: the acting tab's insert deliberately avoids
259+
// invalidating the rows root to prevent flicker, which an unattributed echo would undo.
260+
signalTableRowsChangedByActor(tableId, readClientId(request))
258261

259262
const responseBody = {
260263
success: true,

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

Lines changed: 26 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@ import { toast } from '@sim/emcn'
55
import { createLogger } from '@sim/logger'
66
import { backoffWithJitter } from '@sim/utils/retry'
77
import { useQueryClient } from '@tanstack/react-query'
8+
import { getClientFingerprint } from '@/lib/api/client-id'
89
import type { ActiveDispatch } from '@/lib/api/contracts/tables'
910
import type {
1011
RowData,
@@ -245,6 +246,30 @@ export function useTableEventStream({
245246
}, ROWS_INVALIDATE_DEBOUNCE_MS)
246247
}
247248

249+
/**
250+
* This tab's fingerprint as it appears on a broadcast it caused. Resolved once, asynchronously;
251+
* until it lands `applyEdit` simply takes the refetch path, which is the pre-existing behavior.
252+
*/
253+
let ownFingerprint: string | undefined
254+
void getClientFingerprint().then((fingerprint) => {
255+
ownFingerprint = fingerprint
256+
})
257+
258+
/**
259+
* A manual row edit landed. Refetch the rows so the winning last-write value shows live —
260+
* unless this tab is the one that made it.
261+
*
262+
* The signal names its originator only for writes whose mutation hook already applies the
263+
* server's answer to every cached rows query, active or not (single-row create, update,
264+
* delete). For those the refetch is pure duplication: on a scrolled table it re-fetches every
265+
* loaded page, and on delete it races the refetch the hook itself issued. Other tabs see
266+
* someone else's fingerprint and refetch normally; an unattributed edit refetches everywhere.
267+
*/
268+
const applyEdit = (event: Extract<TableEvent, { kind: 'edit' }>): void => {
269+
if (event.originatorId && event.originatorId === ownFingerprint) return
270+
scheduleRowsInvalidate()
271+
}
272+
248273
const applyCell = (event: Extract<TableEvent, { kind: 'cell' }>): void => {
249274
void snapshotAndMutateRows(queryClient, tableId, (row) => applyCellEventToRow(row, event), {
250275
cancelInFlight: false,
@@ -445,9 +470,7 @@ export function useTableEventStream({
445470
else if (entry.event?.kind === 'dispatch') applyDispatch(entry.event)
446471
else if (entry.event?.kind === 'job') applyJob(entry.event)
447472
else if (entry.event?.kind === 'usageLimitReached') applyUsageLimit(entry.event)
448-
// A collaborator's manual edit: refetch rows (debounced) so the winning
449-
// last-write value shows live, in this client's own wire format.
450-
else if (entry.event?.kind === 'edit') scheduleRowsInvalidate()
473+
else if (entry.event?.kind === 'edit') applyEdit(entry.event)
451474
// A collaborator changed the table structure: mirror the local
452475
// invalidateTableSchema set — the definition (exact, so rows stay on the
453476
// debounce), the run-state + enrichment sibling queries under detail (a group

apps/sim/hooks/queries/tables.ts

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1071,6 +1071,18 @@ export function useUpdateTableRow({ workspaceId, tableId }: RowMutationContext)
10711071
updatedAt: serverRow.updatedAt,
10721072
}
10731073
})
1074+
1075+
// `patchCachedRows` rewrites values in place, which is the whole answer for the default
1076+
// view. It cannot be for a filtered or column-sorted one: editing a cell can move a row in
1077+
// or out of the filter and change its sort position and `totalCount`, none of which a
1078+
// per-row patch can express. Those views are refetched instead — the same split
1079+
// `useCreateTableRow` makes, and previously supplied by the broadcast this write no longer
1080+
// makes the acting tab honor.
1081+
queryClient.invalidateQueries({
1082+
queryKey: tableKeys.rowsRoot(tableId),
1083+
exact: false,
1084+
predicate: (query) => !isDefaultOrderRowsQuery(query.queryKey),
1085+
})
10741086
},
10751087
onError: (error, _vars, context) => {
10761088
if (context?.previousQueries) {

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

Lines changed: 54 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,54 @@
1+
/**
2+
* @vitest-environment node
3+
*/
4+
import { describe, expect, it } from 'vitest'
5+
import { CLIENT_ID_HEADER, fingerprintClientId, readClientId } from '@/lib/api/client-id'
6+
7+
describe('readClientId', () => {
8+
it('reads the sending tab id off the request', () => {
9+
const request = new Request('https://sim.ai/api/table/t1/rows', {
10+
headers: { [CLIENT_ID_HEADER]: 'tab-abc' },
11+
})
12+
expect(readClientId(request)).toBe('tab-abc')
13+
})
14+
15+
/** Absent must read as "unattributed" — the signal then makes every client refetch, as before. */
16+
it('is undefined when the caller sent no id', () => {
17+
const request = new Request('https://sim.ai/api/table/t1/rows')
18+
expect(readClientId(request)).toBeUndefined()
19+
})
20+
21+
/**
22+
* The value is caller-controlled and is broadcast to every subscriber of the table, so an
23+
* over-long one is dropped rather than fanned out.
24+
*/
25+
it('drops an over-long id instead of broadcasting it', () => {
26+
const request = new Request('https://sim.ai/api/table/t1/rows', {
27+
headers: { [CLIENT_ID_HEADER]: 'x'.repeat(65) },
28+
})
29+
expect(readClientId(request)).toBeUndefined()
30+
})
31+
})
32+
33+
/**
34+
* Every subscriber of a table sees every broadcast, so what gets published must not be replayable.
35+
* If the raw id travelled, a collaborator could read it off the stream, send it as their own
36+
* header, and have their write attributed to someone else's tab — which would then suppress a
37+
* refetch it genuinely needed and sit on stale rows.
38+
*/
39+
describe('fingerprintClientId', () => {
40+
it('is stable for the same id, so a tab recognises its own broadcast', async () => {
41+
expect(await fingerprintClientId('tab-abc')).toBe(await fingerprintClientId('tab-abc'))
42+
})
43+
44+
it('differs between tabs, so one tab never suppresses on another tab’s write', async () => {
45+
expect(await fingerprintClientId('tab-abc')).not.toBe(await fingerprintClientId('tab-xyz'))
46+
})
47+
48+
it('does not reveal the id it was derived from', async () => {
49+
const fingerprint = await fingerprintClientId('tab-abc')
50+
expect(fingerprint).not.toContain('tab-abc')
51+
// SHA-256 hex — knowing this cannot produce the header value that would match it.
52+
expect(fingerprint).toMatch(/^[0-9a-f]{64}$/)
53+
})
54+
})

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

Lines changed: 77 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,77 @@
1+
import { generateShortId } from '@sim/utils/id'
2+
3+
/**
4+
* Header naming the browser tab that sent a request.
5+
*
6+
* Shared by the client that sets it and the route handlers that read it. An opaque correlation
7+
* token, never an authorization input.
8+
*/
9+
export const CLIENT_ID_HEADER = 'x-sim-client-id'
10+
11+
/**
12+
* Generated ids are {@link generateShortId} length; the ceiling is slack for that, not a format.
13+
* Bounded because the value is caller-controlled and gets fanned out to every subscriber of a
14+
* table — uncapped, one request could inflate every broadcast payload it triggers.
15+
*/
16+
const MAX_CLIENT_ID_LENGTH = 64
17+
18+
let cachedClientId: string | undefined
19+
20+
/**
21+
* An id for this browser tab, generated once per page load and not stable across reloads.
22+
*
23+
* Deliberately per-TAB rather than per-user or per-session: its only consumer compares it against
24+
* the originator stamped on a broadcast, so two tabs belonging to the same user must not share one.
25+
* A shared id would make the second tab ignore the first tab's edits and silently go stale.
26+
*
27+
* Returns `undefined` on the server, where there is no tab to identify.
28+
*/
29+
export function getClientId(): string | undefined {
30+
if (typeof window === 'undefined') return undefined
31+
cachedClientId ??= generateShortId()
32+
return cachedClientId
33+
}
34+
35+
/**
36+
* The sending tab's id, as seen by a route handler. Absent for server-to-server callers, for any
37+
* client that did not send one, and for an over-long value — all read as "unattributed", never as
38+
* "not the actor".
39+
*
40+
* Untrusted, and never safe to broadcast as-is: see {@link fingerprintClientId}.
41+
*/
42+
export function readClientId(request: Request): string | undefined {
43+
const raw = request.headers.get(CLIENT_ID_HEADER)
44+
return raw && raw.length <= MAX_CLIENT_ID_LENGTH ? raw : undefined
45+
}
46+
47+
/**
48+
* One-way digest of a tab id, for naming the originator of a broadcast.
49+
*
50+
* The raw id must never travel on a broadcast. Every subscriber of a table sees every event, so a
51+
* raw id would be observable by any collaborator, who could then replay it as their own
52+
* `x-sim-client-id` — their write would be attributed to your tab, your tab would suppress its
53+
* refetch, and it would sit on stale rows. Publishing the digest instead means matching it
54+
* requires already knowing the id, which only the tab that generated it does.
55+
*
56+
* Web Crypto rather than `node:crypto` so one implementation serves both sides — the server
57+
* stamping the event and the browser recognising its own — with no chance of the two disagreeing.
58+
*/
59+
export async function fingerprintClientId(clientId: string): Promise<string> {
60+
const digest = await crypto.subtle.digest('SHA-256', new TextEncoder().encode(clientId))
61+
return Array.from(new Uint8Array(digest))
62+
.map((byte) => byte.toString(16).padStart(2, '0'))
63+
.join('')
64+
}
65+
66+
let cachedFingerprint: string | undefined
67+
68+
/**
69+
* This tab's fingerprint, as it appears on a broadcast it caused. `undefined` on the server, and
70+
* until the first digest resolves — callers must treat that as "not me" and take the normal path.
71+
*/
72+
export async function getClientFingerprint(): Promise<string | undefined> {
73+
const clientId = getClientId()
74+
if (!clientId) return undefined
75+
cachedFingerprint ??= await fingerprintClientId(clientId)
76+
return cachedFingerprint
77+
}

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

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@
44
import { afterEach, describe, expect, it, vi } from 'vitest'
55
import { z } from 'zod'
66
import { requestJson } from '@/lib/api/client/request'
7+
import { CLIENT_ID_HEADER } from '@/lib/api/client-id'
78
import { listKnowledgeDocumentsContract } from '@/lib/api/contracts/knowledge'
89
import { defineRouteContract } from '@/lib/api/contracts/types'
910

@@ -87,3 +88,38 @@ describe('requestJson query serialization', () => {
8788
expect(url).toContain('tags=a&tags=b')
8889
})
8990
})
91+
92+
/**
93+
* The tab id rides on every request so a broadcast raised by one can be attributed back to the tab
94+
* that caused it. Asserted here rather than on the reader, because the header being *sent* is the
95+
* half that silently does nothing if it regresses.
96+
*/
97+
describe('requestJson client id header', () => {
98+
const contract = defineRouteContract({
99+
method: 'GET',
100+
path: '/api/test',
101+
response: { mode: 'json', schema: z.object({ ok: z.boolean() }) },
102+
})
103+
104+
function sentHeaders(fetchMock: ReturnType<typeof mockFetchReturning>): Record<string, string> {
105+
return (fetchMock.mock.calls[0][1] as RequestInit).headers as Record<string, string>
106+
}
107+
108+
it('sends the tab id in the browser', async () => {
109+
vi.stubGlobal('window', {})
110+
const fetchMock = mockFetchReturning({ ok: true })
111+
112+
await requestJson(contract, {})
113+
114+
expect(sentHeaders(fetchMock)[CLIENT_ID_HEADER]).toEqual(expect.any(String))
115+
})
116+
117+
it('omits it on the server, where there is no tab to name', async () => {
118+
vi.stubGlobal('window', undefined)
119+
const fetchMock = mockFetchReturning({ ok: true })
120+
121+
await requestJson(contract, {})
122+
123+
expect(sentHeaders(fetchMock)[CLIENT_ID_HEADER]).toBeUndefined()
124+
})
125+
})

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

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
import { ApiClientError } from '@/lib/api/client/errors'
2+
import { CLIENT_ID_HEADER, getClientId } from '@/lib/api/client-id'
23
import type {
34
AnyApiRouteContract,
45
ApiSchema,
@@ -104,6 +105,10 @@ function buildHeaders(headers: unknown, hasBody: boolean): Record<string, string
104105
output['Content-Type'] = 'application/json'
105106
}
106107

108+
/** Set here rather than per call site so every request carries it without a decision to get wrong. */
109+
const clientId = getClientId()
110+
if (clientId) output[CLIENT_ID_HEADER] = clientId
111+
107112
if (headers && typeof headers === 'object') {
108113
for (const [key, value] of Object.entries(headers as Record<string, unknown>)) {
109114
if (typeof value === 'string') output[key] = value
Lines changed: 48 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,48 @@
1+
/**
2+
* @vitest-environment node
3+
*/
4+
import { readdir, readFile } from 'node:fs/promises'
5+
import { join } from 'node:path'
6+
import { describe, expect, it } from 'vitest'
7+
8+
/**
9+
* `signalTableRowsChangedByActor` lets the acting tab skip its own refetch, which is only sound
10+
* where that tab's mutation hook already applies the server's answer to every cached rows query.
11+
* That invariant lives in `hooks/queries/tables.ts` — nothing in the type system ties it to the
12+
* call site, so a well-meaning fourth call would silently strand that client on stale rows.
13+
*
14+
* This pins the allowlist. If you are here because it failed: adding a call means proving the
15+
* calling route's client hook reconciles locally, then adding it below. Removing one is always safe.
16+
*/
17+
const ATTRIBUTED_CALL_SITES = [
18+
'app/api/table/[tableId]/rows/route.ts',
19+
'app/api/table/[tableId]/rows/[rowId]/route.ts',
20+
] as const
21+
22+
const APP_ROOT = join(import.meta.dirname, '../..')
23+
/** Declares the function; matching its own definition would say nothing about call sites. */
24+
const DECLARING_MODULE = 'lib/table/events.ts'
25+
26+
async function* walk(dir: string): AsyncGenerator<string> {
27+
for (const entry of await readdir(dir, { withFileTypes: true })) {
28+
if (entry.name === 'node_modules' || entry.name === '.next') continue
29+
const full = join(dir, entry.name)
30+
if (entry.isDirectory()) yield* walk(full)
31+
else if (entry.name.endsWith('.ts') && !entry.name.includes('.test.')) yield full
32+
}
33+
}
34+
35+
describe('signalTableRowsChangedByActor call sites', () => {
36+
it('is called only where the acting tab reconciles the write locally', async () => {
37+
const callers: string[] = []
38+
for await (const file of walk(APP_ROOT)) {
39+
const source = await readFile(file, 'utf8')
40+
if (!source.includes('signalTableRowsChangedByActor(')) continue
41+
const relative = file.slice(APP_ROOT.length + 1)
42+
if (relative === DECLARING_MODULE) continue
43+
callers.push(relative)
44+
}
45+
46+
expect(callers.sort()).toEqual([...ATTRIBUTED_CALL_SITES].sort())
47+
})
48+
})

0 commit comments

Comments
 (0)