Archives: charge each print as its own run (archive redesign P1b)

- A print's charge goes by its run: the run's billing id and cost
  centre, and how the run itself ended, under a lock on the run row
- A run without a queue item gets its own billing id, so a print of an
  already charged archive started on the printer is charged too
- An end without its queue item frees only the archive's reservations
  whose items are over, never another plate's still pending or printing
- The run write at a print's end is retried on a locked SQLite database

backend/app/main.py

@@ -9512,9 +9512,13 @@ async def _close_out_print(printer_id: int, data: dict):
             async with async_session() as db:
                 from backend.app.models.archive import PrintArchive
                 from backend.app.services.finance_billing import apply_print_charge_for_archive
+                from backend.app.services.print_log import run_being_ended
 
                 archive = await db.get(PrintArchive, archive_id)
-                if archive and billing_run_id is None:
+                # The run this end fills in below: charged as its own, not as
+                # the archive's latest dispatch (another plate of it, maybe).
+                charged_run = await run_being_ended(db, archive_id, printer_id, archive.started_at) if archive else None
+                if archive and billing_run_id is None and charged_run is None:
                     billing_run_id = getattr(archive, "billing_run_id", None)
                 if archive and archive.created_by_id is None and _print_user_info:
                     archive.created_by_id = await _print_user_id_if_still_there(db, _print_user_info)
@@ -9543,6 +9547,8 @@ async def _close_out_print(printer_id: int, data: dict):
                     print_run_id=billing_run_id,
                     base_cost_override=billing_base_cost,
                     filament_usage=filament_usage,
+                    run_entry_id=charged_run.id if charged_run is not None else None,
+                    run_status=run_status,
                 )
                 await db.commit()
                 if charged:
@@ -9601,110 +9607,118 @@ async def _close_out_print(printer_id: int, data: dict):
     # the one started last, which with plates of it on two printers need not
     # be the one that just ended.
     completed_run_id: int | None = None
-    try:
-        async with async_session() as db:
-            from backend.app.models.archive import PrintArchive
-            from backend.app.services.print_log import discard_open_runs, record_archive_wear, write_log_entry
 
-            archive = await db.get(PrintArchive, archive_id)
-            if archive is None:
-                # Deleted while it printed: its run has nothing left to end on.
-                await discard_open_runs(db, printer_id)
-                await db.commit()
-            if archive:
-                # Back-fill created_by_id on reprint (#730): reprint reuses the
-                # source archive row rather than creating a new one, so an
-                # archive that was auto-created from a printer-initiated
-                # print (created_by_id=NULL) would otherwise stay unattributed
-                # forever. When we have a print-session user AND the archive
-                # has no attribution yet, credit the current user. Never
-                # overwrite an existing attribution — the original uploader
-                # keeps ownership.
-                _print_user_id = await _print_user_id_if_still_there(db, _print_user_info)
-                if archive.created_by_id is None and _print_user_id is not None:
-                    archive.created_by_id = _print_user_id
-                p_info = printer_manager.get_printer(printer_id)
-                # Per-run actuals — written to PrintLogEntry so stats reflect
-                # what THIS print actually used, not the source archive's
-                # first-run values (#1378). Helper handles the partial-print
-                # math (failed / cancelled / stopped get scaled to progress
-                # or to tracked spool deltas).
-                _run_status = data.get("status", "completed")
-                # #2614: scope the per-run estimate to the printed plate. For a
-                # multi-plate 3MF dispatched one plate at a time, the archive's
-                # filament/cost are the whole-file totals; the PrintLogEntry must
-                # reflect only this plate. No effect on single-plate archives (the
-                # plate estimate equals the whole-file value) or on the tracker
-                # path (measured spool deltas win in _compute_run_filament_grams).
-                _est_full_path = (
-                    app_settings.base_dir / archive.file_path if archive.file_path else None
-                )  # SEC-PATH-OK: archive.file_path is DB-stored, internally generated
-                # The plate that printed, as billing reads it: a reprint of
-                # another plate must not log the archive's first plate.
-                _run_plate_id = billing_plate_id if billing_plate_id is not None else notify_plate_id
-                _est_grams, _est_cost = _plate_scoped_run_estimate(archive, _est_full_path, _run_plate_id)
-                if _run_plate_id is None:
-                    _run_plate_id = archive.plate_id
-                _run_grams = _compute_run_filament_grams(
-                    _run_status,
-                    _est_grams,
-                    data.get("last_progress", data.get("progress")),
-                    usage_results,
-                )
+    async def _write_run(db):
+        # Retried on a locked SQLite database: a run left open here would be
+        # closed again by the next reconnect's reconcile.
+        nonlocal completed_run_id
+        from backend.app.models.archive import PrintArchive
+        from backend.app.services.print_log import discard_open_runs, record_archive_wear, write_log_entry
 
