diff --git a/src/server.zig b/src/server.zig
index de2320f..2d14743 100644
--- a/src/server.zig
+++ b/src/server.zig
@@ -7713,3 +7713,163 @@ test "server txn_op survives many real concurrent connections writing disjoint r
         try widgets.createIndex("owner");
     }
 
+    const address = try std.net.Address.parseIp("127.0.0.1", 0);
+    var server = try Server.init(allocator, &db, address);
+    try server.start();
+    defer server.stop();
+
+    const client_mod = @import("client.zig");
+    const threadCount = 6;
+    const perThread = 40;
+
+    const Ctx = struct {
+        allocator: Allocator,
+        addr: std.net.Address,
+        thread: i64,
+        ok: bool = false,
+    };
+
+    const worker = struct {
+        fn run(ctx: *Ctx) void {
+            var client = client_mod.Client.connect(ctx.allocator, ctx.addr) catch return;
+            defer client.close();
+            var buf: [32]u8 = undefined;
+            var i: i64 = 0;
+            while (i < perThread) : (i += 1) {
+                const ownerTag = std.fmt.bufPrint(&buf, "owner-{d}", .{ctx.thread}) catch return;
+                const widgetPk: i64 = ctx.thread * 1_000_000 + i;
+                const entries = [_]protocol.TxnOpEntry{
+                    .{ .verb = .put, .table = "widgets", .pk = .{ .int = widgetPk }, .fields = &.{.{ .name = "owner", .value = .{ .text = ownerTag } }} },
+                    .{ .verb = .put, .table = "tags", .pk = .{ .int = widgetPk }, .fields = &.{.{ .name = "seq", .value = .{ .int = i } }} },
+                };
+                const applied = client.txnOp(&entries) catch return;
+                if (applied != 2) return;
+            }
+            ctx.ok = true;
+        }
+    }.run;
+
+    var contexts: [threadCount]Ctx = undefined;
+    var threads: [threadCount]std.Thread = undefined;
+    for (0..threadCount) |t| {
+        contexts[t] = .{ .allocator = allocator, .addr = server.boundAddress(), .thread = @intCast(t) };
+        threads[t] = try std.Thread.spawn(.{}, worker, .{&contexts[t]});
+    }
+    for (0..threadCount) |t| threads[t].join();
+    for (0..threadCount) |t| try std.testing.expect(contexts[t].ok);
+
+    var widgets = try querymod.Table.open(&db, allocator, "widgets");
+    defer widgets.deinit();
+    var tags = try querymod.Table.open(&db, allocator, "tags");
+    defer tags.deinit();
+
+    var total: usize = 0;
+    for (0..threadCount) |t| {
+        var buf: [32]u8 = undefined;
+        const ownerTag = try std.fmt.bufPrint(&buf, "owner-{d}", .{t});
+        const owned = try widgets.select(.{ .eq = .{ .field = "owner", .value = .{ .text = ownerTag } } });
+        defer querymod.freeRows(allocator, owned);
+        // Every widget this thread wrote must show up under its own index
+        // entry, exactly `perThread` of them - not fewer (a lost write)
+        // and not more (a duplicated one).
+        try std.testing.expectEqual(@as(usize, perThread), owned.len);
+        total += owned.len;
+    }
+    try std.testing.expectEqual(@as(usize, threadCount * perThread), total);
+
+    const allTags = try tags.select(null);
+    defer querymod.freeRows(allocator, allTags);
+    try std.testing.expectEqual(@as(usize, threadCount * perThread), allTags.len);
+}
+
+test "server delivers a published message to a subscriber over real TCP" {
+    const allocator = std.testing.allocator;
+    var tmp = std.testing.tmpDir(.{});
+    defer tmp.cleanup();
+    const path = try tmp.dir.realpathAlloc(allocator, ".");
+    defer allocator.free(path);
+
+    var db = try testDb(allocator, path);
+    defer db.close();
+
+    const address = try std.net.Address.parseIp("127.0.0.1", 0);
+    var server = try Server.init(allocator, &db, address);
+    try server.start();
+    defer server.stop();
+
+    const client_mod = @import("client.zig");
+    var sub = try client_mod.Client.connect(allocator, server.boundAddress());
+    defer sub.close();
+    _ = try sub.subscribe(&.{"weather"});
+
+    var pub_ = try client_mod.Client.connect(allocator, server.boundAddress());
+    defer pub_.close();
+    const delivered = try pub_.publish("weather", "sunny");
+    try std.testing.expectEqual(@as(u64, 1), delivered);
+
+    const msg = (try sub.nextMessage(allocator)).?;
+    defer msg.deinit(allocator);
+    try std.testing.expectEqualStrings("weather", msg.channel);
+    try std.testing.expectEqualStrings("sunny", msg.payload);
+    try std.testing.expectEqual(@as(u64, 1), server.messagesPublishedTotal.get());
+}
+
+test "server delivers to a subscriber's own copy per channel it named, every subscriber on a channel reached" {
+    const allocator = std.testing.allocator;
+    var tmp = std.testing.tmpDir(.{});
+    defer tmp.cleanup();
+    const path = try tmp.dir.realpathAlloc(allocator, ".");
+    defer allocator.free(path);
+
+    var db = try testDb(allocator, path);
+    defer db.close();
+
+    const address = try std.net.Address.parseIp("127.0.0.1", 0);
+    var server = try Server.init(allocator, &db, address);
+    try server.start();
+    defer server.stop();
+
+    const client_mod = @import("client.zig");
+    var subA = try client_mod.Client.connect(allocator, server.boundAddress());
+    defer subA.close();
+    _ = try subA.subscribe(&.{ "orders", "alerts" });
+
+    var subB = try client_mod.Client.connect(allocator, server.boundAddress());
+    defer subB.close();
+    _ = try subB.subscribe(&.{"orders"});
+
+    var pub_ = try client_mod.Client.connect(allocator, server.boundAddress());
+    defer pub_.close();
+    try std.testing.expectEqual(@as(u64, 2), try pub_.publish("orders", "order-1"));
+    try std.testing.expectEqual(@as(u64, 1), try pub_.publish("alerts", "disk-full"));
+
+    const aFirst = (try subA.nextMessage(allocator)).?;
+    defer aFirst.deinit(allocator);
+    try std.testing.expectEqualStrings("orders", aFirst.channel);
+    try std.testing.expectEqualStrings("order-1", aFirst.payload);
+    const aSecond = (try subA.nextMessage(allocator)).?;
+    defer aSecond.deinit(allocator);
+    try std.testing.expectEqualStrings("alerts", aSecond.channel);
+    try std.testing.expectEqualStrings("disk-full", aSecond.payload);
+
+    const bOnly = (try subB.nextMessage(allocator)).?;
+    defer bOnly.deinit(allocator);
+    try std.testing.expectEqualStrings("orders", bOnly.channel);
+    try std.testing.expectEqualStrings("order-1", bOnly.payload);
+}
+
+test "server publish to a channel with no subscribers returns 0 without crashing, over real TCP" {
+    const allocator = std.testing.allocator;
+    var tmp = std.testing.tmpDir(.{});
+    defer tmp.cleanup();
+    const path = try tmp.dir.realpathAlloc(allocator, ".");
+    defer allocator.free(path);
+
+    var db = try testDb(allocator, path);
+    defer db.close();
+
+    const address = try std.net.Address.parseIp("127.0.0.1", 0);
+    var server = try Server.init(allocator, &db, address);
+    try server.start();
+    defer server.stop();
+