Repository navigation
Let typed bus subscribers be unsubscribed - #7013
Open
iliya-vidrush wants to merge 2 commits into
Open
iliya-vidrush wants to merge 2 commits into
iliya-vidrush wants to merge 2 commits into
Conversation
Publish delivered to each subscriber with `ch <- event` while holding the bus read lock. A consumer that stopped draining filled its 10k buffer and parked the publisher on that send, so the read lock was never released. Subscribe and Unsubscribe then blocked on Lock(), and because a pending writer excludes new readers, every other Publish blocked behind them. One stuck consumer froze the entire bus. That path is reachable today: the socket handler writes to websockets inline with no write deadline, so a peer that stops reading stalls it and `sst dev` hangs with no error. Commit 9ec92b5 ("dev: potentially fixed freezing issue") added Unsubscribe to make abandoned subscribers rarer, but left the blocking send itself in place. Delivery is now non-blocking for every subscriber, so the read lock is only ever held for bounded work. Policy is expressed as buffer depth: typed subscribers carry control flow and get 16384 slots, while the firehose keeps 10000 and is cheap to lose. Overflow is counted per subscriber and surfaced through Stats(), so a wedged consumer can be named rather than merely detected. Also fixes Unsubscribe never removing typed subscribers: the inner loop reassigned a local slice and never wrote it back to the map. That path was unreachable from outside the package, which is why it went unnoticed. Callers that close their channel now unsubscribe first, matching the ordering diff.go already used. Benchmarks are included and report drops/op alongside ns/op, so a fast number produced by discarding events cannot pass unnoticed.
Subscribe returns <-chan interface{} but Unsubscribe took chan
interface{}, so a Subscribe handle could not be passed back. The typed
removal path was therefore unreachable from outside the package, which
is how it stayed broken from the day it was written.
Unsubscribe now takes a receive-only channel. A bidirectional channel
from SubscribeAll is assignable to that type, so every existing caller
compiles unchanged.
|
Thanks for the fix, @iliya-vidrush. This PR, with #7012 under it, is now in the community fork, The fork accepts issues and PRs, if you have other fixes you'd like carried. |
This was referenced Oct 10, 2026
AlperSakarya
pushed a commit
to AlperSakarya/sst
that referenced
this pull request
Oct 10, 2026
Subscribe returns <-chan interface{} but Unsubscribe took chan
interface{}, so a Subscribe handle could not be passed back. The typed
removal path was therefore unreachable from outside the package, which
is how it stayed broken from the day it was written.
Unsubscribe now takes a receive-only channel. A bidirectional channel
from SubscribeAll is assignable to that type, so every existing caller
compiles unchanged.
Carried from anomalyco/sst#7013.
This branch has not been deployed
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Stacked on #7012 — its commit is included here until that merges.
Summary
Subscribereturned a receive-only channel whileUnsubscribetook a bidirectional one, so a typed subscription could never be removed.Unsubscribenow accepts a receive-only channel;SubscribeAllhandles are assignable to it, so no caller changes.Validation
Unsubscribebody withreceived {after} after unsubscribe, and passes here.go build ./...confirms every existing caller compiles unchanged;go test ./pkg/bus/ -racepasses.Flow
%%{init: {'themeVariables': {'fontSize': '12px'}}}%% flowchart TB classDef added fill:#dcfce7,stroke:#16a34a,color:#166534 classDef changed fill:#fef3c7,stroke:#d97706,color:#92400e sub["Subscribe(eventTypes...)"] --> handle["receive-only handle"] all["SubscribeAll()"] --> bidi["bidirectional handle"] handle ==>|"now accepted"| unsub["Unsubscribe(ch)"] bidi -->|"assignable, unchanged"| unsub unsub ==> recv["subscriber.recv()"] recv --> removal["bus.subscribers[t] removal"] class unsub,removal changed class recv added