[client] let engine context unblock WaitStreamConnected

WaitStreamConnected only watched the signal client's own context, which
derives from the parent engineCtx rather than the engine's run context.
A Start blocked here (signal stream not yet up) could therefore not be
released by Engine.Stop, since Stop only cancels the engine's run
context.

Pass a context into WaitStreamConnected and select on it too, and have
the engine pass e.ctx, so Stop cancelling e.ctx unblocks a parked Start.
Update the Client interface, the mock, and callers accordingly.
This commit is contained in:
Zoltán Papp
2026-06-16 13:11:46 +02:00
parent c370c72d93
commit 002e0b036f
5 changed files with 9 additions and 7 deletions
+2 -1
View File
@@ -1776,7 +1776,8 @@ func (e *Engine) receiveSignalEvents() {
} }
}() }()
e.signal.WaitStreamConnected() // todo: consider to remove this blocker. I do not see benefit to block the Start operations
e.signal.WaitStreamConnected(e.ctx)
} }
func (e *Engine) parseNATExternalIPMappings() []string { func (e *Engine) parseNATExternalIPMappings() []string {
+1 -1
View File
@@ -33,7 +33,7 @@ type Client interface {
Receive(ctx context.Context, msgHandler func(msg *proto.Message) error) error Receive(ctx context.Context, msgHandler func(msg *proto.Message) error) error
Ready() bool Ready() bool
IsHealthy() bool IsHealthy() bool
WaitStreamConnected() WaitStreamConnected(context.Context)
SendToStream(msg *proto.EncryptedMessage) error SendToStream(msg *proto.EncryptedMessage) error
Send(msg *proto.Message) error Send(msg *proto.Message) error
SetOnReconnectedListener(func()) SetOnReconnectedListener(func())
+3 -3
View File
@@ -65,7 +65,7 @@ var _ = Describe("GrpcClient", func() {
return return
} }
}() }()
clientA.WaitStreamConnected() clientA.WaitStreamConnected(context.Background())
// connect PeerB to Signal // connect PeerB to Signal
keyB, _ := wgtypes.GenerateKey() keyB, _ := wgtypes.GenerateKey()
@@ -91,7 +91,7 @@ var _ = Describe("GrpcClient", func() {
} }
}() }()
clientB.WaitStreamConnected() clientB.WaitStreamConnected(context.Background())
// PeerA initiates ping-pong // PeerA initiates ping-pong
err := clientA.Send(&sigProto.Message{ err := clientA.Send(&sigProto.Message{
@@ -129,7 +129,7 @@ var _ = Describe("GrpcClient", func() {
return return
} }
}() }()
client.WaitStreamConnected() client.WaitStreamConnected(context.Background())
Expect(client).NotTo(BeNil()) Expect(client).NotTo(BeNil())
}) })
}) })
+2 -1
View File
@@ -282,7 +282,7 @@ func (c *GrpcClient) IsHealthy() bool {
} }
// WaitStreamConnected waits until the client is connected to the Signal stream // WaitStreamConnected waits until the client is connected to the Signal stream
func (c *GrpcClient) WaitStreamConnected() { func (c *GrpcClient) WaitStreamConnected(ctx context.Context) {
if c.status == StreamConnected { if c.status == StreamConnected {
return return
@@ -290,6 +290,7 @@ func (c *GrpcClient) WaitStreamConnected() {
ch := c.getStreamStatusChan() ch := c.getStreamStatusChan()
select { select {
case <-ctx.Done():
case <-c.ctx.Done(): case <-c.ctx.Done():
case <-ch: case <-ch:
} }
+1 -1
View File
@@ -55,7 +55,7 @@ func (sm *MockClient) Ready() bool {
return sm.ReadyFunc() return sm.ReadyFunc()
} }
func (sm *MockClient) WaitStreamConnected() { func (sm *MockClient) WaitStreamConnected(context.Context) {
if sm.WaitStreamConnectedFunc == nil { if sm.WaitStreamConnectedFunc == nil {
return return
} }