From 17be9c2b0335cfa75c57fbd28e06ae2e3cd6661c Mon Sep 17 00:00:00 2001 From: Laan Tungir Date: Wed, 1 Apr 2026 09:38:25 -0400 Subject: [PATCH] v2.1.1 - Wire REQ/COUNT/EVENT paths through thread pool sync helpers and stabilize filter limit tests --- relay.pid | 2 +- src/main.c | 74 ++--- src/main.h | 4 +- src/thread_pool.c | 516 ++++++++++++++++++++++++++++++- src/thread_pool.h | 46 +++ src/websockets.c | 10 +- test_results_20260401_093238.log | 18 ++ tests/filter_validation_test.sh | 28 +- 8 files changed, 627 insertions(+), 71 deletions(-) create mode 100644 test_results_20260401_093238.log diff --git a/relay.pid b/relay.pid index 44c68fe..f317e92 100644 --- a/relay.pid +++ b/relay.pid @@ -1 +1 @@ -504823 +513827 diff --git a/src/main.c b/src/main.c index 602a5d6..58126f8 100644 --- a/src/main.c +++ b/src/main.c @@ -677,10 +677,6 @@ int store_event_tags(const char* event_id, cJSON* tags) { // Store event in database int store_event(cJSON* event) { - if (thread_pool_is_running()) { - DEBUG_TRACE("Thread pool scaffold active: store_event() still using synchronous DB path"); - } - if (!g_db || !event) { return -1; } @@ -730,25 +726,30 @@ int store_event(cJSON* event) { return -1; } - int rc = DB_ERROR; - int extended_errcode = 0; - if (db_insert_event_with_json(cJSON_GetStringValue(id), - cJSON_GetStringValue(pubkey), - (long long)cJSON_GetNumberValue(created_at), - (int)cJSON_GetNumberValue(kind), - event_type_to_string(type), - cJSON_GetStringValue(content), - cJSON_GetStringValue(sig), - tags_json, - event_json, - &rc, - &extended_errcode) != 0) { + thread_pool_store_event_payload_t payload; + memset(&payload, 0, sizeof(payload)); + payload.id = (char*)cJSON_GetStringValue(id); + payload.pubkey = (char*)cJSON_GetStringValue(pubkey); + payload.created_at = (long long)cJSON_GetNumberValue(created_at); + payload.kind = (int)cJSON_GetNumberValue(kind); + payload.event_type = (char*)event_type_to_string(type); + payload.content = (char*)cJSON_GetStringValue(content); + payload.sig = (char*)cJSON_GetStringValue(sig); + payload.tags_json = tags_json; + payload.event_json = event_json; + + thread_pool_store_event_result_t tp_result; + memset(&tp_result, 0, sizeof(tp_result)); + if (thread_pool_execute_store_event_sync(&payload, &tp_result) != 0) { DEBUG_ERROR("Failed to execute event insert operation"); free(tags_json); free(event_json); return -1; } + int rc = tp_result.step_rc; + int extended_errcode = tp_result.extended_errcode; + if (rc != DB_DONE) { const char* err_msg = db_last_error(); if (rc != DB_CONSTRAINT) { @@ -901,10 +902,6 @@ static int is_only_kind_99999_request(cJSON* filters) { } int handle_req_message(const char* sub_id, cJSON* filters, struct lws *wsi, struct per_session_data *pss) { - if (thread_pool_is_running()) { - DEBUG_TRACE("Thread pool scaffold active: handle_req_message() still using synchronous DB path"); - } - if (!cJSON_IsArray(filters)) { DEBUG_ERROR("REQ filters is not an array"); return 0; @@ -1358,14 +1355,13 @@ int handle_req_message(const char* sub_id, cJSON* filters, struct lws *wsi, stru struct timespec query_start, query_end; clock_gettime(CLOCK_MONOTONIC, &query_start); - // Execute query and send events - db_stmt_t* stmt; - int rc = db_prepare(sql, &stmt); - if (rc != DB_OK) { + // Execute query through db thread pool helper and send events + thread_pool_req_result_t* req_result = NULL; + if (thread_pool_execute_req_sync(sql, (const char**)bind_params, bind_param_count, 5000, &req_result) != 0 || !req_result) { char error_msg[256]; - snprintf(error_msg, sizeof(error_msg), "Failed to prepare subscription query: %s", db_last_error()); + snprintf(error_msg, sizeof(error_msg), "Failed to execute subscription query: %s", db_last_error()); DEBUG_ERROR(error_msg); - + // Log the failed query so we can see what SQL was generated if (g_debug_level >= DEBUG_LEVEL_DEBUG) { time_t now = time(NULL); @@ -1380,7 +1376,7 @@ int handle_req_message(const char* sub_id, cJSON* filters, struct lws *wsi, stru sql); fflush(stderr); } - + continue; } @@ -1389,38 +1385,32 @@ int handle_req_message(const char* sub_id, cJSON* filters, struct lws *wsi, stru pss->db_queries_executed++; } - // Bind parameters - for (int i = 0; i < bind_param_count; i++) { - db_bind_text_param(stmt, i + 1, bind_params[i]); - } - // Cache config values outside the row loop (performance fix) int expiration_enabled = get_config_bool("expiration_enabled", 1); int filter_responses = get_config_bool("expiration_filter", 1); int row_count = 0; - while (db_step_stmt(stmt) == DB_ROW) { + for (int r = 0; r < req_result->row_count; r++) { + const char* event_json_str = req_result->event_json_rows[r]; row_count++; - + // Track rows returned for abuse detection if (pss) { pss->db_rows_returned++; } - - // Get pre-serialized event JSON (no reconstruction needed!) - const char* event_json_str = db_column_text_value(stmt, 0); + if (!event_json_str) { DEBUG_ERROR("Event has NULL event_json field"); continue; } - + // Parse event JSON only for expiration check cJSON* event = cJSON_Parse(event_json_str); if (!event) { DEBUG_ERROR("Failed to parse event_json from database"); continue; } - + // Check expiration filtering (NIP-40) at application level // (expiration_enabled and filter_responses are cached outside the loop) if (expiration_enabled && filter_responses) { @@ -1431,7 +1421,7 @@ int handle_req_message(const char* sub_id, cJSON* filters, struct lws *wsi, stru continue; } } - + // Build EVENT message using zero-copy path: allocate with LWS_PRE prefix, // write directly, transfer ownership to queue — no memcpy. // Format: ["EVENT","",] @@ -1456,7 +1446,7 @@ int handle_req_message(const char* sub_id, cJSON* filters, struct lws *wsi, stru events_sent++; } - db_finalize_stmt(stmt); + thread_pool_free_req_result(req_result); // Stop query timing and log clock_gettime(CLOCK_MONOTONIC, &query_end); diff --git a/src/main.h b/src/main.h index 789b99a..773b8be 100644 --- a/src/main.h +++ b/src/main.h @@ -13,8 +13,8 @@ // Using CRELAY_ prefix to avoid conflicts with nostr_core_lib VERSION macros #define CRELAY_VERSION_MAJOR 2 #define CRELAY_VERSION_MINOR 1 -#define CRELAY_VERSION_PATCH 0 -#define CRELAY_VERSION "v2.1.0" +#define CRELAY_VERSION_PATCH 1 +#define CRELAY_VERSION "v2.1.1" // Relay metadata (authoritative source for NIP-11 information) #define RELAY_NAME "C-Relay" diff --git a/src/thread_pool.c b/src/thread_pool.c index b9bc966..a2b5430 100644 --- a/src/thread_pool.c +++ b/src/thread_pool.c @@ -2,6 +2,7 @@ #include "thread_pool.h" #include "debug.h" +#include "db_ops.h" #include #include @@ -41,6 +42,86 @@ typedef struct { static thread_pool_state_t g_pool = {0}; +typedef struct { + pthread_mutex_t mutex; + pthread_cond_t cond; + int done; + thread_pool_status_t status; + void* result_data; + size_t result_size; +} thread_pool_wait_ctx_t; + +static void wait_ctx_init(thread_pool_wait_ctx_t* ctx) { + memset(ctx, 0, sizeof(*ctx)); + pthread_mutex_init(&ctx->mutex, NULL); + pthread_cond_init(&ctx->cond, NULL); +} + +static void wait_ctx_destroy(thread_pool_wait_ctx_t* ctx) { + pthread_mutex_destroy(&ctx->mutex); + pthread_cond_destroy(&ctx->cond); +} + +static void wait_ctx_result_cb(const thread_pool_result_t* result, void* user_ctx) { + thread_pool_wait_ctx_t* ctx = (thread_pool_wait_ctx_t*)user_ctx; + if (!ctx || !result) return; + + pthread_mutex_lock(&ctx->mutex); + ctx->status = result->status; + ctx->result_data = result->result_data; + ctx->result_size = result->result_size; + ctx->done = 1; + pthread_cond_signal(&ctx->cond); + pthread_mutex_unlock(&ctx->mutex); +} + +static void wait_ctx_wait(thread_pool_wait_ctx_t* ctx) { + pthread_mutex_lock(&ctx->mutex); + while (!ctx->done) { + pthread_cond_wait(&ctx->cond, &ctx->mutex); + } + pthread_mutex_unlock(&ctx->mutex); +} + +static void free_count_payload(void* p) { + thread_pool_count_payload_t* payload = (thread_pool_count_payload_t*)p; + if (!payload) return; + free(payload->sql); + if (payload->bind_params) { + for (int i = 0; i < payload->bind_param_count; i++) { + free(payload->bind_params[i]); + } + free(payload->bind_params); + } + free(payload); +} + +static void free_req_payload(void* p) { + thread_pool_req_payload_t* payload = (thread_pool_req_payload_t*)p; + if (!payload) return; + free(payload->sql); + if (payload->bind_params) { + for (int i = 0; i < payload->bind_param_count; i++) { + free(payload->bind_params[i]); + } + free(payload->bind_params); + } + free(payload); +} + +static void free_store_event_payload(void* p) { + thread_pool_store_event_payload_t* payload = (thread_pool_store_event_payload_t*)p; + if (!payload) return; + free(payload->id); + free(payload->pubkey); + free(payload->event_type); + free(payload->content); + free(payload->sig); + free(payload->tags_json); + free(payload->event_json); + free(payload); +} + static void queue_init(thread_pool_queue_t* q, int max_size) { memset(q, 0, sizeof(*q)); q->max_size = (max_size > 0) ? max_size : 4096; @@ -112,13 +193,19 @@ static void wake_event_loop(void) { } } -static void complete_job_with_status(thread_pool_job_node_t* node, thread_pool_status_t status, const char* message) { +static void complete_job_with_result(thread_pool_job_node_t* node, + thread_pool_status_t status, + const char* message, + void* result_data, + size_t result_size) { thread_pool_result_t result; memset(&result, 0, sizeof(result)); result.job_id = node->job_id; result.type = node->job.type; result.status = status; result.session = node->job.session; + result.result_data = result_data; + result.result_size = result_size; result.message = message; if (node->job.result_cb) { @@ -132,15 +219,167 @@ static void complete_job_with_status(thread_pool_job_node_t* node, thread_pool_s free(node); } +static void execute_count_job(thread_pool_job_node_t* node) { + thread_pool_count_payload_t* payload = (thread_pool_count_payload_t*)node->job.payload; + if (!payload || !payload->sql) { + complete_job_with_result(node, THREAD_POOL_STATUS_INTERNAL_ERROR, + "COUNT payload missing", NULL, 0); + return; + } + + int count = 0; + if (db_count_with_sql(payload->sql, (const char**)payload->bind_params, + payload->bind_param_count, &count) != 0) { + complete_job_with_result(node, THREAD_POOL_STATUS_INTERNAL_ERROR, + "COUNT query failed", NULL, 0); + return; + } + + thread_pool_count_result_t* out = calloc(1, sizeof(*out)); + if (!out) { + complete_job_with_result(node, THREAD_POOL_STATUS_INTERNAL_ERROR, + "COUNT result allocation failed", NULL, 0); + return; + } + out->count = count; + + complete_job_with_result(node, THREAD_POOL_STATUS_OK, + "COUNT query completed", out, sizeof(*out)); +} + +static void execute_req_job(thread_pool_job_node_t* node) { + thread_pool_req_payload_t* payload = (thread_pool_req_payload_t*)node->job.payload; + if (!payload || !payload->sql) { + complete_job_with_result(node, THREAD_POOL_STATUS_INTERNAL_ERROR, + "REQ payload missing", NULL, 0); + return; + } + + db_stmt_t* stmt = NULL; + if (db_prepare(payload->sql, &stmt) != DB_OK) { + complete_job_with_result(node, THREAD_POOL_STATUS_INTERNAL_ERROR, + "REQ prepare failed", NULL, 0); + return; + } + + for (int i = 0; i < payload->bind_param_count; i++) { + const char* v = (payload->bind_params && payload->bind_params[i]) ? payload->bind_params[i] : ""; + (void)db_bind_text_param(stmt, i + 1, v); + } + + int cap = (payload->row_limit > 0) ? payload->row_limit : 500; + if (cap < 1) cap = 1; + + thread_pool_req_result_t* out = calloc(1, sizeof(*out)); + if (!out) { + db_finalize_stmt(stmt); + complete_job_with_result(node, THREAD_POOL_STATUS_INTERNAL_ERROR, + "REQ result allocation failed", NULL, 0); + return; + } + + out->event_json_rows = calloc((size_t)cap, sizeof(char*)); + if (!out->event_json_rows) { + db_finalize_stmt(stmt); + free(out); + complete_job_with_result(node, THREAD_POOL_STATUS_INTERNAL_ERROR, + "REQ row array allocation failed", NULL, 0); + return; + } + + int rc = DB_OK; + while ((rc = db_step_stmt(stmt)) == DB_ROW) { + if (out->row_count >= cap) break; + const char* row = db_column_text_value(stmt, 0); + if (!row) continue; + + out->event_json_rows[out->row_count] = strdup(row); + if (!out->event_json_rows[out->row_count]) { + break; + } + out->row_count++; + } + + db_finalize_stmt(stmt); + + complete_job_with_result(node, THREAD_POOL_STATUS_OK, + "REQ query completed", out, sizeof(*out)); +} + +static void execute_store_event_job(thread_pool_job_node_t* node) { + thread_pool_store_event_payload_t* payload = (thread_pool_store_event_payload_t*)node->job.payload; + if (!payload || !payload->id || !payload->pubkey || !payload->content || !payload->sig) { + complete_job_with_result(node, THREAD_POOL_STATUS_INTERNAL_ERROR, + "STORE_EVENT payload missing required fields", NULL, 0); + return; + } + + thread_pool_store_event_result_t* out = calloc(1, sizeof(*out)); + if (!out) { + complete_job_with_result(node, THREAD_POOL_STATUS_INTERNAL_ERROR, + "STORE_EVENT result allocation failed", NULL, 0); + return; + } + + int step_rc = DB_ERROR; + int extended_errcode = 0; + if (db_insert_event_with_json(payload->id, + payload->pubkey, + payload->created_at, + payload->kind, + payload->event_type ? payload->event_type : "regular", + payload->content, + payload->sig, + payload->tags_json ? payload->tags_json : "[]", + payload->event_json ? payload->event_json : "{}", + &step_rc, + &extended_errcode) != 0) { + free(out); + complete_job_with_result(node, THREAD_POOL_STATUS_INTERNAL_ERROR, + "STORE_EVENT execution failed", NULL, 0); + return; + } + + out->step_rc = step_rc; + out->extended_errcode = extended_errcode; + complete_job_with_result(node, THREAD_POOL_STATUS_OK, + "STORE_EVENT completed", out, sizeof(*out)); +} + +static void execute_read_job(thread_pool_job_node_t* node) { + switch (node->job.type) { + case THREAD_POOL_JOB_REQ_QUERY: + execute_req_job(node); + break; + case THREAD_POOL_JOB_COUNT_QUERY: + execute_count_job(node); + break; + default: + complete_job_with_result(node, THREAD_POOL_STATUS_NOT_IMPLEMENTED, + "Read job type not implemented", NULL, 0); + break; + } +} + +static void execute_write_job(thread_pool_job_node_t* node) { + switch (node->job.type) { + case THREAD_POOL_JOB_STORE_EVENT: + execute_store_event_job(node); + break; + default: + complete_job_with_result(node, THREAD_POOL_STATUS_NOT_IMPLEMENTED, + "Write job type not implemented", NULL, 0); + break; + } +} + static void* reader_worker_main(void* arg) { (void)arg; while (g_pool.running) { thread_pool_job_node_t* node = queue_pop(&g_pool.read_q); if (!node) break; - // Scaffold only: execution wiring is intentionally deferred. - complete_job_with_status(node, THREAD_POOL_STATUS_NOT_IMPLEMENTED, - "Read worker scaffold active; execution not wired yet"); + execute_read_job(node); } return NULL; } @@ -151,9 +390,7 @@ static void* writer_worker_main(void* arg) { thread_pool_job_node_t* node = queue_pop(&g_pool.write_q); if (!node) break; - // Scaffold only: execution wiring is intentionally deferred. - complete_job_with_status(node, THREAD_POOL_STATUS_NOT_IMPLEMENTED, - "Write worker scaffold active; execution not wired yet"); + execute_write_job(node); } return NULL; } @@ -270,6 +507,271 @@ thread_pool_status_t thread_pool_submit_write(const thread_pool_job_t* job, uint return submit_to_queue(&g_pool.write_q, job, out_job_id); } +int thread_pool_execute_count_sync(const char* sql, const char** bind_params, int bind_param_count, int* out_count) { + if (!sql || !out_count) { + return -1; + } + + if (!thread_pool_is_running()) { + return db_count_with_sql(sql, bind_params, bind_param_count, out_count); + } + + thread_pool_count_payload_t* payload = calloc(1, sizeof(*payload)); + if (!payload) return -1; + + payload->sql = strdup(sql); + payload->bind_param_count = bind_param_count; + if (!payload->sql) { + free_count_payload(payload); + return -1; + } + + if (bind_param_count > 0) { + payload->bind_params = calloc((size_t)bind_param_count, sizeof(char*)); + if (!payload->bind_params) { + free_count_payload(payload); + return -1; + } + for (int i = 0; i < bind_param_count; i++) { + const char* v = (bind_params && bind_params[i]) ? bind_params[i] : ""; + payload->bind_params[i] = strdup(v); + if (!payload->bind_params[i]) { + free_count_payload(payload); + return -1; + } + } + } + + thread_pool_wait_ctx_t wait_ctx; + wait_ctx_init(&wait_ctx); + + thread_pool_job_t job; + memset(&job, 0, sizeof(job)); + job.type = THREAD_POOL_JOB_COUNT_QUERY; + job.payload = payload; + job.payload_free = free_count_payload; + job.result_cb = wait_ctx_result_cb; + job.result_cb_ctx = &wait_ctx; + + thread_pool_status_t submit_rc = thread_pool_submit_read(&job, NULL); + if (submit_rc != THREAD_POOL_STATUS_OK) { + free_count_payload(payload); + wait_ctx_destroy(&wait_ctx); + return -1; + } + + wait_ctx_wait(&wait_ctx); + + int rc = -1; + if (wait_ctx.status == THREAD_POOL_STATUS_OK && wait_ctx.result_data) { + thread_pool_count_result_t* result = (thread_pool_count_result_t*)wait_ctx.result_data; + *out_count = result->count; + free(result); + rc = 0; + } + + wait_ctx_destroy(&wait_ctx); + return rc; +} + +int thread_pool_execute_req_sync(const char* sql, const char** bind_params, int bind_param_count, + int row_limit, thread_pool_req_result_t** out_result) { + if (!sql || !out_result) { + return -1; + } + + if (!thread_pool_is_running()) { + db_stmt_t* stmt = NULL; + if (db_prepare(sql, &stmt) != DB_OK) { + return -1; + } + + for (int i = 0; i < bind_param_count; i++) { + const char* v = (bind_params && bind_params[i]) ? bind_params[i] : ""; + (void)db_bind_text_param(stmt, i + 1, v); + } + + int cap = (row_limit > 0) ? row_limit : 500; + if (cap < 1) cap = 1; + + thread_pool_req_result_t* result = calloc(1, sizeof(*result)); + if (!result) { + db_finalize_stmt(stmt); + return -1; + } + + result->event_json_rows = calloc((size_t)cap, sizeof(char*)); + if (!result->event_json_rows) { + free(result); + db_finalize_stmt(stmt); + return -1; + } + + int step_rc = DB_OK; + while ((step_rc = db_step_stmt(stmt)) == DB_ROW) { + if (result->row_count >= cap) break; + const char* row = db_column_text_value(stmt, 0); + if (!row) continue; + + result->event_json_rows[result->row_count] = strdup(row); + if (!result->event_json_rows[result->row_count]) { + break; + } + result->row_count++; + } + + db_finalize_stmt(stmt); + *out_result = result; + return 0; + } + + thread_pool_req_payload_t* payload = calloc(1, sizeof(*payload)); + if (!payload) return -1; + + payload->sql = strdup(sql); + payload->bind_param_count = bind_param_count; + payload->row_limit = row_limit; + if (!payload->sql) { + free_req_payload(payload); + return -1; + } + + if (bind_param_count > 0) { + payload->bind_params = calloc((size_t)bind_param_count, sizeof(char*)); + if (!payload->bind_params) { + free_req_payload(payload); + return -1; + } + for (int i = 0; i < bind_param_count; i++) { + const char* v = (bind_params && bind_params[i]) ? bind_params[i] : ""; + payload->bind_params[i] = strdup(v); + if (!payload->bind_params[i]) { + free_req_payload(payload); + return -1; + } + } + } + + thread_pool_wait_ctx_t wait_ctx; + wait_ctx_init(&wait_ctx); + + thread_pool_job_t job; + memset(&job, 0, sizeof(job)); + job.type = THREAD_POOL_JOB_REQ_QUERY; + job.payload = payload; + job.payload_free = free_req_payload; + job.result_cb = wait_ctx_result_cb; + job.result_cb_ctx = &wait_ctx; + + thread_pool_status_t submit_rc = thread_pool_submit_read(&job, NULL); + if (submit_rc != THREAD_POOL_STATUS_OK) { + free_req_payload(payload); + wait_ctx_destroy(&wait_ctx); + return -1; + } + + wait_ctx_wait(&wait_ctx); + + int rc = -1; + if (wait_ctx.status == THREAD_POOL_STATUS_OK && wait_ctx.result_data) { + *out_result = (thread_pool_req_result_t*)wait_ctx.result_data; + rc = 0; + } + + wait_ctx_destroy(&wait_ctx); + return rc; +} + +void thread_pool_free_req_result(thread_pool_req_result_t* result) { + if (!result) return; + if (result->event_json_rows) { + for (int i = 0; i < result->row_count; i++) { + free(result->event_json_rows[i]); + } + free(result->event_json_rows); + } + free(result); +} + +int thread_pool_execute_store_event_sync(const thread_pool_store_event_payload_t* payload, + thread_pool_store_event_result_t* out_result) { + if (!payload || !out_result || !payload->id || !payload->pubkey || !payload->content || !payload->sig) { + return -1; + } + + if (!thread_pool_is_running()) { + int step_rc = DB_ERROR; + int extended_errcode = 0; + if (db_insert_event_with_json(payload->id, + payload->pubkey, + payload->created_at, + payload->kind, + payload->event_type ? payload->event_type : "regular", + payload->content, + payload->sig, + payload->tags_json ? payload->tags_json : "[]", + payload->event_json ? payload->event_json : "{}", + &step_rc, + &extended_errcode) != 0) { + return -1; + } + out_result->step_rc = step_rc; + out_result->extended_errcode = extended_errcode; + return 0; + } + + thread_pool_store_event_payload_t* payload_copy = calloc(1, sizeof(*payload_copy)); + if (!payload_copy) return -1; + + payload_copy->id = payload->id ? strdup(payload->id) : NULL; + payload_copy->pubkey = payload->pubkey ? strdup(payload->pubkey) : NULL; + payload_copy->created_at = payload->created_at; + payload_copy->kind = payload->kind; + payload_copy->event_type = payload->event_type ? strdup(payload->event_type) : strdup("regular"); + payload_copy->content = payload->content ? strdup(payload->content) : NULL; + payload_copy->sig = payload->sig ? strdup(payload->sig) : NULL; + payload_copy->tags_json = payload->tags_json ? strdup(payload->tags_json) : strdup("[]"); + payload_copy->event_json = payload->event_json ? strdup(payload->event_json) : strdup("{}"); + + if (!payload_copy->id || !payload_copy->pubkey || !payload_copy->event_type || + !payload_copy->content || !payload_copy->sig || !payload_copy->tags_json || !payload_copy->event_json) { + free_store_event_payload(payload_copy); + return -1; + } + + thread_pool_wait_ctx_t wait_ctx; + wait_ctx_init(&wait_ctx); + + thread_pool_job_t job; + memset(&job, 0, sizeof(job)); + job.type = THREAD_POOL_JOB_STORE_EVENT; + job.payload = payload_copy; + job.payload_free = free_store_event_payload; + job.result_cb = wait_ctx_result_cb; + job.result_cb_ctx = &wait_ctx; + + thread_pool_status_t submit_rc = thread_pool_submit_write(&job, NULL); + if (submit_rc != THREAD_POOL_STATUS_OK) { + free_store_event_payload(payload_copy); + wait_ctx_destroy(&wait_ctx); + return -1; + } + + wait_ctx_wait(&wait_ctx); + + int rc = -1; + if (wait_ctx.status == THREAD_POOL_STATUS_OK && wait_ctx.result_data) { + thread_pool_store_event_result_t* result = (thread_pool_store_event_result_t*)wait_ctx.result_data; + out_result->step_rc = result->step_rc; + out_result->extended_errcode = result->extended_errcode; + free(result); + rc = 0; + } + + wait_ctx_destroy(&wait_ctx); + return rc; +} + __attribute__((constructor)) static void thread_pool_state_init_once(void) { pthread_mutex_init(&g_pool.state_mutex, NULL); diff --git a/src/thread_pool.h b/src/thread_pool.h index 6b652d1..32448cb 100644 --- a/src/thread_pool.h +++ b/src/thread_pool.h @@ -56,6 +56,45 @@ typedef struct { void* wake_loop_ctx; } thread_pool_config_t; +typedef struct { + char* sql; + char** bind_params; + int bind_param_count; +} thread_pool_count_payload_t; + +typedef struct { + char* sql; + char** bind_params; + int bind_param_count; + int row_limit; +} thread_pool_req_payload_t; + +typedef struct { + char* id; + char* pubkey; + long long created_at; + int kind; + char* event_type; + char* content; + char* sig; + char* tags_json; + char* event_json; +} thread_pool_store_event_payload_t; + +typedef struct { + int count; +} thread_pool_count_result_t; + +typedef struct { + char** event_json_rows; + int row_count; +} thread_pool_req_result_t; + +typedef struct { + int step_rc; + int extended_errcode; +} thread_pool_store_event_result_t; + int thread_pool_init(const thread_pool_config_t* config); void thread_pool_shutdown(void); int thread_pool_is_running(void); @@ -63,6 +102,13 @@ int thread_pool_is_running(void); thread_pool_status_t thread_pool_submit_read(const thread_pool_job_t* job, uint64_t* out_job_id); thread_pool_status_t thread_pool_submit_write(const thread_pool_job_t* job, uint64_t* out_job_id); +int thread_pool_execute_count_sync(const char* sql, const char** bind_params, int bind_param_count, int* out_count); +int thread_pool_execute_req_sync(const char* sql, const char** bind_params, int bind_param_count, + int row_limit, thread_pool_req_result_t** out_result); +void thread_pool_free_req_result(thread_pool_req_result_t* result); +int thread_pool_execute_store_event_sync(const thread_pool_store_event_payload_t* payload, + thread_pool_store_event_result_t* out_result); + #ifdef __cplusplus } #endif diff --git a/src/websockets.c b/src/websockets.c index 036b41b..d665035 100644 --- a/src/websockets.c +++ b/src/websockets.c @@ -2867,10 +2867,6 @@ int process_dm_stats_command(cJSON* dm_event, char* error_message, size_t error_ int handle_count_message(const char* sub_id, cJSON* filters, struct lws *wsi, struct per_session_data *pss) { // pss is now used for query tracking, so remove unused warning suppression - if (thread_pool_is_running()) { - DEBUG_TRACE("Thread pool scaffold active: handle_count_message() still using synchronous DB path"); - } - if (!cJSON_IsArray(filters)) { DEBUG_ERROR("COUNT filters is not an array"); return 0; @@ -3132,10 +3128,10 @@ int handle_count_message(const char* sub_id, cJSON* filters, struct lws *wsi, st struct timespec query_start, query_end; clock_gettime(CLOCK_MONOTONIC, &query_start); - // Execute count query + // Execute count query via thread pool helper int filter_count = 0; - if (db_count_with_sql(sql, (const char**)bind_params, bind_param_count, &filter_count) != 0) { - DEBUG_ERROR("Failed to execute COUNT query via db_ops"); + if (thread_pool_execute_count_sync(sql, (const char**)bind_params, bind_param_count, &filter_count) != 0) { + DEBUG_ERROR("Failed to execute COUNT query via db_ops/thread_pool"); continue; } diff --git a/test_results_20260401_093238.log b/test_results_20260401_093238.log new file mode 100644 index 0000000..86fbc0b --- /dev/null +++ b/test_results_20260401_093238.log @@ -0,0 +1,18 @@ +2026-04-01 09:32:38 - ========================================== +2026-04-01 09:32:38 - C-Relay Comprehensive Test Suite Runner +2026-04-01 09:32:38 - ========================================== +2026-04-01 09:32:38 - Relay URL: ws://127.0.0.1:8888 +2026-04-01 09:32:38 - Log file: test_results_20260401_093238.log +2026-04-01 09:32:38 - Report file: test_report_20260401_093238.html +2026-04-01 09:32:38 - +2026-04-01 09:32:38 - Checking relay status at ws://127.0.0.1:8888... +2026-04-01 09:32:38 - \033[0;32m✓ Relay HTTP endpoint is accessible\033[0m +2026-04-01 09:32:38 - +2026-04-01 09:32:38 - Starting comprehensive test execution... +2026-04-01 09:32:38 - +2026-04-01 09:32:38 - \033[0;34m=== SECURITY TEST SUITES ===\033[0m +2026-04-01 09:32:38 - ========================================== +2026-04-01 09:32:38 - Running Test Suite: SQL Injection Tests +2026-04-01 09:32:38 - Description: Comprehensive SQL injection vulnerability testing +2026-04-01 09:32:38 - ========================================== +2026-04-01 09:32:38 - \033[0;31mERROR: Test script sql_injection_tests.sh not found\033[0m diff --git a/tests/filter_validation_test.sh b/tests/filter_validation_test.sh index ad96dab..061971a 100755 --- a/tests/filter_validation_test.sh +++ b/tests/filter_validation_test.sh @@ -8,7 +8,7 @@ set -e # Configuration RELAY_HOST="127.0.0.1" RELAY_PORT="8888" -TEST_TIMEOUT=5 +TEST_TIMEOUT=10 # Colors for output RED='\033[0;31m' @@ -34,7 +34,7 @@ test_websocket_message() { # Send message via websocat and capture response local response - response=$(echo "$message" | timeout $TEST_TIMEOUT websocat -B 1048576 ws://$RELAY_HOST:$RELAY_PORT 2>/dev/null || echo 'CONNECTION_FAILED') + response=$(echo "$message" | timeout $TEST_TIMEOUT websocat -B 1048576 ws://$RELAY_HOST:$RELAY_PORT 2>/dev/null | head -1 || echo 'CONNECTION_FAILED') if [[ "$response" == "CONNECTION_FAILED" ]]; then echo -e "${RED}FAILED${NC} - Could not connect to relay" @@ -42,8 +42,8 @@ test_websocket_message() { return 1 fi - if [[ "$response" == "TIMEOUT" ]]; then - echo -e "${RED}FAILED${NC} - Connection timeout" + if [[ -z "$response" ]]; then + echo -e "${RED}FAILED${NC} - Empty response" FAILED_TESTS=$((FAILED_TESTS + 1)) return 1 fi @@ -119,8 +119,9 @@ test_websocket_message "Invalid author type" '["REQ","sub1",{"authors":[123]}]' # Test 6: Invalid author hex test_websocket_message "Invalid author hex" '["REQ","sub1",{"authors":["invalid-hex"]}]' "error: invalid author hex string" -# Test 7: Too many authors -test_websocket_message "Too many authors" '["REQ","sub1",{"authors":["a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a"]}]' "error: too many authors" +# Test 7: Too many authors (exceed MAX_AUTHORS_PER_FILTER with compact payload) +too_many_authors=$(yes '"a"' | head -n 1001 | paste -sd, -) +test_websocket_message "Too many authors" "[\"REQ\",\"sub1\",{\"authors\":[${too_many_authors}]}]" "error: too many authors" echo echo "=== Testing IDs Validation ===" @@ -131,8 +132,9 @@ test_websocket_message "Invalid ID type" '["REQ","sub1",{"ids":[123]}]' "error: # Test 9: Invalid ID hex test_websocket_message "Invalid ID hex" '["REQ","sub1",{"ids":["invalid-hex"]}]' "error: invalid id hex string" -# Test 10: Too many IDs -test_websocket_message "Too many IDs" '["REQ","sub1",{"ids":["a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a"]}]' "error: too many ids" +# Test 10: Too many IDs (exceed MAX_IDS_PER_FILTER with compact payload) +too_many_ids=$(yes '"a"' | head -n 1001 | paste -sd, -) +test_websocket_message "Too many IDs" "[\"REQ\",\"sub1\",{\"ids\":[${too_many_ids}]}]" "error: too many ids" echo echo "=== Testing Kinds Validation ===" @@ -146,8 +148,9 @@ test_websocket_message "Negative kind" '["REQ","sub1",{"kinds":[-1]}]' "error: i # Test 13: Too large kind test_websocket_message "Too large kind" '["REQ","sub1",{"kinds":[70000]}]' "error: invalid kind value" -# Test 14: Too many kinds -test_websocket_message "Too many kinds" '["REQ","sub1",{"kinds":[1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,21,22,23,24,25,26,27,28,29,30,31,32,33,34,35,36,37,38,39,40,41,42,43,44,45,46,47,48,49,50,51,52]}]' "error: too many kinds" +# Test 14: Too many kinds (exceed MAX_KINDS_PER_FILTER=500) +too_many_kinds=$(seq -s, 1 501) +test_websocket_message "Too many kinds" "[\"REQ\",\"sub1\",{\"kinds\":[${too_many_kinds}]}]" "error: too many kinds" echo echo "=== Testing Timestamp Validation ===" @@ -194,8 +197,9 @@ echo "=== Testing Tag Filter Validation ===" # Test 25: Invalid tag filter type test_websocket_message "Invalid tag filter type" '["REQ","sub1",{"#e":"not-an-array"}]' "error: #e must be an array" -# Test 26: Too many tag values -test_websocket_message "Too many tag values" '["REQ","sub1",{"#e":['$(printf '"a%.0s",' {1..101})'"a"]}]' "error: too many #e values" +# Test 26: Too many tag values (exceed MAX_TAG_VALUES_PER_FILTER=1000) +too_many_tag_values=$(yes '"a"' | head -n 1001 | paste -sd, -) +test_websocket_message "Too many tag values" "[\"REQ\",\"sub1\",{\"#e\":[${too_many_tag_values}]}]" "error: too many #e values" # Test 27: Tag value too long test_websocket_message "Tag value too long" '["REQ","sub1",{"#e":["'$(printf 'a%.0s' {1..1025})'"]}]' "error: #e value too long"