Add ipfs.pinning configuration section for separate pinning service with backward compatibility (#859)

Author: aliel

src/aleph/config.py

@@ -171,10 +171,12 @@ def get_defaults():
         "ipfs": {
             # Whether to enable storage and communication on IPFS.
             "enabled": True,
-            # Hostname of the IPFS service.
+            # Hostname of the IPFS.
             "host": "ipfs",
             # Port of the IPFS service.
             "port": 5001,
+            # scheme of the IPFS service
+            "scheme": "http",
             # IPFS pubsub topic used for liveness checks.
             "alive_topic": "ALEPH_ALIVE",
             # Delay between connection attempts to other nodes on the network.
@@ -184,6 +186,17 @@ def get_defaults():
                 "/ip4/51.159.57.71/tcp/4001/p2p/12D3KooWBH3JVSBwHLNzxv7EzniBP3tDmjJaoa3EJBF9wyhZtHt2",
                 "/ip4/62.210.93.220/tcp/4001/p2p/12D3KooWLcmvqojHzUnR7rr8YhFKGDD8z7fmsPyBfAm2rT3sFGAF",
             ],
+            # Pinning service configuration
+            "pinning": {
+                # Hostname of the IPFS pinning service (if different from main IPFS).
+                "host": None,
+                # Port of the IPFS pinning service (if different from main IPFS).
+                "port": 5001,
+                # Scheme of the ipfs pinning service.
+                "scheme": "http",
+                # Timeout for pinning operations (seconds)
+                "timeout": 60,
+            },
         },
         "rabbitmq": {
             # Hostname of the RabbitMQ service.

src/aleph/network.py

@@ -9,7 +9,7 @@
 from aleph.handlers.message_handler import MessagePublisher
 from aleph.services.cache.node_cache import NodeCache
 from aleph.services.ipfs import IpfsService
-from aleph.services.ipfs.common import make_ipfs_client
+from aleph.services.ipfs.common import make_ipfs_p2p_client
 from aleph.services.ipfs.pubsub import incoming_channel as incoming_ipfs_channel
 from aleph.services.storage.fileystem_engine import FileSystemStorageEngine
 from aleph.storage import StorageService
@@ -46,7 +46,7 @@ async def listener_tasks(
     from aleph.services.p2p.protocol import incoming_channel as incoming_p2p_channel
 
     # TODO: these should be passed as parameters. This module could probably be a class instead?
-    ipfs_client = make_ipfs_client(config)
+    ipfs_client = make_ipfs_p2p_client(config)
     ipfs_service = IpfsService(ipfs_client=ipfs_client)
     storage_service = StorageService(
         storage_engine=FileSystemStorageEngine(folder=config.storage.folder.value),

src/aleph/services/ipfs/common.py

@@ -8,20 +8,61 @@ async def get_base_url(config):
     return "http://{}:{}".format(config.ipfs.host.value, config.ipfs.port.value)
 
 
-def make_ipfs_client(config: Config, timeout: int = 60) -> aioipfs.AsyncIPFS:
-    host = config.ipfs.host.value
-    port = config.ipfs.port.value
-
+def make_ipfs_client(
+    host: str,
+    port: int,
+    timeout: int = 60,
+    scheme: str = "http",
+    debug_level: int = logging.WARNING,
+) -> aioipfs.AsyncIPFS:
     return aioipfs.AsyncIPFS(
         host=host,
         port=port,
+        scheme=scheme,
         read_timeout=timeout,
         conns_max=25,
         conns_max_per_host=10,
-        debug=(config.logging.level.value <= logging.DEBUG),
+        debug=debug_level,
     )
 
 
+def make_ipfs_p2p_client(config: Config, timeout: int = 60) -> aioipfs.AsyncIPFS:
+    """Create IPFS client for P2P operations (pubsub, content retrieval)."""
+    # Always use main IPFS config for P2P operations
+    host = config.ipfs.host.value
+    port = int(config.ipfs.port.value)
+    scheme = config.ipfs.scheme.value
+    debug_level = config.logging.level.value <= logging.DEBUG
+
+    return make_ipfs_client(host, port, timeout, scheme, debug_level)
+
+
+def make_ipfs_pinning_client(config: Config, timeout: int = 60) -> aioipfs.AsyncIPFS:
+    """Create IPFS client for pinning operations."""
+    # Use pinning specific config if provided, otherwise use main config
+    if (
+        hasattr(config.ipfs, "pinning")
+        and config.ipfs.pinning.host.value
+        and config.ipfs.pinning.port.value
+    ):
+        host = config.ipfs.pinning.host.value
+        port = int(config.ipfs.pinning.port.value)
+        scheme = config.ipfs.pinning.scheme.value
+    else:
+        # Use main IPFS config as fallback
+        host = config.ipfs.host.value
+        port = int(config.ipfs.port.value)
+        scheme = config.ipfs.scheme.value
+
+    # Get pinning-specific timeout if available
+    if hasattr(config.ipfs, "pinning") and hasattr(config.ipfs.pinning, "timeout"):
+        timeout = int(config.ipfs.pinning.timeout.value)
+
+    debug_level = config.logging.level.value <= logging.DEBUG
+
+    return make_ipfs_client(host, port, timeout, scheme, debug_level)
+
+
 def get_cid_version(ipfs_hash: str) -> int:
     if ipfs_hash.startswith("Qm") and 44 <= len(ipfs_hash) <= 46:  # CIDv0
         return 0

src/aleph/services/ipfs/service.py

@@ -8,7 +8,7 @@
 import aioipfs
 from configmanager import Config
 
-from aleph.services.ipfs.common import make_ipfs_client
+from aleph.services.ipfs.common import make_ipfs_p2p_client, make_ipfs_pinning_client
 from aleph.services.utils import get_IP
 from aleph.types.message_status import FileUnavailable
 from aleph.utils import run_in_executor
@@ -19,19 +19,36 @@
 
 
 class IpfsService:
-    def __init__(self, ipfs_client: aioipfs.AsyncIPFS):
-        self.ipfs_client = ipfs_client
+    def __init__(
+        self,
+        ipfs_client: aioipfs.AsyncIPFS,
+        pinning_client: Optional[aioipfs.AsyncIPFS] = None,
+    ):
+        self.ipfs_client = ipfs_client  # For P2P operations
+        self.pinning_client = pinning_client or ipfs_client  # For pinning operations
 
     @classmethod
     def new(cls, config: Config) -> Self:
-        ipfs_client = make_ipfs_client(config)
-        return cls(ipfs_client=ipfs_client)
+        # Create P2P client (for content retrieval, pubsub)
+        p2p_client = make_ipfs_p2p_client(config)
+
+        # Create separate pinning client if configured differently
+        if _should_use_separate_pinning_client(config):
+            LOGGER.info("Using separate IPFS client for pinning operations")
+            pinning_client = make_ipfs_pinning_client(config)
+        else:
+            pinning_client = p2p_client
+
+        return cls(ipfs_client=p2p_client, pinning_client=pinning_client)
 
     async def __aenter__(self):
         return self
 
     async def close(self):
         await self.ipfs_client.close()
+        # Only close pinning client if it's different from main client
+        if self.pinning_client != self.ipfs_client:
+            await self.pinning_client.close()
 
     async def __aexit__(self, exc_type, exc_val, exc_tb):
         await self.close()
@@ -183,11 +200,11 @@ async def get_json(self, hash, timeout=1, tries=1):
         return result
 
     async def add_json(self, value: bytes) -> str:
-        result = self.ipfs_client.add_json(value)
+        result = self.pinning_client.add_json(value)
         return result["Hash"]
 
     async def add_bytes(self, value: bytes, cid_version: int = 0) -> str:
-        result = await self.ipfs_client.add_bytes(value, cid_version=cid_version)
+        result = await self.pinning_client.add_bytes(value, cid_version=cid_version)
         return result["Hash"]
 
     async def _pin_add(self, cid: str, timeout: int = 30):
@@ -201,7 +218,8 @@ async def _pin_add(self, cid: str, timeout: int = 30):
         tick_timeout = timeout * 2
         last_progress = None
 
-        async for status in self.ipfs_client.pin.add(cid):
+        # Use pinning client instead of main client
+        async for status in self.pinning_client.pin.add(cid):
             # If the Pins key appears, the file is pinned.
             if "Pins" in status:
                 break
@@ -235,15 +253,13 @@ async def pin_add(self, cid: str, timeout: int = 30, tries: int = 1):
                 break
 
     async def add_file(self, file_content: bytes):
-        url = f"{self.ipfs_client.api_url}add"
+        """
+        Add a file to IPFS using bytes as data.
 
-        async with aiohttp.ClientSession() as session:
-            data = aiohttp.FormData()
-            data.add_field("path", file_content)
-
-            resp = await session.post(url, data=data)
-            resp.raise_for_status()
-            return await resp.json()
+        This is a backward-compatible wrapper around add_bytes().
+        Uses the pinning client for write operations.
+        """
+        return await self.add_bytes(file_content)
 
     async def sub(self, topic: str):
         ipfs_client = self.ipfs_client
@@ -269,3 +285,23 @@ async def pub(self, topic: str, message: Union[str, bytes]):
 
         ipfs_client = self.ipfs_client
         await ipfs_client.pubsub.pub(topic, message_str)
+
+
+def _should_use_separate_pinning_client(config: Config) -> bool:
+    """
+    Determine if we should use a separate IPFS client for pinning operations.
+    Returns True if pinning configuration is different from main IPFS configuration.
+    """
+    if not hasattr(config.ipfs, "pinning"):
+        return False
+
+    # Check if pinning host/port are specifically configured and different
+    pinning_host = config.ipfs.pinning.host.value
+    pinning_port = config.ipfs.pinning.port.value
+
+    if pinning_host and pinning_port:
+        main_host = config.ipfs.host.value
+        main_port = config.ipfs.port.value
+        return (pinning_host != main_host) or (pinning_port != main_port)
+
+    return False