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

Make IPFS pinning non-blocking (fire-and-forget)

Browse Source 
IPFS pinning can take a long time if content needs to be fetched
from the network. Changed all pin operations to run in background
threads so they don't block the HTTP response.

This fixes the 30s timeout issue when publishing assets.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
gilesb
2026-01-09 00:26:32 +00:00
parent a7dfdc8a39
commit a0ed1ae5ae
1 changed files with 20 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
server.py
View File

@@ -1599,13 +1599,9 @@ async def update_asset(name: str, req: UpdateAssetRequest, user: User = Depends(
updates["origin"] = req.origin updates["origin"] = req.origin
if req.ipfs_cid is not None: if req.ipfs_cid is not None:
updates["ipfs_cid"] = req.ipfs_cid updates["ipfs_cid"] = req.ipfs_cid
# Pin on IPFS # Pin on IPFS (fire-and-forget, don't block)
try: import threading
import ipfs_client threading.Thread(target=_pin_ipfs_async, args=(req.ipfs_cid,), daemon=True).start()
if ipfs_client.is_available():
ipfs_client.pin(req.ipfs_cid)
except Exception as e:
logger.warning(f"Failed to pin IPFS content {req.ipfs_cid}: {e}")
# Update asset in database # Update asset in database
updated_asset = await db.update_asset(name, updates) updated_asset = await db.update_asset(name, updates)
@@ -1639,20 +1635,27 @@ async def update_asset(name: str, req: UpdateAssetRequest, user: User = Depends(
return {"asset": updated_asset, "activity": activity} return {"asset": updated_asset, "activity": activity}
def _pin_ipfs_async(cid: str):
"""Pin IPFS content in background thread."""
try:
import ipfs_client
if ipfs_client.is_available():
ipfs_client.pin(cid)
logger.info(f"Pinned IPFS content: {cid}")
except Exception as e:
logger.warning(f"Failed to pin IPFS content {cid}: {e}")
async def _register_asset_impl(req: RegisterRequest, owner: str): async def _register_asset_impl(req: RegisterRequest, owner: str):
"""Internal implementation for registering an asset.""" """Internal implementation for registering an asset."""
# Check if name exists # Check if name exists
if await db.asset_exists(req.name): if await db.asset_exists(req.name):
raise HTTPException(400, f"Asset already exists: {req.name}") raise HTTPException(400, f"Asset already exists: {req.name}")
# Pin content on IPFS if CID provided # Pin content on IPFS if CID provided (fire-and-forget, don't block)
if req.ipfs_cid: if req.ipfs_cid:
try: import threading
import ipfs_client threading.Thread(target=_pin_ipfs_async, args=(req.ipfs_cid,), daemon=True).start()
if ipfs_client.is_available():
ipfs_client.pin(req.ipfs_cid)
except Exception as e:
logger.warning(f"Failed to pin IPFS content {req.ipfs_cid}: {e}")
# Create asset # Create asset
now = datetime.now(timezone.utc).isoformat() now = datetime.now(timezone.utc).isoformat()
@@ -1794,14 +1797,10 @@ async def publish_cache(req: PublishCacheRequest, user: User = Depends(get_requi
if await db.asset_exists(req.asset_name): if await db.asset_exists(req.asset_name):
raise HTTPException(400, f"Asset name already exists: {req.asset_name}") raise HTTPException(400, f"Asset name already exists: {req.asset_name}")
# Pin content on IPFS if CID provided # Pin content on IPFS if CID provided (fire-and-forget, don't block)
if req.ipfs_cid: if req.ipfs_cid:
try: import threading
import ipfs_client threading.Thread(target=_pin_ipfs_async, args=(req.ipfs_cid,), daemon=True).start()
if ipfs_client.is_available():
ipfs_client.pin(req.ipfs_cid)
except Exception as e:
logger.warning(f"Failed to pin IPFS content {req.ipfs_cid}: {e}")
# Create asset # Create asset
now = datetime.now(timezone.utc).isoformat() now = datetime.now(timezone.utc).isoformat()