[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

@@ -1,17 +1,26 @@
 #  Drakkar-Software OctoBot
 #  Shared helpers/constants for octobot process functional tests (run_octobot_process, GridTradingMode).
 
+import asyncio
 import copy
 import decimal
+import json
+import os
+import pathlib
+import time
 import typing
 
+import octobot.constants as octobot_constants
+import octobot_commons.configuration as configuration_module
+import octobot_commons.constants as commons_constants
 import octobot_commons.dsl_interpreter as dsl_interpreter
 import octobot_trading.constants as trading_constants
 import octobot_trading.enums as trading_enums
 import pytest
 
 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
@@ -200,6 +209,100 @@ def _get_action_by_id(
     return None
 
 
+_FERNET_ENCRYPTED_PREFIX = "gAAAAA"
+
+
+# --- Child on-disk readiness (poll after init_state_ok; PID can be up before first dump / encrypt) ---
+
+
+def _process_bot_state_path(inner: dict) -> str:
+    return os.path.normpath(
+        os.path.join(
+            inner["user_root"],
+            octobot_constants.PROCESS_BOT_STATE_FILE_NAME,
+        )
+    )
+
+
+async def _wait_for_process_bot_state_file(
+    state_path: str,
+    *,
+    timeout_sec: float = GLOBAL_START_TIMEOUT_SEC,
+    poll_interval_sec: float = SLEEP_BETWEEN_JOB_POLLS_SEC,
+) -> None:
+    """Poll until the child has written at least one process_bot_state.json dump."""
+    deadline = time.monotonic() + timeout_sec
+    while time.monotonic() < deadline:
+        if os.path.isfile(state_path):
+            return
+        await asyncio.sleep(poll_interval_sec)
+    pytest.fail(
+        f"Timed out waiting for process_bot_state.json at {state_path!r} within {timeout_sec}s"
+    )
+
+
+async def _assert_encrypted_exchange_credentials_in_user_config(
+    user_root: typing.Union[pathlib.Path, str],
+    exchange_internal_name: str,
+    expected_api_key: str,
+    expected_api_secret: str,
+    *,
+    timeout_sec: float = GLOBAL_START_TIMEOUT_SEC,
+    poll_interval_sec: float = SLEEP_BETWEEN_JOB_POLLS_SEC,
+) -> None:
+    """
+    Poll child user-root config.json, then assert api key/secret are Fernet-encrypted and decrypt
+    to the expected plaintext.
+
+    run_octobot_process seeds plaintext credentials before spawn; the child re-saves config.json
+    on startup. init_state_ok can become true from PID liveness before encryption completes.
+    """
+    user_root_path = pathlib.Path(user_root)
+    config_path = user_root_path / commons_constants.CONFIG_FILE
+    deadline = time.monotonic() + timeout_sec
+    last_failure_reason = "config.json missing or exchange entry not ready"
+    while time.monotonic() < deadline:
+        if not config_path.is_file():
+            await asyncio.sleep(poll_interval_sec)
+            continue
+        root_cfg = json.loads(config_path.read_text(encoding="utf-8"))
+        exchanges_cfg = root_cfg.get(commons_constants.CONFIG_EXCHANGES) or {}
+        exchange_cfg = exchanges_cfg.get(exchange_internal_name)
+        if not isinstance(exchange_cfg, dict):
+            last_failure_reason = f"exchange {exchange_internal_name!r} missing from config"
+            await asyncio.sleep(poll_interval_sec)
+            continue
+        stored_api_key = exchange_cfg.get(commons_constants.CONFIG_EXCHANGE_KEY)
+        stored_api_secret = exchange_cfg.get(commons_constants.CONFIG_EXCHANGE_SECRET)
+        if not (
+            isinstance(stored_api_key, str)
+            and stored_api_key.startswith(_FERNET_ENCRYPTED_PREFIX)
+            and isinstance(stored_api_secret, str)
+            and stored_api_secret.startswith(_FERNET_ENCRYPTED_PREFIX)
+        ):
+            last_failure_reason = "api key/secret not yet Fernet-encrypted in config.json"
+            await asyncio.sleep(poll_interval_sec)
+            continue
+        try:
+            decrypted_api_key = configuration_module.decrypt(stored_api_key)
+            decrypted_api_secret = configuration_module.decrypt(stored_api_secret)
+        except Exception as decrypt_error:
+            last_failure_reason = f"decrypt failed: {decrypt_error}"
+            await asyncio.sleep(poll_interval_sec)
+            continue
+        assert decrypted_api_key == expected_api_key, (
+            f"decrypted api key mismatch for {exchange_internal_name!r}"
+        )
+        assert decrypted_api_secret == expected_api_secret, (
+            f"decrypted api secret mismatch for {exchange_internal_name!r}"
+        )
+        return
+    pytest.fail(
+        f"Timed out waiting for encrypted exchange credentials in {config_path!r} "
+        f"within {timeout_sec}s ({last_failure_reason})"
+    )
+
+
 def _make_tracked_spawn_managed_with_forward_terminal_output(
     real_spawn_managed: typing.Callable[..., typing.Any],
     popen_calls: dict[str, int],
@@ -213,6 +316,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/flow/tests/functionnal_tests/octobot_process_actions/test_octobot_process_edit_config.py

@@ -185,6 +185,10 @@ async def test_run_octobot_process_grid_refresh_four_to_six_orders(
             initial_spawn_count = popen_calls["count"]
             assert initial_spawn_count >= 1
 
+            # First process_bot_state dump can lag init_state_ok (see shared wait helper).
+            state_path = octobot_process_functional_shared._process_bot_state_path(inner)
+            await octobot_process_functional_shared._wait_for_process_bot_state_file(state_path)
+
             # 3) Wait until at least four open ladder orders exist, then assert a 2×2 grid pattern.
             orders_deadline = time.monotonic() + octobot_process_functional_shared.GRID_ORDERS_TIMEOUT_SEC
             exchange_account_snapshot: typing.Optional[

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

@@ -12,7 +12,6 @@
 
 import mock
 import octobot.constants as octobot_app_constants
-import octobot_commons.configuration as configuration_module
 import octobot_commons.constants as common_constants
 import octobot_commons.dsl_interpreter as dsl_interpreter
 import octobot_commons.process_util as process_util
@@ -154,13 +153,9 @@ async def test_run_octobot_process_lifecycle_grid_trading(
             assert popen_calls["count"] >= 1
 
             # --- process_bot_state path: must exist before poll (child wrote at least one dump) ---
-            state_path = os.path.normpath(
-                os.path.join(
-                    inner["user_root"],
-                    octobot_app_constants.PROCESS_BOT_STATE_FILE_NAME,
-                )
-            )
-            assert os.path.isfile(state_path)
+            # First process_bot_state dump can lag init_state_ok (see shared wait helper).
+            state_path = octobot_process_functional_shared._process_bot_state_path(inner)
+            await octobot_process_functional_shared._wait_for_process_bot_state_file(state_path)
 
             # 1) Poll AutomationJob + dump() until merge yields ≥4 open orders (EAE from automation snapshot,
             #    not from parsing full process_bot_state on disk).
@@ -438,15 +433,14 @@ async def test_run_octobot_process_lifecycle_default_config_no_profile_data(
 
             user_root = pathlib.Path(inner["user_root"])
             assert inner.get("profile_id") == "non-trading"
-            root_cfg = json.loads((user_root / common_constants.CONFIG_FILE).read_text(encoding="utf-8"))
-            exchange_cfg = root_cfg[common_constants.CONFIG_EXCHANGES][
-                octobot_process_functional_shared.EXCHANGE_BINANCEUS
-            ]
             exchange_auth_entry = exchange_auth[0]
-            stored_api_key = exchange_cfg[common_constants.CONFIG_EXCHANGE_KEY]
-            stored_api_secret = exchange_cfg[common_constants.CONFIG_EXCHANGE_SECRET]
-            assert configuration_module.decrypt(stored_api_key) == exchange_auth_entry["api_key"]
-            assert configuration_module.decrypt(stored_api_secret) == exchange_auth_entry["api_secret"]
+            # Child encrypts config.json after spawn; wait + assert in one helper (see shared module).
+            await octobot_process_functional_shared._assert_encrypted_exchange_credentials_in_user_config(
+                user_root,
+                octobot_process_functional_shared.EXCHANGE_BINANCEUS,
+                exchange_auth_entry["api_key"],
+                exchange_auth_entry["api_secret"],
+            )
             profile_json_path = (
                 user_root
                 / common_constants.PROFILES_FOLDER
@@ -455,13 +449,9 @@ async def test_run_octobot_process_lifecycle_default_config_no_profile_data(
             )
             assert profile_json_path.is_file()
 
-            state_path = os.path.normpath(
-                os.path.join(
-                    inner["user_root"],
-                    octobot_app_constants.PROCESS_BOT_STATE_FILE_NAME,
-                )
-            )
-            assert os.path.isfile(state_path)
+            # First process_bot_state dump can lag init_state_ok (see shared wait helper).
+            state_path = octobot_process_functional_shared._process_bot_state_path(inner)
+            await octobot_process_functional_shared._wait_for_process_bot_state_file(state_path)
             with open(state_path, encoding="utf-8") as process_state_file:
                 file_metadata_payload = json.load(process_state_file)
             process_metadata = process_bot_state_import.Metadata.from_dict(

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