art-dag/celery
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Get actor_id from pending_runs when saving completed run

Browse Source 
When a DAG task completes, look up actor_id from pending_runs
(where it was saved when the run started) and include it in
run_cache. Also clean up pending_runs entry after completion.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
gilesb
2026-01-11 21:43:44 +00:00
parent 59de1cf6b5
commit eb1de433b1
1 changed files with 10 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
legacy_tasks.py
View File

@@ -383,13 +383,23 @@ def execute_dag(self, dag_json: str, run_id: str = None) -> dict:
if (node.node_type == NodeType.SOURCE or str(node.node_type) == "SOURCE") if (node.node_type == NodeType.SOURCE or str(node.node_type) == "SOURCE")
and node.config.get("content_hash") and node.config.get("content_hash")
] ]
# Get actor_id from pending_runs (saved when run started)
actor_id = None
pending = await database.get_pending_run(run_id)
if pending:
actor_id = pending.get("actor_id")
await database.save_run_cache( await database.save_run_cache(
run_id=run_id, run_id=run_id,
output_hash=output_hash, output_hash=output_hash,
recipe="dag", recipe="dag",
inputs=input_hashes_for_db, inputs=input_hashes_for_db,
ipfs_cid=output_ipfs_cid, ipfs_cid=output_ipfs_cid,
actor_id=actor_id,
) )
# Clean up pending run
if pending:
await database.complete_pending_run(run_id)
try: try:
loop = asyncio.get_event_loop() loop = asyncio.get_event_loop()