Fix bad merge for prior PR which caused more stalls (#3630)

Due to a bad merge, timeouts were not early-returning and so could cause
a stall in the roomserver. Fix the bad merge and early-out.

Minor fix to a bad merge of my change for #3484 

This also incorporates a suggested tweak to the prior PR #3588 to avoid
holding the lock unnecessarily.

### Pull Request Checklist

<!-- Please read
https://matrix-org.github.io/dendrite/development/contributing before
submitting your pull request -->

* [x] I have added Go unit tests or [Complement integration
tests](https://github.com/matrix-org/complement) for this PR _or_ I have
justified why this PR doesn't need tests
* [x] Pull request includes a [sign off
below](https://element-hq.github.io/dendrite/development/contributing#sign-off)
_or_ I have already signed off privately

Signed-off-by: `Vivianne Langdon <[email protected]>`

---------

Signed-off-by: Vivianne Langdon <[email protected]>

Author: viviicat

roomserver/internal/input/input.go

@@ -292,11 +292,17 @@ func (w *worker) _next() {
 		// 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.Act(nil, w._next)
 			w.Unlock()
+			w.Act(nil, w._next)
 			return
 		}
+
+		if err = w.subscription.Unsubscribe(); err != nil {
+			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()
 		defer w.Unlock()