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

Store cache items by IPFS CID, index by cache_id

Browse Source 
- Files in /data/cache/nodes/ are now stored by IPFS CID only
- cache_id parameter creates index from cache_id -> IPFS CID
- Removed deprecated node_id parameter behavior
- get_by_cid(cache_id) still works via index lookup

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
gilesb
2026-01-13 04:20:34 +00:00
parent c46fcd2308
commit d7d7cd28c2
2 changed files with 28 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
legacy_tasks.py
View File

@@ -346,21 +346,16 @@ def execute_dag(self, dag_json: str, run_id: str = None) -> dict:
else:
cache_node_type = "dag_intermediate"
# Store in cache_manager (indexes by cid, uploads to IPFS)
# put() returns (CachedFile, cid) where cid is IPFS CID if available, else local hash
# Store in cache_manager (stored by IPFS CID, indexed by node_id)
cached, content_cid = cache_manager.put(
Path(node_path),
node_type=cache_node_type,
node_id=node_id,
cache_id=node_id,
)
# content_cid is the primary identifier (IPFS CID or local hash)
# content_cid is always IPFS CID now (IPFS failures are fatal)
node_hashes[node_id] = content_cid
# Track IPFS CIDs separately (they start with Qm or bafy)
if content_cid and (content_cid.startswith("Qm") or content_cid.startswith("bafy")):
node_ipfs_cids[node_id] = content_cid
logger.info(f"Cached node {node_id}: IPFS CID {content_cid}")
else:
logger.info(f"Cached node {node_id}: local hash {content_cid[:16] if content_cid else 'none'}...")
node_ipfs_cids[node_id] = content_cid
logger.info(f"Cached node {node_id}: IPFS CID {content_cid}")
# Get output hash from the output node
# Use the same identifier that's in the cache index (IPFS CID if available)
@@ -840,11 +835,11 @@ def execute_recipe(self, recipe_sexp: str, input_hashes: Dict[str, str], run_id:
import shutil
shutil.copy2(current_input, final_output)
# Upload to IPFS
# Upload to IPFS (stored by IPFS CID, indexed by cache_id)
cached, content_cid = cache_manager.put(
final_output,
node_type="COMPOUND",
node_id=step.cache_id,
cache_id=step.cache_id,
)
# Cleanup temp files
@@ -910,11 +905,11 @@ def execute_recipe(self, recipe_sexp: str, input_hashes: Dict[str, str], run_id:
if result.returncode != 0:
raise RuntimeError(f"FFmpeg concat failed: {result.stderr}")
# Upload to IPFS
# Upload to IPFS (stored by IPFS CID, indexed by cache_id)
cached, content_cid = cache_manager.put(
final_output,
node_type="SEQUENCE",
node_id=step.cache_id,
cache_id=step.cache_id,
)
# Cleanup
@@ -963,7 +958,7 @@ def execute_recipe(self, recipe_sexp: str, input_hashes: Dict[str, str], run_id:
cached, content_cid = cache_manager.put(
result_path,
node_type="EFFECT",
node_id=step.cache_id,
cache_id=step.cache_id,
)
step_results[step.step_id] = {
@@ -995,11 +990,11 @@ def execute_recipe(self, recipe_sexp: str, input_hashes: Dict[str, str], run_id:
logger.info(f"Executing step {step.step_id} ({step.node_type}) with {len(input_paths)} inputs")
result_path = executor.execute(step.config, input_paths, output_path)
# Store result in cache under code-addressed cache_id
# Store result in cache (by IPFS CID, indexed by cache_id)
cached, content_cid = cache_manager.put(
result_path,
node_type=step.node_type,
node_id=step.cache_id, # Use cache_id as node_id
cache_id=step.cache_id,
)
step_results[step.step_id] = {