-                # Per-run cost — prefer usage_results sum. For partial prints
-                # we deliberately skip the topup-to-estimate logic in
-                # usage_tracker (which assumes the print completed); the raw
-                # tracked-spool sum is closer to what THIS run actually cost.
-                _run_cost: float | None = None
-                if usage_results:
-                    _run_cost = sum(r.get("cost") or 0 for r in usage_results) or None
-                if _run_cost is None and _run_status != "completed":
-                    _run_cost = spoolman_run_cost
-                if _run_cost is None and _run_status == "completed":
-                    _run_cost = _est_cost
+        archive = await db.get(PrintArchive, archive_id)
+        if archive is None:
+            # Deleted while it printed: its run has nothing left to end on.
+            await discard_open_runs(db, printer_id)
+            await db.commit()
+        if archive:
+            # Back-fill created_by_id on reprint (#730): reprint reuses the
+            # source archive row rather than creating a new one, so an
+            # archive that was auto-created from a printer-initiated
+            # print (created_by_id=NULL) would otherwise stay unattributed
+            # forever. When we have a print-session user AND the archive
+            # has no attribution yet, credit the current user. Never
+            # overwrite an existing attribution — the original uploader
+            # keeps ownership.
+            _print_user_id = await _print_user_id_if_still_there(db, _print_user_info)
+            if archive.created_by_id is None and _print_user_id is not None:
+                archive.created_by_id = _print_user_id
+            p_info = printer_manager.get_printer(printer_id)
+            # Per-run actuals — written to PrintLogEntry so stats reflect
+            # what THIS print actually used, not the source archive's
+            # first-run values (#1378). Helper handles the partial-print
+            # math (failed / cancelled / stopped get scaled to progress
+            # or to tracked spool deltas).
+            _run_status = data.get("status", "completed")
+            # #2614: scope the per-run estimate to the printed plate. For a
+            # multi-plate 3MF dispatched one plate at a time, the archive's
+            # filament/cost are the whole-file totals; the PrintLogEntry must
+            # reflect only this plate. No effect on single-plate archives (the
+            # plate estimate equals the whole-file value) or on the tracker
+            # path (measured spool deltas win in _compute_run_filament_grams).
+            _est_full_path = (
+                app_settings.base_dir / archive.file_path if archive.file_path else None
+            )  # SEC-PATH-OK: archive.file_path is DB-stored, internally generated
+            # The plate that printed, as billing reads it: a reprint of
+            # another plate must not log the archive's first plate.
+            _run_plate_id = billing_plate_id if billing_plate_id is not None else notify_plate_id
+            _est_grams, _est_cost = _plate_scoped_run_estimate(archive, _est_full_path, _run_plate_id)
+            if _run_plate_id is None:
+                _run_plate_id = archive.plate_id
+            _run_grams = _compute_run_filament_grams(
+                _run_status,
+                _est_grams,
+                data.get("last_progress", data.get("progress")),
+                usage_results,
+            )
 
