mirror of
https://github.com/simstudioai/sim.git
synced 2026-09-24 15:45:35 +08:00
improvement(mothership): make user_table limit cap internal, not model-facing
The model can now pass any limit — no "cannot exceed 1000" rejection. 1000 becomes an internal threshold: query_rows clamps the page to MAX_QUERY_LIMIT (totalCount signals truncation; the model pages with offset), and bulk filter ops above the cap run as background jobs. update_rows_by_filter loads full row data inline, so an explicit limit above the cap escalates to the background worker with a new maxRows budget (the worker stops after maxRows; update has no read mask so the cap is exact). delete only loads ids inline, so an explicit limit (any size) stays inline — only unbounded deletes use the masked background path, which would over-hide a bounded delete. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Fable 5
parent
4ee39374e9
commit
d2c93aef04
@@ -3959,7 +3959,7 @@ export const UserTable: ToolCatalogEntry = {
|
||||
limit: {
|
||||
type: 'number',
|
||||
description:
|
||||
'Maximum rows to return or affect (default 100, max 1000). For update_rows_by_filter / delete_rows_by_filter, omit to act on every match — large match sets run as a background job.',
|
||||
'Maximum rows to return or affect (optional, default 100). Any value is allowed — large operations run in the background automatically. Omit on update_rows_by_filter / delete_rows_by_filter to act on every match.',
|
||||
},
|
||||
mapping: {
|
||||
type: 'object',
|
||||
|
||||
@@ -3687,7 +3687,7 @@ export const TOOL_RUNTIME_SCHEMAS: Record<string, ToolRuntimeSchemaEntry> = {
|
||||
limit: {
|
||||
type: 'number',
|
||||
description:
|
||||
'Maximum rows to return or affect (default 100, max 1000). For update_rows_by_filter / delete_rows_by_filter, omit to act on every match — large match sets run as a background job.',
|
||||
'Maximum rows to return or affect (optional, default 100). Any value is allowed — large operations run in the background automatically. Omit on update_rows_by_filter / delete_rows_by_filter to act on every match.',
|
||||
},
|
||||
mapping: {
|
||||
type: 'object',
|
||||
|
||||
@@ -692,15 +692,15 @@ describe('userTableServerTool.query_rows', () => {
|
||||
})
|
||||
})
|
||||
|
||||
it('rejects limits above MAX_QUERY_LIMIT', async () => {
|
||||
it('clamps an over-large query limit to MAX_QUERY_LIMIT instead of rejecting', async () => {
|
||||
const result = await userTableServerTool.execute(
|
||||
{ operation: 'query_rows', args: { tableId: 'tbl_1', limit: 100000 } },
|
||||
{ userId: 'user-1', workspaceId: 'workspace-1' }
|
||||
)
|
||||
|
||||
expect(result.success).toBe(false)
|
||||
expect(result.message).toBe('Limit cannot exceed 1000')
|
||||
expect(mockQueryRows).not.toHaveBeenCalled()
|
||||
expect(result.success).toBe(true)
|
||||
const options = mockQueryRows.mock.calls[0][1] as Record<string, unknown>
|
||||
expect(options.limit).toBe(1000)
|
||||
})
|
||||
|
||||
it('queries without execution metadata and passes limit/offset through', async () => {
|
||||
@@ -732,7 +732,7 @@ describe('userTableServerTool.delete_rows_by_filter', () => {
|
||||
})
|
||||
})
|
||||
|
||||
it('rejects limits above MAX_BULK_OPERATION_SIZE', async () => {
|
||||
it('runs an explicit large limit inline without escalating (delete loads only ids)', async () => {
|
||||
const result = await userTableServerTool.execute(
|
||||
{
|
||||
operation: 'delete_rows_by_filter',
|
||||
@@ -741,9 +741,11 @@ describe('userTableServerTool.delete_rows_by_filter', () => {
|
||||
{ userId: 'user-1', workspaceId: 'workspace-1' }
|
||||
)
|
||||
|
||||
expect(result.success).toBe(false)
|
||||
expect(result.message).toBe('Limit cannot exceed 1000')
|
||||
expect(mockDeleteRowsByFilter).not.toHaveBeenCalled()
|
||||
expect(result.success).toBe(true)
|
||||
// An explicit limit never counts/escalates — it deletes inline, bounded by the limit.
|
||||
expect(mockQueryRows).not.toHaveBeenCalled()
|
||||
expect(mockDeleteRowsByFilter).toHaveBeenCalledTimes(1)
|
||||
expect(mockDeleteRowsByFilter.mock.calls[0][1]).toMatchObject({ limit: 5000 })
|
||||
})
|
||||
|
||||
it('deletes inline when the unbounded match count is within the cap', async () => {
|
||||
@@ -853,7 +855,14 @@ describe('userTableServerTool.update_rows_by_filter', () => {
|
||||
mockQueryRows.mockResolvedValue({ rows: [], rowCount: 0, totalCount: 5, limit: 1, offset: 0 })
|
||||
})
|
||||
|
||||
it('rejects limits above MAX_BULK_OPERATION_SIZE', async () => {
|
||||
it('escalates an explicit limit above the cap to a background update with maxRows', async () => {
|
||||
mockQueryRows.mockResolvedValueOnce({
|
||||
rows: [],
|
||||
rowCount: 0,
|
||||
totalCount: 20000,
|
||||
limit: 1,
|
||||
offset: 0,
|
||||
})
|
||||
const result = await userTableServerTool.execute(
|
||||
{
|
||||
operation: 'update_rows_by_filter',
|
||||
@@ -861,9 +870,16 @@ describe('userTableServerTool.update_rows_by_filter', () => {
|
||||
},
|
||||
{ userId: 'user-1', workspaceId: 'workspace-1' }
|
||||
)
|
||||
expect(result.success).toBe(false)
|
||||
expect(result.message).toBe('Limit cannot exceed 1000')
|
||||
await flushDetached()
|
||||
|
||||
expect(result.success).toBe(true)
|
||||
// target = min(limit 5000, matchCount 20000) = 5000, above the inline cap → background.
|
||||
expect(result.data?.affectedCount).toBe(5000)
|
||||
expect(mockUpdateRowsByFilter).not.toHaveBeenCalled()
|
||||
const [, , type, payload] = mockMarkTableJobRunning.mock.calls[0]
|
||||
expect(type).toBe('update')
|
||||
expect(payload).toMatchObject({ affectedCount: 5000, maxRows: 5000 })
|
||||
expect(mockRunTableUpdate.mock.calls[0][0]).toMatchObject({ maxRows: 5000 })
|
||||
})
|
||||
|
||||
it('updates inline when the unbounded match count is within the cap', async () => {
|
||||
@@ -909,6 +925,8 @@ describe('userTableServerTool.update_rows_by_filter', () => {
|
||||
cutoff: expect.any(String),
|
||||
data: { age: 1 },
|
||||
})
|
||||
// Unbounded match (no explicit limit) → the worker patches every match, no cap.
|
||||
expect((payload as { maxRows?: number }).maxRows).toBeUndefined()
|
||||
expect(mockRunTableUpdate).toHaveBeenCalledTimes(1)
|
||||
expect(mockRunTableUpdate.mock.calls[0][0]).toMatchObject({
|
||||
jobId,
|
||||
|
||||
@@ -200,8 +200,9 @@ async function dispatchUpdateJob(params: {
|
||||
filter: Filter
|
||||
data: RowData
|
||||
cutoff: Date
|
||||
maxRows?: number
|
||||
}): Promise<void> {
|
||||
const { jobId, tableId, workspaceId, filter, data, cutoff } = params
|
||||
const { jobId, tableId, workspaceId, filter, data, cutoff, maxRows } = params
|
||||
if (isTriggerDevEnabled) {
|
||||
try {
|
||||
const [{ tableUpdateTask }, { tasks }] = await Promise.all([
|
||||
@@ -210,7 +211,7 @@ async function dispatchUpdateJob(params: {
|
||||
])
|
||||
await tasks.trigger<typeof tableUpdateTask>(
|
||||
'table-update',
|
||||
{ jobId, tableId, workspaceId, filter, data, cutoff: cutoff.toISOString() },
|
||||
{ jobId, tableId, workspaceId, filter, data, cutoff: cutoff.toISOString(), maxRows },
|
||||
{ tags: [`tableId:${tableId}`, `jobId:${jobId}`] }
|
||||
)
|
||||
} catch (error) {
|
||||
@@ -219,10 +220,12 @@ async function dispatchUpdateJob(params: {
|
||||
}
|
||||
} else {
|
||||
runDetached('table-update', () =>
|
||||
runTableUpdate({ jobId, tableId, workspaceId, filter, data, cutoff }).catch(async (error) => {
|
||||
await markTableUpdateFailed(tableId, jobId, error)
|
||||
throw error
|
||||
})
|
||||
runTableUpdate({ jobId, tableId, workspaceId, filter, data, cutoff, maxRows }).catch(
|
||||
async (error) => {
|
||||
await markTableUpdateFailed(tableId, jobId, error)
|
||||
throw error
|
||||
}
|
||||
)
|
||||
)
|
||||
}
|
||||
}
|
||||
@@ -280,17 +283,16 @@ function parseDeploymentMode(value: unknown): WorkflowGroupDeploymentMode | unde
|
||||
}
|
||||
|
||||
/**
|
||||
* Validates an optional row limit against the same bounds the HTTP contracts
|
||||
* enforce. Returns an error message, or `null` when the limit is acceptable.
|
||||
* Validates an optional row limit. There's no upper bound the caller must respect — the model may
|
||||
* ask for any number. `MAX_QUERY_LIMIT` / `MAX_BULK_OPERATION_SIZE` are applied internally instead
|
||||
* (query_rows clamps the page; bulk ops above the bound run as a background job). Returns an error
|
||||
* message, or `null` when the limit is acceptable.
|
||||
*/
|
||||
function limitError(limit: unknown, max: number): string | null {
|
||||
function limitError(limit: unknown): string | null {
|
||||
if (limit === undefined) return null
|
||||
if (typeof limit !== 'number' || !Number.isInteger(limit) || limit < 1) {
|
||||
return 'Limit must be an integer of at least 1'
|
||||
}
|
||||
if (limit > max) {
|
||||
return `Limit cannot exceed ${max}`
|
||||
}
|
||||
return null
|
||||
}
|
||||
|
||||
@@ -579,7 +581,7 @@ export const userTableServerTool: BaseServerTool<UserTableArgs, UserTableResult>
|
||||
return { success: false, message: 'Workspace ID is required' }
|
||||
}
|
||||
|
||||
const queryLimitError = limitError(args.limit, TABLE_LIMITS.MAX_QUERY_LIMIT)
|
||||
const queryLimitError = limitError(args.limit)
|
||||
if (queryLimitError) {
|
||||
return { success: false, message: queryLimitError }
|
||||
}
|
||||
@@ -592,12 +594,18 @@ export const userTableServerTool: BaseServerTool<UserTableArgs, UserTableResult>
|
||||
const requestId = generateId().slice(0, 8)
|
||||
const idByName = buildIdByName(table.schema)
|
||||
const nameById = buildNameById(table.schema)
|
||||
// The model may request any number; we serve at most MAX_QUERY_LIMIT per page so a single
|
||||
// tool result can't drain a whole table. `totalCount` in the response signals truncation,
|
||||
// and the model pages with `offset`.
|
||||
const result = await queryRows(
|
||||
table,
|
||||
{
|
||||
filter: args.filter ? filterNamesToIds(args.filter, idByName) : undefined,
|
||||
sort: args.sort ? sortNamesToIds(args.sort, idByName) : undefined,
|
||||
limit: args.limit,
|
||||
limit:
|
||||
args.limit !== undefined
|
||||
? Math.min(args.limit, TABLE_LIMITS.MAX_QUERY_LIMIT)
|
||||
: undefined,
|
||||
offset: args.offset,
|
||||
withExecutions: false,
|
||||
},
|
||||
@@ -700,7 +708,7 @@ export const userTableServerTool: BaseServerTool<UserTableArgs, UserTableResult>
|
||||
if (!workspaceId) {
|
||||
return { success: false, message: 'Workspace ID is required' }
|
||||
}
|
||||
const updateLimitError = limitError(args.limit, TABLE_LIMITS.MAX_BULK_OPERATION_SIZE)
|
||||
const updateLimitError = limitError(args.limit)
|
||||
if (updateLimitError) {
|
||||
return { success: false, message: updateLimitError }
|
||||
}
|
||||
@@ -715,29 +723,34 @@ export const userTableServerTool: BaseServerTool<UserTableArgs, UserTableResult>
|
||||
const idFilter = filterNamesToIds(args.filter, idByName)
|
||||
const idData = rowDataNameToId(args.data, idByName)
|
||||
|
||||
// Unbounded "update everything matching": measure the blast radius first and hand
|
||||
// anything past the inline cap to the background update worker — same escalation as
|
||||
// delete_rows_by_filter, so a broad update on a huge table doesn't load every matching
|
||||
// row into this request. A patch touching a unique column stays inline (the service
|
||||
// rejects bulk-setting a unique value across multiple rows).
|
||||
// Inline handles up to MAX_BULK_OPERATION_SIZE rows in one request; a larger operation
|
||||
// (an explicit limit above the cap, or unbounded "update everything matching") runs in the
|
||||
// background worker so a broad update on a huge table doesn't load every matching row into
|
||||
// this request. A small explicit limit is the fast path — no count needed. A patch
|
||||
// touching a unique column always stays inline (the service rejects bulk-setting a unique
|
||||
// value across multiple rows).
|
||||
const patchTouchesUnique = table.schema.columns.some(
|
||||
(c) => c.unique === true && (c.id ?? c.name) in idData
|
||||
)
|
||||
if (args.limit === undefined && !patchTouchesUnique) {
|
||||
const updateInlineEligible =
|
||||
args.limit !== undefined && args.limit <= TABLE_LIMITS.MAX_BULK_OPERATION_SIZE
|
||||
if (!updateInlineEligible && !patchTouchesUnique) {
|
||||
const { totalCount } = await queryRows(
|
||||
table,
|
||||
{ filter: idFilter, limit: 1, withExecutions: false },
|
||||
requestId
|
||||
)
|
||||
const matchCount = totalCount ?? 0
|
||||
if (matchCount > TABLE_LIMITS.MAX_BULK_OPERATION_SIZE) {
|
||||
const target = args.limit !== undefined ? Math.min(args.limit, matchCount) : matchCount
|
||||
if (target > TABLE_LIMITS.MAX_BULK_OPERATION_SIZE) {
|
||||
const cutoff = new Date()
|
||||
const jobId = generateId()
|
||||
const payload: TableUpdateJobPayload = {
|
||||
filter: idFilter,
|
||||
data: idData,
|
||||
cutoff: cutoff.toISOString(),
|
||||
affectedCount: matchCount,
|
||||
affectedCount: target,
|
||||
maxRows: args.limit,
|
||||
}
|
||||
assertNotAborted()
|
||||
const claimed = await markTableJobRunning(table.id, jobId, 'update', payload)
|
||||
@@ -751,11 +764,12 @@ export const userTableServerTool: BaseServerTool<UserTableArgs, UserTableResult>
|
||||
filter: idFilter,
|
||||
data: idData,
|
||||
cutoff,
|
||||
maxRows: args.limit,
|
||||
})
|
||||
return {
|
||||
success: true,
|
||||
message: `Started background update of ${matchCount} matching rows (job ${jobId}). Rows update in the background — query_rows to check progress. Note: background updates don't auto-recompute workflow/enrichment columns; use run_column afterward if needed.`,
|
||||
data: { jobId, affectedCount: matchCount },
|
||||
message: `Started background update of ${target} matching rows (job ${jobId}). Rows update in the background — query_rows to check progress. Note: background updates don't auto-recompute workflow/enrichment columns; use run_column afterward if needed.`,
|
||||
data: { jobId, affectedCount: target },
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -789,7 +803,7 @@ export const userTableServerTool: BaseServerTool<UserTableArgs, UserTableResult>
|
||||
if (!workspaceId) {
|
||||
return { success: false, message: 'Workspace ID is required' }
|
||||
}
|
||||
const deleteLimitError = limitError(args.limit, TABLE_LIMITS.MAX_BULK_OPERATION_SIZE)
|
||||
const deleteLimitError = limitError(args.limit)
|
||||
if (deleteLimitError) {
|
||||
return { success: false, message: deleteLimitError }
|
||||
}
|
||||
@@ -803,10 +817,11 @@ export const userTableServerTool: BaseServerTool<UserTableArgs, UserTableResult>
|
||||
const idByName = buildIdByName(table.schema)
|
||||
const idFilter = filterNamesToIds(args.filter, idByName)
|
||||
|
||||
// Unbounded "delete everything matching": measure the blast radius
|
||||
// first, and hand anything past the inline cap to the background
|
||||
// delete worker (same path as the UI's select-all delete) instead of
|
||||
// loading every matching row id into this request.
|
||||
// An explicit limit runs inline (delete loads only row ids, so even a large bounded
|
||||
// delete is light). Only an unbounded "delete everything matching" measures the blast
|
||||
// radius and hands off to the background delete worker (same path as the UI's select-all
|
||||
// delete) — the read-path mask hides exactly the all-matching set, which a bounded delete
|
||||
// would over-hide.
|
||||
if (args.limit === undefined) {
|
||||
const { totalCount } = await queryRows(
|
||||
table,
|
||||
|
||||
@@ -231,6 +231,8 @@ export interface TableUpdateJobPayload {
|
||||
/** ISO timestamp; rows created after it are not patched. */
|
||||
cutoff: string
|
||||
affectedCount?: number
|
||||
/** Stop after updating this many rows (an explicit caller-supplied limit). Omitted = every match. */
|
||||
maxRows?: number
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -163,6 +163,23 @@ describe('runTableUpdate', () => {
|
||||
expect(mockUpdatePageByIds).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('stops once maxRows is reached and never over-fetches a page', async () => {
|
||||
// budget 3 with page size 2: first page fills 2, second page is capped to the remaining 1.
|
||||
mockSelectRowDataPage
|
||||
.mockResolvedValueOnce([row('a'), row('b')])
|
||||
.mockResolvedValueOnce([row('c')])
|
||||
|
||||
await runTableUpdate(basePayload({ maxRows: 3 }))
|
||||
|
||||
expect(mockSelectRowDataPage).toHaveBeenCalledTimes(2)
|
||||
expect(mockSelectRowDataPage.mock.calls[0][0]).toMatchObject({ limit: 2 })
|
||||
expect(mockSelectRowDataPage.mock.calls[1][0]).toMatchObject({ limit: 1 })
|
||||
expect(mockUpdatePageByIds).toHaveBeenCalledTimes(2)
|
||||
expect(mockAppendTableEvent).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ status: 'ready', progress: 3 })
|
||||
)
|
||||
})
|
||||
|
||||
it('passes the cutoff and filter clause through to the page query', async () => {
|
||||
mockSelectRowDataPage.mockResolvedValueOnce([])
|
||||
|
||||
|
||||
@@ -37,6 +37,8 @@ export interface TableUpdatePayload {
|
||||
data: RowData
|
||||
/** Only rows created at/before this instant are patched, so mid-job inserts are spared. */
|
||||
cutoff: Date
|
||||
/** Stop after updating this many rows (an explicit caller-supplied limit). Omitted = every match. */
|
||||
maxRows?: number
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -57,8 +59,9 @@ export interface TableUpdatePayload {
|
||||
* failed via `markTableUpdateFailed`. A superseded run returns quietly.
|
||||
*/
|
||||
export async function runTableUpdate(payload: TableUpdatePayload): Promise<void> {
|
||||
const { jobId, tableId, workspaceId, filter, data, cutoff } = payload
|
||||
const { jobId, tableId, workspaceId, filter, data, cutoff, maxRows } = payload
|
||||
const requestId = generateId().slice(0, 8)
|
||||
const budget = maxRows ?? Number.POSITIVE_INFINITY
|
||||
|
||||
try {
|
||||
const table = await getTableById(tableId, { includeArchived: true })
|
||||
@@ -81,7 +84,7 @@ export async function runTableUpdate(payload: TableUpdatePayload): Promise<void>
|
||||
let lastReported = resumed
|
||||
let afterId: string | undefined
|
||||
|
||||
while (true) {
|
||||
while (processed < budget) {
|
||||
const owns = await updateJobProgress(tableId, processed, jobId)
|
||||
if (!owns) throw new JobSupersededError()
|
||||
|
||||
@@ -91,7 +94,7 @@ export async function runTableUpdate(payload: TableUpdatePayload): Promise<void>
|
||||
cutoff,
|
||||
filterClause,
|
||||
afterId,
|
||||
limit: TABLE_LIMITS.DELETE_PAGE_SIZE,
|
||||
limit: Math.min(TABLE_LIMITS.DELETE_PAGE_SIZE, budget - processed),
|
||||
// Skip rows already carrying the patch so a retried run resumes without re-walking /
|
||||
// double-counting the rows an earlier attempt updated (updated rows still exist and may
|
||||
// still match the filter, unlike deletes).
|
||||
|
||||
Reference in New Issue
Block a user