feat(jobs): Phase 2 Job Persistence (DB-backed + Restart Recovery + Cleanup)

[strausmann]

Summary

Hub Phase 2 — Jobs werden jetzt in der jobs-Tabelle persistiert statt nur im In-Memory-Zustand gehalten. Der Neustart des Services erholt sich automatisch: PRINTING-Jobs werden als FAILED_RESTART markiert, QUEUED-Jobs werden re-enqueued. Ein CleanupTask entfernt terminal Jobs nach konfigurierbarer Retention-Zeit.

Contributor License Agreement (CLA)

By opening this pull request you affirm that you have read and agree to the
project's Contributor License Agreement for the contribution(s)
included here.

• I have read and agree to the Contributor License Agreement.

Linked issue

Closes #93

Type of change

• Bug fix (non-breaking)
• New feature (non-breaking)
• Breaking change (feat! / fix!) — describe migration path below
• New printer model plugin
• Documentation only
• Refactor / chore (no behaviour change)
• Test only
• CI / release tooling

Änderungen im Überblick

• JobStore Protocol mit MemoryJobStore (In-Memory, default) und SQLiteJobStore (DB-backed)
• PrintQueue ruft store.mark_* bei jeder State-Transition synchron auf
• PrintService legt erst die DB-Row an, dann queue.submit (atomisch mit Rollback bei queue-Fehler)
• PrintQueue.start() Recovery: PRINTING → FAILED_RESTART, QUEUED → re-enqueued via Rerender
• CleanupTask räumt terminal Jobs älter als PRINTER_HUB_JOB_RETENTION_DAYS (Default 30 Tage)
• GET /api/batches/{id} Snapshot-Endpoint mit summary.all_terminal
• recover_inflight_jobs() aus Lifespan entfernt — Phase 2 ersetzt die Funktion mit korrekter QUEUED/PRINTING-Differenzierung

Hardware tested on

• Brother PT-Series (model: ___)
• Brother QL-Series (model: ___)
• Other: ___
• No hardware impact

Test coverage

• Added/updated unit tests
• Added/updated integration tests
• Hardware tests added (if applicable)
• Existing tests still pass

~50 neue Tests (Unit + Integration), 870 Tests grün, 5 expected Skips.

Checklist

• PR title follows Conventional Commits (feat(...): ... etc.)
• All commits are signed off (if applicable)
• CHANGELOG entry isn't needed (semantic-release generates it)
• Documentation updated (if behaviour, API, or config changed)
• No private IPs, hostnames, domains, real tokens, or PII in commits
• CI is green (will be checked again after CI runs)

Spec & Plan

• Spec: docs/superpowers/specs/2026-05-31-phase-2-job-persistence-design.md
• Plan: docs/superpowers/plans/2026-05-31-phase-2-job-persistence.md

Beide gemerged auf den Branch mit Reviewer-Patches aus 2 unabhängigen Spec+Plan-Reviews.

Migration / breaking change notes

Keine Breaking Changes. Neues optionales Env-Var PRINTER_HUB_JOB_RETENTION_DAYS (Default: 30). Bestehende Deployments ohne diese Variable verhalten sich wie bisher (In-Memory-Store bleibt Default bis DATABASE_URL gesetzt ist).

Screenshots / output

870 passed, 5 skipped in 143.18s
mypy --strict app: Success: no issues found in 107 source files
ruff check app tests: All checks passed!

Entblockt strausmann/hangar#81 (Result-Page Live-Updates).

[gemini-code-assist]

Summary of Changes

Hello, I'm Gemini Code Assist¹! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed!

This pull request introduces robust job persistence to the printing service, transitioning from purely in-memory job tracking to a database-backed system. By integrating a JobStore protocol, the service now ensures that job states are durable, supports automatic recovery after service restarts, and provides a new API endpoint for efficient batch status monitoring. A background cleanup task has also been added to manage database growth by pruning old terminal jobs.

Highlights

• Job Persistence: Implemented a new JobStore protocol with a SQLiteJobStore implementation to persist job state transitions in the database, ensuring durability across service restarts.
• Restart Recovery: Added automatic recovery in PrintQueue.start(): PRINTING jobs are marked as FAILED_RESTART and QUEUED jobs are re-enqueued using a rerender process.
• Cleanup Task: Introduced a background CleanupTask that periodically removes terminal jobs (DONE, FAILED, etc.) older than a configurable retention period (default 30 days).
• New API Endpoint: Added GET /api/batches/{id} to provide a snapshot of batch status and job states, including an all_terminal summary for Hangar's frontend to optimize SSE connections.

