diff --git a/README.md b/README.md index 43bdfcd..cc6cf37 100644 --- a/README.md +++ b/README.md @@ -18,9 +18,9 @@ Be kind. Be clear. Assume good intent. Keep feedback constructive. - 👤 **Creator:** [Gilles Romer](https://romerg.de/) - ✉️ **Contact:** [info@romerg.de](mailto:info@romerg.de) -![Version](https://img.shields.io/badge/version-v1.27.0-e3000b?style=flat-square) ![Dark Theme](https://img.shields.io/badge/theme-dark-1a1a2e?style=flat-square) ![TCGdex](https://img.shields.io/badge/card%20data-TCGdex-e3000b?style=flat-square) ![Docker](https://img.shields.io/badge/deploy-Docker-2496ed?style=flat-square) ![FastAPI](https://img.shields.io/badge/backend-FastAPI-009688?style=flat-square) ![React](https://img.shields.io/badge/frontend-React%2018-61dafb?style=flat-square) [![Ko-fi](https://img.shields.io/badge/support-Ko--fi-ff5e5b?style=flat-square&logo=ko-fi&logoColor=white)](https://ko-fi.com/gillesromer) +![Version](https://img.shields.io/badge/version-v1.27.1-e3000b?style=flat-square) ![Dark Theme](https://img.shields.io/badge/theme-dark-1a1a2e?style=flat-square) ![TCGdex](https://img.shields.io/badge/card%20data-TCGdex-e3000b?style=flat-square) ![Docker](https://img.shields.io/badge/deploy-Docker-2496ed?style=flat-square) ![FastAPI](https://img.shields.io/badge/backend-FastAPI-009688?style=flat-square) ![React](https://img.shields.io/badge/frontend-React%2018-61dafb?style=flat-square) [![Ko-fi](https://img.shields.io/badge/support-Ko--fi-ff5e5b?style=flat-square&logo=ko-fi&logoColor=white)](https://ko-fi.com/gillesromer) -**Current version:** `v1.27.0` · Releases are tracked on the [GitHub Releases page](https://github.com/Git-Romer/pokecollector/releases). +**Current version:** `v1.27.1` · Releases are tracked on the [GitHub Releases page](https://github.com/Git-Romer/pokecollector/releases). ![WebApp Preview](preview-homescreen.png) diff --git a/VERSION b/VERSION index 5db08bf..08002f8 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -1.27.0 +1.27.1 diff --git a/backend/services/sync_service.py b/backend/services/sync_service.py index 8d63e0e..52301d9 100644 --- a/backend/services/sync_service.py +++ b/backend/services/sync_service.py @@ -2,9 +2,11 @@ import datetime import math from contextlib import contextmanager -from typing import Any, Mapping +from typing import Any, Iterable, Mapping from sqlalchemy.orm import Session, load_only from sqlalchemy import func, or_, text +from sqlalchemy.dialects.postgresql import insert as pg_insert +from sqlalchemy.dialects.sqlite import insert as sqlite_insert from models import Card, Set, CollectionItem, WishlistItem, BinderCard, PriceHistory, SyncLog, PortfolioSnapshot, CustomCardMatch, ProductPurchase, User, UserSetting from services import pokemon_api, telegram from services.card_fallbacks import apply_cross_language_fallbacks, build_missing_language_cards_for_set @@ -25,6 +27,8 @@ MISSING_PRICE_SYNC_RATIO = 0.7 NO_PRICE_RETRY_COOLDOWN = datetime.timedelta(hours=24) PRICE_SYNC_DB_CHUNK_SIZE = 400 # Stay below SQLite's common 999-parameter limit. +PRICE_HISTORY_UPSERT_CHUNK_SIZE = 100 # Seven bound values per row; safe for older SQLite builds. +PRICE_HISTORY_ADVISORY_LOCK_ID = 0x506F6B6550726963 # "PokePric" as a signed 64-bit int. COMPLETE_SET_CARD_REFRESH_LIMIT = 25 @@ -456,25 +460,58 @@ def _sets_for_card_catalogue_sync(db: Session) -> list[Set]: return db.query(Set).filter(sync_set_filter(db)).order_by(Set.updated_at.asc(), Set.id.asc()).all() -def record_price_history(db: Session, card: Card): - """Record today's price for a card.""" - today = datetime.date.today() - existing = db.query(PriceHistory).filter( - PriceHistory.card_id == card.id, - PriceHistory.date == today - ).first() +def record_price_history_batch( + db: Session, + cards: Iterable[Card], + *, + recorded_on: datetime.date | None = None, +) -> int: + """Upsert one daily price-history row per card near the sync commit. + + Full and price syncs may overlap. Buffering their history rows until all + network work is complete avoids holding database locks throughout the + long-running sync. PostgreSQL serializes only this short final write phase; + sorted batches provide a deterministic lock order on every database, while + ON CONFLICT handles rows committed by an earlier sync. + """ + cards_by_id = { + card.id: card + for card in cards + if card is not None and card.id + } + if not cards_by_id: + return 0 + + recorded_on = recorded_on or datetime.date.today() + rows = [ + { + "card_id": card_id, + "date": recorded_on, + "price_low": card.price_low, + "price_mid": card.price_mid, + "price_high": card.price_high, + "price_market": card.price_market, + "price_trend": card.price_trend, + } + for card_id, card in sorted(cards_by_id.items()) + ] + + if _db_dialect_name(db) == "postgresql": + db.execute( + text("SELECT pg_advisory_xact_lock(:lock_id)"), + {"lock_id": PRICE_HISTORY_ADVISORY_LOCK_ID}, + ) - if not existing: - history = PriceHistory( - card_id=card.id, - date=today, - price_low=card.price_low, - price_mid=card.price_mid, - price_high=card.price_high, - price_market=card.price_market, - price_trend=card.price_trend, + # Keep card inserts/updates ahead of their history rows for FK safety. The + # application session disables autoflush, so make that ordering explicit. + db.flush() + insert_fn = pg_insert if _db_dialect_name(db) == "postgresql" else sqlite_insert + for row_chunk in _chunks(rows, PRICE_HISTORY_UPSERT_CHUNK_SIZE): + stmt = insert_fn(PriceHistory).values(row_chunk).on_conflict_do_nothing( + index_elements=["card_id", "date"] ) - db.add(history) + db.execute(stmt) + return len(rows) def check_wishlist_alerts(db: Session, updated_card_ids: list): @@ -699,6 +736,7 @@ def _perform_full_sync_locked(db: Session) -> dict: cards_updated = 0 sets_updated = 0 updated_card_ids = [] + price_history_cards = [] try: # 1. Sync all sets first @@ -878,7 +916,7 @@ def _perform_full_sync_locked(db: Session) -> dict: ) parsed["set_id"] = None card = upsert_card(db, parsed) - record_price_history(db, card) + price_history_cards.append(card) logger.debug( "Full sync price card saved: requested_card_id=%s saved_card_id=%s saved_prices=%s saved_price_source_lang=%s", card_id, @@ -891,6 +929,7 @@ def _perform_full_sync_locked(db: Session) -> dict: except Exception as e: logger.warning(f"Failed to sync card {card_id}: {e}") + record_price_history_batch(db, price_history_cards) db.commit() # 3. Check wishlist alerts @@ -938,6 +977,7 @@ def perform_price_sync(db: Session, *, force: bool = False) -> dict: cards_updated = 0 updated_card_ids = [] + price_history_cards = [] try: price_plan = _price_sync_plan(db, force=force) @@ -1010,7 +1050,7 @@ def perform_price_sync(db: Session, *, force: bool = False) -> dict: ) parsed["set_id"] = None card = upsert_card(db, parsed) - record_price_history(db, card) + price_history_cards.append(card) logger.debug( "Price sync card saved: requested_card_id=%s saved_card_id=%s saved_prices=%s saved_price_source_lang=%s", card_id, @@ -1023,6 +1063,7 @@ def perform_price_sync(db: Session, *, force: bool = False) -> dict: except Exception as e: logger.warning(f"Failed to sync card {card_id}: {e}") + record_price_history_batch(db, price_history_cards) db.commit() # Check wishlist alerts @@ -1043,6 +1084,7 @@ def perform_price_sync(db: Session, *, force: bool = False) -> dict: except Exception as e: logger.error(f"Price sync failed: {e}") + db.rollback() log.finished_at = datetime.datetime.utcnow() log.status = "error" log.error_message = str(e) diff --git a/backend/tests/test_price_history_upsert.py b/backend/tests/test_price_history_upsert.py new file mode 100644 index 0000000..3ac30d0 --- /dev/null +++ b/backend/tests/test_price_history_upsert.py @@ -0,0 +1,238 @@ +import datetime +import os +import queue +import threading +import time +import unittest +import uuid +from unittest.mock import patch + +try: + from sqlalchemy import create_engine, event, text + from sqlalchemy.orm import sessionmaker + + from database import Base + from models import Card, PriceHistory, SyncLog + from services.sync_service import ( + PRICE_HISTORY_UPSERT_CHUNK_SIZE, + perform_price_sync, + record_price_history_batch, + ) + + DEPS = True +except ModuleNotFoundError: + DEPS = False + + +def _card(card_id: str, price: float) -> Card: + tcg_card_id = card_id.removesuffix("_en") + return Card( + id=card_id, + tcg_card_id=tcg_card_id, + name=tcg_card_id, + lang="en", + price_market=price, + ) + + +@unittest.skipUnless(DEPS, "SQLAlchemy not installed in this lightweight test environment") +class RecordPriceHistoryBatchTests(unittest.TestCase): + def setUp(self): + self.engine = create_engine("sqlite:///:memory:") + Base.metadata.create_all(self.engine) + self.Session = sessionmaker(bind=self.engine, autoflush=False) + + def tearDown(self): + self.engine.dispose() + + def test_empty_batch_is_a_noop(self): + db = self.Session() + self.assertEqual(record_price_history_batch(db, []), 0) + self.assertEqual(db.query(PriceHistory).count(), 0) + db.close() + + def test_batch_deduplicates_cards_and_existing_daily_rows(self): + db = self.Session() + first = _card("sv1-1_en", 1.0) + second = _card("sv1-2_en", 2.0) + db.add_all([first, second]) + + attempted = record_price_history_batch(db, [second, first, first]) + db.commit() + self.assertEqual(attempted, 2) + + first.price_market = 9.0 + record_price_history_batch(db, [first, second]) + db.commit() + + rows = db.query(PriceHistory).order_by(PriceHistory.card_id).all() + self.assertEqual([row.card_id for row in rows], ["sv1-1_en", "sv1-2_en"]) + self.assertEqual(rows[0].price_market, 1.0) + db.close() + + def test_price_sync_rolls_back_before_recording_database_error(self): + db = self.Session() + with patch.object(db, "rollback", wraps=db.rollback) as rollback, \ + patch("services.sync_service.logger.error"), \ + patch( + "services.sync_service._price_sync_plan", + side_effect=RuntimeError("database write failed"), + ): + with self.assertRaisesRegex(RuntimeError, "database write failed"): + perform_price_sync(db) + + rollback.assert_called_once() + log = db.query(SyncLog).one() + self.assertEqual(log.sync_type, "price") + self.assertEqual(log.status, "error") + self.assertEqual(log.error_message, "database write failed") + db.close() + + def test_large_batch_uses_bounded_bulk_statements(self): + db = self.Session() + card_count = PRICE_HISTORY_UPSERT_CHUNK_SIZE * 2 + 5 + cards = [ + _card(f"sv1-{index:04d}_en", float(index)) + for index in range(card_count) + ] + db.add_all(cards) + history_inserts = [] + + def capture_history_inserts(_conn, _cursor, statement, _parameters, _context, _many): + if statement.startswith("INSERT INTO price_history"): + history_inserts.append(statement) + + event.listen(self.engine, "before_cursor_execute", capture_history_inserts) + try: + record_price_history_batch(db, reversed(cards)) + db.commit() + finally: + event.remove(self.engine, "before_cursor_execute", capture_history_inserts) + + self.assertEqual(len(history_inserts), 3) + self.assertEqual(db.query(PriceHistory).count(), card_count) + db.close() + + +POSTGRES_TEST_URL = os.getenv("TEST_POSTGRES_URL", "") + + +@unittest.skipUnless( + DEPS and POSTGRES_TEST_URL.startswith("postgresql"), + "TEST_POSTGRES_URL is required for PostgreSQL concurrency coverage", +) +class RecordPriceHistoryPostgresConcurrencyTests(unittest.TestCase): + def setUp(self): + self.schema = f"price_history_test_{uuid.uuid4().hex}" + self.admin_engine = create_engine(POSTGRES_TEST_URL, isolation_level="AUTOCOMMIT") + with self.admin_engine.connect() as conn: + conn.execute(text(f'CREATE SCHEMA "{self.schema}"')) + + self.engine = create_engine( + POSTGRES_TEST_URL, + connect_args={"options": f"-csearch_path={self.schema}"}, + ) + Base.metadata.create_all(self.engine) + self.Session = sessionmaker(bind=self.engine, autoflush=False) + + db = self.Session() + db.add_all([_card("sv1-1_en", 1.0), _card("sv1-2_en", 2.0)]) + db.commit() + db.close() + + def tearDown(self): + self.engine.dispose() + with self.admin_engine.connect() as conn: + conn.execute(text(f'DROP SCHEMA IF EXISTS "{self.schema}" CASCADE')) + self.admin_engine.dispose() + + def test_uncommitted_overlapping_batches_wait_then_finish_without_deadlock(self): + """The second writer must wait, then safely ignore the committed rows. + + Passing the cards in opposite orders also proves the batch helper imposes + its own stable lock order instead of trusting caller order. + """ + first_has_written = threading.Event() + release_first = threading.Event() + second_lock_attempted = threading.Event() + second_finished = threading.Event() + failures = queue.Queue() + + def observe_second_lock(_conn, _cursor, statement, _parameters, _context, _many): + if ( + threading.current_thread().name == "price-history-second" + and "pg_advisory_xact_lock" in statement + ): + second_lock_attempted.set() + + event.listen(self.engine, "before_cursor_execute", observe_second_lock) + + def write_first(): + db = self.Session() + try: + cards = { + card.id: card + for card in db.query(Card).filter(Card.id.in_(["sv1-1_en", "sv1-2_en"])).all() + } + record_price_history_batch(db, [cards["sv1-1_en"], cards["sv1-2_en"]]) + first_has_written.set() + if not release_first.wait(timeout=5): + raise TimeoutError("timed out waiting to release the first transaction") + db.commit() + except Exception as exc: + db.rollback() + failures.put(exc) + finally: + db.close() + + def write_second(): + db = self.Session() + try: + if not first_has_written.wait(timeout=5): + raise TimeoutError("first transaction never reached its uncommitted insert") + cards = { + card.id: card + for card in db.query(Card).filter(Card.id.in_(["sv1-1_en", "sv1-2_en"])).all() + } + record_price_history_batch(db, [cards["sv1-2_en"], cards["sv1-1_en"]]) + db.commit() + second_finished.set() + except Exception as exc: + db.rollback() + failures.put(exc) + finally: + db.close() + + first = threading.Thread(target=write_first, name="price-history-first", daemon=True) + second = threading.Thread(target=write_second, name="price-history-second", daemon=True) + first.start() + second.start() + + try: + self.assertTrue(second_lock_attempted.wait(timeout=5)) + time.sleep(0.2) + self.assertFalse( + second_finished.is_set(), + "second writer did not wait for the uncommitted conflicting batch", + ) + finally: + release_first.set() + first.join(timeout=5) + second.join(timeout=5) + event.remove(self.engine, "before_cursor_execute", observe_second_lock) + + self.assertFalse(first.is_alive()) + self.assertFalse(second.is_alive()) + if not failures.empty(): + raise failures.get() + + db = self.Session() + rows = db.query(PriceHistory).filter( + PriceHistory.date == datetime.date.today() + ).order_by(PriceHistory.card_id).all() + self.assertEqual([row.card_id for row in rows], ["sv1-1_en", "sv1-2_en"]) + db.close() + + +if __name__ == "__main__": + unittest.main() diff --git a/frontend/package-lock.json b/frontend/package-lock.json index b2d9a78..8aec010 100644 --- a/frontend/package-lock.json +++ b/frontend/package-lock.json @@ -1,12 +1,12 @@ { "name": "pokemon-tcg-collection", - "version": "1.27.0", + "version": "1.27.1", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "pokemon-tcg-collection", - "version": "1.27.0", + "version": "1.27.1", "dependencies": { "@tanstack/react-query": "^5.18.0", "axios": "^1.18.0", diff --git a/frontend/package.json b/frontend/package.json index cf83d0f..f256994 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -1,7 +1,7 @@ { "name": "pokemon-tcg-collection", "private": true, - "version": "1.27.0", + "version": "1.27.1", "type": "module", "scripts": { "dev": "vite",