From a91ac6246864b7fbc149ba3c0dd8b3d1fecb7926 Mon Sep 17 00:00:00 2001 From: DualFroz Date: Tue, 8 Sep 2026 17:41:29 +0200 Subject: [PATCH] feat: add Publisher and run to pubsub --- src/pubsub.zig | 218 +++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 218 insertions(+) diff --git a/src/pubsub.zig b/src/pubsub.zig index 18652b5..e6395f0 100644 --- a/src/pubsub.zig +++ b/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"})); +}