Using Gemini Code Assist

The full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips.

Invoking Gemini

You can request assistance from Gemini at any point by creating a comment using either /gemini <command> or @gemini-code-assist <command>. Below is a summary of the supported commands on the current page.

Feature	Command	Description
Code Review	/gemini review	Performs a code review for the current pull request in its current state.
Pull Request Summary	/gemini summary	Provides a summary of the current pull request in its current state.
Comment	@gemini-code-assist	Responds in comments when explicitly tagged, both in pull request comments and review comments.
Help	/gemini help	Displays a list of available commands.

Customization

To customize the Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a .gemini/ folder in the base of the repository. Detailed instructions can be found here.

Limitations & Feedback

Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counterproductive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for GitHub and other Google products, sign up here.

Footnotes

1. Review the Privacy Notices, Generative AI Prohibited Use Policy, Terms of Service, and learn how to configure Gemini Code Assist in GitHub here. Gemini can make mistakes, so double check it and use code with caution. ↩

[codecov]

Codecov Report

❌ Patch coverage is 88.98551% with 38 lines in your changes missing coverage. Please review.
✅ Project coverage is 89.92%. Comparing base (280f46d) to head (77e4450).
✅ All tests successful. No failed tests found.

Files with missing lines	Patch %	Lines
backend/app/services/print_queue.py	81.81%	8 Missing and 4 partials ⚠️
backend/app/api/routes/batches.py	77.77%	6 Missing ⚠️
backend/app/services/cleanup_task.py	86.04%	4 Missing and 2 partials ⚠️
backend/app/services/job_store_sqlite.py	89.47%	3 Missing and 3 partials ⚠️
backend/app/services/job_store.py	93.44%	3 Missing and 1 partial ⚠️
backend/app/main.py	89.47%	2 Missing ⚠️
backend/app/repositories/jobs.py	92.00%	1 Missing and 1 partial ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main      #94      +/-   ##
==========================================
- Coverage   90.20%   89.92%   -0.29%     
==========================================
  Files          84       89       +5     
  Lines        3684     4008     +324     
  Branches      314      343      +29     
==========================================
+ Hits         3323     3604     +281     
- Misses        279      315      +36     
- Partials       82       89       +7     

Components	Coverage Δ
Printer Backends (transport)	87.50% <ø> (ø)
Printer Models (drivers)	91.42% <ø> (ø)
Services	91.38% <88.61%> (-0.65%)	⬇️
REST API	85.08% <78.57%> (-0.39%)	⬇️
Pydantic Schemas	100.00% <100.00%> (ø)
Integration Plugins	100.00% <ø> (ø)

Files with missing lines	Coverage Δ
backend/app/api/routes/print.py	97.50% <100.00%> (-1.25%)	⬇️
backend/app/config.py	100.00% <100.00%> (ø)
backend/app/schemas/batch_read.py	100.00% <100.00%> (ø)
backend/app/services/print_service.py	100.00% <100.00%> (ø)
backend/app/main.py	85.19% <89.47%> (-1.40%)	⬇️
backend/app/repositories/jobs.py	89.52% <92.00%> (-0.96%)	⬇️
backend/app/services/job_store.py	93.44% <93.44%> (ø)
backend/app/api/routes/batches.py	77.77% <77.77%> (ø)
backend/app/services/cleanup_task.py	86.04% <86.04%> (ø)
backend/app/services/job_store_sqlite.py	89.47% <89.47%> (ø)
... and 1 more

... and 1 file with indirect coverage changes

Flag	Coverage Δ
backend	89.92% <88.98%> (-0.29%)	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

---

Continue to review full report in Codecov by Sentry.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 280f46d...77e4450. Read the comment docs.

🚀 New features to boost your workflow:

• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[gemini-code-assist]

Code Review

