Skip to content

LocalBroker: design and operations

The local broker (BROKER_TYPE=local, the default) shares one in-memory queue between the processes of a single host. This page describes how it starts, how processes find it, and how it fails. For which broker to pick, read broker-matrix.md first.

Code: src/broker/local.py (backend), src/broker/contracts.py (interface), src/broker/factory.py (get_broker()).

Process model

  • Server (fp-brokerd): a child process running a multiprocessing.managers.BaseManager that serves one _State object: per-topic queues, running counters, workers, locks, topic configs, dead letters and stats. All state lives in this process; nothing is persisted.
  • Clients: every LocalBroker instance is a thin client holding one shared proxy per process (_shared_proxy, guarded by _client_lock). Each broker call goes through that proxy.
  • Address: TCP 127.0.0.1:50050 by default. BROKER_DSN overrides it (host:port); the authkey is b"fp-broker".

There is no Unix-socket leader election. Whoever calls setup() first on a free port spawns the server; everyone else finds the port answering and reuses it.

setup() and reuse

LocalBroker.setup() (called by TaskWorker.setup_broker() and by the plugin's broker-setup thread) does, under a per-process _server_lock:

  1. If this process already spawned a live server, return.
  2. Probe the address. If something answers as a fastpluggy broker with our authkey, reuse it (_try_reuse_existing_broker). The server records its owner pid, so a reuse is logged once as either a live sibling (normal: e.g. a worker with --beat plus the web server in one container) or an orphan from a previous run (#15).
  3. Otherwise spawn fp-brokerd (fork on Linux, spawn on Windows), register an atexit hook that terminates it, and wait until it answers.

Something listening on the port that does not answer within FP_BROKER_PROBE_TIMEOUT (default 5 s), or answers with another authkey, raises RuntimeError instead of hanging. BaseManager.connect() cannot be bounded from outside, so every readiness check goes through a disposable, timeout-bounded probe first (#17).

A forked child of the process that spawned the server does not count as the owner (_our_server_is_running), so it takes the reuse path instead of crashing on is_alive() (uvicorn/gunicorn with several workers).

Connection and reconnect

  • The proxy connects lazily on first use (_ensure_connected, retries/delay, default 5 × 0.5 s).
  • Each call that fails with BrokenPipeError, EOFError, ConnectionResetError or OSError drops the shared proxy and retries once on a fresh connection (_reconnect_once).
  • LocalBroker.close() frees the process's connection on shutdown.

Limits

  • Single host only, and the queue dies with fp-brokerd: queued messages are lost on restart.
  • The server lives as long as the process that spawned it. If that process exits, every sibling loses the broker until one of them calls setup() again.
  • Orphaned in-flight messages are handled by the reaper, see orphan-recovery.md.

Usage

from fastpluggy_plugin.tasks_worker import TaskWorker

# At application startup, before the web server; returns the broker instance.
broker = TaskWorker.setup_broker(topic_settings={"ocr": {"global_concurrency": 1}})

fastpluggy_plugin.tasks_worker.broker.factory.get_broker() builds a client from the settings without starting anything.

Tests

tests/brokers/test_local*.py cover the paths above: multi-process sharing (test_local_mp.py), reuse vs wrong authkey (test_local_port_collision.py), a wedged peer (test_local.py::test_setup_does_not_hang_on_a_wedged_peer), reuse reporting (test_local_reuse_reporting.py), and races (test_local_race.py).