fix: allow deeploy update from persisted pipeline (#466)

extensions/business/deeploy/deeploy_manager_api.py

@@ -269,6 +269,85 @@ def _redact_failed_request_payload(self, payload):
     """
     return self._redact_per_node_config_for_log(payload)
 
+  def _gather_persisted_pipeline_update_context(self, owner, app_id, job_id, project_id=None):
+    pipeline_cid = self._get_pipeline_from_cstore(job_id) if job_id is not None else None
+    if not pipeline_cid:
+      raise ValueError(
+        f"{DEEPLOY_ERRORS.NODES3}: No running workers or persisted pipeline found for "
+        f"{f'app_id {app_id}' if app_id else f'job_id {job_id}'} and owner '{owner}'."
+      )
+
+    pipeline = self.get_pipeline_from_r1fs(
+      pipeline_cid,
+      timeout=30,
+      pin=False,
+      raise_on_error=False,
+      show_logs=False,
+    )
+    if not isinstance(pipeline, dict):
+      raise ValueError(f"{DEEPLOY_ERRORS.NODES3}: Persisted pipeline '{pipeline_cid}' is unavailable.")
+
+    pipeline_owner = pipeline.get(NetMonCt.OWNER.upper())
+    if str(pipeline_owner).lower() != str(owner).lower():
+      raise ValueError(f"{DEEPLOY_ERRORS.REQUEST3}. Persisted pipeline owner does not match request owner.")
+
+    pipeline_app_id = pipeline.get("NAME")
+    if str(pipeline_app_id).lower() != str(app_id).lower():
+      raise ValueError(f"{DEEPLOY_ERRORS.REQUEST3}. Persisted pipeline app_id does not match request app_id.")
+
+    deeploy_specs = pipeline.get(NetMonCt.DEEPLOY_SPECS.upper()) or {}
+    if not isinstance(deeploy_specs, dict):
+      deeploy_specs = {}
+
+    if str(deeploy_specs.get(DEEPLOY_KEYS.JOB_ID)).lower() != str(job_id).lower():
+      raise ValueError(f"{DEEPLOY_ERRORS.REQUEST3}. Persisted pipeline job_id does not match request job_id.")
+    if project_id is not None and str(deeploy_specs.get(DEEPLOY_KEYS.PROJECT_ID)).lower() != str(project_id).lower():
+      raise ValueError(f"{DEEPLOY_ERRORS.REQUEST3}. Persisted pipeline project_id does not match request project_id.")
+
+    nodes = deeploy_specs.get(DEEPLOY_KEYS.CURRENT_TARGET_NODES) or []
+    if not isinstance(nodes, list):
+      nodes = []
+
+    discovered_instances = []
+    plugins = pipeline.get(NetMonCt.PLUGINS.upper()) or []
+    if not isinstance(plugins, list):
+      plugins = []
+
+    for plugin in plugins:
+      if not isinstance(plugin, dict):
+        continue
+      signature = plugin.get(ct.CONFIG_PLUGIN.K_SIGNATURE) or plugin.get("signature")
+      if not signature:
+        continue
+      instances = plugin.get(ct.CONFIG_PLUGIN.K_INSTANCES, [])
+      if not isinstance(instances, list):
+        continue
+      for instance in instances:
+        if not isinstance(instance, dict):
+          continue
+        instance_id = instance.get(ct.CONFIG_INSTANCE.K_INSTANCE_ID) or instance.get(DEEPLOY_KEYS.PLUGIN_INSTANCE_ID)
+        discovered_instances.append({
+          DEEPLOY_PLUGIN_DATA.APP_ID: pipeline_app_id,
+          DEEPLOY_PLUGIN_DATA.INSTANCE_ID: instance_id,
+          DEEPLOY_PLUGIN_DATA.PLUGIN_SIGNATURE: signature,
+          DEEPLOY_PLUGIN_DATA.PLUGIN_INSTANCE: {
+            "instance_conf": self.deepcopy(instance),
+          },
+          DEEPLOY_PLUGIN_DATA.NODE: nodes[0] if nodes else None,
+          DEEPLOY_PLUGIN_DATA.CHAINSTORE_RESPONSE_KEY: instance.get(ct.BIZ_PLUGIN_DATA.CHAINSTORE_RESPONSE_KEY),
+        })
+
+    if not discovered_instances:
+      raise ValueError(f"{DEEPLOY_ERRORS.NODES3}: Persisted pipeline '{pipeline_cid}' has no plugin instances.")
+
+    return {
+      "discovered_instances": discovered_instances,
+      "nodes": nodes,
+      "discovered_nodes": [],
+      "deeploy_specs": deeploy_specs,
+      "from_persisted_pipeline": True,
+    }
+
   def __handle_error(self, exc, request, extra_error_code=DEEPLOY_ERRORS.GENERIC):
     """
     Handle the error and return a response.
@@ -658,6 +737,7 @@ def _process_pipeline_request(
       prepared_create_deploy_plan = None
       skip_create_response_key_reset = False
       previous_pipeline_cid = None
+      update_context_from_persisted_pipeline = False
       if is_create:
         is_valid = self.deeploy_check_payment_and_job_owner(inputs, auth_result[DEEPLOY_KEYS.ESCROW_OWNER], is_create=is_create, debug=self.cfg_deeploy_verbose > 1)
         if not is_valid:
@@ -677,11 +757,22 @@ def _process_pipeline_request(
         nodes_changed = True
       else:
         # Discover the live deployment so we can validate node affinity and reuse existing specs.
-        pipeline_context = self._gather_running_pipeline_context(
-          owner=auth_result[DEEPLOY_KEYS.ESCROW_OWNER],
-          app_id=app_id,
-          job_id=job_id,
-        )
+        try:
+          pipeline_context = self._gather_running_pipeline_context(
+            owner=auth_result[DEEPLOY_KEYS.ESCROW_OWNER],
+            app_id=app_id,
+            job_id=job_id,
+          )
+        except ValueError as exc:
+          if DEEPLOY_ERRORS.NODES3 not in str(exc):
+            raise
+          pipeline_context = self._gather_persisted_pipeline_update_context(
+            owner=auth_result[DEEPLOY_KEYS.ESCROW_OWNER],
+            app_id=app_id,
+            job_id=job_id,
+            project_id=inputs.get(DEEPLOY_KEYS.PROJECT_ID, None),
+          )
+          update_context_from_persisted_pipeline = True
         discovered_plugin_instances = pipeline_context["discovered_instances"]
         current_nodes = pipeline_context["nodes"]
         deeploy_specs_for_update = pipeline_context["deeploy_specs"]
@@ -850,12 +941,20 @@ def _process_pipeline_request(
           skip_create_response_key_reset = True
 
         # All validations and response-key resets passed; remove the running job and redeploy.
-        self.delete_pipeline_from_nodes(
-          app_id=app_id,
-          job_id=job_id,
-          owner=auth_result[DEEPLOY_KEYS.ESCROW_OWNER],
-          discovered_instances=discovered_plugin_instances,
-        )
+        if update_context_from_persisted_pipeline:
+          # TODO: stop stale offline old-node pipelines through ChainDist reconciliation when they return.
+          self.Pd(
+            f"Skipping live pipeline stop for offline update fallback on job_id={job_id}; "
+            "new pipeline will be deployed to validated target nodes.",
+            color='y',
+          )
+        else:
+          self.delete_pipeline_from_nodes(
+            app_id=app_id,
+            job_id=job_id,
+            owner=auth_result[DEEPLOY_KEYS.ESCROW_OWNER],
+            discovered_instances=discovered_plugin_instances,
+          )
 
         deployment_nodes = list(validated_nodes)
         confirmation_nodes = list(validated_nodes)

extensions/business/deeploy/tests/test_update_requests.py

@@ -32,6 +32,7 @@ def decorator(fn):
 from extensions.business.deeploy.deeploy_const import (
   DEEPLOY_DYNAMIC_ENV_KEYS,
   DEEPLOY_DYNAMIC_ENV_TYPES,
+  DEEPLOY_ERRORS,
   DEEPLOY_KEYS,
   DEEPLOY_PLUGIN_DATA,
   DEEPLOY_STATUS,
@@ -80,7 +81,11 @@ def _make_process_update_plugin(self, discovered_instances, nodes=None, deeploy_
     plugin._ensure_plugin_instance_ids = lambda *args, **kwargs: None
     plugin._check_nodes_availability = lambda inputs: nodes or ["node-1"]
 
-    called = {"delete": 0, "deploy": 0, "deploy_kwargs": None, "queued": 0}
+    called = {"delete": 0, "deploy": 0, "deploy_kwargs": None, "queued": 0, "bc_update": 0}
+    plugin.bc = types.SimpleNamespace(
+      node_addr_to_eth_addr=lambda node: node,
+      submit_node_update=lambda **kwargs: called.__setitem__("bc_update", called["bc_update"] + 1),
+    )
     plugin.delete_pipeline_from_nodes = lambda **kwargs: called.__setitem__("delete", called["delete"] + 1)
 
     def check_and_deploy_pipelines(**kwargs):
@@ -985,6 +990,80 @@ def check_nodes_availability(inputs):
     self.assertEqual(called["deploy"], 1)
     self.assertEqual([context for context, _, _ in validation_calls], ["payment", "nodes"])
 
+  def test_process_update_uses_persisted_pipeline_when_all_old_nodes_are_offline(self):
+    plugin, called = self._make_process_update_plugin(
+      discovered_instances=[],
+      nodes=["old-node-1", "old-node-2"],
+      deeploy_specs={
+        DEEPLOY_KEYS.JOB_ID: 11,
+        DEEPLOY_KEYS.PROJECT_ID: "0xProject",
+        DEEPLOY_KEYS.CURRENT_TARGET_NODES: ["old-node-1", "old-node-2"],
+        DEEPLOY_KEYS.JOB_APP_TYPE: "generic",
+      },
+    )
+    plugin._gather_running_pipeline_context = lambda **kwargs: (_ for _ in ()).throw(
+      ValueError(f"{DEEPLOY_ERRORS.NODES3}: No running workers found")
+    )
+    plugin._get_pipeline_from_cstore = lambda job_id: "cid-old-pipeline"
+    plugin.get_pipeline_from_r1fs = lambda *args, **kwargs: {
+      "NAME": "app-123",
+      "OWNER": "0xOwner",
+      "DEEPLOY_SPECS": {
+        DEEPLOY_KEYS.JOB_ID: 11,
+        DEEPLOY_KEYS.PROJECT_ID: "0xProject",
+        DEEPLOY_KEYS.CURRENT_TARGET_NODES: ["old-node-1", "old-node-2"],
+        DEEPLOY_KEYS.JOB_APP_TYPE: "generic",
+      },
+      "PLUGINS": [
+        {
+          plugin.ct.CONFIG_PLUGIN.K_SIGNATURE: "CONTAINER_APP_RUNNER",
+          plugin.ct.CONFIG_PLUGIN.K_INSTANCES: [
+            {
+              plugin.ct.CONFIG_INSTANCE.K_INSTANCE_ID: "current-instance",
+              DEEPLOY_KEYS.PLUGIN_NAME: "worker",
+              "IMAGE": "repo/app:1.0",
+              "CONTAINER_RESOURCES": {"cpu": 1, "memory": "256m"},
+            },
+          ],
+        },
+      ],
+    }
+    plugin._check_nodes_availability = lambda inputs: ["new-node-1"]
+
+    response = plugin._process_pipeline_request(
+      {
+        DEEPLOY_KEYS.APP_ID: "app-123",
+        DEEPLOY_KEYS.APP_ALIAS: "app",
+        DEEPLOY_KEYS.JOB_ID: 11,
+        DEEPLOY_KEYS.PROJECT_ID: "0xProject",
+        DEEPLOY_KEYS.JOB_APP_TYPE: "generic",
+        DEEPLOY_KEYS.PIPELINE_INPUT_TYPE: "void",
+        DEEPLOY_KEYS.CHAINSTORE_RESPONSE: False,
+        DEEPLOY_KEYS.TARGET_NODES: ["new-node-1"],
+        DEEPLOY_KEYS.TARGET_NODES_COUNT: 1,
+        DEEPLOY_KEYS.PLUGINS: [
+          {
+            DEEPLOY_KEYS.PLUGIN_SIGNATURE: "CONTAINER_APP_RUNNER",
+            DEEPLOY_KEYS.PLUGIN_INSTANCE_ID: "current-instance",
+            DEEPLOY_KEYS.PLUGIN_NAME: "worker",
+            "IMAGE": "repo/app:2.0",
+            "CONTAINER_RESOURCES": {"cpu": 1, "memory": "256m"},
+          },
+        ],
+      },
+      is_create=False,
+      async_mode=True,
+    )
+
+    self.assertEqual(response[DEEPLOY_KEYS.STATUS], DEEPLOY_STATUS.COMMAND_DELIVERED)
+    self.assertEqual(called["delete"], 0)
+    self.assertEqual(called["deploy"], 1)
+    self.assertEqual(called["queued"], 1)
+    self.assertEqual(called["bc_update"], 1)
+    self.assertEqual(called["deploy_kwargs"]["new_nodes"], ["new-node-1"])
+    redeploy_plugins = called["deploy_kwargs"]["inputs"][DEEPLOY_KEYS.PLUGINS]
+    self.assertEqual(redeploy_plugins[0]["IMAGE"], "repo/app:2.0")
+
   def test_process_update_rejects_job_app_type_change_before_payment_or_delete(self):
     plugin, called = self._make_process_update_plugin(
       discovered_instances=[

ver.py

@@ -1 +1 @@
-__VER__ = '2.10.380'
+__VER__ = '2.10.390'