This happens only when TelemetryManager.attach_backend() is active.
Environment
- RHAPSODY: ConcurrentExecutionBackend, TelemetryManager
- AsyncFlow: WorkflowEngine.create(backend) + engine.start_telemetry()
- Trigger: concurrent function_task spawning (1–4 tasks per iteration, many iterations)
Reproduction
backend = await ConcurrentExecutionBackend()
engine = await WorkflowEngine.create(backend)
telemetry = await engine.start_telemetry(resource_poll_interval=1.0)
for x in batch:
fut = engine.function_task(my_impl)(x=x)
fut.add_done_callback(lambda f, uid=uid: handle_result(f, uid))Run for enough iterations to observe the warning (non-deterministic, typically within 50–100 tasks):
WARNING Attempted to handle an already finished task "task.000039"
WARNING Attempted to handle an already finished task "task.000052"
Does not reproduce when start_telemetry() is omitted.
Root Cause
WorkflowEngine._attach_backend() calls:
backend.register_callback(engine.task_callbacks) # registration 1
TelemetryManager.attach_backend(), called later by engine.start_telemetry(), calls register_callback on the same backend instance a second time. ConcurrentExecutionBackend maintains a list of callbacks and appends rather than replaces, so after start_telemetry() there are two independent listeners on the backend.
When a task reaches a terminal state, the backend dispatches to both. The telemetry adapter's dispatch path re-routes the same completion event back through engine.task_callbacks, resulting in two DONE deliveries for the same task UID. AsyncFlow's handle_task_success guard (if not task_fut.done()) prevents double-resolution of the future, but logs a warning on the second delivery.
Expected Behaviour
Each task UID receives exactly one callback invocation per state transition, regardless of how many listeners are registered on the backend.
Actual Behaviour
With start_telemetry() active, engine.task_callbacks is invoked twice with DONE for the same UID. The second invocation hits the already-resolved future and emits a warning that is indistinguishable in logs from a genuine state-conflict warning.
Impact
Log pollution — spurious warnings appear at a rate proportional to task throughput, making it difficult to distinguish them from real state conflicts (e.g. a DONE event arriving for a CANCELLED future).
Masked diagnostics — AsyncFlow's handle_task_success warning is the only signal for both benign duplicates and genuine backend misbehaviour. The duplicate delivery makes that signal unreliable.
No data corruption — futures are not double-resolved; the guard in handle_task_success holds.
Suggested Fix
ConcurrentTelemetryAdapter (or TelemetryManager.attach_backend) should intercept the completion event, record telemetry, and forward once — either by wrapping the existing callback rather than appending a second independent one, or by deduplicating at the ConcurrentExecutionBackend dispatch layer: once a terminal state has been dispatched for a given UID, further deliveries of the same state are dropped before reaching any registered listener.
This happens only when
TelemetryManager.attach_backend()is active.Environment
Reproduction
Run for enough iterations to observe the warning (non-deterministic, typically within 50–100 tasks):
Does not reproduce when start_telemetry() is omitted.
Root Cause
WorkflowEngine._attach_backend()calls:backend.register_callback(engine.task_callbacks)# registration 1TelemetryManager.attach_backend(), called later byengine.start_telemetry(), calls register_callback on the same backend instance a second time.ConcurrentExecutionBackendmaintains a list of callbacks and appends rather than replaces, so afterstart_telemetry()there are two independent listeners on the backend.When a task reaches a terminal state, the backend dispatches to both. The telemetry adapter's dispatch path re-routes the same completion event back through engine.task_callbacks, resulting in two
DONEdeliveries for the same task UID. AsyncFlow'shandle_task_successguard (if not task_fut.done()) prevents double-resolution of the future, but logs a warning on the second delivery.Expected Behaviour
Each task UID receives exactly one callback invocation per state transition, regardless of how many listeners are registered on the backend.
Actual Behaviour
With start_telemetry() active, engine.task_callbacks is invoked twice with DONE for the same UID. The second invocation hits the already-resolved future and emits a warning that is indistinguishable in logs from a genuine state-conflict warning.
Impact
Log pollution — spurious warnings appear at a rate proportional to task throughput, making it difficult to distinguish them from real state conflicts (e.g. a DONE event arriving for a CANCELLED future).
Masked diagnostics — AsyncFlow's handle_task_success warning is the only signal for both benign duplicates and genuine backend misbehaviour. The duplicate delivery makes that signal unreliable.
No data corruption — futures are not double-resolved; the guard in handle_task_success holds.
Suggested Fix
ConcurrentTelemetryAdapter (or TelemetryManager.attach_backend) should intercept the completion event, record telemetry, and forward once — either by wrapping the existing callback rather than appending a second independent one, or by deduplicating at the ConcurrentExecutionBackend dispatch layer: once a terminal state has been dispatched for a given UID, further deliveries of the same state are dropped before reaching any registered listener.