"""An OS-level lock so at most one `trader run`/`run-once` runs against a
given database at a time (issue #43).
2026-08-20: a rogue `trader run` daemon (started manually, never registered
with launchd), the launchd-managed daemon, and a manually invoked
`trader run-once` all ran concurrently against the same live database and
broker account. Three processes competed for Ollama's single request slot,
serializing/stalling all of them, and two of them raced inside
`trader/marketdata/cache.py`'s `_replace`: both refreshing the same symbol's
bar cache at once, one's insert collided with the other's already-committed
rows, and the exception propagated uncaught and crashed a whole daemon
cycle. Nothing in the app stopped this from happening.
**Postgres session-level advisory lock (issue #26), not `fcntl.flock`.**
Before the Postgres migration this held an OS file lock next to the SQLite
file; there is no file to lock next to a network database, so this now
acquires `pg_try_advisory_lock(key)` on one dedicated connection, opened
directly with `psycopg` and kept **outside the SQLAlchemy pool** for the
process's entire lifetime — deliberately not borrowed from the app's normal
`Engine`, so pool recycling/`pool_pre_ping` can never silently swap out the
connection holding the lock out from under it.
There is still no "is the recorded PID still alive" staleness check to get
wrong: Postgres releases every session-level advisory lock a backend holds
the instant that backend's connection closes, for any reason — a clean
`with`-block exit, an uncaught exception, or the process being `SIGKILL`ed
out from under an open socket. This module does call `pg_advisory_unlock`
explicitly on the clean-exit path (faster release than waiting on socket
teardown), but nothing depends on that call actually running.
**One real behavioural difference from `flock`, not a staleness check but
worth naming**: a `brew services restart postgresql`, or any transient loss
of the connection holding the lock, releases it — unlike an OS file
descriptor, which only depended on this process's own table, not on a
remote service staying up. This is not a silent hazard: the same dropped
connection is also the one issuing every other query in the cycle, so a
mid-cycle restart already breaks the cycle loudly through the normal
`DatabaseError` paths rather than continuing as if nothing happened. See
`docs/invariants.md` for the fuller writeup.
Scope is the *resolved database name*, never global: two processes pointed
at different databases (a dev checkout's `trader_ro` role against
`trader_live`, and a parallel test run against its own throwaway database)
must not conflict with each other, so the lock key is derived from each
process's own resolved database name rather than a fixed value.
On failure to acquire, this raises immediately — no retry, no queueing. That
mirrors the rest of this codebase's "everything that can go wrong resolves
to a safe default" discipline in shape only, not in outcome: the LLM path's
guiding rule is that a failure must never block a trade, but here the
opposite is correct — failing loudly and refusing to start *is* the safe
behaviour, because starting is what let three processes collide in the first
place.
"""
from __future__ import annotations
import contextlib
import hashlib
import struct
from collections.abc import Iterator
import psycopg
from sqlalchemy.engine import make_url
__all__ = ["InstanceAlreadyRunningError", "advisory_lock_key_for", "single_instance_lock"]
[docs]
class InstanceAlreadyRunningError(Exception):
"""Raised when another process already holds the single-instance lock."""
[docs]
def advisory_lock_key_for(database_url: str) -> int:
"""A stable `bigint` lock key derived from `database_url`'s resolved
database name.
`pg_try_advisory_lock` takes a signed 64-bit integer, so the database
name is hashed (`sha256`, not a weaker/faster hash — collision risk here
means two *different* databases silently sharing a lock, so this is
computed Python-side and kept explicit rather than leaned on implicitly
via Postgres's own per-database advisory-lock catalog, which would make
the "scoped to the resolved database, never global" guarantee harder to
audit) and the first 8 bytes are reinterpreted as a signed int64.
"""
database_name = make_url(database_url).database or ""
digest = hashlib.sha256(database_name.encode("utf-8")).digest()
(key,) = struct.unpack(">q", digest[:8])
return key
def _psycopg_dsn(database_url: str) -> str:
"""Strip SQLAlchemy's `+psycopg` driver suffix for a raw `psycopg.connect`
call. `psycopg.connect` does not understand `postgresql+psycopg://` —
that scheme is a SQLAlchemy dialect selector, not a libpq URL — and
raises a confusing "missing '=' after ... in connection info string"
instead of a clear "unknown scheme" error. Every DSN in this app is
built with that suffix (`config/loader.py`'s `assemble_database_url`),
so anything opening a connection outside SQLAlchemy needs this first.
"""
return (
make_url(database_url)
.set(drivername="postgresql")
.render_as_string(hide_password=False)
)
[docs]
@contextlib.contextmanager
def single_instance_lock(database_url: str) -> Iterator[None]:
"""Hold an exclusive Postgres advisory lock scoped to `database_url`'s
resolved database name.
Raises `InstanceAlreadyRunningError` immediately if another process
already holds it. Opens its own `psycopg` connection directly (not via
`create_db_engine`/the SQLAlchemy pool this app uses everywhere else),
autocommit, held open for the entire `with` block so the lock cannot be
silently dropped by pool recycling.
"""
key = advisory_lock_key_for(database_url)
connection = psycopg.connect(_psycopg_dsn(database_url), autocommit=True)
try:
(acquired,) = connection.execute(
"SELECT pg_try_advisory_lock(%s)", (key,)
).fetchone()
if not acquired:
raise InstanceAlreadyRunningError(
f"Another trader run/run-once is already running against this "
f"database (advisory lock key {key}). Refusing to start a "
"second instance against the same database."
)
except InstanceAlreadyRunningError:
connection.close()
raise
except Exception:
connection.close()
raise
try:
yield
finally:
try:
connection.execute("SELECT pg_advisory_unlock(%s)", (key,))
finally:
connection.close()