fix(maintenance): a printer's time offline is not counted as print time; duplicate maintenance items are merged on upgrade and prevented by a unique index; the "performed" response honours the item's own interval type

CHANGELOG.md

@@ -164,6 +164,9 @@ All notable changes to Bambuddy will be documented in this file.
 - **The frontend build no longer warns about `path` and `crypto` being externalized for the STEP previewer (#2976)** — `occt-import-js`, the Emscripten build behind STEP previews, requires both modules, but only inside its `ENVIRONMENT_IS_NODE` branches; in the browser it loads its `.wasm` from the URL the preview worker passes and draws randomness from `crypto.getRandomValues`. Vite still externalized both and printed two warnings on every build. `vite.config.ts` now drops exactly those two warnings for that one package through `build.rolldownOptions.onLog`, so an externalization anywhere else, or of any other module, still shows.
 
 ### Fixed
+- **Maintenance: a printer's time offline was counted as print time** — A printer that dropped off mid-print and came back printing hours later had the whole gap added to its runtime, so hours-based maintenance came due early. Time while Bambuddy itself was restarting is still counted, as before.
+- **Maintenance: duplicate maintenance items** — A printer's items are created when its maintenance is first shown, and two pages (or a page and the background check) could each create the same one; marking it performed could then update the copy not shown, and assigning that type again failed. Existing duplicates are merged on upgrade (the most recently performed one is kept, or the one the page showed when none was, with the history of all), and they can no longer be created.
+- **Maintenance API: "performed" answered in hours for an item set to days** — Its response used the maintenance type's interval type instead of the item's own setting, which the list already used.
 - **Notifications: the daily digest could be lost** — When the digest could not be sent (a network error, say), its events were deleted anyway. A digest time inside the provider's quiet hours was skipped, logged as sent, and its events deleted too. The digest now goes out at its time even during quiet hours, and a failed send keeps the events for the next day's digest (events more than two days old are dropped, so a provider that keeps failing does not pile them up). Lines from an earlier day show their date.
 - **Notifications: a later print could be attributed to the user of an earlier queued one** — When a queued print's archive was deleted while it was being sent to the printer, the record of who queued it was left behind, and an unrelated later print of the same name on that printer named and emailed that user.
 - **Notifications: "Print Stopped" after a printer came back online showed the time it was offline as the print's duration** — A print that ended while the printer was offline is closed when it reconnects; its real end time is unknown, so the notification now shows the sliced estimate instead of, say, 14 hours.

backend/app/api/routes/maintenance.py

@@ -5,6 +5,7 @@ from datetime import datetime, timezone
 
 from fastapi import APIRouter, Depends, HTTPException
 from sqlalchemy import select
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import selectinload
 
@@ -299,6 +300,36 @@ def maintenance_alert_values(item: MaintenanceStatus) -> tuple[float, float, str
     return (item.hours_since_maintenance, item.interval_hours, "hours")
 
 
+async def _get_or_create_maintenance_item(db: AsyncSession, printer_id: int, type_id: int) -> PrinterMaintenance:
+    """The printer's row for this type, created with defaults if missing.
+
+    Insert first, in a savepoint: two reads of a new printer (two dashboards,
+    or a dashboard and the background check) each used to create the row. The
+    unique index now refuses the second, which takes the first's (FO-C14).
+    """
+    item = PrinterMaintenance(
+        printer_id=printer_id,
+        maintenance_type_id=type_id,
+        enabled=True,
+        last_performed_hours=0.0,
+    )
+    try:
+        async with db.begin_nested():
+            db.add(item)
+            await db.flush()
+        return item
+    except IntegrityError:
+        existing = await db.scalar(
+            select(PrinterMaintenance).where(
+                PrinterMaintenance.printer_id == printer_id,
+                PrinterMaintenance.maintenance_type_id == type_id,
+            )
+        )
+        if existing is None:
+            raise
+        return existing
+
+
 async def _get_printer_maintenance_internal(
     printer_id: int,
     db: AsyncSession,
@@ -356,21 +387,15 @@ async def _get_printer_maintenance_internal(
             if not maint_type.is_system:
                 continue
 
-            # Create default entry for this printer/type
-            item = PrinterMaintenance(
-                printer_id=printer_id,
-                maintenance_type_id=maint_type.id,
-                enabled=True,
-                last_performed_hours=0.0,
-            )
-            db.add(item)
-            await db.flush()
+            # Create default entry for this printer/type, or take the one a
+            # concurrent read just created.
+            item = await _get_or_create_maintenance_item(db, printer_id, maint_type.id)
 
-            interval = maint_type.default_interval_hours
-            interval_type = default_interval_type
-            enabled = True
-            last_performed_hours = 0.0
-            last_performed_at = None
+            interval = item.custom_interval_hours or maint_type.default_interval_hours
+            interval_type = getattr(item, "custom_interval_type", None) or default_interval_type
+            enabled = item.enabled
+            last_performed_hours = item.last_performed_hours
+            last_performed_at = item.last_performed_at
             item_id = item.id
 
         # Calculate status based on interval type
@@ -552,7 +577,13 @@ async def assign_maintenance_type(
         last_performed_hours=0.0,
     )
     db.add(item)
-    await db.commit()
+    try:
+        await db.commit()
+    except IntegrityError:
+        # A concurrent assign (a double click) passed the check above too; the
+        # unique index refuses this one (FO-C14).
+        await db.rollback()
+        raise HTTPException(status_code=400, detail="Maintenance type already assigned to this printer") from None
 
     # Re-fetch with relationship loaded for response serialization
     from sqlalchemy.orm import selectinload
@@ -647,7 +678,8 @@ async def perform_maintenance(
 
     # Calculate status
     interval = item.custom_interval_hours or item.maintenance_type.default_interval_hours
-    interval_type = getattr(item.maintenance_type, "interval_type", "hours") or "hours"
+    # The item's own interval type first, as the status list reads it (dbcol-09).
+    interval_type = item.custom_interval_type or getattr(item.maintenance_type, "interval_type", "hours") or "hours"
     hours_since = current_hours - item.last_performed_hours
     hours_until = interval - hours_since
 

backend/app/core/database.py

@@ -5682,6 +5682,7 @@ async def _run_migrations(conn):
     # IntegrityError the route maps back to the same 400. create_all() adds it
     # on fresh installs only — this covers databases whose table predates it.
     await _migrate_location_ha_sensor_unique_binding(conn)
+    await _migrate_printer_maintenance_unique(conn)
 
     # Migration: the stock alert templates name the colour and subtype (#2955).
     # The forecast groups by colour, so two colours of one product would
@@ -6901,6 +6902,59 @@ async def _migrate_location_ha_sensor_unique_binding(conn) -> None:
     )
 
 
+async def _migrate_printer_maintenance_unique(conn) -> None:
+    """Unique index on printer_maintenance (printer_id, maintenance_type_id) (FO-C14).
+
+    A printer's maintenance rows are created on first read with no constraint,
+    so two concurrent reads could each create one. Same name and shape as the
+    Index in the model, so fresh and upgraded databases end up identical.
+
+    Duplicates are merged first, since CREATE UNIQUE INDEX refuses to build
+    over them: the most recently performed row is kept (the highest id, the
+    one the page showed, when none was performed), and the history of the
+    others moves onto it.
+    """
+    from sqlalchemy import text
+
+    async with conn.begin_nested():
+        rows = (
+            await conn.execute(
+                text(
+                    "SELECT id, printer_id, maintenance_type_id, last_performed_at FROM printer_maintenance "
+                    "WHERE (printer_id, maintenance_type_id) IN ("
+                    "SELECT printer_id, maintenance_type_id FROM printer_maintenance "
+                    "GROUP BY printer_id, maintenance_type_id HAVING COUNT(*) > 1)"
+                )
+            )
+        ).all()
+        groups: dict[tuple[int, int], list] = {}
+        for row in rows:
+            groups.setdefault((row.printer_id, row.maintenance_type_id), []).append(row)
+        for group in groups.values():
+            # Latest performance first; never-performed rows last, oldest first.
+            performed = sorted(
+                (r for r in group if r.last_performed_at is not None), key=lambda r: str(r.last_performed_at)
+            )
+            # Never performed: the highest id, the row the page showed (it
+            # kept the last one read) and so the one holding the user's settings.
+            keep = performed[-1] if performed else max(group, key=lambda r: r.id)
+            for row in group:
+                if row.id == keep.id:
+                    continue
+                await conn.execute(
+                    text(
+                        "UPDATE maintenance_history SET printer_maintenance_id = :keep WHERE printer_maintenance_id = :gone"
+                    ),
+                    {"keep": keep.id, "gone": row.id},
+                )
+                await conn.execute(text("DELETE FROM printer_maintenance WHERE id = :gone"), {"gone": row.id})
+    await _safe_execute(
+        conn,
+        "CREATE UNIQUE INDEX IF NOT EXISTS uq_printer_maintenance_printer_type "
+        "ON printer_maintenance (printer_id, maintenance_type_id)",
+    )
+
+
 async def _migrate_normalize_printer_locations(conn) -> None:
     """Trim printer locations and store a blank one as NULL (#2962).
 

backend/app/main.py

@@ -1196,6 +1196,7 @@ async def forget_printer(printer_id: int) -> None:
         _kill_switch_stopped,
         _printer_offline_published,
         _nozzle_count_updated,
+        _runtime_seen_connected,
     ):
         marked.discard(printer_id)
 
@@ -10157,6 +10158,16 @@ def stop_printer_sensor_history_recording():
 _runtime_tracking_task: asyncio.Task | None = None
 RUNTIME_TRACKING_INTERVAL = 30  # Update every 30 seconds
 
+# Printers the runtime tracker has seen connected since startup. One of them
+# going offline clears its timestamp, so the time it was away is not counted
+# when it comes back printing (DB-C5). A printer not yet seen -- still
+# connecting after a restart -- keeps it for a grace period: a print that ran
+# on through a Bambuddy restart is counted from the stored timestamp, as before.
+_runtime_seen_connected: set[int] = set()
+# A printer never seen connected this long after the tracker started is
+# offline, not still connecting: its stored timestamp predates a gap too.
+RUNTIME_STARTUP_GRACE_SECONDS = 120
+
 
 async def track_printer_runtime():
     """Background task to track printer active runtime (RUNNING state only).
@@ -10170,6 +10181,7 @@ async def track_printer_runtime():
 
     # Wait for MQTT connections to establish on startup
     await asyncio.sleep(15)
+    started = time.monotonic()
 
     while True:
         try:
@@ -10197,7 +10209,23 @@ async def track_printer_runtime():
                     continue
                 if not state.connected:
                     logger.debug("[%s] Runtime tracking: not connected", pname)
+                    offline = pid in _runtime_seen_connected or (
+                        time.monotonic() - started > RUNTIME_STARTUP_GRACE_SECONDS
+                    )
+                    if offline and last_update is not None:
+                        try:
+                            async with async_session() as db:
+                                printer = await db.get(Printer, pid)
+                                if printer:
+                                    printer.last_runtime_update = None
+                                    await db.commit()
+                        except Exception as e:
+                            # Still marked: the next offline tick tries again.
+                            logger.warning("[%s] Runtime tracking commit failed: %s", pname, e)
+                            continue
+                    _runtime_seen_connected.discard(pid)
                     continue
+                _runtime_seen_connected.add(pid)
 
                 needs_commit = False
                 new_runtime = runtime_secs

backend/app/models/maintenance.py

@@ -2,7 +2,7 @@
 
 from datetime import datetime
 
-from sqlalchemy import Boolean, DateTime, Float, ForeignKey, String, Text, func
+from sqlalchemy import Boolean, DateTime, Float, ForeignKey, Index, String, Text, func
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from backend.app.core.database import Base
@@ -35,6 +35,8 @@ class PrinterMaintenance(Base):
     """Tracks maintenance status for a specific printer."""
 
     __tablename__ = "printer_maintenance"
+    # One row per printer and type: they are created on first read (FO-C14).
+    __table_args__ = (Index("uq_printer_maintenance_printer_type", "printer_id", "maintenance_type_id", unique=True),)
 
     id: Mapped[int] = mapped_column(primary_key=True)
     printer_id: Mapped[int] = mapped_column(ForeignKey("printers.id", ondelete="CASCADE"))

backend/tests/integration/test_maintenance_api.py

@@ -196,6 +196,25 @@ class TestMaintenanceItemsAPI:
         assert response.status_code == 200
         assert response.json()["enabled"] is False
 
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_perform_reports_the_item_s_own_interval_type(self, async_client: AsyncClient, printer_factory):
+        """dbcol-09: an item set to days on an hours-based type was answered in
+        hours by "performed", while the list honoured the item's setting."""
+        printer = await printer_factory(name="Interval Type Printer")
+        overview = (await async_client.get(f"/api/v1/maintenance/printers/{printer.id}")).json()
+        item = next(i for i in overview["maintenance_items"] if i["interval_type"] == "hours")
+        patched = await async_client.patch(
+            f"/api/v1/maintenance/items/{item['id']}",
+            json={"custom_interval_type": "days", "custom_interval_hours": 30},
+        )
+        assert patched.status_code == 200
+
+        data = (await async_client.post(f"/api/v1/maintenance/items/{item['id']}/perform", json={})).json()
+
+        assert data["interval_type"] == "days"
+        assert data["days_until_due"] == 30
+
     @pytest.mark.asyncio
     @pytest.mark.integration
     async def test_perform_maintenance(self, async_client: AsyncClient, maintenance_item):

backend/tests/integration/test_maintenance_item_get_or_create.py

@@ -0,0 +1,77 @@
+"""A maintenance row created on first read is taken, not duplicated (FO-C14).
+
+Two reads of a new printer's maintenance -- two dashboards, or a dashboard
+and the background check -- could each create the same row. With the unique
+index the second insert is refused; the read then uses the row the other one
+created, and its session carries on.
+"""
+
+import pytest
+from sqlalchemy import func, select
+
+from backend.app.api.routes.maintenance import _get_or_create_maintenance_item, ensure_default_types
+from backend.app.models.maintenance import MaintenanceType, PrinterMaintenance
+
+pytestmark = [pytest.mark.asyncio, pytest.mark.integration]
+
+
+async def test_the_row_another_read_created_is_used(db_session, printer_factory):
+    printer = await printer_factory()
+    await ensure_default_types(db_session)
+    type_id = (await db_session.scalars(select(MaintenanceType.id).limit(1))).first()
+    theirs = PrinterMaintenance(
+        printer_id=printer.id, maintenance_type_id=type_id, enabled=True, last_performed_hours=3.0
+    )
+    db_session.add(theirs)
+    await db_session.commit()
+
+    item = await _get_or_create_maintenance_item(db_session, printer.id, type_id)
+
+    assert item.id == theirs.id
+    count = await db_session.scalar(
+        select(func.count()).select_from(PrinterMaintenance).where(PrinterMaintenance.printer_id == printer.id)
+    )
+    assert count == 1
+    # The refused insert was rolled back to its savepoint only.
+    await db_session.commit()
+
+
+async def test_a_missing_row_is_created(db_session, printer_factory):
+    printer = await printer_factory()
+    await ensure_default_types(db_session)
+    type_id = (await db_session.scalars(select(MaintenanceType.id).limit(1))).first()
+
+    item = await _get_or_create_maintenance_item(db_session, printer.id, type_id)
+
+    assert item.id is not None
+    assert (item.enabled, item.last_performed_hours) == (True, 0.0)
+
+
+async def test_a_second_assign_racing_the_first_is_refused_as_assigned(db_session, printer_factory):
+    """Both pass the "already assigned?" check (a double click); the index now
+    refuses the second insert, which answers the route's own 400, not a 500."""
+    from unittest.mock import MagicMock, patch
+
+    from fastapi import HTTPException
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+    from backend.app.api.routes.maintenance import assign_maintenance_type
+
+    printer = await printer_factory()
+    await ensure_default_types(db_session)
+    type_id = (await db_session.scalars(select(MaintenanceType.id).limit(1))).first()
+    db_session.add(PrinterMaintenance(printer_id=printer.id, maintenance_type_id=type_id, enabled=True))
+    await db_session.commit()
+
+    real_execute = AsyncSession.execute
+
+    async def check_sees_nothing(self, statement, *args, **kwargs):
+        sql = str(statement)
+        if "FROM printer_maintenance" in sql and "maintenance_type_id" in sql:
+            return MagicMock(scalar_one_or_none=MagicMock(return_value=None))
+        return await real_execute(self, statement, *args, **kwargs)
+
+    with patch.object(AsyncSession, "execute", check_sees_nothing), pytest.raises(HTTPException) as refused:
+        await assign_maintenance_type(printer.id, type_id, db_session, None)
+
+    assert refused.value.status_code == 400

backend/tests/integration/test_runtime_disconnect_gap.py

@@ -0,0 +1,120 @@
+"""A printer's time offline is not counted as runtime (DB-C5).
+
+The runtime counter adds the time since its last tick while a printer is
+RUNNING. When a printer dropped off mid-print and came back hours later with a
+print running, that first tick added the whole gap: hours-based maintenance
+came due early. The timestamp is now cleared when a printer this process saw
+connected goes offline. A Bambuddy restart is still counted from the stored
+timestamp, as before -- those printers have not been seen connected yet.
+"""
+
+import asyncio
+from datetime import datetime, timedelta
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import pytest
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from backend.app import main as main_module
+from backend.app.models.printer import Printer
+
+pytestmark = [pytest.mark.asyncio, pytest.mark.integration]
+
+
+@pytest.fixture(autouse=True)
+def _forget_seen():
+    main_module._runtime_seen_connected.clear()
+    yield
+    main_module._runtime_seen_connected.clear()
+
+
+async def _ticks(test_engine, states: list) -> None:
+    """Run the tracker for one tick per state, then stop it."""
+    maker = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
+    feed = iter(states)
+    current = {"state": None}
+    sleeps = {"n": 0}
+
+    async def fake_sleep(_seconds):
+        sleeps["n"] += 1
+        if sleeps["n"] > len(states):
+            raise asyncio.CancelledError
+        current["state"] = next(feed)
+
+    with (
+        patch.object(main_module, "async_session", maker),
+        patch.object(main_module.asyncio, "sleep", fake_sleep),
+        patch.object(main_module.printer_manager, "get_status", lambda _pid: current["state"]),
+    ):
+        try:
+            await main_module.track_printer_runtime()
+        except asyncio.CancelledError:
+            pass
+
+
+async def _printer(db_session, printer_factory, last_update):
+    printer = await printer_factory()
+    printer.runtime_seconds = 1000
+    printer.last_runtime_update = last_update
+    await db_session.commit()
+    return printer
+
+
+async def test_a_disconnect_clears_the_timestamp(test_engine, db_session, printer_factory):
+    printer = await _printer(db_session, printer_factory, datetime.utcnow() - timedelta(seconds=60))
+    running = SimpleNamespace(connected=True, state="RUNNING")
+    offline = SimpleNamespace(connected=False, state="RUNNING")
+
+    await _ticks(test_engine, [running, offline])
+
+    await db_session.refresh(printer)
+    assert printer.last_runtime_update is None
+
+
+async def test_a_restart_still_counts_the_stored_timestamp(test_engine, db_session, printer_factory):
+    stored = datetime.utcnow() - timedelta(minutes=30)
+    printer = await _printer(db_session, printer_factory, stored)
+    # Still connecting after the restart, then printing.
+    connecting = SimpleNamespace(connected=False, state="unknown")
+    running = SimpleNamespace(connected=True, state="RUNNING")
+
+    await _ticks(test_engine, [connecting, running])
+
+    await db_session.refresh(printer)
+    assert printer.runtime_seconds >= 1000 + 29 * 60
+
+
+async def test_a_printer_still_offline_after_the_startup_grace_is_not_credited(
+    test_engine, db_session, printer_factory, monkeypatch
+):
+    """It dropped off before the restart and never came back during it: the
+    stored timestamp is from before the gap, not from a print that ran on."""
+    monkeypatch.setattr(main_module, "RUNTIME_STARTUP_GRACE_SECONDS", 0)
+    printer = await _printer(db_session, printer_factory, datetime.utcnow() - timedelta(hours=10))
+    offline = SimpleNamespace(connected=False, state="unknown")
+
+    await _ticks(test_engine, [offline])
+
+    await db_session.refresh(printer)
+    assert printer.last_runtime_update is None
+
+
+async def test_a_clear_that_failed_is_tried_again(test_engine, db_session, printer_factory):
+    printer = await _printer(db_session, printer_factory, datetime.utcnow() - timedelta(seconds=60))
+    running = SimpleNamespace(connected=True, state="RUNNING")
+    offline = SimpleNamespace(connected=False, state="RUNNING")
+    real_get = AsyncSession.get
+    calls = {"n": 0}
+
+    async def flaky_get(self, *args, **kwargs):
+        calls["n"] += 1
+        if calls["n"] == 1:
+            raise RuntimeError("database is locked")
+        return await real_get(self, *args, **kwargs)
+
+    with patch.object(AsyncSession, "get", flaky_get):
+        await _ticks(test_engine, [running, offline, offline])
+
+    await db_session.refresh(printer)
+    assert printer.last_runtime_update is None

backend/tests/unit/test_forget_printer.py

@@ -85,6 +85,7 @@ def _entries(printer_id: int) -> dict[str, tuple]:
         "_unknown_tag_last_broadcast": (printer_id, {(0, 0): ("uid", "x")}),
         "_last_status_broadcast": (printer_id, "hash"),
         "_nozzle_count_updated": (printer_id, None),
+        "_runtime_seen_connected": (printer_id, None),
         "_fallback_3mf_retry_tasks": (printer_id, _sleeper()),
         "_connection_watchdog_last_attempt": (printer_id, 1.0),
         "_ams_index_only_logged": ((printer_id, 0), None),

backend/tests/unit/test_printer_maintenance_unique_migration.py

@@ -0,0 +1,117 @@
+"""Duplicate maintenance rows are merged, and a unique index keeps them out (FO-C14).
+
+A printer's maintenance rows are created on first read, with no constraint:
+two dashboards, or a dashboard and the background check, could each create
+one. "Performed" could then update the copy that is not shown, and assigning
+the type again answered 500. The migration merges existing duplicates --
+keeping the most recently performed row and moving the others' history onto
+it -- and adds the index, which CREATE UNIQUE INDEX would refuse to build over
+duplicates.
+"""
+
+from __future__ import annotations
+
+import pytest
+from sqlalchemy import text
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.ext.asyncio import create_async_engine
+
+from backend.app.core.database import _migrate_printer_maintenance_unique
+
+pytestmark = pytest.mark.asyncio
+
+
+@pytest.fixture
+async def engine():
+    """In-memory SQLite with the two tables as they were before the index."""
+    engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
+    async with engine.begin() as conn:
+        await conn.execute(
+            text(
+                "CREATE TABLE printer_maintenance ("
+                "id INTEGER PRIMARY KEY AUTOINCREMENT, printer_id INTEGER NOT NULL, "
+                "maintenance_type_id INTEGER NOT NULL, last_performed_at DATETIME, "
+                "last_performed_hours FLOAT DEFAULT 0)"
+            )
+        )
+        await conn.execute(
+            text(
+                "CREATE TABLE maintenance_history ("
+                "id INTEGER PRIMARY KEY AUTOINCREMENT, printer_maintenance_id INTEGER NOT NULL)"
+            )
+        )
+    try:
+        yield engine
+    finally:
+        await engine.dispose()
+
+
+async def _item(conn, printer_id: int, type_id: int, performed_at: str | None = None) -> int:
+    result = await conn.execute(
+        text(
+            "INSERT INTO printer_maintenance (printer_id, maintenance_type_id, last_performed_at) "
+            "VALUES (:p, :t, :at) RETURNING id"
+        ),
+        {"p": printer_id, "t": type_id, "at": performed_at},
+    )
+    return result.scalar_one()
+
+
+async def _history(conn, item_id: int) -> None:
+    await conn.execute(text("INSERT INTO maintenance_history (printer_maintenance_id) VALUES (:i)"), {"i": item_id})
+
+
+async def test_duplicates_merge_into_the_most_recently_performed_row(engine):
+    async with engine.begin() as conn:
+        never = await _item(conn, 1, 5)
+        performed = await _item(conn, 1, 5, "2026-10-01 10:00:00")
+        older = await _item(conn, 1, 5, "2026-09-01 10:00:00")
+        other_type = await _item(conn, 1, 6)
+        for item_id in (never, performed, older):
+            await _history(conn, item_id)
+
+    async with engine.begin() as conn:
+        await _migrate_printer_maintenance_unique(conn)
+
+    async with engine.begin() as conn:
+        ids = (await conn.execute(text("SELECT id FROM printer_maintenance ORDER BY id"))).scalars().all()
+        history = (await conn.execute(text("SELECT printer_maintenance_id FROM maintenance_history"))).scalars().all()
+    assert ids == [performed, other_type]
+    assert history == [performed, performed, performed]
+
+
+async def test_without_a_performance_the_row_the_page_showed_is_kept(engine):
+    """The page took the last row read -- the highest id -- so a user's
+    settings went there; keeping another would bring the defaults back."""
+    async with engine.begin() as conn:
+        await _item(conn, 1, 5)
+        shown = await _item(conn, 1, 5)
+
+    async with engine.begin() as conn:
+        await _migrate_printer_maintenance_unique(conn)
+
+    async with engine.begin() as conn:
+        ids = (await conn.execute(text("SELECT id FROM printer_maintenance"))).scalars().all()
+    assert ids == [shown]
+
+
+async def test_the_index_then_refuses_a_second_row(engine):
+    async with engine.begin() as conn:
+        await _migrate_printer_maintenance_unique(conn)
+        await _item(conn, 1, 5)
+
+    with pytest.raises(IntegrityError):
+        async with engine.begin() as conn:
+            await _item(conn, 1, 5)
+
+
+async def test_is_idempotent(engine):
+    async with engine.begin() as conn:
+        await _item(conn, 1, 5)
+        await _item(conn, 2, 5)
+    for _ in range(2):
+        async with engine.begin() as conn:
+            await _migrate_printer_maintenance_unique(conn)
+
+    async with engine.begin() as conn:
+        assert (await conn.execute(text("SELECT COUNT(*) FROM printer_maintenance"))).scalar_one() == 2