diff --git a/roomserver/internal/input/input.go b/roomserver/internal/input/input.go index 18c622ef..7abc6ffd 100644 --- a/roomserver/internal/input/input.go +++ b/roomserver/internal/input/input.go @@ -289,10 +289,10 @@ func (w *worker) _next() { // down the subscriber to free up resources. It'll get started // again if new activity happens. w.Lock() - defer w.Unlock() // inside the lock, let's check if the ephemeral consumer saw something new! // If so, we do have new messages after all, they just came at a bad time. if w.ephemeralSeq > w.durableSeq { + w.Unlock(); w.Act(nil, w._next) return } @@ -301,6 +301,7 @@ func (w *worker) _next() { logrus.WithError(err).Errorf("Failed to unsubscribe to stream for room %q", w.roomID) } w.subscription = nil + w.Unlock() return case nats.ErrConsumerDeleted, nats.ErrConsumerNotFound: w.Lock()