mirror of
https://github.com/stablyai/orca.git
synced 2026-10-18 08:03:22 +00:00
fix(orchestration): stamp fallback-resolved ids onto null-id worker_done
A null-id worker_done resolved via the sole-active-dispatch fallback left its payload id-less, so a later re-read of the same message re-resolved against current sender state. Once the pane received a fresh dispatch B, the stale message wrongly completed B (potentially releasing B's dependents early). Persist the resolved taskId/dispatchId onto the message payload before completing the task, reusing the existing payload-update mechanism. A re-read now takes the stable id-bearing path and re-targets the original dispatch A instead of B. Idempotent and recoverable on partial failure. Co-authored-by: Orca <help@stably.ai>
This commit is contained in:
@@ -61,6 +61,23 @@ function addLifecycleRejectionMarker(payload: string | null, reason: string): st
|
||||
})
|
||||
}
|
||||
|
||||
function stampResolvedLifecycleIds(
|
||||
payload: string | null,
|
||||
taskId: string,
|
||||
dispatchId: string
|
||||
): string {
|
||||
let parsed: Record<string, unknown> = {}
|
||||
try {
|
||||
const value: unknown = payload ? JSON.parse(payload) : {}
|
||||
if (value && typeof value === 'object' && !Array.isArray(value)) {
|
||||
parsed = value as Record<string, unknown>
|
||||
}
|
||||
} catch {
|
||||
// Fallback resolution only reaches this path with object/empty payloads.
|
||||
}
|
||||
return JSON.stringify({ ...parsed, taskId, dispatchId })
|
||||
}
|
||||
|
||||
const SQLITE_UTC_TIMESTAMP_RE = /^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}(?:\.\d+)?$/
|
||||
|
||||
function exposeUtcTimestamp(timestamp: string | null): string | null {
|
||||
@@ -385,6 +402,16 @@ export class OrchestrationDb {
|
||||
return this.getMessageById(messageId)
|
||||
}
|
||||
|
||||
// Why: stamp fallback-resolved ids onto a null-id worker_done so a later re-read takes the stable id path instead of re-resolving against a newer dispatch (#7429).
|
||||
stampResolvedWorkerDoneIds(messageId: string, taskId: string, dispatchId: string): void {
|
||||
const message = this.getMessageById(messageId)
|
||||
if (!message || message.type !== 'worker_done') {
|
||||
return
|
||||
}
|
||||
const payload = stampResolvedLifecycleIds(message.payload, taskId, dispatchId)
|
||||
this.db.prepare('UPDATE messages SET payload = ? WHERE id = ?').run(payload, messageId)
|
||||
}
|
||||
|
||||
// Why: delivered_at IS NULL filter — push-on-idle delivers each row at most once; read (set only by check) wouldn't prevent replay.
|
||||
getUndeliveredUnreadMessages(toHandle: string, types?: MessageType[]): MessageRow[] {
|
||||
if (types && types.length > 0) {
|
||||
|
||||
@@ -432,6 +432,41 @@ describe('lifecycle reconciliation', () => {
|
||||
expect(db.getDispatchContextById(dispatch.id)?.status).toBe('completed')
|
||||
})
|
||||
|
||||
it('does not let a re-read of a fallback-resolved worker_done complete a later dispatch to the same pane', () => {
|
||||
db = new OrchestrationDb(':memory:')
|
||||
const taskA = db.createTask({ spec: 'work A' })
|
||||
const dispatchA = db.createDispatchContext(taskA.id, 'term_worker', `tab_w:${LEAF_A}`)
|
||||
const done = db.insertMessage({
|
||||
from: 'term_worker',
|
||||
to: 'term_coordinator',
|
||||
subject: 'Done',
|
||||
type: 'worker_done',
|
||||
payload: JSON.stringify({}),
|
||||
senderPaneKey: `tab_w:${LEAF_A}`
|
||||
})
|
||||
|
||||
// First reconcile completes A via the sole-active-dispatch fallback.
|
||||
expect(reconcileLifecycleMessage(db, done)).toMatchObject({
|
||||
action: 'completed',
|
||||
taskId: taskA.id,
|
||||
dispatchId: dispatchA.id
|
||||
})
|
||||
|
||||
// The same pane now owns a fresh dispatch B.
|
||||
const taskB = db.createTask({ spec: 'work B' })
|
||||
const dispatchB = db.createDispatchContext(taskB.id, 'term_worker', `tab_w:${LEAF_A}`)
|
||||
|
||||
// Re-reading the ORIGINAL message (as the coordinator does) must re-target A, not B.
|
||||
const reread = db.getMessageById(done.id)
|
||||
expect(reread && reconcileLifecycleMessage(db, reread)).toMatchObject({
|
||||
action: 'completed',
|
||||
taskId: taskA.id,
|
||||
dispatchId: dispatchA.id
|
||||
})
|
||||
expect(db.getTask(taskB.id)?.status).toBe('dispatched')
|
||||
expect(db.getDispatchContextById(dispatchB.id)?.status).toBe('dispatched')
|
||||
})
|
||||
|
||||
it('ignores a null-id worker_done sent from a foreign pane and handle', () => {
|
||||
db = new OrchestrationDb(':memory:')
|
||||
const task = db.createTask({ spec: 'work' })
|
||||
|
||||
@@ -252,7 +252,10 @@ function resolveWorkerDoneTarget(
|
||||
|
||||
const fallback = db.getSoleActiveDispatchForSender(msg.from_handle, msg.sender_pane_key)
|
||||
if (fallback && hasLifecycleAuthority(fallback, msg)) {
|
||||
return { taskId: taskId || fallback.task_id, dispatchId: dispatchId || fallback.id }
|
||||
const resolved = { taskId: taskId || fallback.task_id, dispatchId: dispatchId || fallback.id }
|
||||
// Why: persist before completion so a re-read of this same message uses the recorded ids; else the sole-dispatch fallback would re-resolve onto a later dispatch B and wrongly complete it (#7429).
|
||||
db.stampResolvedWorkerDoneIds(msg.id, resolved.taskId, resolved.dispatchId)
|
||||
return resolved
|
||||
}
|
||||
|
||||
onLog(
|
||||
|
||||
Reference in New Issue
Block a user