lukas.pupkalipinski/BanGUI
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add explicit database transaction isolation to multi-step operations

Browse Source 
This commit addresses race conditions in multi-step database operations by:

1. Wrap write operations in BEGIN IMMEDIATE ... COMMIT transactions:
   - import_run_repo: create_pending, mark_completed, mark_failed
   - geo_cache_repo: all upsert_*_and_commit functions
   - geo_cache_repo: bulk_upsert_entries_and_neg_entries_and_commit

2. Handle concurrent write collisions gracefully:
   - import_run_repo.create_pending can now raise IntegrityError
   - blocklist_import_workflow catches IntegrityError and retries lookup
   - Logs 'blocklist_import_lost_race' event when another request wins the race

3. Add comprehensive documentation:
   - Backend-Development.md § 6.3 Database Transactions
   - Explains when to use BEGIN IMMEDIATE
   - Shows transaction pattern with try-except-rollback
   - Documents race condition error handling pattern

The solution leverages SQLite's UNIQUE constraint for data integrity while
handling the concurrent case gracefully in application logic. This is more
efficient than using BEGIN EXCLUSIVE which would serialize all writers.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
Lukas
2026-04-30 22:04:15 +02:00
parent 94d6352d1d
commit f9e283541b
5 changed files with 259 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files

92
backend/app/repositories/geo_cache_repo.py
View File

