c-relay-pg/plans/c_relay_thread_pool_plan.md

c-relay-pg Thread Pool Plan — db_ops Abstraction + SQLite Thread Pool

Overview

This plan covers improvements to c-relay-pg (this repo) — the lean, single-binary, embedded-SQLite Nostr relay. Two phases:

1. Phase 1: Database Abstraction Layer — Consolidate all scattered sqlite3_* calls into a single db_ops.h / db_ops.c module. Pure refactor, no behavior change.
2. Phase 2: SQLite Thread Pool — Add worker threads for concurrent reads, unblocking the lws_service() event loop.

This is the final form of c-relay-pg: an embedded-SQLite relay with clean architecture and non-blocking database access. The abstraction layer also enables a future fork to PostgreSQL — see c-relay-pg-pg plan.

Why This First?

The current architecture has a fundamental bottleneck documented in the database architecture analysis: the single-threaded event loop blocks on every database call. A 672ms REQ query freezes every connected client. The thread pool fixes this without changing the database engine or deployment model.

Metric	Current	After Thread Pool
Event loop blocking	0.1–672ms per query	Near-zero
Concurrent reads	1	4–8
Deployment	Single binary + DB file	Same
Database	SQLite	Same
Code risk	N/A	Low — additive change

Architecture

flowchart TD
    subgraph c-relay-pg - Final Form
        RELAY[c-relay-pg binary] -->|db_ops API| DBOPS[db_ops.c - abstraction layer]
        DBOPS -->|sqlite3 calls| SQLITE[SQLite WAL database]
    end

flowchart TD
    subgraph c-relay-pg with Thread Pool
        LWS[lws_service event loop] -->|REQ arrives| Q[Thread-safe job queue]
        LWS -->|EVENT arrives| WQ[Write queue - single writer]
        Q --> T1[Reader Thread 1 - own sqlite3*]
        Q --> T2[Reader Thread 2 - own sqlite3*]
        Q --> T3[Reader Thread 3 - own sqlite3*]
        Q --> T4[Reader Thread 4 - own sqlite3*]
        WQ --> TW[Writer Thread - own sqlite3*]
        T1 -->|results| CB[Callback to lws event loop]
        T2 -->|results| CB
        T3 -->|results| CB
        T4 -->|results| CB
        TW -->|OK/error| CB
        CB -->|queue_message| LWS
    end

---

Phase 1: Database Abstraction Layer

Goal

Consolidate all 258 scattered sqlite3_* calls across 8 files into a single db_ops.h / db_ops.c module with a backend-agnostic interface. SQLite continues to work — this is a pure refactor.

Files That Currently Touch SQLite Directly

File	sqlite3_* calls	Operations
src/main.c	~80	Event store/retrieve, tag storage, REQ queries, COUNT queries, DB init, schema migration
src/config.c	~60	Config CRUD, auth rules CRUD, WoT sync, relay key storage, startup sequence
src/api.c	~40	Stats queries, monitoring views, admin SQL execution, config management
src/dm_admin.c	~20	Auth rule management, WoT whitelist operations
src/websockets.c	~15	COUNT queries, IP ban stats, connection tracking
src/subscriptions.c	~15	Subscription logging — create/close/disconnect
src/nip009.c	~10	Event deletion — by ID and by address
src/request_validator.c	~8	Blacklist/whitelist checks
src/ip_ban.c	~15	IP ban table CRUD, persistence

Abstraction Layer Design

New files: src/db_ops.h and src/db_ops.c

// src/db_ops.h — Database operations abstraction layer
#ifndef DB_OPS_H
#define DB_OPS_H

#include "../nostr_core_lib/cjson/cJSON.h"

// ============================================================
// Lifecycle
// ============================================================
int db_init(const char* connection_string);  // SQLite: file path, PG: connection string
void db_close(void);
int db_is_available(void);

// ============================================================
// Schema / Migration
// ============================================================
int db_ensure_schema(void);
int db_get_schema_version(void);
int db_execute_raw(const char* sql);  // For schema DDL only

// ============================================================
// Event Operations
// ============================================================
int db_store_event(cJSON* event);
int db_event_exists(const char* event_id);
char* db_get_event_json(const char* event_id);  // Caller must free
int db_delete_event_by_id(const char* event_id, const char* requester_pubkey);
int db_delete_events_by_address(const char* pubkey, int kind, 
                                 const char* d_tag, long before_timestamp);

