router: a stale response callback can permanently deadlock the router's message dispatch loop

What happened:

listener.Process() is the router's single message dispatch goroutine, started from router.go:52.

When a message has a ParentID, HandleMessage() invokes the registered callback synchronously. The callback registered by servicebus.GoToTarget performs a blocking send on an unbuffered channel:

if stop != nil {
    listener.MessageHandlerInstance.SetCallback(messageID, func(message *model.Message) {
        response = message
        stop <- struct{}{}
    })
    <-stop
    listener.MessageHandlerInstance.DelCallback(messageID)
}

The problem is that stop has one receiver but two possible senders: the response callback and the timeout path in Rest.Forward.

This creates two ways for the callback to block permanently:

1. Timeout and response race If the timeout signal and the edge response arrive at nearly the same time, <-stop consumes one of them. If it consumes the timeout signal, the response callback is left trying to send to a channel with no receiver.

2. DelCallback window After <-stop returns, the callback is still registered until DelCallback() executes. If a response arrives in that small window, HandleMessage() invokes the callback even though GoToTarget is no longer receiving from stop.

In both cases, the callback blocks inside listener.Process(). Since callback execution is synchronous, Process() never gets back to beehiveContext.Receive(), so the router stops processing all subsequent messages.

This is therefore not limited to a single request. A stale response can stall the entire router module until CloudCore is restarted.

There are also two related issues in the same code path:

response can be written by the callback while being read by GoToTarget after a timeout wakeup without synchronization. This is a data race under -race.If GoToTarget remains blocked, DelCallback() is never reached, so the corresponding entry remains in callbackHandlers.

Relation to #7016:

Making the stop signal non-blocking alone does not completely solve this.

SetCallback() happens before <-stop. If the timeout signal is sent during that window and the send is made non-blocking, the signal can simply be dropped. GoToTarget then waits forever with the callback still registered, and a later response can still block listener.Process().

The callback delivery and callback lifetime therefore need to be handled together.

What should happen:

A late, duplicate, or otherwise stale response should never be able to block the router's message dispatch loop.

Once the request has timed out or the receiver has gone away, a response should be safely ignored rather than blocking the dispatch goroutine. Callback registration should also be cleaned up when the waiting goroutine exits, including the timeout/abandoned-request path.

How to reproduce:

1. Configure a router rule with a REST source and a servicebus target.
2. Send a request whose edge response arrives at approximately the same time as restTimeout (60s by default).
3. The request returns 408, but the router stops processing subsequent messages.
4. A goroutine profile shows listener.Process blocked on a channel send inside the callback registered by servicebus.GoToTarget.

The following unit test reproduces the underlying problem without requiring a cluster:

func TestHandleMessageWedgesDispatchLoopOnStaleCallback(t *testing.T) {
	stop := make(chan struct{}) // unbuffered, nobody receiving

	MessageHandlerInstance.SetCallback("parent-1", func(*model.Message) {
		stop <- struct{}{} // same blocking send used by servicebus.GoToTarget
	})

	reply := model.NewMessage("parent-1")
	reply.Header.ParentID = "parent-1"

	done := make(chan struct{})
	go func() {
		_ = MessageHandlerInstance.HandleMessage(reply)
		close(done)
	}()

	select {
	case <-done:
	case <-time.After(2 * time.Second):
		t.Fatal("HandleMessage blocked on a stale callback; " +
			"Process() invokes callbacks synchronously, so this can deadlock " +
			"the router's message dispatch loop")
	}
}

Environment

• KubeEdge: master (172a4b5d3)
• Affected area: cloud/pkg/router
• Feature: Rule / RuleEndpoint
• Related: #7016