Add query timeouts

deno.json

@@ -26,7 +26,7 @@
     "@hono/hono": "jsr:@hono/hono@^4.4.6",
     "@isaacs/ttlcache": "npm:@isaacs/ttlcache@^1.4.1",
     "@noble/secp256k1": "npm:@noble/secp256k1@^2.0.0",
-    "@nostrify/nostrify": "jsr:@nostrify/nostrify@^0.23.3",
+    "@nostrify/nostrify": "jsr:@nostrify/nostrify@^0.25.0",
     "@scure/base": "npm:@scure/base@^1.1.6",
     "@sentry/deno": "https://deno.land/x/sentry@7.112.2/index.mjs",
     "@soapbox/kysely-deno-sqlite": "jsr:@soapbox/kysely-deno-sqlite@^2.1.0",

deno.lock

@@ -12,7 +12,7 @@
       "jsr:@nostrify/nostrify@^0.22.1": "jsr:@nostrify/nostrify@0.22.5",
       "jsr:@nostrify/nostrify@^0.22.4": "jsr:@nostrify/nostrify@0.22.4",
       "jsr:@nostrify/nostrify@^0.22.5": "jsr:@nostrify/nostrify@0.22.5",
-      "jsr:@nostrify/nostrify@^0.23.3": "jsr:@nostrify/nostrify@0.23.3",
+      "jsr:@nostrify/nostrify@^0.25.0": "jsr:@nostrify/nostrify@0.25.0",
       "jsr:@soapbox/kysely-deno-sqlite@^2.1.0": "jsr:@soapbox/kysely-deno-sqlite@2.2.0",
       "jsr:@soapbox/stickynotes@^0.4.0": "jsr:@soapbox/stickynotes@0.4.0",
       "jsr:@std/assert@^0.217.0": "jsr:@std/assert@0.217.0",
@@ -136,8 +136,8 @@
           "npm:zod@^3.23.8"
         ]
       },
-      "@nostrify/nostrify@0.23.3": {
+      "@nostrify/nostrify@0.25.0": {
-        "integrity": "868b10dd094801e28f4982ef9815f0d43f2a807b6f8ad291c78ecb3eb291605a",
+        "integrity": "98f26f44e95ac87fc91b3f3809d38432e1a7f6aebf10380b2554b6f9526313c6",
         "dependencies": [
           "jsr:@std/encoding@^0.224.1",
           "npm:@scure/base@^1.1.6",
@@ -1420,7 +1420,7 @@
       "jsr:@bradenmacdonald/s3-lite-client@^0.7.4",
       "jsr:@db/sqlite@^0.11.1",
       "jsr:@hono/hono@^4.4.6",
-      "jsr:@nostrify/nostrify@^0.23.3",
+      "jsr:@nostrify/nostrify@^0.25.0",
       "jsr:@soapbox/kysely-deno-sqlite@^2.1.0",
       "jsr:@soapbox/stickynotes@^0.4.0",
       "jsr:@std/assert@^0.225.1",

src/controllers/nostr/relay.ts

@@ -73,11 +73,15 @@ function connectStream(socket: WebSocket) {
     const pubsub = await Storages.pubsub();
 
     try {
-      for (const event of await store.query(filters, { limit: FILTER_LIMIT })) {
+      for (const event of await store.query(filters, { limit: FILTER_LIMIT, timeout: 500 })) {
         send(['EVENT', subId, event]);
       }
     } catch (e) {
-      send(['CLOSED', subId, e.message]);
+      if (e instanceof RelayError) {
+        send(['CLOSED', subId, e.message]);
+      } else {
+        send(['CLOSED', subId, 'error: something went wrong']);
+      }
       controllers.delete(subId);
       return;
     }
@@ -124,7 +128,7 @@ function connectStream(socket: WebSocket) {
   /** Handle COUNT. Return the number of events matching the filters. */
   async function handleCount([_, subId, ...filters]: NostrClientCOUNT): Promise<void> {
     const store = await Storages.db();
-    const { count } = await store.count(filters);
+    const { count } = await store.count(filters, { timeout: 500 });
     send(['COUNT', subId, { count, approximate: false }]);
   }
 

src/storages/EventsDB.ts

@@ -51,7 +51,7 @@ class EventsDB implements NStore {
   }
 
   /** Insert an event (and its tags) into the database. */
-  async event(event: NostrEvent, _opts?: { signal?: AbortSignal }): Promise<void> {
+  async event(event: NostrEvent, opts?: { signal?: AbortSignal; timeout?: number }): Promise<void> {
     event = purifyEvent(event);
     this.console.debug('EVENT', JSON.stringify(event));
     dbEventCounter.inc({ kind: event.kind });
@@ -63,7 +63,7 @@ class EventsDB implements NStore {
     await this.deleteEventsAdmin(event);
 
     try {
-      await this.store.event(event);
+      await this.store.event(event, { timeout: opts?.timeout ?? 3000 });
     } catch (e) {
       if (e.message === 'Cannot add a deleted event') {
         throw new RelayError('blocked', 'event deleted by user');
@@ -137,7 +137,10 @@ class EventsDB implements NStore {
   }
 
   /** Get events for filters from the database. */
-  async query(filters: NostrFilter[], opts: { signal?: AbortSignal; limit?: number } = {}): Promise<NostrEvent[]> {
+  async query(
+    filters: NostrFilter[],
+    opts: { signal?: AbortSignal; timeout?: number; limit?: number } = {},
+  ): Promise<NostrEvent[]> {
     filters = await this.expandFilters(filters);
     dbQueryCounter.inc();
 
@@ -160,28 +163,28 @@ class EventsDB implements NStore {
 
     this.console.debug('REQ', JSON.stringify(filters));
 
-    return this.store.query(filters, opts);
+    return this.store.query(filters, { timeout: opts.timeout ?? 3000 });
   }
 
   /** Delete events based on filters from the database. */
-  async remove(filters: NostrFilter[], _opts?: { signal?: AbortSignal }): Promise<void> {
+  async remove(filters: NostrFilter[], opts?: { signal?: AbortSignal; timeout?: number }): Promise<void> {
     if (!filters.length) return Promise.resolve();
     this.console.debug('DELETE', JSON.stringify(filters));
 
-    return this.store.remove(filters);
+    return this.store.remove(filters, opts);
   }
 
   /** Get number of events that would be returned by filters. */
   async count(
     filters: NostrFilter[],
-    opts: { signal?: AbortSignal } = {},
+    opts: { signal?: AbortSignal; timeout?: number } = {},
   ): Promise<{ count: number; approximate: boolean }> {
     if (opts.signal?.aborted) return Promise.reject(abortError());
     if (!filters.length) return Promise.resolve({ count: 0, approximate: false });
 
     this.console.debug('COUNT', JSON.stringify(filters));
 
-    return this.store.count(filters);
+    return this.store.count(filters, { timeout: opts.timeout ?? 1000 });
   }
 
   /** Return only the tags that should be indexed. */