If a client disconnects before its pending EVENT write messages have been processed, drop those messages instead of trying to write them

CHANGES

@@ -1,3 +1,7 @@
+0.9.4
+  * If a client disconnects before its pending EVENT write messages have been processed,
+    drop those messages instead of trying to write them
+
 0.9.3
   * Limit on events that can be processed by a sync
     - Configurable with relay.negentropy.maxSyncEvents

TODO

@@ -15,7 +15,6 @@ features
     * delete by receivedAt, IP addrs, etc
     * inverted filter: delete events that *don't* match the provided filter
   ? less verbose default logging
-  ? if a client disconnects, delete all its pending write messages
   ? kill plugin if it times out
 
 rate limits (maybe not needed now that we have plugins?)

src/apps/relay/RelayIngester.cpp

@@ -71,6 +71,7 @@ void RelayServer::runIngester(ThreadPool<MsgIngester>::Thread &thr) {
                 }
             } else if (auto msg = std::get_if<MsgIngester::CloseConn>(&newMsg.msg)) {
                 auto connId = msg->connId;
+                tpWriter.dispatch(connId, MsgWriter{MsgWriter::CloseConn{connId}});
                 tpReqWorker.dispatch(connId, MsgReqWorker{MsgReqWorker::CloseConn{connId}});
                 tpNegentropy.dispatch(connId, MsgNegentropy{MsgNegentropy::CloseConn{connId}});
             }

src/apps/relay/RelayServer.h

@@ -70,7 +70,11 @@ struct MsgWriter : NonCopyable {
         std::string jsonStr;
     };
 
-    using Var = std::variant<AddEvent>;
+    struct CloseConn {
+        uint64_t connId;
+    };
+
+    using Var = std::variant<AddEvent, CloseConn>;
     Var msg;
     MsgWriter(Var &&msg_) : msg(std::move(msg_)) {}
 };

src/apps/relay/RelayWriter.cpp

@@ -9,6 +9,28 @@ void RelayServer::runWriter(ThreadPool<MsgWriter>::Thread &thr) {
     while(1) {
         auto newMsgs = thr.inbox.pop_all();
 
+        // Filter out messages from already closed sockets
+
+        {
+            flat_hash_set<uint64_t> closedConns;
+
+            for (auto &newMsg : newMsgs) {
+                if (auto msg = std::get_if<MsgWriter::CloseConn>(&newMsg.msg)) closedConns.insert(msg->connId);
+            }
+
+            if (closedConns.size()) {
+                decltype(newMsgs) newMsgsFiltered;
+
+                for (auto &newMsg : newMsgs) {
+                    if (auto msg = std::get_if<MsgWriter::AddEvent>(&newMsg.msg)) {
+                        if (!closedConns.contains(msg->connId)) newMsgsFiltered.emplace_back(std::move(newMsg));
+                    }
+                }
+
+                std::swap(newMsgs, newMsgsFiltered);
+            }
+        }
+
         // Prepare messages
 
         std::vector<EventToWrite> newEvents;
@@ -33,6 +55,10 @@ void RelayServer::runWriter(ThreadPool<MsgWriter>::Thread &thr) {
             }
         }
 
+        if (!newEvents.size()) continue;
+
+        // Do write
+
         try {
             auto txn = env.txn_rw();
             writeEvents(txn, newEvents);