diff --git a/gns3server/api/routes/compute/docker_nodes.py b/gns3server/api/routes/compute/docker_nodes.py
index 4c7f319c3..a5e48c57d 100644
--- a/gns3server/api/routes/compute/docker_nodes.py
+++ b/gns3server/api/routes/compute/docker_nodes.py
@@ -81,6 +81,7 @@ async def create_docker_node(project_id: UUID, node_data: schemas.DockerCreate)
         extra_configs=node_data.get("extra_configs"),
         memory=node_data.get("memory", 0),
         cpus=node_data.get("cpus", 0),
+        image_digest=node_data.get("image_digest"),
     )
     # Pop keys already consumed by create_node above so the setattr
     # fallback loop below only applies truly extra keys and does not
@@ -89,7 +90,7 @@ async def create_docker_node(project_id: UUID, node_data: schemas.DockerCreate)
         "console", "console_type", "console_resolution", "console_http_port",
         "console_http_path", "aux", "aux_type", "start_command", "environment",
         "adapters", "mac_address", "extra_hosts", "extra_volumes", "extra_configs",
-        "memory", "cpus",
+        "memory", "cpus", "image_digest",
     ):
         node_data.pop(key, None)
     for name, value in node_data.items():
diff --git a/gns3server/compute/docker/docker_vm.py b/gns3server/compute/docker/docker_vm.py
index c364bc817..bfcc53a7e 100644
--- a/gns3server/compute/docker/docker_vm.py
+++ b/gns3server/compute/docker/docker_vm.py
@@ -122,6 +122,7 @@ class DockerVM(BaseNode):
         extra_configs=None,
         memory=0,
         cpus=0,
+        image_digest=None,
     ):
 
         if not is_rfc1123_hostname_valid(name):
@@ -135,6 +136,10 @@ class DockerVM(BaseNode):
         if ":" not in image:
             image = f"{image}:latest"
         self._image = image
+        # image id expected by the controller (set when the image is available on
+        # the controller host): a different local image under the same tag is
+        # treated as missing so the controller re-syncs the expected version
+        self._image_digest = image_digest
         # assign through the property setters so creation and updates apply
         # the same value normalization (e.g. "" -> None)
         self.start_command = start_command
@@ -583,6 +588,16 @@ class DockerVM(BaseNode):
         if image_infos is None:
             raise DockerError(f"Cannot get information for image '{self._image}', please try again.")
 
