File size: 4,468 Bytes
6aecb2e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 | """
One-time application startup initialization for multi-worker deployments.
Why:
- Uvicorn/Gunicorn workers each run FastAPI startup handlers.
- Database init, migrations, and default-template bootstrap should run once per app startup,
not once per worker.
"""
from __future__ import annotations
import logging
import os
import tempfile
import time
from contextlib import contextmanager
from typing import Iterator
from ..core.config import app_config
from .create_default_template import ensure_default_templates_exist
from .database import init_db
from .startup_migrations import run_startup_migrations
logger = logging.getLogger(__name__)
_PROCESS_START_TS = time.time()
def _startup_lock_settings() -> tuple[int, int]:
timeout_seconds = int(getattr(app_config, "auto_migrate_lock_timeout_seconds", 300) or 300)
stale_seconds = int(getattr(app_config, "auto_migrate_lock_stale_seconds", 900) or 900)
return timeout_seconds, stale_seconds
def _startup_done_for_current_process(done_path: str) -> bool:
try:
return os.path.getmtime(done_path) >= (_PROCESS_START_TS - 1.0)
except OSError:
return False
@contextmanager
def _startup_owner_gate(
lock_path: str,
*,
done_path: str,
timeout_seconds: int,
stale_seconds: int,
) -> Iterator[bool]:
"""
Yield True only for the worker that should run one-time startup tasks.
Other workers wait for the current owner to finish and then skip the one-time work.
"""
lock_path = os.path.abspath(lock_path)
done_path = os.path.abspath(done_path)
if _startup_done_for_current_process(done_path):
yield False
return
deadline = time.time() + max(1, int(timeout_seconds))
owner = False
while True:
if _startup_done_for_current_process(done_path):
yield False
return
try:
fd = os.open(lock_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY)
try:
os.write(fd, f"{os.getpid()} {time.time()}\n".encode("utf-8", errors="ignore"))
finally:
os.close(fd)
owner = True
break
except FileExistsError:
try:
st = os.stat(lock_path)
if (time.time() - st.st_mtime) > max(30, int(stale_seconds)):
try:
os.unlink(lock_path)
continue
except OSError:
pass
except OSError:
pass
if time.time() >= deadline:
raise TimeoutError(f"Timed out waiting for startup initialization lock: {lock_path}")
time.sleep(0.5)
try:
if _startup_done_for_current_process(done_path):
yield False
return
yield True
try:
with open(done_path, "w", encoding="utf-8") as f:
f.write(f"{os.getpid()} {time.time()}\n")
except OSError:
logger.warning("Startup initialization: failed to write done marker", exc_info=True)
finally:
if owner:
try:
os.unlink(lock_path)
except OSError:
pass
async def run_startup_initialization() -> bool:
"""Run one-time startup initialization exactly once across local workers."""
timeout_seconds, stale_seconds = _startup_lock_settings()
temp_dir = tempfile.gettempdir()
lock_path = os.path.join(temp_dir, "landppt_startup_init.lock")
done_path = os.path.join(temp_dir, "landppt_startup_init.done")
with _startup_owner_gate(
lock_path,
done_path=done_path,
timeout_seconds=timeout_seconds,
stale_seconds=stale_seconds,
) as is_owner:
if not is_owner:
logger.info("Startup initialization: another worker already completed one-time startup tasks")
return False
logger.info(
"Startup initialization: initializing database (configured=%s)",
getattr(app_config, "database_url", ""),
)
await init_db()
logger.info("Startup initialization: database initialized successfully")
await run_startup_migrations()
template_ids = await ensure_default_templates_exist()
logger.info(
"Startup initialization: template bootstrap completed, available=%s",
len(template_ids or []),
)
return True
|