[Node] fix process bot restart issues

Author: GuillaumeDSM

octobot/storage/process_bot_state_dumper.py

@@ -73,7 +73,11 @@ async def _write_state_file_async(
 ) -> None:
     now = time.time()
     state = process_bot_state_import.ProcessBotState(
-        metadata=process_bot_state_import.Metadata(updated_at=now, next_updated_at=now + interval),
+        metadata=process_bot_state_import.Metadata(
+            updated_at=now,
+            next_updated_at=now + interval,
+            pid=os.getpid(),
+        ),
         exchange_account_elements=_synced_exchange_account_elements_for_first_trading_exchange(
             bot,
         ),

packages/flow/octobot_flow/entities/accounts/process_bot_state.py

@@ -10,11 +10,14 @@
 @dataclasses.dataclass
 class Metadata(octobot_commons.dataclasses.MinimizableDataclass):
     """
-    Timestamps written with process bot state dumps; used for file-based liveness checks.
+    Timestamps and child PID written with process bot state dumps. Liveness checks use
+    updated_at / next_updated_at only; pid is the authoritative child PID for parent binding
+    after restarts.
     """
 
     updated_at: float = 0.0
     next_updated_at: float = 0.0
+    pid: int = 0
 
 
 @dataclasses.dataclass

packages/flow/octobot_flow/environment.py

@@ -1,8 +1,21 @@
+import typing
+
 import octobot.constants  # will load .env file and init constants
 
 import octobot_flow.repositories.community
 import octobot_trading.constants
 
+_EXECUTOR_ID: typing.Optional[str] = None
+
+
+def register_executor_id(executor_id: str) -> None:
+    global _EXECUTOR_ID
+    _EXECUTOR_ID = executor_id
+
+
+def get_executor_id() -> typing.Optional[str]:
+    return _EXECUTOR_ID
+
 
 def initialize_environment(allow_funds_transfer: bool = False) -> None:
     octobot_flow.repositories.community.initialize_community_authentication()

packages/flow/octobot_flow/logic/dsl/dsl_executor.py

@@ -12,6 +12,7 @@
 import octobot_evaluators.evaluators as evaluators
 
 import octobot_flow.entities
+import octobot_flow.environment
 import octobot_flow.errors
 import octobot_flow.enums
 import octobot_flow.logic.dsl.action_error_util
@@ -24,21 +25,23 @@
 import tentacles.Meta.DSL_operators.octobot_process_operators.octobot_process_ops as octobot_process_ops
 
 
-
 class DSLExecutor(AbstractActionExecutor):
     def __init__(
         self,
         profile_data: octobot_commons.profiles.ProfileData,
         exchange_manager: typing.Optional[octobot_trading.exchanges.ExchangeManager],
         dsl_script: typing.Optional[str],
         dependencies: typing.Optional[octobot_commons.signals.SignalDependencies] = None,
+        executor_id: typing.Optional[str] = None,
     ):
         super().__init__()
         self._exchange_manager = exchange_manager
         self._dependencies = dependencies
         self._dependencies_config: dict = profile_data.to_profile("").config
         self._interpreter_signals: octobot_commons.dsl_interpreter.OperatorSignals = None # type: ignore (reset when interpreter is created)
-        self._interpreter: octobot_commons.dsl_interpreter.Interpreter = self._create_interpreter(None)
+        self._interpreter: octobot_commons.dsl_interpreter.Interpreter = self._create_interpreter(
+            None, executor_id
+        )
         if dsl_script:
             self._interpreter.prepare(dsl_script)
 
@@ -49,7 +52,15 @@ def _get_matrix_id(self) -> typing.Optional[str]:
 
     def get_flow_operator_classes(
         self,
+        executor_id: typing.Optional[str] = None,
     ) -> list[typing.Type[octobot_commons.dsl_interpreter.Operator]]:
+        resolved_executor_id = (
+            executor_id or octobot_flow.environment.get_executor_id()
+        )
+        if not resolved_executor_id:
+            raise octobot_flow.errors.MissingDSLExecutorDependencyError(
+                "executor_id is required for run_octobot_process"
+            )
         return (
             octobot_commons.dsl_interpreter.get_all_operators()
             + dsl_operators.create_ohlcv_operators(self._exchange_manager, None, None)
@@ -74,16 +85,19 @@ def get_flow_operator_classes(
                 copier_trading_mode=None,
             )
             + octobot_process_ops.create_octobot_process_operators(
-                self._interpreter_signals
+                self._interpreter_signals,
+                resolved_executor_id,
             )
         ) # type: ignore (list[type[Operator]])
 
     def _create_interpreter(
-        self, previous_execution_result: typing.Optional[dict]
+        self,
+        previous_execution_result: typing.Optional[dict],
+        executor_id: typing.Optional[str] = None,
     ) -> octobot_commons.dsl_interpreter.Interpreter:
         self._interpreter_signals = octobot_commons.dsl_interpreter.OperatorSignals()
         return octobot_commons.dsl_interpreter.Interpreter(
-            self.get_flow_operator_classes()
+            self.get_flow_operator_classes(executor_id)
         )
 
     def get_dependencies(self) -> list[
@@ -110,7 +124,8 @@ async def execute_action(
         ] = None,
     ) -> octobot_commons.dsl_interpreter.DSLCallResult:
         self._interpreter = self._create_interpreter(
-            action.previous_execution_result
+            action.previous_execution_result,
+            None,
         )
         expression = action.get_resolved_dsl_script()
         try:

packages/flow/tests/functionnal_tests/octobot_process_actions/octobot_process_functional_shared.py

@@ -12,6 +12,7 @@
 
 import octobot_flow.jobs
 import octobot_flow.entities
+import octobot_flow.environment
 import octobot_flow.enums
 import tests.functionnal_tests as functionnal_tests
 import tests.functionnal_tests.tentacle_test_configs as tentacle_test_configs
@@ -213,6 +214,12 @@ def _tracked(*args: typing.Any, **kwargs: typing.Any) -> typing.Any:
     return _tracked
 
 
+@pytest.fixture(autouse=True)
+def register_functional_executor_id():
+    octobot_flow.environment.register_executor_id("func-test-executor")
+    yield
+
+
 @pytest.fixture
 def init_action():
     # Automation apply_configuration: seed automation state to match expected exchange + portfolio.

packages/node/octobot_node/constants.py

@@ -14,6 +14,7 @@
 #  You should have received a copy of the GNU General Public
 #  License along with OctoBot. If not, see <https://www.gnu.org/licenses/>.
 import os
+import uuid
 
 try:
     import octobot.constants as octobot_constants
@@ -67,3 +68,5 @@
 )
 
 DEFAULT_PORTFOLIO_VALUATION_UNIT = "USDT"
+
+SCHEDULER_EXECUTOR_ID = str(uuid.uuid4()) # unique for each worker

packages/node/octobot_node/scheduler/octobot_flow_client.py

@@ -20,6 +20,7 @@
 
 import octobot_commons.dataclasses
 
+import octobot_node.constants
 import octobot_node.errors as errors
 import octobot_node.scheduler.workflows_util as workflows_util
 
@@ -33,6 +34,9 @@
 
     # ensure environment is initialized
     octobot_flow.environment.initialize_environment(True)
+    octobot_flow.environment.register_executor_id(
+        octobot_node.constants.SCHEDULER_EXECUTOR_ID
+    )
 
 except ImportError:
     pass # OctoBot Flow is not available

packages/node/octobot_node/scheduler/scheduler.py

@@ -48,7 +48,8 @@
 _BASE_CONFIG = dbos.DBOSConfig(
     name=DEFAULT_NAME,
     max_executor_threads=octobot_node.config.settings.SCHEDULER_MAX_EXECUTOR_THREADS,
-    application_version=VERSION,
+    application_version=VERSION, # octobot version
+    executor_id=octobot_node.constants.SCHEDULER_EXECUTOR_ID,
 )