This pull request implements Phase 2 of job persistence for the label printer hub, introducing a JobStore protocol with SQLite and in-memory implementations to persist print job state transitions. It refactors PrintQueue and PrintService to save jobs before submission, adds a restart-recovery mechanism to re-enqueue queued jobs upon startup, introduces a periodic CleanupTask to evict old terminal jobs, and adds a new GET /api/batches/{batch_id} snapshot endpoint. The reviewer's feedback focuses on improving robustness and efficiency, suggesting that database state transitions in the worker loop be wrapped in try-except blocks to prevent queue blockage, that potential TypeError and AttributeError exceptions be handled defensively during recovery, and that redundant database queries in SQLiteJobStore be eliminated. Additionally, the reviewer recommends explicitly awaiting canceled tasks in CleanupTask to avoid resource leaks and wrapping rollback database calls in PrintService to prevent exception masking.

[gemini-code-assist]

Wenn self._store.mark_printing eine Exception auslöst (z. B. aufgrund einer temporären Datenbanksperre bei SQLite), wird diese außerhalb des try-Blocks nicht abgefangen. Dies führt dazu, dass der gesamte Worker-Task (_worker) unerwartet beendet wird und die Druckerwarteschlange dauerhaft blockiert. Es wird empfohlen, diesen Aufruf in den try-Block zu verschieben, damit Fehler abgefangen und protokolliert werden können, ohne den Worker-Loop zu beenden.

Suggested change

	# Phase 2: DB-State QUEUED->PRINTING persistieren (bridge: dataclass.id ist str)
	await self._store.mark_printing(UUID(job.id))
	try:
	_from = job.state
	JobStateMachine.transition(job, JobState.PRINTING)
	try:
	# Phase 2: DB-State QUEUED->PRINTING persistieren (bridge: dataclass.id ist str)
	await self._store.mark_printing(UUID(job.id))
	_from = job.state
	JobStateMachine.transition(job, JobState.PRINTING)

[gemini-code-assist]

Falls db_job.payload in der Datenbank beschädigt oder None ist, kann _rerender_from_db_job eine TypeError auslösen (da None nicht indizierbar ist). Da TypeError nicht in der Liste der abgefangenen Ausnahmen enthalten ist, würde dies den gesamten Recovery-Prozess beim Start abbrechen. Es wird empfohlen, TypeError ebenfalls abzufangen.

Suggested change

	try:
	image = await self._rerender_from_db_job(db_job)
	except (KeyError, ValidationError, TemplateNotFoundError) as exc:
	logger.warning(
	"Recovery: Job %s rerender fehlgeschlagen (%s), FAILED",
	db_job.id,
	exc.__class__.__name__,
	)
	await self._store.mark_failed(db_job.id, f"recovery_rerender_failed: {exc}")
	continue
	try:
	image = await self._rerender_from_db_job(db_job)
	except (KeyError, ValidationError, TemplateNotFoundError, TypeError) as exc:
	logger.warning(
	"Recovery: Job %s rerender fehlgeschlagen (%s), FAILED",
	db_job.id,
	exc.__class__.__name__,
	)
	await self._store.mark_failed(db_job.id, f"recovery_rerender_failed: {exc}")
	continue

[gemini-code-assist]

Falls db_job.payload in der Datenbank aus irgendeinem Grund None ist (z. B. bei fehlerhaften oder unvollständigen Altdaten), führt der direkte Aufruf von db_job.payload.get(...) außerhalb des try-Blocks zu einem AttributeError und bringt den gesamten Recovery-Prozess zum Absturz. Es wird empfohlen, den Payload defensiv zu behandeln.

Suggested change

	wrapper = Job(
	id=str(db_job.id),
	printer_id=db_job.printer_id,
	image_payload=payload_bytes,
	tape_mm=db_job.payload.get("tape_mm"),
	options=db_job.payload.get("options", {}),
	)
	payload = db_job.payload or {}
	wrapper = Job(
	id=str(db_job.id),
	printer_id=db_job.printer_id,
	image_payload=payload_bytes,
	tape_mm=payload.get("tape_mm"),
	options=payload.get("options", {}),
	)

[gemini-code-assist]

In SQLiteJobStore.mark_printing wird zuerst jobs_repo.get aufgerufen, um den Job-Status zu prüfen, und anschließend jobs_repo.mark_printing, welches intern den Job erneut aus der Datenbank lädt. Dies führt zu zwei redundanten Datenbankabfragen pro Zustandsübergang. Da jobs_repo.mark_printing bei Fehlern (Job nicht gefunden oder ungültiger Zustand) eine ValueError auslöst, kann dies effizienter mit einem einzigen try-except-Block gelöst werden. Das gleiche Muster lässt sich auch auf mark_done und mark_failed anwenden.

    async def mark_printing(self, job_id: UUID) -> None:
        """Transition QUEUED -> PRINTING. Silently no-op if job not found."""
        async with self._session_factory() as session:
            try:
                await jobs_repo.mark_printing(session, job_id)
            except ValueError as exc:
                if "not found" in str(exc):
                    return
                logger.warning(str(exc))