-                from backend.app.models.printer import Printer as _Printer
+            # Per-run cost — prefer usage_results sum. For partial prints
+            # we deliberately skip the topup-to-estimate logic in
+            # usage_tracker (which assumes the print completed); the raw
+            # tracked-spool sum is closer to what THIS run actually cost.
+            _run_cost: float | None = None
+            if usage_results:
+                _run_cost = sum(r.get("cost") or 0 for r in usage_results) or None
+            if _run_cost is None and _run_status != "completed":
+                _run_cost = spoolman_run_cost
+            if _run_cost is None and _run_status == "completed":
+                _run_cost = _est_cost
+
+            from backend.app.models.printer import Printer as _Printer
+
+            _wear_rate = await db.scalar(select(_Printer.wear_cost_per_hour).where(_Printer.id == printer_id))
+            run_entry = await write_log_entry(
+                db,
+                archive_id=archive.id,
+                # Captured by _update_queue_status above; None for
+                # printer-initiated prints with no queue row. Batch
+                # cost/energy roll-up joins on it (#342).
+                queue_item_id=queue_item_id,
+                status=_run_status,
+                print_name=archive.print_name,
+                # The plate this run printed, as the estimate above takes
+                # it; the archive's own is the latest run's.
+                plate_id=_run_plate_id,
+                printer_name=p_info.name if p_info else None,
+                printer_id=printer_id,
+                started_at=archive.started_at,
+                completed_at=archive.completed_at,
+                filament_type=archive.filament_type,
+                filament_color=archive.filament_color,
+                filament_used_grams=_run_grams,
+                cost=_run_cost,
+                wear_cost_per_hour=_wear_rate,
+                failure_reason=archive.failure_reason,
+                thumbnail_path=archive.thumbnail_path,
+                created_by_id=archive.created_by_id,
+                created_by_username=_print_user_info.get("username") if _print_user_info else None,
+                # Reconciled completions have an unknown real end time —
+                # log 0 duration instead of the whole disconnect gap (#2592),
+                # unless it was estimated (#3310).
+                reconciled=bool(data.get("_reconciled")) and not data.get("_reconciled_completed_at"),
+                # Its time is the estimate or nothing, never measured (#3310).
+                duration_unmeasured=bool(data.get("_reconciled")),
+                # The real end of a print stopped while its printer was offline.
+                replaces_offline_stop=True,
+            )
+            await record_archive_wear(db, archive, run_entry)
+            await db.commit()
+            completed_run_id = run_entry.id
+            logger.info("[PRINT_LOG] Log entry written for archive %s", archive_id)
 
-                _wear_rate = await db.scalar(select(_Printer.wear_cost_per_hour).where(_Printer.id == printer_id))
-                run_entry = await write_log_entry(
-                    db,
-                    archive_id=archive.id,
-                    # Captured by _update_queue_status above; None for
-                    # printer-initiated prints with no queue row. Batch
-                    # cost/energy roll-up joins on it (#342).
-                    queue_item_id=queue_item_id,
-                    status=_run_status,
-                    print_name=archive.print_name,
-                    # The plate this run printed, as the estimate above takes
-                    # it; the archive's own is the latest run's.
-                    plate_id=_run_plate_id,
-                    printer_name=p_info.name if p_info else None,
-                    printer_id=printer_id,
-                    started_at=archive.started_at,
-                    completed_at=archive.completed_at,
-                    filament_type=archive.filament_type,
-                    filament_color=archive.filament_color,
-                    filament_used_grams=_run_grams,
-                    cost=_run_cost,
-                    wear_cost_per_hour=_wear_rate,
-                    failure_reason=archive.failure_reason,
-                    thumbnail_path=archive.thumbnail_path,
-                    created_by_id=archive.created_by_id,
-                    created_by_username=_print_user_info.get("username") if _print_user_info else None,
-                    # Reconciled completions have an unknown real end time —
-                    # log 0 duration instead of the whole disconnect gap (#2592),
-                    # unless it was estimated (#3310).
-                    reconciled=bool(data.get("_reconciled")) and not data.get("_reconciled_completed_at"),
-                    # Its time is the estimate or nothing, never measured (#3310).
-                    duration_unmeasured=bool(data.get("_reconciled")),
-                    # The real end of a print stopped while its printer was offline.
-                    replaces_offline_stop=True,
-                )
-                await record_archive_wear(db, archive, run_entry)
-                await db.commit()
-                completed_run_id = run_entry.id
-                logger.info("[PRINT_LOG] Log entry written for archive %s", archive_id)
+    try:
+        from backend.app.core.database import run_with_retry
+
+        await run_with_retry(_write_run, label="print log entry")
     except Exception as e:
         logger.warning("[PRINT_LOG] Failed to write log entry for archive %s: %s", archive_id, e)
 

backend/app/services/finance_billing.py

@@ -7,6 +7,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from backend.app.models.archive import PrintArchive
 from backend.app.models.finance import TransactionType, UserWallet, WalletTransaction
+from backend.app.models.print_log import INCLUDE_OPEN_RUNS, PrintLogEntry
 from backend.app.services.finance_balance import sync_personal_wallet_balance
 from backend.app.services.finance_budget import is_billing_enabled, release_budget_reservation
 
@@ -63,16 +64,20 @@ def _calculate_partial_charge(
     base_cost: float,
     *,
     filament_usage: tuple[float | None, float | None] | None = None,
+    status: str | None = None,
 ) -> tuple[float, str]:
     """Calculate proportional charge for partial prints based on filament usage.
 
+    ``status`` is how the charged run ended; without it, the archive's.
+
     Returns (charge_amount, description_suffix) where:
     - charge_amount: absolute cost to charge (0 if insufficient data)
     - description_suffix: reason/details for transaction description
     """
     try:
