laantungir/nostr_core_lib
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add per-relay EOSE timeout fallback and robust subscription EOSE completion

Browse Source
This commit is contained in:
Laan Tungir
2026-04-07 07:41:50 -04:00
parent 2e7eacc02e
commit f478a780d7
3 changed files with 102 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
VERSION
+1 -1
View File
@@ -1 +1 @@
0.5.13
0.5.14
nostr_core/core_relay_pool.c
+99 -26
View File
@@ -178,6 +178,9 @@ struct nostr_pool_subscription {
// Per-relay timeout tracking
time_t* relay_last_activity; // Last activity time per relay
// Guard to ensure on_eose is emitted only once per subscription
int eose_callback_fired;
};
struct nostr_relay_pool {
@@ -957,9 +960,9 @@ int nostr_pool_subscription_close(nostr_pool_subscription_t* subscription) {
if (!subscription || subscription->closed) {
return NOSTR_ERROR_INVALID_INPUT;
}
subscription->closed = 1;
// Send CLOSE messages to all relays
for (int i = 0; i < subscription->relay_count; i++) {
relay_connection_t* relay = find_relay_by_url(subscription->pool, subscription->relay_urls[i]);
@@ -967,7 +970,7 @@ int nostr_pool_subscription_close(nostr_pool_subscription_t* subscription) {
nostr_relay_send_close(relay->ws_client, subscription->subscription_id);
}
}
// Remove from pool
nostr_relay_pool_t* pool = subscription->pool;
for (int i = 0; i < pool->subscription_count; i++) {
@@ -980,7 +983,7 @@ int nostr_pool_subscription_close(nostr_pool_subscription_t* subscription) {
break;
}
}
// Cleanup subscription
cJSON_Delete(subscription->filter);
for (int i = 0; i < subscription->relay_count; i++) {
@@ -988,11 +991,55 @@ int nostr_pool_subscription_close(nostr_pool_subscription_t* subscription) {
}
free(subscription->relay_urls);
free(subscription->eose_received);
free(subscription->relay_last_activity);
if (subscription->collected_events) {
for (int i = 0; i < subscription->collected_event_count; i++) {
cJSON_Delete(subscription->collected_events[i]);
}
free(subscription->collected_events);
}
free(subscription);
return NOSTR_SUCCESS;
}
static int maybe_complete_subscription_eose(nostr_pool_subscription_t* sub) {
if (!sub || sub->closed || sub->eose_callback_fired) {
return 0;
}
int all_eose = 1;
for (int j = 0; j < sub->relay_count; j++) {
if (!sub->eose_received[j]) {
all_eose = 0;
break;
}
}
if (!all_eose) {
return 0;
}
sub->eose_callback_fired = 1;
if (sub->on_eose) {
// FIRST mode: no events collected, pass NULL/0
if (sub->result_mode == NOSTR_POOL_EOSE_FIRST) {
sub->on_eose(NULL, 0, sub->user_data);
} else {
// FULL_SET or MOST_RECENT: pass collected events
sub->on_eose(sub->collected_events, sub->collected_event_count, sub->user_data);
}
}
if (sub->close_on_eose && !sub->closed) {
nostr_pool_subscription_close(sub);
return 1;
}
return 0;
}
// Event processing
static void process_relay_message(nostr_relay_pool_t* pool, relay_connection_t* relay, const char* message) {
if (!pool || !relay || !message) {
@@ -1032,6 +1079,15 @@ static void process_relay_message(nostr_relay_pool_t* pool, relay_connection_t*
}
if (sub) {
// Track per-relay activity for timeout fallback.
time_t now = time(NULL);
for (int j = 0; j < sub->relay_count; j++) {
if (strcmp(sub->relay_urls[j], relay->url) == 0) {
sub->relay_last_activity[j] = now;
break;
}
}
// Check for duplicate (per-subscription deduplication)
int is_duplicate = 0;
if (sub->enable_deduplication) {
@@ -1099,32 +1155,15 @@ static void process_relay_message(nostr_relay_pool_t* pool, relay_connection_t*
}
}
// Check if all relays have sent EOSE
int all_eose = 1;
// Keep activity updated on explicit EOSE from this relay
for (int j = 0; j < sub->relay_count; j++) {
if (!sub->eose_received[j]) {
all_eose = 0;
if (strcmp(sub->relay_urls[j], relay->url) == 0) {
sub->relay_last_activity[j] = time(NULL);
break;
}
}
if (all_eose) {
if (sub->on_eose) {
// Pass collected events based on result mode
if (sub->result_mode == NOSTR_POOL_EOSE_FIRST) {
// FIRST mode: no events collected, pass NULL/0
sub->on_eose(NULL, 0, sub->user_data);
} else {
// FULL_SET or MOST_RECENT: pass collected events
sub->on_eose(sub->collected_events, sub->collected_event_count, sub->user_data);
}
}
// Auto-close subscription if close_on_eose is enabled
if (sub->close_on_eose && !sub->closed) {
nostr_pool_subscription_close(sub);
}
}
(void)maybe_complete_subscription_eose(sub);
break;
}
}
@@ -2018,5 +2057,39 @@ int nostr_relay_pool_poll(nostr_relay_pool_t* pool, int timeout_ms) {
}
}
// Timeout fallback for relays that never emit EOSE:
// if no activity from a relay for relay_timeout_seconds, treat it as done.
time_t now = time(NULL);
for (int i = 0; i < pool->subscription_count; i++) {
nostr_pool_subscription_t* sub = pool->subscriptions[i];
if (!sub || sub->closed || sub->eose_callback_fired) {
continue;
}
if (sub->relay_timeout_seconds <= 0) {
continue;
}
int changed = 0;
for (int j = 0; j < sub->relay_count; j++) {
if (sub->eose_received[j]) {
continue;
}
if (sub->relay_last_activity &&
now - sub->relay_last_activity[j] >= sub->relay_timeout_seconds) {
sub->eose_received[j] = 1;
changed = 1;
}
}
if (changed) {
if (maybe_complete_subscription_eose(sub)) {
// Subscription list may have shifted due to auto-close.
i--;
}
}
}
return events_processed;
}
nostr_core/nostr_core.h
+2 -2
View File
@@ -2,10 +2,10 @@
#define NOSTR_CORE_H
// Version information (auto-updated by increment_and_push.sh)
#define VERSION "v0.5.13"
#define VERSION "v0.5.14"
#define VERSION_MAJOR 0
#define VERSION_MINOR 5
#define VERSION_PATCH 13
#define VERSION_PATCH 14
/*
* NOSTR Core Library - Complete API Reference