diff --git a/build.sh b/build.sh index c26cb78b..c4f8ddfd 100755 --- a/build.sh +++ b/build.sh @@ -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 ' ' ',')" diff --git a/nostr_core/core_relay_pool.c b/nostr_core/core_relay_pool.c index ea3691fe..05e8cbe1 100644 --- a/nostr_core/core_relay_pool.c +++ b/nostr_core/core_relay_pool.c @@ -19,6 +19,7 @@ #include #include #include +#include // 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++) { diff --git a/tests/async_publish_test b/tests/async_publish_test index b8458a77..2afa607c 100755 Binary files a/tests/async_publish_test and b/tests/async_publish_test differ diff --git a/tests/backward_compat_test b/tests/backward_compat_test index b2c05b76..d02d184c 100755 Binary files a/tests/backward_compat_test and b/tests/backward_compat_test differ diff --git a/tests/nip42_pool_test b/tests/nip42_pool_test index a23f8637..22ff5cce 100755 Binary files a/tests/nip42_pool_test and b/tests/nip42_pool_test differ diff --git a/tests/relay_synchronous_test b/tests/relay_synchronous_test index 17f1eeaa..b002f423 100755 Binary files a/tests/relay_synchronous_test and b/tests/relay_synchronous_test differ diff --git a/tests/repeated_sync_query_test b/tests/repeated_sync_query_test new file mode 100755 index 00000000..249137df Binary files /dev/null and b/tests/repeated_sync_query_test differ diff --git a/tests/repeated_sync_query_test.c b/tests/repeated_sync_query_test.c new file mode 100644 index 00000000..32be5ae8 --- /dev/null +++ b/tests/repeated_sync_query_test.c @@ -0,0 +1,141 @@ +#define _POSIX_C_SOURCE 200809L + +#include "../nostr_core/nostr_core.h" +#include "../cjson/cJSON.h" + +#include +#include +#include + +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; +} diff --git a/tests/simple_async_test b/tests/simple_async_test index f9b3b75d..fdf36618 100755 Binary files a/tests/simple_async_test and b/tests/simple_async_test differ