14 files changed, 341 insertions, 26 deletions

diff --git a/Mk/tests.mk b/Mk/tests.mk
index 46f308b..2efdd72 100644
--- a/Mk/tests.mk
+++ b/Mk/tests.mk
@@ -29,7 +29,13 @@ LDFLAGS_integration=$(LINK) -fsanitize=address,undefined
 CFLAGS_tsan=$(OPTS_debug) $(CPPFLAGS) $(OT_CFLAGS) $(FEATURES_integration) -fsanitize=thread -fno-omit-frame-pointer
 LDFLAGS_tsan=$(LINK) -fsanitize=thread
 
-BINARIES+=$(TESTBIN) $(TESTBIN).poll $(TESTBIN).sanitize opentracker.integration opentracker.tsan
+# Live sync is a compile time feature, so reaching it from a black box test
+# needs a binary of its own. tests/integration/test_livesync.py skips itself
+# against a build without it, which is what -v is for.
+FEATURES_livesync=$(FEATURES_integration) -DWANT_SYNC_LIVE
+CFLAGS_livesync=$(OPTS_debug) $(CPPFLAGS) $(OT_CFLAGS) $(FEATURES_livesync) -fsanitize=address,undefined -fno-omit-frame-pointer
+
+BINARIES+=$(TESTBIN) $(TESTBIN).poll $(TESTBIN).sanitize opentracker.integration opentracker.tsan opentracker.livesync
 
 $(TESTBIN): $(TEST_SOURCES) $(TEST_UNITS) $(TEST_HEADERS) $(BUILDDEPS)
         $(CC) -o $@ $(CFLAGS_test) $(TEST_SOURCES) $(TEST_UNITS)
@@ -46,6 +52,9 @@ opentracker.integration: $(SOURCES) $(HEADERS) $(BUILDDEPS)
 opentracker.tsan: $(SOURCES) $(HEADERS) $(BUILDDEPS)
         $(CC) -o $@ $(CFLAGS_tsan) $(SOURCES) $(LDFLAGS_tsan)
 
+opentracker.livesync: $(SOURCES) $(HEADERS) $(BUILDDEPS)
+        $(CC) -o $@ $(CFLAGS_livesync) $(SOURCES) $(LDFLAGS_integration)
+
 check: $(TESTBIN) $(BINARY)
         ./$(TESTBIN)
 
@@ -61,12 +70,18 @@ check-integration: opentracker.integration
 check-tsan: opentracker.tsan
         OPENTRACKER=./opentracker.tsan $(PYTHON) -m unittest discover -s tests/integration
 
+# The whole suite against a build that has live sync, which is also the only
+# way the live sync tests stop skipping.
+check-livesync: opentracker.livesync
+        OPENTRACKER=./opentracker.livesync $(PYTHON) -m unittest discover -s tests/integration
+
 # Cheapest first, so a quick failure comes back quickly.
 check-all:
         $(MAKE) check
         $(MAKE) check-poll
         $(MAKE) check-sanitize
         $(MAKE) check-integration
+        $(MAKE) check-livesync
         $(MAKE) check-tsan
 
-.PHONY: check check-poll check-sanitize check-integration check-tsan check-all
+.PHONY: check check-poll check-sanitize check-integration check-livesync check-tsan check-all
diff --git a/man5/opentracker.conf.5 b/man5/opentracker.conf.5
index 67e624f..9a836ae 100644
--- a/man5/opentracker.conf.5
+++ b/man5/opentracker.conf.5
@@ -79,7 +79,18 @@ Specifies the path to the stats location. You can configure opentracker to appea
 Specifies the IP address or network of the reverse proxies. Opentracker will take the X-Forwarded-For address instead of the source IP address. Can be added more than once.
 .Pp
 .It livesync.cluster.listen Ar ip_address:port
