Lukas
52f237d5d4
CRITICAL FIX: Background tasks (especially blocklist_import) crashed mid-execution, leaving partial state. On retry, the same bans were applied again, causing duplicates. Solution: Content-hash based operation tracking for blocklist imports: - Added import_runs table (migration 6) to track operations by source + content hash - Before banning, check if this exact content has already been imported - If completed: skip banning (already done), optionally re-warm cache - If new or failed: proceed with ban and mark as completed or failed Changes: - Database: Migration 6 adds import_runs table with operation state tracking - Model: Added ImportRunEntry for import run records - Repository: New import_run_repo module with CRUD operations - Workflow: Updated blocklist_import_workflow to check operation history before banning - Dependencies: Registered import_run_repo for dependency injection - Tests: Added test_import_source_idempotent_on_retry and test_import_source_different_content_not_reused - Documentation: Added Task Idempotency section to Backend-Development.md Verification: - All 7 import tests pass (5 existing + 2 new idempotency tests) - Type checking: mypy --strict ✅ - Linting: ruff ✅ - No API changes, backwards compatible via automatic migration Fixes: Background tasks not idempotent #CRITICAL Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
141 lines
3.6 KiB
Python
141 lines
3.6 KiB
Python
"""Import run repository for blocklist import idempotency tracking.
|
|
|
|
Persists and queries import run records in the ``import_runs`` table.
|
|
Enables detection of duplicate import attempts and prevents re-running bans
|
|
on scheduler retry after a crash.
|
|
|
|
All methods are plain async functions that accept an :class:`aiosqlite.Connection`.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import TYPE_CHECKING
|
|
|
|
if TYPE_CHECKING:
|
|
import aiosqlite
|
|
|
|
from app.models.blocklist import ImportRunEntry
|
|
|
|
|
|
async def get_by_source_and_hash(
|
|
db: aiosqlite.Connection,
|
|
source_id: int,
|
|
content_hash: str,
|
|
) -> ImportRunEntry | None:
|
|
"""Check if a specific import (by source and content hash) already exists.
|
|
|
|
Args:
|
|
db: Active aiosqlite connection.
|
|
source_id: FK to ``blocklist_sources.id``.
|
|
content_hash: SHA256 hash of the downloaded blocklist content.
|
|
|
|
Returns:
|
|
ImportRunEntry if found, None otherwise.
|
|
"""
|
|
async with db.execute(
|
|
"""
|
|
SELECT
|
|
id, source_id, content_hash, status,
|
|
imported_count, skipped_count, error_message,
|
|
created_at, updated_at
|
|
FROM import_runs
|
|
WHERE source_id = ? AND content_hash = ?
|
|
""",
|
|
(source_id, content_hash),
|
|
) as cursor:
|
|
row = await cursor.fetchone()
|
|
|
|
if not row:
|
|
return None
|
|
|
|
return ImportRunEntry(
|
|
id=row[0],
|
|
source_id=row[1],
|
|
content_hash=row[2],
|
|
status=row[3],
|
|
imported_count=row[4],
|
|
skipped_count=row[5],
|
|
error_message=row[6],
|
|
created_at=row[7],
|
|
updated_at=row[8],
|
|
)
|
|
|
|
|
|
async def create_pending(
|
|
db: aiosqlite.Connection,
|
|
source_id: int,
|
|
content_hash: str,
|
|
) -> int:
|
|
"""Create a pending import run entry.
|
|
|
|
Args:
|
|
db: Active aiosqlite connection.
|
|
source_id: FK to ``blocklist_sources.id``.
|
|
content_hash: SHA256 hash of the downloaded blocklist content.
|
|
|
|
Returns:
|
|
Primary key of the inserted row.
|
|
"""
|
|
cursor = await db.execute(
|
|
"""
|
|
INSERT INTO import_runs (source_id, content_hash, status)
|
|
VALUES (?, ?, 'pending')
|
|
""",
|
|
(source_id, content_hash),
|
|
)
|
|
await db.commit()
|
|
return int(cursor.lastrowid) # type: ignore[arg-type]
|
|
|
|
|
|
async def mark_completed(
|
|
db: aiosqlite.Connection,
|
|
run_id: int,
|
|
imported_count: int,
|
|
skipped_count: int,
|
|
) -> None:
|
|
"""Mark an import run as completed with final counts.
|
|
|
|
Args:
|
|
db: Active aiosqlite connection.
|
|
run_id: Primary key of the import run.
|
|
imported_count: Number of IPs successfully banned.
|
|
skipped_count: Number of entries skipped (invalid or CIDR).
|
|
"""
|
|
await db.execute(
|
|
"""
|
|
UPDATE import_runs
|
|
SET status = 'completed',
|
|
imported_count = ?,
|
|
skipped_count = ?,
|
|
updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
|
|
WHERE id = ?
|
|
""",
|
|
(imported_count, skipped_count, run_id),
|
|
)
|
|
await db.commit()
|
|
|
|
|
|
async def mark_failed(
|
|
db: aiosqlite.Connection,
|
|
run_id: int,
|
|
error_message: str,
|
|
) -> None:
|
|
"""Mark an import run as failed with error details.
|
|
|
|
Args:
|
|
db: Active aiosqlite connection.
|
|
run_id: Primary key of the import run.
|
|
error_message: Error description.
|
|
"""
|
|
await db.execute(
|
|
"""
|
|
UPDATE import_runs
|
|
SET status = 'failed',
|
|
error_message = ?,
|
|
updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
|
|
WHERE id = ?
|
|
""",
|
|
(error_message, run_id),
|
|
)
|
|
await db.commit()
|