[gemini-code-assist]

Analog zu mark_printing führt auch mark_done zwei redundante Datenbankabfragen durch. Durch die direkte Nutzung von jobs_repo.mark_done innerhalb eines try-except-Blocks lässt sich die Datenbanklast halbieren.

    async def mark_done(self, job_id: UUID) -> None:
        """Transition PRINTING -> DONE.

        Delegiert an jobs_repo.mark_done mit result={} — JobStore-Protocol
        speichert kein structured result (Phase-2 YAGNI; ergänzbar via
        `set_result` Methode wenn Hangar das später braucht).
        """
        async with self._session_factory() as session:
            try:
                await jobs_repo.mark_done(session, job_id, result={})
            except ValueError as exc:
                if "not found" in str(exc):
                    return
                logger.warning(str(exc))

[gemini-code-assist]

Auch in mark_failed lässt sich die doppelte Datenbankabfrage durch direktes Aufrufen von jobs_repo.mark_failed mit entsprechender Ausnahmebehandlung vermeiden.

Suggested change

	async def mark_failed(self, job_id: UUID, error: str) -> None:
	"""Transition any non-terminal -> FAILED. Silently no-op if job not found."""
	_terminal = {
	JobState.DONE.value,
	JobState.FAILED.value,
	JobState.CANCELLED.value,
	JobState.FAILED_RESTART.value,
	}
	async with self._session_factory() as session:
	job = await jobs_repo.get(session, job_id)
	if job is None:
	return # silent no-op (Protocol contract)
	if job.state in _terminal:
	logger.warning(
	"mark_failed called on job %s in terminal state %s",
	job_id,
	job.state,
	)
	return
	await jobs_repo.mark_failed(session, job_id, error)
	async def mark_failed(self, job_id: UUID, error: str) -> None:
	"""Transition any non-terminal -> FAILED. Silently no-op if job not found."""
	async with self._session_factory() as session:
	try:
	await jobs_repo.mark_failed(session, job_id, error)
	except ValueError as exc:
	if "not found" in str(exc):
	return
	logger.warning(str(exc))

[gemini-code-assist]

Nach dem Aufruf von self._task.cancel() im Fehlerfall wird der Task nicht mit await abgewartet. Dies kann dazu führen, dass der Task beim Beenden der Anwendung noch kurzzeitig im Hintergrund läuft oder Ressourcen nicht rechtzeitig freigegeben werden. Es wird empfohlen, den Task nach dem Abbruch explizit abzuwarten und eine eventuelle asyncio.CancelledError abzufangen.

Suggested change

	async def stop(self, timeout_s: float = 5.0) -> None:
	"""Signalisiert dem Loop zu stoppen und wartet bis zu timeout_s Sekunden.
	Nach Ablauf des Timeouts wird der Task gecancelled.
	"""
	self._stopping.set()
	if self._task is not None:
	try:
	await asyncio.wait_for(self._task, timeout=timeout_s)
	except TimeoutError:
	self._task.cancel()
	logger.warning(
	"CleanupTask hat sich nicht in %ss beendet, Task gecancelled",
	timeout_s,
	)
	self._task = None
	async def stop(self, timeout_s: float = 5.0) -> None:
	"""Signalisiert dem Loop zu stoppen und wartet bis zu timeout_s Sekunden.
	Nach Ablauf des Timeouts wird der Task gecancelled.
	"""
	self._stopping.set()
	if self._task is not None:
	try:
	await asyncio.wait_for(self._task, timeout=timeout_s)
	except TimeoutError:
	self._task.cancel()
	try:
	await self._task
	except asyncio.CancelledError:
	pass
	logger.warning(
	"CleanupTask hat sich nicht in %ss beendet, Task gecancelled",
	timeout_s,
	)
	self._task = None

[gemini-code-assist]

