Files

T

Laan Tungir aedd22436e v2.1.16 - Implemented api-worker timer + LISTEN/NOTIFY monitoring (Phases 1,2,5) and fixed thread-connection binding bug that stopped dashboard updates

2026-07-25 10:07:08 -04:00

13 KiB

Raw Blame History

API Upgrade Implementation Plan — Phases 1, 2, and 5

Goal: finish the remaining work from plans/api_upgrade_plan.md:

Phase 1 (design fix): convert the api-worker thread from job-queue-driven to timer + subscriber-check driven.
Phase 2: remove the per-event and per-subscription monitoring triggers so monitoring no longer runs on the event-processing path.
Phase 5: add PostgreSQL LISTEN/NOTIFY so the worker wakes on real data changes instead of pure polling.

Target test relay: port 7777 (local).

Current State (verified in code)

api_worker_main() (src/api.c:172) blocks on api_worker_pop_job_blocking() and reacts to three job types: API_WORK_JOB_EVENT_STORED, API_WORK_JOB_SUBSCRIPTION_CHANGE, API_WORK_JOB_STATUS_POST.
Monitoring hooks monitoring_on_event_stored() (src/api.c:752) and monitoring_on_subscription_change() (src/api.c:759) are called from src/main.c:1169 and src/subscriptions.c:483, src/subscriptions.c:536. They enqueue jobs; sync fallback runs if the worker is down.
Throttling lives in monitoring_on_event_stored_sync() / monitoring_on_subscription_change_sync() (src/api.c:711, src/api.c:731).
Round-robin selection in generate_round_robin_monitoring_event() (src/api.c:571); per-type build/sign/broadcast in generate_monitoring_event_for_type() (src/api.c:621).
has_subscriptions_for_kind(int) (src/subscriptions.c:980) checks only subscriptions with an explicit kinds filter — it does not consider the no_kind_filter_subs list, so it alone is insufficient for a "is anyone listening to kind 24567" check.
Worker DB connection is a PGconn* opened via db_open_worker_connection() (src/db_ops_postgres.c:226). Schema is embedded in src/pg_schema.h (version "4") and mirrored in src/pg_schema.sql.
No LISTEN/NOTIFY/pg_notify/PQnotifies usage exists anywhere in src/.

Design

Phase 1 + 2 + 5 combined worker loop

The three phases are coupled, so they are implemented together in one rewrite of api_worker_main() and its helpers.

