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

Fix IPFS stream caching - return live playlist CID from database

Browse Source 
- Return ipfs_playlist_cid from pending_runs while task is running
- Add Cache-Control: no-cache headers to prevent browser/CDN caching
- Fix streaming clients getting stale playlist CIDs

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
giles
2026-02-04 12:32:56 +00:00
parent cd95e62899
commit baf79f453f
1 changed files with 53 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

68
app/routers/runs.py
View File

@@ -1152,6 +1152,13 @@ async def get_ipfs_stream_info(run_id: str, request: Request):
# Try to get pending run to find the Celery task ID # Try to get pending run to find the Celery task ID
pending = await database.get_pending_run(run_id) pending = await database.get_pending_run(run_id)
from fastapi.responses import JSONResponse
no_cache_headers = {
"Cache-Control": "no-cache, no-store, must-revalidate",
"Pragma": "no-cache",
"Expires": "0"
}
if not pending: if not pending:
# Try completed runs # Try completed runs
run = await database.get_run_cache(run_id) run = await database.get_run_cache(run_id)
@@ -1161,11 +1168,14 @@ async def get_ipfs_stream_info(run_id: str, request: Request):
ipfs_cid = run.get("ipfs_cid") ipfs_cid = run.get("ipfs_cid")
if ipfs_cid: if ipfs_cid:
gateway = os.environ.get("IPFS_GATEWAY_URL", "https://ipfs.io/ipfs") gateway = os.environ.get("IPFS_GATEWAY_URL", "https://ipfs.io/ipfs")
return { return JSONResponse(
"run_id": run_id, content={
"status": "completed", "run_id": run_id,
"ipfs_video_url": f"{gateway}/{ipfs_cid}", "status": "completed",
} "ipfs_video_url": f"{gateway}/{ipfs_cid}",
},
headers=no_cache_headers
)
raise HTTPException(404, "No IPFS stream info available") raise HTTPException(404, "No IPFS stream info available")
task_id = pending.get("celery_task_id") task_id = pending.get("celery_task_id")
@@ -1175,6 +1185,14 @@ async def get_ipfs_stream_info(run_id: str, request: Request):
# Get the Celery task result # Get the Celery task result
result = celery_app.AsyncResult(task_id) result = celery_app.AsyncResult(task_id)
from fastapi.responses import JSONResponse
gateway = os.environ.get("IPFS_GATEWAY_URL", "https://ipfs.io/ipfs")
no_cache_headers = {
"Cache-Control": "no-cache, no-store, must-revalidate",
"Pragma": "no-cache",
"Expires": "0"
}
if result.ready(): if result.ready():
# Task is complete - check the result for IPFS playlist info # Task is complete - check the result for IPFS playlist info
task_result = result.result task_result = result.result
@@ -1182,17 +1200,37 @@ async def get_ipfs_stream_info(run_id: str, request: Request):
ipfs_playlist_cid = task_result.get("ipfs_playlist_cid") ipfs_playlist_cid = task_result.get("ipfs_playlist_cid")
ipfs_playlist_url = task_result.get("ipfs_playlist_url") ipfs_playlist_url = task_result.get("ipfs_playlist_url")
if ipfs_playlist_url: if ipfs_playlist_url:
return { return JSONResponse(
"run_id": run_id, content={
"status": "completed", "run_id": run_id,
"ipfs_playlist_cid": ipfs_playlist_cid, "status": "completed",
"ipfs_playlist_url": ipfs_playlist_url, "ipfs_playlist_cid": ipfs_playlist_cid,
"segment_count": task_result.get("ipfs_segment_count", 0), "ipfs_playlist_url": ipfs_playlist_url,
} "segment_count": task_result.get("ipfs_segment_count", 0),
},
headers=no_cache_headers
)
# Task is still running or no IPFS info available # Task is still running - check database for live playlist updates
return { ipfs_playlist_cid = pending.get("ipfs_playlist_cid")
response_data = {
"run_id": run_id, "run_id": run_id,
"status": pending.get("status", "pending"), "status": pending.get("status", "pending"),
"message": "IPFS streaming info not yet available"
} }
if ipfs_playlist_cid:
response_data["ipfs_playlist_cid"] = ipfs_playlist_cid
response_data["ipfs_playlist_url"] = f"{gateway}/{ipfs_playlist_cid}"
else:
response_data["message"] = "IPFS streaming info not yet available"
# No caching for live streaming data
return JSONResponse(
content=response_data,
headers={
"Cache-Control": "no-cache, no-store, must-revalidate",
"Pragma": "no-cache",
"Expires": "0"
}
)