3 files changed, 139 insertions, 1 deletions

diff --git a/tests/accept_filter_probe.py b/tests/accept_filter_probe.py
new file mode 100644
index 0000000..22f585a
--- /dev/null
+++ b/tests/accept_filter_probe.py
@@ -0,0 +1,110 @@
+"""What a deferred accept does to connections that never send anything.
+
+Those connections never reach opentracker, so its own client timeout cannot
+clean them up: whatever bound exists is the kernel's. This measures it, and
+then asks the question that follows from it -- whether enough of them keep
+the tracker from answering anybody else.
+
+    make opentracker
+    python3 tests/accept_filter_probe.py
+
+On FreeBSD the filter does nothing unless accf_data is loaded, so run it both
+ways to see whether the numbers differ:
+
+    kldload accf_data
+"""
+
+import os
+import socket
+import signal
+import subprocess
+import sys
+import tempfile
+import time
+import urllib.error
+import urllib.request
+
+TRACKER = os.environ.get("OPENTRACKER", "./opentracker")
+IDLE = int(os.environ.get("IDLE", "300"))
+PATIENCE = float(os.environ.get("PATIENCE", "90"))
+HASH = "".join("%%%02x" % b for b in range(1, 21))
+
+
+def start(port):
+    directory = tempfile.mkdtemp(prefix="ot-accf-")
+    with open(os.path.join(directory, "conf"), "w") as handle:
+        handle.write("listen.tcp 127.0.0.1:%d\nlisten.udp 127.0.0.1:%d\n" % (port, port))
+    log = open(os.path.join(directory, "log"), "w+b")
+    tracker = subprocess.Popen([TRACKER, "-f", os.path.join(directory, "conf")], stdout=log, stderr=log)
+    for _ in range(300):
+        try:
+            socket.create_connection(("127.0.0.1", port), timeout=0.2).close()
+            return tracker, log
+        except OSError:
+            time.sleep(0.02)
+    tracker.kill()
+    raise SystemExit("tracker never listened on %d" % port)
+
+
+def announce(port, timeout=5):
+    """One ordinary request, so we can see whether the tracker still serves."""
+    started = time.time()
+    try:
+        urllib.request.urlopen(
+            "http://127.0.0.1:%d/announce?info_hash=%s&peer_id=-OT0001-probe0000001&port=6881&left=0" % (port, HASH),
+            timeout=timeout,
+        ).read()
+        return time.time() - started, None
+    except Exception as problem:  # noqa: BLE001 - reported, not swallowed
+        return time.time() - started, problem
+
+
+def main():
+    port = int(os.environ.get("PORT", "28000"))
+    tracker, log = start(port)
+    try:
+        print("  how long a connection that says nothing is held")
+        quiet = socket.create_connection(("127.0.0.1", port), timeout=5)
+        quiet.settimeout(PATIENCE)
+        started = time.time()
+        try:
+            held = "dropped after %.1f s" % (time.time() - started) if quiet.recv(1) == b"" else "sent us something"
+        except socket.timeout:
+            held = "still open after %.0f s" % PATIENCE
+        except OSError as problem:
+            held = "ended after %.1f s: %s" % (time.time() - started, problem)
+        quiet.close()
+        print("    %s" % held)
+
+        print("  and whether %d of them keep anybody else out" % IDLE)
+        before, problem = announce(port)
+        print("    a request before: %.0f ms%s" % (before * 1000, "" if problem is None else " (%s)" % problem))
+
+        idlers = []
+        for _ in range(IDLE):
+            try:
+                idlers.append(socket.create_connection(("127.0.0.1", port), timeout=2))
+            except OSError as problem:
+                print("    could only open %d of them: %s" % (len(idlers), problem))
+                break
+        print("    holding %d connections that have sent nothing" % len(idlers))
+
+        during, problem = announce(port, timeout=10)
+        print("    a request during: %.0f ms%s" % (during * 1000, "" if problem is None else "  REFUSED: %s" % problem))
+
+        for idler in idlers:
+            idler.close()
+        after, problem = announce(port)
+        print("    a request after:  %.0f ms%s" % (after * 1000, "" if problem is None else " (%s)" % problem))
+    finally:
+        tracker.send_signal(signal.SIGINT)
+        try:
+            tracker.wait(timeout=10)
+        except subprocess.TimeoutExpired:
+            tracker.kill()
+            tracker.wait()
+        log.close()
+
+
+if __name__ == "__main__":
+    main()
diff --git a/tests/integration/test_expiry.py b/tests/integration/test_expiry.py
index 5a21e57..5e6dc0a 100644
--- a/tests/integration/test_expiry.py
+++ b/tests/integration/test_expiry.py
@@ -27,6 +27,11 @@ class ExpirySweep(TrackerTestCase):
     def test_the_sweep_takes_the_idle_and_leaves_the_answered(self):
         idle = socket.create_connection(("127.0.0.1", self.tracker.port), timeout=5)
         try:
+            # Enough for a kernel that holds a connection back until data
+            # arrives to hand it over, and far too little to be answered, so
+            # the tracker has accepted it and is waiting for the rest
+            idle.sendall(b"GET /ann")
+
             self.assertEqual(self.tracker.announce(self.hash_for("expiry-sweep"))[b"complete"], 1)
 
             time.sleep(CLIENT_TIMEOUT + 3 * CHECK_INTERVAL)
@@ -34,6 +39,6 @@ class ExpirySweep(TrackerTestCase):
             self.assertEqual(self.tracker.torrent_count(), 1, "the tracker stopped serving while sweeping")
 
             idle.settimeout(5)
-            self.assertEqual(idle.recv(64), b"", "a connection that never sent a request was not swept")
+            self.assertEqual(idle.recv(64), b"", "a connection that went quiet mid-request was not swept")
         finally:
             idle.close()
diff --git a/tests/t_event.c b/tests/t_event.c
index ad3948f..706a285 100644
--- a/tests/t_event.c
+++ b/tests/t_event.c
@@ -241,6 +241,28 @@ static void case_close_takes_it_out(void) {
   pair_close();
 }
 
+/* A descriptor can be taken on before anybody waits for anything on it: the
+   full scrape parks a connection while a worker thread builds the reply. The
+   mechanisms disagree about what registering nothing means, so what is
+   checked is that asking for a filter afterwards still works. */
+static void case_add_nothing_then_ask(void) {
+  test_case("a descriptor added with no interest can be given some later");
+  pair_open();
+  CHECK(sv[0] != -1);
+  CHECK_INT(event_add(sv[0], 0, 7), 0);
+  CHECK_INT((int)write(sv[1], "x", 1), 1);
+  CHECK_INT(wait_for(sv[0], EVENT_IN, 200, NULL), 0);
+
+  CHECK_INT(event_modify(sv[0], EVENT_IN, 7), 0);
+  {
+    uint32_t generation = 0;
+    CHECK_INT(wait_for(sv[0], EVENT_IN, 500, &generation), 1);
+    CHECK_INT((int)generation, 7);
+  }
+  event_del(sv[0]);
+  pair_close();
+}
+
 void test_event(void) {
   CHECK_INT(event_init(), 0);
   CHECK(event_mechanism() != NULL);
@@ -254,6 +276,7 @@ void test_event(void) {
   case_timeout();
   case_unknown_descriptor();
   case_close_takes_it_out();
+  case_add_nothing_then_ask();
 
   event_deinit();
 }