Skip to content

fix(run): emit the queued NDJSON event only after the job state file is persisted #663

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mattmillerai merged 1 commit into from
Aug 3, 2026
+322 −26
Merged

fix(run): emit the queued NDJSON event only after the job state file is persisted #663

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
80 changes: 66 additions & 14 deletions comfy_cli/command/run/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -385,6 +385,12 @@ def execute(
wait_state.status = "running"
_write_state(wait_state)

@github-actions github-actions Bot Aug 2, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚪ Nit — The event stream is now sequenced behind jobs_state.write, which takes locking.file_lock with no timeout (fcntl.flock(fd, LOCK_EX) blocks indefinitely). A stuck holder of <prompt_id>.lock — a SIGSTOPped watcher, a hung NFS mount — now stalls the queued announcement itself, so a consumer blocking on that line hangs unbounded; passing a timeout= would keep the announcement live. Raised by 1 of 8 reviewers (claude-opus-5-thinking-max adversarial).

@mattmillerai mattmillerai Aug 3, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Real, but deferring to a follow-up ticket rather than fixing here.

Confirmed the mechanics: jobs_state.write (comfy_cli/jobs_state.py:125) takes locking.file_lock(path.with_suffix(".lock")) with no timeout, and locking.file_lock (comfy_cli/locking.py:49) blocks indefinitely when timeout is None. So a stuck holder does stall the announcement, and a consumer blocking on that line hangs unbounded.

Out of scope for this PR because the lock has always been unbounded — this PR only changed which line is sequenced behind the write (the async path already wrote state before emitting its envelope). Fixing it means threading a timeout through a primitive shared by the foreground run paths, the detached watcher (command/job_watcher.py), and jobs ls's stale-watcher reap (command/jobs.py:211-222), and deciding per call site what a timeout means. The run command's submit-time writes now go through the best-effort _write_state, so they can degrade to state_file: null cleanly — but the watcher and the reap write assume the write lands, and making them fail silently on timeout would quietly stop jobs ls reaping. That needs a decision before coding, so it isn't something to bolt on here.

Recorded as a follow-up (low severity) with that analysis and the proposed write(state, *, lock_timeout=…) shape. Worth noting file_lock's own docstring already flags that timeout is best-effort and that flock on NFS is often a silent no-op, so the hung-NFS half may need a different primitive rather than a timeout.


# Only now, with the durable record on disk, announce the queue.
# The event-order contract (docs/json-output.md) is
# `prompt_preview → queued → node events`, so this must also
# precede `watch_execution`'s WebSocket stream.
_emit_queued(renderer, execution)

@github-actions github-actions Bot Aug 2, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 Medium — This emit now sits inside the try whose handler treats nearly every OSError as a WebSocket failure, and only BrokenPipeError is exempted. An ENOSPC/EIO from a redirected or closed stdout will therefore rewrite a perfectly healthy job record to error/server_died even though the server connection is intact. Raised by 3 of 8 reviewers (gemini-3.1-pro adversarial, gpt-5.6-sol-max edge-case, kimi-k2.7-code edge-case).

@mattmillerai mattmillerai Aug 3, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed by the same change as the BrokenPipeError gate above.

The narrow exemption was the wrong shape for exactly this reason — the guard now keys on who owns the failing fd, not on which OSError subclass it is. Renderer._write_json_line sets _stream_broken on any OSError from its own machine_stream write, so an ENOSPC/EIO from a redirected or closed stdout is attributed to us and re-raised, instead of rewriting a healthy record to error/server_died.

Because the flag lives in the renderer rather than at this one call site, it also covers a stdout failure during the node events watch_execution streams later in the same try — which the narrow exemption never did.

Covered by test_a_failed_write_flags_it_and_still_raises in tests/comfy_cli/output/test_renderer.py, parametrized over BrokenPipeError and OSError(ENOSPC).


# `watch_execution` reports a terminal server event by rendering
# the error and raising `typer.Exit` (1 for `execution_error`, 130
# for `execution_interrupted`) — the ordinary failure path, not an
Expand Down Expand Up @@ -460,6 +466,10 @@ def execute(
state_file = jobs_state.write(state)
watcher_spawned = _spawn_watcher(execution.prompt_id, where="local", host=host, port=port, notify=notify)

# Emitted after the state file and the watcher spawn attempt, so a
# consumer reading this line can already `comfy jobs status` it.
_emit_queued(renderer, execution)

@github-actions github-actions Bot Aug 2, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟠 High — The local async branch now sequences the only pre-envelope carrier of prompt_id behind the unguarded jobs_state.write(state) at line 466, which raises ValueError when the server's prompt_id fails state_path's ^[a-zA-Z0-9_\-]{1,128}$ check and OSError on an unwritable state dir or full disk. ValueError isn't in the handler tuple (traceback, no envelope); OSError is misreported as ws_disconnected and, since wait_state is None here, the details omit the prompt_id — so an already-accepted job becomes unaddressable. Use the existing best-effort _write_state helper, which swallows both. Raised by 3 of 8 reviewers (claude-opus-5-thinking-max adversarial, claude-opus-5-thinking-max edge-case, gpt-5.6-sol-max adversarial).

@mattmillerai mattmillerai Aug 3, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed — routed through _write_state.

You're right that this PR is what makes it matter: before the reordering, queued (the only pre-envelope carrier of the prompt_id) was emitted before the write, so a write failure still left the caller holding the id of an accepted job. Sequencing the emit behind an unguarded jobs_state.write handed that guarantee back. ValueError escaped as a traceback and OSError was misreported as ws_disconnected with the prompt_id omitted (since wait_state is None on this branch), exactly as described.

state_file is now _write_state(state), and the emit reports the outcome as a new nullable state_file field on the event rather than being gated on it — losing the id of an accepted job is the worse failure. See the reply on the _emit_queued thread for that part.

Pinned by test_a_failed_write_still_announces_but_reports_null, parametrized over OSError and ValueError: asserts exit 0, prompt_id still present, state_file: null. Verified it fails when the call is reverted to jobs_state.write.


if renderer.is_pretty():
from comfy_cli.output.glyphs import status_glyph

Expand Down Expand Up @@ -527,6 +537,12 @@ def execute(
)
raise typer.Exit(code=1)
except (WebSocketException, ConnectionError, OSError) as e:
if isinstance(e, BrokenPipeError):

@github-actions github-actions Bot Aug 2, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟠 High — isinstance(e, BrokenPipeError) cannot tell a closed stdout from a genuine EPIPE on another fd: this try wraps execution.connect() and execution.watch_execution(), and websocket-client writes pong/close frames back to the server during recv(), so a server that vanishes mid-run can surface here as BrokenPipeError and get laundered into click's silent exit 0 with no ws_disconnected envelope and the record left non-terminal. It also short-circuits the token.is_set() check below, so a Ctrl-C that ends in EPIPE skips _mark_cancelled and the cancelled/130 envelope. Gate on stdout ownership (e.g. check the renderer's stream is the failing one) rather than the exception type alone. Raised by 4 of 8 reviewers (claude-opus-5-thinking-max adversarial, claude-opus-5-thinking-max edge-case, gemini-3.1-pro adversarial, gpt-5.6-sol-max edge-case).

@mattmillerai mattmillerai Aug 3, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed in the follow-up commit — you're right that the exception type can't attribute the failing fd, and the short-circuit was the worse half of it.

The bail-out now gates on stdout ownership rather than the type: Renderer._write_json_line flags _stream_broken = True when one of its own machine_stream writes raises OSError, and the handler checks renderer.stream_broken. A BrokenPipeError that websocket-client raised from its own pong/close sends never sets that flag, so it now falls through to the ws_disconnected path with the server_died record write intact.

I also moved the token.is_set() check above the bail-out, so a Ctrl-C that ends in EPIPE still runs _mark_cancelled and emits the cancelled/130 envelope.

Both are pinned by tests that I verified fail against the old arrangement:

  • test_a_socket_epipe_is_still_reported_as_a_lost_server — socket-side EPIPE, asserts exit 1 + ws_disconnected + error/server_died. Reverting the gate to isinstance(e, BrokenPipeError) fails it.
  • test_ctrl_c_ending_in_epipe_still_reports_cancelled — fails against the exact pre-fix arrangement.
  • test_cancellation_outranks_a_closed_stdout — both signals at once; fails if the gate is moved back above the token check.

# A closed stdout is the consumer leaving, not the server dying —
# let click/__main__ turn it into the documented silent exit 0 and
# leave the just-written state record untouched. Same principle as
# `completed_payload` being emitted outside this try.
raise

@github-actions github-actions Bot Aug 2, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟠 High — Re-raising BrokenPipeError on the --wait path removes the only code path that finalized the on-disk record, leaving the submit-time record from line 386 at status="running" with watcher_pid=None forever: jobs ls only reaps non-terminal records whose recorded watcher_pid is dead, and --wait never sets one, so the phantom job is unreapable and jobs wait on it burns its full timeout. Consider marking the record (or spawning a detached watcher) before re-raising. Raised by 4 of 8 reviewers (claude-opus-5-thinking-max adversarial, claude-opus-5-thinking-max edge-case, gemini-3.1-pro adversarial, gemini-3.1-pro edge-case).

@mattmillerai mattmillerai Aug 3, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed — the bail-out now hands the record off before re-raising.

Confirmed the phantom is this PR's doing: pre-PR a BrokenPipeError here was caught by the OSError handler and finalized the record as error/server_died, so re-raising removed the only path that moved it off running. And --wait never sets watcher_pid, so jobs ls's reap (jobs.py:211-222) could never touch it.

Took your first suggestion, since the job genuinely is still running server-side — marking it terminal would be a lie. New _hand_off_record helper: spawn a detached watcher for the prompt, and if that fails, record watcher_pid = os.getpid() so the reaper finishes the job off once this process is gone. Guarded against raising, since it runs while the BrokenPipeError is already unwinding toward the silent exit 0 — replacing it would turn that 0 into a traceback.

The same hand-off is wired into the cloud --wait announcement (see the _emit_queued_cloud thread).

Pinned by test_local_wait_leaves_the_running_record_intact (asserts the watcher spawn) and test_local_wait_names_itself_reapable_when_no_watcher_spawns (asserts the os.getpid() fallback). Both fail when the _hand_off_record call is removed.

# If we closed the WebSocket ourselves in response to Ctrl-C, the recv
# loop exits with a WebSocketException that *looks* like the server
# vanished. Check the cancellation token first so we emit the right
Expand Down Expand Up @@ -594,6 +610,24 @@ def execute(
renderer.emit(completed_payload, command="run", where="local")


def _emit_queued(renderer, execution) -> None:

@github-actions github-actions Bot Aug 2, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 Medium — _emit_queued takes no signal of whether the preceding persistence actually succeeded, and every caller emits unconditionally — _write_state is best-effort and returns None when the write was skipped or the state dir was unwritable. That silently falsifies the new docs guarantee that a consumer may run comfy jobs status <prompt_id> the moment it reads the line; gate the emit on the returned path or carry the outcome in the event. Raised by 5 of 8 reviewers (gpt-5.6-sol-max adversarial, gpt-5.6-sol-max edge-case, claude-opus-5-thinking-max both types, gemini-3.1-pro edge-case, kimi-k2.7-code edge-case).

@mattmillerai mattmillerai Aug 3, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed — the outcome is now carried in the event.

Agreed the docs guarantee was overstated. Of your two options I took carry the outcome rather than gate the emit: gating would mean a failed state write also costs the caller the prompt_id of a job the server has already accepted (and on cloud already billed), which is the larger loss and the one this reordering exists to prevent.

So queued gains a nullable state_file field — the path, or null when _write_state skipped/failed. Also updated:

  • comfy_cli/schemas/run_event.json declares it as ["string", "null"], so a consumer validating the stream doesn't reject the very case the field signals.
  • docs/json-output.md now says the write is best-effort and tells consumers to key off state_file rather than treat the presence of the queued line as proof the record is there.
  • Pretty mode prints not written instead of a bare None where the path used to go.

Pinned by TestQueuedReportsWhetherTheRecordLanded (happy path + OSError/ValueError failures, local and cloud) and test_queued_declares_a_nullable_state_file. Verified they fail when the field is dropped from the emit.

"""Emit the contractual local `queued` event (docs/json-output.md#queued).

Lives in the caller rather than ``WorkflowExecution.queue()`` so it can be
emitted *after* the job's state file is persisted — a consumer that sees
this line can immediately `comfy jobs status <prompt_id>` / see the job in
`jobs ls`, and a `BrokenPipeError` here can no longer abort setup before
the durable record exists. The field set is unchanged.
"""
renderer.event(
"queued",
prompt_id=execution.prompt_id,
client_id=execution.client_id,
validation_warnings=execution.validation_warnings,
nodes=execution.workflow_manifest(),
)


def _write_state(state):
"""Best-effort ``jobs_state.write``. Returns the path, or None if the
write was skipped, the state dir was unwritable, or the prompt_id was
Expand Down Expand Up @@ -860,7 +894,8 @@ def execute_cloud(
# with `queued` (async) / `executing` (--wait) carrying {workflow, base_url}
# — both wrong per docs/json-output.md: `queued` means "the server accepted
# the prompt" (so it cannot precede the POST), and `executing` is a per-node
# event. The contractual `queued` is emitted after submit, below.
# event. The contractual `queued` is emitted after submit *and* after the
# job state file is written, below.

try:
if not wait and renderer.is_pretty():
Expand Down Expand Up @@ -915,19 +950,27 @@ def execute_cloud(
raise typer.Exit(code=1)

# The contractual `queued`: the server has the prompt. Same shape the local
# path emits from `WorkflowExecution.queue()` (docs/json-output.md#queued),
# plus `base_url` so a cloud consumer still learns the target.
# `validation_warnings` is always empty here — unlike local, the cloud
# treats any `node_errors` on an accepted submit as a hard `prompt_rejected`
# above, so a partially-valid graph never reaches this line.
renderer.event(
"queued",
prompt_id=submit.prompt_id,
client_id=client_id,
validation_warnings=[],
nodes=workflow_manifest(parsed_workflow),
base_url=target.base_url,
)
# path emits (docs/json-output.md#queued), plus `base_url` so a cloud
# consumer still learns the target. `validation_warnings` is always empty
# here — unlike local, the cloud treats any `node_errors` on an accepted
# submit as a hard `prompt_rejected` above, so a partially-valid graph
# never reaches this line.
#
# Deliberately NOT emitted at this point: both branches below emit it only
# once the job's state file (and, on the async branch, the watcher) exist.
# A `BrokenPipeError` here — `comfy run --where cloud --json-stream … |
# head -n1`, where the NDJSON renderer flushes every line — used to abort
# setup before `jobs_state.write` ran while `__main__` still exited 0,
# leaving a billable cloud job with no journal entry, state file or watcher.
def _emit_queued_cloud() -> None:
renderer.event(
"queued",
prompt_id=submit.prompt_id,
client_id=client_id,
validation_warnings=[],
nodes=workflow_manifest(parsed_workflow),
base_url=target.base_url,
)

if not wait:
state = jobs_state.new(
Expand All @@ -942,6 +985,9 @@ def execute_cloud(
_journal_run(workflow_name, submit.prompt_id, "cloud")
watcher_spawned = _spawn_watcher(submit.prompt_id, where="cloud", notify=notify)

# Durable record + watcher exist: safe to announce.
_emit_queued_cloud()

@github-actions github-actions Bot Aug 2, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟠 High — Same hazard on both cloud branches: jobs_state.write(state) at lines 984 and 1037 is unguarded and execute_cloud catches neither ValueError (prompt_id rejected by state_path) nor OSError, so a storage failure after the billable submit kills the run with a traceback, no envelope, and no queued line — the caller never learns the prompt_id of a job it is already paying for, which is exactly the loss this reordering is meant to prevent. Route both through a best-effort wrapper like _write_state. Raised by 3 of 8 reviewers (claude-opus-5-thinking-max edge-case, gpt-5.6-sol-max edge-case, kimi-k2.7-code edge-case).

@mattmillerai mattmillerai Aug 3, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed — both cloud submit-time writes now go through the best-effort _write_state.

Agreed on the framing: this is the loss the reordering exists to prevent, and doing it after a billable submit is the worst place for it. execute_cloud catches neither ValueError (from state_path() rejecting the id) nor OSError, so either killed the run with a traceback and no queued line.

Pinned by test_cloud_survives_a_failed_write_after_the_billable_submit, parametrized over both exception types: asserts exit 0, prompt_id still emitted, state_file: null. Verified it fails when the async-branch call is reverted to jobs_state.write.


if renderer.is_pretty():
from comfy_cli.output.glyphs import status_glyph

Expand Down Expand Up @@ -991,6 +1037,12 @@ def execute_cloud(
state_file = jobs_state.write(state)
_journal_run(workflow_name, submit.prompt_id, "cloud")

# Announced once the record is durable, and *outside* the polling try below
# — none of its handlers should ever see (and rewrite state for) a
# BrokenPipeError raised by this emit. It propagates to click/`__main__`,
# which exits 0 silently, with the job already recorded.
_emit_queued_cloud()

@github-actions github-actions Bot Aug 2, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 Medium — Placing _emit_queued_cloud() outside the polling try means an EPIPE here exits 0 while the cloud job keeps running and billing, with the record frozen at queued. Unlike the async branch this path never spawns a watcher, so nothing ever advances or reaps that record and jobs status/wait/ls report a permanently queued job. Raised by 3 of 8 reviewers (claude-opus-5-thinking-max adversarial, gemini-3.1-pro adversarial, gemini-3.1-pro edge-case).

@mattmillerai mattmillerai Aug 3, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed — same hand-off as the local --wait path.

You're right that this branch is the worst case for it: it spawns no watcher by design, so nothing would ever advance or reap the record, and the job keeps running and billing. The emit stays outside the polling try (so none of those handlers can rewrite state for a stdout failure), but it's now wrapped in its own narrow handler:

try:
    _emit_queued_cloud(state_file)
except OSError:
    _hand_off_record(state, where="cloud", notify=notify)
    raise

OSError rather than BrokenPipeError so an ENOSPC/EIO on a redirected stdout gets the hand-off too; the re-raise is unchanged, so __main__ still turns a broken pipe into the documented silent exit 0 while a genuine lost-output error surfaces.

Pinned by test_cloud_wait_hands_off_the_queued_record, which fails when the hand-off is removed.


try:

def _probe():
Expand Down
19 changes: 10 additions & 9 deletions comfy_cli/command/run/execution.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -163,6 +163,12 @@ def __init__(
# compute nodes that never fire a server-side `executed` event.
self.cached_node_ids: list[str] = []
self.executed_node_ids: list[str] = []
# Per-node issues the server reported alongside a successful (HTTP 200)
# queue, stashed by ``queue()`` for the caller's `queued` event. The
# event itself is emitted by ``comfy_cli.command.run.execute`` — only
# *after* the job's state file is persisted — so a consumer sees it and
# can immediately rely on `comfy jobs status` / `jobs ls` finding it.
self.validation_warnings: list[dict] = []
# Classified verdict of a terminal server-side `execution_error`,
# stashed by `on_error` before it raises `typer.Exit`. The `--wait`
# caller only sees the exit, so this is how the real cause reaches
Expand Down Expand Up @@ -289,16 +295,11 @@ def queue(self):

# 200 may still carry node_errors if some output chains failed
# validation but others passed — surface as warnings, not a failure.
# Stored rather than emitted here: the contractual `queued` event is
# emitted by the caller once the job state file exists (see
# ``validation_warnings`` above and ``run._emit_queued``).
node_errors = body.get("node_errors") if isinstance(body, dict) else None
validation_warnings = _node_errors_to_list(node_errors)

self.renderer.event(
"queued",
prompt_id=prompt_id,
client_id=self.client_id,
validation_warnings=validation_warnings,
nodes=self.workflow_manifest(),
)
self.validation_warnings = _node_errors_to_list(node_errors)

def watch_execution(self):
if self.ws is None:
Expand Down
11 changes: 8 additions & 3 deletions docs/json-output.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -238,7 +238,8 @@ final envelope and exits 0 without queuing.
### `queued`

Emitted after the submit request returns success — `POST /prompt` returning
200 locally, the equivalent cloud submit under `--where cloud`.
200 locally, the equivalent cloud submit under `--where cloud` — **and** after
the job's state file has been persisted.

```json
{
Expand All @@ -262,8 +263,12 @@ Emitted after the submit request returns success — `POST /prompt` returning
| `nodes` | array of dict | Manifest of every node in the submitted (post-conversion) workflow: `node_id` (str), `class_type` (str), `title` (str). Lets piped consumers render a per-node UI without the workflow file. |
| `base_url` | str | **Cloud only.** The cloud endpoint the prompt was submitted to. Absent on `--where local`. |

`queued` is emitted **after** the submit call returns successfully, on both
targets — it means the server has the prompt. A run whose submit fails emits
`queued` is emitted **after** the submit call returns successfully **and after
the job's state file has been persisted** — on the async (`--no-wait`) paths,
after the background watcher spawn attempt too. On both targets it therefore
means more than "the server has the prompt": the durable record exists, so a
consumer may run `comfy jobs status <prompt_id>` — or expect the job in
`comfy jobs ls` — the moment it reads this line. A run whose submit fails emits
its error envelope with no `queued` line at all.

### `executing`
Expand Down
Loading
Loading