feat(relay): add maxTotalEventsPerReq to cap aggregate events across filters

src/DBQuery.h

@@ -290,6 +290,8 @@ struct DBQuery : NonCopyable {
     flat_hash_set<uint64_t> sentEventsCurr;
     uint64_t lastWorkChecked = 0;
 
+    uint64_t maxTotalEvents = 0;
+    bool hitMaxTotalEvents = false;
     uint64_t currScanTime = 0;
     uint64_t currScanSaveRestores = 0;
     uint64_t totalTime = 0;
@@ -319,6 +321,7 @@ struct DBQuery : NonCopyable {
                 }
 
                 sentEventsCurr.insert(levId);
+                if (maxTotalEvents > 0 && sentEventsFull.size() >= maxTotalEvents) return true;
                 return sentEventsCurr.size() >= f.limit;
             }, [&](uint64_t approxWork){
                 if (approxWork > lastWorkChecked + 2'000) {
@@ -355,6 +358,12 @@ struct DBQuery : NonCopyable {
 
             currScanTime = 0;
             currScanSaveRestores = 0;
+
+            if (maxTotalEvents > 0 && sentEventsFull.size() >= maxTotalEvents) {
+                LW << "[" << sub.connId << "] REQ='" << sub.subId.sv() << "' hit maxTotalEventsPerReq limit (" << maxTotalEvents << ")";
+                hitMaxTotalEvents = true;
+                break;
+            }
         }
 
         if (logMetrics) {

src/QueryScheduler.h

@@ -6,7 +6,7 @@
 struct QueryScheduler : NonCopyable {
     std::function<void(lmdb::txn &txn, const Subscription &sub, uint64_t levId, std::string_view eventPayload)> onEvent;
     std::function<void(lmdb::txn &txn, const Subscription &sub, const std::vector<uint64_t> &levIds)> onEventBatch;
-    std::function<void(lmdb::txn &txn, Subscription &sub, uint64_t total)> onComplete;
+    std::function<void(lmdb::txn &txn, Subscription &sub, uint64_t total, bool hitMaxTotalEvents)> onComplete;
 
     // If false, then levIds returned to above callbacks can be stale (because they were deleted)
     // If false, then onEvent's eventPayload will always be ""
@@ -33,6 +33,7 @@ struct QueryScheduler : NonCopyable {
         }
 
         DBQuery *q = new DBQuery(sub);
+        q->maxTotalEvents = cfg().relay__maxTotalEventsPerReq;
 
         connQueries.try_emplace(q->sub.subId, q);
         running.push_front(q);
@@ -101,7 +102,7 @@ struct QueryScheduler : NonCopyable {
             auto connId = q->sub.connId;
             removeSub(connId, q->sub.subId);
 
-            if (onComplete) onComplete(txn, q->sub, q->sentEventsFull.size());
+            if (onComplete) onComplete(txn, q->sub, q->sentEventsFull.size(), q->hitMaxTotalEvents);
 
             delete q;
         } else {

src/apps/relay/RelayNegentropy.cpp

@@ -137,7 +137,7 @@ void RelayServer::runNegentropy(ThreadPool<MsgNegentropy>::Thread &thr) {
         }
     };
 
-    queries.onComplete = [&](lmdb::txn &txn, Subscription &sub, uint64_t){
+    queries.onComplete = [&](lmdb::txn &txn, Subscription &sub, uint64_t, bool){
         auto *userView = views.findView(sub.connId, sub.subId);
         if (!userView) return;
 

src/apps/relay/RelayReqWorker.cpp

@@ -21,7 +21,7 @@ void RelayServer::runReqWorker(ThreadPool<MsgReqWorker>::Thread &thr) {
         sendEvent(sub.connId, sub.subId, decodeEventPayload(txn, decomp, eventPayload, nullptr, nullptr));
     };
 
-    queries.onComplete = [&](lmdb::txn &, Subscription &sub, uint64_t total){
+    queries.onComplete = [&](lmdb::txn &, Subscription &sub, uint64_t total, bool hitMaxTotalEvents){
         if (sub.countOnly) {
             bool limited = false;
 
@@ -37,6 +37,11 @@ void RelayServer::runReqWorker(ThreadPool<MsgReqWorker>::Thread &thr) {
             if (limited) countBody["limited"] = true;
 
             sendToConn(sub.connId, tao::json::to_string(tao::json::value::array({ "COUNT", sub.subId.str(), countBody })));
+        } else if (hitMaxTotalEvents) {
+            // Per NIP-01, terminate with CLOSED + a machine-readable "rate-limited:"
+            // reason so historical-sync clients don't mistake a truncated result
+            // for a complete one.
+            sendClosed(sub.connId, sub.subId.str(), "rate-limited: max aggregate events reached for request");
         } else {
             PROM_INC_RELAY_MSG("EOSE");
             sendToConn(sub.connId, tao::json::to_string(tao::json::value::array({ "EOSE", sub.subId.str() })));

src/apps/relay/RelayServer.h

@@ -252,14 +252,18 @@ struct RelayServer {
         hubTrigger->send();
     }
 
-    void sendClosedError(uint64_t connId, const std::string &subId, std::string &&payload) {
+    void sendClosed(uint64_t connId, const std::string &subId, std::string &&reason) {
         PROM_INC_RELAY_MSG("CLOSED");
-        LI << "sending closed to [" << connId << "]: " << payload;
-        auto reply = tao::json::value::array({ "CLOSED", subId, std::string("ERROR: ") + payload });
+        LI << "sending closed to [" << connId << "]: " << reason;
+        auto reply = tao::json::value::array({ "CLOSED", subId, std::move(reason) });
         tpWebsocket.dispatch(0, MsgWebsocket{MsgWebsocket::Send{connId, std::move(tao::json::to_string(reply))}});
         hubTrigger->send();
     }
 
+    void sendClosedError(uint64_t connId, const std::string &subId, std::string &&payload) {
+        sendClosed(connId, subId, std::string("ERROR: ") + payload);
+    }
+
     void sendOKResponse(uint64_t connId, std::string_view eventIdHex, bool written, std::string_view message) {
         PROM_INC_RELAY_MSG("OK");
         auto reply = tao::json::value::array({ "OK", eventIdHex, written, message });

src/apps/relay/golpe.yaml

@@ -79,6 +79,9 @@ config:
   - name: relay__maxFilterLimit
     desc: "Maximum records that can be returned per filter"
     default: 500
+  - name: relay__maxTotalEventsPerReq
+    desc: "Maximum total events returned across all filters in a REQ (0 = unlimited)"
+    default: 2000
   - name: relay__maxTagsPerFilter
     desc: "Maximum number of tag filters allowed per filter (O(N^2) in matching, so keep it small)"
     default: 3

strfry.conf

@@ -116,6 +116,9 @@ relay {
     # Maximum records that can be returned per filter
     maxFilterLimit = 500
 
+    # Maximum total events returned across all filters in a REQ (0 = unlimited)
+    maxTotalEventsPerReq = 2000
+
     # Maximum number of tag filters allowed per filter (O(N^2) in matching, so keep it small)
     maxTagsPerFilter = 3