From 59a367b88d8ea01972a996952447f2f189c5d97e Mon Sep 17 00:00:00 2001
From: Cristi <cristian.ciobanu.it@gmail.com>
Date: Fri, 18 Sep 2026 15:06:17 +0300
Subject: [PATCH] Allow projects containing missing images to open in a
 degraded state instead of failing to load

---
 gns3server/controller/node.py             | 213 ++++++++++++++-
 gns3server/controller/project.py          |  91 ++++++-
 gns3server/controller/udp_link.py         |  13 +
 gns3server/schemas/controller/nodes.py    |  17 ++
 tests/api/routes/controller/test_nodes.py |  13 +
 tests/controller/test_node.py             | 313 +++++++++++++++++++++-
 tests/controller/test_project.py          | 243 ++++++++++++++++-
 tests/controller/test_udp_link.py         |  38 +++
 8 files changed, 923 insertions(+), 18 deletions(-)

diff --git a/gns3server/controller/node.py b/gns3server/controller/node.py
index 452dfb984..8b7b83193 100644
--- a/gns3server/controller/node.py
+++ b/gns3server/controller/node.py
@@ -40,6 +40,26 @@ import logging
 log = logging.getLogger(__name__)
 
 
+# Image properties of a node, mapped by node type. The value is the image
+# "type" used by the image manager (see utils.images). Docker images are not
+# registered in the image database, they are matched by tag instead.
+IMAGE_PROPERTIES_BY_NODE_TYPE = {
+    "qemu": {
+        "hda_disk_image": "qemu",
+        "hdb_disk_image": "qemu",
+        "hdc_disk_image": "qemu",
+        "hdd_disk_image": "qemu",
+        "cdrom_image": "qemu",
+        "initrd": "qemu",
+        "kernel_image": "qemu",
+        "bios_image": "qemu",
+    },
+    "dynamips": {"image": "ios"},
+    "iou": {"path": "iou"},
+    "docker": {"image": "docker"},
+}
+
+
 def _extract_iol_startup_config_knob(environment):
     """
     Split the GNS3_IOL_STARTUP_CONFIG knob out of a Docker environment
@@ -143,6 +163,11 @@ class Node:
         self._netmiko_device_type = None
         self._default_username = None
         self._default_password = None
+        # Set when the node cannot be created on its compute because one or
+        # more of its images are missing. The node is kept on the controller
+        # (topology and links are preserved) until the user provides a
+        # compatible replacement image.
+        self._missing_images = []
 
         # This properties will be recomputed
         ignore_properties = ("width", "height", "hover_symbol")
@@ -187,6 +212,18 @@ class Node:
     def status(self):
         return self._status
 
+    @property
+    def missing_image(self):
+        """
+        :returns: True if at least one image required by this node is missing
+            and the node could therefore not be created on its compute.
+        """
+        return len(self._missing_images) > 0
+
+    @property
+    def missing_images(self):
+        return self._missing_images
+
     @property
     def template_id(self):
         return self._template_id
@@ -457,9 +494,14 @@ class Node:
     def links(self):
         return self._links
 
-    async def create(self):
+    async def create(self, allow_missing_image=False):
         """
         Create the node on the compute
+
+        :param allow_missing_image: when True, if an image is missing and
+            cannot be uploaded/synced, the node is not created on the compute
+            but is kept on the controller and flagged with the missing image(s)
+            instead of raising.
         """
         data = self._node_data()
         data["node_id"] = self._id
@@ -469,6 +511,8 @@ class Node:
         else:
             timeout = 1200
         trial = 0
+        last_missing_image = None
+        last_missing_error = None
         while trial != 6:
             try:
                 response = await self._compute.post(
@@ -477,17 +521,126 @@ class Node:
             except ComputeConflictError as e:
                 response = e.response()
                 if response.get("exception") == "ImageMissingError":
-                    res = await self._upload_missing_image(self._node_type, response["image"])
+                    last_missing_error = e
+                    last_missing_image = response.get("image")
+                    res = False
+                    if last_missing_image:
+                        try:
+                            res = await self._upload_missing_image(self._node_type, last_missing_image)
+                        except ControllerError as upload_error:
+                            # For Docker the image is pulled from a registry: a
+                            # failed pull (unknown tag, no network, private image)
+                            # must not abort the whole project. Treat it as a
+                            # missing image so the user can pick another one.
+                            if not allow_missing_image:
+                                raise
+                            log.warning(
+                                "Could not provide missing image '%s' for node '%s' [%s]: %s",
+                                last_missing_image, self._name, self._id, upload_error
+                            )
                     if not res:
+                        if allow_missing_image:
+                            missing_images = self._compute_missing_images(last_missing_image)
+                            # A degraded node must always identify at least one
+                            # missing image. Otherwise the controller would keep
+                            # a node that was never created on the compute while
+                            # exposing it as healthy.
+                            if not missing_images:
+                                raise e
+                            self._missing_images = missing_images
+                            log.warning(
+                                "Node '%s' [%s] is kept in degraded state, missing image(s): %s",
+                                self._name, self._id, ", ".join(m["image"] for m in self._missing_images)
+                            )
+                            return False
                         raise e
                 else:
                     raise e
             else:
                 await self.parse_node_response(response.json)
+                self._missing_images = []
                 return True
             trial += 1
+        if allow_missing_image:
+            # The image was repeatedly reported missing (e.g. the upload/sync
+            # seemed to succeed but did not). Keep the node in degraded state.
+            self._missing_images = self._compute_missing_images(last_missing_image)
+            log.warning(
+                "Node '%s' [%s] could not be created, missing image(s): %s",
+                self._name, self._id, ", ".join(m["image"] for m in self._missing_images)
+            )
+        elif last_missing_error is not None:
+            # Uploading appeared to succeed, but the compute rejected every
+            # retry. Do not let the caller register a controller-only node as
+            # healthy.
+            raise last_missing_error
         return False
 
+    def _image_available(self, image_type, image):
+        """
+        Check whether an image is available on the controller (and can
+        therefore be uploaded to the compute when needed).
+
+        :param image_type: image type (e.g. "qemu", "ios", "iou")
+        :param image: image filename or path
+        """
+
+        if not image:
+            return True
+        if image_type == "docker":
+            # Docker images are not stored on the controller filesystem
+            return True
+        try:
+            directories = images_directories(image_type)
+        except NotImplementedError:
+            return True
+        for directory in directories:
+            if os.path.exists(os.path.join(directory, image)):
+                return True
+        return False
+
+    def _compute_missing_images(self, fallback_image=None):
+        """
+        Build the list of images referenced by this node that are not
+        available on the controller.
+
+        :param fallback_image: image reported as missing by the compute, always
+            included (used for Docker images and images not directly mapped to
+            a node property).
+        """
+
+        missing = []
+        mapping = IMAGE_PROPERTIES_BY_NODE_TYPE.get(self._node_type, {})
+        properties = self._properties or {}
+        for prop, image_type in mapping.items():
+            value = properties.get(prop)
+            if not value:
+                continue
+            if not self._image_available(image_type, value):
+                missing.append({"property": prop, "image": value, "image_type": image_type})
+        if fallback_image:
+            prop = next((p for p in mapping if properties.get(p) == fallback_image), None)
+            if prop is None:
+                fallback_basename = os.path.basename(fallback_image)
+                prop = next(
+                    (
+                        p
+                        for p in mapping
+                        if properties.get(p) and os.path.basename(properties[p]) == fallback_basename
+                    ),
+                    None,
+                )
+            if prop is None and mapping:
+                prop = next(iter(mapping))
+            already_reported = any(
+                m["image"] == fallback_image or (prop is not None and m["property"] == prop)
+                for m in missing
+            )
+            if not already_reported:
+                image_type = mapping.get(prop, self._node_type)
+                missing.append({"property": prop, "image": fallback_image, "image_type": image_type})
+        return missing
+
     async def update(self, **kwargs):
         """
         Update the node on the compute