+        if self._image_digest and image_infos.get("Id") != self._image_digest:
+            # the tag exists but points to a different image (e.g. a moved :latest):
+            # report it as missing so the controller re-syncs the expected version
+            local_id = image_infos.get("Id")
+            log.info(
+                f"Image '{self._image}' version mismatch on this compute: "
+                f"local id '{local_id}' != expected '{self._image_digest}'"
+            )
+            raise ImageMissingError(self._image)
+
         available_cpus = psutil.cpu_count(logical=True)
         if self._cpus > available_cpus:
             raise DockerError(
diff --git a/gns3server/controller/node.py b/gns3server/controller/node.py
index b674d1e6b..452dfb984 100644
--- a/gns3server/controller/node.py
+++ b/gns3server/controller/node.py
@@ -465,6 +465,7 @@ class Node:
         data["node_id"] = self._id
         if self._node_type == "docker":
             timeout = None
+            await self._add_docker_image_digest(data)
         else:
             timeout = 1200
         trial = 0
@@ -788,6 +789,29 @@ class Node:
                 return True
         return False
 
+    async def _add_docker_image_digest(self, data):
+        """
+        Pin the Docker image of a node to the image id found on the Docker daemon
+        of the controller host, so that a compute holding a different image under
+        the same tag (e.g. a moved :latest) gets it re-synced instead of silently
+        reusing the stale copy. Not set when the image is not on the controller host.
+        """
+
+        from gns3server.compute.docker import Docker
+        from gns3server.compute.docker.docker_error import DockerError
+
+        image = data.get("image")
+        if not image:
+            return
+        try:
+            image_info = await Docker.instance().query("GET", f"images/{image}/json")
+        except DockerError:
+            # not available on the controller host: nothing to pin against
+            return
+        image_id = image_info.get("Id")
+        if image_id:
+            data["image_digest"] = image_id
+
     async def _sync_missing_docker_image(self, image):
         """
         Export a Docker image from the Docker daemon on the controller host and
diff --git a/gns3server/schemas/compute/docker_nodes.py b/gns3server/schemas/compute/docker_nodes.py
index e2023d407..d752b7586 100644
--- a/gns3server/schemas/compute/docker_nodes.py
+++ b/gns3server/schemas/compute/docker_nodes.py
@@ -75,6 +75,12 @@ class DockerCreate(DockerBase):
     application_id: Optional[int] = Field(
         None, ge=1, le=1022, description="IOL application ID for iol-runner images (allocated by the controller)"
     )
+    image_digest: Optional[str] = Field(
+        None, pattern=r"^sha256:[a-f0-9]{64}$",
+        description="Image id the controller expects for 'image' (sha256:<hex>, resolved from the Docker "
+                    "daemon on the controller host). When set and the compute holds a different image "
+                    "under the same tag, the image is reported as missing so the controller re-syncs it"
+    )
 
 
 class DockerUpdate(DockerBase):
diff --git a/tests/compute/docker/test_docker_vm.py b/tests/compute/docker/test_docker_vm.py
index a5d44d82b..51359b024 100644
--- a/tests/compute/docker/test_docker_vm.py
+++ b/tests/compute/docker/test_docker_vm.py
@@ -564,6 +564,38 @@ async def test_create_image_not_available(compute_project, manager):
         query_mock.assert_not_called()
 
 
+@pytest.mark.asyncio
+async def test_create_image_digest_match(compute_project, manager):
+
+    response = {
+        "Id": "sha256:" + "a" * 64,
+        "Warnings": []
+    }
+    with asyncio_patch("gns3server.compute.docker.Docker.query", return_value=response) as mock:
+        vm = DockerVM("test", str(uuid.uuid4()), compute_project, manager, "ubuntu:latest",
+                      image_digest="sha256:" + "a" * 64)
+        await vm.create()
+        # the last query is the container creation: the digest check let it through
+        assert mock.call_args[0] == ("POST", "containers/create?name={}".format(vm.docker_name))
+        assert vm._cid == "sha256:" + "a" * 64
+
+
+@pytest.mark.asyncio
+async def test_create_image_digest_mismatch(compute_project, manager):
+
+    response = {
+        "Id": "sha256:" + "b" * 64,
+        "Warnings": []
+    }
+    vm = DockerVM("test", str(uuid.uuid4()), compute_project, manager, "ubuntu:latest",
+                  image_digest="sha256:" + "a" * 64)
+    with asyncio_patch("gns3server.compute.docker.Docker.query", return_value=response) as query_mock:
+        with pytest.raises(ImageMissingError, match="ubuntu:latest"):
+            await vm.create()
+        # only the image inspect happened: no container was created from the stale image
+        query_mock.assert_called_once_with("GET", "images/ubuntu:latest/json")
+
+
 @pytest.mark.asyncio
 async def test_create_with_user(compute_project, manager):
 
diff --git a/tests/controller/test_node.py b/tests/controller/test_node.py
index b650470e7..b62784052 100644
--- a/tests/controller/test_node.py
+++ b/tests/controller/test_node.py
@@ -726,6 +726,28 @@ async def test_sync_missing_docker_image_pull_fallback(compute, controller):
     compute.post.assert_called_with("/docker/images/pull", data={"image": "nginx:latest"}, timeout=None)
 
 
+@pytest.mark.asyncio
+async def test_create_docker_node_pins_image_digest(compute, controller):
+
+    project = Project(str(uuid.uuid4()), controller=controller)
+    node = Node(project, compute, "demo",
+                node_id=str(uuid.uuid4()),
+                node_type="docker",
+                properties={"image": "gns3/frr:latest", "adapters": 1})
+
+    response = MagicMock()
+    response.status = 200
+    response.json = {}
+    compute.post = AsyncioMagicMock(return_value=response)
+
+    image_id = "sha256:" + "a" * 64
+    with asyncio_patch("gns3server.compute.docker.Docker.query", return_value={"Id": image_id}):
+        assert await node.create() is True
+    # the image id from the controller host daemon is pinned into the create payload
+    data = compute.post.call_args[1]["data"]
+    assert data["image_digest"] == image_id
+
+
 def test_update_label(node):
     """
     The text in label need to be always the