From 0d21dc3255cd384aa43a8892aed4cf1e379f0ee6 Mon Sep 17 00:00:00 2001
From: Doug Hoyte <doug@hcsw.org>
Date: Thu, 20 Jul 2023 17:45:00 -0400
Subject: [PATCH] limit on events that can be processed by a sync

---
 docs/negentropy.md                 |  1 +
 src/apps/mesh/cmd_sync.cpp         |  3 +++
 src/apps/relay/RelayIngester.cpp   |  2 +-
 src/apps/relay/RelayNegentropy.cpp | 14 ++++++++++++++
 src/apps/relay/golpe.yaml          |  4 ++++
 strfry.conf                        |  5 +++++
 6 files changed, 28 insertions(+), 1 deletion(-)

diff --git a/docs/negentropy.md b/docs/negentropy.md
index ac79133..469b6dd 100644
--- a/docs/negentropy.md
+++ b/docs/negentropy.md
@@ -54,6 +54,7 @@ Current reason codes are:
 
 * `RESULTS_TOO_BIG`
   * Relays can optionally reject queries that would require them to process too many records, or records that are too old
+  * The maximum number of records that can be processed can optionally be returned as the 4th element in the response
 * `CLOSED`
   * Because the `NEG-OPEN` queries are stateful, relays may choose to time-out inactive queries to recover memory resources
 * `FILTER_NOT_FOUND`
diff --git a/src/apps/mesh/cmd_sync.cpp b/src/apps/mesh/cmd_sync.cpp
index b13d937..97eb76f 100644
--- a/src/apps/mesh/cmd_sync.cpp
+++ b/src/apps/mesh/cmd_sync.cpp
@@ -149,6 +149,9 @@ void cmd_sync(const std::vector<std::string> &subArgs) {
             } else if (msg.at(0) == "EOSE") {
                 inFlightDown = false;
                 writer.wait();
+            } else if (msg.at(0) == "NEG-ERR") {
+                LE << "Got NEG-ERR response from relay: " << msg;
+                ::exit(1);
             } else {
                 LW << "Unexpected message from relay: " << msg;
             }
diff --git a/src/apps/relay/RelayIngester.cpp b/src/apps/relay/RelayIngester.cpp
index c352d74..b81654d 100644
--- a/src/apps/relay/RelayIngester.cpp
+++ b/src/apps/relay/RelayIngester.cpp
@@ -121,7 +121,7 @@ void RelayServer::ingesterProcessNegentropy(lmdb::txn &txn, Decompressor &decomp
         if (arr.get_array().size() < 5) throw herr("negentropy query missing elements");
 
         NostrFilterGroup filter;
-        auto maxFilterLimit = MAX_U64;
+        auto maxFilterLimit = cfg().relay__negentropy__maxSyncEvents + 1;
 
         if (arr.at(2).is_string()) {
             auto ev = lookupEventById(txn, from_hex(arr.at(2).get_string()));
diff --git a/src/apps/relay/RelayNegentropy.cpp b/src/apps/relay/RelayNegentropy.cpp
index 5aa4bce..2a58410 100644
--- a/src/apps/relay/RelayNegentropy.cpp
+++ b/src/apps/relay/RelayNegentropy.cpp
@@ -80,6 +80,20 @@ void RelayServer::runNegentropy(ThreadPool<MsgNegentropy>::Thread &thr) {
         LI << "[" << sub.connId << "] Negentropy query matched " << view->ne.items.size() << " events in "
            << (hoytech::curr_time_us() - view->startTime) << "us";
 
+        if (view->ne.items.size() > cfg().relay__negentropy__maxSyncEvents) {
+            LI << "[" << sub.connId << "] Negentropy query size exceeeded " << cfg().relay__negentropy__maxSyncEvents;
+
+            sendToConn(sub.connId, tao::json::to_string(tao::json::value::array({
+                "NEG-ERR",
+                sub.subId.str(),
+                "RESULTS_TOO_BIG",
+                cfg().relay__negentropy__maxSyncEvents
+            })));
+
+            views.removeView(sub.connId, sub.subId);
+            return;
+        }
+
         view->ne.seal();
 
         auto resp = view->ne.reconcile(view->initialMsg);
diff --git a/src/apps/relay/golpe.yaml b/src/apps/relay/golpe.yaml
index 6053d5c..bbe5dc9 100644
--- a/src/apps/relay/golpe.yaml
+++ b/src/apps/relay/golpe.yaml
@@ -94,3 +94,7 @@ config:
     desc: negentropy threads: Handle negentropy protocol messages
     default: 2
     noReload: true
+
+  - name: relay__negentropy__maxSyncEvents
+    desc: "Maximum records that sync will process before returning an error"
+    default: 1000000
diff --git a/strfry.conf b/strfry.conf
index 390aa9f..4b04eed 100644
--- a/strfry.conf
+++ b/strfry.conf
@@ -124,4 +124,9 @@ relay {
         # negentropy threads: Handle negentropy protocol messages (restart required)
         negentropy = 2
     }
+
+    negentropy {
+        # Maximum records that sync will process before returning an error
+        maxSyncEvents = 1000000
+    }
 }