@@ -499,6 +652,9 @@ class Node:
         update_compute = False
         old_json = self.asdict()
         old_name = self._name
+        old_properties = copy.deepcopy(self._properties)
+        old_custom_adapters = copy.deepcopy(self._custom_adapters)
+        old_missing_images = copy.deepcopy(self._missing_images)
 
         compute_properties = None
         # Update node properties with additional elements
@@ -522,8 +678,32 @@ class Node:
         if compute_properties and "custom_adapters" in compute_properties:
             # we need to check custom adapters to update the custom port names
             self.custom_adapters = compute_properties["custom_adapters"]
+        if self.missing_image and compute_properties is not None:
+            # The node was never created on the compute (missing image). Apply
+            # the new properties locally so create() can use them.
+            self._properties = compute_properties
         self._list_ports()
         if update_compute:
+            if self.missing_image:
+                # Try to create the node on the compute now that an image may
+                # have been provided. If it is still missing the node simply
+                # remains in its degraded state.
+                try:
+                    created = await self.create(allow_missing_image=True)
+                except Exception:
+                    # create() can reject the replacement for reasons other
+                    # than a missing image. Keep the controller state aligned
+                    # with the node that is still absent from the compute.
+                    self._properties = old_properties
+                    self._custom_adapters = old_custom_adapters
+                    self._missing_images = old_missing_images
+                    self._list_ports()
+                    raise
+                if created:
+                    await self.project.restore_deferred_links(self)
+                self.project.emit_notification("node.updated", self.asdict())
+                self.project.dump()
+                return
             data = self._node_data(properties=compute_properties)
             try:
                 response = await self.put(None, data=data)
@@ -666,6 +846,25 @@ class Node:
         """
         Start a node
         """
+        if self.missing_image:
+            images = ", ".join(m["image"] for m in self._missing_images)
+            raise ControllerError(
+                f"Cannot start node '{self._name}': the required image(s) ({images}) "
+                f"are missing. Please provide a compatible image."
+            )
+        deferred_links = [link for link in self.links if link.deferred]
+        if deferred_links:
+            await self.project.restore_deferred_links(self)
+            failed_links = [
+                link
+                for link in deferred_links
+                if link.deferred and not any(endpoint["node"].missing_image for endpoint in link._nodes)
+            ]
+            if failed_links:
+                raise ControllerError(
+                    f"Cannot start node '{self._name}': {len(failed_links)} deferred link(s) "
+                    "could not be restored. Please try again."
+                )
         try:
             # For IOU: we need to send the licence everytime we start a node
             if self.node_type == "iou":
@@ -681,6 +880,8 @@ class Node:
         """
         Stop a node
         """
+        if self.missing_image:
+            return
         try:
             await self.post("/stop", timeout=240, dont_connect=True)
         # We don't care if a node is down at this step
@@ -756,6 +957,10 @@ class Node:
         """
         HTTP post on the node
         """
+        if self.missing_image and path is None:
+            # The node was never created on the compute (missing image). Any
+            # partial object there is cleaned up when the project is closed.
+            return None
         if path is None:
             return await self._compute.delete(
                 f"/projects/{self._project.id}/{self._node_type}/nodes/{self._id}", **kwargs
@@ -1018,7 +1223,9 @@ class Node:
             "status": self._status,
             "console_host": str(self._compute.console_host),
             "node_directory": self._node_directory,
-            "ports": [port.asdict() for port in self.ports]
+            "ports": [port.asdict() for port in self.ports],
+            "missing_image": self.missing_image,
+            "missing_images": self._missing_images,
         }
         topology.update(additional_data)
         return topology
