mirror of
https://github.com/netbirdio/netbird.git
synced 2026-09-25 16:19:07 +02:00
fix(client): track which connection run is current in the daemon
Nothing recorded which run of the connection was current, so three defects followed from the same gap. A ConnectClient is single-use, and the daemon builds a fresh one per outer-retry turn (server.go connect). Each turn overwrote s.connectClient and nothing stopped the one it replaced — the outgoing run loop had returned, which is what brought control back to the retry, but that was assumed rather than enforced, and any teardown its error path left half-done got no second chance. cleanupConnection read s.connectClient, cancelled, then stopped that engine. Nothing established the client it read was still current by the time it stopped it, so a teardown could target a client a newer turn had already replaced and leave the live one running untracked. Down's wait on clientGiveUpChan and Up's refusal to start a second loop kept the window narrow, but by arrangement rather than by construction. Third, the engine was stopped twice concurrently: actCancel woke the run loop, which stops the engine on its way out, while cleanupConnection stopped the same engine directly. The TODO there said ConnectClient.Stop was the right call and that its unbounded wait was what ruled it out. RunSupervisor records the generation of the current run. Publish refuses a client from a superseded run and stops the client it displaces, so no ConnectClient is dropped without being stopped. Stop invalidates whatever run is in flight, stops the published client and waits for the run to exit. ConnectClient.StopWithContext bounds that wait, which removes the TODO's obstacle: cleanupConnection now hands the run loop sole ownership of engine shutdown and passes Down's 5s budget down. Stop() keeps its signature and its unbounded wait, so callers outside this change are untouched. embed.Client.Stop had built the same bound by hand with a goroutine and a select purely to watch its caller's context; it passes the context down instead. clientGiveUpChan and connectClient are gone — the supervisor answers both. The MDM restart path drops its hand-rolled 10s channel wait for the same Stop, which additionally stops the client the previous run left behind. Its deliberate choice to leave clientRunning set is unchanged. Down now waits inside cleanupConnection, under s.mutex, where it previously waited after releasing it. That is what pins the client being stopped to the one current when the call started; the cost is that Down can hold the mutex for up to its 5s budget. Found while fixing the iOS wifi-to-cellular black-hole (#7329), which was the same class of defect in the mobile SDKs. No bug report backs the daemon findings — they are read off the code, and the narrow windows above may be why they have not been observed.
This commit is contained in:
Zoltán Papp
@@ -599,11 +599,25 @@ func (c *ConnectClient) Status() StatusType {
|
||||
}
|
||||
|
||||
func (c *ConnectClient) Stop() error {
|
||||
return c.StopWithContext(context.Background())
|
||||
}
|
||||
|
||||
// StopWithContext cancels the run loop and waits for it to exit, giving up the
|
||||
// wait when ctx is done and returning ctx.Err(). Giving up only abandons the
|
||||
// wait: the run stays cancelled and finishes its teardown in the background, so
|
||||
// a caller that returns early must not assume the engine is already gone.
|
||||
func (c *ConnectClient) StopWithContext(ctx context.Context) error {
|
||||
c.runCancel()
|
||||
if c.runStarted.Load() {
|
||||
<-c.runExited
|
||||
if !c.runStarted.Load() {
|
||||
return nil
|
||||
}
|
||||
|
||||
select {
|
||||
case <-c.runExited:
|
||||
return nil
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// SetSyncResponsePersistence enables or disables sync response persistence.
|
||||
|
||||
@@ -0,0 +1,132 @@
|
||||
package internal
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
// RunSupervisor tracks which run of a logical connection is current.
|
||||
//
|
||||
// A ConnectClient is single-use, so every attempt builds a fresh one. Without a
|
||||
// record of which attempt is current, a stale one can publish its client over a
|
||||
// newer one's, and a teardown can target a client that has already been
|
||||
// replaced — leaving the live one running with nothing tracking it.
|
||||
//
|
||||
// A run claims a generation with Begin, publishes its client with Publish, and
|
||||
// closes the channel Begin returned when it exits. Publish refuses a client from
|
||||
// a run that is no longer current and stops the client it displaces, so no
|
||||
// ConnectClient is dropped without being stopped.
|
||||
//
|
||||
// The zero value is ready to use.
|
||||
type RunSupervisor struct {
|
||||
mu sync.Mutex
|
||||
generation uint64
|
||||
current *ConnectClient
|
||||
done chan struct{}
|
||||
}
|
||||
|
||||
// Begin claims a generation for a starting run. The caller must close the
|
||||
// returned channel when the run exits, whether or not it published a client.
|
||||
func (s *RunSupervisor) Begin() (uint64, chan struct{}) {
|
||||
done := make(chan struct{})
|
||||
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
|
||||
s.generation++
|
||||
s.done = done
|
||||
|
||||
return s.generation, done
|
||||
}
|
||||
|
||||
// Publish installs cc as the current client and reports whether it took effect.
|
||||
// It returns false once a newer Begin or a Stop has superseded the generation,
|
||||
// and the caller must then abandon its startup. A client it displaces within the
|
||||
// same run is stopped, with stopCtx bounding that wait.
|
||||
func (s *RunSupervisor) Publish(ctx context.Context, generation uint64, cc *ConnectClient) bool {
|
||||
s.mu.Lock()
|
||||
if s.generation != generation {
|
||||
s.mu.Unlock()
|
||||
return false
|
||||
}
|
||||
displaced := s.current
|
||||
s.current = cc
|
||||
s.mu.Unlock()
|
||||
|
||||
if displaced != nil && displaced != cc {
|
||||
if err := displaced.StopWithContext(ctx); err != nil {
|
||||
log.Warnf("stopping the displaced connect client: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
return true
|
||||
}
|
||||
|
||||
// Current returns the published client, or nil while no run has published one.
|
||||
func (s *RunSupervisor) Current() *ConnectClient {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
|
||||
return s.current
|
||||
}
|
||||
|
||||
// Done returns the channel the current run closes when it exits, or nil when no
|
||||
// run has been started. Callers that only need a yes/no answer should use Alive.
|
||||
func (s *RunSupervisor) Done() <-chan struct{} {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
|
||||
return s.done
|
||||
}
|
||||
|
||||
// Alive reports whether a run has claimed a generation and not yet signalled its
|
||||
// exit.
|
||||
func (s *RunSupervisor) Alive() bool {
|
||||
s.mu.Lock()
|
||||
done := s.done
|
||||
s.mu.Unlock()
|
||||
|
||||
if done == nil {
|
||||
return false
|
||||
}
|
||||
|
||||
select {
|
||||
case <-done:
|
||||
return false
|
||||
default:
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
// Stop invalidates any run in flight, stops the published client, and waits for
|
||||
// the run to signal its exit. It gives up the wait when ctx is done and returns
|
||||
// ctx.Err(); the run stays cancelled and finishes tearing down in the
|
||||
// background, so an early return does not mean the engine is gone.
|
||||
func (s *RunSupervisor) Stop(ctx context.Context) error {
|
||||
s.mu.Lock()
|
||||
s.generation++
|
||||
cc := s.current
|
||||
done := s.done
|
||||
s.current = nil
|
||||
s.done = nil
|
||||
s.mu.Unlock()
|
||||
|
||||
if cc != nil {
|
||||
if err := cc.StopWithContext(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
if done == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
select {
|
||||
case <-done:
|
||||
return nil
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,116 @@
|
||||
package internal
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func newSupervisedClient() *ConnectClient {
|
||||
return NewConnectClient(context.Background(), nil, nil)
|
||||
}
|
||||
|
||||
func TestRunSupervisorPublishRejectsSupersededRun(t *testing.T) {
|
||||
var s RunSupervisor
|
||||
ctx := context.Background()
|
||||
|
||||
staleGen, staleDone := s.Begin()
|
||||
defer close(staleDone)
|
||||
|
||||
freshGen, freshDone := s.Begin()
|
||||
defer close(freshDone)
|
||||
|
||||
fresh := newSupervisedClient()
|
||||
require.True(t, s.Publish(ctx, freshGen, fresh))
|
||||
|
||||
assert.False(t, s.Publish(ctx, staleGen, newSupervisedClient()))
|
||||
assert.Same(t, fresh, s.Current())
|
||||
}
|
||||
|
||||
func TestRunSupervisorPublishStopsDisplacedClient(t *testing.T) {
|
||||
var s RunSupervisor
|
||||
ctx := context.Background()
|
||||
|
||||
generation, done := s.Begin()
|
||||
defer close(done)
|
||||
|
||||
displaced := newSupervisedClient()
|
||||
require.True(t, s.Publish(ctx, generation, displaced))
|
||||
|
||||
replacement := newSupervisedClient()
|
||||
require.True(t, s.Publish(ctx, generation, replacement))
|
||||
|
||||
assert.Same(t, replacement, s.Current())
|
||||
assert.Error(t, displaced.ctx.Err(), "the displaced client's run context should be cancelled")
|
||||
}
|
||||
|
||||
func TestRunSupervisorStopInvalidatesRunInFlight(t *testing.T) {
|
||||
var s RunSupervisor
|
||||
ctx := context.Background()
|
||||
|
||||
generation, done := s.Begin()
|
||||
close(done)
|
||||
|
||||
require.NoError(t, s.Stop(ctx))
|
||||
|
||||
assert.False(t, s.Publish(ctx, generation, newSupervisedClient()))
|
||||
assert.Nil(t, s.Current())
|
||||
}
|
||||
|
||||
func TestRunSupervisorStopWaitsForRunExit(t *testing.T) {
|
||||
var s RunSupervisor
|
||||
|
||||
_, done := s.Begin()
|
||||
|
||||
stopped := make(chan error, 1)
|
||||
go func() { stopped <- s.Stop(context.Background()) }()
|
||||
|
||||
select {
|
||||
case <-stopped:
|
||||
t.Fatal("Stop returned before the run signalled its exit")
|
||||
case <-time.After(50 * time.Millisecond):
|
||||
}
|
||||
|
||||
close(done)
|
||||
|
||||
select {
|
||||
case err := <-stopped:
|
||||
require.NoError(t, err)
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("Stop did not return after the run exited")
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunSupervisorStopGivesUpWaitOnContext(t *testing.T) {
|
||||
var s RunSupervisor
|
||||
|
||||
_, done := s.Begin()
|
||||
defer close(done)
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
|
||||
defer cancel()
|
||||
|
||||
assert.ErrorIs(t, s.Stop(ctx), context.DeadlineExceeded)
|
||||
}
|
||||
|
||||
func TestRunSupervisorAliveTracksRun(t *testing.T) {
|
||||
var s RunSupervisor
|
||||
|
||||
assert.False(t, s.Alive(), "no run has started")
|
||||
|
||||
_, done := s.Begin()
|
||||
assert.True(t, s.Alive(), "a run is in flight")
|
||||
|
||||
close(done)
|
||||
assert.False(t, s.Alive(), "the run signalled its exit")
|
||||
}
|
||||
|
||||
func TestRunSupervisorStopWithoutRunIsNoop(t *testing.T) {
|
||||
var s RunSupervisor
|
||||
|
||||
require.NoError(t, s.Stop(context.Background()))
|
||||
assert.Nil(t, s.Current())
|
||||
}
|
||||
Reference in New Issue
Block a user