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

fix relay pool sync query repeat behavior and add regression test

Browse Source
This commit is contained in:
Your Name
2026-03-16 05:27:23 -04:00
parent adc7aa9dcf
commit 20006df228
9 changed files with 236 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
build.sh
View File

@@ -195,7 +195,7 @@ print_info "Auto-detecting needed NIPs from your source code..."
NEEDED_NIPS=""
if [ -n "$FORCE_NIPS" ]; then
if [ "$FORCE_NIPS" = "all" ]; then
NEEDED_NIPS="001 004 005 006 011 013 017 019 042 044 046 059"
NEEDED_NIPS="001 004 005 006 011 013 017 019 021 042 044 046 059"
print_info "Forced: Building all available NIPs"
else
# Convert comma-separated list to space-separated with 3-digit format
@@ -214,7 +214,7 @@ else
# Check for nostr_core.h (includes everything)
if grep -q '#include[[:space:]]*["\<]nostr_core\.h["\>]' *.c 2>/dev/null; then
print_info "Found #include \"nostr_core.h\" - building all NIPs"
NEEDED_NIPS="001 004 005 006 011 013 019 042 044 046 059"
NEEDED_NIPS="001 004 005 006 011 013 019 021 042 044 046 059"
elif [ -n "$DETECTED" ]; then
NEEDED_NIPS="$DETECTED"
print_success "Auto-detected NIPs: $(echo $NEEDED_NIPS | tr ' ' ',')"

98
nostr_core/core_relay_pool.c
View File