diff --git a/gns3server/controller/project.py b/gns3server/controller/project.py
index b69a587b6..24511d40b 100644
--- a/gns3server/controller/project.py
+++ b/gns3server/controller/project.py
@@ -632,7 +632,7 @@ class Project:
         node = await self.add_node(compute, name, node_id, node_type=node_type, **template)
         return node
 
-    async def _create_node(self, compute, name, node_id, node_type=None, **kwargs):
+    async def _create_node(self, compute, name, node_id, node_type=None, allow_missing_image=False, **kwargs):
 
         node = Node(self, compute, name, node_id=node_id, node_type=node_type, **kwargs)
         # Hold the lock across the check + POST + register so that concurrent
@@ -651,17 +651,21 @@ class Project:
                 await compute.post("/projects", data=data)
                 self._project_created_on_compute.add(compute)
 
-        await node.create()
+        await node.create(allow_missing_image=allow_missing_image)
         self._nodes[node.id] = node
 
         return node
 
     @open_required
-    async def add_node(self, compute, name, node_id, dump=True, node_type=None, **kwargs):
+    async def add_node(
+        self, compute, name, node_id, dump=True, node_type=None, allow_missing_image=False, **kwargs
+    ):
         """
         Create a node or return an existing node
 
         :param dump: Dump topology to disk
+        :param allow_missing_image: Keep the node on the controller in a
+            degraded state instead of failing when its image is missing
         :param kwargs: See the documentation of node
         """
 
@@ -683,7 +687,9 @@ class Project:
                     )
                 elif "application_id" not in kwargs.keys() and not kwargs.get("properties"):
                     kwargs["application_id"] = get_next_application_id(self._controller.projects, self._computes)
-                node = await self._create_node(compute, name, node_id, node_type, **kwargs)
+                node = await self._create_node(
+                    compute, name, node_id, node_type, allow_missing_image=allow_missing_image, **kwargs
+                )
         elif node_type == "docker" and _is_iol_docker_kwargs(kwargs):
             # IOL Docker nodes derive interface MACs from the application ID
             # exactly like IOU; they draw from the disjoint upper half of the
@@ -700,9 +706,13 @@ class Project:
                     kwargs["application_id"] = get_next_application_id(
                         self._controller.projects, self._computes, iol_docker=True
                     )
-                node = await self._create_node(compute, name, node_id, node_type, **kwargs)
+                node = await self._create_node(
+                    compute, name, node_id, node_type, allow_missing_image=allow_missing_image, **kwargs
+                )
         else:
-            node = await self._create_node(compute, name, node_id, node_type, **kwargs)
+            node = await self._create_node(
+                compute, name, node_id, node_type, allow_missing_image=allow_missing_image, **kwargs
+            )
         self.emit_notification("node.created", node.asdict())
         if dump:
             self.dump()
@@ -975,6 +985,20 @@ class Project:
             # a link should have 2 attached nodes, this can happen with corrupted projects
             await self.delete_link(link.id, force_delete=True)
             return None
+        if any(n["node"].missing_image for n in link._nodes):
+            # One of the endpoints could not be created on its compute because
+            # an image is missing. Keep the link on the controller (so the
+            # topology is preserved) but defer the NIO creation until the
+            # missing image is resolved.
+            for n in link._nodes:
+                n["node"].add_link(link)
+                n["port"].link = link
+            link._deferred = True
+            log.info(
+                "Project '%s' [%s]: deferring link %s until missing image(s) are resolved",
+                self._name, self._id, link.id,
+            )
+            return None
         # Apply project-level marker definitions onto the link's memory
         # (memory_only) before _prepare() so the inherited markers ride the
         # batch NIO dispatch — zero extra HTTP round-trips.  The final
@@ -987,6 +1011,33 @@ class Project:
         entries = await link._prepare()
         return (link, entries)
 
+    async def restore_deferred_links(self, node):
+        """
+        Create on the computes the NIOs of the links that were deferred while
+        one of their endpoints had a missing image. Called once the node has
+        been successfully created.
+
+        :param node: node that has just been created on its compute
+        """
+
+        restored = []
+        for link in list(node.links):
+            if not link.deferred:
+                continue
+            if any(n["node"].missing_image for n in link._nodes):
+                # the other endpoint is still missing an image
+                continue
+            try:
+                await link.create()
+                link._deferred = False
+                restored.append(link)
+            except Exception as e:
+                log.exception("Could not restore deferred link %s: %s", link.id, e)
+        for link in restored:
+            self.emit_notification("link.updated", link.asdict())
+        if restored:
+            self.dump()
+
     @open_required
     async def add_link(self, link_id=None, dump=True):
         """
@@ -1867,7 +1918,15 @@ class Project:
             log.info("Project '%s' [%s]: loading %d nodes...", self._name, self._id, len(nodes_to_create))
             pool = Pool(concurrency=100)
             for compute, name, node_id, node_data in nodes_to_create:
-                pool.append(self.add_node, compute, name, node_id, dump=False, **node_data)
+                pool.append(
+                    self.add_node,
+                    compute,
+                    name,
+                    node_id,
+                    dump=False,
+                    allow_missing_image=True,
+                    **node_data,
+                )
             await pool.join()
             log.info("Project '%s' [%s]: loaded %d nodes", self._name, self._id, len(nodes_to_create))
             # Pre-allocate UDP ports for all links in batch to reduce HTTP round-trips
@@ -1875,9 +1934,12 @@ class Project:
             for link_data in topology.get("links", []):
                 if "link_id" not in link_data.keys():
                     continue
-                for node_link in link_data.get("nodes", []):
-                    node = self._nodes.get(node_link["node_id"])
-                    if node:
+                link_nodes = [self._nodes.get(nl["node_id"]) for nl in link_data.get("nodes", [])]
+                if any(node is not None and node.missing_image for node in link_nodes):
+                    # the link will be deferred, no NIO/port will be created now
+                    continue
+                for node in link_nodes:
+                    if node is not None:
                         ports_per_compute[node.compute.id] = ports_per_compute.get(node.compute.id, 0) + 1
             for compute in self.computes:
                 count = ports_per_compute.get(compute.id, 0)
@@ -2215,7 +2277,7 @@ class Project:
         """
         Start all nodes (except always-running types like Ethernet switch, Cloud, NAT, etc.)
         """
-        nodes_to_start = [n for n in self.nodes.values() if not n.is_always_running()]
+        nodes_to_start = [n for n in self.nodes.values() if not n.is_always_running() and not n.missing_image]
         if not nodes_to_start:
             return
         log.info("Project '%s' [%s]: starting %d nodes...", self._name, self._id, len(nodes_to_start))
@@ -2230,7 +2292,7 @@ class Project:
         """
         Stop all nodes (except always-running types like Ethernet switch, Cloud, NAT, etc.)
         """
-        nodes_to_stop = [n for n in self.nodes.values() if not n.is_always_running()]
+        nodes_to_stop = [n for n in self.nodes.values() if not n.is_always_running() and not n.missing_image]
         if not nodes_to_stop:
             return
         log.info("Project '%s' [%s]: stopping %d nodes...", self._name, self._id, len(nodes_to_stop))
@@ -2247,6 +2309,8 @@ class Project:
         """
         pool = Pool(concurrency=50)
         for node in self.nodes.values():
