Skip to content

Commit 283f404

Browse files
committed
fix(tables): refresh open tables when a marker cleanup loses its reply
A repeat after a lost reply finds nothing left to update, so it cannot name the cells the lost attempt failed. The cleanup now owns its retry and signals a refetch. Correct the retry budget comment: maxElapsedMs only decides whether another attempt starts.
1 parent 9353f58 commit 283f404

4 files changed

Lines changed: 61 additions & 26 deletions

File tree

‎apps/sim/background/workflow-column-execution.ts‎

Lines changed: 10 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,7 @@ import {
2525
} from '@/lib/core/execution-limits'
2626
import { withResourceOutboundScope } from '@/lib/core/network/resource-scope.server'
2727
import { RateLimiter } from '@/lib/core/rate-limiter/rate-limiter'
28-
import { withDatabaseReadRetry, withIdempotentWriteRetry } from '@/lib/db/read-retry'
28+
import { withDatabaseReadRetry } from '@/lib/db/read-retry'
2929
import {
3030
registerManualExecutionAborter,
3131
unregisterManualExecutionAborter,
@@ -422,16 +422,15 @@ export async function runRowCascadeLoop(
422422
} catch (err) {
423423
const { tableId, rowId } = payload
424424
const { terminalizeUnclaimedRowMarkers } = await import('@/lib/table/cell-write')
425-
await withIdempotentWriteRetry(
426-
() => terminalizeUnclaimedRowMarkers(tableId, rowId, CELL_STOPPED_MESSAGE),
427-
{ label: 'terminalizeUnclaimedRowMarkers', ...CELL_DATABASE_RETRY }
428-
).catch((markerErr) => {
429-
logger.error('Failed to terminalize the unclaimed markers of a failed row', {
430-
tableId,
431-
rowId,
432-
cause: describeError(markerErr),
433-
})
434-
})
425+
await terminalizeUnclaimedRowMarkers(tableId, rowId, CELL_STOPPED_MESSAGE).catch(
426+
(markerErr) => {
427+
logger.error('Failed to terminalize the unclaimed markers of a failed row', {
428+
tableId,
429+
rowId,
430+
cause: describeError(markerErr),
431+
})
432+
}
433+
)
435434
throw err
436435
}
437436
}

‎apps/sim/lib/table/cell-write.integration.ts‎

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -433,6 +433,33 @@ describe('a durable cell state write around a lost connection', { timeout: 90_00
433433
expect(tableEventsMockFns.mockAppendTableEvent).toHaveBeenCalledOnce()
434434
})
435435

436+
it('tells open tables to refetch when the reply to failing the markers is lost', async () => {
437+
const { tableId, rowId } = await createCell({ pickedUp: false })
438+
await writeExecutionsPatch(db, tableId, rowId, {
439+
[groupId]: {
440+
status: 'pending',
441+
executionId: null,
442+
jobId: null,
443+
workflowId: 'workflow-1',
444+
error: null,
445+
},
446+
})
447+
/** An unprepared statement waits on a reply before it executes, so the commit is applied here. */
448+
interception = {
449+
match: /update "table_row_executions"/i,
450+
forward: false,
451+
beforeReset: async () => {
452+
await control`UPDATE table_row_executions SET status = 'error' WHERE row_id = ${rowId}`
453+
},
454+
}
455+
456+
await terminalizeUnclaimedRowMarkers(tableId, rowId, 'Stopped')
457+
458+
expect((await readCell(rowId)).status).toBe('error')
459+
expect(tableEventsMockFns.mockAppendTableEvent).not.toHaveBeenCalled()
460+
expect(tableEventsMockFns.mockSignalTableRowsChanged).toHaveBeenCalledWith(tableId)
461+
})
462+
436463
it('does not repeat a cancel stamp, which could kill a run the user just started', async () => {
437464
const { ctx, rowId } = await createCell()
438465
interception = {

‎apps/sim/lib/table/cell-write.ts‎

Lines changed: 21 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,7 @@ import { getValueAtPath } from '@sim/utils/object'
1616
import { and, eq, isNull } from 'drizzle-orm'
1717
import { withIdempotentWriteRetry } from '@/lib/db/read-retry'
1818
import { CELL_DATABASE_RETRY } from '@/lib/table/constants'
19-
import { appendTableEvent } from '@/lib/table/events'
19+
import { appendTableEvent, signalTableRowsChanged } from '@/lib/table/events'
2020
import { TableRowNotFoundError } from '@/lib/table/rows/errors'
2121
import { loadExecutionsForRow, writeExecutionsPatch } from '@/lib/table/rows/executions'
2222
import {
@@ -204,24 +204,32 @@ export async function confirmWorkflowGroupStateCommitted(
204204
* Fails every marker on a row that no worker has claimed. `executionId` stays null, so a carrier
205205
* still queued for one of them can claim it and run. Claimed, cancelled, and finished cells are
206206
* untouched, and so is a paused one, which holds an execution id.
207+
*
208+
* Open tables are told to refetch too: a repeat after a lost reply finds nothing left to update, so
209+
* it cannot name the cells the lost attempt failed.
207210
*/
208211
export async function terminalizeUnclaimedRowMarkers(
209212
tableId: string,
210213
rowId: string,
211214
error: string
212215
): Promise<void> {
213-
const failed = await db
214-
.update(tableRowExecutions)
215-
.set({ status: 'error', jobId: null, error, runningBlockIds: [], updatedAt: new Date() })
216-
.where(
217-
and(
218-
eq(tableRowExecutions.tableId, tableId),
219-
eq(tableRowExecutions.rowId, rowId),
220-
eq(tableRowExecutions.status, 'pending'),
221-
isNull(tableRowExecutions.executionId)
222-
)
223-
)
224-
.returning({ groupId: tableRowExecutions.groupId })
216+
const failed = await withIdempotentWriteRetry(
217+
() =>
218+
db
219+
.update(tableRowExecutions)
220+
.set({ status: 'error', jobId: null, error, runningBlockIds: [], updatedAt: new Date() })
221+
.where(
222+
and(
223+
eq(tableRowExecutions.tableId, tableId),
224+
eq(tableRowExecutions.rowId, rowId),
225+
eq(tableRowExecutions.status, 'pending'),
226+
isNull(tableRowExecutions.executionId)
227+
)
228+
)
229+
.returning({ groupId: tableRowExecutions.groupId }),
230+
{ label: 'terminalizeUnclaimedRowMarkers', ...CELL_DATABASE_RETRY }
231+
)
232+
signalTableRowsChanged(tableId)
225233
for (const { groupId } of failed) {
226234
void appendTableEvent({
227235
kind: 'cell',

‎apps/sim/lib/table/constants.ts‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -523,8 +523,9 @@ export function generateUniqueTableName(existingNames: string[]): string {
523523
* How a cell rides out lost database connections. A reset takes individual sockets and a fresh one
524524
* works, but resets arrive in bursts of about a second, so backoff totals about four seconds to
525525
* outlast one. A reset can also stall the replacement connection until `connect_timeout`, so
526-
* `maxElapsedMs` admits one more attempt after such a stall and no more, which bounds how long each
527-
* failing operation holds the cell's concurrency slot and cascade lock.
526+
* `maxElapsedMs` leaves room to try again after such a stall. It only decides whether another attempt
527+
* starts, so one operation can hold the cell's concurrency slot and cascade lock for up to
528+
* `maxElapsedMs` plus a final stall, about 70 seconds.
528529
*/
529530
export const CELL_DATABASE_RETRY = {
530531
maxAttempts: 5,

0 commit comments

Comments
 (0)