fix: worker waits for schema before claiming; npm ci for reproducible builds
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
+1
-1
@@ -4,7 +4,7 @@ RUN apt-get update && apt-get install -y openssl && rm -rf /var/lib/apt/lists/*
|
|||||||
COPY package.json package-lock.json* ./
|
COPY package.json package-lock.json* ./
|
||||||
# Copy schema before install so package.json's postinstall (prisma generate) can run
|
# Copy schema before install so package.json's postinstall (prisma generate) can run
|
||||||
COPY prisma ./prisma
|
COPY prisma ./prisma
|
||||||
RUN npm install
|
RUN npm ci
|
||||||
COPY . .
|
COPY . .
|
||||||
RUN npx prisma generate && npm run build
|
RUN npx prisma generate && npm run build
|
||||||
|
|
||||||
|
|||||||
@@ -1,4 +1,5 @@
|
|||||||
import os
|
import os
|
||||||
|
import time
|
||||||
|
|
||||||
import psycopg
|
import psycopg
|
||||||
|
|
||||||
@@ -6,3 +7,34 @@ import psycopg
|
|||||||
def connect() -> psycopg.Connection:
|
def connect() -> psycopg.Connection:
|
||||||
dsn = os.environ["DATABASE_URL"]
|
dsn = os.environ["DATABASE_URL"]
|
||||||
return psycopg.connect(dsn)
|
return psycopg.connect(dsn)
|
||||||
|
|
||||||
|
|
||||||
|
def wait_for_db(timeout: float = 60.0, interval: float = 2.0) -> psycopg.Connection:
|
||||||
|
"""Block until the database is up and the `Job` table exists.
|
||||||
|
|
||||||
|
Repeatedly opens a connection and probes for the `Job` table's presence.
|
||||||
|
Tolerates the DB not being reachable yet (`psycopg.OperationalError`) and
|
||||||
|
migrations not having run yet (`psycopg.errors.UndefinedTable`), retrying
|
||||||
|
with a fixed `interval` until `timeout` elapses. Returns the open, usable
|
||||||
|
connection on success; re-raises the last error on timeout.
|
||||||
|
"""
|
||||||
|
deadline = time.monotonic() + timeout
|
||||||
|
printed_waiting = False
|
||||||
|
last_error: Exception | None = None
|
||||||
|
while time.monotonic() < deadline:
|
||||||
|
conn: psycopg.Connection | None = None
|
||||||
|
try:
|
||||||
|
conn = connect()
|
||||||
|
with conn.cursor() as cur:
|
||||||
|
cur.execute('SELECT 1 FROM "Job" LIMIT 0')
|
||||||
|
return conn
|
||||||
|
except (psycopg.OperationalError, psycopg.errors.UndefinedTable) as exc:
|
||||||
|
last_error = exc
|
||||||
|
if conn is not None:
|
||||||
|
conn.close()
|
||||||
|
if not printed_waiting:
|
||||||
|
print("worker: waiting for database schema...", flush=True)
|
||||||
|
printed_waiting = True
|
||||||
|
time.sleep(interval)
|
||||||
|
assert last_error is not None
|
||||||
|
raise last_error
|
||||||
|
|||||||
@@ -1,7 +1,7 @@
|
|||||||
import time
|
import time
|
||||||
|
|
||||||
from lyra_worker.claim import claim_next
|
from lyra_worker.claim import claim_next
|
||||||
from lyra_worker.db import connect
|
from lyra_worker.db import wait_for_db
|
||||||
from lyra_worker.pipeline import run_pipeline
|
from lyra_worker.pipeline import run_pipeline
|
||||||
|
|
||||||
IDLE_SLEEP = 2.0
|
IDLE_SLEEP = 2.0
|
||||||
@@ -9,7 +9,7 @@ STAGE_DELAY = 1.0 # visible progress in the UI
|
|||||||
|
|
||||||
|
|
||||||
def run_forever() -> None:
|
def run_forever() -> None:
|
||||||
conn = connect()
|
conn = wait_for_db()
|
||||||
print("worker: waiting for jobs", flush=True)
|
print("worker: waiting for jobs", flush=True)
|
||||||
try:
|
try:
|
||||||
while True:
|
while True:
|
||||||
|
|||||||
@@ -0,0 +1,11 @@
|
|||||||
|
from lyra_worker.db import wait_for_db
|
||||||
|
|
||||||
|
|
||||||
|
def test_wait_for_db_returns_usable_connection_when_schema_ready(conn):
|
||||||
|
# `conn` fixture guarantees the DB is up and the schema is migrated.
|
||||||
|
result = wait_for_db(timeout=5)
|
||||||
|
try:
|
||||||
|
with result.cursor() as cur:
|
||||||
|
cur.execute('SELECT 1 FROM "Job" LIMIT 0')
|
||||||
|
finally:
|
||||||
|
result.close()
|
||||||
Reference in New Issue
Block a user