mirror of
https://github.com/simstudioai/sim.git
synced 2026-09-24 15:45:35 +08:00
fix(tables): cap row pages and use native cursors (#6582)
This commit is contained in:
@@ -710,7 +710,7 @@
|
||||
"get": {
|
||||
"operationId": "listTableRows",
|
||||
"summary": "List Rows",
|
||||
"description": "List a plain cursor page in default row order. Use the query endpoint for predicate filtering and sorting.",
|
||||
"description": "List a plain cursor page in default row order. Pages are capped at 5MB by default and may contain fewer rows than the requested limit; continue until nextCursor is null. Use the query endpoint for predicate filtering and sorting.",
|
||||
"tags": ["Tables"],
|
||||
"parameters": [
|
||||
{
|
||||
@@ -1383,7 +1383,7 @@
|
||||
"post": {
|
||||
"operationId": "queryTableRows",
|
||||
"summary": "Query Rows",
|
||||
"description": "Query rows with a typed predicate, ordered sort specification, and opaque cursor pagination.",
|
||||
"description": "Query rows with a typed predicate, ordered sort specification, and opaque cursor pagination. Bounded pages are capped at 5MB by default and may contain fewer rows than the requested limit; continue until nextCursor is null.",
|
||||
"tags": ["Tables"],
|
||||
"parameters": [
|
||||
{
|
||||
|
||||
@@ -94,7 +94,7 @@ describe('/api/v2/tables/[tableId]/rows', () => {
|
||||
mocks.preauthRate.mockResolvedValue(RATE)
|
||||
mocks.operationRate.mockResolvedValue(RATE)
|
||||
mocks.gate.mockResolvedValue(null)
|
||||
mocks.listRows.mockResolvedValue({ table: TABLE, rows: [ROW], nextOffset: null })
|
||||
mocks.listRows.mockResolvedValue({ table: TABLE, rows: [ROW], nextCursor: null })
|
||||
mocks.createRows.mockResolvedValue({ kind: 'single', table: TABLE, row: ROW })
|
||||
mocks.updateRows.mockResolvedValue({
|
||||
table: TABLE,
|
||||
@@ -111,31 +111,13 @@ describe('/api/v2/tables/[tableId]/rows', () => {
|
||||
})
|
||||
})
|
||||
|
||||
/**
|
||||
* Coercing an undecodable cursor to offset 0 re-served page one while the
|
||||
* client believed it was paging forward, which loops a paging client forever.
|
||||
* Every sibling v2 cursor list rejects instead, so this one does too.
|
||||
*/
|
||||
it.each([
|
||||
['undecodable base64-JSON', 'malformed'],
|
||||
['a payload with no offset', Buffer.from(JSON.stringify({ o: 5 })).toString('base64')],
|
||||
['a non-integer offset', Buffer.from(JSON.stringify({ offset: 1.5 })).toString('base64')],
|
||||
['a negative offset', Buffer.from(JSON.stringify({ offset: -1 })).toString('base64')],
|
||||
])('rejects a GET cursor with %s instead of restarting pagination', async (_label, cursor) => {
|
||||
const req = request(
|
||||
'GET',
|
||||
undefined,
|
||||
`?workspaceId=${WORKSPACE_ID}&limit=25&cursor=${encodeURIComponent(cursor)}`
|
||||
)
|
||||
const response = await GET(req, CONTEXT)
|
||||
|
||||
expect(response.status).toBe(400)
|
||||
expect((await response.json()).error).toMatchObject({ message: 'Invalid cursor' })
|
||||
expect(mocks.listRows).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('resumes at the encoded offset for a well-formed cursor', async () => {
|
||||
const cursor = Buffer.from(JSON.stringify({ offset: 50 })).toString('base64')
|
||||
it('passes the opaque native row cursor through the route unchanged', async () => {
|
||||
const cursor = 'native-row-cursor'
|
||||
mocks.listRows.mockResolvedValue({
|
||||
table: TABLE,
|
||||
rows: [ROW],
|
||||
nextCursor: 'next-native-cursor',
|
||||
})
|
||||
const req = request(
|
||||
'GET',
|
||||
undefined,
|
||||
@@ -150,10 +132,11 @@ describe('/api/v2/tables/[tableId]/rows', () => {
|
||||
tableId: 'table-1',
|
||||
assertedWorkspaceId: WORKSPACE_ID,
|
||||
limit: 25,
|
||||
offset: 50,
|
||||
cursor,
|
||||
},
|
||||
request: req,
|
||||
})
|
||||
expect((await response.json()).nextCursor).toBe('next-native-cursor')
|
||||
})
|
||||
|
||||
it('delegates single and batch creation through one semantic use case', async () => {
|
||||
|
||||
@@ -14,7 +14,6 @@ import {
|
||||
updateTableRows,
|
||||
} from '@/lib/table/application/rows'
|
||||
import { namedRowMapper } from '@/lib/table/cell-format'
|
||||
import { decodeOffsetCursor, encodeCursor } from '@/app/api/v2/lib/response'
|
||||
import { toApiRow } from '@/app/api/v2/tables/utils'
|
||||
|
||||
export const dynamic = 'force-dynamic'
|
||||
@@ -30,14 +29,14 @@ export const GET = defineV2JsonRoute({
|
||||
tableId: params.tableId,
|
||||
assertedWorkspaceId: query.workspaceId,
|
||||
limit: query.limit,
|
||||
offset: decodeOffsetCursor(query.cursor),
|
||||
cursor: query.cursor,
|
||||
}),
|
||||
useCase: listTableRows,
|
||||
present: ({ table, rows, nextOffset }) => {
|
||||
present: ({ table, rows, nextCursor }) => {
|
||||
const toNamedRow = namedRowMapper(table.schema.columns)
|
||||
return {
|
||||
data: rows.map((row) => toApiRow(row, toNamedRow)),
|
||||
nextCursor: nextOffset === null ? null : encodeCursor({ offset: nextOffset }),
|
||||
nextCursor,
|
||||
}
|
||||
},
|
||||
})
|
||||
|
||||
@@ -358,7 +358,7 @@ const routes = [
|
||||
operationId: 'listTableRows',
|
||||
summary: 'List Rows',
|
||||
description:
|
||||
'List a plain cursor page in default row order. Use the query endpoint for predicate filtering and sorting.',
|
||||
'List a plain cursor page in default row order. Pages are capped at 5MB by default and may contain fewer rows than the requested limit; continue until nextCursor is null. Use the query endpoint for predicate filtering and sorting.',
|
||||
errors: RESOURCE_ERRORS,
|
||||
success: { description: 'A page of table rows.' },
|
||||
}),
|
||||
@@ -619,7 +619,7 @@ const routes = [
|
||||
operationId: 'queryTableRows',
|
||||
summary: 'Query Rows',
|
||||
description:
|
||||
'Query rows with a typed predicate, ordered sort specification, and opaque cursor pagination.',
|
||||
'Query rows with a typed predicate, ordered sort specification, and opaque cursor pagination. Bounded pages are capped at 5MB by default and may contain fewer rows than the requested limit; continue until nextCursor is null.',
|
||||
errors: RESOURCE_ERRORS,
|
||||
success: { description: 'A page of matching table rows.' },
|
||||
}),
|
||||
|
||||
@@ -644,9 +644,9 @@ export const v2DeleteTableColumnContract = defineRouteContract({
|
||||
/**
|
||||
* Row list query: a plain cursor page over the default row order. Filtering and
|
||||
* sorting are NOT part of this surface — rich reads go through the dedicated
|
||||
* `POST /query` endpoint's predicate grammar. The opaque cursor encodes the
|
||||
* underlying offset today; it can move to a keyset implementation later without
|
||||
* an interface change. Total row count is available as `rowCount` on the table.
|
||||
* `POST /query` endpoint's predicate grammar. The opaque cursor uses the
|
||||
* `(order_key, id)` keyset when possible and handles legacy rows without an order
|
||||
* key internally. Total row count is available as `rowCount` on the table.
|
||||
*/
|
||||
export const v2TableRowsQuerySchema = tableRowsQueryBaseSchema
|
||||
.pick({ workspaceId: true, limit: true })
|
||||
|
||||
@@ -138,7 +138,7 @@ export const env = createEnv({
|
||||
ENTERPRISE_TABLES_LIMIT: z.number().optional(), // Max user tables per workspace on enterprise tier (default: 10000)
|
||||
ENTERPRISE_TABLE_ROWS_LIMIT: z.number().optional(), // Max rows per table on enterprise tier (default: 1000000)
|
||||
TABLE_MAX_ROW_SIZE_BYTES: z.number().optional(), // Max serialized size in bytes of a single user-table row (default: 409600)
|
||||
TABLE_MAX_PAGE_BYTES: z.number().optional(), // Byte budget per row-page read; pages cut early past it (unset = disabled)
|
||||
TABLE_MAX_PAGE_BYTES: z.number().optional(), // Byte budget per row-page read; pages cut early past it (default: 5242880)
|
||||
TABLE_DISPATCH_CONCURRENCY_FREE: z.number().optional(), // Rows one table run executes in parallel on free tier (default: 20)
|
||||
TABLE_DISPATCH_CONCURRENCY_PAID: z.number().optional(), // Rows one table run executes in parallel on paid tiers (default: 50)
|
||||
|
||||
|
||||
@@ -204,9 +204,7 @@ describe('queryRows byte budget', () => {
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks()
|
||||
resetDbChainMock()
|
||||
// The bounded-page byte cut is opt-in; pin it on rather than inheriting
|
||||
// whatever the developer's local `.env` happens to set.
|
||||
setEnv({ TABLE_MAX_PAGE_BYTES: TABLE_LIMITS.MAX_QUERY_RESULT_BYTES })
|
||||
setEnv({ TABLE_MAX_PAGE_BYTES: undefined })
|
||||
})
|
||||
|
||||
const row = (i: number, blobBytes: number) => ({
|
||||
@@ -265,12 +263,9 @@ describe('queryRows byte budget', () => {
|
||||
})
|
||||
})
|
||||
|
||||
it('does NOT byte-cut a bounded page when TABLE_MAX_PAGE_BYTES is unset', async () => {
|
||||
// Default-off: a short page is only safe for a client that terminates on
|
||||
// `nextCursor === null`. A pre-existing v1 pager stopping at
|
||||
// `rows.length < limit` would read the cut as end-of-data and truncate.
|
||||
setEnv({ TABLE_MAX_PAGE_BYTES: undefined })
|
||||
const perRow = Math.floor(TABLE_LIMITS.MAX_QUERY_RESULT_BYTES * 0.6)
|
||||
it('honors a smaller bounded-page byte override', async () => {
|
||||
setEnv({ TABLE_MAX_PAGE_BYTES: 3 * 1024 * 1024 })
|
||||
const perRow = 2 * 1024 * 1024
|
||||
dbChainMockFns.limit.mockResolvedValueOnce([])
|
||||
dbChainMockFns.limit.mockResolvedValueOnce([row(1, perRow), row(2, perRow)])
|
||||
|
||||
@@ -280,8 +275,8 @@ describe('queryRows byte budget', () => {
|
||||
'req-1'
|
||||
)
|
||||
|
||||
expect(result.rows).toHaveLength(2)
|
||||
expect(result.nextCursor).toBeNull()
|
||||
expect(result.rows).toHaveLength(1)
|
||||
expect(result.nextCursor).not.toBeNull()
|
||||
})
|
||||
|
||||
it('still fails fast on an UNBOUNDED query with TABLE_MAX_PAGE_BYTES unset', async () => {
|
||||
|
||||
@@ -142,6 +142,7 @@ vi.mock('@/lib/table/events', () => ({
|
||||
import {
|
||||
createTableRows,
|
||||
deleteTableRows,
|
||||
listTableRows,
|
||||
ProjectedWireRowsValidationError,
|
||||
queryTableRows,
|
||||
replaceProjectedWireRows,
|
||||
@@ -152,6 +153,7 @@ import {
|
||||
updateTableRows,
|
||||
upsertTableRow,
|
||||
} from '@/lib/table/application/rows'
|
||||
import { encodeCursor } from '@/lib/table/rows/cursor'
|
||||
|
||||
const TABLE: TableDefinition = {
|
||||
id: 'table-1',
|
||||
@@ -584,6 +586,49 @@ describe('row query and upsert application semantics', () => {
|
||||
expect(mockLoadSecretProvenance).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('rejects a malformed list cursor before querying storage', async () => {
|
||||
await expect(
|
||||
listTableRows.execute({
|
||||
principal: PRINCIPAL,
|
||||
input: { tableId: TABLE.id, limit: 25, cursor: 'malformed' },
|
||||
})
|
||||
).rejects.toMatchObject({ details: { code: 'INVALID_CURSOR' } })
|
||||
|
||||
expect(mockQueryRows).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('passes the native row cursor through a short list page', async () => {
|
||||
const cursor = encodeCursor({
|
||||
lastRow: { id: 'row-50', orderKey: 'a50' },
|
||||
keysetValid: true,
|
||||
nextOffset: 50,
|
||||
})
|
||||
mockQueryRows.mockResolvedValueOnce({
|
||||
rows: [{ id: 'row-51', data: {} }],
|
||||
rowCount: 1,
|
||||
totalCount: null,
|
||||
nextCursor: 'native-next-cursor',
|
||||
})
|
||||
|
||||
const result = await listTableRows.execute({
|
||||
principal: PRINCIPAL,
|
||||
input: { tableId: TABLE.id, limit: 25, cursor },
|
||||
})
|
||||
|
||||
expect(mockQueryRows).toHaveBeenCalledWith(
|
||||
TABLE,
|
||||
{
|
||||
limit: 25,
|
||||
after: { id: 'row-50', orderKey: 'a50' },
|
||||
offset: undefined,
|
||||
includeTotal: false,
|
||||
withExecutions: false,
|
||||
},
|
||||
expect.any(String)
|
||||
)
|
||||
expect(result.nextCursor).toBe('native-next-cursor')
|
||||
})
|
||||
|
||||
it('loads requested persisted provenance inside the authorized application query', async () => {
|
||||
const row = {
|
||||
id: 'row-1',
|
||||
|
||||
@@ -192,12 +192,12 @@ function rethrowQueryValidation(error: unknown): never {
|
||||
|
||||
export interface ListTableRowsInput extends TableScopedInput {
|
||||
limit: number
|
||||
offset: number
|
||||
cursor?: string
|
||||
}
|
||||
|
||||
export interface ListTableRowsResult extends TableResult {
|
||||
rows: TableRow[]
|
||||
nextOffset: number | null
|
||||
nextCursor: string | null
|
||||
}
|
||||
|
||||
export const listTableRows = defineAuthorizedTableUseCase({
|
||||
@@ -205,25 +205,24 @@ export const listTableRows = defineAuthorizedTableUseCase({
|
||||
resolveContext: ({ input }: { input: ListTableRowsInput }) => resolveActiveTableContext(input),
|
||||
async execute({ input, context }): Promise<ListTableRowsResult> {
|
||||
requireIntegerInRange(input.limit, 1, TABLE_LIMITS.MAX_QUERY_LIMIT, 'Limit')
|
||||
if (!Number.isSafeInteger(input.offset) || input.offset < 0) {
|
||||
throw new TableRowsValidationError('Offset must be 0 or greater')
|
||||
}
|
||||
try {
|
||||
const cursor = input.cursor ? decodeCursor(input.cursor) : undefined
|
||||
if (cursor) assertCursorSortBinding(cursor, undefined)
|
||||
const result = await queryRows(
|
||||
context.table,
|
||||
{
|
||||
limit: input.limit,
|
||||
offset: input.offset,
|
||||
includeTotal: true,
|
||||
after: cursor?.after,
|
||||
offset: cursor?.offset,
|
||||
includeTotal: false,
|
||||
withExecutions: false,
|
||||
},
|
||||
requestId(input)
|
||||
)
|
||||
const total = result.totalCount ?? 0
|
||||
return {
|
||||
table: context.table,
|
||||
rows: result.rows,
|
||||
nextOffset: input.offset + result.rowCount < total ? input.offset + input.limit : null,
|
||||
nextCursor: result.nextCursor,
|
||||
}
|
||||
} catch (error) {
|
||||
rethrowQueryValidation(error)
|
||||
|
||||
@@ -37,7 +37,11 @@ declare module '@/lib/table/constants?constants-test' {
|
||||
export * from '@/lib/table/constants'
|
||||
}
|
||||
|
||||
import { getBillingDisabledTableLimits } from '@/lib/table/constants?constants-test'
|
||||
import {
|
||||
getBillingDisabledTableLimits,
|
||||
getMaxPageBytes,
|
||||
TABLE_LIMITS,
|
||||
} from '@/lib/table/constants?constants-test'
|
||||
|
||||
describe('getBillingDisabledTableLimits', () => {
|
||||
beforeEach(() => {
|
||||
@@ -66,3 +70,19 @@ describe('getBillingDisabledTableLimits', () => {
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
describe('getMaxPageBytes', () => {
|
||||
beforeEach(() => {
|
||||
for (const key of Object.keys(mockEnv)) delete mockEnv[key]
|
||||
})
|
||||
|
||||
it('defaults bounded pages to the 5MB query-result budget', () => {
|
||||
expect(getMaxPageBytes()).toBe(TABLE_LIMITS.MAX_QUERY_RESULT_BYTES)
|
||||
})
|
||||
|
||||
it('allows a positive integer environment override', () => {
|
||||
mockEnv.TABLE_MAX_PAGE_BYTES = String(2 * 1024 * 1024)
|
||||
|
||||
expect(getMaxPageBytes()).toBe(2 * 1024 * 1024)
|
||||
})
|
||||
})
|
||||
|
||||
@@ -74,17 +74,19 @@ export const DEFAULT_TABLE_PLAN_LIMITS = {
|
||||
|
||||
/**
|
||||
* Byte budget at which a **bounded** page (one with an explicit `limit`) is cut
|
||||
* short, or `null` when disabled — the default. Opt in with `TABLE_MAX_PAGE_BYTES`.
|
||||
* short. Defaults to the 5MB query-result budget and can be overridden with
|
||||
* `TABLE_MAX_PAGE_BYTES`. Callers must terminate on `nextCursor === null`, not
|
||||
* page fullness, because a byte-limited page may contain fewer rows than requested.
|
||||
*
|
||||
* Off by default because a short page is only safe for a client that terminates
|
||||
* on `nextCursor === null`; a pre-existing v1 pager terminating on
|
||||
* `rows.length < limit` would read the cut as end-of-data and silently truncate.
|
||||
* Unbounded queries (no `limit`) are unaffected — they always fail fast at
|
||||
* `TABLE_LIMITS.MAX_QUERY_RESULT_BYTES` rather than return a partial result.
|
||||
* Unbounded queries (no `limit`) are unaffected by the override — they always
|
||||
* fail fast at `TABLE_LIMITS.MAX_QUERY_RESULT_BYTES` rather than return a partial
|
||||
* result.
|
||||
*/
|
||||
export function getMaxPageBytes(): number | null {
|
||||
const value = envNumber(env.TABLE_MAX_PAGE_BYTES, 0, { min: 0, integer: true })
|
||||
return value > 0 ? value : null
|
||||
export function getMaxPageBytes(): number {
|
||||
return envNumber(env.TABLE_MAX_PAGE_BYTES, TABLE_LIMITS.MAX_QUERY_RESULT_BYTES, {
|
||||
min: 1,
|
||||
integer: true,
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -1202,7 +1202,7 @@ export async function queryRows(
|
||||
startOffset: offset,
|
||||
limit,
|
||||
budgetBytes: TABLE_LIMITS.MAX_QUERY_RESULT_BYTES,
|
||||
pageCutBytes: getMaxPageBytes() ?? undefined,
|
||||
pageCutBytes: getMaxPageBytes(),
|
||||
})
|
||||
|
||||
const [fetched, totalCount] = await Promise.all([drainPromise, countPromise])
|
||||
@@ -1271,11 +1271,8 @@ interface BoundedFetchParams {
|
||||
limit?: number
|
||||
/** Drain ceiling: sizes batches, and the fail-fast bound for an unbounded query. */
|
||||
budgetBytes: number
|
||||
/**
|
||||
* Opt-in byte cut for a **bounded** page (`TABLE_MAX_PAGE_BYTES`); `undefined`
|
||||
* disables it, so a bounded page always returns its full `limit`.
|
||||
*/
|
||||
pageCutBytes?: number
|
||||
/** Byte cut for a **bounded** page; defaults to 5MB and is environment-overridable. */
|
||||
pageCutBytes: number
|
||||
}
|
||||
|
||||
interface BoundedFetchResult {
|
||||
@@ -1300,9 +1297,9 @@ const MAX_QUERY_BATCHES = 1000
|
||||
*
|
||||
* Byte ceiling: an **unbounded** query (no `limit`) always fails fast at
|
||||
* `budgetBytes` — returning part of a result that promised everything would be
|
||||
* silent truncation. A **bounded** page cuts short only when `pageCutBytes` is
|
||||
* set (`TABLE_MAX_PAGE_BYTES`), because a short page is only safe for clients
|
||||
* that terminate on `nextCursor === null` rather than on page fullness.
|
||||
* silent truncation. A **bounded** page cuts short at `pageCutBytes` and returns
|
||||
* a cursor, so clients must terminate on `nextCursor === null` rather than on
|
||||
* page fullness.
|
||||
*
|
||||
* Advance strategy: when `keysetValid`, the loop re-anchors on each consumed
|
||||
* keyed row and seeks `(order_key, id) > (anchor)` — delete-tolerant and an
|
||||
@@ -1319,7 +1316,7 @@ async function fetchRowsBounded(params: BoundedFetchParams): Promise<BoundedFetc
|
||||
const firstBatchCap = Math.max(1, Math.floor((4 * budgetBytes) / TABLE_LIMITS.MAX_ROW_SIZE_BYTES))
|
||||
|
||||
// The byte ceiling that ends the drain: an unbounded query fails fast at the
|
||||
// budget; a bounded page cuts only when the operator opted in.
|
||||
// budget; a bounded page cuts at the configured page budget.
|
||||
const cutBytes = limit === undefined ? budgetBytes : pageCutBytes
|
||||
|
||||
const rows: Array<typeof userTableRows.$inferSelect> = []
|
||||
|
||||
Reference in New Issue
Block a user