diff --git a/src/DBQuery.h b/src/DBQuery.h index e3ec1665..2591f29a 100644 --- a/src/DBQuery.h +++ b/src/DBQuery.h @@ -290,6 +290,8 @@ struct DBQuery : NonCopyable { flat_hash_set 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) { diff --git a/src/QueryScheduler.h b/src/QueryScheduler.h index 72722c82..99ba2d69 100644 --- a/src/QueryScheduler.h +++ b/src/QueryScheduler.h @@ -6,7 +6,7 @@ struct QueryScheduler : NonCopyable { std::function onEvent; std::function &levIds)> onEventBatch; - std::function onComplete; + std::function 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 { diff --git a/src/apps/relay/RelayNegentropy.cpp b/src/apps/relay/RelayNegentropy.cpp index 017c0be3..a766d4d5 100644 --- a/src/apps/relay/RelayNegentropy.cpp +++ b/src/apps/relay/RelayNegentropy.cpp @@ -137,7 +137,7 @@ void RelayServer::runNegentropy(ThreadPool::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; diff --git a/src/apps/relay/RelayReqWorker.cpp b/src/apps/relay/RelayReqWorker.cpp index 56021aea..af3d7729 100644 --- a/src/apps/relay/RelayReqWorker.cpp +++ b/src/apps/relay/RelayReqWorker.cpp @@ -21,7 +21,7 @@ void RelayServer::runReqWorker(ThreadPool::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::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() }))); diff --git a/src/apps/relay/RelayServer.h b/src/apps/relay/RelayServer.h index 7767d8c6..aa747990 100644 --- a/src/apps/relay/RelayServer.h +++ b/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 }); diff --git a/src/apps/relay/golpe.yaml b/src/apps/relay/golpe.yaml index 553d0b44..01a43c38 100644 --- a/src/apps/relay/golpe.yaml +++ b/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 diff --git a/strfry.conf b/strfry.conf index 81da7afa..2b3ae774 100644 --- a/strfry.conf +++ b/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