mirror of
https://github.com/netbirdio/netbird.git
synced 2026-10-27 06:49:07 +01:00
[client] Wait for signal receive watchdog to stop before reconnect
The per-stream watchReceiveStream goroutine was started fire-and-forget and never joined. On reconnect a lingering watchdog could still flip shared client state (receiveStalled, the disconnect notifier) on the freshly established stream, since cancelStream only cancels its own stream context. Track the watchdog with a WaitGroup and wait for it to exit (after cancelling its stream) before the operation returns, so each reconnect starts with no stale watchdog.
This commit is contained in:
@@ -85,6 +85,7 @@ type GrpcClient struct {
|
|||||||
// receive backpressure as a dead stream: reconnecting cannot help, since the
|
// receive backpressure as a dead stream: reconnecting cannot help, since the
|
||||||
// new stream feeds the same worker, and only triggers a reconnect storm.
|
// new stream feeds the same worker, and only triggers a reconnect storm.
|
||||||
receiveHandoffBlocked atomic.Bool
|
receiveHandoffBlocked atomic.Bool
|
||||||
|
watchdogWg sync.WaitGroup
|
||||||
}
|
}
|
||||||
|
|
||||||
// NewClient creates a new Signal client
|
// NewClient creates a new Signal client
|
||||||
@@ -200,10 +201,18 @@ func (c *GrpcClient) Receive(ctx context.Context, msgHandler func(msg *proto.Mes
|
|||||||
// Guard the receive direction: the transport can stay healthy while the
|
// Guard the receive direction: the transport can stay healthy while the
|
||||||
// server stops delivering messages. The watchdog reconnects via cancelStream.
|
// server stops delivering messages. The watchdog reconnects via cancelStream.
|
||||||
c.markReceived()
|
c.markReceived()
|
||||||
go c.watchReceiveStream(streamCtx, cancelStream)
|
c.watchdogWg.Add(1)
|
||||||
|
go func() {
|
||||||
|
defer c.watchdogWg.Done()
|
||||||
|
c.watchReceiveStream(streamCtx, cancelStream)
|
||||||
|
}()
|
||||||
|
|
||||||
// start receiving messages from the Signal stream (from other peers through signal)
|
// start receiving messages from the Signal stream (from other peers through signal)
|
||||||
err = c.receive(stream)
|
err = c.receive(stream)
|
||||||
|
|
||||||
|
cancelStream()
|
||||||
|
c.watchdogWg.Wait()
|
||||||
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
// Check the parent context, not streamCtx: a watchdog-triggered
|
// Check the parent context, not streamCtx: a watchdog-triggered
|
||||||
// cancelStream must reconnect, only a parent cancel is shutdown.
|
// cancelStream must reconnect, only a parent cancel is shutdown.
|
||||||
|
|||||||
Reference in New Issue
Block a user