Fix: Chain writer failed due to TypeError (#549)

Solution: Use `bytes` in memory everywhere. Not elegant but fixes the problem.

---

The Chain writer task for Chain.ETH failed with the following stacktrace:

```
aleph-pyaleph-1 | 2024-01-29 14:09:32 [ERROR] aleph.chains.connector: Chain writer task for Chain.ETH failed, relaunching in 10 seconds.
aleph-pyaleph-1 | Traceback (most recent call last):
aleph-pyaleph-1 |   File "/opt/pyaleph/src/aleph/chains/connector.py", line 63, in chain_writer_task
aleph-pyaleph-1 |     await connector.packer(config)
aleph-pyaleph-1 |   File "/opt/pyaleph/src/aleph/chains/ethereum.py", line 351, in packer
aleph-pyaleph-1 |     await self.chain_data_service.prepare_sync_event_payload(
aleph-pyaleph-1 |   File "/opt/pyaleph/src/aleph/chains/chain_data_service.py", line 75, in prepare_sync_event_payload
aleph-pyaleph-1 |     ipfs_cid = await self.storage_service.add_file(
aleph-pyaleph-1 |                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
aleph-pyaleph-1 |   File "/opt/pyaleph/src/aleph/storage.py", line 293, in add_file
aleph-pyaleph-1 |     await self.add_file_content_to_local_storage(
aleph-pyaleph-1 |   File "/opt/pyaleph/src/aleph/storage.py", line 269, in add_file_content_to_local_storage
aleph-pyaleph-1 |     await self.storage_engine.write(filename=file_hash, content=file_content)
aleph-pyaleph-1 |   File "/opt/pyaleph/src/aleph/services/storage/fileystem_engine.py", line 26, in write
aleph-pyaleph-1 |     file_path.write_bytes(content)
aleph-pyaleph-1 |   File "/usr/lib/python3.11/pathlib.py", line 1066, in write_bytes
aleph-pyaleph-1 |     view = memoryview(data)
aleph-pyaleph-1 |            ^^^^^^^^^^^^^^^^
aleph-pyaleph-1 | TypeError: memoryview: a bytes-like object is required, not 'str'
```

It appears that a `StringIO` was passed to `aleph.storage.add_file(...)`, resulting in `file_content` being of type `str`, where the underlying function `aleph.services.storage.fileystem_engine.FileSystemStorageEngine.write` expects the type `bytes`.

Solution:
1. Encode the `archive_content` in UTF-8 before adding it in the storage service.
2. Modify the type annotations accordingly
3. Modify the signature of `aleph.storage.StorageService.add_file` to expect `bytes` instead of any `IO` type.

src/aleph/chains/chain_data_service.py

@@ -70,10 +70,10 @@ async def prepare_sync_event_payload(
                 messages=[OnChainMessage.from_orm(message) for message in messages]
             ),
         )
-        archive_content = archive.json()
+        archive_content: bytes = archive.json().encode("utf-8")
 
         ipfs_cid = await self.storage_service.add_file(
-            session=session, fileobject=StringIO(archive_content), engine=ItemType.ipfs
+            session=session, file_content=archive_content, engine=ItemType.ipfs
         )
         return OffChainSyncEventPayload(
             protocol=ChainSyncProtocol.OFF_CHAIN_SYNC, version=1, content=ipfs_cid

src/aleph/services/ipfs/service.py

@@ -154,12 +154,12 @@ async def pin_add(self, cid: str, timeout: int = 30, tries: int = 1):
             else:
                 break
 
-    async def add_file(self, fileobject: IO):
+    async def add_file(self, file_content: bytes):
         url = f"{self.ipfs_client.api_url}add"
 
         async with aiohttp.ClientSession() as session:
             data = aiohttp.FormData()
-            data.add_field("path", fileobject)
+            data.add_field("path", file_content)
 
             resp = await session.post(url, data=data)
             resp.raise_for_status()

src/aleph/storage.py

@@ -275,16 +275,13 @@ async def add_file_content_to_local_storage(
         )
 
     async def add_file(
-        self, session: DbSession, fileobject: IO, engine: ItemType = ItemType.ipfs
+        self, session: DbSession, file_content: bytes, engine: ItemType = ItemType.ipfs
     ) -> str:
         if engine == ItemType.ipfs:
-            output = await self.ipfs_service.add_file(fileobject)
+            output = await self.ipfs_service.add_file(file_content)
             file_hash = output["Hash"]
-            fileobject.seek(0)
-            file_content = fileobject.read()
 
         elif engine == ItemType.storage:
-            file_content = fileobject.read()
             file_hash = sha256(file_content).hexdigest()
 
         else:

src/aleph/web/controllers/ipfs.py

@@ -1,6 +1,7 @@
 import asyncio
 
 from aiohttp import web
+from aiohttp.web_request import FileField
 
 from aleph.db.accessors.files import upsert_file
 from aleph.types.files import FileType
@@ -29,7 +30,23 @@ async def ipfs_add_file(request: web.Request):
     except KeyError:
         raise web.HTTPUnprocessableEntity(reason="Missing 'file' in multipart form.")
 
-    ipfs_add_response = await ipfs_service.add_file(file_field_to_io(file_field))
+    file_content: bytes
+    if isinstance(file_field, bytes):
+        file_content = file_field
+    elif isinstance(file_field, str):
+        file_content = file_field.encode()
+    elif isinstance(file_field, FileField):
+        if file_field.content_type != "application/octet-stream":
+            raise web.HTTPUnprocessableEntity(
+                reason="Invalid content-type for 'file' field. Must be 'application/octet-stream'."
+            )
+        file_content = file_field.file.read()
+    else:
+        raise web.HTTPUnprocessableEntity(
+            reason="Invalid type for 'file' field. Must be bytes, str or FileField."
+        )
+
+    ipfs_add_response = await ipfs_service.add_file(file_content)
 
     cid = ipfs_add_response["Hash"]
     name = ipfs_add_response["Name"]

tests/api/test_storage.py

@@ -104,7 +104,7 @@ async def add_file(
 
     post_response = await api_client.post(uri, data=form_data)
     response_text = await post_response.text()
-    assert post_response.status == 200, await response_text
+    assert post_response.status == 200, response_text
     post_response_json = await post_response.json()
     assert post_response_json["status"] == "success"
     file_hash = post_response_json["hash"]

tests/chains/test_chain_data_service.py

@@ -42,9 +42,9 @@ async def test_prepare_sync_event_payload(mocker):
     ]
 
     async def mock_add_file(
-        session: DbSession, fileobject: IO, engine: ItemType = ItemType.ipfs
+        session: DbSession, file_content: bytes, engine: ItemType = ItemType.ipfs
     ) -> str:
-        content = fileobject.read()
+        content = file_content
         archive = OnChainSyncEventPayload.parse_raw(content)
 
         assert archive.version == 1