From 59b6262c70c16a9f6854de18137193fc383f8ece Mon Sep 17 00:00:00 2001 From: Mikan <72257910+Mikan-DS@users.noreply.github.com> Date: Fri, 19 Jun 2026 11:30:38 +0300 Subject: [PATCH] fix bd --- .env.example | 10 ++ backend/app/migrations/init_db.py | 130 +++++++++++++++++++------- backend/app/workers/main.py | 16 +++- backend/app/workers/trigger_runner.py | 21 ++++- docker-compose.yml | 21 +++++ 5 files changed, 160 insertions(+), 38 deletions(-) diff --git a/.env.example b/.env.example index e898788..cb45687 100644 --- a/.env.example +++ b/.env.example @@ -20,6 +20,16 @@ DEFAULT_LLM_BASE_URL=http://host.docker.internal:1234/v1 DEFAULT_LLM_API_KEY=dummy DEFAULT_LLM_MODEL=local-model +# === Default embeddings / RAG (overridable via admin panel) === +# provider: "hash" (offline fallback, no semantic quality) or "openai" (real /embeddings endpoint). +# When provider=openai and base_url/api_key are empty, they fall back to the LLM settings above. +DEFAULT_EMBEDDING_PROVIDER=hash +DEFAULT_EMBEDDING_BASE_URL= +DEFAULT_EMBEDDING_API_KEY= +DEFAULT_EMBEDDING_MODEL=text-embedding-3-small +DEFAULT_EMBEDDING_DIM=0 +DEFAULT_EMBEDDING_REQUEST_TIMEOUT=60 + # === Frontend === VITE_API_URL=http://localhost:8000 FRONTEND_BUILD_TARGET=dev diff --git a/backend/app/migrations/init_db.py b/backend/app/migrations/init_db.py index 9631fa7..168cbfe 100644 --- a/backend/app/migrations/init_db.py +++ b/backend/app/migrations/init_db.py @@ -1,4 +1,11 @@ -"""Database initialization: create all tables and seed defaults.""" +"""Database initialization: create all tables and seed defaults. + +Idempotent: safe to call from multiple processes (backend lifespan + worker +startup) thanks to a PostgreSQL advisory lock that serializes the seeding +phase. `create_all` itself is already `CREATE TABLE IF NOT EXISTS`, so the +only race is on seed inserts — guarded by `pg_advisory_xact_lock` plus +per-row IntegrityError handling. +""" from __future__ import annotations import asyncio @@ -6,6 +13,7 @@ import json from pathlib import Path from sqlalchemy import select, text +from sqlalchemy.exc import IntegrityError from app.db import AsyncSessionLocal, Base, engine from app.models import GlossaryEntry, Preset, Setting, User @@ -17,6 +25,11 @@ from app.prompts.fantasy_preset import FANTASY_PRESET_RU, FANTASY_PRESET_EN log = get_logger("migrations") +# Stable advisory lock key so backend + worker don't race on seeding. +# (key1, key2) — arbitrary 64-bit integers, kept constant across runs. +_ADVISORY_LOCK_KEY = (42424201, 1) + + DEFAULT_SETTINGS = [ ("llm.base_url", settings.default_llm_base_url, "OpenAI-compatible base URL"), ("llm.api_key", settings.default_llm_api_key, "API key for LLM endpoint"), @@ -46,38 +59,31 @@ DEFAULT_SETTINGS = [ async def init_db() -> None: setup_logging() log.info("creating_tables") + # CREATE TABLE IF NOT EXISTS — safe to run concurrently. async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) log.info("tables_ready") - async with AsyncSessionLocal() as session: - # Seed settings - result = await session.execute(select(Setting).limit(1)) - if result.scalars().first() is None: - for key, value, desc in DEFAULT_SETTINGS: - session.add(Setting(key=key, value=value, description=desc)) + # Seed phase: serialize across processes via PG advisory transaction lock. + # On non-PG backends (SQLite for tests) the lock statement is a no-op + # (we catch the error and proceed without locking). + try: + async with AsyncSessionLocal() as session: + await session.execute( + text("SELECT pg_advisory_xact_lock(:k1, :k2)").bindparams( + k1=_ADVISORY_LOCK_KEY[0], k2=_ADVISORY_LOCK_KEY[1] + ) + ) + await _seed_settings(session) + await _seed_builtin_presets(session) await session.commit() - log.info("settings_seeded", count=len(DEFAULT_SETTINGS)) - else: - log.info("settings_already_exist") - - # Seed built-in Fantasy preset - result = await session.execute(select(Preset).where(Preset.is_builtin.is_(True))) - if result.scalars().first() is None: - for preset_def in (FANTASY_PRESET_RU, FANTASY_PRESET_EN): - session.add(Preset( - slug=preset_def["slug"], - title=preset_def["title"], - description=preset_def["description"], - language=preset_def["language"], - is_public=True, - is_builtin=True, - payload=preset_def["payload"], - )) + except Exception as e: + # Non-PG backend (SQLite) or transient error — retry without the lock. + log.warning("advisory_lock_unavailable_proceeding", error=f"{type(e).__name__}: {e}") + async with AsyncSessionLocal() as session: + await _seed_settings(session) + await _seed_builtin_presets(session) await session.commit() - log.info("builtin_presets_seeded") - else: - log.info("builtin_presets_already_exist") # Ensure admin_setup_token is set; if empty, generate and print token = settings.admin_setup_token.strip() @@ -89,15 +95,73 @@ async def init_db() -> None: existing_obj = existing.scalars().first() if existing_obj is None: session.add(Setting(key="admin.setup_token", value=token, description="One-time token for /admin/setup")) - await session.commit() - print("=" * 60) - print("ADMIN SETUP TOKEN (use at /admin/setup):") - print(token) - print("=" * 60) - log.info("admin_setup_token_generated") + try: + await session.commit() + print("=" * 60) + print("ADMIN SETUP TOKEN (use at /admin/setup):") + print(token) + print("=" * 60) + log.info("admin_setup_token_generated") + except IntegrityError: + # Another process inserted it concurrently — fine. + await session.rollback() + log.info("admin_setup_token_already_set") else: log.info("admin_setup_token_already_set") +async def _seed_settings(session) -> None: + """Insert default settings that don't yet exist (per-row, race-safe).""" + result = await session.execute(select(Setting).limit(1)) + if result.scalars().first() is not None: + log.info("settings_already_exist") + return + seeded = 0 + for key, value, desc in DEFAULT_SETTINGS: + # Check existence per-row to avoid IntegrityError on concurrent inserts + existing = await session.execute(select(Setting).where(Setting.key == key)) + if existing.scalars().first() is not None: + continue + session.add(Setting(key=key, value=value, description=desc)) + seeded += 1 + if seeded: + try: + await session.commit() + log.info("settings_seeded", count=seeded) + except IntegrityError: + await session.rollback() + log.info("settings_seed_skipped_concurrent") + else: + log.info("settings_already_exist") + + +async def _seed_builtin_presets(session) -> None: + """Insert built-in presets if none exist yet.""" + result = await session.execute(select(Preset).where(Preset.is_builtin.is_(True))) + if result.scalars().first() is not None: + log.info("builtin_presets_already_exist") + return + for preset_def in (FANTASY_PRESET_RU, FANTASY_PRESET_EN): + # Check by slug to avoid race on unique constraint + existing = await session.execute(select(Preset).where(Preset.slug == preset_def["slug"])) + if existing.scalars().first() is not None: + continue + session.add(Preset( + slug=preset_def["slug"], + title=preset_def["title"], + description=preset_def["description"], + language=preset_def["language"], + is_public=True, + is_builtin=True, + payload=preset_def["payload"], + )) + try: + await session.commit() + log.info("builtin_presets_seeded") + except IntegrityError: + await session.rollback() + log.info("builtin_presets_seed_skipped_concurrent") + + if __name__ == "__main__": asyncio.run(init_db()) diff --git a/backend/app/workers/main.py b/backend/app/workers/main.py index c83d6e0..fda106c 100644 --- a/backend/app/workers/main.py +++ b/backend/app/workers/main.py @@ -1,8 +1,15 @@ -"""Worker entrypoint: runs trigger checker + future background jobs.""" +"""Worker entrypoint: runs trigger checker + future background jobs. + +Waits for the database (and required tables) to be ready before starting +the background loops. This prevents the worker from crashing when the +backend hasn't finished running `init_db()` yet (typical docker-compose +race condition where both services start in parallel). +""" from __future__ import annotations import asyncio +from app.db_wait import wait_for_db_or_exit from app.logging_setup import get_logger, setup_logging from app.workers.trigger_runner import main_loop as trigger_loop @@ -12,6 +19,13 @@ log = get_logger("worker") async def main(): setup_logging() log.info("worker_starting") + + # Wait until DB is reachable and the `settings` table exists. + # The backend's lifespan runs `init_db()` which creates tables; if the + # worker comes up first, this loop will retry until that completes. + await wait_for_db_or_exit(max_retries=60, delay=2.0) + + log.info("worker_db_ready_starting_loops") # Run all background loops concurrently await asyncio.gather( trigger_loop(), diff --git a/backend/app/workers/trigger_runner.py b/backend/app/workers/trigger_runner.py index 691aa6a..664e880 100644 --- a/backend/app/workers/trigger_runner.py +++ b/backend/app/workers/trigger_runner.py @@ -155,21 +155,34 @@ async def _fire_trigger(db, trigger: DeferredTrigger, session: Session, world: W async def main_loop(): - """Main worker loop. Polls every N seconds for due triggers.""" + """Main worker loop. Polls every N seconds for due triggers. + + Resilient to transient DB errors: any error inside an iteration is logged + and the loop sleeps for a fallback interval before retrying, instead of + crashing the worker process. + """ setup_logging() log.info("trigger_worker_started") + fallback_interval = 30 # used when settings table is unreadable while True: + interval = fallback_interval try: async with AsyncSessionLocal() as db: enabled = await _get_setting(db, "triggers.enabled", True) - interval = int(await _get_setting(db, "triggers.check_interval", 30)) + interval = int(await _get_setting(db, "triggers.check_interval", fallback_interval)) if enabled: fired = await check_and_fire_triggers() if fired: log.info("triggers_fired", count=fired) except Exception as e: - log.error("trigger_worker_iteration_failed", error=str(e)) - await asyncio.sleep(max(5, int(await _get_setting_sleep()))) + log.error("trigger_worker_iteration_failed", error=f"{type(e).__name__}: {e}") + # Always sleep with a safe positive interval; never let a DB error + # escape this loop and crash the worker. + try: + sleep_for = max(5, int(interval)) + except Exception: + sleep_for = fallback_interval + await asyncio.sleep(sleep_for) async def _get_setting(db, key: str, default): diff --git a/docker-compose.yml b/docker-compose.yml index 3939c09..b52a0f6 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -65,6 +65,13 @@ services: DEFAULT_LLM_BASE_URL: ${DEFAULT_LLM_BASE_URL:-http://host.docker.internal:1234/v1} DEFAULT_LLM_API_KEY: ${DEFAULT_LLM_API_KEY:-dummy} DEFAULT_LLM_MODEL: ${DEFAULT_LLM_MODEL:-local-model} + # Default embeddings settings (overridable via admin panel) + DEFAULT_EMBEDDING_PROVIDER: ${DEFAULT_EMBEDDING_PROVIDER:-hash} + DEFAULT_EMBEDDING_BASE_URL: ${DEFAULT_EMBEDDING_BASE_URL:-} + DEFAULT_EMBEDDING_API_KEY: ${DEFAULT_EMBEDDING_API_KEY:-} + DEFAULT_EMBEDDING_MODEL: ${DEFAULT_EMBEDDING_MODEL:-text-embedding-3-small} + DEFAULT_EMBEDDING_DIM: ${DEFAULT_EMBEDDING_DIM:-0} + DEFAULT_EMBEDDING_REQUEST_TIMEOUT: ${DEFAULT_EMBEDDING_REQUEST_TIMEOUT:-60} ports: - "8000:8000" volumes: @@ -72,6 +79,12 @@ services: # Print admin setup token to console on first run stdin_open: true tty: true + healthcheck: + test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8000/health', timeout=3)"] + interval: 5s + timeout: 5s + retries: 30 + start_period: 30s worker: build: @@ -85,6 +98,8 @@ services: condition: service_healthy qdrant: condition: service_started + backend: + condition: service_healthy environment: DATABASE_URL: postgresql+asyncpg://${POSTGRES_USER:-airpg}:${POSTGRES_PASSWORD:-airpg_secret}@postgres:5432/${POSTGRES_DB:-airpg} REDIS_URL: redis://redis:6379/0 @@ -94,6 +109,12 @@ services: DEFAULT_LLM_BASE_URL: ${DEFAULT_LLM_BASE_URL:-http://host.docker.internal:1234/v1} DEFAULT_LLM_API_KEY: ${DEFAULT_LLM_API_KEY:-dummy} DEFAULT_LLM_MODEL: ${DEFAULT_LLM_MODEL:-local-model} + DEFAULT_EMBEDDING_PROVIDER: ${DEFAULT_EMBEDDING_PROVIDER:-hash} + DEFAULT_EMBEDDING_BASE_URL: ${DEFAULT_EMBEDDING_BASE_URL:-} + DEFAULT_EMBEDDING_API_KEY: ${DEFAULT_EMBEDDING_API_KEY:-} + DEFAULT_EMBEDDING_MODEL: ${DEFAULT_EMBEDDING_MODEL:-text-embedding-3-small} + DEFAULT_EMBEDDING_DIM: ${DEFAULT_EMBEDDING_DIM:-0} + DEFAULT_EMBEDDING_REQUEST_TIMEOUT: ${DEFAULT_EMBEDDING_REQUEST_TIMEOUT:-60} WORKER_MODE: "1" command: ["python", "-m", "app.workers.main"] volumes: