Merge branch '3.1' into ruff/trivia-fixups

gns3server/api/routes/controller/images.py

@@ -21,12 +21,14 @@ API routes for images.
 import os
 import logging
 import urllib.parse
+import tempfile
 
-from fastapi import APIRouter, Request, Depends, status
+from fastapi import APIRouter, Request, Response, Depends, Query, status
 from fastapi.encoders import jsonable_encoder
 from starlette.requests import ClientDisconnect
 from sqlalchemy.orm.exc import MultipleResultsFound
-from typing import List, Optional
+from sqlalchemy.exc import SQLAlchemyError
+from typing import List, Optional, Literal
 
 from gns3server import schemas
 from gns3server.config import Config
@@ -43,6 +45,8 @@ from gns3server.db.repositories.images import ImagesRepository
 from gns3server.db.repositories.templates import TemplatesRepository
 from gns3server.db.repositories.rbac import RbacRepository
 from gns3server.controller import Controller
+from gns3server.services.image_reconciliation import get_image_reconciliation_service
+from gns3server.utils.image_inventory import contained_path, image_lock, publish_image, fingerprint, ImageLockBusy
 from gns3server.controller.controller_error import (
     ControllerError,
     ControllerNotFoundError,
@@ -59,6 +63,43 @@ log = logging.getLogger(__name__)
 router = APIRouter()
 
 
+def image_destination(image_path):
+    root = os.path.realpath(os.path.expanduser(Config.instance().settings.Server.images_path))
+    full_path = os.path.abspath(os.path.join(root, image_path))
+    if not contained_path(os.path.realpath(full_path), root):
+        raise ControllerForbiddenError(f"Cannot write image, '{image_path}' is forbidden")
+    return full_path
+
+
+@router.post(
+    "/sync",
+    response_model=schemas.ImageSyncJob,
+    status_code=status.HTTP_202_ACCEPTED,
+    dependencies=[Depends(has_privilege("Image.Allocate"))],
+)
+async def sync_images(options: schemas.ImageSyncRequest, request: Request, response: Response):
+    """Reconcile configured image directories without deleting files or references."""
+    try:
+        job = await get_image_reconciliation_service(request.app).start(**options.model_dump())
+    except ImageLockBusy:
+        raise ControllerError("Image synchronization is already running or shutting down")
+    response.headers["Location"] = str(request.url_for("get_image_sync_job", job_id=job["job_id"]))
+    return job
+
+
+@router.get(
+    "/sync/jobs/{job_id}", response_model=schemas.ImageSyncJob, dependencies=[Depends(has_privilege("Image.Audit"))]
+)
+async def get_image_sync_job(
+    job_id: str, request: Request, offset: int = Query(0, ge=0), limit: int = Query(100, ge=1, le=1000)
+):
+    """Get progress and a page of errors (first 1,000 errors retained per job)."""
+    job = await get_image_reconciliation_service(request.app).get_job(job_id, offset, limit)
+    if job is None:
+        raise ControllerNotFoundError(f"Image synchronization job '{job_id}' not found")
+    return job
+
+
 @router.post(
     "/qemu/{image_path:path}",
     response_model=schemas.Image,
@@ -69,7 +110,7 @@ async def create_qemu_image(
     image_path: str,
     image_data: schemas.QemuDiskImageCreate,
     images_repo: ImagesRepository = Depends(get_repository(ImagesRepository)),
-) -> models.Image:
+) -> Optional[models.Image]:
     """
     Create a new blank Qemu image.
 
@@ -83,39 +124,41 @@ async def create_qemu_image(
     disk_image_path = urllib.parse.unquote(image_path)
     image_dir, image_name = os.path.split(disk_image_path)
     # check if the path is within the default images directory
-    base_images_directory = os.path.expanduser(Config.instance().settings.Server.images_path)
-    full_path = os.path.abspath(os.path.join(base_images_directory, image_dir, image_name))
-    if os.path.commonprefix([base_images_directory, full_path]) != base_images_directory:
-        raise ControllerForbiddenError(f"Cannot write disk image, '{disk_image_path}' is forbidden")
+    disk_image_path = image_destination(disk_image_path)
 
     if not image_dir:
         # put the image in the default images directory for Qemu
         directory = default_images_directory(image_type="qemu")
         os.makedirs(directory, exist_ok=True)
-        disk_image_path = os.path.abspath(os.path.join(directory, disk_image_path))
+        disk_image_path = image_destination(os.path.join(directory, image_name))
 
-    if await images_repo.get_image(disk_image_path):
-        raise ControllerBadRequestError(f"Disk image '{disk_image_path}' already exists")
-
-    options = jsonable_encoder(image_data, exclude_unset=True)
-    # FIXME: should we have the create_disk_image in the compute code since
-    # this code is used to create images on the controller?
-    await Qemu.instance().create_disk_image(disk_image_path, options)
-
-    image_info = await read_image_info(disk_image_path, "qemu")
-
-    image = await images_repo.get_image(disk_image_path)
-    if image:
-        # the image has already been added to the database
-        return image
-    else:
-        return await images_repo.add_image(**image_info)
+    async with image_lock(disk_image_path):
+        if os.path.lexists(disk_image_path):
+            raise ControllerBadRequestError(f"Disk image '{disk_image_path}' already exists")
+        os.makedirs(os.path.dirname(disk_image_path), exist_ok=True)
+        fd, temporary = tempfile.mkstemp(prefix=".gns3-create-", suffix=".tmp", dir=os.path.dirname(disk_image_path))
+        os.close(fd)
+        try:
+            options = jsonable_encoder(image_data, exclude_unset=True)
+            await Qemu.instance().create_disk_image(temporary, options)
+            image_info = await read_image_info(temporary, "qemu", allow_raw_image=allow_raw_image)
+            publish_image(temporary, disk_image_path)
+            image_info.update(
+                path=disk_image_path, image_name=image_name, file_fingerprint=fingerprint(disk_image_path)
+            )
+            return await images_repo.save_verified_image(image_info)
+        except (OSError, InvalidImageError, SQLAlchemyError) as e:
+            raise ControllerError(f"Could not create disk image '{disk_image_path}': {e}") from e
+        finally:
+            if os.path.exists(temporary):
+                os.unlink(temporary)
 
 
 @router.get("", response_model=List[schemas.Image], dependencies=[Depends(has_privilege("Image.Audit"))])
 async def get_images(
     images_repo: ImagesRepository = Depends(get_repository(ImagesRepository)),
     image_type: Optional[schemas.ImageType] = None,
+    availability: Optional[Literal["unknown", "available", "missing", "unavailable", "invalid"]] = None,
 ) -> List[models.Image]:
     """
     Return all images.
@@ -123,7 +166,7 @@ async def get_images(
     Required privilege: Image.Audit
     """
 
-    return await images_repo.get_images(image_type)
+    return await images_repo.get_images(image_type, availability)
 
 
 @router.post(
@@ -153,10 +196,7 @@ async def upload_image(
     image_path = urllib.parse.unquote(image_path)
     image_dir, image_name = os.path.split(image_path)
     # check if the path is within the default images directory
-    base_images_directory = os.path.expanduser(Config.instance().settings.Server.images_path)
-    full_path = os.path.abspath(os.path.join(base_images_directory, image_dir, image_name))
-    if os.path.commonprefix([base_images_directory, full_path]) != base_images_directory:
-        raise ControllerForbiddenError(f"Cannot write image, '{image_path}' is forbidden")
+    full_path = image_destination(image_path)
 
     # If the client sends X-MD5-Checksum, check for a duplicate before consuming the upload stream
     checksum_header = request.headers.get("X-MD5-Checksum")
@@ -170,7 +210,10 @@ async def upload_image(
     try:
         allow_raw_image = Config.instance().settings.Server.allow_raw_images
         image = await write_image(image_path, full_path, request.stream(), images_repo, allow_raw_image=allow_raw_image)
-    except (OSError, InvalidImageError, ClientDisconnect) as e:
+    except (OSError, InvalidImageError, ClientDisconnect, SQLAlchemyError) as e:
+        service = getattr(request.app.state, "image_reconciliation", None)
+        if service:
+            service.dirty.set()
         raise ControllerError(f"Could not save image '{image_path}': {e}")
 
     if install_appliances:
@@ -206,7 +249,9 @@ async def prune_images(
     # a single pass over all projects' node properties protects every
     # referenced file name at once
     referenced_filenames = Controller.instance().collect_referenced_image_filenames()
-    await images_repo.prune_images(list(skip_images) + list(referenced_filenames))
+    await images_repo.prune_images(
+        list(skip_images) + list(referenced_filenames), is_in_use=Controller.instance().find_projects_using_image
+    )
 
 
 @router.post("/install", status_code=status.HTTP_200_OK, dependencies=[Depends(has_privilege("Image.Allocate"))])
@@ -228,6 +273,11 @@ async def install_images(
     skip_images = get_builtin_disks()
     images = await images_repo.get_images()
     for image in images:
+        if not await images_repo.is_usable(image):
+            skipped.append(
+                {"name": image.filename, "reason": "image is missing, changed or unreadable; synchronize images first"}
+            )
+            continue
         if skip_images and image.filename in skip_images:
             log.debug(f"Skipping image '{image.path}' for image installation")
             continue
@@ -307,11 +357,22 @@ async def delete_image(
     if project_names:
         raise ControllerError(f"Image '{image_path}' is used by one or more projects: {', '.join(project_names)}")
 
-    try:
-        os.remove(image.path)
-    except OSError:
-        log.warning(f"Could not delete image file {image.path}")
-
-    success = await images_repo.delete_image(image_path)
-    if not success:
-        raise ControllerError(f"Image '{image_path}' could not be deleted")
+    path = image.path
+    revision = (image.image_id, image.checksum, image.file_fingerprint)
+    async with image_lock(path):
+        image = await images_repo.get_image(path, refresh=True)
+        if image is None or (image.image_id, image.checksum, image.file_fingerprint) != revision:
+            raise ControllerError(f"Image '{image_path}' changed while waiting for deletion; refresh and retry")
+        # Recheck usage after waiting for a concurrent writer/scanner.
+        if await images_repo.get_image_templates(image.image_id) or Controller.instance().find_projects_using_image(
+            image.filename
+        ):
+            raise ControllerError(f"Image '{image_path}' is in use")
+        try:
+            os.remove(image.path)
+        except FileNotFoundError:
+            pass
+        except OSError as e:
+            raise ControllerError(f"Could not delete image file '{image.path}': {e}") from e
+        if not await images_repo.delete_image_exact(image.image_id):
+            raise ControllerError(f"Image '{image_path}' could not be deleted")

gns3server/controller/appliance_manager.py

@@ -35,6 +35,7 @@ from ..utils.http_client import HTTPClient
 from .controller_error import ControllerBadRequestError, ControllerNotFoundError, ControllerError
 from .appliance_to_template import ApplianceToTemplate
 from ..utils.images import InvalidImageError, write_image, read_image_info
+from ..utils.image_inventory import image_lock
 
 from gns3server import schemas
 from gns3server.schemas.controller.appliances import ApplianceModel
@@ -178,13 +179,16 @@ class ApplianceManager:
                             # check if the image is on disk but it not yet in the database
                             image_path = os.path.join(image_dir, appliance_file)
                             if os.path.exists(image_path):
-                                image_info = await read_image_info(image_path)
-                                if image_info.get("checksum") == image_checksum:
-                                    log.info(f"Adding image '{image_path}' to the database")
+                                async with image_lock(image_path):
+                                    image_info = await read_image_info(image_path, allow_raw_image=True)
+                                    if image_info["checksum"] != image_checksum:
+                                        raise ControllerError(
+                                            f"Image '{image_path}' does not match the appliance checksum"
+                                        )
                                     try:
-                                        await images_repo.add_image(**image_info)
+                                        await images_repo.save_verified_image(image_info)
                                     except SQLAlchemyError as e:
-                                        log.warning(f"Error while adding image '{image['path']}' to the database: {e}")
+                                        raise ControllerError(f"Could not register image '{image_path}': {e}") from e
                             else:
                                 # download the image if there is a direct download URL
                                 direct_download_url = image.get("direct_download_url")

gns3server/core/tasks.py

@@ -26,22 +26,23 @@ from gns3server.compute import MODULES
 from gns3server.compute.port_manager import PortManager
 from gns3server.compute.marker.marker_manager import MarkerManager
 from gns3server.utils.http_client import HTTPClient
-from gns3server.db.tasks import connect_to_db, get_computes, disconnect_from_db, discover_images_on_filesystem
+from gns3server.db.tasks import connect_to_db, get_computes, disconnect_from_db
+from gns3server.services.image_reconciliation import get_image_reconciliation_service
 
 
 import logging
 
 log = logging.getLogger(__name__)
 
-auto_discover_images_task_handle = None
-
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
 
     await startup(app)
-    yield
-    await shutdown(app)
+    try:
+        yield
+    finally:
+        await shutdown(app)
 
 
 async def startup(app: FastAPI) -> None:
@@ -66,18 +67,7 @@ async def startup(app: FastAPI) -> None:
 
     await Controller.instance().start(computes)
 
-    # Because with a large image collection
-    # without md5sum already computed we start the
-    # computing with server start
-    from gns3server.compute.qemu import Qemu
-
-    if Config.instance().settings.Server.auto_discover_images is True:
-        # Start the discovering new images on file system 5 seconds after the server has started
-        # to give it a chance to process API requests
-        global auto_discover_images_task_handle
-        auto_discover_images_task_handle = asyncio.get_event_loop().call_later(
-            5, lambda: asyncio.create_task(discover_images_on_filesystem(app))
-        )
+    get_image_reconciliation_service(app).start_background()
 
     for module in MODULES:
         log.debug(f"Loading module {module.__name__}")
@@ -107,8 +97,10 @@ async def shutdown(app: FastAPI) -> None:
     Tasks to be performed when the server is exiting.
     """
 
-    if auto_discover_images_task_handle is not None and not auto_discover_images_task_handle.cancelled():
-        auto_discover_images_task_handle.cancel()
+    service = getattr(app.state, "image_reconciliation", None)
+    if service is not None:
+        await service.close()
+        del app.state.image_reconciliation
     await HTTPClient.close_session()
     await MarkerManager.instance().stop()
     # Kill resident sharkd sessions (marker replay) and drop their /tmp

gns3server/db/models/__init__.py

@@ -21,7 +21,7 @@ from .users import User, UserGroup
 from .roles import Role
 from .privileges import Privilege
 from .computes import Compute
-from .images import Image
+from .images import Image, ImageSyncJob
 from .pools import Resource, ResourcePool
 from .llm_model_configs import LLMModelConfig
 from .api_keys import ApiKey

gns3server/db/models/images.py

@@ -15,7 +15,7 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
-from sqlalchemy import Table, Column, String, ForeignKey, BigInteger, Integer
+from sqlalchemy import Table, Column, String, ForeignKey, BigInteger, Integer, DateTime, JSON, Boolean
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from .base import Base, BaseTable, GUID
@@ -39,4 +39,21 @@ class Image(BaseTable):
     image_size = Column(BigInteger)
     checksum = Column(String, index=True)
     checksum_algorithm = Column(String)
+    availability = Column(String, nullable=False, default="unknown", server_default="unknown")
+    file_fingerprint = Column(String)
+    last_seen_at = Column(DateTime)
+    last_verified_at = Column(DateTime)
+    last_error = Column(String)
     templates = relationship("Template", secondary=image_template_map, back_populates="images")
+
+
+class ImageSyncJob(BaseTable):
+    __tablename__ = "image_sync_jobs"
+
+    job_id = Column(String, primary_key=True)
+    status = Column(String, nullable=False)
+    dry_run = Column(Boolean, nullable=False)
+    force_checksum = Column(Boolean, nullable=False)
+    finished_at = Column(DateTime)
+    counts = Column(JSON, nullable=False)
+    errors = Column(JSON, nullable=False)

gns3server/db/repositories/images.py

@@ -16,11 +16,15 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 import os
+import asyncio
+from datetime import datetime, timezone
 
-from typing import Optional, List, cast
+from typing import Optional, List, Callable, cast
 from sqlalchemy import select, delete, update
 from sqlalchemy.engine import CursorResult
 from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.exc import IntegrityError
+from gns3server.utils.image_inventory import fingerprint, image_lock, normalized_path
 
 from .base import BaseRepository
 
@@ -36,19 +40,21 @@ class ImagesRepository(BaseRepository):
 
         super().__init__(db_session)
 
-    async def get_image(self, image_path: str) -> Optional[models.Image]:
+    async def get_image(self, image_path: str, *, refresh: bool = False) -> Optional[models.Image]:
         """
         Get an image by its path.
         """
 
         image_dir, image_name = os.path.split(image_path)
-        if image_dir:
+        if os.path.isabs(image_path):
+            query = select(models.Image).where(models.Image.path == image_path)
+        elif image_dir:
             query = select(models.Image).where(
-                models.Image.filename == image_name, models.Image.path.endswith(image_path)
+                models.Image.filename == image_name, models.Image.path.endswith(os.sep + image_path, autoescape=True)
             )
         else:
             query = select(models.Image).where(models.Image.filename == image_name)
-        result = await self._db_session.execute(query)
+        result = await self._db_session.execute(query.execution_options(populate_existing=refresh))
         return result.scalars().one_or_none()
 
     async def get_image_by_checksum(self, checksum: str, image_dir: Optional[str] = None) -> Optional[models.Image]:
@@ -56,18 +62,29 @@ class ImagesRepository(BaseRepository):
         Get an image by its checksum.
         """
 
-        if image_dir:
-            query = select(models.Image).where(
-                models.Image.checksum == checksum, models.Image.path.startswith(image_dir)
-            )
-            result = await self._db_session.execute(query)
-            return result.scalars().one_or_none()
-        else:
-            query = select(models.Image).where(models.Image.checksum == checksum)
-            result = await self._db_session.execute(query)
-            return result.scalars().first()
+        query = select(models.Image).where(models.Image.checksum == checksum).order_by(models.Image.image_id)
+        result = await self._db_session.execute(query)
+        for image in result.scalars().all():
+            if image_dir and normalized_path(os.path.dirname(image.path)) != normalized_path(image_dir):
+                continue
+            if await self.is_usable(image):
+                return image
+        return None
 
-    async def get_images(self, image_type=None) -> List[models.Image]:
+    async def is_usable(self, image: models.Image) -> bool:
+        """Validate a checksum candidate without trusting stale catalog/sidecar data."""
+        from gns3server.utils.images import inspect_image_file, InvalidImageError
+
+        try:
+            current = await asyncio.to_thread(fingerprint, image.path)
+            if image.availability == "available" and image.file_fingerprint == current:
+                return True
+            info = await asyncio.to_thread(inspect_image_file, image.path, image.image_type, True)
+            return info["checksum"] == image.checksum and info["image_size"] == image.image_size
+        except (OSError, InvalidImageError):
+            return False
+
+    async def get_images(self, image_type=None, availability=None) -> List[models.Image]:
         """
         Get all images.
         """
@@ -76,6 +93,8 @@ class ImagesRepository(BaseRepository):
             query = select(models.Image).where(models.Image.image_type == image_type)
         else:
             query = select(models.Image)
+        if availability:
+            query = query.where(models.Image.availability == availability)
         result = await self._db_session.execute(query)
         return list(result.scalars().all())
 
@@ -89,7 +108,9 @@ class ImagesRepository(BaseRepository):
         result = await self._db_session.execute(query)
         return list(result.scalars().all())
 
-    async def add_image(self, image_name, image_type, image_size, path, checksum, checksum_algorithm) -> models.Image:
+    async def add_image(
+        self, image_name, image_type, image_size, path, checksum, checksum_algorithm, file_fingerprint=None
+    ) -> models.Image:
         """
         Create a new image.
         """
@@ -102,10 +123,16 @@ class ImagesRepository(BaseRepository):
             path=path,
             checksum=checksum,
             checksum_algorithm=checksum_algorithm,
+            file_fingerprint=file_fingerprint,
+            availability="available" if file_fingerprint else "unknown",
         )
 
         self._db_session.add(db_image)
-        await self._db_session.commit()
+        try:
+            await self._db_session.commit()
+        except Exception:
+            await self._db_session.rollback()
+            raise
         await self._db_session.refresh(db_image)
         return db_image
 
@@ -127,16 +154,91 @@ class ImagesRepository(BaseRepository):
             await self._db_session.refresh(image_db)  # force refresh of updated_at value
         return image_db
 
+    async def save_verified_image(self, info: dict) -> Optional[models.Image]:
+        """Upsert an exact path, preserving template associations and the image ID.
+
+        Callers coordinate publication/inspection with image_lock(). Each commit
+        is short, and a concurrent insert from a legacy caller is retried safely.
+        """
+        values = dict(info)
+        values["filename"] = values.pop("image_name")
+        values.update(
+            availability="available",
+            last_error=None,
+            last_seen_at=datetime.now(timezone.utc).replace(tzinfo=None),
+            last_verified_at=datetime.now(timezone.utc).replace(tzinfo=None),
+        )
+        for attempt in range(2):
+            try:
+                image = (
+                    await self._db_session.execute(
+                        select(models.Image)
+                        .where(models.Image.path == info["path"])
+                        .execution_options(populate_existing=True)
+                    )
+                ).scalar_one_or_none()
+                if image is None:
+                    # A legacy spelling (e.g. /images/QEMU/./disk.qcow2) can
+                    # refer to the same destination restored by an API upload.
+                    candidates = (
+                        (
+                            await self._db_session.execute(
+                                select(models.Image).where(models.Image.filename == values["filename"])
+                            )
+                        )
+                        .scalars()
+                        .all()
+                    )
+                    aliases = [
+                        candidate
+                        for candidate in candidates
+                        if normalized_path(candidate.path) == normalized_path(info["path"])
+                    ]
+                    if len(aliases) > 1:
+                        from sqlalchemy.exc import MultipleResultsFound
+
+                        raise MultipleResultsFound("Ambiguous image path aliases; manual review required")
+                    if aliases:
+                        image = aliases[0]
+                        values["path"] = image.path
+                if image is None:
+                    image = models.Image(**values)
+                    self._db_session.add(image)
+                else:
+                    for key, value in values.items():
+                        setattr(image, key, value)
+                await self._db_session.commit()
+                await self._db_session.refresh(image)
+                return image
+            except IntegrityError:
+                await self._db_session.rollback()
+                if attempt:
+                    raise
+            except Exception:
+                await self._db_session.rollback()
+                raise
+        return None
+
+    async def delete_image_exact(self, image_id: int) -> bool:
+        result = await self._db_session.execute(delete(models.Image).where(models.Image.image_id == image_id))
+        await self._db_session.commit()
+        return cast(CursorResult, result).rowcount > 0
+
     async def delete_image(self, image_path: str) -> bool:
         """
         Delete an image.
         """
 
         image_dir, image_name = os.path.split(image_path)
-        if image_dir:
+        if os.path.isabs(image_path):
+            query = delete(models.Image).where(models.Image.path == image_path)
+        elif image_dir:
             query = (
                 delete(models.Image)
-                .where(models.Image.filename == image_name, models.Image.path.endswith(image_path))
+                .where(
+                    models.Image.filename == image_name,
+                    models.Image.path.endswith(os.sep + image_path, autoescape=True),
+                )
                 .execution_options(synchronize_session=False)
             )
         else:
@@ -145,25 +247,45 @@ class ImagesRepository(BaseRepository):
         await self._db_session.commit()
         return cast(CursorResult, result).rowcount > 0
 
-    async def prune_images(self, skip_images: Optional[list[str]] = None) -> int:
+    async def prune_images(self, skip_images: Optional[list[str]] = None, is_in_use: Optional[Callable] = None) -> int:
         """
         Prune images not attached to any template.
         """
 
         query = select(models.Image).filter(~models.Image.templates.any())
         result = await self._db_session.execute(query)
-        images = result.scalars().all()
+        # Snapshot scalar values; commits can expire ORM instances.
+        images = [
+            (image.image_id, image.filename, image.path, image.checksum, image.file_fingerprint)
+            for image in result.scalars().all()
+        ]
         images_deleted = 0
-        for image in images:
-            if skip_images and image.filename in skip_images:
-                log.debug(f"Skipping image '{image.path}' for pruning")
+        errors = []
+        for image_id, filename, path, checksum, file_fingerprint in images:
+            if skip_images and filename in skip_images:
                 continue
-            try:
-                log.debug(f"Deleting image '{image.path}'")
-                os.remove(image.path)
-            except OSError:
-                log.warning(f"Could not delete image file {image.path}")
-            if await self.delete_image(image.path):
-                images_deleted += 1
+            async with image_lock(path):
+                current = await self.get_image(path, refresh=True)
+                if current is None or (current.image_id, current.checksum, current.file_fingerprint) != (
+                    image_id,
+                    checksum,
+                    file_fingerprint,
+                ):
+                    continue  # A concurrent request removed or replaced this image.
+                if await self.get_image_templates(image_id) or (is_in_use and is_in_use(filename)):
+                    continue
+                try:
+                    os.remove(path)
+                except FileNotFoundError:
+                    pass
+                except OSError as e:
+                    errors.append(f"{path}: {e}")
+                    continue
+                if await self.delete_image_exact(image_id):
+                    images_deleted += 1
         log.info(f"{images_deleted} image(s) have been deleted")
+        if errors:
+            from gns3server.controller.controller_error import ControllerError
+
+            raise ControllerError("Could not delete image files: " + "; ".join(errors))
         return images_deleted

gns3server/db/repositories/templates.py

@@ -31,6 +31,8 @@ from .base import BaseRepository
 
 import gns3server.db.models as models
 from gns3server import schemas
+from gns3server.utils.image_inventory import image_lock
+from gns3server.controller.controller_error import ControllerNotFoundError
 
 log = logging.getLogger(__name__)
 
@@ -136,18 +138,22 @@ class TemplatesRepository(BaseRepository):
             await self._db_session.refresh(db_template)
         return db_template
 
-    async def get_image(self, image_path: str) -> Optional[models.Image]:
+    async def get_image(self, image_path: str, *, include_unavailable: bool = False) -> Optional[models.Image]:
         """
         Get an image by its path.
         """
 
         image_dir, image_name = os.path.split(image_path)
-        if image_dir:
+        if os.path.isabs(image_path):
+            query = select(models.Image).where(models.Image.path == image_path)
+        elif image_dir:
             query = select(models.Image).where(
-                models.Image.filename == image_name, models.Image.path.endswith(image_path)
+                models.Image.filename == image_name, models.Image.path.endswith(os.sep + image_path, autoescape=True)
             )
         else:
             query = select(models.Image).where(models.Image.filename == image_name)
+        if not include_unavailable:
+            query = query.where(models.Image.availability.in_(["unknown", "available"]))
         query = query.order_by(models.Image.image_id)
         result = await self._db_session.execute(query)
         images = result.scalars().all()
@@ -164,20 +170,28 @@ class TemplatesRepository(BaseRepository):
         Add an image to template.
         """
 
-        query = (
-            select(models.Template)
-            .options(selectinload(models.Template.images))
-            .where(models.Template.template_id == template_id)
-        )
-        result = await self._db_session.execute(query)
-        template_in_db = result.scalars().first()
-        if not template_in_db:
-            return None
+        async with image_lock(image.path):
+            exists = (
+                await self._db_session.execute(
+                    select(models.Image.image_id).where(models.Image.image_id == image.image_id)
+                )
+            ).scalar_one_or_none()
+            if exists is None:
+                raise ControllerNotFoundError(f"Image '{image.path}' was removed while creating the template")
+            query = (
+                select(models.Template)
+                .options(selectinload(models.Template.images))
+                .where(models.Template.template_id == template_id)
+            )
+            result = await self._db_session.execute(query)
+            template_in_db = result.scalars().first()
+            if not template_in_db:
+                return None
 
-        template_in_db.images.append(image)
-        await self._db_session.commit()
-        await self._db_session.refresh(template_in_db)
-        return template_in_db
+            template_in_db.images.append(image)
+            await self._db_session.commit()
+            await self._db_session.refresh(template_in_db)
+            return template_in_db
 
     async def remove_image_from_template(self, template_id: UUID, image: models.Image) -> Union[None, models.Template]:
         """

gns3server/db/tasks.py

@@ -15,8 +15,6 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
-import asyncio
-import time
 import os
 
 from fastapi import FastAPI
@@ -31,13 +29,10 @@ from alembic import command, config
 from alembic.script import ScriptDirectory
 from alembic.runtime.migration import MigrationContext
 from alembic.util.exc import CommandError
-from watchdog.observers import Observer
-from watchdog.events import FileSystemEvent, PatternMatchingEventHandler
 
 from gns3server.db.repositories.computes import ComputesRepository
 from gns3server.db.repositories.images import ImagesRepository
-from gns3server.utils.images import md5sum, discover_images, read_image_info, InvalidImageError
-from gns3server.utils.asyncio import wait_run_in_executor
+from gns3server.utils.images import read_image_info
 from gns3server import schemas
 
 from .models import Base
@@ -142,7 +137,18 @@ async def connect_to_db(app: FastAPI) -> None:
                     log.info("Created new database and stamped to head revision")
                 elif db_state == "new_with_llm_configs":
                     # Database already has llm_model_configs table (from Base.metadata.create_all)
-                    # Just stamp the version
+                    # Older unversioned metadata databases lack inventory fields.
+                    # Ensure the additive schema before stamping the current head.
+                    def upgrade_inventory(connection):
+                        from alembic.operations import Operations
+                        from gns3server.db_migrations.versions.d9e8a2b7c401_image_inventory_reconciliation import (
+                            upgrade,
+                        )
+
+                        with Operations.context(MigrationContext.configure(connection)):
+                            upgrade()
+
+                    await conn.run_sync(upgrade_inventory)
                     await conn.run_sync(run_stamp, alembic_cfg)
                     await conn.commit()
                     log.info("Database has llm_model_configs table, stamped to head revision")
@@ -186,145 +192,25 @@ async def get_computes(app: FastAPI) -> List[schemas.Compute]:
     return computes
 
 
-async def discover_images_on_filesystem(app: FastAPI) -> None:
-
-    async with AsyncSession(app.state._db_engine) as db_session:
-        images_repository = ImagesRepository(db_session)
-        db_images = await images_repository.get_images()
-        existing_image_paths = []
-        for db_image in db_images:
-            try:
-                image = schemas.Image.model_validate(db_image)
-                existing_image_paths.append(image.path)
-            except ValidationError as e:
-                log.error(f"Could not load image '{db_image.filename}' from database: {e}")
-                continue
-        for image_type in ("qemu", "ios", "iou"):
-            discovered_images = await discover_images(image_type, existing_image_paths)
-            for image_info in discovered_images:
-                log.info(f"Adding discovered image '{image_info['path']}' to the database")
-                try:
-                    await images_repository.add_image(**image_info)
-                except SQLAlchemyError as e:
-                    log.warning(f"Error while adding image '{image_info['path']}' to the database: {e}")
-
-    # monitor if images have been manually added
-    asyncio.create_task(monitor_images_on_filesystem(app))
-
-
 async def update_disk_checksums(updated_disks: List[str]) -> None:
-    """
-    Update the checksum of a list of disks in the database.
-
-    :param updated_disks: list of updated disks
-    """
-
+    """Refresh complete metadata after a server-managed disk modification."""
     from gns3server.api.server import app
+    from gns3server.utils.image_inventory import image_lock
 
-    async with AsyncSession(app.state._db_engine) as db_session:
-        images_repository = ImagesRepository(db_session)
-        for path in updated_disks:
-            image = await images_repository.get_image(path)
-            if image:
-                log.info(f"Updating image '{path}' in the database")
-                checksum = await wait_run_in_executor(md5sum, path, cache_to_md5file=False)
-                if image.checksum != checksum:
-                    await images_repository.update_image(path, checksum, "md5")
-
-
-class EventHandler(PatternMatchingEventHandler):
-    """
-    Watchdog event handler.
-    """
-
-    def __init__(self, queue: asyncio.Queue, loop: asyncio.AbstractEventLoop, **kwargs):
-
-        self._loop = loop
-        self._queue = queue
-
-        # ignore temporary files, md5sum files, hidden files and directories
-        super().__init__(ignore_patterns=["*.tmp", "*.md5sum", ".*"], ignore_directories=True, **kwargs)
-
-    def on_closed(self, event: FileSystemEvent) -> None:
-        # monitor for closed files (e.g. when a file has finished to be copied)
-        if "/lib/" in event.src_path or "/lib64/" in event.src_path:
-            return  # ignore custom IOU libraries
-        self._loop.call_soon_threadsafe(self._queue.put_nowait, event)
-
-
-class EventIterator:
-    """
-    Watchdog Event iterator.
-    """
-
-    def __init__(self, queue: asyncio.Queue):
-        self.queue = queue
-
-    def __aiter__(self):
-        return self
-
-    async def __anext__(self):
-
-        item = await self.queue.get()
-        if item is None:
-            raise StopAsyncIteration
-        return item
-
-
-async def monitor_images_on_filesystem(app: FastAPI):
-
-    def watchdog(
-        path: str, queue: asyncio.Queue, loop: asyncio.AbstractEventLoop, app: FastAPI, recursive: bool = False
-    ) -> None:
-        """
-        Thread to monitor a directory for new images.
-        """
-
-        handler = EventHandler(queue, loop)
-        observer = Observer()
-        observer.schedule(handler, str(path), recursive=recursive)
-        observer.start()
-        log.info(f"Monitoring for new images in '{path}'")
-        while True:
-            time.sleep(1)
-            # stop when the app is exiting
-            if app.state.exiting:
-                observer.stop()
-                observer.join(10)
-                log.info(f"Stopping monitoring for new images in '{path}'")
-                loop.call_soon_threadsafe(queue.put_nowait, None)
-                break
-
-    queue: asyncio.Queue = asyncio.Queue()
-    loop = asyncio.get_event_loop()
-    server_config = Config.instance().settings.Server
-    image_dir = os.path.expanduser(server_config.images_path)
-    asyncio.get_event_loop().run_in_executor(None, watchdog, image_dir, queue, loop, app, True)
-
-    async for filesystem_event in EventIterator(queue):
-        # read the file system event from the queue
-        image_path = filesystem_event.src_path
-        expected_image_type = None
-        if "IOU" in image_path:
-            expected_image_type = "iou"
-        elif "QEMU" in image_path:
-            expected_image_type = "qemu"
-        elif "IOS" in image_path:
-            expected_image_type = "ios"
-        async with AsyncSession(app.state._db_engine) as db_session:
-            images_repository = ImagesRepository(db_session)
-            try:
-                image = await read_image_info(image_path, expected_image_type)
-            except InvalidImageError as e:
-                log.warning(str(e))
-                continue
-            try:
-                if await images_repository.get_image(image_path):
-                    continue
-                await images_repository.add_image(**image)
-                log.info(f"Discovered image '{image_path}' has been added to the database")
-            except SQLAlchemyError as e:
-                log.warning(f"Error while adding image '{image_path}' to the database: {e}")
+    for path in updated_disks:
+        async with image_lock(path):
+            async with AsyncSession(app.state._db_engine, expire_on_commit=False) as db_session:
+                repository = ImagesRepository(db_session)
+                image = await repository.get_image(path)
+                if image:
+                    info = await read_image_info(path, str(image.image_type), allow_raw_image=True)
+                    try:
+                        os.unlink(path + ".md5sum")
+                    except FileNotFoundError:
+                        pass
+                    except OSError as e:
+                        log.warning("Could not invalidate checksum cache for '%s': %s", path, e)
+                    await repository.save_verified_image(info)
 
 
 async def get_user_llm_config_full(user_id: str, app: FastAPI) -> Optional[dict]:

gns3server/db_migrations/versions/d9e8a2b7c401_image_inventory_reconciliation.py

@@ -0,0 +1,51 @@
+"""Track image availability and persistent synchronization jobs.
+
+Revision ID: d9e8a2b7c401
+Revises: c7e4a9f1d2b6
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+revision = "d9e8a2b7c401"
+down_revision = "c7e4a9f1d2b6"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # Startup also uses this for unversioned databases previously created from
+    # metadata, which may already contain some/all of these additive fields.
+    inspector = sa.inspect(op.get_bind())
+    existing = {column["name"] for column in inspector.get_columns("images")}
+    for column in (
+        sa.Column("availability", sa.String(), nullable=False, server_default="unknown"),
+        sa.Column("file_fingerprint", sa.String()),
+        sa.Column("last_seen_at", sa.DateTime()),
+        sa.Column("last_verified_at", sa.DateTime()),
+        sa.Column("last_error", sa.String()),
+    ):
+        if column.name not in existing:
+            op.add_column("images", column)
+    if inspector.has_table("image_sync_jobs"):
+        return
+    op.create_table(
+        "image_sync_jobs",
+        sa.Column("job_id", sa.String(), primary_key=True),
+        sa.Column("status", sa.String(), nullable=False),
+        sa.Column("dry_run", sa.Boolean(), nullable=False),
+        sa.Column("force_checksum", sa.Boolean(), nullable=False),
+        sa.Column("finished_at", sa.DateTime()),
+        sa.Column("counts", sa.JSON(), nullable=False),
+        sa.Column("errors", sa.JSON(), nullable=False),
+        sa.Column("created_at", sa.DateTime(), server_default=sa.func.current_timestamp()),
+        sa.Column("updated_at", sa.DateTime(), server_default=sa.func.current_timestamp()),
+    )
+
+
+def downgrade():
+    op.drop_table("image_sync_jobs")
+    # Do not rebuild images: dropping the old table during a SQLite batch
+    # rebuild would cascade-delete image_template_map entries with FK enabled.
+    for name in ("last_error", "last_verified_at", "last_seen_at", "file_fingerprint", "availability"):
+        op.drop_column("images", name)

gns3server/schemas/__init__.py

@@ -41,7 +41,7 @@ from .controller.computes import (
     Compute,
 )
 from .controller.templates import TemplateCreate, TemplateUpdate, TemplateUsage, Template
-from .controller.images import Image, ImageType
+from .controller.images import Image, ImageType, ImageSyncRequest, ImageSyncJob
 from .controller.appliances import ApplianceVersion, ApplianceVersionV8, Appliance
 from .controller.drawings import Drawing
 from .controller.gns3vm import GNS3VM

gns3server/schemas/config.py

@@ -184,6 +184,7 @@ class ServerSettings(BaseModel):
     )
     allow_raw_images: bool = Field(True, description="Allow raw images to be uploaded to the server")
     auto_discover_images: bool = Field(True, description="Automatically discover images in the images directory")
+    image_sync_interval: int = Field(900, ge=10, description="Seconds between automatic image inventory scans")
     report_errors: bool = Field(True, description="Automatically send crash reports to the GNS3 team")
     additional_images_paths: List[str] = Field(
         default_factory=list,

gns3server/schemas/controller/images.py

@@ -16,6 +16,8 @@
 
 from pydantic import ConfigDict, BaseModel, Field
 from enum import Enum
+from datetime import datetime
+from typing import Optional, Literal
 
 from .base import DateTimeModelMixin
 
@@ -37,7 +39,28 @@ class ImageBase(BaseModel):
     image_size: int = Field(..., description="Image size in bytes")
     checksum: str = Field(..., description="Checksum value")
     checksum_algorithm: str = Field(..., description="Checksum algorithm")
+    availability: Literal["unknown", "available", "missing", "unavailable", "invalid"] = "unknown"
+    last_seen_at: Optional[datetime] = None
+    last_verified_at: Optional[datetime] = None
+    last_error: Optional[str] = None
 
 
 class Image(DateTimeModelMixin, ImageBase):
     model_config = ConfigDict(from_attributes=True)
+
+
+class ImageSyncRequest(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+    dry_run: bool = False
+    force_checksum: bool = False
+
+
+class ImageSyncJob(DateTimeModelMixin):
+    model_config = ConfigDict(from_attributes=True)
+    job_id: str
+    status: Literal["queued", "running", "completed", "partial", "failed", "cancelled", "interrupted"]
+    dry_run: bool
+    force_checksum: bool
+    finished_at: Optional[datetime] = None
+    counts: dict[str, int]
+    errors: list[dict[str, str]]

gns3server/schemas/controller/settings.py

@@ -120,6 +120,7 @@ class ServerSettingsUpdate(BaseModel):
     default_symbol_theme: Optional[BuiltinSymbolTheme] = None
     allow_raw_images: Optional[bool] = None
     auto_discover_images: Optional[bool] = None
+    image_sync_interval: Optional[int] = Field(None, ge=10)
     report_errors: Optional[bool] = None
     additional_images_paths: Optional[List[str]] = None
     console_start_port_range: Optional[int] = Field(None, gt=0, le=65535)

gns3server/services/image_reconciliation.py

@@ -0,0 +1,510 @@
+"""Recoverable controller image inventory, shared by manual and automatic sync."""
+
+import asyncio
+import logging
+import os
+import stat
+import threading
+import uuid
+from datetime import datetime, timezone
+
+from sqlalchemy import select, update, delete
+from sqlalchemy.ext.asyncio import AsyncSession
+from watchdog.events import FileSystemEventHandler
+from watchdog.observers import Observer
+
+from gns3server.config import Config
+from gns3server.db.models import Image, ImageSyncJob
+from gns3server.db.repositories.images import ImagesRepository
+from gns3server.utils.image_inventory import (
+    ImageLock,
+    ImageLockBusy,
+    contained_path,
+    fingerprint,
+    image_lock,
+    normalized_path,
+)
+from gns3server.utils.images import inspect_image_file, InvalidImageError, ImageChangedError
+
+log = logging.getLogger(__name__)
+COUNTERS = (
+    "scanned",
+    "added",
+    "updated",
+    "unchanged",
+    "missing",
+    "unavailable",
+    "invalid",
+    "deferred",
+    "out_of_scope",
+    "errors",
+    "bytes_hashed",
+)
+RAW_EXTENSIONS = {".raw", ".img", ".iso", ".fd", ".vhd", ".vdi", ".bin"}
+
+
+def utcnow():
+    return datetime.now(timezone.utc).replace(tzinfo=None)
+
+
+def ignored(name):
+    return name.startswith(".") or name.endswith((".tmp", ".md5sum"))
+
+
+def configured_roots():
+    settings = Config.instance().settings.Server
+    roots = sorted({normalized_path(p) for p in [settings.images_path, *settings.additional_images_paths]})
+    # Keep parent roots only, avoiding duplicate scans and conflicting absence passes.
+    return [root for root in roots if not any(root != other and contained_path(root, other) for other in roots)]
+
+
+def root_identity(root):
+    info = os.stat(root)
+    # Opening the directory tests readability, not just existence.
+    with os.scandir(root):
+        pass
+    return info.st_dev, info.st_ino, os.path.realpath(root)
+
+
+def enumerate_root(root):
+    """Observe a root without creating directories or silently dropping failures."""
+    files, errors = {}, []
+    try:
+        root_stat = os.stat(root)
+        if not stat.S_ISDIR(root_stat.st_mode):
+            raise NotADirectoryError(root)
+        for directory, dirs, names in os.walk(
+            root,
+            onerror=lambda e: errors.append({"path": str(e.filename or root), "reason": str(e)}),
+            followlinks=False,
+        ):
+            dirs[:] = [
+                name
+                for name in dirs
+                if not ignored(name)
+                and name not in ("lib", "lib64")
+                and not os.path.islink(os.path.join(directory, name))
+            ]
+            for name in names:
+                if ignored(name):
+                    continue
+                path = normalized_path(os.path.join(directory, name))
+                try:
+                    files[path] = fingerprint(path)
+                except OSError as e:
+                    errors.append({"path": path, "reason": str(e)})
+        # A mount/root replacement during traversal makes the absence pass unsafe.
+        after = os.stat(root)
+        if (root_stat.st_dev, root_stat.st_ino) != (after.st_dev, after.st_ino):
+            errors.append({"path": root, "reason": "Image root changed during scan"})
+    except OSError as e:
+        errors.append({"path": root, "reason": str(e)})
+    return files, errors
+
+
+class InventoryEvents(FileSystemEventHandler):
+    def __init__(self, loop, dirty):
+        self.loop, self.dirty = loop, dirty
+        self._pending = False
+        self._lock = threading.Lock()
+
+    def _mark_dirty(self):
+        with self._lock:
+            self._pending = False
+        self.dirty.set()
+
+    def on_any_event(self, event):
+        if event.event_type not in ("created", "modified", "closed", "deleted", "moved"):
+            return
+        paths = [event.src_path, getattr(event, "dest_path", "")]
+        if any(path and not ignored(os.path.basename(path)) for path in paths):
+            # A single dirty bit coalesces events with bounded memory, including
+            # directory renames and bursts. The next scan observes current state.
+            with self._lock:
+                if self._pending:
+                    return
+                self._pending = True
+            self.loop.call_soon_threadsafe(self._mark_dirty)
+
+
+class ImageReconciliationService:
+    def __init__(self, engine, settle_seconds=1.0):
+        self.engine = engine
+        self.settle_seconds = settle_seconds
+        self.task = None
+        self.scheduler = None
+        self.observer = None
+        self.dirty = asyncio.Event()
+        self.stopping = threading.Event()
+        self._watched_roots = None
+        self._catalog_paths = {}
+        self._ambiguous_paths = set()
+
+    async def start(self, dry_run=False, force_checksum=False):
+        if self.stopping.is_set():
+            raise ImageLockBusy("Image synchronization is shutting down")
+        lock = ImageLock("image-inventory", wait=False)
+        await lock.__aenter__()
+        try:
+            async with AsyncSession(self.engine, expire_on_commit=False) as session:
+                # The OS lock proves no previous worker using this catalog is
+                # active. Recover jobs abandoned by a crash before admitting one.
+                await session.execute(
+                    update(ImageSyncJob)
+                    .where(ImageSyncJob.status.in_(["queued", "running"]))
+                    .values(status="interrupted", finished_at=utcnow())
+                )
+                job = ImageSyncJob(
+                    job_id=str(uuid.uuid4()),
+                    status="queued",
+                    dry_run=dry_run,
+                    force_checksum=force_checksum,
+                    counts=dict.fromkeys(COUNTERS, 0),
+                    errors=[],
+                )
+                session.add(job)
+                # Bound history; active jobs are never removed.
+                old_jobs = (
+                    select(ImageSyncJob.job_id)
+                    .where(ImageSyncJob.status.notin_(["queued", "running"]))
+                    .order_by(ImageSyncJob.created_at.desc())
+                    .offset(99)
+                )
+                await session.execute(delete(ImageSyncJob).where(ImageSyncJob.job_id.in_(old_jobs)))
+                await session.commit()
+                await session.refresh(job)
+                result = job.asdict()
+            self.task = asyncio.create_task(self._run(result, lock), name="image-inventory-sync")
+            self.task.add_done_callback(self._finished)
+            return result
+        except BaseException:
+            await lock.__aexit__()
+            raise
+
+    @staticmethod
+    def _finished(task):
+        if not task.cancelled() and task.exception() is not None:
+            log.error("Could not persist image synchronization outcome: %s", task.exception())
+
+    async def get_job(self, job_id, offset=0, limit=100):
+        # A client can resume polling after a restart even with automatic scans
+        # disabled. Do not leave an abandoned job looking permanently active.
+        try:
+            async with ImageLock("image-inventory", wait=False):
+                async with AsyncSession(self.engine) as session:
+                    await session.execute(
+                        update(ImageSyncJob)
+                        .where(ImageSyncJob.status.in_(["queued", "running"]))
+                        .values(status="interrupted", finished_at=utcnow())
+                    )
+                    await session.commit()
+        except ImageLockBusy:
+            pass
+        async with AsyncSession(self.engine) as session:
+            job = await session.get(ImageSyncJob, job_id)
+            if job is None:
+                return None
+            result = job.asdict()
+            result["errors"] = result["errors"][offset : offset + limit]
+            return result
+
+    async def _persist(self, job):
+        async with AsyncSession(self.engine) as session:
+            await session.execute(
+                update(ImageSyncJob)
+                .where(ImageSyncJob.job_id == job["job_id"])
+                .values(
+                    status=job["status"],
+                    counts=dict(job["counts"]),
+                    errors=list(job["errors"]),
+                    finished_at=job.get("finished_at"),
+                )
+            )
+            await session.commit()
+
+    def _error(self, job, path, reason):
+        job["counts"]["errors"] += 1
+        # The total count remains accurate even when detailed errors are capped.
+        if len(job["errors"]) < 1000:
+            job["errors"].append({"path": path, "reason": str(reason)})
+
+    async def _inspect(self, path, expected, allow_raw):
+        worker = asyncio.create_task(asyncio.to_thread(inspect_image_file, path, expected, allow_raw, self.stopping))
+        try:
+            return await asyncio.shield(worker)
+        except asyncio.CancelledError:
+            self.stopping.set()
+            try:
+                await worker
+            except (OSError, InvalidImageError):
+                pass
+            raise
+
+    async def _state(self, row, availability, reason, dry_run):
+        if row is None or dry_run:
+            return
+        async with AsyncSession(self.engine) as session:
+            # Revision guard for legacy writers not yet taking the path lock.
+            await session.execute(
+                update(Image)
+                .where(
+                    Image.image_id == row["image_id"],
+                    Image.path == row["path"],
+                    Image.file_fingerprint == row["file_fingerprint"],
+                    Image.checksum == row["checksum"],
+                )
+                .values(availability=availability, last_error=str(reason))
+            )
+            await session.commit()
+
+    async def _row(self, path):
+        async with AsyncSession(self.engine) as session:
+            stored = self._catalog_paths.get(normalized_path(path), path)
+            row = (await session.execute(select(Image).where(Image.path.in_([stored, path])))).scalar_one_or_none()
+            return row.asdict() if row else None
+
+    async def _file(self, path, observed, job, root):  # noqa: C901 - keep per-file outcomes under one path lock
+        if path in self._ambiguous_paths:
+            job["counts"]["deferred"] += 1
+            return
+        async with image_lock(path):
+            row = await self._row(path)
+            counts = job["counts"]
+            counts["scanned"] += 1
+            try:
+                if not contained_path(os.path.realpath(path), root):
+                    raise OSError("Image path resolves outside its configured root")
+                current = await asyncio.to_thread(fingerprint, path)
+                if current != observed:
+                    raise ImageChangedError("File changed during settling interval")
+                if (
+                    row
+                    and row["availability"] == "available"
+                    and row["file_fingerprint"] == current
+                    and not job["force_checksum"]
+                ):
+                    counts["unchanged"] += 1
+                    return
+                expected = row["image_type"] if row else None
+                main_root = normalized_path(Config.instance().settings.Server.images_path)
+                if not expected and contained_path(path, main_root):
+                    component = os.path.relpath(path, main_root).split(os.sep)[0]
+                    expected = {"QEMU": "qemu", "IOS": "ios", "IOU": "iou"}.get(component)
+                allow_raw = Config.instance().settings.Server.allow_raw_images and (
+                    expected == "qemu"
+                    or (expected in (None, "qemu") and os.path.splitext(path)[1].lower() in RAW_EXTENSIONS)
+                )
+                info = await self._inspect(path, expected, allow_raw)
+                if row:
+                    # Preserve legacy path spelling as well as image identity.
+                    info["path"] = row["path"]
+                counts["bytes_hashed"] += info["image_size"]
+                if self.stopping.is_set():
+                    raise asyncio.CancelledError
+                if await asyncio.to_thread(fingerprint, path) != info["file_fingerprint"]:
+                    raise ImageChangedError("File changed before catalog update")
+                if not job["dry_run"]:
+                    # Do not leave stale sidecars for legacy compute consumers.
+                    try:
+                        os.unlink(path + ".md5sum")
+                    except FileNotFoundError:
+                        pass
+                    except OSError as e:
+                        self._error(job, path + ".md5sum", e)
+                    async with AsyncSession(self.engine, expire_on_commit=False) as session:
+                        await ImagesRepository(session).save_verified_image(info)
+                counts["updated" if row else "added"] += 1
+            except ImageChangedError as e:
+                counts["deferred"] += 1
+                await self._state(row, "unavailable", e, job["dry_run"])
+            except InvalidImageError as e:
+                counts["invalid"] += 1
+                await self._state(row, "invalid", e, job["dry_run"])
+                self._error(job, path, e)
+            except OSError as e:
+                if self.stopping.is_set():
+                    raise asyncio.CancelledError from e
+                counts["unavailable"] += 1
+                await self._state(row, "unavailable", e, job["dry_run"])
+                self._error(job, path, e)
+
+    async def _run(self, job, lock):  # noqa: C901 - one job owns root health, progress and lock lifetime
+        job["status"] = "running"
+        try:
+            await self._persist(job)
+            async with AsyncSession(self.engine) as session:
+                paths = (await session.execute(select(Image.path))).scalars().all()
+            self._catalog_paths = {}
+            self._ambiguous_paths = set()
+            roots = configured_roots()
+            for path in paths:
+                normalized = normalized_path(path)
+                if not any(contained_path(path, root) for root in roots):
+                    job["counts"]["out_of_scope"] += 1
+                    self._error(job, path, "Catalog path is outside the configured image directories; record retained")
+                if normalized in self._catalog_paths:
+                    self._ambiguous_paths.add(normalized)
+                    self._error(job, path, "Multiple catalog paths resolve to the same image; manual review required")
+                self._catalog_paths[normalized] = path
+            for root in roots:
+                if self.stopping.is_set():
+                    raise asyncio.CancelledError
+                resolved_root = os.path.realpath(root)
+                try:
+                    identity = await asyncio.to_thread(root_identity, root)
+                except OSError:
+                    identity = None
+                files, errors = await asyncio.to_thread(enumerate_root, root)
+                if resolved_root != os.path.realpath(root):
+                    errors.append({"path": root, "reason": "Image root target changed during scan"})
+                for error in errors:
+                    self._error(job, error["path"], error["reason"])
+                # One settling wait per root, not one per large image collection.
+                if files and self.settle_seconds:
+                    await asyncio.sleep(self.settle_seconds)
+                for path, observed in files.items():
+                    if self.stopping.is_set():
+                        raise asyncio.CancelledError
+                    try:
+                        await self._file(path, observed, job, resolved_root)
+                    except Exception as e:
+                        self._error(job, path, e)
+                        log.warning("Could not reconcile image %s: %s", path, e)
+                    if job["counts"]["scanned"] % 25 == 0:
+                        await self._persist(job)
+                try:
+                    if identity != await asyncio.to_thread(root_identity, root):
+                        raise OSError("Image root changed during reconciliation")
+                except OSError as e:
+                    errors.append({"path": root, "reason": str(e)})
+                    self._error(job, root, e)
+                async with AsyncSession(self.engine) as session:
+                    rows = [image.asdict() for image in (await session.execute(select(Image))).scalars()]
+                for row in rows:
+                    if self.stopping.is_set():
+                        raise asyncio.CancelledError
+                    path = row["path"]
+                    if (
+                        not contained_path(path, root)
+                        or normalized_path(path) in files
+                        or normalized_path(path) in self._ambiguous_paths
+                    ):
+                        continue
+                    async with image_lock(path):
+                        row = await self._row(path)
+                        if row is None:
+                            continue
+                        if errors:
+                            job["counts"]["unavailable"] += 1
+                            await self._state(
+                                row, "unavailable", "Image root was not completely scanned", job["dry_run"]
+                            )
+                            continue
+                        try:
+                            if not contained_path(os.path.realpath(path), resolved_root):
+                                raise OSError("Image path resolves outside its configured root")
+                            await asyncio.to_thread(fingerprint, path)
+                        except FileNotFoundError:
+                            try:
+                                if identity != await asyncio.to_thread(root_identity, root):
+                                    raise OSError("Image root changed during reconciliation")
+                            except OSError as e:
+                                job["counts"]["unavailable"] += 1
+                                await self._state(row, "unavailable", e, job["dry_run"])
+                                self._error(job, root, e)
+                            else:
+                                job["counts"]["missing"] += 1
+                                await self._state(row, "missing", "Image file is missing", job["dry_run"])
+                        except OSError as e:
+                            job["counts"]["unavailable"] += 1
+                            await self._state(row, "unavailable", e, job["dry_run"])
+                            self._error(job, path, e)
+            if self.stopping.is_set():
+                raise asyncio.CancelledError
+            job["status"] = "partial" if job["counts"]["errors"] or job["counts"]["deferred"] else "completed"
+        except asyncio.CancelledError:
+            job["status"] = "cancelled"
+        except Exception as e:
+            job["status"] = "failed"
+            self._error(job, "", e)
+            log.exception("Image reconciliation failed")
+        finally:
+            job["finished_at"] = utcnow()
+            try:
+                await self._persist(job)
+            finally:
+                await lock.__aexit__()
+
+    async def _watch(self):
+        roots = configured_roots() if Config.instance().settings.Server.auto_discover_images else []
+        # Include availability in the signature so newly mounted roots get watched.
+        roots = [root for root in roots if os.path.isdir(root)]
+        if roots == self._watched_roots:
+            return
+        if self.observer:
+            self.observer.stop()
+            await asyncio.to_thread(self.observer.join)
+            self.observer = None
+        self._watched_roots = roots
+        if roots:
+            observer = Observer()
+            handler = InventoryEvents(asyncio.get_running_loop(), self.dirty)
+            try:
+                for root in roots:
+                    observer.schedule(handler, root, recursive=True)
+                observer.start()
+                self.observer = observer
+            except OSError:
+                observer.stop()
+                if observer.is_alive():
+                    await asyncio.to_thread(observer.join)
+                self._watched_roots = None
+                log.warning("Image watcher unavailable; periodic scanning remains active", exc_info=True)
+
+    async def _automatic(self):
+        await asyncio.sleep(5)
+        while not self.stopping.is_set():
+            try:
+                await self._watch()
+                self.dirty.clear()
+                if Config.instance().settings.Server.auto_discover_images:
+                    try:
+                        await self.start()
+                        await asyncio.shield(self.task)
+                    except ImageLockBusy:
+                        pass
+                try:
+                    await asyncio.wait_for(self.dirty.wait(), Config.instance().settings.Server.image_sync_interval)
+                    await asyncio.sleep(1)  # debounce bursts of file/directory events
+                except asyncio.TimeoutError:
+                    pass
+            except asyncio.CancelledError:
+                break
+            except Exception:
+                log.exception("Automatic image synchronization failed; will retry")
+                await asyncio.sleep(10)
+
+    def start_background(self):
+        self.scheduler = asyncio.create_task(self._automatic(), name="image-inventory-scheduler")
+
+    async def close(self):
+        self.stopping.set()
+        if self.scheduler:
+            self.scheduler.cancel()
+            await asyncio.gather(self.scheduler, return_exceptions=True)
+        if self.observer:
+            self.observer.stop()
+            await asyncio.to_thread(self.observer.join)
+            self.observer = None
+        if self.task and not self.task.done():
+            # Cooperative cancellation also covers a task not yet entered; its
+            # finally block must run to release the already-acquired job lock.
+            await asyncio.gather(self.task, return_exceptions=True)
+
+
+def get_image_reconciliation_service(app):
+    service = getattr(app.state, "image_reconciliation", None)
+    if service is None:
+        service = ImageReconciliationService(app.state._db_engine)
+        app.state.image_reconciliation = service
+    return service

gns3server/services/templates.py

@@ -279,7 +279,8 @@ class TemplatesService:
 
         if not image_path:
             return
-        image = await self._templates_repo.get_image(image_path)
+        # Removing an association must also find missing/invalid image rows.
+        image = await self._templates_repo.get_image(image_path, include_unavailable=True)
         if image is None:
             return
         await self._templates_repo.remove_image_from_template(template_id, image)

gns3server/utils/image_inventory.py

@@ -0,0 +1,99 @@
+"""Filesystem primitives shared by image writers and inventory reconciliation."""
+
+import asyncio
+import hashlib
+import os
+import stat
+
+from gns3server.config import Config
+
+
+def normalized_path(path):
+    return os.path.normcase(os.path.abspath(os.path.expanduser(path)))
+
+
+def contained_path(path, root):
+    try:
+        return os.path.commonpath((normalized_path(path), normalized_path(root))) == normalized_path(root)
+    except ValueError:
+        return False
+
+
+def fingerprint(path):
+    info = os.stat(path, follow_symlinks=True)
+    if not stat.S_ISREG(info.st_mode):
+        raise OSError(f"Not a regular image file: {path}")
+    return stat_fingerprint(info)
+
+
+def stat_fingerprint(info):
+    # Store as text: inode/device numbers need not fit a signed SQL BIGINT.
+    return ":".join(
+        str(value) for value in (info.st_dev, info.st_ino, info.st_size, info.st_mtime_ns, info.st_ctime_ns)
+    )
+
+
+class ImageLockBusy(Exception):
+    pass
+
+
+class ImageLock:
+    """Advisory lock shared by controller processes using the same config directory.
+
+    Lock files are deliberately retained: unlinking them permits two processes to
+    lock different inodes for the same name. OS locks are released on process exit.
+    """
+
+    def __init__(self, key, wait=True):
+        self.key = key
+        self.wait = wait
+        self._file = None
+
+    async def __aenter__(self):
+        directory = os.path.join(Config.instance().config_dir, ".image-locks")
+        os.makedirs(directory, exist_ok=True)
+        path = os.path.join(directory, hashlib.sha256(self.key.encode()).hexdigest() + ".lock")
+        self._file = open(path, "a+b")
+        if os.name == "nt" and os.fstat(self._file.fileno()).st_size == 0:
+            self._file.write(b"\0")
+            self._file.flush()
+        try:
+            while True:
+                try:
+                    if os.name == "nt":
+                        import msvcrt
+
+                        self._file.seek(0)
+                        msvcrt.locking(self._file.fileno(), msvcrt.LK_NBLCK, 1)
+                    else:
+                        import fcntl
+
+                        fcntl.flock(self._file, fcntl.LOCK_EX | fcntl.LOCK_NB)
+                    return self
+                except (BlockingIOError, PermissionError):
+                    if not self.wait:
+                        raise ImageLockBusy(self.key)
+                    await asyncio.sleep(0.05)
+        except BaseException:
+            self._file.close()
+            self._file = None
+            raise
+
+    async def __aexit__(self, *args):
+        if self._file is not None:
+            self._file.close()
+            self._file = None
+
+
+def image_lock(path):
+    return ImageLock("image:" + normalized_path(os.path.realpath(path)))
+
+
+def publish_image(temporary, destination):
+    """Atomically publish a complete file without replacing an existing image.
+
+    Both paths must be on the same filesystem. Hard linking provides the atomic
+    no-overwrite operation that os.replace()/shutil.move() cannot provide.
+    """
+    os.link(temporary, destination)
+    os.unlink(temporary)

gns3server/utils/images.py

@@ -18,7 +18,10 @@ import os
 import hashlib
 import stat
 import aiofiles
-import shutil
+import asyncio
+import tempfile
+
+from gns3server.utils.image_inventory import fingerprint, stat_fingerprint, image_lock, publish_image, contained_path
 
 try:
     import importlib_resources
@@ -126,33 +129,46 @@ def get_builtin_disks() -> List[str]:
     return builtin_disks
 
 
-async def read_image_info(path: str, expected_image_type: str | None = None) -> dict:
+def inspect_image_file(path, expected_image_type=None, allow_raw_image=False, stopped_event=None):
+    """Read a stable regular file once, never trusting checksum sidecars."""
+    before = fingerprint(path)
+    flags = os.O_RDONLY | getattr(os, "O_BINARY", 0)
+    with os.fdopen(os.open(path, flags), "rb") as f:
+        info = os.fstat(f.fileno())
+        if not stat.S_ISREG(info.st_mode) or stat_fingerprint(info) != before:
+            raise ImageChangedError(f"Image changed while opening: {path}")
+        header = f.read(7)
+        if len(header) < 7:
+            raise InvalidImageError(f"Image '{path}' is too small to be valid")
+        image_type = check_valid_image_header(path, header, allow_raw_image)
+        if expected_image_type and image_type != expected_image_type:
+            raise InvalidImageError(f"Detected image type for '{path}' is {image_type}, expected {expected_image_type}")
+        digest = hashlib.md5(header)
+        while True:
+            if stopped_event is not None and stopped_event.is_set():
+                raise InterruptedError("Image inspection cancelled")
+            chunk = f.read(1024 * 1024)
+            if not chunk:
+                break
+            digest.update(chunk)
+        if stat_fingerprint(os.fstat(f.fileno())) != before or fingerprint(path) != before:
+            raise ImageChangedError(f"Image changed while reading: {path}")
+    return dict(
+        image_name=os.path.basename(path),
+        image_type=image_type,
+        image_size=info.st_size,
+        path=path,
+        checksum=digest.hexdigest(),
+        checksum_algorithm="md5",
+        file_fingerprint=before,
+    )
 
-    header_magic_len = 7
+
+async def read_image_info(path: str, expected_image_type: str | None = None, allow_raw_image=False) -> dict:
     try:
-        async with aiofiles.open(path, "rb") as f:
-            image_header = await f.read(header_magic_len)  # read the first 7 bytes of the file
-            if len(image_header) >= header_magic_len:
-                detected_image_type = check_valid_image_header(path, image_header)
-                if expected_image_type and detected_image_type != expected_image_type:
-                    raise InvalidImageError(
-                        f"Detected image type for '{path}' is {detected_image_type}, "
-                        f"expected type is {expected_image_type}"
-                    )
-            else:
-                raise InvalidImageError(f"Image '{path}' is too small to be valid")
+        return await asyncio.to_thread(inspect_image_file, path, expected_image_type, allow_raw_image)
     except OSError as e:
-        raise InvalidImageError(f"Cannot read image '{path}': {e}")
-
-    image_info = {
-        "image_name": os.path.basename(path),
-        "image_type": detected_image_type,
-        "image_size": os.stat(path).st_size,
-        "path": path,
-        "checksum": await wait_run_in_executor(md5sum, path, cache_to_md5file=False),
-        "checksum_algorithm": "md5",
-    }
-    return image_info
+        raise InvalidImageError(f"Cannot read image '{path}': {e}") from e
 
 
 async def discover_images(image_type: str, skip_image_paths: list | None = None) -> List[dict]:
@@ -244,7 +260,7 @@ def images_directories(image_type, include_parent_directory=True):
     return [force_unix_path(p) for p in paths if os.path.exists(p)]
 
 
-def md5sum(path, working_dir=None, stopped_event=None, cache_to_md5file=True):
+def md5sum(path, working_dir=None, stopped_event=None, cache_to_md5file=True, use_cache=True):
     """
     Return the md5sum of an image and cache it on disk
 
@@ -263,7 +279,7 @@ def md5sum(path, working_dir=None, stopped_event=None, cache_to_md5file=True):
     else:
         md5sum_file = path + ".md5sum"
 
-    if os.path.exists(md5sum_file):
+    if use_cache and os.path.exists(md5sum_file):
         try:
             with open(md5sum_file) as f:
                 md5 = f.read().strip()
@@ -319,6 +335,10 @@ class InvalidImageError(Exception):
         return self._message
 
 
+class ImageChangedError(InvalidImageError):
+    """An observation must be retried because the file is still changing."""
+
+
 def check_valid_image_header(path: str, data: bytes, allow_raw_image: bool = False) -> str:
 
     if data[:7] == b"\x7fELF\x01\x02\x01":
@@ -347,54 +367,66 @@ async def write_image(
 ) -> models.Image:
 
     image_dir, image_name = os.path.split(image_filename)
-    # Store the file under its final name only when the upload is completed
-    tmp_path = image_path + ".tmp"
-    log.info(f"Writing image file to '{tmp_path}'")
+    # HTTP chunk boundaries need not align with the seven-byte image header.
+    iterator = stream.__aiter__()
+    prefix = bytearray()
+    async for chunk in iterator:
+        prefix.extend(chunk)
+        if len(prefix) >= 7:
+            break
+    if len(prefix) < 7:
+        raise InvalidImageError("The image content is empty or too small to be valid")
+    image_type = check_valid_image_header(image_path, bytes(prefix), allow_raw_image or not check_image_header)
+    if not image_dir:
+        image_path = os.path.abspath(os.path.join(default_images_directory(image_type), image_name))
+        root = os.path.realpath(os.path.expanduser(Config.instance().settings.Server.images_path))
+        if not contained_path(os.path.realpath(image_path), root):
+            raise InvalidImageError(f"Image destination is outside the configured image directory: {image_path}")
     os.makedirs(os.path.dirname(image_path), exist_ok=True)
+    descriptor, tmp_path = tempfile.mkstemp(prefix=".gns3-upload-", suffix=".tmp", dir=os.path.dirname(image_path))
+    os.close(descriptor)
     checksum = hashlib.md5()
-    header_magic_len = 7
-    image_type = None
-    image_size = 0
     try:
         async with aiofiles.open(tmp_path, "wb") as f:
-            async for chunk in stream:
-                if check_image_header and len(chunk) >= header_magic_len:
-                    check_image_header = False
-                    image_type = check_valid_image_header(image_path, chunk, allow_raw_image)
+            await f.write(prefix)
+            checksum.update(prefix)
+            async for chunk in iterator:
                 await f.write(chunk)
                 checksum.update(chunk)
-
         image_size = os.path.getsize(tmp_path)
-        if not image_size or image_size < header_magic_len:
-            raise InvalidImageError("The image content is empty or too small to be valid")
-
-        if not image_dir:
-            directory = default_images_directory(image_type)
-            os.makedirs(directory, exist_ok=True)
-            image_path = os.path.abspath(os.path.join(directory, image_filename))
-
-        if os.path.exists(image_path):
-            raise InvalidImageError(
-                f"File '{image_path}' already exists, please choose a different name or remove the existing image"
+        async with image_lock(image_path):
+            if os.path.lexists(image_path):
+                raise InvalidImageError(
+                    f"File '{image_path}' already exists, please choose a different name or remove the existing image"
+                )
+            checksum_str: str = checksum.hexdigest()
+            duplicate_image = await images_repo.get_image_by_checksum(checksum_str, os.path.dirname(image_path))
+            if duplicate_image:
+                raise InvalidImageError(
+                    f"Image '{duplicate_image.filename}' with the same checksum "
+                    f"already exists in '{os.path.dirname(image_path)}'"
+                )
+            os.chmod(tmp_path, stat.S_IWRITE | stat.S_IREAD | stat.S_IEXEC)
+            publish_image(tmp_path, image_path)
+            # Complete files survive a database failure so the next scan can
+            # recover them. Never compensate by unlinking a published image.
+            image = await images_repo.save_verified_image(
+                dict(
+                    image_name=image_name,
+                    image_type=image_type,
+                    image_size=image_size,
+                    path=image_path,
+                    checksum=checksum_str,
+                    checksum_algorithm="md5",
+                    file_fingerprint=fingerprint(image_path),
+                )
             )
-
-        checksum_hex = checksum.hexdigest()
-        image_dir = os.path.dirname(image_path)
-        duplicate_image = await images_repo.get_image_by_checksum(checksum_hex, image_dir)
-        if duplicate_image:
-            raise InvalidImageError(
-                f"Image '{duplicate_image.filename}' with the same checksum already exists in '{image_dir}'"
-            )
-
-        shutil.move(tmp_path, image_path)
-        os.chmod(image_path, stat.S_IWRITE | stat.S_IREAD | stat.S_IEXEC)
+            if image is None:
+                raise InvalidImageError(f"Failed to save image '{image_name}' to database")
+            return image
     finally:
         try:
             if os.path.exists(tmp_path):
                 os.remove(tmp_path)
         except OSError:
             log.warning(f"Could not remove '{tmp_path}'")
-
-    return await images_repo.add_image(
-        image_name, image_type, image_size, image_path, checksum_hex, checksum_algorithm="md5"
-    )