-Specifies the IP address and port opentracker will listen on for incoming live sync packets to keep a cluster of opentrackers synchronized.
+Specifies the IP address and port opentracker will listen on for incoming live
+sync packets to keep a cluster of opentrackers synchronized. Every member of a
+cluster has to agree on the port, which is also the one packets are sent to.
+.Pp
+.It livesync.cluster.group Ar ip_address
+Specifies the multicast group a cluster talks in. For reaosns of backwards
+comaptibility with legacy trackers, it defaults to 224.0.23.5, which,
+technically sits in a range IANA assigns on request.
+
+A cluster of its own is better served by an address out of 239.192.0.0/14,
+which RFC 2365 sets aside for private use. Packets carry a ttl of 1 either
+way and never leave the local link.
 .Pp
 .It livesync.cluster.node_ip Ar ip_address
 Specifies one trusted IP address for sync between trackers running in a cluster. Can be added more than once.
diff --git a/opentracker.conf.sample b/opentracker.conf.sample
index 491c88a..7b35369 100644
--- a/opentracker.conf.sample
+++ b/opentracker.conf.sample
@@ -138,12 +138,22 @@
 # livesync.cluster.listen 192.168.0.1:9696
 #
 #      Note that two udp sockets will be opened. One on ip address 0.0.0.0
-#      port 9696, that will join the multicast group 224.0.42.23 for incoming
-#      udp packets and one on ip address 192.168.0.1 port 9696 for outgoing
-#      udp packets.
+#      port 9696, that will join the multicast group for incoming udp packets
+#      and one on ip address 192.168.0.1 port 9696 for outgoing ones. Every
+#      member of a cluster has to agree on the port as well as the group.
+#
+#      The group a cluster talks in.
+#
+# livesync.cluster.group 239.192.23.5
+#
+#      For reaosns of backwards comaptibility with legacy trackers, it
+#      defaults to 224.0.23.5, which, technically sits in a range IANA
+#      assigns on request.
+#
+#      A cluster of your own is better served by an address out of
+#      239.192.0.0/14, which RFC 2365 sets aside for private use. Packets
+#      carry a ttl of 1 either way, so they never leave the local link.
 #
-#      As of now one and only one ip address must be given, if opentracker
-#      was built with the WANT_SYNC_LIVE feature.
 #
 
 # IV)  Sync between trackers running in a cluster is restricted to packets