Wenn self._store.mark_failed im except-Block fehlschlägt (z. B. wegen Verbindungsproblemen zur DB), wird die ursprüngliche Exception von submit_with_id maskiert. Es wird empfohlen, den Rollback-Datenbankaufruf in ein eigenes try-except zu hüllen, um sicherzustellen, dass die primäre Fehlerursache immer korrekt an den Aufrufer propagiert wird.

Suggested change

	except Exception as exc:
	# I-1-Fix: in-memory Submit fehlgeschlagen nach DB-Persist — Rollback.
	# Ohne diesen Rollback bliebe eine stale QUEUED-Row in der DB ohne
	# Worker-Gegenstück, die nach Hub-Restart fälschlicherweise re-enqueued
	# würde. mark_failed markiert die Row als FAILED und verhindert das.
	await self._store.mark_failed(
	db_job.id,
	f"submit_failed: {exc.__class__.__name__}: {exc}",
	)
	raise
	except Exception as exc:
	# I-1-Fix: in-memory Submit fehlgeschlagen nach DB-Persist — Rollback.
	# Ohne diesen Rollback bliebe eine stale QUEUED-Row in der DB ohne
	# Worker-Gegenstück, die nach Hub-Restart fälschlicherweise re-enqueued
	# würde. mark_failed markiert die Row als FAILED und verhindert das.
	try:
	await self._store.mark_failed(
	db_job.id,
	f"submit_failed: {exc.__class__.__name__}: {exc}",
	)
	except Exception as db_exc:
	logger.error("Rollback-mark_failed fehlgeschlagen für Job %s: %s", db_job.id, db_exc)
	raise

[Copilot]

Pull request overview

This PR adds Phase 2 DB-backed job persistence so submitted jobs survive restarts, batch/job snapshots can be served from SQLite, and terminal jobs are cleaned up by retention.

Changes:

• Adds JobStore abstractions, SQLite-backed persistence, and print-queue recovery/transition writes.
• Adds GET /api/batches/{id} snapshot support with batch summary state.
• Wires retention cleanup and expands unit/integration coverage for persistence, recovery, and snapshots.

Reviewed changes

Copilot reviewed 26 out of 27 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
docs/superpowers/specs/2026-05-31-phase-2-job-persistence-design.md	Adds Phase 2 persistence design/spec.
docs/superpowers/plans/2026-05-31-phase-2-job-persistence.md	Adds implementation plan and task breakdown.
backend/app/services/job_store.py	Introduces JobStore protocol and memory implementation.
backend/app/services/job_store_sqlite.py	Adds SQLite-backed JobStore.
backend/app/services/cleanup_task.py	Adds retention cleanup background task.
backend/app/services/print_queue.py	Persists queue state transitions and performs restart recovery.
backend/app/services/print_service.py	Persists jobs before queue handoff.
backend/app/repositories/jobs.py	Adds repository helpers for recovery, lookup, and cleanup.
backend/app/api/routes/batches.py	Adds batch snapshot endpoint.
backend/app/api/routes/print.py	Preserves string job IDs in API response.
backend/app/schemas/batch_read.py	Adds batch snapshot response schemas.
backend/app/main.py	Wires store, cleanup task, recovery renderer, and new route.
backend/app/config.py	Adds job retention setting.
backend/tests/conftest.py	Adds shared async session factory fixture.
backend/tests/unit/repositories/conftest.py	Adds repository test DB fixtures.
backend/tests/unit/repositories/init.py	Adds repository test package marker.
backend/tests/unit/repositories/test_jobs_phase2.py	Tests new jobs repository helpers.
backend/tests/unit/services/test_job_store_memory.py	Tests memory store behavior.
backend/tests/unit/services/test_job_store_sqlite.py	Tests SQLite store behavior.
backend/tests/unit/services/test_cleanup_task.py	Tests cleanup task lifecycle/fail-soft behavior.
backend/tests/unit/services/test_print_queue_persistence.py	Tests queue persistence callbacks.
backend/tests/unit/services/test_print_service.py	Updates print service tests for UUID/persistence flow.
backend/tests/integration/conftest.py	Patches integration DB session binding.
backend/tests/integration/test_print_service_persistence.py	Tests service-level DB persistence before queueing.
backend/tests/integration/test_print_queue_recovery.py	Tests restart recovery behavior.
backend/tests/integration/test_batch_snapshot_endpoint.py	Tests batch snapshot endpoint summaries/order.
backend/tests/integration/test_phase6b_sse_with_batch.py	Adjusts printer setup for Phase 2 stub row behavior.