// ============================================================
// Event Queries - REQ handling
// ============================================================
typedef struct {
    int* kinds;           int kind_count;
    char** authors;       int author_count;
    char** ids;           int id_count;
    char** tag_names;     // Parallel arrays: tag_names[i] has tag_values[i][]
    char*** tag_values;
    int* tag_value_counts;
    int tag_filter_count;
    long since;           // 0 = not set
    long until;           // 0 = not set
    int limit;            // 0 = default 500
    char* search;         // NIP-50 search term, NULL = not set
} db_event_filter_t;

typedef struct db_result db_result_t;

db_result_t* db_query_events(const db_event_filter_t* filter);
const char* db_result_next_json(db_result_t* result);  // Returns event_json, NULL when done
int db_result_row_count(db_result_t* result);
void db_result_free(db_result_t* result);

int db_count_events(const db_event_filter_t* filter);

// ============================================================
// Config Operations
// ============================================================
char* db_get_config(const char* key);  // Caller must free, NULL if not found
int db_set_config(const char* key, const char* value, const char* data_type,
                  const char* description, const char* category, int requires_restart);
int db_update_config(const char* key, const char* value);
int db_config_exists(const char* key);
int db_get_config_count(void);

// ============================================================
// Auth Rules Operations
// ============================================================
int db_add_auth_rule(const char* rule_type, const char* pattern_type, const char* pattern_value);
int db_remove_auth_rule(const char* rule_type, const char* pattern_type, const char* pattern_value);
int db_auth_rule_exists(const char* rule_type, const char* pattern_type, const char* pattern_value);
int db_clear_auth_rules(const char* rule_type);  // NULL = clear all
int db_is_blacklisted(const char* pubkey);
int db_is_whitelisted(const char* pubkey);
int db_has_any_whitelist(void);

// ============================================================
// Relay Key Storage
// ============================================================
int db_store_relay_private_key(const char* privkey_hex);
char* db_get_relay_private_key(void);  // Caller must free

// ============================================================
// Subscription Logging
// ============================================================
int db_log_subscription_created(const char* sub_id, const char* wsi_ptr, 
                                 const char* client_ip, const char* filter_json);
int db_log_subscription_closed(const char* sub_id, const char* client_ip);
int db_log_subscription_disconnected(const char* client_ip);
int db_update_subscription_events_sent(const char* sub_id, int events_sent);
int db_cleanup_orphaned_subscriptions(void);

// ============================================================
// IP Ban Persistence
// ============================================================
int db_ensure_ip_ban_table(void);
int db_load_ip_bans(void* ban_table, int table_size);  // Populates in-memory table
int db_save_ip_bans(const void* ban_table, int table_size);

// ============================================================
// Analytics / Stats - for dashboard
// ============================================================
cJSON* db_get_event_kind_distribution(void);
cJSON* db_get_time_based_stats(void);
cJSON* db_get_top_pubkeys(int limit);
cJSON* db_get_subscription_details(void);
int db_get_total_event_count(void);

// ============================================================
// Admin SQL Query
// ============================================================
cJSON* db_execute_admin_query(const char* sql, char* error_msg, size_t error_size);

#endif // DB_OPS_H

Migration Strategy for Phase 1

The key principle: change the interface, not the behavior. Each function in db_ops.c initially just wraps the existing SQLite calls.

Step-by-step for each file:

1. Create db_ops.h and db_ops.c with the interface above
2. Implement each db_ops function by moving the existing SQLite code from the source files into db_ops.c
3. Replace direct sqlite3_* calls in each source file with db_ops_* calls
4. Remove extern sqlite3* g_db from each file — only db_ops.c knows about g_db
5. Remove #include <sqlite3.h> from each file — only db_ops.c includes it
6. Update Makefile to compile db_ops.c

Order of Migration (by risk, lowest first)

1. src/ip_ban.c — Self-contained, simple CRUD. Good warmup.
2. src/subscriptions.c — Logging only, no critical path.
3. src/nip009.c — Event deletion, small file.
4. src/request_validator.c — Auth checks, small file.
5. src/api.c — Stats/monitoring queries. Larger but read-only.
6. src/dm_admin.c — Auth rules + WoT. Medium complexity.
7. src/config.c — Config CRUD. Large but well-structured.
8. src/websockets.c — COUNT queries, minimal DB usage.
9. src/main.c — Event store/retrieve, REQ queries. The big one — do last.

Testing Phase 1

After Phase 1, the relay should behave identically to before. Run:

• tests/run_all_tests.sh — Full test suite
• tests/performance_benchmarks.sh — Verify no performance regression
• Manual testing with production-like traffic

---

Phase 2: SQLite Thread Pool

Goal

Add a pool of N worker threads, each with its own sqlite3* connection to the same database file. SQLite WAL mode (already enabled) supports concurrent readers. The event loop dispatches database work to the pool and remains responsive.

