laantungir/c-relay
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 10 Wiki Activity

v1.2.6 - Add MAX_MESSAGE_QUEUE_SIZE=500 limit to prevent OOM from unbounded write queue; fix wasted buf allocation in broadcast

Browse Source
This commit is contained in:
Your Name
2026-02-23 09:12:09 -04:00
parent 1adabdbc4e
commit e94b1a81e3
5 changed files with 10020 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9970
crelay.log Normal file
View File

File diff suppressed because it is too large Load Diff

4
src/main.h
View File

@@ -13,8 +13,8 @@
// Using CRELAY_ prefix to avoid conflicts with nostr_core_lib VERSION macros
#define CRELAY_VERSION_MAJOR 1
#define CRELAY_VERSION_MINOR 2
#define CRELAY_VERSION_PATCH 5
#define CRELAY_VERSION "v1.2.5"
#define CRELAY_VERSION_PATCH 6
#define CRELAY_VERSION "v1.2.6"
// Relay metadata (authoritative source for NIP-11 information)
#define RELAY_NAME "C-Relay"

70
src/subscriptions.c
View File

@@ -873,30 +873,38 @@ int broadcast_event_to_subscriptions(cJSON* event) {
// Second pass: send messages without holding lock
temp_sub_t* current_temp = matching_subs;
while (current_temp) {
// Create EVENT message for this subscription
cJSON* event_msg = cJSON_CreateArray();
cJSON_AddItemToArray(event_msg, cJSON_CreateString("EVENT"));
cJSON_AddItemToArray(event_msg, cJSON_CreateString(current_temp->id));
cJSON_AddItemToArray(event_msg, cJSON_Duplicate(event, 1));
char* msg_str = cJSON_Print(event_msg);
if (msg_str) {
size_t msg_len = strlen(msg_str);
unsigned char* buf = malloc(LWS_PRE + msg_len);
if (buf) {
memcpy(buf + LWS_PRE, msg_str, msg_len);
// DEBUG: Log WebSocket frame details before sending
DEBUG_TRACE("WS_FRAME_SEND: type=EVENT sub=%s len=%zu data=%.100s%s",
current_temp->id,
msg_len,
msg_str,
msg_len > 100 ? "..." : "");
// Serialize event once per subscription using pre-serialized event_json if available,
// otherwise fall back to cJSON serialization.
// Format: ["EVENT","<sub_id>",<event_json>]
cJSON* event_id_obj = cJSON_GetObjectItem(event, "id");
const char* event_json_str = NULL;
char* event_json_allocated = NULL;
// Try to get pre-serialized event_json field first (fast path)
cJSON* event_json_field = cJSON_GetObjectItem(event, "event_json");
if (event_json_field && cJSON_IsString(event_json_field)) {
event_json_str = cJSON_GetStringValue(event_json_field);
} else {
// Fall back to serializing the event
event_json_allocated = cJSON_PrintUnformatted(event);
event_json_str = event_json_allocated;
}
if (event_json_str) {
size_t sub_id_len = strlen(current_temp->id);
size_t event_json_len = strlen(event_json_str);
// ["EVENT","<sub_id>",<event_json>]
size_t msg_len = 10 + sub_id_len + 3 + event_json_len + 1;
char* msg_str = malloc(msg_len + 1);
if (msg_str) {
snprintf(msg_str, msg_len + 1, "[\"EVENT\",\"%s\",%s]", current_temp->id, event_json_str);
size_t actual_len = strlen(msg_str);
DEBUG_TRACE("WS_FRAME_SEND: type=EVENT sub=%s len=%zu", current_temp->id, actual_len);
// Queue message for proper libwebsockets pattern
struct per_session_data* pss = (struct per_session_data*)lws_wsi_user(current_temp->wsi);
if (queue_message(current_temp->wsi, pss, msg_str, msg_len, LWS_WRITE_TEXT) == 0) {
// Message queued successfully
if (queue_message(current_temp->wsi, pss, msg_str, actual_len, LWS_WRITE_TEXT) == 0) {
broadcasts++;
// Update events sent counter for this subscription
@@ -905,30 +913,22 @@ int broadcast_event_to_subscriptions(cJSON* event) {
while (update_sub) {
if (update_sub->wsi == current_temp->wsi &&
strcmp(update_sub->id, current_temp->id) == 0 &&
update_sub->active) { // Add active check to prevent use-after-free
update_sub->active) {
update_sub->events_sent++;
break;
}
update_sub = update_sub->next;
}
pthread_mutex_unlock(&g_subscription_manager.subscriptions_lock);
// Log event broadcast to database (optional - can be disabled for performance)
// NOTE: event_broadcasts table removed due to FOREIGN KEY constraint issues
// cJSON* event_id_obj = cJSON_GetObjectItem(event, "id");
// if (event_id_obj && cJSON_IsString(event_id_obj)) {
// log_event_broadcast(cJSON_GetStringValue(event_id_obj), current_temp->id, current_temp->client_ip);
// }
} else {
DEBUG_ERROR("Failed to queue EVENT message for sub=%s", current_temp->id);
}
free(buf);
free(msg_str);
}
free(msg_str);
}
cJSON_Delete(event_msg);
if (event_json_allocated) {
free(event_json_allocated);
}
current_temp = current_temp->next;
}

8
src/websockets.c
View File

@@ -135,6 +135,14 @@ int queue_message(struct lws* wsi, struct per_session_data* pss, const char* mes
return -1;
}
// Drop message if queue is full to prevent unbounded memory growth under load.
// Slow or disconnected clients should not cause the relay to OOM.
if (pss->message_queue_count >= MAX_MESSAGE_QUEUE_SIZE) {
DEBUG_WARN("queue_message: queue full (%d), dropping message for slow/disconnected client",
pss->message_queue_count);
return -1;
}
// Allocate message queue node
struct message_queue_node* node = malloc(sizeof(struct message_queue_node));
if (!node) {

5
src/websockets.h
View File

@@ -19,6 +19,11 @@
#define MALFORMED_REQUEST_BLOCK_DURATION 3600 // 1 hour in seconds
#define RATE_LIMIT_CLEANUP_INTERVAL 300 // 5 minutes
// Maximum number of messages allowed in a per-client write queue.
// When exceeded, new messages are dropped to prevent unbounded memory growth
// under high-traffic or slow-client conditions.
#define MAX_MESSAGE_QUEUE_SIZE 500
// Filter validation constants
#define MAX_FILTERS_PER_REQUEST 10
#define MAX_AUTHORS_PER_FILTER 1000