diff --git a/src/opentracker.c b/src/opentracker.c
index 14b817e..60a6d8c 100644
--- a/src/opentracker.c
+++ b/src/opentracker.c
@@ -109,7 +109,7 @@ static void *server_mainloop(void *args) {
 
   (void)args;
 
-  if (workstruct_init(&ws) == -1)
+  if (workstruct_init(&ws, G_INBUF_SIZE) == -1)
     panic("Initializing worker failed");
 
   while (g_opentracker_running) {
diff --git a/src/ot_config.c b/src/ot_config.c
index 95136b5..4064d1b 100644
--- a/src/ot_config.c
+++ b/src/ot_config.c
@@ -208,6 +208,7 @@ typedef enum {
     CONFIG_TRACKER_REDIRECT_URL,
     CONFIG_LIVESYNC_CLUSTER_NODE_IP,
     CONFIG_LIVESYNC_CLUSTER_LISTEN,
+    CONFIG_LIVESYNC_CLUSTER_GROUP,
 } config_kind;
 
 typedef struct {
@@ -247,6 +248,7 @@ static const config_key config_keys[] = {
 #ifdef WANT_SYNC_LIVE
     CONFIG_KEY("livesync.cluster.node_ip", CONFIG_LIVESYNC_CLUSTER_NODE_IP),
     CONFIG_KEY("livesync.cluster.listen",  CONFIG_LIVESYNC_CLUSTER_LISTEN),
+    CONFIG_KEY("livesync.cluster.group",   CONFIG_LIVESYNC_CLUSTER_GROUP),
 #endif
 };
 static const size_t config_entry_count = sizeof(config_keys) / sizeof(config_keys[0]);
@@ -378,6 +380,10 @@ static int parse_configfile(char *config_filename) {
         set_config_option(&g_redirecturl, p);
         break;
 #ifdef WANT_SYNC_LIVE
+      case CONFIG_LIVESYNC_CLUSTER_GROUP:
+        if (!ip_parse(p, tmpip) || livesync_set_group(tmpip) == -1)
+          goto parse_error;
+        break;
       case CONFIG_LIVESYNC_CLUSTER_NODE_IP:
         if (!scan_ip6_net(p, &tmpnet))
           goto parse_error;
diff --git a/src/ot_livesync.c b/src/ot_livesync.c
index 083fe1a..85577f7 100644
--- a/src/ot_livesync.c
+++ b/src/ot_livesync.c
@@ -4,6 +4,7 @@
 /* System */
 #include <arpa/inet.h>
 #include <pthread.h>
+#include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
 #include <sys/socket.h>
@@ -21,7 +22,10 @@
 
 #ifdef WANT_SYNC_LIVE
 
-ot_ip6 groupip_1 = {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xff, 0xff, 224, 0, 23, 5};
+/* The multicast group, legacy opentrackers joined by default 224.0.23.5, so it
+   stays the default */
+static ot_ip6   g_group = {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xff, 0xff, 224, 0, 23, 5};
+static uint16_t g_port  = LIVESYNC_PORT;
 
 #define LIVESYNC_INCOMING_BUFFSIZE        (256 * 256)
 
@@ -59,7 +63,7 @@ void livesync_init(void) {
   uint32_t packet_type4 = htonl(OT_SYNC_PEER4);
 
   if (g_socket_in == -1)
-    exerr("No socket address for live sync specified.");
+    return;
 
   /* Prepare outgoing peers buffer */
   memcpy(g_v6_buf.data, &g_tracker_id, sizeof(g_tracker_id));
@@ -94,6 +98,14 @@ void livesync_deinit(void) {
     close(g_socket_out);
 }
 
+int livesync_set_group(const ot_ip6 group) {
+  /* v4 multicast is 224.0.0.0/4, and the socket layer has nothing else */
+  if (!ip_is_v4mapped(group) || (group[12] & 0xf0) != 0xe0)
+    return -1;
+  memcpy(g_group, group, sizeof(ot_ip6));
+  return 0;
+}
+
 void livesync_bind_mcast(ot_ip6 ip, uint16_t port) {
   if (!ip_is_v4mapped(ip))
     exerr("v6 mcast support not yet available.");
@@ -101,7 +113,9 @@ void livesync_bind_mcast(ot_ip6 ip, uint16_t port) {
   if (g_socket_in != -1)
     exerr("Error: Livesync listen ip specified twice.");
 
-  if ((g_socket_in = socket_mcast_join(groupip_1, ip, port)) == -1)
+  g_port = port;
+
+  if ((g_socket_in = socket_mcast_join(g_group, ip, port)) == -1)
     exerr("Error: Cant bind live sync incoming socket.");
 
   if ((g_socket_out = socket_mcast_bind(ip, port, 0)) == -1)
@@ -121,7 +135,7 @@ static void livesync_issue_peersync(sync_buffer *buf) {
      has modified the protected element */
   pthread_mutex_unlock(&g_outbuf_mutex);
 
-  socket_mcast_send(g_socket_out, mycopy, fill, groupip_1, LIVESYNC_PORT);
+  socket_mcast_send(g_socket_out, mycopy, fill, g_group, g_port);
 }
 
 static void livesync_handle_peersync(struct ot_workstruct *ws, size_t peer_size) {
@@ -159,6 +173,9 @@ static void livesync_handle_peersync(struct ot_workstruct *ws, size_t peer_size)
    stuck when there's not enough traffic to fill udp packets fast
    enough */
 void livesync_ticker(void) {
+  if (g_socket_out == -1)
+    return;
+
   /* livesync_issue_peersync sets g_next_packet_time */
   pthread_mutex_lock(&g_outbuf_mutex);
   if (time_now_seconds() > g_v6_buf.next_packet_time && g_v6_buf.fill > sizeof(g_tracker_id) + sizeof(uint32_t))
@@ -176,8 +193,14 @@ void livesync_ticker(void) {
 /* Inform live sync about whats going on. */
 void livesync_tell(struct ot_workstruct *ws) {
   size_t       peer_size; /* initialized in next line */
-  ot_peer     *peer_src = peer_from_peer6(&ws->peer, &peer_size);
-  sync_buffer *dest_buf = peer_size == OT_PEER_SIZE6 ? &g_v6_buf : &g_v4_buf;
+  ot_peer     *peer_src;
+  sync_buffer *dest_buf;
+
+  if (g_socket_out == -1)
+    return;
+
+  peer_src = peer_from_peer6(&ws->peer, &peer_size);
+  dest_buf = peer_size == OT_PEER_SIZE6 ? &g_v6_buf : &g_v4_buf;
 
   pthread_mutex_lock(&g_outbuf_mutex);
 
@@ -200,9 +223,12 @@ static void *livesync_worker(void *args) {
 
   (void)args;
 
-  /* Initialize our "thread local storage" */
-  ws.inbuf = ws.request = malloc(LIVESYNC_INCOMING_BUFFSIZE);
-  ws.outbuf = ws.reply = 0;
+  if (workstruct_init(&ws, LIVESYNC_INCOMING_BUFFSIZE) == -1) {
+    fprintf(stderr, "Warning: live sync could not get its buffers and will not run.\n");
+    atomic_store(&thread_left, 1);
+    return NULL;
+  }
+  ws.request = ws.inbuf;
 
   while (g_opentracker_running) {
     uint32_t packet_type;
@@ -233,7 +259,7 @@ static void *livesync_worker(void *args) {
     }
   }
 
-  free(ws.inbuf);
+  workstruct_deinit(&ws);
   atomic_store(&thread_left, 1);
   return NULL;
 }
diff --git a/src/ot_livesync.h b/src/ot_livesync.h
index 4bd3af0..cf3c64c 100644
--- a/src/ot_livesync.h
+++ b/src/ot_livesync.h
@@ -7,7 +7,8 @@
 #include "trackerlogic.h"
 
 /*
-  Syncing is done as udp packets in the multicast domain 224.0.42.5 port 9696
+  Syncing is done as udp packets in the multicast group 224.0.23.5, on the
+  port livesync.cluster.listen names, 9696 by default
 
   Each tracker should join the multicast group and send its live sync packets
   to that group, using a ttl of 1
@@ -47,6 +48,10 @@
 void livesync_init(void);
 void livesync_deinit(void);
 
+/* The group to join and to send to, (defaults to 224.0.23.5).
+   Returns -1 if group is not a v4 multicast address. */
+int  livesync_set_group(const ot_ip6 group);
+
 /* Join multicast group for listening and create sending socket */
 void livesync_bind_mcast(ot_ip6 ip, uint16_t port);
 
diff --git a/src/ot_worker.c b/src/ot_worker.c
index 432d272..ef50945 100644
--- a/src/ot_worker.c
+++ b/src/ot_worker.c
@@ -20,7 +20,7 @@ static size_t             g_worker_count;
 static void *worker_trampoline(void *args) {
   struct ot_worker *me = args;
 
-  if (workstruct_init(&me->ws) == -1)
+  if (workstruct_init(&me->ws, G_INBUF_SIZE) == -1)
     fprintf(stderr, "Warning: a worker could not get its buffers and will not run.\n");
   else {
     me->run(me);
diff --git a/src/trackerlogic.c b/src/trackerlogic.c
index 9565090..8e30a48 100644
--- a/src/trackerlogic.c
+++ b/src/trackerlogic.c
@@ -567,9 +567,11 @@ void trackerlogic_init(void) {
   stats_init();
 }
 
-int workstruct_init(struct ot_workstruct *ws) {
+/* Live sync reads whole packets rather than requests and wants a larger
+   inbuf than the request path does, hence the size. */
+int workstruct_init(struct ot_workstruct *ws, size_t inbuf_size) {
   memset(ws, 0, sizeof(*ws));
-  ws->inbuf  = malloc(G_INBUF_SIZE);
+  ws->inbuf  = malloc(inbuf_size);
   ws->outbuf = malloc(G_OUTBUF_SIZE);
 #ifdef _DEBUG_HTTPERROR
   ws->debugbuf = malloc(G_DEBUGBUF_SIZE);
diff --git a/src/trackerlogic.h b/src/trackerlogic.h
index 7292d36..692b6af 100644
--- a/src/trackerlogic.h
+++ b/src/trackerlogic.h
@@ -242,7 +242,7 @@ struct ot_workstruct {
 /* Everything a thread that serves requests needs for itself. Set up and torn
    down in one place so that a new kind of worker cannot miss a piece.
    Returns -1 if the buffers could not be had. */
-int  workstruct_init(struct ot_workstruct *ws);
+int  workstruct_init(struct ot_workstruct *ws, size_t inbuf_size);
 void workstruct_deinit(struct ot_workstruct *ws);
 
 /*
diff --git a/tests/integration/harness.py b/tests/integration/harness.py
index b98ce83..10aa6de 100644
--- a/tests/integration/harness.py
+++ b/tests/integration/harness.py
@@ -20,6 +20,9 @@ import urllib.error
 import urllib.parse
 import urllib.request
 
+# groupip_1 in src/ot_livesync.c
+LIVESYNC_GROUP = "224.0.23.5"
+
 TRACKER = os.environ.get("OPENTRACKER", os.path.join(os.path.dirname(__file__), "..", "..", "opentracker.debug"))
 
 
@@ -76,6 +79,9 @@ class Tracker:
 
     def __init__(self, extra_config="", files=None):
         self.port = free_port()
+        # Only used by a config that asks for live sync, but allocated either
+        # way so {livesync_port} always resolves.
+        self.livesync_port = free_port()
         self.dir = tempfile.mkdtemp(prefix="ot-integration-")
         self.log = None
         self.process = None
@@ -87,7 +93,7 @@ class Tracker:
             with open(self.config, "w") as handle:
                 handle.write("listen.tcp 127.0.0.1:%d\n" % self.port)
                 handle.write("listen.udp 127.0.0.1:%d\n" % self.port)
-                handle.write(extra_config.replace("{dir}", self.dir))
+                handle.write(extra_config.replace("{dir}", self.dir).replace("{livesync_port}", str(self.livesync_port)))
             self.log = open(os.path.join(self.dir, "tracker.log"), "w+b")
             # Leak detection is left at the sanitizer's own default, which is on
             # where LeakSanitizer exists and off where it does not. Asking for it
@@ -186,6 +192,32 @@ class Tracker:
     def torrent_count(self):
         return int(self.stats("torr").split()[0])
 
+    def livesync_send(self, payload, group=LIVESYNC_GROUP):
+        """Put a packet on the live sync group.
+
+        It has to go to the group rather than straight at the port: the
+        tracker binds two sockets there, the reader on the wildcard address
+        and the writer on the configured one, and the kernel gives a unicast
+        datagram to the more specific of the two, which nobody reads."""
+        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        try:
+            sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_IF, socket.inet_aton("127.0.0.1"))
+            sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_LOOP, 1)
+            sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 1)
+            sock.sendto(payload, (group, self.livesync_port))
+        finally:
+            sock.close()
+
+    def wait_until(self, predicate, timeout=5):
+        """A live sync packet is handled by another thread, so an effect of
+        one arrives some time after the send."""
+        deadline = time.time() + timeout
+        while time.time() < deadline:
+            if predicate():
+                return True
+            time.sleep(0.02)
+        return predicate()
+
     # ------------------------------------------------------------ over udp
 
     def udp_connect(self):
diff --git a/tests/integration/test_livesync.py b/tests/integration/test_livesync.py
new file mode 100644
index 0000000..52e871f
--- /dev/null
+++ b/tests/integration/test_livesync.py
@@ -0,0 +1,208 @@
+"""Live sync, driven over the multicast group a cluster would use.
+
+Live sync sits behind WANT_SYNC_LIVE, so these only run against a binary
+built with it: make check-livesync. Against any other build they skip, which
+is what the feature list in -v is for.
+"""
+
+import socket
+import struct
+import subprocess
+import unittest
+
+from harness import TRACKER, TrackerTestCase
+
+# enum { OT_SYNC_PEER4, OT_SYNC_PEER6 } in src/ot_livesync.c
+SYNC_PEER4 = 0
+SYNC_PEER6 = 1
+
+# src/trackerlogic.h
+FLAG_SEEDING = 0x80
+FLAG_COMPLETED = 0x40
+FLAG_STOPPED = 0x20
+
+
+def has_livesync():
+    done = subprocess.run([TRACKER, "-v"], capture_output=True, timeout=30)
+    return b"WANT_SYNC_LIVE" in done.stderr
+
+
+def peer4(address, port, flags):
+    """Four bytes of address, two of port, one of flags, one the tracker
+    overwrites with its own idea of the time."""
+    return socket.inet_aton(address) + struct.pack("!H", port) + bytes([flags, 0])
+
+
+def peer6(address, port, flags):
+    return socket.inet_pton(socket.AF_INET6, address) + struct.pack("!H", port) + bytes([flags, 0])
+
+
+def packet(kind, entries, tracker_id=0xDEADBEEF):
+    body = b"".join(info_hash + peer for info_hash, peer in entries)
+    return struct.pack("!I", tracker_id) + struct.pack("!I", kind) + body
+
+
+@unittest.skipUnless(has_livesync(), "this build has no WANT_SYNC_LIVE")
+class LiveSyncIn(TrackerTestCase):
+    config = "livesync.cluster.listen 127.0.0.1:{livesync_port}\nlivesync.cluster.node_ip 127.0.0.1\n"
+
+    def counts(self, info_hash):
+        """Seeders and leechers a scrape reports for one torrent."""
+        files = self.tracker.scrape(info_hash)[b"files"]
+        if info_hash not in files:
+            return None
+        entry = files[info_hash]
+        return entry[b"complete"], entry[b"incomplete"]
+
+    def test_a_synced_seeder_arrives(self):
+        info_hash = self.hash_for("sync-seed")
+        self.tracker.livesync_send(packet(SYNC_PEER4, [(info_hash, peer4("10.1.2.3", 6881, FLAG_SEEDING))]))
+        self.assertTrue(self.tracker.wait_until(lambda: self.counts(info_hash) == (1, 0)),
+                        "scrape says %r" % (self.counts(info_hash),))
+
+    def test_a_synced_leecher_arrives(self):
+        info_hash = self.hash_for("sync-leech")
+        self.tracker.livesync_send(packet(SYNC_PEER4, [(info_hash, peer4("10.1.2.4", 6881, 0))]))
+        self.assertTrue(self.tracker.wait_until(lambda: self.counts(info_hash) == (0, 1)),
+                        "scrape says %r" % (self.counts(info_hash),))
+
+    def test_a_v6_peer_arrives(self):
+        info_hash = self.hash_for("sync-v6")
+        self.tracker.livesync_send(packet(SYNC_PEER6, [(info_hash, peer6("2001:db8::1", 6881, FLAG_SEEDING))]))
+        self.assertTrue(self.tracker.wait_until(lambda: self.counts(info_hash) == (1, 0)),
+                        "scrape says %r" % (self.counts(info_hash),))
+
+    def test_one_packet_carries_several_torrents(self):
+        hashes = [self.hash_for("sync-many-%d" % i) for i in range(4)]
+        self.tracker.livesync_send(packet(SYNC_PEER4, [(h, peer4("10.1.2.%d" % (5 + i), 6881, FLAG_SEEDING))
+                                                       for i, h in enumerate(hashes)]))
+        self.assertTrue(self.tracker.wait_until(lambda: all(self.counts(h) == (1, 0) for h in hashes)))
+
+    def test_a_stopped_peer_is_taken_away_again(self):
+        info_hash = self.hash_for("sync-stop")
+        peer = peer4("10.1.2.9", 6881, FLAG_SEEDING)
+        self.tracker.livesync_send(packet(SYNC_PEER4, [(info_hash, peer)]))
+        self.assertTrue(self.tracker.wait_until(lambda: self.counts(info_hash) == (1, 0)))
+
+        self.tracker.livesync_send(packet(SYNC_PEER4, [(info_hash, peer4("10.1.2.9", 6881, FLAG_STOPPED))]))
+        self.assertTrue(self.tracker.wait_until(lambda: self.counts(info_hash) in (None, (0, 0))),
+                        "scrape still says %r" % (self.counts(info_hash),))
+
+    def test_a_truncated_entry_is_dropped(self):
+        """The loop takes whole hash and peer pairs and leaves a short tail
+        where it is, so the complete entry before it still lands."""
+        good, short = self.hash_for("sync-good"), self.hash_for("sync-short")
+        payload = packet(SYNC_PEER4, [(good, peer4("10.1.2.10", 6881, FLAG_SEEDING))]) + short + b"\x0a\x01"
+        self.tracker.livesync_send(payload)
+        self.assertTrue(self.tracker.wait_until(lambda: self.counts(good) == (1, 0)))
+        self.assertIsNone(self.counts(short))
+
+    def test_an_unknown_packet_type_is_ignored(self):
+        info_hash = self.hash_for("sync-unknown")
+        self.tracker.livesync_send(packet(99, [(info_hash, peer4("10.1.2.11", 6881, FLAG_SEEDING))]))
+        self.assertFalse(self.tracker.wait_until(lambda: self.counts(info_hash) is not None, timeout=1))
+
+    def test_synced_peers_are_counted(self):
+        """The live sync worker keeps its own tally like every other worker,
+        and ?mode=syncs is the sum of them."""
+        before = int(self.tracker.stats("syncs").split()[0])
+        hashes = [self.hash_for("sync-count-%d" % i) for i in range(3)]
+        self.tracker.livesync_send(packet(SYNC_PEER4, [(h, peer4("10.1.2.%d" % (20 + i), 6881, FLAG_SEEDING))
+                                                       for i, h in enumerate(hashes)]))
+        self.assertTrue(self.tracker.wait_until(
+            lambda: int(self.tracker.stats("syncs").split()[0]) == before + 3),
+            "syncs says %s" % self.tracker.stats("syncs").split()[0])
+
+    def test_a_packet_with_nothing_but_a_header_is_ignored(self):
+        before = self.tracker.torrent_count()
+        self.tracker.livesync_send(struct.pack("!I", 0xDEADBEEF) + struct.pack("!I", SYNC_PEER4))
+        self.assertFalse(self.tracker.wait_until(lambda: self.tracker.torrent_count() != before, timeout=1))
+
+
+@unittest.skipUnless(has_livesync(), "this build has no WANT_SYNC_LIVE")
+class LiveSyncAccess(TrackerTestCase):
+    """Nothing blesses 127.0.0.1 here, so the tracker should take nothing
+    from it. Source address checking is all live sync has."""
+
+    config = "livesync.cluster.listen 127.0.0.1:{livesync_port}\nlivesync.cluster.node_ip 10.0.0.1\n"
+
+    def test_an_unblessed_sender_is_ignored(self):
+        info_hash = self.hash_for("sync-unblessed")
+        self.tracker.livesync_send(packet(SYNC_PEER4, [(info_hash, peer4("10.1.2.12", 6881, FLAG_SEEDING))]))
+        self.assertFalse(self.tracker.wait_until(lambda: self.tracker.torrent_count() != 0, timeout=1))
+
+
+@unittest.skipUnless(has_livesync(), "this build has no WANT_SYNC_LIVE")
+class LiveSyncUnconfigured(TrackerTestCase):
+    """A binary that has the feature but no cluster to join. Somebody whose
+    packager turned WANT_SYNC_LIVE on did not ask for live sync and should
+    not be able to tell that it is there."""
+
+    config = ""
+
+    def test_the_tracker_runs_anyway(self):
+        info_hash = self.hash_for("no-cluster")
+        self.tracker.announce(info_hash, port=6881, left=0)
+        self.assertEqual(self.tracker.torrent_count(), 1)
+
+    def test_nothing_is_said_about_the_missing_cluster(self):
+        self.assertNotIn("sync", self.tracker.output().lower())
+
+    def test_no_socket_is_bound_for_it(self):
+        """A configured tracker holds the wildcard address on that port, so
+        taking it here proves nothing was started."""
+        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        try:
+            sock.bind(("0.0.0.0", self.tracker.livesync_port))
+        finally:
+            sock.close()
+
+    def test_announcing_stays_clean(self):
+        """Enough announces to pass the watermark that would flush an
+        outgoing buffer, if there were one to flush. The sanitizers in this
+        build turn anything the attempt touched into a failure."""
+        for i in range(200):
+            self.tracker.announce(self.hash_for("no-cluster-%d" % i), port=6881 + i, left=0)
+        self.assertEqual(self.tracker.torrent_count(), 200)
+
+
+# RFC 2365 sets 239.192.0.0/14 aside for exactly this kind of use
+OWN_GROUP = "239.192.0.7"
+
+
+@unittest.skipUnless(has_livesync(), "this build has no WANT_SYNC_LIVE")
+class LiveSyncGroup(TrackerTestCase):
+    config = ("livesync.cluster.group %s\n"
+              "livesync.cluster.listen 127.0.0.1:{livesync_port}\n"
+              "livesync.cluster.node_ip 127.0.0.1\n" % OWN_GROUP)
+
+    def test_the_configured_group_is_joined(self):
+        info_hash = self.hash_for("own-group")
+        self.tracker.livesync_send(packet(SYNC_PEER4, [(info_hash, peer4("10.1.3.1", 6881, FLAG_SEEDING))]),
+                                   group=OWN_GROUP)
+        self.assertTrue(self.tracker.wait_until(lambda: self.tracker.torrent_count() == 1))
+
+    def test_the_default_group_is_left_alone(self):
+        """A cluster that picked its own group should not hear the one every
+        other opentracker is on."""
+        info_hash = self.hash_for("not-our-group")
+        self.tracker.livesync_send(packet(SYNC_PEER4, [(info_hash, peer4("10.1.3.2", 6881, FLAG_SEEDING))]))
+        self.assertFalse(self.tracker.wait_until(lambda: self.tracker.torrent_count() != 0, timeout=1))
+
+
+@unittest.skipUnless(has_livesync(), "this build has no WANT_SYNC_LIVE")
+class LiveSyncGroupRefused(TrackerTestCase):
+    """A unicast address is not a group. The line is refused like any other
+    bad config line, which leaves the default standing."""
+
+    config = ("livesync.cluster.group 10.0.0.1\n"
+              "livesync.cluster.listen 127.0.0.1:{livesync_port}\n"
+              "livesync.cluster.node_ip 127.0.0.1\n")
+
+    def test_the_line_is_refused(self):
+        self.assertIn("Parse error", self.tracker.output())
+
+    def test_and_the_default_group_still_works(self):
+        info_hash = self.hash_for("fallback-group")
+        self.tracker.livesync_send(packet(SYNC_PEER4, [(info_hash, peer4("10.1.3.3", 6881, FLAG_SEEDING))]))
+        self.assertTrue(self.tracker.wait_until(lambda: self.tracker.torrent_count() == 1))
diff --git a/sync_daemon.pl b/tests/sync_daemon.pl
index 9e4bdb9..9e4bdb9 100644
--- a/sync_daemon.pl
+++ b/tests/sync_daemon.pl
diff --git a/tests/t_trackerlogic.c b/tests/t_trackerlogic.c
index f64977f..1cf9141 100644
--- a/tests/t_trackerlogic.c
+++ b/tests/t_trackerlogic.c
@@ -224,7 +224,7 @@ void test_trackerlogic(void) {
   pthread_sigmask(SIG_BLOCK, &signal_mask, NULL);
 
   trackerlogic_init();
-  CHECK_INT(workstruct_init(&ws), 0);
+  CHECK_INT(workstruct_init(&ws, G_INBUF_SIZE), 0);
 
   case_an_announce_makes_a_torrent();
   case_the_same_peer_twice_is_one_peer();