fix(queue): stop a library-file delete from destroying the jobs queued against it (#2819)

Nothing tied a library file to the queue rows pointing at it, and the FK
that describes the relationship is ON DELETE CASCADE -- which SQLite does
not enforce and PostgreSQL does. So the same fault had two faces: rows
left pointing at a file that no longer existed, failing at the printer
with "Library file not found" days later, or rows deleted outright with
no error and no history.

Two routes into it, both fixed by taking the queue off the file before
the row goes.

Dispatch (the reported case): quantity>1 on the printer-card
upload-and-print flow puts cleanup_library_after_dispatch on every copy,
and _clone_queue_item copies library_file_id onto batch clones, so the
first dispatch consumed the file the rest were waiting on. The copies are
now pointed at the archive that dispatch just created -- it holds its own
copy of the 3MF -- and the consume flag is cleared on them. A copy already
printing from its own archive keeps it, a finished one keeps its outcome,
and a cross-model item (#671) keeps any candidate this does not consume.

Deletion: the File Manager, bulk delete, folder delete, emptying the trash
and the retention sweeper all removed rows with queued work against them.
Folder delete did not even clear the cross-model candidates, because the
file-id walk it already performs threw its result away. Jobs waiting on a
deleted file are now cancelled at that moment, naming the file, and every
other row referring to it is detached rather than destroyed -- print
history and batch progress are counted from those rows. A job that is
printing is left alone: what is deleted is the library copy, not the copy
on the machine. The trash is reversible so it still changes nothing about
the queue, and a job dispatched while its file is in the trash now says so
instead of "not found".

Verified row for row on PostgreSQL 16 as well as SQLite: without this,
PostgreSQL deletes every queue row referencing the file.

backend/app/api/routes/library.py

@@ -371,6 +371,36 @@ def _resolve_slice_destination(target_folder: LibraryFolder | None, out_filename
     return dest, True, None
 
 
+async def _folder_tree_file_ids(db: AsyncSession, folder_id: int) -> list[int]:
+    """Every ``LibraryFile`` id under ``folder_id``, at any depth.
+
+    Deleting a folder cascades to its whole subtree, so anything that has to be
+    released before that delete (queue items, cross-model candidates) needs the
+    subtree, not just the folder's own files.
+
+    Trashed rows are included deliberately: they are still real rows and the
+    cascade takes them too.
+    """
+    file_ids: list[int] = []
+    pending = [folder_id]
+    # The API refuses to make a folder its own ancestor, so a loop here would
+    # mean the table is already corrupt -- but this walk runs inside a delete
+    # request, and hanging one is worse than the cost of a set.
+    seen: set[int] = set()
+    while pending:
+        current = pending.pop()
+        if current in seen:
+            continue
+        seen.add(current)
+        file_ids.extend(
+            (await db.execute(select(LibraryFile.id).where(LibraryFile.folder_id == current))).scalars().all()
+        )
+        pending.extend(
+            (await db.execute(select(LibraryFolder.id).where(LibraryFolder.parent_id == current))).scalars().all()
+        )
+    return file_ids
+
+
 def _stored_file_path(abs_path: Path, is_external: bool) -> str:
     """Produce the value to persist in ``LibraryFile.file_path``.
 
@@ -1422,7 +1452,16 @@ async def delete_folder(
 
         return file_ids
 
-    await get_all_file_ids(folder_id)
+    doomed_file_ids = await get_all_file_ids(folder_id)
+
+    # The folder cascade hard-deletes every file row under it, so the queue has
+    # to be taken off them first — same as the single-file delete below (#2819).
+    # The return value used to be discarded here, which is why this never
+    # happened for a folder delete.
+    from backend.app.services.library_trash import delete_dependent_variants, release_queue_references
+
+    await delete_dependent_variants(db, doomed_file_ids)
+    await release_queue_references(db, doomed_file_ids)
 
     # Delete folder (cascade will handle files and subfolders)
     await db.delete(folder)
@@ -4885,9 +4924,10 @@ async def delete_file(
                 abs_thumb_path.unlink()
             except OSError as e:
                 logger.warning("Failed to delete thumbnail from disk: %s", e)
-        from backend.app.services.library_trash import delete_dependent_variants
+        from backend.app.services.library_trash import delete_dependent_variants, release_queue_references
 
         await delete_dependent_variants(db, [file.id])
+        await release_queue_references(db, [file.id])
         await db.delete(file)
         await db.commit()
         return {"status": "success", "message": "File deleted", "trashed": False}
@@ -5205,10 +5245,15 @@ async def bulk_delete(
 
     Files not owned by the user are skipped (unless user has *_all permission).
     """
+    from backend.app.services.library_trash import delete_dependent_variants, release_queue_references
+
     user, can_modify_all = auth_result
     deleted_files = 0
     deleted_folders = 0
     skipped_files = 0
+    # External files bypass the trash and are removed for good, so the queue has
+    # to come off them. Collected here and dealt with once, below the loop.
+    hard_deleted: list[LibraryFile] = []
 
     # Delete files first. Managed files go to trash (sweeper hard-deletes bytes
     # later); external files bypass trash since their disk state is outside our
@@ -5230,11 +5275,22 @@ async def bulk_delete(
                     abs_thumb_path.unlink()
                 except OSError as e:
                     logger.warning("Failed to delete thumbnail from disk: %s", e)
-            await db.delete(file)
+            hard_deleted.append(file)
         else:
             file.deleted_at = now
         deleted_files += 1
 
+    # After the loop and before any delete is issued (#2819). Order matters
+    # twice over: a query run while a delete is pending autoflushes it, taking
+    # the cascade with it, and releasing once for the whole set is a couple of
+    # statements rather than a couple per file.
+    if hard_deleted:
+        hard_deleted_ids = [f.id for f in hard_deleted]
+        await delete_dependent_variants(db, hard_deleted_ids)
+        await release_queue_references(db, hard_deleted_ids)
+        for file in hard_deleted:
+            await db.delete(file)
+
     # Delete folders (cascade will handle contents). Folders have no ownership
     # tracking, so users without *_all permission may only delete empty,
     # non-external, non-linked folders (#1781) — same rule as DELETE /folders/{id}.
@@ -5252,6 +5308,9 @@ async def bulk_delete(
                 )
             )
             deleted_files += file_count_result.scalar() or 0
+            tree_file_ids = await _folder_tree_file_ids(db, folder_id)
+            await delete_dependent_variants(db, tree_file_ids)
+            await release_queue_references(db, tree_file_ids)
             await db.delete(folder)
             deleted_folders += 1
 

backend/app/services/library_trash.py

@@ -27,8 +27,9 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from backend.app.core.config import settings as app_settings
 from backend.app.core.database import async_session
 from backend.app.models.library import LibraryFile
-from backend.app.models.print_queue import PrintQueueVariant
+from backend.app.models.print_queue import PrintQueueItem, PrintQueueVariant
 from backend.app.models.settings import Settings
+from backend.app.utils.local_time import utcnow_naive
 
 logger = logging.getLogger(__name__)
 
@@ -353,6 +354,7 @@ class LibraryTrashService:
             self._unlink_on_disk(row)
             deleted += 1
         await delete_dependent_variants(db, [r.id for r in rows])
+        await release_queue_references(db, [r.id for r in rows])
         # Single DELETE is faster than N await db.delete() round-trips; we
         # still need the Python loop above to unlink bytes on disk.
         await db.execute(delete(LibraryFile).where(LibraryFile.id.in_([r.id for r in rows])))
@@ -386,10 +388,86 @@ class LibraryTrashService:
         """Bypass retention and delete this trashed file + its bytes immediately."""
         self._unlink_on_disk(file)
         await delete_dependent_variants(db, [file.id])
+        await release_queue_references(db, [file.id])
         await db.delete(file)
         await db.commit()
 
 
+async def release_queue_references(db: AsyncSession, file_ids: list[int]) -> int:
+    """Take queued work off files that are about to be hard-deleted (#2819).
+
+    Call this before any statement that removes ``library_files`` rows — the
+    plain deletes in the routes, the folder cascade, and the sweeper. It is the
+    same repair the scheduler does when a dispatch consumes its own library row
+    (``_repoint_siblings_at_archive``), minus the part that cannot apply here:
+    nothing is being printed, so there is no archive to hand the work to.
+
+    Two things happen, and both matter on a different database:
+
+    * Items still waiting on one of these files are cancelled, saying which file
+      went. Without it a queued job sat there looking dispatchable and failed at
+      the printer with "Library file not found", days later and with nothing
+      naming the delete that caused it.
+    * Every remaining row referencing the file has ``library_file_id`` cleared.
+      That is what keeps it: ``print_queue.library_file_id`` is ``ON DELETE
+      CASCADE``, which SQLite does not enforce and PostgreSQL does, so those rows
+      were silently deleted there -- including finished ones, which is what a
+      batch order counts its progress from.
+
+    Rows already printing are left in place. One of those is a job on a machine
+    right now; the file being deleted is the copy in the library, not the copy
+    the printer is working from. Returns the number of items cancelled.
+    """
+    if not file_ids:
+        return 0
+
+    doomed: dict[int, list[int]] = {}
+    rows = (
+        await db.execute(
+            select(PrintQueueItem.id, PrintQueueItem.library_file_id)
+            .where(PrintQueueItem.library_file_id.in_(file_ids))
+            .where(PrintQueueItem.archive_id.is_(None))
+            # "skipped" is not terminal: clearing a printer's previous-success
+            # gate puts those items back to pending, onto a file that by then
+            # is gone.
+            .where(PrintQueueItem.status.in_(("pending", "skipped")))
+        )
+    ).all()
+    for item_id, lib_id in rows:
+        doomed.setdefault(lib_id, []).append(item_id)
+
+    if doomed:
+        names = dict(
+            (
+                await db.execute(select(LibraryFile.id, LibraryFile.filename).where(LibraryFile.id.in_(list(doomed))))
+            ).all()
+        )
+        # Naive UTC: `completed_at` is a naive column, and asyncpg rejects an
+        # aware value outright where SQLite silently drops the offset.
+        now = utcnow_naive()
+        # One statement per file rather than per item: the case this exists for
+        # is many copies of one file, and a folder delete can reach a lot of
+        # them at once.
+        for lib_id, item_ids in doomed.items():
+            await db.execute(
+                PrintQueueItem.__table__.update()
+                .where(PrintQueueItem.id.in_(item_ids))
+                .values(
+                    status="cancelled",
+                    completed_at=now,
+                    error_message=f"'{names.get(lib_id, 'The library file')}' was deleted from the library",
+                )
+            )
+        logger.info("Library delete: cancelled %d queued item(s) whose file was removed", len(rows))
+
+    await db.execute(
+        PrintQueueItem.__table__.update()
+        .where(PrintQueueItem.library_file_id.in_(file_ids))
+        .values(library_file_id=None)
+    )
+    return len(rows)
+
+
 async def delete_dependent_variants(db: AsyncSession, file_ids: list[int]) -> None:
     """Drop cross-model queue candidates that pointed at these files (#671).
 

backend/app/services/print_scheduler.py

@@ -11,7 +11,7 @@ from datetime import datetime, timezone
 from pathlib import Path
 
 from fastapi import HTTPException
-from sqlalchemy import func, select, update
+from sqlalchemy import delete, false, func, or_, select, true, update
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import selectinload
 
@@ -4751,6 +4751,141 @@ class PrintScheduler:
 
         return prev_item.status in ("completed", "cancelled")
 
+    async def _repoint_siblings_at_archive(
+        self,
+        db: AsyncSession,
+        *,
+        consumed_library_file_id: int,
+        archive_id: int,
+        dispatched_item_id: int,
+    ) -> int:
+        """Move the other queue items off a library row that is about to be deleted (#2819).
+
+        ``cleanup_library_after_dispatch`` consumes the library row: the printer-card
+        upload-and-print flow uploads a transient file, prints it, and deletes it.
+        Queue creation happily puts that flag on every copy of a ``quantity > 1``
+        request, and ``_clone_queue_item`` copies ``library_file_id`` onto batch
+        clones, so the first dispatch could pull the file out from under rows that
+        had not run yet. What those rows did next depended on the database, and
+        neither answer was right: SQLite ships with ``PRAGMA foreign_keys`` off, so
+        the ``ON DELETE CASCADE`` on ``print_queue.library_file_id`` never fired and
+        they were left pointing at a row that no longer existed, failing with
+        "Library file not found" whenever someone started them -- or sitting
+        ``pending`` forever under ``manual_start``. PostgreSQL enforces the same
+        constraint, so there the rows were deleted outright and the queued copies
+        simply vanished, with no error and no history.
+
+        The archive holds its own copy of the 3MF, so the remaining copies can print
+        from it instead. Two things happen here, and both must happen *before* the
+        delete -- afterwards there is nothing left to repair on PostgreSQL:
+
+        * every row still naming the file has ``library_file_id`` cleared, which is
+          what takes it out of the cascade's reach. That covers rows this cannot
+          re-point as well -- a copy already printing from its own archive, and the
+          finished ones, which are not spare parts but the record a batch order
+          counts its progress from.
+        * the rows that still need something to print are pointed at the archive.
+
+        Returns how many items were re-pointed.
+        """
+        variant_item_ids = (
+            (
+                await db.execute(
+                    select(PrintQueueVariant.queue_item_id).where(
+                        PrintQueueVariant.library_file_id == consumed_library_file_id
+                    )
+                )
+            )
+            .scalars()
+            .all()
+        )
+        # Candidate rows naming the consumed file have to go rather than be
+        # cleared: `library_file_id` is NOT NULL there, so there is no way to keep
+        # one out of the cascade. This is what PostgreSQL already does, and
+        # _candidates_for skips such a variant on SQLite anyway, so no selection
+        # outcome changes -- the two backends simply stop disagreeing about
+        # whether the row is still there.
+        #
+        # It also has to happen before the re-point below: a cross-model item
+        # (#671) picks a variant every pass and folds it onto the row, and
+        # _resolve_variant clears archive_id as it does so, which would undo the
+        # re-point on the very next lap.
+        await db.execute(delete(PrintQueueVariant).where(PrintQueueVariant.library_file_id == consumed_library_file_id))
+
+        # An item left with other candidates still has somewhere to go, and those
+        # carry their own target model -- pointing it at this archive would print a
+        # file the matcher never chose. It is excluded from the re-point and simply
+        # re-resolves against what is left.
+        surviving_variant_item_ids = set(
+            (
+                await db.execute(
+                    select(PrintQueueVariant.queue_item_id).where(
+                        PrintQueueVariant.queue_item_id.in_(variant_item_ids) if variant_item_ids else false()
+                    )
+                )
+            )
+            .scalars()
+            .all()
+        )
+        repoint_ids = set(
+            (
+                await db.execute(
+                    select(PrintQueueItem.id)
+                    .where(PrintQueueItem.id != dispatched_item_id)
+                    .where(PrintQueueItem.archive_id.is_(None))
+                    # "skipped" belongs here with the two live states: it is not
+                    # terminal. Clearing a printer's previous-success gate puts
+                    # every item skipped by it back to "pending"
+                    # (resume_after_failure), and one restored onto a deleted
+                    # file is the same orphan by a slower route. "failed",
+                    # "cancelled", "aborted" and "completed" never return.
+                    .where(PrintQueueItem.status.in_(("pending", "printing", "skipped")))
+                    .where(
+                        PrintQueueItem.id.notin_(surviving_variant_item_ids) if surviving_variant_item_ids else true()
+                    )
+                    .where(
+                        or_(
+                            PrintQueueItem.library_file_id == consumed_library_file_id,
+                            PrintQueueItem.id.in_(variant_item_ids) if variant_item_ids else false(),
+                        )
+                    )
+                )
+            )
+            .scalars()
+            .all()
+        )
+        if repoint_ids:
+            await db.execute(
+                update(PrintQueueItem)
+                .where(PrintQueueItem.id.in_(repoint_ids))
+                .values(
+                    archive_id=archive_id,
+                    library_file_id=None,
+                    # The file this flag named is already consumed. Leaving it set
+                    # would arm every re-pointed copy to delete whatever library
+                    # row it is next given.
+                    cleanup_library_after_dispatch=False,
+                )
+            )
+            logger.info(
+                "Queue items %s: re-pointed at archive %s -- library file %s was consumed by item %s",
+                sorted(repoint_ids),
+                archive_id,
+                consumed_library_file_id,
+                dispatched_item_id,
+            )
+
+        # Everything else that still names the file: taken out of the cascade's
+        # reach without touching what it prints. The file is gone either way; what
+        # this preserves is the row.
+        await db.execute(
+            update(PrintQueueItem)
+            .where(PrintQueueItem.id != dispatched_item_id)
+            .where(PrintQueueItem.library_file_id == consumed_library_file_id)
+            .values(library_file_id=None)
+        )
+        return len(repoint_ids)
+
     async def _power_off_if_needed(self, db: AsyncSession, item: PrintQueueItem):
         """Schedule power-off if the queue item enabled auto_off_after.
 
@@ -5085,11 +5220,28 @@ class PrintScheduler:
             result = await db.execute(LibraryFile.active().where(LibraryFile.id == item.library_file_id))
             library_file = result.scalar_one_or_none()
             if not library_file:
+                # "Not found" covers two different situations and only one of
+                # them is recoverable, so say which. A trashed file is still
+                # there and restoring it makes a re-queued job work; a file that
+                # is really gone needs a different one. Neither is knowable from
+                # the queue, which is the whole complaint about this message.
+                trashed = (
+                    await db.execute(select(LibraryFile.filename).where(LibraryFile.id == item.library_file_id))
+                ).scalar_one_or_none()
                 item.status = "failed"
-                item.error_message = "Library file not found"
+                item.error_message = (
+                    f"'{trashed}' is in the library trash — restore it and queue the print again"
+                    if trashed
+                    else "Library file not found — it was deleted after this job was queued"
+                )
                 item.completed_at = datetime.now(timezone.utc)
                 await db.commit()
-                logger.error("Queue item %s: Library file %s not found", item.id, item.library_file_id)
+                logger.error(
+                    "Queue item %s: library file %s is %s",
+                    item.id,
+                    item.library_file_id,
+                    "in the trash" if trashed else "gone",
+                )
                 await self._power_off_if_needed(db, item)
                 return
             # Library files store absolute paths
@@ -5099,6 +5251,11 @@ class PrintScheduler:
 
             # Create archive from library file so usage tracking has access to the 3MF
             queue_item_id = item.id
+            # Held separately: a cleanup dispatch clears item.library_file_id
+            # below, and the log line at the end of this block reported that
+            # cleared field -- so every consumed print logged "from library
+            # file None", which is the one case worth being able to trace.
+            source_library_file_id = item.library_file_id
             try:
                 from backend.app.services.archive import ArchiveService
 
@@ -5118,6 +5275,7 @@ class PrintScheduler:
                     if budget_reservation is not None:
                         budget_reservation.print_archive_id = archive.id
                     if item.cleanup_library_after_dispatch and not library_file.is_external:
+                        consumed_library_file_id = library_file.id
                         item.library_file_id = None
                         cleanup_disk_paths.append(file_path)
                         if library_file.thumbnail_path:
@@ -5125,6 +5283,14 @@ class PrintScheduler:
                             if not thumb_path.is_absolute():
                                 thumb_path = settings.base_dir / library_file.thumbnail_path
                             cleanup_disk_paths.append(thumb_path)
+                        # Before the delete, not after: on PostgreSQL the FK
+                        # cascade would already have taken these rows (#2819).
+                        await self._repoint_siblings_at_archive(
+                            db,
+                            consumed_library_file_id=consumed_library_file_id,
+                            archive_id=archive.id,
+                            dispatched_item_id=item.id,
+                        )
                         await db.delete(library_file)
                         file_path = settings.base_dir / archive.file_path
                         filename = archive.filename
@@ -5138,7 +5304,7 @@ class PrintScheduler:
                         "Queue item %s: Created archive %s from library file %s",
                         item.id,
                         archive.id,
-                        item.library_file_id,
+                        source_library_file_id,
                     )
             except Exception as e:
                 logger.warning(

backend/tests/integration/test_library_trash_api.py

@@ -415,3 +415,184 @@ async def test_trashed_file_hidden_from_makerworld_dedupe(async_client: AsyncCli
         select(LibraryFile).where(LibraryFile.source_url == "https://makerworld.com/en/models/99#profileId-1")
     )
     assert direct.scalar_one_or_none() is not None
+
+
+# ---------------------------------------------------------------------------
+# Queued work must not be destroyed by a library delete (#2819)
+#
+# `print_queue.library_file_id` is ON DELETE CASCADE. SQLite does not enforce
+# it, so a queued job was left pointing at a row that no longer existed and
+# failed at the printer with "Library file not found", days later and with
+# nothing naming the delete that caused it. PostgreSQL does enforce it, so the
+# rows were deleted outright -- including finished ones, which is what a batch
+# order counts its progress from.
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture
+async def queue_factory(db_session):
+    """Queue items on a throwaway printer."""
+    from backend.app.models.print_queue import PrintQueueItem
+    from backend.app.models.printer import Printer
+
+    printer = Printer(
+        name="Queue test printer",
+        serial_number="QUEUE-TEST-2819",
+        ip_address="127.0.0.1",
+        access_code="access-code",
+        model="X1C",
+    )
+    db_session.add(printer)
+    await db_session.commit()
+
+    async def _create(library_file, status="pending", **kwargs):
+        item = PrintQueueItem(printer_id=printer.id, library_file_id=library_file.id, status=status, **kwargs)
+        db_session.add(item)
+        await db_session.commit()
+        await db_session.refresh(item)
+        return item
+
+    return _create
+
+
+async def _reload(db_session, item):
+    """Re-read the row the request's own session wrote.
+
+    Expunged rather than expired: the delete happens in the app's session, so
+    this one holds a stale copy that must be dropped instead of refreshed.
+    """
+    from sqlalchemy import select
+
+    from backend.app.models.print_queue import PrintQueueItem
+
+    item_id = item.id
+    db_session.expunge_all()
+    return (await db_session.execute(select(PrintQueueItem).where(PrintQueueItem.id == item_id))).scalar_one_or_none()
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_hard_delete_cancels_queued_items_and_names_the_file(
+    async_client: AsyncClient, file_factory, queue_factory, db_session
+):
+    """An external file is hard-deleted, so the jobs waiting on it are cancelled."""
+    f = await file_factory(filename="doomed.3mf", is_external=True)
+    waiting = await queue_factory(f)
+
+    response = await async_client.delete(f"/api/v1/library/files/{f.id}")
+    assert response.status_code == 200
+    assert response.json()["trashed"] is False
+
+    item = await _reload(db_session, waiting)
+    assert item is not None, "the cascade must not take the row with the file"
+    assert item.status == "cancelled"
+    # The point of cancelling here rather than letting it fail at the printer
+    # later: the message can still name what went.
+    assert "doomed.3mf" in item.error_message
+    assert item.library_file_id is None
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_hard_delete_leaves_finished_and_running_items_their_outcome(
+    async_client: AsyncClient, file_factory, queue_factory, db_session
+):
+    f = await file_factory(is_external=True)
+    done = await queue_factory(f, status="completed")
+    running = await queue_factory(f, status="printing")
+
+    await async_client.delete(f"/api/v1/library/files/{f.id}")
+
+    for item, expected in ((done, "completed"), (running, "printing")):
+        row = await _reload(db_session, item)
+        assert row is not None
+        # A finished run is a record, and a printing one is a job on a machine
+        # right now -- neither is cancelled. Only the dangling reference goes,
+        # which is what keeps the row out of the cascade.
+        assert row.status == expected
+        assert row.library_file_id is None
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_trashing_a_file_leaves_queued_items_alone(
+    async_client: AsyncClient, file_factory, queue_factory, db_session
+):
+    """Trash is reversible, so the queue is not rewritten on the way in."""
+    f = await file_factory()
+    waiting = await queue_factory(f)
+
+    response = await async_client.delete(f"/api/v1/library/files/{f.id}")
+    assert response.json()["trashed"] is True
+
+    item = await _reload(db_session, waiting)
+    assert item.status == "pending"
+    assert item.library_file_id == f.id
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_folder_delete_releases_items_anywhere_in_the_subtree(
+    async_client: AsyncClient, file_factory, queue_factory, db_session
+):
+    """The folder cascade reaches the whole tree, so the release has to as well."""
+    from backend.app.models.library import LibraryFolder
+
+    parent = LibraryFolder(name="parent")
+    db_session.add(parent)
+    await db_session.commit()
+    child = LibraryFolder(name="child", parent_id=parent.id)
+    db_session.add(child)
+    await db_session.commit()
+
+    nested = await file_factory(filename="nested.3mf", folder_id=child.id)
+    waiting = await queue_factory(nested)
+
+    response = await async_client.delete(f"/api/v1/library/folders/{parent.id}")
+    assert response.status_code == 200
+
+    item = await _reload(db_session, waiting)
+    assert item is not None
+    assert item.status == "cancelled"
+    assert item.library_file_id is None
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_sweeper_releases_items_before_hard_deleting(file_factory, queue_factory, db_session):
+    """The retention sweep is the usual way a managed file finally goes."""
+    from backend.app.services.library_trash import library_trash_service
+
+    f = await file_factory(filename="swept.3mf")
+    f.deleted_at = datetime.now(timezone.utc) - timedelta(days=400)
+    await db_session.commit()
+    waiting = await queue_factory(f)
+
+    swept = await library_trash_service._sweep(db_session)
+    assert swept == 1
+
+    item = await _reload(db_session, waiting)
+    assert item is not None
+    assert item.status == "cancelled"
+    assert "swept.3mf" in item.error_message
+    assert item.library_file_id is None
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_release_reports_and_cancels_every_copy_of_one_file(file_factory, queue_factory, db_session):
+    """The case this exists for: several copies queued from one file."""
+    from backend.app.services.library_trash import release_queue_references
+
+    f = await file_factory(filename="many-copies.3mf")
+    copies = [await queue_factory(f) for _ in range(3)]
+
+    # Counted in items, not files -- the caller logs it.
+    assert await release_queue_references(db_session, [f.id]) == 3
+    await db_session.commit()
+
+    for copy in copies:
+        row = await _reload(db_session, copy)
+        assert row.status == "cancelled"
+        assert "many-copies.3mf" in row.error_message
+        assert row.library_file_id is None

backend/tests/unit/test_scheduler_cleanup_library.py

@@ -4,6 +4,7 @@ from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
+from sqlalchemy import select
 from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
 
 import backend.app.models  # noqa: F401 - populate Base.metadata
@@ -11,7 +12,7 @@ import backend.app.services.print_scheduler as scheduler_module
 from backend.app.core.database import Base
 from backend.app.models.archive import PrintArchive
 from backend.app.models.library import LibraryFile
-from backend.app.models.print_queue import PrintQueueItem
+from backend.app.models.print_queue import PrintQueueItem, PrintQueueVariant
 from backend.app.models.printer import Printer
 from backend.app.services.print_scheduler import PrintScheduler
 
@@ -25,7 +26,7 @@ async def queue_factory(tmp_path):
     session_maker = async_sessionmaker(engine, expire_on_commit=False)
     case_counter = 0
 
-    async def make_case(*, cleanup=True, is_external=False, thumbnail_path=None):
+    async def make_case(*, cleanup=True, is_external=False, thumbnail_path=None, siblings=()):
         nonlocal case_counter
         case_counter += 1
 
@@ -86,6 +87,65 @@ async def queue_factory(tmp_path):
                 nozzle_offset_cali="on",
             )
             db.add(item)
+            await db.flush()
+
+            # The other copies of a quantity>1 dispatch (#2819). Each entry is a
+            # dict of overrides: `status`, `own_archive` for a copy that already
+            # holds one, and `extra_variant` for a cross-model copy that keeps a
+            # candidate this cleanup does not consume.
+            sibling_ids = []
+            other_file = None
+            for spec in siblings:
+                sibling = PrintQueueItem(
+                    printer_id=printer.id,
+                    library_file_id=None if spec.get("variants") else library_file.id,
+                    status=spec.get("status", "pending"),
+                    cleanup_library_after_dispatch=cleanup,
+                )
+                if spec.get("own_archive"):
+                    own = PrintArchive(
+                        printer_id=printer.id,
+                        filename="already-dispatched.3mf",
+                        file_path="archives/already-dispatched.3mf",
+                        file_size=1,
+                        status="printing",
+                    )
+                    db.add(own)
+                    await db.flush()
+                    sibling.archive_id = own.id
+                db.add(sibling)
+                await db.flush()
+                if spec.get("variants"):
+                    db.add(
+                        PrintQueueVariant(
+                            queue_item_id=sibling.id,
+                            library_file_id=library_file.id,
+                            target_model="X1C",
+                            position=0,
+                        )
+                    )
+                    if spec.get("extra_variant"):
+                        if other_file is None:
+                            other_path = base_dir / "library" / f"other-{case_counter}.3mf"
+                            other_path.write_bytes(b"other source")
+                            other_file = LibraryFile(
+                                filename=f"other-{case_counter}.3mf",
+                                file_path=str(other_path),
+                                file_type="3mf",
+                                file_size=other_path.stat().st_size,
+                            )
+                            db.add(other_file)
+                            await db.flush()
+                        db.add(
+                            PrintQueueVariant(
+                                queue_item_id=sibling.id,
+                                library_file_id=other_file.id,
+                                target_model="P1S",
+                                position=1,
+                            )
+                        )
+                sibling_ids.append(sibling.id)
+
             await db.commit()
 
             return SimpleNamespace(
@@ -96,6 +156,8 @@ async def queue_factory(tmp_path):
                 printer_id=printer.id,
                 library_file_id=library_file.id,
                 queue_item_id=item.id,
+                sibling_ids=sibling_ids,
+                other_library_file_id=other_file.id if other_file is not None else None,
                 archive_path=None,
                 upload=AsyncMock(return_value=True),
                 start_print=MagicMock(return_value=True),
@@ -262,6 +324,150 @@ async def test_archive_copy_survives_library_cleanup(queue_factory):
     assert uploaded_path == ctx.archive_path
 
 
+async def _sibling_snapshot(ctx):
+    async with ctx.session_maker() as db:
+        return [await db.get(PrintQueueItem, sid) for sid in ctx.sibling_ids]
+
+
+async def _variant_files(ctx, sibling_id):
+    async with ctx.session_maker() as db:
+        rows = await db.execute(
+            select(PrintQueueVariant.library_file_id).where(PrintQueueVariant.queue_item_id == sibling_id)
+        )
+        return sorted(rows.scalars().all())
+
+
+# ---------------------------------------------------------------------------
+# Sibling copies of the same library row (#2819)
+#
+# `quantity > 1` on the printer-card upload-and-print flow puts the cleanup flag
+# on every copy, and batch clones inherit `library_file_id`. Consuming the row
+# for the first copy used to leave the others pointing at it, which failed with
+# "Library file not found" on SQLite and deleted the rows outright on
+# PostgreSQL, where the FK cascade is enforced.
+#
+# These run on SQLite, so they cover the orphan half directly. The cascade half
+# was verified by hand against a real PostgreSQL 16, building this same fixture
+# on both backends and comparing every row: without the fix the copies were gone
+# after the delete -- including the finished ones a batch order counts its
+# progress from, and a copy already printing from its own archive. With it, the
+# two backends agree row for row. `print_archives.library_file_id` is SET NULL,
+# so it is cleared by the same delete, which is why looking the archive up by
+# the consumed library id -- the obvious alternative fix -- cannot work there.
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_pending_copies_are_repointed_at_the_archive(queue_factory):
+    ctx = await queue_factory(cleanup=True, siblings=({}, {}))
+
+    await _dispatch_library_item(ctx)
+
+    item, library_file, archive = await _queue_snapshot(ctx)
+    assert library_file is None
+    for sibling in await _sibling_snapshot(ctx):
+        # Still queued -- the point is that they can now run, not that they run now.
+        assert sibling.status == "pending"
+        assert sibling.archive_id == archive.id
+        assert sibling.library_file_id is None
+        # Their file is already consumed; leaving this armed would delete
+        # whatever library row they were next given.
+        assert sibling.cleanup_library_after_dispatch is False
+
+
+@pytest.mark.asyncio
+async def test_copy_that_already_has_its_own_archive_keeps_it(queue_factory):
+    ctx = await queue_factory(cleanup=True, siblings=({"own_archive": True, "status": "printing"},))
+
+    await _dispatch_library_item(ctx)
+
+    _, _, archive = await _queue_snapshot(ctx)
+    (sibling,) = await _sibling_snapshot(ctx)
+    # It is mid-print from its own archive and does not need the library file.
+    # Re-pointing it would swap the file under a job already running.
+    assert sibling.archive_id != archive.id
+    assert sibling.status == "printing"
+    # Cleared all the same: on PostgreSQL a row still naming the file goes with
+    # it, and this one is a job that is currently printing.
+    assert sibling.library_file_id is None
+
+
+@pytest.mark.parametrize("status", ["completed", "failed", "cancelled", "aborted"])
+@pytest.mark.asyncio
+async def test_finished_copies_keep_their_outcome_and_survive_the_delete(queue_factory, status):
+    ctx = await queue_factory(cleanup=True, siblings=({"status": status},))
+
+    await _dispatch_library_item(ctx)
+
+    (sibling,) = await _sibling_snapshot(ctx)
+    # A finished row is a record of what happened, not a spare part -- it keeps
+    # its outcome and is not handed the archive.
+    assert sibling.status == status
+    assert sibling.archive_id is None
+    # But the reference has to go: it is the only thing tying the row to the
+    # cascade that would otherwise delete it, and a batch order counts its
+    # progress from rows exactly like this one.
+    assert sibling.library_file_id is None
+
+
+@pytest.mark.asyncio
+async def test_skipped_copy_is_repointed_because_it_can_come_back(queue_factory):
+    ctx = await queue_factory(cleanup=True, siblings=({"status": "skipped"},))
+
+    await _dispatch_library_item(ctx)
+
+    _, _, archive = await _queue_snapshot(ctx)
+    (sibling,) = await _sibling_snapshot(ctx)
+    # Clearing the printer's previous-success gate puts skipped items back to
+    # pending, so this one is only waiting -- not finished.
+    assert sibling.status == "skipped"
+    assert sibling.archive_id == archive.id
+    assert sibling.library_file_id is None
+
+
+@pytest.mark.asyncio
+async def test_copies_are_untouched_when_the_dispatch_does_not_consume_the_file(queue_factory):
+    ctx = await queue_factory(cleanup=False, siblings=({},))
+
+    await _dispatch_library_item(ctx)
+
+    item, library_file, _ = await _queue_snapshot(ctx)
+    assert library_file is not None
+    (sibling,) = await _sibling_snapshot(ctx)
+    assert sibling.library_file_id == ctx.library_file_id
+    assert sibling.archive_id is None
+
+
+@pytest.mark.asyncio
+async def test_cross_model_copy_keeps_its_other_candidate_instead_of_the_archive(queue_factory):
+    ctx = await queue_factory(cleanup=True, siblings=({"variants": True, "extra_variant": True},))
+
+    await _dispatch_library_item(ctx)
+
+    (sibling,) = await _sibling_snapshot(ctx)
+    # It still has somewhere to go, and that candidate carries its own target
+    # model -- pointing it at this archive would print a file the matcher never
+    # chose.
+    assert sibling.archive_id is None
+    assert sibling.library_file_id is None
+    assert await _variant_files(ctx, sibling.id) == [ctx.other_library_file_id]
+
+
+@pytest.mark.asyncio
+async def test_copy_whose_only_candidate_was_consumed_is_repointed(queue_factory):
+    ctx = await queue_factory(cleanup=True, siblings=({"variants": True},))
+
+    await _dispatch_library_item(ctx)
+
+    _, _, archive = await _queue_snapshot(ctx)
+    (sibling,) = await _sibling_snapshot(ctx)
+    # Its one candidate is gone. Without the re-point the resolver would hold it
+    # pending forever with nothing left to dispatch.
+    assert sibling.archive_id == archive.id
+    assert sibling.library_file_id is None
+    assert await _variant_files(ctx, sibling.id) == []
+
+
 @pytest.mark.asyncio
 async def test_oserror_during_unlink_logs_orphan_path_and_does_not_crash_dispatch(queue_factory, caplog):
     ctx = await queue_factory(cleanup=True, thumbnail_path="relative")