+            if node.missing_image:
+                continue
             pool.append(node.suspend)
         await pool.join()
 
@@ -2258,6 +2322,8 @@ class Project:
 
         pool = Pool(concurrency=3)
         for node in self.nodes.values():
+            if node.missing_image:
+                continue
             pool.append(node.reset_console)
         await pool.join()
 
@@ -2347,4 +2413,3 @@ class Project:
 
     def __repr__(self):
         return f"<gns3server.controller.Project {self._name} {self._id}>"
-
diff --git a/gns3server/controller/udp_link.py b/gns3server/controller/udp_link.py
index bc6cf2d52..766958bc4 100644
--- a/gns3server/controller/udp_link.py
+++ b/gns3server/controller/udp_link.py
@@ -42,6 +42,10 @@ class UDPLink(Link):
         super().__init__(project, link_id=link_id)
         self._created = False
         self._link_data = []
+        # True when the link could not be created on the computes because one
+        # of its endpoints has a missing image. The NIO creation is retried
+        # once the image is resolved.
+        self._deferred = False
 
     @property
     def debug_link_data(self):
@@ -49,6 +53,11 @@ class UDPLink(Link):
         Use for the debug exports
         """
         return self._link_data
+
+    @property
+    def deferred(self):
+        """Whether NIO creation is waiting for missing node images."""
+        return self._deferred
     
     def _get_node_filters(self, node1, node2):
         """
@@ -258,6 +267,10 @@ class UDPLink(Link):
         Delete the link and free the resources
         """
         if not self._created:
+            if self._deferred:
+                # There is no NIO on the computes to delete, but the local
+                # back-references created while loading must be cleared.
+                await super().delete()
             return
         try:
             node1 = self._nodes[0]["node"]
diff --git a/gns3server/schemas/controller/nodes.py b/gns3server/schemas/controller/nodes.py
index 85672ffb8..048ff1364 100644
--- a/gns3server/schemas/controller/nodes.py
+++ b/gns3server/schemas/controller/nodes.py
@@ -99,6 +99,16 @@ class NodePort(BaseModel):
     mac_address: Union[str, None] = Field(None, pattern="^([0-9a-fA-F]{2}[:]){5}([0-9a-fA-F]{2})$")
 
 
+class MissingImage(BaseModel):
+    """
+    A missing image referenced by a node.
+    """
+
+    property: Optional[str] = Field(None, description="Node property referencing the image")
+    image: str = Field(..., description="Requested image filename")
+    image_type: Optional[str] = Field(None, description="Type of image (qemu/ios/iou/docker)")
+
+
 class NodeBase(BaseModel):
     """
     Node data.
@@ -182,6 +192,13 @@ class Node(NodeBase):
         None,
         description="Console host. Warning if the host is 0.0.0.0 or :: (listen on all interfaces) you need to use the same address you use to connect to the controller",
     )
+    missing_image: bool = Field(
+        False,
+        description="True when the node could not be created on its compute because a required image is missing. Read only",
+    )
+    missing_images: List[MissingImage] = Field(
+        default_factory=list, description="List of missing images referenced by the node. Read only"
+    )
 
 
 class NodeDuplicate(BaseModel):
diff --git a/tests/api/routes/controller/test_nodes.py b/tests/api/routes/controller/test_nodes.py
index be115b1dd..6183503d7 100644
--- a/tests/api/routes/controller/test_nodes.py
+++ b/tests/api/routes/controller/test_nodes.py
@@ -103,7 +103,20 @@ class TestNodeRoutes:
         response = await client.get(app.url_path_for("get_nodes", project_id=project.id))
         assert response.status_code == status.HTTP_200_OK
         assert response.json()[0]["name"] == "test"
+    async def test_list_node_with_missing_image(
+        self, app: FastAPI, client: AsyncClient, project: Project, node: Node
+    ) -> None:
+        node._missing_images = [
+            {"property": "path", "image": "i86bi-linux-l2.bin", "image_type": "iou"}
+        ]
 
