refactor(core): Use shared memory instead of file-based transfer in native Python runner (#20150)

packages/@n8n/task-runner-python/src/constants.py

@@ -77,7 +77,7 @@ IGNORED_ERROR_TYPES = (
 # Logging
 LOG_FORMAT = "%(asctime)s.%(msecs)03d\t%(levelname)s\t%(message)s"
 LOG_TIMESTAMP_FORMAT = "%Y-%m-%d %H:%M:%S"
-LOG_TASK_COMPLETE = 'Completed task {task_id} in {duration} for node "{node_name}" ({node_id}) in workflow "{workflow_name}" ({workflow_id})'
+LOG_TASK_COMPLETE = 'Completed task {task_id} in {duration} ({result_size}) for node "{node_name}" ({node_id}) in workflow "{workflow_name}" ({workflow_id})'
 LOG_TASK_CANCEL = 'Cancelled task {task_id} for node "{node_name}" ({node_id}) in workflow "{workflow_name}" ({workflow_id})'
 LOG_TASK_CANCEL_UNKNOWN = (
     "Received cancel for unknown task: {task_id}. Discarding message."

packages/@n8n/task-runner-python/src/task_executor.py

@@ -5,7 +5,6 @@ import textwrap
 import json
 import os
 import sys
-import tempfile
 import logging
 
 from src.errors import (
@@ -27,6 +26,7 @@ from src.constants import (
 from typing import Any, Set
 
 from multiprocessing.context import ForkServerProcess
+from multiprocessing import shared_memory
 
 logger = logging.getLogger(__name__)
 
@@ -79,7 +79,7 @@ class TaskExecutor:
         queue: multiprocessing.Queue,
         task_timeout: int,
         continue_on_fail: bool,
-    ) -> tuple[list, PrintArgs]:
+    ) -> tuple[list, PrintArgs, int]:
         """Execute a subprocess for a Python code task."""
 
         print_args: PrintArgs = []
@@ -119,23 +119,29 @@ class TaskExecutor:
             if "error" in returned:
                 raise TaskRuntimeError(returned["error"])
 
-            if "result_file" not in returned:
+            if "shm_name" not in returned:
                 raise TaskResultMissingError()
 
-            result_file = returned["result_file"]
+            shm_name = returned["shm_name"]
+            shm_size = returned["shm_size"]
             try:
-                with open(result_file, "r", encoding="utf-8") as f:
-                    result = json.load(f)
-            finally:
-                os.unlink(result_file)
+                shm = shared_memory.SharedMemory(name=shm_name)
+                try:
+                    json_str = bytes(shm.buf[:shm_size]).decode("utf-8")
+                    result = json.loads(json_str)
+                finally:
+                    shm.close()
+                    shm.unlink()
+            except FileNotFoundError:
+                raise TaskResultMissingError()
 
             print_args = returned.get("print_args", [])
 
-            return result, print_args
+            return result, print_args, shm_size
 
         except Exception as e:
             if continue_on_fail:
-                return [{"json": {"error": str(e)}}], print_args
+                return [{"json": {"error": str(e)}}], print_args, 0
             raise
 
     @staticmethod
@@ -249,18 +255,22 @@ class TaskExecutor:
     def _put_result(
         queue: multiprocessing.Queue, result: list[Any], print_args: PrintArgs
     ):
-        with tempfile.NamedTemporaryFile(
-            mode="w",
-            delete=False,
-            prefix="n8n_result_",
-            suffix=".json",
-            encoding="utf-8",
-        ) as f:
-            json.dump(result, f, default=str, ensure_ascii=False)
-            result_file = f.name
+        json_bytes = json.dumps(result, default=str, ensure_ascii=False).encode("utf-8")
+        json_bytes_size = len(json_bytes)
+
+        shm = shared_memory.SharedMemory(create=True, size=json_bytes_size)
+        shm.buf[:json_bytes_size] = json_bytes
 
         print_args_to_send = TaskExecutor._truncate_print_args(print_args)
-        queue.put({"result_file": result_file, "print_args": print_args_to_send})
+        queue.put(
+            {
+                "shm_name": shm.name,
+                "shm_size": json_bytes_size,  # stay exact, shm.size can round up for alignment
+                "print_args": print_args_to_send,
+            }
+        )
+
+        shm.close()
 
     @staticmethod
     def _put_error(queue: multiprocessing.Queue, e: Exception, print_args: PrintArgs):

packages/@n8n/task-runner-python/src/task_runner.py

@@ -303,7 +303,7 @@ class TaskRunner:
 
             task_state.process = process
 
-            result, print_args = await asyncio.to_thread(
+            result, print_args, result_size_bytes = await asyncio.to_thread(
                 self.executor.execute_process,
                 process=process,
                 queue=queue,
@@ -323,6 +323,7 @@ class TaskRunner:
                 LOG_TASK_COMPLETE.format(
                     task_id=task_id,
                     duration=self._get_duration(start_time),
+                    result_size=self._get_result_size(result_size_bytes),
                     **task_state.context(),
                 )
             )
@@ -380,6 +381,8 @@ class TaskRunner:
         serialized = self.serde.serialize_runner_message(message)
         await self.websocket_connection.send(serialized)
 
+    # ========== Formatting ==========
+
     def _get_duration(self, start_time: float) -> str:
         elapsed = time.time() - start_time
 
@@ -391,6 +394,14 @@ class TaskRunner:
 
         return f"{int(elapsed) // 60}m"
 
+    def _get_result_size(self, size_bytes: int) -> str:
+        if size_bytes < 1024:
+            return f"{size_bytes} bytes"
+        elif size_bytes < 1024 * 1024:
+            return f"{size_bytes / 1024:.1f} KB"
+        else:
+            return f"{size_bytes / (1024 * 1024):.1f} MB"
+
     # ========== Offers ==========
 
     async def _send_offers_loop(self) -> None: