router/dealer: don't block callee session on slow caller; per-call retry queue (gammazero#324)

The dealer's yield() had a synchronous retry loop that ran on the
callee's session-handler goroutine. When the caller's outbound queue
filled (slow / unresponsive caller), that retry blocked for up to
sendResultDeadline (one minute), freezing all other communication for
the otherwise-healthy callee — even unrelated calls, subscriptions,
and meta procedures.

Fix: move the retry loop onto a fresh goroutine per invocation, with
a per-invocation in-order pending-yield queue tracked on the
invocation struct itself.

Properties of the new design:
  - Fast path unchanged: a YIELD that delivers on the first attempt
    finishes synchronously; no goroutine spawn, no queueing.
  - Slow path: the YIELD is appended to invocation.pendingYields and
    a single retry goroutine drains it. Subsequent YIELDs for the
    same invocation that arrive while a retry is in progress are
    appended (preserving in-call ordering, which progressive results
    depend on) rather than racing on out-of-band delivery.
  - Backoff: starts at yieldRetryDelay, doubles up to a 1 s cap, and
    resets to yieldRetryDelay whenever a queued YIELD is delivered
    (caller is making progress).
  - Shutdown: dealer gains a 'closing' channel that is signaled
    before actionChan is closed, so retry goroutines exit cleanly via
    select on closing rather than racing to a send-on-closed-channel
    panic.

Test: TestDealerYieldDoesNotBlockCalleeOnSlowCaller exercises the
exact scenario from gammazero#324 — a caller with a 1-slot outbound queue that
the test never drains, two CALLs whose RESULTs would queue, then a
SUBSCRIBE from the callee that must complete promptly. Without the
fix the test trips a 1-second deadline (callee deadlocked); with the
fix the SUBSCRIBE returns in milliseconds. The existing
TestProgressStress (16 progressive calls × variable chunk sizes,
caller pauses mid-stream) continues to pass — chunks stay in order
and counts match.

Author: Marko Petzold

router/dealer.go