CRITICAL BUG FIX (why the web API isn't getting updates today)

postgres_db_open_worker_connection() (src/db_ops_postgres.c:226) calls postgres_db_set_thread_connection(conn) at line 243 — but g_thread_pg_conn is declared __thread (src/db_ops_postgres.c:45), so the binding is set in the caller's thread (lws-main, which calls start_api_worker), NOT in the api-worker thread. When api_worker_main runs and the monitoring query functions call postgres_db_active_connection(), g_thread_pg_conn is NULL in that thread, so they fall back to the shared global g_pg_conn — concurrent access with the main thread. This causes monitoring queries to fail or interleave, which is why the dashboard stops receiving kind 24567 events.

Fix: inside api_worker_main(), after opening worker_db, call db_set_thread_connection(worker_db) so the __thread pointer is bound in the api-worker thread itself. Call db_clear_thread_connection() before closing the connection on shutdown. This must be done in the new rewrite regardless of the loop design.

api-worker thread:
  1. Open own PG connection.
  2. db_set_thread_connection(worker_db)   // bind __thread pointer in THIS thread
  3. If PG backend: LISTEN event_stored on this connection.
  4. Loop while running:
     a. Determine if anyone is listening to kind 24567
        (has_subscriptions_for_kind(24567) OR no_kind_filter_subs non-empty).
     b. If NO subscribers:
        - condvar_timedwait(throttle_sec)  // wakeable for shutdown / STATUS_POST job
        - on wake: process any pending STATUS_POST job, then continue.
        // Zero DB work, zero notify polling.
     c. If subscribers exist:
        - wait_for_notify_or_timeout(throttle_sec):
            select() on { PQsocket, self-pipe read fd } with timeout = throttle_sec.
            - If PQsocket readable: PQconsumeInput + drain PQnotifies.
            - If self-pipe readable: drain (shutdown or STATUS_POST signal).
            - If timeout: fall through.
        - generate_round_robin_monitoring_event()   // one d-tag per tick
        - process any pending STATUS_POST job.
  4. On shutdown: UNLISTEN, close PG connection.

Key properties:

Zero overhead when no subscribers — condvar sleep, no select, no LISTEN polling.
Reactive when subscribers exist — wakes within throttle_sec of an event_stored NOTIFY, but never more than once per throttle_sec (rate-limited by the select timeout + the round-robin cadence already in generate_round_robin_monitoring_event).
STATUS_POST preserved — still enqueued from src/websockets.c:3470; the self-pipe wakes the worker's select so status posts are not delayed by a pending notify wait.
SQLite fallback — db_worker_poll_notify() returns "not supported" on SQLite, so the worker falls back to pure timer behavior (condvar timed wait) — identical to the no-PG path.

Self-pipe for job/shutdown signalling

A self-pipe (or eventfd on Linux) is added to api.c:

api_worker_enqueue_job(STATUS_POST) writes one byte to the pipe → wakes select.
stop_api_worker() sets g_api_worker_running = 0, writes a byte, and broadcasts the condvar (covers both the subscriber and no-subscriber wait paths).

Subscriber-check helper

Add int has_any_subscription_for_kind(int event_kind) in src/subscriptions.c that returns true if has_subscriptions_for_kind(event_kind) OR the no_kind_filter_subs list is non-empty. Export it in src/subscriptions.h. The worker calls has_any_subscription_for_kind(24567).

DB abstraction for LISTEN/NOTIFY

Add to src/db_ops.h / src/db_ops_postgres.c / src/db_ops_sqlite.c:

int db_worker_listen(void* conn, const char* channel); — issues LISTEN <channel>.
int db_worker_poll_notify(void* conn, int timeout_ms); — returns 1 if a notification was consumed, 0 on timeout, -1 on error/unsupported. Uses PQsocket(), select(), PQconsumeInput(), PQnotifies().
SQLite stubs return -1 (unsupported) so the worker falls back to timer mode.

Schema change (Phase 5)

Add to both src/pg_schema.sql and src/pg_schema.h:

CREATE OR REPLACE FUNCTION notify_event_stored() RETURNS trigger AS $$
BEGIN
  PERFORM pg_notify('event_stored', json_build_object(
    'kind', NEW.kind,
    'pubkey', substring(NEW.pubkey, 1, 8)
  )::text);
  RETURN NEW;
END;
$$ LANGUAGE plpgsql;

DROP TRIGGER IF EXISTS trg_notify_event_stored ON events;
CREATE TRIGGER trg_notify_event_stored
  AFTER INSERT ON events
  FOR EACH ROW EXECUTE FUNCTION notify_event_stored();

Bump EMBEDDED_PG_SCHEMA_VERSION from "4" to "5" in src/pg_schema.h and the matching schema_info version write in src/pg_schema.sql. The trigger is idempotent (DROP TRIGGER IF EXISTS + CREATE), so existing databases upgrade automatically on next startup via postgres_db_apply_schema() (src/db_ops_postgres.c:134).

Phase 2 removals

Delete the monitoring_on_event_stored() call at src/main.c:1169 (and its forward decl at src/main.c:459).
Delete the monitoring_on_subscription_change() calls at src/subscriptions.c:483 and src/subscriptions.c:536 (and its forward decl at src/subscriptions.c:58).
Remove from src/api.h: monitoring_on_event_stored, monitoring_on_subscription_change.
Remove from src/api.c: API_WORK_JOB_EVENT_STORED, API_WORK_JOB_SUBSCRIPTION_CHANGE, monitoring_on_event_stored, monitoring_on_subscription_change, monitoring_on_event_stored_sync, monitoring_on_subscription_change_sync. Keep API_WORK_JOB_STATUS_POST, generate_round_robin_monitoring_event, generate_monitoring_event_for_type, generate_event_driven_monitoring/generate_subscription_driven_monitoring (still used by generate_monitoring_event legacy path and tests).
The generate_event_driven_monitoring / generate_subscription_driven_monitoring wrappers become unused by the worker but remain as public helpers for any external/test callers; the worker calls generate_round_robin_monitoring_event() directly.

Thread layout (unchanged from plan target)

c_relay_pg process
├── lws-main           — WebSocket event loop (no monitoring DB work)
├── db-read-1..N       — Async REQ/COUNT queries
├── event-worker       — Async EVENT ingestion (no monitoring hook)
├── db-write-1..N      — Async sub logging, misc writes
└── api-worker         — Timer + LISTEN/NOTIFY driven monitoring (own PG conn)

Mermaid: api-worker state machine

stateDiagram-v2
   [*] --> OpenConn
   OpenConn --> BindThread: db_set_thread_connection
   BindThread --> Listen: PG backend
   BindThread --> Ready: SQLite/no-PG
   Listen --> Ready
    Ready --> CheckSubs
    CheckSubs --> NoSubSleep: no kind-24567 subs
    CheckSubs --> WaitNotify: subs present
    NoSubSleep --> CheckSubs: throttle_sec or wake
    WaitNotify --> Generate: NOTIFY or throttle_sec
    Generate --> CheckSubs: after round-robin tick
    WaitNotify --> NoSubSleep: subs dropped + wake
    Ready --> [*]: stop_api_worker

Implementation Todo List

Add has_any_subscription_for_kind() to subscriptions.c/subscriptions.h (checks kind index + no-kind-filter list).
Add db_worker_listen() and db_worker_poll_notify() to db_ops.h, db_ops_postgres.c (libpq), and db_ops_sqlite.c (stub).
Add notify_event_stored() function + trg_notify_event_stored trigger to pg_schema.sql and pg_schema.h; bump schema version to 5.
Rewrite api_worker_main() in api.c: bind the worker PG connection to the thread via db_set_thread_connection() (fixes the dashboard-not-updating bug), then run the timer + LISTEN/NOTIFY + subscriber-check loop; add self-pipe for STATUS_POST/shutdown wakeups; keep start_api_worker/stop_api_worker/api_worker_process_completions/api_worker_enqueue_status_post signatures.
Remove API_WORK_JOB_EVENT_STORED/API_WORK_JOB_SUBSCRIPTION_CHANGE job types and the monitoring_on_event_stored/monitoring_on_subscription_change functions + sync wrappers from api.c and api.h.
Remove the monitoring_on_event_stored() call from main.c (event storage path).
Remove the monitoring_on_subscription_change() calls from subscriptions.c (subscription create/close paths).
Build with ./make_and_restart_relay.sh on port 7777; fix any compile errors.
Test: subscribe to kind 24567 on ws://localhost:7777 → confirm monitoring events arrive on throttle cadence (this validates the thread-connection fix).
Test: publish a kind-1 event → confirm a NOTIFY-driven monitoring event fires within throttle_sec.
Test: with no kind-24567 subscribers → confirm no monitoring events generated (zero overhead) and event storage latency unaffected.
Run tests/quick_error_tests.sh and tests/subscribe_all.sh against port 7777 to confirm no regressions.

Risks & Mitigations

Self-pipe + condvar mix complexity. Mitigation: the no-subscriber path uses only the condvar; the subscriber path uses select on the self-pipe + PQ socket. The condvar is only used to wake the no-subscriber sleep, so the two wait mechanisms never overlap.
LISTEN on a worker connection that also runs monitoring queries. LISTEN is connection-local and persists; running SELECTs on the same connection does not cancel it. PQconsumeInput must be called before PQnotifies. Mitigation: encapsulate all of this in db_worker_poll_notify().
Schema trigger on every insert adds per-event overhead. pg_notify is cheap (in-memory queue) and the payload is tiny. The existing sync_event_tags_from_events AFTER INSERT trigger already does far more work per row, so this is negligible by comparison.
SQLite has no LISTEN/NOTIFY. Mitigation: db_worker_poll_notify returns -1 on SQLite and the worker falls back to pure timer mode — functionally identical to the no-notify path.
Removing the per-event monitoring hook changes dashboard freshness behavior. Mitigation: with subscribers present, NOTIFY now drives near-real-time updates (better than before); with no subscribers, nothing is generated (matches the plan's "zero overhead" goal). Dashboard already handles event-driven updates via its kind-24567 subscription.

13 KiB Raw Blame History