Key Design Points

1. Read path: REQ queries dispatched to thread pool. Each worker opens its own sqlite3* connection. SQLite WAL allows unlimited concurrent readers.

2. Write path: EVENT inserts go through a single dedicated writer thread. SQLite only allows one writer at a time anyway — this serializes writes cleanly.

3. Result delivery: Worker threads cannot call lws_write() directly (libwebsockets is not thread-safe). Instead, they push results into a per-session message queue and call lws_cancel_service() to wake the event loop, which then drains the queue.

4. Connection lifecycle: Each thread opens its own connection with PRAGMA journal_mode=WAL and PRAGMA busy_timeout=5000.

What Changes in the Codebase

Component	Current	Thread Pool
g_db	Single global connection	One per thread + writer connection
handle_req_message()	Synchronous SQL in callback	Package filter into job, dispatch to pool
store_event()	Synchronous INSERT in callback	Dispatch to writer thread
handle_count_message()	Synchronous COUNT in callback	Dispatch to pool
Result delivery	Direct queue_message()	Worker pushes to queue + lws_cancel_service()
Config reads	Direct sqlite3_prepare on g_db	Can stay synchronous with own connection or cache

New Files

• src/thread_pool.h — Thread pool interface
• src/thread_pool.c — Thread pool implementation (job queue, worker lifecycle, result delivery)

Thread Pool Interface (sketch)

// src/thread_pool.h
#ifndef THREAD_POOL_H
#define THREAD_POOL_H

typedef enum {
    JOB_TYPE_REQ_QUERY,
    JOB_TYPE_COUNT_QUERY,
    JOB_TYPE_STORE_EVENT,
    JOB_TYPE_DELETE_EVENT,
} job_type_t;

typedef struct {
    job_type_t type;
    void* session;              // lws per-session data pointer
    db_event_filter_t filter;   // For REQ/COUNT jobs
    cJSON* event;               // For STORE jobs
    char event_id[65];          // For DELETE jobs
} db_job_t;

int thread_pool_init(int num_readers, const char* db_path);
void thread_pool_shutdown(void);
int thread_pool_submit_read(db_job_t* job);
int thread_pool_submit_write(db_job_t* job);

#endif // THREAD_POOL_H

Performance Characteristics

Metric	Value
Concurrent reads	N readers in parallel (N = thread count, typically 4–8)
Write throughput	Same as current — SQLite serializes writes regardless
Event loop latency	Near-zero — REQ no longer blocks the loop
Max theoretical read throughput	~4–8x current (limited by disk I/O, not CPU)
Memory overhead	~50–100 MB per connection (page cache)
Latency per query	Same as current per-query, but no head-of-line blocking

Limitations

• Write contention: SQLite still allows only ONE writer at a time. With WAL, readers don't block writers and writers don't block readers, but two simultaneous writes will serialize. At the current write rate (~56 events/hour), this is a non-issue.
• Database size: The 2.7 GB index bloat problem remains. Thread pool doesn't fix the schema — it fixes the concurrency.
• Scaling ceiling: Beyond ~8 reader threads, diminishing returns due to disk I/O contention on a single SQLite file.
• Complexity: Need a proper job queue, thread lifecycle management, and careful handling of the lws ↔ worker thread boundary.

---

Risk Mitigation

Risk	Mitigation
Phase 1 introduces bugs	Each file migrated independently, tested after each. Full test suite runs after each file.
Thread pool race conditions	Worker threads only touch their own sqlite3* connection. Result delivery uses a mutex-protected queue. lws_cancel_service() is documented as thread-safe.
Performance regression from abstraction	Abstraction layer is thin wrappers — no extra allocations or copies. Benchmark before/after.
lws thread safety issues	Workers never call lws_write() directly. They push to a queue and wake the event loop. This is the documented pattern for libwebsockets multi-threading.
Rollback needed	Phase 1 is a pure refactor — easy to revert. Phase 2 thread pool is additive — can be disabled with a compile flag.

---

Relationship to c-relay-pg-pg

After Phase 1 (abstraction layer) is complete, the codebase is ready to be forked into a separate c-relay-pg-pg project that replaces the SQLite backend with PostgreSQL. See c-relay-pg-pg plan for details.

The db_ops.h interface is designed to work with any backend:

• SQLite (this plan): db_result_next_json() copies from sqlite3_column_text()
• PostgreSQL (c-relay-pg-pg): db_result_next_json() returns from PQgetvalue()
• LMDB (future possibility): db_result_next_json() returns a zero-copy pointer into mmap'd memory