@@ -19,6 +19,7 @@
#include <string.h>
#include <time.h>
#include <unistd.h>
#include <stdarg.h>
// Our production-ready WebSocket implementation
#include "../nostr_websocket/nostr_websocket_tls.h"
@@ -43,6 +44,36 @@
#define NOSTR_POOL_MAX_PENDING_SUBSCRIPTIONS 8 // Max concurrent subscription timings per relay
#define NOSTR_POOL_MAX_PENDING_PUBLISHES 32 // Max concurrent publish operations
static int g_query_sync_debug_cached = -1;
static int query_sync_debug_enabled(void) {
if (g_query_sync_debug_cached >= 0) {
return g_query_sync_debug_cached;
}
const char* env = getenv("NOSTR_POOL_QUERY_SYNC_DEBUG");
if (!env || env[0] == '\0' || strcmp(env, "0") == 0 || strcasecmp(env, "false") == 0) {
g_query_sync_debug_cached = 0;
} else {
g_query_sync_debug_cached = 1;
}
return g_query_sync_debug_cached;
}
static void query_sync_debug_log(const char* format, ...) {
if (!query_sync_debug_enabled() || !format) {
return;
}
va_list args;
va_start(args, format);
fprintf(stderr, "[nostr_core][query_sync] ");
vfprintf(stderr, format, args);
fprintf(stderr, "\n");
va_end(args);
}
// High-resolution timing helper
static double get_current_time_ms(void) {
struct timespec ts;
@@ -1247,6 +1278,11 @@ cJSON** nostr_relay_pool_query_sync(
cJSON** events = NULL;
*event_count = 0;
int events_capacity = 0;
// Query-local event deduplication (do not reuse pool-global seen cache)
char** query_seen_ids = NULL;
int query_seen_count = 0;
int query_seen_capacity = 0;
// Generate unique subscription ID
char subscription_id[NOSTR_POOL_SUBSCRIPTION_ID_SIZE];
@@ -1272,7 +1308,9 @@ cJSON** nostr_relay_pool_query_sync(
add_subscription_timing(relay, subscription_id);
// Send REQ message
if (nostr_relay_send_req(relay->ws_client, subscription_id, filter) < 0) {
int send_rc = nostr_relay_send_req(relay->ws_client, subscription_id, filter);
query_sync_debug_log("send_req relay=%s sub_id=%s rc=%d", relay->url, subscription_id, send_rc);
if (send_rc < 0) {
// Remove timing if send failed
remove_subscription_timing(relay, subscription_id);
connected_count--; // Don't count failed connections
@@ -1281,19 +1319,25 @@ cJSON** nostr_relay_pool_query_sync(
}
if (connected_count == 0) {
free(query_seen_ids);
return NULL;
}
// Wait for responses
time_t start_time = time(NULL);
int eose_count = 0;
while (time(NULL) - start_time < (timeout_ms / 1000) && eose_count < connected_count) {
int timeout_seconds = (timeout_ms + 999) / 1000;
if (timeout_seconds <= 0) timeout_seconds = 1;
query_sync_debug_log("query_loop_start sub_id=%s timeout_ms=%d timeout_seconds=%d connected_relays=%d", subscription_id, timeout_ms, timeout_seconds, connected_count);
while (time(NULL) - start_time < timeout_seconds && eose_count < connected_count) {
for (int i = 0; i < connected_count; i++) {
relay_connection_t* relay = connected_relays[i];
char buffer[65536];
int len = nostr_ws_receive(relay->ws_client, buffer, sizeof(buffer) - 1, 100);
query_sync_debug_log("receive relay=%s len=%d", relay->url, len);
if (len > 0) {
buffer[len] = '\0';
@@ -1308,6 +1352,16 @@ cJSON** nostr_relay_pool_query_sync(
cJSON* parsed = NULL;
if (nostr_parse_relay_message(buffer, &msg_type, &parsed) == 0) {
int handled_for_query = 0;
const char* parsed_sub_id = NULL;
if (cJSON_IsArray(parsed) && cJSON_GetArraySize(parsed) >= 2) {
cJSON* parsed_sub_id_json = cJSON_GetArrayItem(parsed, 1);
if (cJSON_IsString(parsed_sub_id_json)) {
parsed_sub_id = cJSON_GetStringValue(parsed_sub_id_json);
}
}
query_sync_debug_log("parsed relay=%s type=%s sub_id=%s", relay->url, msg_type ? msg_type : "(null)", parsed_sub_id ? parsed_sub_id : "(none)");
if (msg_type && strcmp(msg_type, "EVENT") == 0) {
if (cJSON_IsArray(parsed) && cJSON_GetArraySize(parsed) >= 3) {
@@ -1322,8 +1376,33 @@ cJSON** nostr_relay_pool_query_sync(
if (event_id_json && cJSON_IsString(event_id_json)) {
const char* event_id = cJSON_GetStringValue(event_id_json);
if (!is_event_seen(pool, event_id)) {
mark_event_seen(pool, event_id);
int duplicate_in_query = 0;
for (int q = 0; q < query_seen_count; q++) {
if (query_seen_ids[q] && strcmp(query_seen_ids[q], event_id) == 0) {
duplicate_in_query = 1;
break;
}
}
if (!duplicate_in_query) {
if (query_seen_count >= query_seen_capacity) {
int new_capacity = (query_seen_capacity == 0) ? 16 : query_seen_capacity * 2;
char** grown_seen = realloc(query_seen_ids, (size_t)new_capacity * sizeof(char*));
if (!grown_seen) {
*event_count = 0;
break;
}
query_seen_ids = grown_seen;
query_seen_capacity = new_capacity;
}
query_seen_ids[query_seen_count] = strdup(event_id);
if (!query_seen_ids[query_seen_count]) {
*event_count = 0;
break;
}
query_seen_count++;
relay->stats.events_received++;
if (*event_count >= events_capacity) {
@@ -1347,6 +1426,7 @@ cJSON** nostr_relay_pool_query_sync(
strcmp(cJSON_GetStringValue(sub_id_json), subscription_id) == 0) {
handled_for_query = 1;
eose_count++;
query_sync_debug_log("eose relay=%s sub_id=%s eose_count=%d/%d event_count=%d", relay->url, cJSON_GetStringValue(sub_id_json), eose_count, connected_count, *event_count);
}
}
} else if (msg_type && strcmp(msg_type, "AUTH") == 0) {
@@ -1386,6 +1466,7 @@ cJSON** nostr_relay_pool_query_sync(
if (parsed) cJSON_Delete(parsed);
if (!handled_for_query) {
query_sync_debug_log("forward_to_pool relay=%s type=%s", relay->url, msg_type ? msg_type : "(null)");
process_relay_message(pool, relay, buffer);
}
}
@@ -1393,6 +1474,13 @@ cJSON** nostr_relay_pool_query_sync(
}
}
}
int elapsed_sec = (int)(time(NULL) - start_time);
if (eose_count >= connected_count) {
query_sync_debug_log("query_loop_end reason=all_eose sub_id=%s elapsed_s=%d eose=%d/%d events=%d", subscription_id, elapsed_sec, eose_count, connected_count, *event_count);
} else {
query_sync_debug_log("query_loop_end reason=timeout sub_id=%s elapsed_s=%d timeout_s=%d eose=%d/%d events=%d", subscription_id, elapsed_sec, timeout_seconds, eose_count, connected_count, *event_count);
}
// Send CLOSE messages
for (int i = 0; i < connected_count; i++) {

BIN
tests/async_publish_test
View File

Binary file not shown.

BIN
tests/backward_compat_test
View File

Binary file not shown.

BIN
tests/nip42_pool_test
View File

Binary file not shown.

BIN
tests/relay_synchronous_test
View File

Binary file not shown.

BIN
tests/repeated_sync_query_test Executable file
View File

Binary file not shown.

141
tests/repeated_sync_query_test.c Normal file
View File

@@ -0,0 +1,141 @@
#define _POSIX_C_SOURCE 200809L
#include "../nostr_core/nostr_core.h"
#include "../cjson/cJSON.h"
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
typedef struct {
const char** relays;
int relay_count;
const char* pubkey;
int timeout_ms;
} test_config_t;
static cJSON* build_kind10002_filter(const char* pubkey_hex) {
if (!pubkey_hex || pubkey_hex[0] == '\0') {
return NULL;
}
cJSON* filter = cJSON_CreateObject();
cJSON* kinds = cJSON_CreateArray();
cJSON* authors = cJSON_CreateArray();
if (!filter || !kinds || !authors) {
cJSON_Delete(filter);
cJSON_Delete(kinds);
cJSON_Delete(authors);
return NULL;
}
cJSON_AddItemToArray(kinds, cJSON_CreateNumber(10002));
cJSON_AddItemToObject(filter, "kinds", kinds);
cJSON_AddItemToArray(authors, cJSON_CreateString(pubkey_hex));
cJSON_AddItemToObject(filter, "authors", authors);
cJSON_AddItemToObject(filter, "limit", cJSON_CreateNumber(16));
return filter;
}
static void free_events(cJSON** events, int event_count) {
if (!events) return;
for (int i = 0; i < event_count; i++) {
if (events[i]) cJSON_Delete(events[i]);
}
free(events);
}
static void print_event_summary(const char* label, cJSON** events, int event_count) {
printf("%s event_count=%d\n", label, event_count);
for (int i = 0; i < event_count; i++) {
cJSON* id = cJSON_GetObjectItemCaseSensitive(events[i], "id");
cJSON* created_at = cJSON_GetObjectItemCaseSensitive(events[i], "created_at");
const char* id_str = (id && cJSON_IsString(id) && id->valuestring) ? id->valuestring : "(no-id)";
long long ts = (created_at && cJSON_IsNumber(created_at)) ? (long long)created_at->valuedouble : 0;
printf(" [%d] id=%.16s... created_at=%lld\n", i, id_str, ts);
}
}
static int run_query(nostr_relay_pool_t* pool, const test_config_t* cfg, const char* label) {
int event_count = 0;
cJSON* filter = build_kind10002_filter(cfg->pubkey);
if (!filter) {
fprintf(stderr, "[%s] failed to build filter\n", label);
return -1;
}
cJSON** events = nostr_relay_pool_query_sync(
pool,
cfg->relays,
cfg->relay_count,
filter,
&event_count,
cfg->timeout_ms
);
cJSON_Delete(filter);
print_event_summary(label, events, event_count);
free_events(events, event_count);
return event_count;
}
int main(int argc, char** argv) {
const char* pubkey = "52a3e82f7b3743852fbe804cfcbf4db3448115887895247c001f2b50e790acb8";
if (argc > 1 && argv[1] && argv[1][0] != '\0') {
pubkey = argv[1];
}
const char* relays[] = {
"wss://relay.damus.io",
"wss://relay.primal.net"
};
test_config_t cfg = {
.relays = relays,
.relay_count = 2,
.pubkey = pubkey,
.timeout_ms = 5000
};
printf("=== repeated_sync_query_test ===\n");
printf("pubkey=%s\n", cfg.pubkey);
printf("relay[0]=%s\n", cfg.relays[0]);
printf("relay[1]=%s\n", cfg.relays[1]);
printf("timeout_ms=%d\n", cfg.timeout_ms);
printf("Set NOSTR_POOL_QUERY_SYNC_DEBUG=1 for internal query-sync tracing.\n\n");
if (nostr_init() != NOSTR_SUCCESS) {
fprintf(stderr, "nostr_init failed\n");
return 1;
}
nostr_relay_pool_t* pool = nostr_relay_pool_create(NULL);
if (!pool) {
fprintf(stderr, "nostr_relay_pool_create failed\n");
nostr_cleanup();
return 1;
}
int first_count = run_query(pool, &cfg, "first_query");
int second_count = run_query(pool, &cfg, "second_query");
int exit_code = 0;
if (first_count <= 0) {
fprintf(stderr, "FAIL: first query returned no events\n");
exit_code = 2;
} else if (second_count <= 0) {
fprintf(stderr, "FAIL: second query returned no events (regression candidate)\n");
exit_code = 3;
} else {
printf("PASS: both queries returned events\n");
}
nostr_relay_pool_destroy(pool);
nostr_cleanup();
return exit_code;
}

BIN
tests/simple_async_test
View File

Binary file not shown.