+        response = await client.get(app.url_path_for("get_nodes", project_id=project.id))
+        assert response.status_code == status.HTTP_200_OK
+        data = response.json()[0]
+        assert data["missing_image"] is True
+        assert data["missing_images"] == [
+            {"property": "path", "image": "i86bi-linux-l2.bin", "image_type": "iou"}
+        ]
 
     @pytest.mark.parametrize(
         "tags, expected_match",
diff --git a/tests/controller/test_node.py b/tests/controller/test_node.py
index b62784052..42fa889b5 100644
--- a/tests/controller/test_node.py
+++ b/tests/controller/test_node.py
@@ -26,6 +26,7 @@ from gns3server.compute.docker.docker_error import DockerError
 
 from gns3server.controller.node import Node
 from gns3server.controller.project import Project
+from gns3server.controller.controller_error import ComputeConflictError, ControllerError
 
 
 @pytest.fixture
@@ -234,7 +235,9 @@ def test_json(node, compute):
                 "port_number": 0,
                 "short_name": "e0"
             }
-        ]
+        ],
+        "missing_image": False,
+        "missing_images": []
     }
 
     assert node.asdict(topology_dump=True) == {
@@ -319,6 +322,314 @@ async def test_create_image_missing(node, compute):
     #assert node._upload_missing_image.called is True
 
 
+@pytest.mark.asyncio
+async def test_create_image_missing_kept_in_degraded_state(project, compute, tmpdir, config):
+    """
+    With allow_missing_image=True a node whose image cannot be provided is kept
+    on the controller and flagged instead of aborting the whole project open.
+    """
+
+    config.settings.Server.images_path = str(tmpdir)
+    node = Node(project, compute, "r1",
+                node_id=str(uuid.uuid4()),
+                node_type="qemu",
+                properties={"hda_disk_image": "missing.qcow2", "ram": 256})
+
+    async def resp(*args, **kwargs):
+        raise ComputeConflictError(
+            "/projects/{}/qemu/nodes".format(project.id),
+            {"message": "The image is missing", "image": "missing.qcow2", "exception": "ImageMissingError"}
+        )
+
+    compute.post = AsyncioMagicMock(side_effect=resp)
+    node._upload_missing_image = AsyncioMagicMock(return_value=False)
+
+    assert await node.create(allow_missing_image=True) is False
+    assert node.missing_image is True
+    assert node.missing_images == [
+        {"property": "hda_disk_image", "image": "missing.qcow2", "image_type": "qemu"}
+    ]
+
+
+@pytest.mark.asyncio
+async def test_create_image_missing_raises_by_default(project, compute, tmpdir, config):
+    """
+    Without allow_missing_image the historical behaviour is preserved: the
+    ImageMissingError is re-raised.
+    """
+
+    config.settings.Server.images_path = str(tmpdir)
+    node = Node(project, compute, "r1",
+                node_id=str(uuid.uuid4()),
+                node_type="qemu",
+                properties={"hda_disk_image": "missing.qcow2", "ram": 256})
+
+    async def resp(*args, **kwargs):
+        raise ComputeConflictError(
+            "/projects/{}/qemu/nodes".format(project.id),
+            {"message": "The image is missing", "image": "missing.qcow2", "exception": "ImageMissingError"}
+        )
+
+    compute.post = AsyncioMagicMock(side_effect=resp)
+    node._upload_missing_image = AsyncioMagicMock(return_value=False)
+
+    with pytest.raises(ComputeConflictError):
+        await node.create()
+    assert node.missing_image is False
+
+
+@pytest.mark.asyncio
+async def test_create_image_missing_raises_after_upload_retries(project, compute):
+    node = Node(
+        project,
+        compute,
+        "r1",
+        node_id=str(uuid.uuid4()),
+        node_type="qemu",
+        properties={"hda_disk_image": "missing.qcow2", "ram": 256},
+    )
+
+    async def resp(*args, **kwargs):
+        raise ComputeConflictError(
+            f"/projects/{project.id}/qemu/nodes",
+            {"message": "missing", "image": "missing.qcow2", "exception": "ImageMissingError"},
+        )
+
+    compute.post = AsyncioMagicMock(side_effect=resp)
+    node._upload_missing_image = AsyncioMagicMock(return_value=True)
+
+    with pytest.raises(ComputeConflictError):
+        await node.create()
+    assert compute.post.call_count == 6
+    assert node.missing_image is False
+
+
+@pytest.mark.asyncio
+async def test_create_image_missing_without_image_name_is_not_degraded(project, compute):
+    """A malformed conflict must not leave a controller-only node marked healthy."""
+
+    node = Node(
+        project,
+        compute,
+        "r1",
+        node_id=str(uuid.uuid4()),
+        node_type="qemu",
+        properties={"hda_disk_image": "present.qcow2", "ram": 256},
+    )
+
+    async def resp(*args, **kwargs):
+        raise ComputeConflictError(
+            f"/projects/{project.id}/qemu/nodes",
+            {"message": "The image is missing", "exception": "ImageMissingError"},
+        )
+
+    compute.post = AsyncioMagicMock(side_effect=resp)
+    node._image_available = MagicMock(return_value=True)
+
+    with pytest.raises(ComputeConflictError):
+        await node.create(allow_missing_image=True)
+    assert node.missing_image is False
+
+
+def test_compute_missing_images_matches_reported_basename_to_property(project, compute):
+    node = Node(
+        project,
+        compute,
+        "r1",
+        node_id=str(uuid.uuid4()),
+        node_type="qemu",
+        properties={
+            "hda_disk_image": "/images/qemu/disk.qcow2",
+            "cdrom_image": "/images/qemu/installer.iso",
+            "ram": 256,
+        },
+    )
+    node._image_available = MagicMock(return_value=True)
+
+    assert node._compute_missing_images("installer.iso") == [
+        {"property": "cdrom_image", "image": "installer.iso", "image_type": "qemu"}
+    ]
+
+
+def test_compute_missing_images_deduplicates_path_and_reported_basename(project, compute):
+    node = Node(
+        project,
+        compute,
+        "r1",
+        node_id=str(uuid.uuid4()),
+        node_type="qemu",
+        properties={"cdrom_image": "/images/qemu/installer.iso", "ram": 256},
+    )
+    node._image_available = MagicMock(return_value=False)
+
+    assert node._compute_missing_images("installer.iso") == [
+        {
+            "property": "cdrom_image",
+            "image": "/images/qemu/installer.iso",
+            "image_type": "qemu",
+        }
+    ]
+
+
+@pytest.mark.asyncio
+async def test_create_docker_image_missing_after_failed_pull(project, compute):
+    """
+    A Docker image that cannot be pulled from the registry keeps the node in a
+    degraded state instead of aborting the project open.
+    """
+
+    node = Node(project, compute, "web",
+                node_id=str(uuid.uuid4()),
+                node_type="docker",
+                properties={"image": "ghost:latest", "adapters": 1})
+
+    async def resp(*args, **kwargs):
+        raise ComputeConflictError(
+            "/projects/{}/docker/nodes".format(project.id),
+            {"message": "The image is missing", "image": "ghost:latest", "exception": "ImageMissingError"}
+        )
+
+    compute.post = AsyncioMagicMock(side_effect=resp)
+    node._upload_missing_image = AsyncioMagicMock(
+        side_effect=ControllerError("Failed to pull Docker image 'ghost:latest'")
+    )
+
+    assert await node.create(allow_missing_image=True) is False
+    assert node.missing_image is True
+    assert node.missing_images == [
+        {"property": "image", "image": "ghost:latest", "image_type": "docker"}
+    ]
+
+
+@pytest.mark.asyncio
+async def test_create_docker_image_missing_pull_error_raises_by_default(project, compute):
+    """
+    Without allow_missing_image a failed Docker pull is still surfaced.
+    """
+
+    node = Node(project, compute, "web",
+                node_id=str(uuid.uuid4()),
+                node_type="docker",
+                properties={"image": "ghost:latest", "adapters": 1})
+
+    async def resp(*args, **kwargs):
+        raise ComputeConflictError(
+            "/projects/{}/docker/nodes".format(project.id),
+            {"message": "The image is missing", "image": "ghost:latest", "exception": "ImageMissingError"}
+        )
+
+    compute.post = AsyncioMagicMock(side_effect=resp)
+    node._upload_missing_image = AsyncioMagicMock(
+        side_effect=ControllerError("Failed to pull Docker image 'ghost:latest'")
+    )
+
+    with pytest.raises(ControllerError):
+        await node.create()
+
+
+def test_compute_missing_images_lists_all_unavailable_slots(project, compute, tmpdir, config):
+    """
+    All the image slots of a multi-image node are reported, not only the first
+    one the compute complained about.
+    """
+
+    config.settings.Server.images_path = str(tmpdir)
+    node = Node(project, compute, "r1",
+                node_id=str(uuid.uuid4()),
+                node_type="qemu",
+                properties={
+                    "hda_disk_image": "present.qcow2",
+                    "hdc_disk_image": "missing.qcow2",
+                    "initrd": "missing.initrd",
+                    "ram": 256,
+                })
+    # make one image available on the controller
+    os.makedirs(os.path.join(str(tmpdir), "QEMU"), exist_ok=True)
+    with open(os.path.join(str(tmpdir), "QEMU", "present.qcow2"), "w") as f:
+        f.write("x")
+
+    missing = node._compute_missing_images()
+    assert {m["property"] for m in missing} == {"hdc_disk_image", "initrd"}
+    assert all(m["image_type"] == "qemu" for m in missing)
+
+
+@pytest.mark.asyncio
+async def test_start_missing_image(node):
+
+    node._missing_images = [
+        {"property": "hda_disk_image", "image": "missing.qcow2", "image_type": "qemu"}
+    ]
+    with pytest.raises(ControllerError):
+        await node.start()
+
+
+@pytest.mark.asyncio
+async def test_update_recreates_missing_image_node(project, compute, node):
+    """
+    Updating the image of a degraded node creates it on the compute and clears
+    the missing image state.
+    """
+
+    node._missing_images = [
+        {"property": "image", "image": "missing.image", "image_type": "ios"}
+    ]
+    response = MagicMock()
+    response.json = {"console": 2048}
+    compute.post = AsyncioMagicMock(return_value=response)
+    project.restore_deferred_links = AsyncioMagicMock()
+
+    await node.update(properties={"image": "available.image"})
+    assert node.missing_image is False
+    assert node.missing_images == []
+    assert project.restore_deferred_links.called is True
+
+
+@pytest.mark.asyncio
+async def test_update_missing_image_node_rolls_back_after_create_error(project, compute, node):
+    original_properties = {"image": "missing.image"}
+    node._properties = original_properties.copy()
+    node._missing_images = [
+        {"property": "image", "image": "missing.image", "image_type": "ios"}
+    ]
+    compute.post = AsyncioMagicMock(side_effect=ControllerError("create failed"))
+
+    with pytest.raises(ControllerError):
+        await node.update(properties={"image": "invalid.image"})
+
+    assert node.properties == original_properties
+    assert node.missing_images == [
+        {"property": "image", "image": "missing.image", "image_type": "ios"}
+    ]
+
+
+@pytest.mark.asyncio
+async def test_start_retries_ready_deferred_links(node, project):
+    peer = MagicMock()
+    peer.missing_image = False
+    link = MagicMock()
+    link.deferred = True
+    link._nodes = [{"node": node}, {"node": peer}]
+    node._links.add(link)
+    project.restore_deferred_links = AsyncioMagicMock()
+
+    with pytest.raises(ControllerError, match="deferred link"):
+        await node.start()
+
+    project.restore_deferred_links.assert_called_once_with(node)
+
+
+@pytest.mark.asyncio
+async def test_stop_missing_image_node_does_not_contact_compute(node, compute):
+    node._missing_images = [
+        {"property": "image", "image": "missing.image", "image_type": "ios"}
+    ]
+    compute.post = AsyncioMagicMock()
+
+    await node.stop()
+
+    compute.post.assert_not_called()
+
+
 @pytest.mark.asyncio
 async def test_create_base_script(node, config, compute, tmpdir):
 
diff --git a/tests/controller/test_project.py b/tests/controller/test_project.py
index 373d05e10..978a5d4d8 100644
--- a/tests/controller/test_project.py
+++ b/tests/controller/test_project.py
@@ -17,6 +17,7 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 import os
+import json
 import uuid
 import pytest
 import pytest_asyncio
@@ -28,7 +29,12 @@ from uuid import uuid4
 from gns3server.controller.project import Project
 from gns3server.controller.node import Node
 from gns3server.controller.ports.ethernet_port import EthernetPort
-from gns3server.controller.controller_error import ControllerError, ControllerNotFoundError, ControllerForbiddenError
+from gns3server.controller.controller_error import (
+    ControllerError,
+    ControllerNotFoundError,
+    ControllerForbiddenError,
+    ComputeConflictError,
+)
 from gns3server.config import Config
 
 
@@ -1006,6 +1012,19 @@ async def test_stop_all(project):
     assert len(compute.post.call_args_list) == 10
 
 
+@pytest.mark.asyncio
+async def test_stop_all_skips_nodes_with_missing_images(project):
+    node = MagicMock()
+    node.is_always_running.return_value = False
+    node.missing_image = True
+    node.stop = AsyncioMagicMock()
+    project._nodes[node.id] = node
+
+    await project.stop_all()
+
+    node.stop.assert_not_called()
+
+
 @pytest.mark.asyncio
 async def test_suspend_all(project):
 
@@ -1023,6 +1042,18 @@ async def test_suspend_all(project):
     assert len(compute.post.call_args_list) == 10
 
 
+@pytest.mark.asyncio
+async def test_suspend_all_skips_nodes_with_missing_images(project):
+    node = MagicMock()
+    node.missing_image = True
+    node.suspend = AsyncioMagicMock()
+    project._nodes[node.id] = node
+
+    await project.suspend_all()
+
+    node.suspend.assert_not_called()
+
+
 @pytest.mark.asyncio
 async def test_console_reset_all(project):
 
@@ -1040,6 +1071,18 @@ async def test_console_reset_all(project):
     assert len(compute.post.call_args_list) == 10
 
 
+@pytest.mark.asyncio
+async def test_console_reset_all_skips_nodes_with_missing_images(project):
+    node = MagicMock()
+    node.missing_image = True
+    node.reset_console = AsyncioMagicMock()
+    project._nodes[node.id] = node
+
+    await project.reset_console_all()
+
+    node.reset_console.assert_not_called()
+
+
 @pytest.mark.asyncio
 async def test_node_name(project):
 
@@ -1085,3 +1128,201 @@ async def test_duplicate_node(project):
         })
     new_node = await project.duplicate_node(original, 42, 10, 11)
     assert new_node.x == 42