@@ -18,7 +18,6 @@ if TYPE_CHECKING:
from app.models.geo import GeoCacheEntry
# Alias for backward compatibility with protocols
GeoCacheRow = GeoCacheEntry
@@ -109,9 +108,17 @@ async def upsert_entry_and_commit(
asn: str | None,
org: str | None,
) -> None:
"""Insert or update a resolved geo cache entry and commit."""
await upsert_entry(db, ip, country_code, country_name, asn, org)
await db.commit()
"""Insert or update a resolved geo cache entry and commit.
Wraps the upsert in an explicit transaction to ensure atomicity.
"""
try:
await db.execute("BEGIN IMMEDIATE")
await upsert_entry(db, ip, country_code, country_name, asn, org)
await db.commit()
except Exception:
await db.rollback()
raise
async def upsert_neg_entry(db: aiosqlite.Connection, ip: str) -> None:
@@ -127,9 +134,17 @@ async def upsert_neg_entry(db: aiosqlite.Connection, ip: str) -> None:
async def upsert_neg_entry_and_commit(db: aiosqlite.Connection, ip: str) -> None:
"""Record a failed lookup attempt and commit the transaction."""
await upsert_neg_entry(db, ip)
await db.commit()
"""Record a failed lookup attempt and commit the transaction.
Wraps the upsert in an explicit transaction to ensure atomicity.
"""
try:
await db.execute("BEGIN IMMEDIATE")
await upsert_neg_entry(db, ip)
await db.commit()
except Exception:
await db.rollback()
raise
async def bulk_upsert_entries(
@@ -173,17 +188,33 @@ async def bulk_upsert_entries_and_commit(
db: aiosqlite.Connection,
rows: Sequence[tuple[str, str | None, str | None, str | None, str | None]],
) -> int:
"""Bulk insert or update multiple geo cache entries and commit."""
count = await bulk_upsert_entries(db, rows)
await db.commit()
return count
"""Bulk insert or update multiple geo cache entries and commit.
Wraps the bulk upsert in an explicit transaction to ensure atomicity.
"""
try:
await db.execute("BEGIN IMMEDIATE")
count = await bulk_upsert_entries(db, rows)
await db.commit()
return count
except Exception:
await db.rollback()
raise
async def bulk_upsert_neg_entries_and_commit(db: aiosqlite.Connection, ips: list[str]) -> int:
"""Bulk insert negative lookup entries and commit."""
count = await bulk_upsert_neg_entries(db, ips)
await db.commit()
return count
"""Bulk insert negative lookup entries and commit.
Wraps the bulk upsert in an explicit transaction to ensure atomicity.
"""
try:
await db.execute("BEGIN IMMEDIATE")
count = await bulk_upsert_neg_entries(db, ips)
await db.commit()
return count
except Exception:
await db.rollback()
raise
async def bulk_upsert_entries_and_neg_entries_and_commit(
@@ -191,17 +222,34 @@ async def bulk_upsert_entries_and_neg_entries_and_commit(
rows: Sequence[tuple[str, str | None, str | None, str | None, str | None]],
ips: list[str],
) -> tuple[int, int]:
"""Persist positive and negative geo cache rows together, then commit."""
"""Persist positive and negative geo cache rows together, then commit.
Wraps both upserts in a single transaction to ensure atomicity.
Either all rows are persisted or none are.
Args:
db: Active aiosqlite connection.
rows: Sequence of (ip, country_code, country_name, asn, org) tuples.
ips: List of IP strings for negative entries (failed lookups).
Returns:
A tuple (positive_count, negative_count) of rows persisted.
"""
positive_count = 0
negative_count = 0
if rows:
positive_count = await bulk_upsert_entries(db, rows)
if ips:
negative_count = await bulk_upsert_neg_entries(db, ips)
try:
await db.execute("BEGIN IMMEDIATE")
if rows:
positive_count = await bulk_upsert_entries(db, rows)
if ips:
negative_count = await bulk_upsert_neg_entries(db, ips)
if rows or ips:
await db.commit()
if rows or ips:
await db.commit()
except Exception:
await db.rollback()
raise
return positive_count, negative_count

92
backend/app/repositories/import_run_repo.py
View File

@@ -68,6 +68,10 @@ async def create_pending(
) -> int:
"""Create a pending import run entry.
Wraps the insert in an explicit transaction to ensure atomicity and enable
proper error handling if a UNIQUE(source_id, content_hash) constraint
violation occurs due to concurrent requests.
Args:
db: Active aiosqlite connection.
source_id: FK to ``blocklist_sources.id``.
@@ -75,16 +79,26 @@ async def create_pending(
Returns:
Primary key of the inserted row.
Raises:
aiosqlite.IntegrityError: If a row with this (source_id, content_hash)
already exists (constraint violation). The caller should catch this
and retry the lookup to get the existing run's ID.
"""
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]
try:
await db.execute("BEGIN IMMEDIATE")
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]
except Exception:
await db.rollback()
raise
async def mark_completed(
@@ -95,24 +109,31 @@ async def mark_completed(
) -> None:
"""Mark an import run as completed with final counts.
Wraps the update in an explicit transaction to ensure atomicity.
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()
try:
await db.execute("BEGIN IMMEDIATE")
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()
except Exception:
await db.rollback()
raise
async def mark_failed(
@@ -122,19 +143,26 @@ async def mark_failed(
) -> None:
"""Mark an import run as failed with error details.
Wraps the update in an explicit transaction to ensure atomicity.
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()
try:
await db.execute("BEGIN IMMEDIATE")
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()
except Exception:
await db.rollback()
raise

49
backend/app/services/blocklist_import_workflow.py
View File

@@ -15,6 +15,7 @@ import hashlib
from typing import TYPE_CHECKING
import aiohttp
import aiosqlite
import structlog
from app.models.blocklist import BlocklistSource, ImportSourceResult
@@ -26,8 +27,6 @@ from app.services.blocklist_parser import BlocklistParser
if TYPE_CHECKING:
from collections.abc import Awaitable, Callable
import aiosqlite
from app.services.geo_cache import GeoCache
log: structlog.stdlib.BoundLogger = structlog.get_logger()
@@ -191,17 +190,41 @@ class BlocklistImportWorkflow:
# --- Create or update pending import run entry ---
if existing_run is None:
run_id = await import_run_repo.create_pending(
db,
source.id,
content_hash,
)
log.info(
"blocklist_import_tracking_created",
source_id=source.id,
run_id=run_id,
content_hash=content_hash[:8],
)
try:
run_id = await import_run_repo.create_pending(
db,
source.id,
content_hash,
)
log.info(
"blocklist_import_tracking_created",
source_id=source.id,
run_id=run_id,
content_hash=content_hash[:8],
)
except aiosqlite.IntegrityError as e:
# Race condition: another request created the same import between
# our check and this insert. Fetch the existing run and use its ID.
existing_run = await import_run_repo.get_by_source_and_hash(
db,
source.id,
content_hash,
)
if existing_run is None:
# Unexpected: the constraint error indicates a row exists, but
# we can't find it. This should not happen in normal operation.
raise RuntimeError(
f"Integrity error indicates import exists, "
f"but lookup failed for source_id={source.id}, "
f"content_hash={content_hash[:8]}"
) from e
run_id = existing_run.id
log.info(
"blocklist_import_lost_race",
source_id=source.id,
run_id=run_id,
content_hash=content_hash[:8],
)
else:
# Retry case: existing run is pending or failed, try again
run_id = existing_run.id