@@ -62,6 +62,28 @@ type invocation struct {
 	inProgress  bool
 	timerCancel context.CancelFunc
 	options     wamp.Dict
+
+	// pendingYields buffers YIELD messages that arrived while a retry
+	// goroutine is draining the queue for this invocation. Drained in
+	// arrival order so progressive results stay correctly sequenced.
+	// Accessed only from the dealer actor goroutine.
+	pendingYields []pendingYield
+
+	// retrying is true while a retry goroutine is actively delivering
+	// queued YIELDs for this invocation. Prevents duplicate retry
+	// goroutines from racing on the same call.
+	retrying bool
+
+	// retryDeadline is the absolute time after which the retry goroutine
+	// stops trying and cancels the call instead.
+	retryDeadline time.Time
+}
+
+// pendingYield carries a queued YIELD with the cached progress flag so
+// the retry goroutine doesn't have to re-extract it from msg.Options.
+type pendingYield struct {
+	msg      *wamp.Yield
+	progress bool
 }
 
 type requestID struct {
@@ -98,6 +120,10 @@ type dealer struct {
 
 	actionChan chan func()
 	stopped    chan struct{}
+	// closing is signaled before actionChan is closed so background
+	// goroutines (e.g. yield-retry workers) can exit cleanly without
+	// racing on a send-on-closed-channel panic.
+	closing chan struct{}
 
 	// Generate registration IDs.
 	idGen *wamp.IDGen
@@ -139,6 +165,7 @@ func newDealer(logger stdlog.StdLog, strictURI, allowDisclose, debug bool) *deal
 		// channel is appropriate.
 		actionChan: make(chan func()),
 		stopped:    make(chan struct{}),
+		closing:    make(chan struct{}),
 
 		idGen: new(wamp.IDGen),
 		prng:  rand.New(rand.NewSource(time.Now().Unix())), //nolint:gosec // used for call invocation
@@ -318,52 +345,159 @@ func (d *dealer) cancel(caller *wamp.Session, msg *wamp.Cancel) {
 // yield handles the result of successfully processing and finishing the
 // execution of a call, send from callee to dealer.
 //
-// If the RESULT could not be sent to the caller because the caller was blocked
-// (send queue full), then retry sending until timeout. If timeout while trying
-// to send RESULT, then cancel call.
+// Fast path: deliver the RESULT on the dealer goroutine via syncYield. If
+// the caller's outbound queue is full, the YIELD is enqueued on the
+// invocation's per-call retry queue and a retry goroutine is started
+// (one per invocation, at most). All subsequent YIELDs for the same
+// invocation that arrive while a retry is in progress are appended to
+// the queue and drained in arrival order, so progressive results stay
+// correctly sequenced.
+//
+// Critically, this function returns to the callee's session-handler
+// goroutine as soon as the dealer has either delivered or queued the
+// YIELD. The retry loop runs on a fresh goroutine. Without this split,
+// a single unresponsive caller would freeze all further communication
+// for an otherwise healthy callee — see gammazero/nexus#324.
 func (d *dealer) yield(callee *wamp.Session, msg *wamp.Yield) {
 	if callee == nil || msg == nil {
 		panic("dealer.Yield with nil session or message")
 	}
 
-	var again bool
 	progress, _ := msg.Options[wamp.OptProgress].(bool)
+	invkReqID := requestID{session: callee.ID, request: msg.Request}
 
+	var startRetry bool
 	done := make(chan struct{})
-	d.actionChan <- func() {
-		again = d.syncYield(callee, msg, progress, true)
+	action := func() {
+		invk, ok := d.invocations[invkReqID]
+		if ok && invk.retrying {
+			// A retry goroutine is already draining this call's queue;
+			// preserve in-call order by appending instead of trying to
+			// deliver out-of-band.
+			invk.pendingYields = append(invk.pendingYields, pendingYield{msg: msg, progress: progress})
+			done <- struct{}{}
+			return
+		}
+		// No active retry — try direct delivery first.
+		if !d.syncYield(callee, msg, progress, true) {
+			done <- struct{}{}
+			return
+		}
+		// Caller's queue is full. Queue this YIELD and start a retry
+		// goroutine. Re-fetch invk because syncYield may have changed
+		// state (it shouldn't have deleted the entry on canRetry=true,
+		// but be defensive).
+		invk, ok = d.invocations[invkReqID]
+		if !ok {
+			// Invocation gone (raced with cancel/timeout) — nothing to retry.
+			done <- struct{}{}
+			return
+		}
+		invk.pendingYields = append(invk.pendingYields, pendingYield{msg: msg, progress: progress})
+		invk.retrying = true
+		invk.retryDeadline = time.Now().Add(sendResultDeadline)
+		startRetry = true
 		done <- struct{}{}
 	}
+	select {
+	case d.actionChan <- action:
+	case <-d.closing:
+		return
+	}
 	<-done
 
-	// If blocked, retry
-	if again {
-		retry := true
-		delay := yieldRetryDelay
-		start := time.Now()
-		// Retry processing YIELD until caller gone or deadline reached.
-		for {
-			if d.debug {
-				d.log.Println("Retry sending RESULT after", delay)
-			}
-			<-time.After(delay)
-			// Do not retry if the elapsed time exceeds deadline.
-			if time.Since(start) >= sendResultDeadline {
-				retry = false
-			}
-			d.actionChan <- func() {
-				again = d.syncYield(callee, msg, progress, retry)
-				done <- struct{}{}
-			}
-			<-done
-			if !again {
-				break
-			}
+	if startRetry {
+		go d.drainPendingYields(callee, invkReqID)
+	}
+}
+
+// drainPendingYields runs on a dedicated goroutine per invocation while
+// that invocation has YIELDs queued for retry. It serializes through the
+// dealer actor (actionChan), so dealer state stays single-threaded.
+//
+// Backoff: starts at yieldRetryDelay, doubles on each failed attempt
+// up to a 1-second ceiling, and resets to yieldRetryDelay whenever a
+// queued YIELD is successfully delivered (caller is making progress).
+//
+// Exits when the queue is drained, the invocation is gone, or the dealer
+// is closing.
+func (d *dealer) drainPendingYields(callee *wamp.Session, invkReqID requestID) {
+	delay := yieldRetryDelay
+	for {
+		if d.debug {
+			d.log.Println("Retry sending RESULT after", delay)
+		}
+		// Sleep, but watch for dealer shutdown so we never block the
+		// realm.close path.
+		timer := time.NewTimer(delay)
+		select {
+		case <-timer.C:
+		case <-d.closing:
+			timer.Stop()
+			return
+		}
+
+		var keepGoing, delivered bool
+		done := make(chan struct{})
+		select {
+		case d.actionChan <- func() {
+			keepGoing, delivered = d.tryDrainOneYield(callee, invkReqID)
+			done <- struct{}{}
+		}:
+		case <-d.closing:
+			return
+		}
+		<-done
+
+		if !keepGoing {
+			return
+		}
+		if delivered {
+			// Caller is making progress; retry the next item quickly.
+			delay = yieldRetryDelay
+		} else if delay < time.Second {
 			delay *= 2
 		}
 	}
 }
 
+// tryDrainOneYield is invoked on the dealer actor goroutine. It attempts
+// to deliver the head of the invocation's pending-yield queue.
+// Returns:
+//
+//	keepGoing — true if the retry goroutine should keep looping.
+//	delivered — true if the head was delivered on this attempt
+//	            (whether or not more remain).
+func (d *dealer) tryDrainOneYield(callee *wamp.Session, invkReqID requestID) (keepGoing, delivered bool) {
+	invk, ok := d.invocations[invkReqID]
+	if !ok {
+		// Invocation gone — call canceled, callee disconnected, etc.
+		return false, false
+	}
+	if len(invk.pendingYields) == 0 {
+		invk.retrying = false
+		return false, false
+	}
+	head := invk.pendingYields[0]
+	canRetry := time.Now().Before(invk.retryDeadline)
+	if d.syncYield(callee, head.msg, head.progress, canRetry) {
+		// Still blocked — keep the head and retry after backoff.
+		return true, false
+	}
+	// Delivered (or canceled at deadline). Pop the head. For a final
+	// non-progress YIELD, syncYield's deferred cleanup deletes the
+	// invocation; re-fetch before touching it again.
+	invk.pendingYields = invk.pendingYields[1:]
+	invk, ok = d.invocations[invkReqID]
+	if !ok || len(invk.pendingYields) == 0 {
+		if ok {
+			invk.retrying = false
+		}
+		return false, true
+	}
+	return true, true
+}
+
 // error handles an invocation error returned by the callee.
 func (d *dealer) error(callee *wamp.Session, msg *wamp.Error) {
 	if msg == nil {
@@ -402,6 +536,10 @@ func (d *dealer) removeSession(sess *wamp.Session) {
 
 // close stops the dealer, letting already queued actions finish.
 func (d *dealer) close() {
+	// Signal background goroutines (e.g. yield-retry workers) before
+	// closing actionChan, so they exit via the closing channel rather
+	// than racing on a send-on-closed-channel panic.
+	close(d.closing)
 	close(d.actionChan)
 	<-d.stopped
 	if d.debug {

router/router_test.go

@@ -255,6 +255,71 @@ func TestPublishNoAcknowledge(t *testing.T) {
 	})
 }
 
+// TestDealerYieldDoesNotBlockCalleeOnSlowCaller pins gammazero/nexus#324:
+// when a caller's outbound queue fills (the caller is slow or unresponsive
+// reading its Recv channel), the dealer's RESULT-delivery retry loop must
+// not stall the callee's session goroutine. Otherwise a single bad caller
+// freezes communication for the callee, even though the callee has
+// nothing wrong with it.
+//
+// Without the fix this test trips its 1-second deadline; with the fix the
+// callee remains responsive within milliseconds.
+func TestDealerYieldDoesNotBlockCalleeOnSlowCaller(t *testing.T) {
+	r := newTestRouter(t)
+
+	// Caller with the smallest practical outbound queue. After WELCOME is
+	// drained the queue is empty; one stalled RESULT will fill it.
+	callerClient, callerServer := transport.LinkedPeersQSize(1)
+	go func() {
+		callerClient.Send() <- &wamp.Hello{Realm: testRealm, Details: clientRoles}
+	}()
+	require.NoError(t, r.Attach(callerServer))
+	welcome := <-callerClient.Recv()
+	_, ok := welcome.(*wamp.Welcome)
+	require.True(t, ok, "expected WELCOME, got %T", welcome)
+
+	// Callee with the normal default queue.
+	callee := testClient(t, r)
+
+	const proc = wamp.URI("nexus.test.slowcaller.proc")
+	callee.Send() <- &wamp.Register{Request: 1, Procedure: proc}
+	msg, err := wamp.RecvTimeout(callee, time.Second)
+	require.NoError(t, err)
+	_, ok = msg.(*wamp.Registered)
+	require.True(t, ok, "expected REGISTERED, got %T", msg)
+
+	deliverYield := func(callRequest wamp.ID) {
+		t.Helper()
+		callerClient.Send() <- &wamp.Call{Request: callRequest, Procedure: proc}
+		msg, err := wamp.RecvTimeout(callee, time.Second)
+		require.NoErrorf(t, err, "callee never received INVOCATION for call %d", callRequest)
+		inv, ok := msg.(*wamp.Invocation)
+		require.True(t, ok, "expected INVOCATION, got %T", msg)
+		callee.Send() <- &wamp.Yield{Request: inv.Request}
+	}
+
+	// First call: RESULT 1 fills caller's 1-slot queue (test never reads it).
+	deliverYield(1)
+
+	// Second call: RESULT 2 cannot be queued — caller is now blocked.
+	// dealer.yield enters its retry loop. With the bug, this retry runs on
+	// the callee's session goroutine and blocks it.
+	deliverYield(2)
+
+	// The callee must still be able to do other work while the dealer
+	// retries delivering RESULT 2 in the background. SUBSCRIBE → SUBSCRIBED
+	// must complete promptly; without the fix it waits for the retry
+	// deadline (~1 minute).
+	callee.Send() <- &wamp.Subscribe{Request: 100, Topic: testTopic}
+	select {
+	case msg := <-callee.Recv():
+		_, ok := msg.(*wamp.Subscribed)
+		require.True(t, ok, "expected SUBSCRIBED, got %T", msg)
+	case <-time.After(time.Second):
+		t.Fatal("callee blocked for >1s while caller's queue is full — see #324")
+	}
+}
+
 func TestRouterCall(t *testing.T) {
 	synctest.Test(t, func(t *testing.T) {
 		r := newTestRouter(t)