+        status = status or archive.status
         # Only apply proportional calculation for non-completed prints
-        if archive.status == "completed":
+        if status == "completed":
             return round(float(base_cost), 2), ""
 
         if filament_usage is not None:
@@ -96,7 +101,7 @@ def _calculate_partial_charge(
         # If we don't have reliable planned filament data, do not guess a partial charge.
         # Charging a failed/aborted print without an estimated baseline can overcharge users.
         if filament_planned is None or filament_planned <= 0:
-            return 0.0, f"[{archive.status}: insufficient filament data]"
+            return 0.0, f"[{status}: insufficient filament data]"
 
         # Calculate proportional cost
         filament_ratio = min(1.0, max(0.0, filament_used / filament_planned))  # Clamp to [0, 1]
@@ -105,13 +110,25 @@ def _calculate_partial_charge(
         # Round charges to 2 decimals for consistent persistence
         charge = round(charge, 2)
 
-        suffix = f"[{archive.status}: {filament_ratio:.1%} filament ({filament_used:.1f}g/{filament_planned:.1f}g)]"
+        suffix = f"[{status}: {filament_ratio:.1%} filament ({filament_used:.1f}g/{filament_planned:.1f}g)]"
         return charge, suffix
     except ValueError as e:
         logger.error(f"Value error in _calculate_partial_charge: {e}", exc_info=True)
         raise
 
 
+async def _settle_reservation(db: AsyncSession, status: str, *, archive_id: int, print_queue_id: int | None) -> None:
+    """Release or consume what the charged print reserved. Reservations are
+    made per queue item. Without the print's own (no queue item, or its item
+    closed already, e.g. cancelled by a stop before the printer reported the
+    start) the archive's whose items are over, never those of another plate
+    of it still pending or printing."""
+    if print_queue_id is not None:
+        await release_budget_reservation(db, source_type="print_queue", source_id=print_queue_id, status=status)
+    else:
+        await release_budget_reservation(db, print_archive_id=archive_id, status=status)
+
+
 async def apply_print_charge_for_archive(
     db: AsyncSession,
     archive_id: int,
@@ -122,43 +139,72 @@ async def apply_print_charge_for_archive(
     print_run_id: str | None = None,
     base_cost_override: float | None = None,
     filament_usage: tuple[float | None, float | None] | None = None,
+    run_entry_id: int | None = None,
+    run_status: str | None = None,
 ) -> bool:
     """Apply an idempotent wallet charge for a print archive.
 
     Charges completed prints at full cost, and partial/failed prints proportionally
     based on actual filament used vs. planned filament.
 
+    ``run_entry_id`` is the run being charged (its print log entry) and
+    ``run_status`` how it ended. The charge is then the run's: its billing run
+    id and cost centre, under a lock on the run row, so plates of
+    one archive printing on two printers are each charged once. Without a run
+    (a print started before runs were written at the start) the archive's.
+
     Returns True when a new wallet transaction was created.
     """
     try:
         if not await is_billing_enabled(db):
-            if print_queue_id is not None:
-                await release_budget_reservation(
-                    db, source_type="print_queue", source_id=print_queue_id, status="released"
-                )
-            else:
-                await release_budget_reservation(db, print_archive_id=archive_id, status="released")
+            await _settle_reservation(db, "released", archive_id=archive_id, print_queue_id=print_queue_id)
             logger.info("Billing is disabled; skipping print charge for archive ID %s.", archive_id)
             return False
 
-        archive = (
-            await db.execute(select(PrintArchive).where(PrintArchive.id == archive_id).with_for_update())
-        ).scalar_one_or_none()
+        run = None
+        if run_entry_id is not None:
+            # Locked, and read again under the lock (the caller loaded it):
+            # two ends of the same run (a repeated completion) mint and charge
+            # its billing run id once.
+            run = (
+                await db.execute(
+                    select(PrintLogEntry)
+                    .where(PrintLogEntry.id == run_entry_id)
+                    .with_for_update()
+                    .execution_options(populate_existing=True, **{INCLUDE_OPEN_RUNS: True})
+                )
+            ).scalar_one_or_none()
+
+        archive_query = select(PrintArchive).where(PrintArchive.id == archive_id)
+        if run is None:
+            archive_query = archive_query.with_for_update()
+        archive = (await db.execute(archive_query)).scalar_one_or_none()
         if archive is None:
             logger.warning(f"Archive with ID {archive_id} not found.")
             return False
 
-        effective_run_id = print_run_id or archive.billing_run_id
-        # The archive-level flag is retained only for legacy deleted charges.
-        # A new scheduler dispatch clears it while persisting its new run UUID;
-        # current deletions are represented by a voided transaction instead.
-        if archive.wallet_charge_skipped:
+        if run is not None:
+            # A run opened at its start has its own id. One written only at
+            # its end (no start seen) has none, and is charged as before by the
+            # archive's, which a repeated end then finds charged.
+            effective_run_id = print_run_id or run.billing_run_id or archive.billing_run_id
+            skipped = bool(run.wallet_charge_skipped)
+        else:
+            effective_run_id = print_run_id or archive.billing_run_id
+            # The archive-level flag is retained only for legacy deleted charges.
+            # A new scheduler dispatch clears it while persisting its new run UUID;
+            # current deletions are represented by a voided transaction instead.
+            skipped = archive.wallet_charge_skipped
+        if skipped:
             logger.info(f"Wallet charge skipped for archive ID {archive_id}.")
             return False
 
-        # Accept completed, aborted, cancelled, and failed prints
-        if archive.status not in ("completed", "aborted", "cancelled", "failed"):
-            logger.info(f"Archive ID {archive_id} has status {archive.status}, which is not chargeable.")
+        # Accept completed, aborted, cancelled, and failed prints. A run's own
+        # end: with another plate of the archive ending at the same time, the
+        # archive's status may be that one's.
+        charged_status = run_status or archive.status
+        if charged_status not in ("completed", "aborted", "cancelled", "failed"):
+            logger.info(f"Archive ID {archive_id} has status {charged_status}, which is not chargeable.")
             return False
 
         # A queue print's user is the one who queued it, also when deleted
@@ -176,7 +222,12 @@ async def apply_print_charge_for_archive(
             actual_user_id = None
         else:
             actual_user_id = archive.created_by_id
-        actual_cost_center_id = cost_center_id if cost_center_id is not None else archive.cost_center_id
+        if cost_center_id is not None:
+            actual_cost_center_id = cost_center_id
+        elif run is not None and run.cost_center_id is not None:
+            actual_cost_center_id = run.cost_center_id
+        else:
+            actual_cost_center_id = archive.cost_center_id
         # No user: deleted while the print ran, or never had one (login turned
         # off). A cost centre's print is still its cost: charged to the cost
         # centre without a user (shown as "No user"), which also consumes its
@@ -194,21 +245,20 @@ async def apply_print_charge_for_archive(
             # Nothing to charge, but the print is over: settle what it held
             # (reserved from an estimate, e.g. an older archive with grams but
             # no stored cost), as the zero-charge case below does.
-            if print_queue_id is not None:
-                await release_budget_reservation(
-                    db, source_type="print_queue", source_id=print_queue_id, status="released"
-                )
-            else:
-                await release_budget_reservation(db, print_archive_id=archive.id, status="released")
+            await _settle_reservation(db, "released", archive_id=archive.id, print_queue_id=print_queue_id)
             logger.info(f"Base cost for archive ID {archive_id} is zero or negative.")
             return False
 
-        # New dispatches persist a UUID before sending the printer command.
-        # Generate one here only for legacy/in-flight rows created before that
-        # migration; the locked archive row makes this fallback durable.
+        # New dispatches persist a UUID before sending the printer command,
+        # and a run gets its own when it opens. Generate one here only for
+        # legacy/in-flight rows created before that; the locked row makes this
+        # fallback durable.
         if not effective_run_id:
             effective_run_id = str(uuid.uuid4())
-            archive.billing_run_id = effective_run_id
+            if run is None:
+                archive.billing_run_id = effective_run_id
+        if run is not None and run.billing_run_id != effective_run_id:
+            run.billing_run_id = effective_run_id
 
         tx_conditions = [
             WalletTransaction.transaction_type == TransactionType.PRINT_CHARGE.value,
@@ -237,14 +287,10 @@ async def apply_print_charge_for_archive(
             archive,
             base_cost,
             filament_usage=filament_usage,
+            status=charged_status,
         )
         if charge <= 0:
-            if print_queue_id is not None:
-                await release_budget_reservation(
-                    db, source_type="print_queue", source_id=print_queue_id, status="released"
-                )
-            else:
-                await release_budget_reservation(db, print_archive_id=archive.id, status="released")
+            await _settle_reservation(db, "released", archive_id=archive.id, print_queue_id=print_queue_id)
             logger.info(f"Calculated charge for archive ID {archive_id} is zero or negative.")
             return False
 
@@ -313,10 +359,7 @@ async def apply_print_charge_for_archive(
         new_wallet_balance = await sync_personal_wallet_balance(db, wallet) if wallet is not None else None
 
         # Consume matching budget reservations after the transaction is persisted
-        if print_queue_id is not None:
-            await release_budget_reservation(db, source_type="print_queue", source_id=print_queue_id, status="consumed")
-        else:
-            await release_budget_reservation(db, print_archive_id=archive.id, status="consumed")
+        await _settle_reservation(db, "consumed", archive_id=archive.id, print_queue_id=print_queue_id)
         logger.info(f"Applied print charge for archive ID {archive_id}. New balance: {new_wallet_balance}.")
         return True
     except SQLAlchemyError as e:

backend/app/services/finance_budget.py

@@ -6,7 +6,7 @@ from typing import TYPE_CHECKING
 from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
 
 from fastapi import HTTPException
-from sqlalchemy import case, func, select
+from sqlalchemy import case, func, or_, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from backend.app.models.finance import BudgetReservation, CostCenter, CostCenterMember, WalletTransaction
@@ -281,9 +281,19 @@ async def release_budget_reservation(
     print_archive_id: int | None = None,
     status: str = "released",
 ) -> int:
+    """By the archive, the reservations of its queue items that are over: one
+    still pending or printing is another print's (another plate of it)."""
     conditions = [BudgetReservation.status == "active"]
     if print_archive_id is not None:
         conditions.append(BudgetReservation.print_archive_id == print_archive_id)
+        conditions.append(
+            or_(
+                BudgetReservation.source_type != "print_queue",
+                BudgetReservation.source_id.not_in(
+                    select(PrintQueueItem.id).where(PrintQueueItem.status.in_(("pending", "printing")))
+                ),
+            )
+        )
     else:
         conditions.extend(
             [

backend/app/services/print_log.py

@@ -4,6 +4,7 @@ Log entries are written to a separate table and never touch archives or queue it
 """
 
 import logging
+import uuid
 from collections.abc import Set as AbstractSet
 from datetime import datetime, timezone
 
@@ -78,9 +79,10 @@ async def write_log_entry(
     such a run of its printer, the printer's latest, which its archive's start
     (another plate's since, maybe) and its closed queue item no longer find.
     """
-    entry = await _entry_of_this_run(db, archive_id, printer_id, started_at)
-    if entry is None and replaces_offline_stop and archive_id is not None and printer_id is not None:
-        entry = await _stopped_while_offline(db, archive_id, printer_id)
+    if replaces_offline_stop:
+        entry = await run_being_ended(db, archive_id, printer_id, started_at)
+    else:
+        entry = await _entry_of_this_run(db, archive_id, printer_id, started_at)
     filling_open_run = entry is not None and entry.status == RUN_OPEN
     if entry is not None and entry.started_at is not None:
         # The run's own start. The archive's is its latest run's: another
@@ -296,6 +298,18 @@ async def _entry_of_this_run(
     return await _open_run_on(db, archive_id, printer_id)
 
 
+async def run_being_ended(
+    db: AsyncSession, archive_id: int | None, printer_id: int | None, started_at: datetime | None
+) -> PrintLogEntry | None:
+    """The run a print's end fills in (``write_log_entry`` with
+    ``replaces_offline_stop``): the one written for it, else the one a stop
+    logged while the printer was offline. Billing charges the same run."""
+    entry = await _entry_of_this_run(db, archive_id, printer_id, started_at)
+    if entry is None and archive_id is not None and printer_id is not None:
+        entry = await _stopped_while_offline(db, archive_id, printer_id)
+    return entry
+
+
 async def _stopped_while_offline(db: AsyncSession, archive_id: int, printer_id: int) -> PrintLogEntry | None:
     """The run a print's real end replaces after a stop logged while the
     printer was offline: the printer's latest run, when it is of this
@@ -369,7 +383,9 @@ async def open_run(
     Idempotent: a print start seen again (a restart resuming the print, a
     repeated report) finds the run it opened, keyed as its end will key it, by
     archive, printer and start time. The queue item running it, if any, gives
-    the dispatch's own billing run and cost centre. Does not commit.
+    the dispatch's own billing run and cost centre; a run without one gets its
+    own billing run id, so each print of the archive is charged as its own.
+    Does not commit.
     """
     from backend.app.models.print_queue import PrintQueueItem
 
@@ -407,7 +423,9 @@ async def open_run(
         thumbnail_path=archive.thumbnail_path,
         created_by_id=archive.created_by_id,
         subtask_id=archive.subtask_id,
-        billing_run_id=(item.billing_run_id if item is not None else None) or archive.billing_run_id,
+        # Not the archive's: that is the latest dispatch's, or the id an
+        # earlier print of the archive was charged under.
+        billing_run_id=(item.billing_run_id if item is not None else None) or str(uuid.uuid4()),
         cost_center_id=(item.cost_center_id if item is not None else None) or archive.cost_center_id,
         confirm_requested=archive.confirm_requested,
         quantity=archive.quantity,

backend/tests/integration/test_run_lifecycle.py

@@ -1177,3 +1177,196 @@ async def test_a_stale_archive_cancelled_at_the_next_start_leaves_no_run_open(te
     archive = await db_session.get(PrintArchive, farm.archive)
     assert archive.status == "cancelled"
     assert await _all_runs(db_session, farm.archive) == []
+
+
+async def _charges(db_session) -> list[WalletTransaction]:
+    db_session.expire_all()
+    return list(
+        (
+            await db_session.scalars(
+                select(WalletTransaction)
+                .where(WalletTransaction.transaction_type == "print_charge")
+                .order_by(WalletTransaction.id)
+            )
+        ).all()
+    )
+
+
+async def _print_on_the_printer(test_engine, printer_id: int, status: str = "completed") -> None:
+    """A print of the archive started at the printer, no queue item, to its end."""
+    await _drive_print_start(test_engine, SimpleNamespace(id=printer_id), download_ok=False)
+    await _complete(test_engine, printer_id, status, name=SUBTASK)
+
+
+async def test_each_print_of_an_archive_started_on_the_printer_is_charged(test_engine, db_session, farm):
+    """A print without a queue item is charged under its run's own id, not
+    under the id an earlier print of the archive was charged under."""
+    await _run(test_engine, db_session, farm, plate=1, status="completed", run_id="run-1")
+    # As the scheduler's dispatch left it.
+    (await db_session.get(PrintArchive, farm.archive)).billing_run_id = "run-1"
+    await db_session.commit()
+
+    register_expected_print(farm.printer, FILE, farm.archive, plate_id=1, cost_center_id=farm.center)
+    await _print_on_the_printer(test_engine, farm.printer)
+    register_expected_print(farm.printer, FILE, farm.archive, plate_id=1, cost_center_id=farm.center)
+    await _print_on_the_printer(test_engine, farm.printer)
+
+    run_ids = [r.billing_run_id for r in await _runs(db_session, farm.archive)]
+    charged = [c.print_run_id for c in await _charges(db_session)]
+    assert len(set(charged)) == 3
+    assert charged == run_ids and run_ids[0] == "run-1"
+
+
+async def test_a_plates_end_leaves_the_budget_held_for_another_plate(test_engine, db_session, farm, printer_factory):
+    """Plate 2 queued on another printer holds a reservation on the same
+    archive: plate 1's end, without a queue item, does not free it."""
+    from backend.app.models.finance import BudgetReservation
+
+    other = (await printer_factory(name="Q")).id
+    item = PrintQueueItem(
+        printer_id=other,
+        archive_id=farm.archive,
+        status="printing",
+        position=1,
+        plate_id=2,
+        created_by_id=farm.user,
+        cost_center_id=farm.center,
+        billing_run_id="run-2",
+    )
+    db_session.add(item)
+    await db_session.flush()
+    hold = BudgetReservation(
+        cost_center_id=farm.center,
+        amount=5.0,
+        source_type="print_queue",
+        source_id=item.id,
+        print_archive_id=farm.archive,
+    )
+    db_session.add(hold)
+    await db_session.commit()
+    hold_id = hold.id
+
+    register_expected_print(farm.printer, FILE, farm.archive, plate_id=1, cost_center_id=farm.center)
+    await _print_on_the_printer(test_engine, farm.printer)
+
+    db_session.expire_all()
+    assert len(await _charges(db_session)) == 1
+    assert (await db_session.get(BudgetReservation, hold_id)).status == "active"
+
+
+async def test_a_run_is_charged_for_how_it_ended(db_session, farm):
+    """Another plate of the archive ending at the same moment may have set
+    the archive's status: the charge goes by the run's own end."""
+    from backend.app.services.finance_billing import apply_print_charge_for_archive
+
+    archive = await db_session.get(PrintArchive, farm.archive)
+    archive.status = "failed"
+    archive.extra_data = {"filament_grams_total": 100.0}
+    run = PrintLogEntry(
+        archive_id=farm.archive,
+        printer_id=farm.printer,
+        status="printing",
+        billing_run_id="run-1",
+        cost_center_id=farm.center,
+    )
+    db_session.add(run)
+    await db_session.commit()
+
+    assert await apply_print_charge_for_archive(
+        db_session,
+        farm.archive,
+        charged_user_id=farm.user,
+        run_entry_id=run.id,
+        run_status="completed",
+        filament_usage=(10.0, 100.0),
+    )
+    await db_session.commit()
+
+    [charge] = await _charges(db_session)
+    assert (charge.amount, charge.print_run_id, charge.cost_center_id) == (-5.0, "run-1", farm.center)
+
+
+async def test_a_locked_database_does_not_leave_the_run_open(test_engine, db_session, farm):
+    """The run write at a print's end is retried, or the next reconnect would
+    close the print a second time."""
+    from unittest.mock import patch
+
+    from sqlalchemy.exc import OperationalError
+
+    from backend.app.core.database import is_sqlite
+    from backend.app.services import print_log
+
+    if not is_sqlite():
+        pytest.skip("SQLite's single writer only")
+    real = print_log.write_log_entry
+    calls = []
+
+    async def locked_once(*args, **kwargs):
+        calls.append(1)
+        if len(calls) == 1:
+            raise OperationalError("UPDATE print_log_entries", {}, Exception("database is locked"))
+        return await real(*args, **kwargs)
+
+    register_expected_print(farm.printer, FILE, farm.archive, plate_id=1)
+    await _drive_print_start(test_engine, SimpleNamespace(id=farm.printer), download_ok=False)
+    with patch.object(print_log, "write_log_entry", side_effect=locked_once):
+        await _complete(test_engine, farm.printer, "completed", name=SUBTASK)
+
+    assert [r.status for r in await _all_runs(db_session, farm.archive)] == ["completed"]
+
+
+async def test_a_run_without_a_billing_id_is_charged_once(db_session, farm):
+    """A run opened before runs had their own billing id, of an archive with
+    none: the id minted at its charge stays on the run, so its end seen again
+    is not charged a second time."""
+    from backend.app.services.finance_billing import apply_print_charge_for_archive
+
+    run = PrintLogEntry(archive_id=farm.archive, printer_id=farm.printer, status="printing")
+    db_session.add(run)
+    await db_session.commit()
+
+    for _ in range(2):
+        await apply_print_charge_for_archive(
+            db_session, farm.archive, charged_user_id=farm.user, run_entry_id=run.id, run_status="completed"
+        )
+        await db_session.commit()
+
+    run_id = (await db_session.get(PrintLogEntry, run.id)).billing_run_id
+    assert [c.print_run_id for c in await _charges(db_session)] == [run_id]
+
+
+async def test_a_print_stopped_before_its_start_was_reported_frees_its_budget(test_engine, db_session, farm):
+    """Stopped while the printer still prepared: its queue item is cancelled
+    before the start report, so its run has no queue item. Its end still
+    frees what the item reserved."""
+    from backend.app.models.finance import BudgetReservation
+
+    item = PrintQueueItem(
+        printer_id=farm.printer,
+        archive_id=farm.archive,
+        status="cancelled",
+        position=1,
+        plate_id=1,
+        created_by_id=farm.user,
+        cost_center_id=farm.center,
+        billing_run_id="run-1",
+    )
+    db_session.add(item)
+    await db_session.flush()
+    hold = BudgetReservation(
+        cost_center_id=farm.center,
+        amount=5.0,
+        source_type="print_queue",
+        source_id=item.id,
+        print_archive_id=farm.archive,
+    )
+    db_session.add(hold)
+    (await db_session.get(PrintArchive, farm.archive)).cost = 0.0
+    await db_session.commit()
+    hold_id = hold.id
+
+    register_expected_print(farm.printer, FILE, farm.archive, plate_id=1, cost_center_id=farm.center)
+    await _print_on_the_printer(test_engine, farm.printer, status="cancelled")
+
+    db_session.expire_all()
+    assert (await db_session.get(BudgetReservation, hold_id)).status == "released"