Fix: Undo the thread split and put exception handler to see ipfs errors.

Author: Andres D. Molins

src/aleph/commands.py

@@ -61,23 +61,6 @@ def run_db_migrations(config: Config):
     alembic.command.upgrade(alembic_cfg, "head", tag=db_url)
 
 
-def run_coroutine_in_new_thread_loop(coro: Coroutine):
-    """
-    Executes a given coroutine in a new event loop in the current thread.
-    This thread is typically one managed by an executor.
-    """
-    new_loop = asyncio.new_event_loop()
-    asyncio.set_event_loop(new_loop)
-    try:
-        LOGGER.debug(f"Running coroutine {getattr(coro, '__name__', str(coro))} in thread event loop: {new_loop}")
-        result = new_loop.run_until_complete(coro)
-        LOGGER.debug(f"Coroutine {getattr(coro, '__name__', str(coro))} completed in thread event loop: {new_loop}")
-        return result
-    finally:
-        LOGGER.debug(f"Closing thread event loop: {new_loop}")
-        new_loop.close()
-
-
 async def init_node_cache(config: Config) -> NodeCache:
     node_cache = NodeCache(
         redis_host=config.redis.host.value, redis_port=config.redis.port.value
@@ -214,24 +197,11 @@ async def main(args: List[str]) -> None:
         tasks.append(refresh_cache_materialized_views(session_factory))
         LOGGER.debug("Initialized cache tasks")
 
-        LOGGER.debug("Preparing garbage collector task to run in a separate thread")
-        # 1. Create the coroutine object as you normally would
-        gc_coroutine_object = garbage_collector_task(
-            config=config, garbage_collector=garbage_collector
-        )
-
-        # 2. Get the current event loop to use run_in_executor
-        current_loop = asyncio.get_running_loop()
-
-        # 3. Schedule the execution of `run_coroutine_in_new_thread_loop` with your coroutine in the executor
-        # This will return a Future that `asyncio.gather` can await.
-        gc_future = current_loop.run_in_executor(
-            None,  # Use the default ThreadPoolExecutor
-            run_coroutine_in_new_thread_loop,
-            gc_coroutine_object  # The coroutine to execute
+        LOGGER.debug("Initializing garbage collector task")
+        tasks.append(
+            garbage_collector_task(config=config, garbage_collector=garbage_collector)
         )
-        tasks.append(gc_future) # Add the Future to the list of tasks
-        LOGGER.debug("Garbage collector task submitted to run in its own thread event loop.")
+        LOGGER.debug("Initialized garbage collector task")
 
         LOGGER.debug("Running event loop")
         await asyncio.gather(*tasks)

src/aleph/services/storage/garbage_collector.py

@@ -35,6 +35,8 @@ async def _delete_from_ipfs(self, file_hash: ItemHash):
 
         except NotPinnedError:
             LOGGER.warning("File not pinned: %s", file_hash)
+        except Exception as err:
+            LOGGER.error("Failed to unpin file %s: %s", file_hash, str(err))
 
         # Smaller IPFS files are cached in local storage
         LOGGER.debug("Deleting %s from local storage", file_hash)