feat: add Publisher and run to pubsub

src/pubsub.zig

@@ -628,3 +628,221 @@ test "closing a subscriber wakes a thread already blocked in waitMessage" {
     try std.testing.expect(result == null);
 }
 
+test "publishing after closeSubscriber but before destroySubscriber never touches freed memory" {
+    const allocator = std.testing.allocator;
+    var hub = Hub.init(allocator);
+    defer hub.deinit();
+
+    const sub = try hub.subscribe(&.{"ch"}, defaultCapacity);
+    hub.closeSubscriber(sub);
+    // The channel entry is gone entirely now (empty list cleanup), so this
+    // publish must find zero subscribers rather than dereferencing `sub`.
+    try std.testing.expectEqual(@as(usize, 0), try hub.publish(allocator, "ch", "x"));
+    hub.destroySubscriber(sub);
+}
+
+test "concurrent publishers hammering one subscriber deliver every message intact" {
+    const allocator = std.testing.allocator;
+    var hub = Hub.init(allocator);
+    defer hub.deinit();
+
+    const publisherCount = 4;
+    const perPublisher = 200;
+    const sub = try hub.subscribe(&.{"ch"}, publisherCount * perPublisher);
+    defer destroy(&hub, sub);
+
+    const Publisher = struct {
+        fn run(h: *Hub, alloc: Allocator, tag: usize) void {
+            var buf: [32]u8 = undefined;
+            var i: usize = 0;
+            while (i < perPublisher) : (i += 1) {
+                const text = std.fmt.bufPrint(&buf, "{d}:{d}", .{ tag, i }) catch unreachable;
+                _ = h.publish(alloc, "ch", text) catch unreachable;
+            }
+        }
+    };
+
+    var threads: [publisherCount]std.Thread = undefined;
+    for (0..publisherCount) |i| {
+        threads[i] = try std.Thread.spawn(.{}, Publisher.run, .{ &hub, allocator, i });
+    }
+    for (threads) |t| t.join();
+
+    try std.testing.expectEqual(@as(usize, 0), sub.droppedCount());
+
+    var seenPerTag = [_]usize{0} ** publisherCount;
+    var total: usize = 0;
+    while (sub.waitMessage()) |msg| {
+        defer msg.deinit(allocator);
+        total += 1;
+        var it = std.mem.splitScalar(u8, msg.payload, ':');
+        const tag = std.fmt.parseInt(usize, it.next().?, 10) catch unreachable;
+        seenPerTag[tag] += 1;
+        if (total == publisherCount * perPublisher) break;
+    }
+    try std.testing.expectEqual(@as(usize, publisherCount * perPublisher), total);
+    for (seenPerTag) |count| try std.testing.expectEqual(@as(usize, perPublisher), count);
+}
+
+test "subscribe leaves nothing registered when it fails partway through" {
+    var failing = std.testing.FailingAllocator.init(std.testing.allocator, .{ .fail_index = 0 });
+    var hub = Hub.init(failing.allocator());
+    defer hub.deinit();
+
+    try std.testing.expectError(error.OutOfMemory, hub.subscribe(&.{ "a", "b", "c", "d" }, defaultCapacity));
+    // Nothing should be left registered anywhere after the unwind.
+    try std.testing.expectEqual(@as(usize, 0), hub.channelSubscriberCount("a"));
+    try std.testing.expectEqual(@as(usize, 0), hub.channelSubscriberCount("b"));
+}
+
+test "push under memory pressure drops the message and counts it instead of leaking" {
+    var failing = std.testing.FailingAllocator.init(std.testing.allocator, .{ .fail_index = 0 });
+    const allocator = failing.allocator();
+    const channels = try std.testing.allocator.alloc([]u8, 0);
+    var sub = Subscriber.init(allocator, defaultCapacity, channels);
+    defer sub.deinitContents();
+
+    const channel = try std.testing.allocator.dupe(u8, "ch");
+    const payload = try std.testing.allocator.dupe(u8, "x");
+    sub.push(.{ .channel = channel, .payload = payload });
+
+    try std.testing.expectEqual(@as(usize, 1), sub.droppedCount());
+    try std.testing.expectEqual(@as(usize, 0), sub.queue.items.len);
+}
+
+test "publish tolerates an out-of-memory snapshot allocation without leaking" {
+    const allocator = std.testing.allocator;
+    var hub = Hub.init(allocator);
+    defer hub.deinit();
+
+    const sub = try hub.subscribe(&.{"ch"}, defaultCapacity);
+    defer destroy(&hub, sub);
+
+    var failing = std.testing.FailingAllocator.init(std.testing.allocator, .{ .fail_index = 0 });
+    try std.testing.expectError(error.OutOfMemory, hub.publish(failing.allocator(), "ch", "x"));
+}
+
+test "subscribe assigns each subscriber a distinct, increasing id" {
+    const allocator = std.testing.allocator;
+    var hub = Hub.init(allocator);
+    defer hub.deinit();
+
+    const a = try hub.subscribe(&.{"ch"}, defaultCapacity);
+    defer destroy(&hub, a);
+    const b = try hub.subscribe(&.{"ch"}, defaultCapacity);
+    defer destroy(&hub, b);
+
+    try std.testing.expect(a.id != b.id);
+    try std.testing.expect(b.id > a.id);
+}
+
+test "unsubscribe from a subset of channels leaves the rest delivering" {
+    const allocator = std.testing.allocator;
+    var hub = Hub.init(allocator);
+    defer hub.deinit();
+
+    const sub = try hub.subscribe(&.{ "weather", "orders", "alerts" }, defaultCapacity);
+    defer destroy(&hub, sub);
+
+    const removed = try hub.unsubscribe(sub.id, &.{"orders"});
+    try std.testing.expectEqual(@as(usize, 1), removed);
+
+    try std.testing.expectEqual(@as(usize, 0), try hub.publish(allocator, "orders", "too-late"));
+    try std.testing.expectEqual(@as(usize, 1), try hub.publish(allocator, "weather", "sunny"));
+    try std.testing.expectEqual(@as(usize, 1), try hub.publish(allocator, "alerts", "fire"));
+
+    const m1 = sub.waitMessage().?;
+    defer m1.deinit(allocator);
+    try std.testing.expectEqualStrings("sunny", m1.payload);
+    const m2 = sub.waitMessage().?;
+    defer m2.deinit(allocator);
+    try std.testing.expectEqualStrings("fire", m2.payload);
+
+    // The channel-name registry itself must not have been disturbed, so a
+    // brand new subscriber can still register under the vacated channel.
+    try std.testing.expectEqual(@as(usize, 0), hub.channelSubscriberCount("orders"));
+    try std.testing.expectEqual(@as(usize, 1), hub.channelSubscriberCount("weather"));
+}
+
+test "unsubscribe naming a channel the subscriber was never on counts zero, no error" {
+    const allocator = std.testing.allocator;
+    var hub = Hub.init(allocator);
+    defer hub.deinit();
+
+    const sub = try hub.subscribe(&.{"weather"}, defaultCapacity);
+    defer destroy(&hub, sub);
+
+    const removed = try hub.unsubscribe(sub.id, &.{"nonexistent"});
+    try std.testing.expectEqual(@as(usize, 0), removed);
+    try std.testing.expectEqual(@as(usize, 1), try hub.publish(allocator, "weather", "still-here"));
+    const msg = sub.waitMessage().?;
+    defer msg.deinit(allocator);
+}
+
+test "unsubscribe from every remaining channel closes the subscriber outright" {
+    const allocator = std.testing.allocator;
+    var hub = Hub.init(allocator);
+    defer hub.deinit();
+
+    const sub = try hub.subscribe(&.{ "a", "b" }, defaultCapacity);
+    defer hub.destroySubscriber(sub);
+
+    const removed = try hub.unsubscribe(sub.id, &.{ "a", "b" });
+    try std.testing.expectEqual(@as(usize, 2), removed);
+
+    try std.testing.expect(sub.isClosed());
+    try std.testing.expect(sub.waitMessage() == null);
+    try std.testing.expectEqual(@as(usize, 0), hub.channelSubscriberCount("a"));
+    try std.testing.expectEqual(@as(usize, 0), hub.channelSubscriberCount("b"));
+
+    // The id is gone too - a second unsubscribe by the same id must not
+    // find a dangling subscriber.
+    try std.testing.expectError(error.UnknownSubscriber, hub.unsubscribe(sub.id, &.{"a"}));
+
+    // Mirrors the normal disconnect path: `closeSubscriber` still runs
+    // afterward (e.g. from `handleSubscribe`'s push loop noticing the
+    // close) and must be a harmless no-op over an already-empty channel set.
+    hub.closeSubscriber(sub);
+}
+
+test "unsubscribe wakes a thread already blocked in waitMessage when it empties the set" {
+    const allocator = std.testing.allocator;
+    var hub = Hub.init(allocator);
+    defer hub.deinit();
+
+    const sub = try hub.subscribe(&.{"ch"}, defaultCapacity);
+    defer hub.destroySubscriber(sub);
+
+    const Waiter = struct {
+        fn run(s: *Subscriber, result: *?Message) void {
+            result.* = s.waitMessage();
+        }
+    };
+    var result: ?Message = undefined;
+    result = Message{ .channel = &.{}, .payload = &.{} };
+    const t = try std.Thread.spawn(.{}, Waiter.run, .{ sub, &result });
+    _ = try hub.unsubscribe(sub.id, &.{"ch"});
+    t.join();
+    try std.testing.expect(result == null);
+}
+
+test "unsubscribe rejects an id nobody ever issued" {
+    const allocator = std.testing.allocator;
+    var hub = Hub.init(allocator);
+    defer hub.deinit();
+
+    try std.testing.expectError(error.UnknownSubscriber, hub.unsubscribe(999, &.{"ch"}));
+}
+
+test "closeSubscriber removes the id, so a later unsubscribe by that id is rejected" {
+    const allocator = std.testing.allocator;
+    var hub = Hub.init(allocator);
+    defer hub.deinit();
+
+    const sub = try hub.subscribe(&.{"ch"}, defaultCapacity);
+    const id = sub.id;
+    hub.closeSubscriber(sub);
+    defer hub.destroySubscriber(sub);
+
+    try std.testing.expectError(error.UnknownSubscriber, hub.unsubscribe(id, &.{"ch"}));
+}