privatevoid.net/depot
1
1
Fork 0
Code Issues 23 Pull requests 2 Projects 3 Releases Packages Wiki Activity Actions

packages/reflex-cache: use ipfs cluster for pinning

Browse source
This commit is contained in:
Max Headroom 2022-12-19 00:09:59 +01:00
parent 226fc6da9a
commit 2889b55808
3 changed files with 25 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
packages/servers/reflex-cache/pyproject.toml
View file

@ -1,6 +1,6 @@
[tool.poetry] [tool.poetry]
name = "reflex-cache" name = "reflex-cache"
version = "0.1.0" version = "0.2.0"
description = "Controller for Nix binary caches on IPFS" description = "Controller for Nix binary caches on IPFS"
authors = ["Max <max@privatevoid.net>"] authors = ["Max <max@privatevoid.net>"]
license = "AGPL-3.0" license = "AGPL-3.0"

14
packages/servers/reflex-cache/reflex_cache/ipfs.py
View file

@ -1,3 +1,4 @@
from datetime import datetime, timedelta, timezone
from urllib.parse import quote_plus from urllib.parse import quote_plus
import requests import requests
@ -5,8 +6,9 @@ import requests_unixsocket
class IPFSController: class IPFSController:
def __init__(self, apiAddress, nixCache, db): def __init__(self, nodeApiAddress, clusterApiAddress, nixCache, db):
self.__addr = f'http+unix://{quote_plus(apiAddress.get("unix"))}' self.__nodeAddr = f'http+unix://{quote_plus(nodeApiAddress.get("unix"))}'
self.__clusterAddr = f'http+unix://{quote_plus(clusterApiAddress.get("unix"))}'
self.__nix = nixCache self.__nix = nixCache
self.__db = db self.__db = db
@ -17,11 +19,17 @@ class IPFSController:
upload = {"file": ("FILE", content, "application/octet-stream")} upload = {"file": ("FILE", content, "application/octet-stream")}
try: try:
rIpfs = requests_unixsocket.post( rIpfs = requests_unixsocket.post(
f"{self.__addr}/api/v0/add?pin=false&quieter=true", files=upload f"{self.__nodeAddr}/api/v0/add?pin=false&quieter=true", files=upload
) )
hash = rIpfs.json()["Hash"] hash = rIpfs.json()["Hash"]
print(f"Mapped: {nar} -> /ipfs/{hash}") print(f"Mapped: {nar} -> /ipfs/{hash}")
self.__db.set_path(nar, hash) self.__db.set_path(nar, hash)
expireAt = datetime.now(timezone.utc) + timedelta(hours=24)
rClusterPin = requests_unixsocket.post(
f"{self.__clusterAddr}/pins/ipfs/{hash}?expire-at={quote_plus(expireAt.isoformat())}&mode=recursive&name=reflex-{quote_plus(nar)}&replication-max=2&replication-min=1", files=upload
)
if rClusterPin.status_code != 200:
print(f"Warning: failed to pin {hash} on IPFS cluster")
callback() callback()
return (nar, 200, hash) return (nar, 200, hash)
except requests.ConnectionError as e: except requests.ConnectionError as e:

15
packages/servers/reflex-cache/reflex_cache/service_handler.py
View file

@ -22,7 +22,11 @@ class ReflexHTTPServiceHandler(BaseHTTPRequestHandler):
util.envOr("NIX_CACHES", "https://cache.nixos.org").split(" ") util.envOr("NIX_CACHES", "https://cache.nixos.org").split(" ")
) )
_ipfs = ipfs.IPFSController(Multiaddr(util.envOrRaise("IPFS_API")), _nix, _db) _ipfs = ipfs.IPFSController(
Multiaddr(util.envOrRaise("IPFS_API")),
Multiaddr(util.envOrRaise("IPFS_CLUSTER_API")),
_nix,
_db)
def do_HEAD(self): def do_HEAD(self):
if self.path.endswith(".narinfo"): if self.path.endswith(".narinfo"):
@ -124,8 +128,15 @@ class ReflexHTTPServiceHandler(BaseHTTPRequestHandler):
if not found and len(self._workSet) < 8: if not found and len(self._workSet) < 8:
print(f"Pre-flight: creating IPFS fetch task for {nar}") print(f"Pre-flight: creating IPFS fetch task for {nar}")
def cb():
with self._workSetLock:
try:
self._workSet.remove((nar, f))
except KeyError:
# already removed
pass
f = self._executor_nar.submit( f = self._executor_nar.submit(
self._ipfs.ipfs_fetch_task, nar self._ipfs.ipfs_fetch_task, cb, nar
) )
self._workSet.add((nar, f)) self._workSet.add((nar, f))
return return