JobOrchestrator turns one user action into N command messages — one per source, all
sharing a message_id — and then reassembles the N acknowledgements by counting them.
Both producers do it:
commit_workflow builds one WorkflowConfig per staged source and registers
expected_count=len(state.staged_jobs)._send_job_commands (stop/reset) builds one JobCommand per JobId and registers
expected_count=len(state.current.jobs).
PendingCommandTracker exists for that accumulation: expected_count, success_count,
error_count, is_complete, and expiry for the responses that never arrive. A worker
handles its sources in a synchronous loop and knows the outcome for all of them at once,
so the fan-out buys nothing — it splits one synchronous result into N Kafka round-trips
and N acks, then spends dashboard state putting them back together.
One command carrying its source set, acknowledged once per worker, removes the
accumulation. In the single-worker deployment we run today that is one ack per command,
and the tracker collapses to a timeout. Multi-worker needs the ack to name the sources
it covers, so the client tracks coverage instead of a count.
What each command needs:
WorkflowConfig would carry the per-source config set (dict[SourceName, JobConfig]),
which is what WorkflowState.staged_jobs already holds.- Stop and reset already have selector targeting:
JobCommand supports workflow_id
(what NICOS uses for reset, ADR 0006) and broadcast, and JobManager._perform_action
returns the affected count. The dashboard is the only producer still enumerating
JobIds.
To settle as part of it: whether partial failure reports per-source detail or just fails
the command, and whether stop keeps per-source targeting at all — ADR 0006 rejected a
per-source reset selector as unneeded, but stop_workflow's orphan sweep currently
leans on per-JobId stops.
Extracted from #445. This is the part of that design that stands on its own: it needs
the config message to carry its source set, but not the config/start split, whose
premise ADR 0006 settled the other way.
JobOrchestratorturns one user action into N command messages — one per source, allsharing a
message_id— and then reassembles the N acknowledgements by counting them.Both producers do it:
commit_workflowbuilds oneWorkflowConfigper staged source and registersexpected_count=len(state.staged_jobs)._send_job_commands(stop/reset) builds oneJobCommandperJobIdand registersexpected_count=len(state.current.jobs).PendingCommandTrackerexists for that accumulation:expected_count,success_count,error_count,is_complete, and expiry for the responses that never arrive. A workerhandles its sources in a synchronous loop and knows the outcome for all of them at once,
so the fan-out buys nothing — it splits one synchronous result into N Kafka round-trips
and N acks, then spends dashboard state putting them back together.
One command carrying its source set, acknowledged once per worker, removes the
accumulation. In the single-worker deployment we run today that is one ack per command,
and the tracker collapses to a timeout. Multi-worker needs the ack to name the sources
it covers, so the client tracks coverage instead of a count.
What each command needs:
WorkflowConfigwould carry the per-source config set (dict[SourceName, JobConfig]),which is what
WorkflowState.staged_jobsalready holds.JobCommandsupportsworkflow_id(what NICOS uses for reset, ADR 0006) and broadcast, and
JobManager._perform_actionreturns the affected count. The dashboard is the only producer still enumerating
JobIds.To settle as part of it: whether partial failure reports per-source detail or just fails
the command, and whether stop keeps per-source targeting at all — ADR 0006 rejected a
per-source reset selector as unneeded, but
stop_workflow's orphan sweep currentlyleans on per-
JobIdstops.Extracted from #445. This is the part of that design that stands on its own: it needs
the config message to carry its source set, but not the config/start split, whose
premise ADR 0006 settled the other way.