submit persisted the full request body on the durable job row for faithful
replay, but request-level globals can carry inline secrets (API keys), landing
plaintext in the job table (JSONB on Postgres) and widening the blast radius of
any DB read. Redact globals from the persisted copy (the live in-memory run
still uses them). Tradeoff: a background re-enqueue and a scaled worker run drop
inline globals, so reference stored global variables by name for background runs
rather than passing secrets inline.
stop() set a redis cancel marker and PUBLISHed on a cancel channel as a fast
path, but the background worker (run_worker_loop -> WorkerJobRunner -> JobRunner)
never subscribes to that channel or checks the marker — only the v1
RedisJobQueueService dispatcher does, which the worker does not run. So the
marker + PUBLISH were no-ops in production, a misleading dead fast-path. Remove
them: the durable STOP signal polled at each vertex boundary is the single
mechanism. Add a real-redis latency test proving scaled stop is bounded by the
durable poll cadence, and drop the v1-dispatcher fast-path test that gave false
confidence.
The first _closed bound evicted the marker on last-subscriber drop and skipped
marking when no subscriber existed, which broke a direct reattach to a finished
job that closed with no subscriber (it blocked on a tail that never produces).
Always record close() and bound the map with an LRU cap instead, so the
standalone reattach-terminates contract holds while a long-lived process still
cannot leak one key per job.
RedisStreamLiveBus called expire() on every XADD, doubling redis round-trips per
token frame and capping single-job streaming throughput. Mirror the v1 Streams
bridge: refresh the TTL on the first frame, then every 100 frames or 30s, and
always on close, preserving the TTL semantics at ~1/100 the EXPIRE rate.
In scaled mode the facade built a redis client for the backend but teardown()
only stopped the executor, leaking the API replica's background-execution
connection pool on shutdown. Add a backend teardown() that closes its client and
call it from the facade teardown (the worker process already closes its own).
close() set _closed[job_id]=True and nothing ever evicted it, so a long-lived
API process leaked one key per completed job. The marker only needs to outlive
close() until existing subscribers drain (a late reattach to a finished job is
gated on the persisted JobStatus by the facade, not this in-process flag), so
set it only when subscribers exist and evict it with the last subscriber drop.
_row_to_frame serialized durable payloads with json.dumps default (spaced)
separators, but the agui live path uses pydantic model_dump_json (compact
separators), so replayed agui frames were not byte-identical to live frames.
Pass the run's stream protocol (read off the persisted submit request) into
_row_to_frame and pick compact separators for agui, spaced for langflow, so
replay == live byte-for-byte on both wires.
The scaled events() replayed durable job_events from the DB then tailed the
redis Stream from 0-0 with no seq filter, so every milestone the worker wrote to
BOTH the DB and the Stream was delivered twice on reattach. Track the highest
durable seq replayed and skip any Stream frame whose worker-stamped seq is
<= highest, mirroring the in-memory bus's dedup-at-the-seam rule, so each
milestone is delivered exactly once on both paths.
The live SSE stream baked its own per-frame stream-seq counter into the id:
line while durable replay used job_events.seq, two different namespaces. A
client reconnecting with a live Last-Event-ID resumed against the wrong cursor
and silently missed or duplicated milestones. The runner now re-stamps every
published frame's id: with its durable seq (the row seq for a milestone, the
last milestone's seq for an ephemeral token) so live ids and durable replay ids
share one cursor and a mid-run reattach resumes exactly after the durable seq.
Drain the claim queue after a concurrent-reconcile race and assert the lost job
is claimable exactly once (a side-effect-style run counter), so a draining worker
re-runs it at most once more, never twice.
The startup re-enqueue flipped a QUEUED row to IN_PROGRESS at claim time, so a
crash before the runner started left a stranded IN_PROGRESS the next sweep failed
worker_lost -- a job that never ran ended FAILED. claim_queued_lease stamps a
single-flight owner+heartbeat WITHOUT changing status, so the row stays QUEUED
(re-runnable) and the runner's execute_with_status performs the real
QUEUED->IN_PROGRESS flip only when it actually starts.
execute_with_status writes the TIMED_OUT/CANCELLED status but no durable error
blob or terminal job_events row, breaking the design invariant that every
terminal path writes result/error + a terminal event. The runner now backfills
run_timed_out (+error) / run_cancelled in its finally. A late stop that wins over
a racing completion clears the completed-run result and sets error=cancelled so a
CANCELLED row never carries a completed result.
requeue_lost only reconciles processing-list ids whose lease is stale/absent, so
a booting/scaled-up worker can no longer re-claim and double-run (or fail) a job
a live worker is mid-running. A periodic watchdog runs requeue_lost on an interval
inside run_worker_loop so a dead worker's in-flight job is reaped under a steady
fleet WITHOUT a restart. The worker stamps a heartbeat on claim and the in-flight
runner keeps it fresh. Retry-safe attempt accounting is now an atomic conditional
increment (increment_attempt_if) so concurrent reconcilers cannot exceed
max_attempts; the LREM count is the single-flight token so two watchdogs cannot
double-enqueue. recover_stranded_queued re-enqueues QUEUED rows present on neither
redis list (API-crash window) so a job is never stuck QUEUED forever.
sweep_orphans only fails IN_PROGRESS rows whose heartbeat is stale/absent, so a
booting worker can no longer flip a sibling's actively-running, freshly-
heartbeated job FAILED(worker_lost) under gunicorn -w N. The runner now refreshes
a job-row heartbeat (owner+ts) on an interval while in flight. The startup sweep
is single-flighted with a FileLock (the starter-projects pattern), and its
terminal-event insert reuses append_event so a seq collision can't roll back the
whole sweep.
create_job's dedupe count was global, so a client-controlled idempotency_key
let user A collide with / DoS user B's key and the 500 leaked its existence.
Scope the count by user_id (ownerless rows keep a global floor for AUTO_LOGIN),
matching the already-user-scoped lookup. Defense-in-depth: map DuplicateJobError
to a 409 with a generic body in execute_workflow_background so the residual
create/lookup race cannot 500-leak the key.
Lease+heartbeat liveness in job_metadata (no new column): heartbeat() stamps
owner+timestamp via shallow merge; is_lease_stale() tells a live in-flight run
from a genuinely orphaned one; increment_attempt_if() is an atomic conditional
bump so concurrent reconcilers cannot push a job past max_attempts.
Submits POST /api/v2/workflows with mode=background, polls
GET /api/v2/workflows?job_id=... to a terminal status, and records a
success only when the job reaches COMPLETED. Tracks submitted vs
completed counts in the quitting summary so the run can be cross-checked
against the job table for lost or stuck jobs.
The runner read unconsumed_signals (a DB query) on every frame including each
ephemeral token, so stop polling scaled with the token stream. A cooperative
stop is only honored at vertex/milestone boundaries, so poll only on durable
frames; the post-loop check still catches a STOP that lands after the last
frame.
The v2 stop handler called update_job_status(CANCELLED) unconditionally, so
stopping an already-COMPLETED/FAILED/TIMED_OUT job flipped it to CANCELLED while
its result/error blob stayed put. Return the existing terminal state instead of
transitioning when the row is already finished.
stop() cancelled in-flight tasks but returned without awaiting them, so a job's
shielded terminal reconcile (a DB write) could race a closing engine and leave a
'Task was destroyed but it is pending' warning. Cancel and gather the in-flight
job tasks first, then tear down the workers. Cancelling the job tasks directly
(rather than relying on cancellation propagating through the worker's await)
avoids the worker sitting in cancellation limbo behind a job that swallows its
cancel on the user-stop path, which otherwise stalls teardown.
unconsumed_signals filters consumed_at IS NULL but nothing stamped it, so STOP
rows lingered: the table grew and a re-enqueued job self-cancelled off a stale
STOP. Add JobService.consume_signals and stamp the STOP in the runner's
_reconcile_stop, the single point where a stop is finalized.
Two uvicorn workers booting against the same DB both re-enqueued the same QUEUED
row, so a non-idempotent flow executed twice. Add JobService.claim_queued_job, a
conditional UPDATE ... WHERE status='QUEUED' that only one racer can win
(rowcount==1), and gate sweep re-enqueue on it. Works on SQLite and Postgres.
The background_job_timeout setting was read by nobody, so a runaway run never
timed out. Bound the runner's drive with asyncio.wait_for(timeout=...) when the
setting is configured; execute_with_status already maps asyncio.TimeoutError to
TIMED_OUT, so the run ends TIMED_OUT with a terminal event. None keeps the prior
unbounded behaviour. The facade passes settings.background_job_timeout through.
events() decided 'is the job finished' from the process-local live bus, whose
_closed marker is empty after a restart. A reattach to an already-terminal job
replayed durable rows then blocked forever on the live tail. Key the decision off
the persisted job status instead: when terminal, replay the durable log and
return. Re-frame durable rows through the same format_sse_event formatter the
live path uses (with id=str(seq)) so replayed and live frames are byte-compatible
and Last-Event-ID resumes correctly.
The worker used asyncio.Task.cancelling() to tell a job-cancel apart from a
worker-cancel, but that API is Python 3.11+. On 3.10 the first job cancellation
raised AttributeError inside the cancel handler and permanently killed the
worker, hanging later jobs QUEUED. Replace it with an explicit _closed flag set
by stop(): a CancelledError while shutting down re-raises to exit the worker,
otherwise it is a per-job cancel that is swallowed so the pool keeps serving.
