fix(elements): correct element status actualization after upgrade

Element statuses remained ACTIVE too long after an upgrade because
the em_incorrect_resource_statuses_view only compared resource
statuses without checking target/actual resource hashes. Now the
view considers a resource ACTIVE only when both the actual resource
status is ACTIVE and target/actual hashes match, otherwise the
resource is IN_PROGRESS.

Additionally:
- Set element status to IN_PROGRESS explicitly in Manifest.upgrade
  for better responsiveness
- Pass session to ElementIncorrectStatusesView.actualize_status
  for transactional consistency with resource status updates
- Propagate IN_PROGRESS status to Node when machine config changes
  in NodeBuilderService._update_machine
- Check machine images before transitioning to ACTIVE in
  PoolBuilderService._actualize_machine_status
- Add migration 0068 with the updated view definition
- Add unit tests for status actualization and Manifest.upgrade

Closes #515

Signed-off-by: Anton Kremenetsky <anton.kremenetsky@gmail.com>

Author: akremenetsky

exordos_core/compute/builders/node.py

@@ -120,6 +120,10 @@ def _update_machine(
         machine.status = nc.MachineStatus.IN_PROGRESS.value
         machine.update(force=force)
 
+        if target_node.status != nc.NodeStatus.IN_PROGRESS.value:
+            target_node.status = nc.NodeStatus.IN_PROGRESS.value
+            target_node.save()
+
     def _is_root_volume(self, volume: models.Volume) -> bool:
         return volume.index == 0
 

exordos_core/compute/builders/pool.py

@@ -421,8 +421,17 @@ def _actualize_machine_status(
             pool_machine.status == nc.MachineStatus.ACTIVE
             and guest_machine.status == nc.MachineStatus.ACTIVE
         ):
-            machine.status = nc.MachineStatus.ACTIVE.value
-            return
+            # Transition to ACTIVE only when the machine images are actual.
+            # Without this check the machine could be marked ACTIVE while
+            # pool_machine or guest_machine still use stale images from a
+            # previous version.
+            resolved_image = self._resolve_image(machine.image)
+            if (
+                machine.image == pool_machine.image
+                and resolved_image == guest_machine.image
+            ):
+                machine.status = nc.MachineStatus.ACTIVE.value
+                return
 
         # TODO(akremenetsky): Support more statuses
         machine.status = nc.MachineStatus.IN_PROGRESS.value

exordos_core/elements/dm/models.py

@@ -31,7 +31,6 @@
 from restalchemy.dm import relationships
 from restalchemy.dm import types as ra_types
 from restalchemy.dm import types_dynamic as ra_types_dyn
-from restalchemy.storage.sql import engines
 from restalchemy.storage.sql import orm
 
 from exordos_core.common import exceptions
@@ -209,6 +208,7 @@ def upgrade(self, session: tp.Any | None = None) -> tuple["Manifest", "Element"]
         element.api_version = self.api_version
         element.description = self.description
         element.manifest = self
+        element.status = Status.IN_PROGRESS.value
         element.save(session=session)
         return self.apply_element(element)
 
@@ -533,20 +533,18 @@ class ElementIncorrectStatusesView(
         read_only=True,
     )
 
-    def actualize_status(self):
-        engine = engines.engine_factory.get_engine()
-        with engine.session_manager() as s:
-            s.execute(
-                f"""
-                UPDATE {Element.__tablename__}
-                SET status = %s
-                WHERE uuid = %s;
-                """,
-                (
-                    self.actual_status,
-                    self.uuid,
-                ),
-            )
+    def actualize_status(self, session):
+        session.execute(
+            f"""
+            UPDATE {Element.__tablename__}
+            SET status = %s
+            WHERE uuid = %s;
+            """,
+            (
+                self.actual_status,
+                self.uuid,
+            ),
+        )
 
 
 class Requirement(

exordos_core/elements/services/builders.py

@@ -67,7 +67,7 @@ def _actualize_statuses(self, session):
                 em_status_model.api_status,
                 em_status_model.actual_status,
             )
-            em_status_model.actualize_status()
+            em_status_model.actualize_status(session)
 
     def _iteration(self):
         with contexts.Context().session_manager() as session:

exordos_core/tests/unit/compute/test_node_builder.py

@@ -0,0 +1,76 @@
+#    Copyright 2025 Genesis Corporation.
+#
+#    All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+from unittest import mock
+import uuid as sys_uuid
+
+from restalchemy.storage.sql import orm
+
+from exordos_core.compute import constants as nc
+from exordos_core.compute.builders import node as node_builder
+from exordos_core.compute.dm import models as compute_models
+
+
+class TestNodeBuilderService:
+    def test_update_machine_sets_node_status_to_in_progress(
+        self,
+        monkeypatch,
+    ):
+        service = node_builder.NodeBuilderService()
+        node_uuid = sys_uuid.uuid4()
+
+        target_node = mock.MagicMock()
+        target_node.uuid = node_uuid
+        target_node.cores = 2
+        target_node.ram = 1024
+        target_node.name = "target-name"
+        target_node.description = "target-desc"
+        target_node.hostname = "host-a"
+        target_node.status = nc.NodeStatus.ACTIVE.value
+
+        actual_node = mock.MagicMock()
+        actual_node.uuid = node_uuid
+        actual_node.cores = 4
+        actual_node.ram = 2048
+        actual_node.name = "actual-name"
+        actual_node.description = "actual-desc"
+        actual_node.hostname = "host-b"
+
+        machine = mock.MagicMock()
+        machine.cores = 1
+        machine.ram = 512
+        machine.name = "old-name"
+        machine.description = "old-desc"
+
+        def fake_get_one_or_none(self, *args, **kwargs):
+            if self.model_cls is compute_models.Machine:
+                return machine
+            raise NotImplementedError
+
+        monkeypatch.setattr(
+            orm.ObjectCollection, "get_one_or_none", fake_get_one_or_none
+        )
+
+        service._update_machine(target_node, actual_node)
+
+        assert machine.cores == target_node.cores
+        assert machine.ram == target_node.ram
+        assert machine.name == target_node.name
+        assert machine.description == target_node.description
+        assert machine.status == nc.MachineStatus.IN_PROGRESS.value
+        machine.update.assert_called_once_with(force=True)
+        assert target_node.status == nc.NodeStatus.IN_PROGRESS.value
+        target_node.save.assert_called_once_with()

exordos_core/tests/unit/elements/dm/test_element_engine.py

@@ -14,16 +14,19 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+from unittest import mock
 import uuid as sys_uuid
 
 import pytest
 
 from exordos_core.common import exceptions
 from exordos_core.elements.dm.models import Element
 from exordos_core.elements.dm.models import ElementEngine
+from exordos_core.elements.dm.models import ElementIncorrectStatusesView
 from exordos_core.elements.dm.models import Export
 from exordos_core.elements.dm.models import Manifest
 from exordos_core.elements.dm.models import Resource
+from exordos_core.elements.dm.models import Status
 from exordos_core.elements.dm.models import element_engine
 
 
@@ -768,3 +771,58 @@ def test_same_element_export_with_different_paths_returns_correct_resource(
         assert result is compute_resource
         assert result.value["type"] == "compute"
         assert result is not storage_resource
+
+
+class TestElementIncorrectStatusesView:
+    def test_actualize_status_uses_passed_session(self):
+        session = mock.MagicMock()
+        view = ElementIncorrectStatusesView(
+            uuid=sys_uuid.uuid4(),
+            actual_status=Status.IN_PROGRESS.value,
+        )
+
+        view.actualize_status(session)
+
+        session.execute.assert_called_once()
+        args, _ = session.execute.call_args
+        assert "UPDATE em_elements" in args[0]
+        assert args[1] == (view.actual_status, view.uuid)
+
+
+class TestManifestUpgrade:
+    def test_upgrade_sets_element_status_to_in_progress(self, monkeypatch):
+        manifest = Manifest(
+            uuid=sys_uuid.uuid4(),
+            name="test-element",
+            version="2.0.0",
+            api_version="v1",
+            description="test",
+        )
+        element = Element(
+            uuid=sys_uuid.uuid4(),
+            name="test-element",
+            version="1.0.0",
+            api_version="v1",
+            description="test",
+            status=Status.ACTIVE.value,
+        )
+
+        def fake_get_one_or_none(self, *args, **kwargs):
+            if self.model_cls is Element:
+                return element
+            raise NotImplementedError
+
+        from restalchemy.storage.sql import orm
+
+        monkeypatch.setattr(
+            orm.ObjectCollection, "get_one_or_none", fake_get_one_or_none
+        )
+        monkeypatch.setattr(element_engine, "load_from_database", lambda: None)
+        monkeypatch.setattr(Element, "save", lambda *args, **kwargs: None)
+        monkeypatch.setattr(
+            Manifest, "apply_element", lambda self, element: (self, element)
+        )
+
+        manifest.upgrade()
+
+        assert element.status == Status.IN_PROGRESS.value

migrations/0068-fix-resource-status-hash-check-437c89.py

@@ -0,0 +1,108 @@
+#    Copyright 2026 Genesis Corporation.
+#
+#    All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+from restalchemy.storage.sql import migrations
+
+
+class MigrationStep(migrations.AbstractMigrationStep):
+    def __init__(self):
+        self._depends = [
+            "0067-init-border-ec37b4.py",
+        ]
+
+    @property
+    def migration_id(self):
+        return "437c8950-3580-406a-aaae-48f9aeac42c2"
+
+    @property
+    def is_manual(self):
+        return False
+
+    def upgrade(self, session):
+        session.execute(
+            """
+                CREATE OR REPLACE VIEW "em_incorrect_resource_statuses_view" AS
+                SELECT
+                    "er"."uuid" AS "uuid",
+                    "er"."status" AS "current_status",
+                    (
+                        CASE
+                            WHEN "utr"."hash" IS NULL THEN "uar"."status"
+                            WHEN "uar"."status" = 'ACTIVE'
+                                AND "utr"."hash" = "uar"."hash" THEN 'ACTIVE'
+                            WHEN "uar"."status" IS NULL THEN NULL
+                            ELSE 'IN_PROGRESS'
+                        END
+                    )::varchar(32) AS "actual_status"
+                FROM
+                    "em_resources" "er"
+                LEFT JOIN (
+                    SELECT
+                        "uuid",
+                        "hash"
+                    FROM "ua_target_resources"
+                    WHERE "kind" LIKE 'em_%'
+                ) AS "utr"
+                    ON "er"."uuid" = "utr"."uuid"
+                LEFT JOIN (
+                    SELECT
+                        "uuid",
+                        "status",
+                        "hash"
+                    FROM "ua_actual_resources"
+                    WHERE "kind" LIKE 'em_%'
+                ) AS "uar"
+                    ON "er"."uuid" = "uar"."uuid"
+                WHERE
+                    "er"."status" IS DISTINCT FROM (
+                        CASE
+                            WHEN "utr"."hash" IS NULL THEN "uar"."status"
+                            WHEN "uar"."status" = 'ACTIVE'
+                                AND "utr"."hash" = "uar"."hash" THEN 'ACTIVE'
+                            WHEN "uar"."status" IS NULL THEN NULL
+                            ELSE 'IN_PROGRESS'
+                        END
+                    )::varchar(32);
+            """,
+            None,
+        )
+
+    def downgrade(self, session):
+        session.execute(
+            """
+                CREATE OR REPLACE VIEW "em_incorrect_resource_statuses_view" AS
+                SELECT
+                    "er"."uuid" AS "uuid",
+                    "er"."status" AS "current_status",
+                    "uar"."status" AS "actual_status"
+                FROM
+                    "em_resources" "er"
+                LEFT JOIN (
+                    SELECT
+                        "uuid",
+                        "status"
+                    FROM "ua_actual_resources"
+                    WHERE "kind" LIKE 'em_%'
+                ) AS "uar"
+                    ON "er"."uuid" = "uar"."uuid"
+                WHERE
+                    "er"."status" IS DISTINCT FROM "uar"."status";
+            """,
+            None,
+        )
+
+
+migration_step = MigrationStep()