Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 8 additions & 1 deletion supervisor/supervisor.go
Original file line number Diff line number Diff line change
Expand Up @@ -124,11 +124,17 @@ func (s *Supervisor) Run(
s.log.Logger().Error().Err(err).Msg("initial tunnel connection failed")
return err
}

return s.runLoop(ctx)
}

func (s *Supervisor) runLoop(ctx context.Context) error {
var tunnelsWaiting []int
tunnelsActive := s.config.HAConnections

backoff := retry.NewBackoff(s.config.Retries, tunnelRetryDuration, true)
var backoffTimer <-chan time.Time
gracefulShutdownC := s.gracefulShutdownC

shuttingDown := false
for {
Expand Down Expand Up @@ -175,8 +181,9 @@ func (s *Supervisor) Run(
// No more tunnels outstanding, clear backoff timer
backoff.SetGracePeriod()
}
case <-s.gracefulShutdownC:
case <-gracefulShutdownC:
shuttingDown = true
gracefulShutdownC = nil
}
}
}
Expand Down
63 changes: 63 additions & 0 deletions supervisor/tunnel_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ import (
"context"
"errors"
"fmt"
"sync/atomic"
"testing"
"time"

Expand Down Expand Up @@ -298,6 +299,68 @@ func newTestSupervisor(t *testing.T, server TunnelServer) *Supervisor {
}
}

type countingDoneContext struct {
context.Context
doneCalls atomic.Int32
doneCallC chan struct{}
}

func (c *countingDoneContext) Done() <-chan struct{} {
c.doneCalls.Add(1)
select {
case c.doneCallC <- struct{}{}:
default:
}
return nil
}

func TestRunLoopHandlesGracefulShutdownOnce(t *testing.T) {
gracefulShutdownC := make(chan struct{})
s := newTestSupervisor(t, nil)
s.config.HAConnections = 1
s.gracefulShutdownC = gracefulShutdownC

ctx := &countingDoneContext{
Context: context.Background(),
doneCallC: make(chan struct{}, 3),
}
runDone := make(chan error, 1)
go func() {
runDone <- s.runLoop(ctx)
}()

select {
case <-ctx.doneCallC:
case <-time.After(time.Second):
t.Fatal("run loop did not start")
}

close(gracefulShutdownC)
select {
case <-ctx.doneCallC:
case <-time.After(time.Second):
t.Fatal("run loop did not handle graceful shutdown")
}

busyLoop := false
select {
case <-ctx.doneCallC:
busyLoop = true
case <-time.After(100 * time.Millisecond):
}

s.tunnelErrors <- tunnelError{index: 0}
select {
case err := <-runDone:
require.NoError(t, err)
case <-time.After(time.Second):
t.Fatal("run loop did not exit after the active tunnel stopped")
}

assert.False(t, busyLoop, "run loop repeatedly selected the closed graceful shutdown channel")
assert.Equal(t, int32(2), ctx.doneCalls.Load())
}

func TestStartFirstTunnelRetryLoop(t *testing.T) {
t.Parallel()

Expand Down