+
+
+@pytest.mark.asyncio
+async def test_add_node_missing_image_kept_in_degraded_state(controller):
+    """
+    A node whose image is missing is kept on the controller with the missing
+    image list instead of failing, when allow_missing_image is set.
+    """
+
+    compute = MagicMock()
+    compute.id = "local"
+    compute.connected = True
+    project = Project(controller=controller, name="Test")
+    project.emit_notification = MagicMock()
+
+    async def post(url, data=None, **kwargs):
+        if url.endswith("/qemu/nodes"):
+            raise ComputeConflictError(
+                url, {"message": "missing", "image": "missing.qcow2", "exception": "ImageMissingError"}
+            )
+        response = MagicMock()
+        response.json = {}
+        return response
+
+    compute.post = AsyncioMagicMock(side_effect=post)
+
+    node = await project.add_node(
+        compute,
+        "r1",
+        None,
+        node_type="qemu",
+        allow_missing_image=True,
+        properties={"hda_disk_image": "missing.qcow2", "ram": 256},
+    )
+    assert node.missing_image is True
+    assert node.id in project._nodes
+    assert node.missing_images == [
+        {"property": "hda_disk_image", "image": "missing.qcow2", "image_type": "qemu"}
+    ]
+
+
+@pytest.mark.asyncio
+async def test_open_with_missing_image_defers_links(controller, projects_dir):
+    """
+    Opening a project with a missing image succeeds, keeps the degraded node
+    and its links in the topology, but does not create the NIOs.
+    """
+
+    project_id = "3c1be6f9-b4ba-4737-b209-63c47c23359f"
+    qemu_id = "11111111-1111-1111-1111-111111111111"
+    iou_id = "33333333-3333-3333-3333-333333333333"
+    vpcs_id = "22222222-2222-2222-2222-222222222222"
+    link_id = "5a3e3a64-e853-4055-9503-4a14e01290f1"
+
+    topology = {
+        "auto_close": True,
+        "auto_open": False,
+        "auto_start": False,
+        "name": "demo",
+        "project_id": project_id,
+        "revision": 5,
+        "topology": {
+            "computes": [],
+            "drawings": [],
+            "links": [
+                {
+                    "link_id": link_id,
+                    "nodes": [
+                        {"adapter_number": 0, "node_id": qemu_id, "port_number": 0},
+                        {"adapter_number": 0, "node_id": vpcs_id, "port_number": 0},
+                    ],
+                }
+            ],
+            "nodes": [
+                {
+                    "compute_id": "local",
+                    "name": "R1",
+                    "node_id": qemu_id,
+                    "node_type": "qemu",
+                    "properties": {"hda_disk_image": "missing.qcow2", "adapters": 1, "ram": 256},
+                    "symbol": ":/symbols/router.svg",
+                    "x": 0,
+                    "y": 0,
+                },
+                {
+                    "compute_id": "local",
+                    "name": "IOU1",
+                    "node_id": iou_id,
+                    "node_type": "iou",
+                    "properties": {"path": "missing.bin"},
+                    "symbol": ":/symbols/router.svg",
+                    "x": 200,
+                    "y": 0,
+                },
+                {
+                    "compute_id": "local",
+                    "name": "PC1",
+                    "node_id": vpcs_id,
+                    "node_type": "vpcs",
+                    "properties": {},
+                    "symbol": ":/symbols/computer.svg",
+                    "x": 100,
+                    "y": 0,
+                },
+            ],
+        },
+        "type": "topology",
+        "version": "2.0.0",
+    }
+
+    project_dir = os.path.join(projects_dir, "demo")
+    os.makedirs(project_dir, exist_ok=True)
+    with open(os.path.join(project_dir, "demo.gns3"), "w+") as f:
+        json.dump(topology, f)
+
+    compute = MagicMock()
+    compute.id = "local"
+    compute.connected = True
+    compute.name = "local"
+    compute.console_host = "127.0.0.1"
+    controller._computes["local"] = compute
+
+    async def post(url, data=None, **kwargs):
+        if url.endswith("/qemu/nodes"):
+            raise ComputeConflictError(
+                url, {"message": "missing", "image": "missing.qcow2", "exception": "ImageMissingError"}
+            )
+        if url.endswith("/iou/nodes"):
+            raise ComputeConflictError(
+                url, {"message": "missing", "image": "missing.bin", "exception": "ImageMissingError"}
+            )
+        response = MagicMock()
+        if "ports/udp/batch" in url:
+            response.json = {"udp_ports": [20000]}
+        else:
+            response.json = {}
+        return response
+
+    compute.post = AsyncioMagicMock(side_effect=post)
+
+    project = Project(
+        name="demo",
+        project_id=project_id,
+        path=project_dir,
+        controller=controller,
+        filename="demo.gns3",
+        status="closed",
+    )
+
+    await project.open()
+    assert project.status == "opened"
+
+    qemu_node = project.get_node(qemu_id)
+    assert qemu_node.missing_image is True
+    assert qemu_node.missing_images[0]["image"] == "missing.qcow2"
+
+    iou_node = project.get_node(iou_id)
+    assert iou_node.missing_image is True
+    assert iou_node.missing_images[0]["image"] == "missing.bin"
+    assert iou_node.missing_images[0]["image_type"] == "iou"
+
+    link = project._links[link_id]
+    assert link._deferred is True
+    assert link.created is False
+    assert link in qemu_node.links
+
+
+@pytest.mark.asyncio
+async def test_restore_deferred_links(project):
+    """
+    Once a degraded node is created, the deferred links are created on the
+    computes and no longer flagged as deferred.
+    """
+
+    compute = MagicMock()
+    compute.id = "local"
+    response = MagicMock()
+    response.json = {"console": 2048}
+    compute.post = AsyncioMagicMock(return_value=response)
+
+    node1 = await project.add_node(compute, "n1", None, node_type="vpcs", properties={})
+    node2 = await project.add_node(compute, "n2", None, node_type="vpcs", properties={})
+
+    link = await project.add_link()
+    await link.add_node(node1, 0, 0, batch=True)
+    await link.add_node(node2, 0, 0, batch=True)
+    link._nodes[0]["port"].link = link
+    link._nodes[1]["port"].link = link
+    node1.add_link(link)
+    node2.add_link(link)
+    link._deferred = True
+    link.create = AsyncioMagicMock()
+
+    project.emit_notification = MagicMock()
+    await project.restore_deferred_links(node1)
+
+    assert link._deferred is False
+    assert link.create.called is True
diff --git a/tests/controller/test_udp_link.py b/tests/controller/test_udp_link.py
index e5045cc36..f8934b1b4 100644
--- a/tests/controller/test_udp_link.py
+++ b/tests/controller/test_udp_link.py
@@ -187,6 +187,44 @@ async def test_delete(project):
     compute2.delete.assert_any_call("/projects/{}/vpcs/nodes/{}/adapters/3/ports/1/nio".format(project.id, node2.id), timeout=120)
 
 
