mirror of
https://github.com/simstudioai/sim.git
synced 2026-09-24 15:45:35 +08:00
fix(tables): heartbeat export job before upload so the stale janitor can't kill a live finalize (#5017)
* fix(tables): heartbeat export job before upload so the stale janitor can't kill a live finalize * chore(tables): rename stale-janitor counters to cover all table job types
This commit is contained in:
@@ -113,15 +113,15 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
|
||||
})
|
||||
}
|
||||
|
||||
// Mark stale table jobs (import or delete) as failed. Jobs run detached on the web container
|
||||
// Mark stale table jobs (import, export, or delete) as failed. Jobs run detached on the web container
|
||||
// and are lost if the pod is killed mid-run. `updated_at` is bumped by progress updates, so a
|
||||
// `running` job with no recent update has stalled (not merely slow). Committed work is left in
|
||||
// place (no rollback); the user retries. Also prune long-settled terminal jobs so the table
|
||||
// doesn't grow unbounded (the latest job per table is what list/detail reads surface).
|
||||
let staleImportsMarkedFailed = 0
|
||||
let staleTableJobsMarkedFailed = 0
|
||||
try {
|
||||
const now = new Date()
|
||||
const staleImports = await db
|
||||
const staleJobs = await db
|
||||
.update(tableJobs)
|
||||
.set({
|
||||
status: 'failed',
|
||||
@@ -132,9 +132,9 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
|
||||
.where(and(eq(tableJobs.status, 'running'), lt(tableJobs.updatedAt, staleThreshold)))
|
||||
.returning({ id: tableJobs.id })
|
||||
|
||||
staleImportsMarkedFailed = staleImports.length
|
||||
if (staleImportsMarkedFailed > 0) {
|
||||
logger.info(`Marked ${staleImportsMarkedFailed} stale table jobs as failed`)
|
||||
staleTableJobsMarkedFailed = staleJobs.length
|
||||
if (staleTableJobsMarkedFailed > 0) {
|
||||
logger.info(`Marked ${staleTableJobsMarkedFailed} stale table jobs as failed`)
|
||||
}
|
||||
|
||||
const terminalRetention = new Date(Date.now() - TABLE_JOB_RETENTION_HOURS * 60 * 60 * 1000)
|
||||
@@ -236,8 +236,8 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
|
||||
staleThresholdMinutes: STALE_THRESHOLD_MINUTES,
|
||||
retentionHours: JOB_RETENTION_HOURS,
|
||||
},
|
||||
tableImports: {
|
||||
staleMarkedFailed: staleImportsMarkedFailed,
|
||||
tableJobs: {
|
||||
staleMarkedFailed: staleTableJobsMarkedFailed,
|
||||
},
|
||||
})
|
||||
} catch (error) {
|
||||
|
||||
@@ -104,6 +104,17 @@ describe('runTableExport', () => {
|
||||
expect(mockMarkJobFailed).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('stops before the upload when ownership is lost at the finalize gate', async () => {
|
||||
mockUpdateJobProgress.mockResolvedValueOnce(true).mockResolvedValue(false)
|
||||
|
||||
await runTableExport(payload)
|
||||
|
||||
expect(mockSelectExportRowPage).toHaveBeenCalledTimes(1)
|
||||
expect(mockUploadFile).not.toHaveBeenCalled()
|
||||
expect(mockMarkJobReady).not.toHaveBeenCalled()
|
||||
expect(mockMarkJobFailed).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('cleans up an orphaned upload when the job was canceled at the wire', async () => {
|
||||
mockMarkJobReady.mockResolvedValue(false)
|
||||
|
||||
|
||||
@@ -93,6 +93,9 @@ export async function runTableExport(payload: TableExportPayload): Promise<void>
|
||||
}
|
||||
if (format === 'json') chunks.push(']')
|
||||
|
||||
const ownsFinalize = await updateJobProgress(tableId, exported, jobId)
|
||||
if (!ownsFinalize) throw new JobSupersededError()
|
||||
|
||||
const fileName = `${sanitizeExportFilename(table.name)}.${format}`
|
||||
const key = `workspace/${workspaceId}/exports/${tableId}/${jobId}/${fileName}`
|
||||
// `preserveKey` keeps the custom key verbatim (without it the provider rewrites the key to a
|
||||
|
||||
Reference in New Issue
Block a user