+@pytest.mark.asyncio
+async def test_delete_deferred_link_clears_local_references(project):
+    node1 = Node(project, MagicMock(), "node1", node_type="vpcs")
+    node1._ports = [EthernetPort("E0", 0, 0, 0)]
+    node2 = Node(project, MagicMock(), "node2", node_type="vpcs")
+    node2._ports = [EthernetPort("E0", 0, 0, 0)]
+    link = UDPLink(project)
+
+    await link.add_node(node1, 0, 0, batch=True)
+    await link.add_node(node2, 0, 0, batch=True)
+    for entry in link._nodes:
+        entry["node"].add_link(link)
+        entry["port"].link = link
+    link._deferred = True
+
+    await link.delete()
+
+    assert link not in node1.links
+    assert link not in node2.links
+    assert node1.get_port(0, 0).link is None
+    assert node2.get_port(0, 0).link is None
+
+
+@pytest.mark.asyncio
+async def test_delete_uncreated_non_deferred_link_preserves_existing_behavior(project):
+    node = Node(project, MagicMock(), "node1", node_type="vpcs")
+    node._ports = [EthernetPort("E0", 0, 0, 0)]
+    link = UDPLink(project)
+    await link.add_node(node, 0, 0, batch=True)
+    node.add_link(link)
+    node.get_port(0, 0).link = link
+
+    await link.delete()
+
+    assert link in node.links
+    assert node.get_port(0, 0).link is link
+
+
 @pytest.mark.asyncio
 async def test_reset(project):
     """