Skip to content

Commit b8a2659

Browse files
committed
Remove arbitrary sleeps from shutdown handling
Replace king's 1s sleep with event-driven QUIC drain wait that returns as soon as all lings disconnect (with 5s timeout as safety net). Remove ling's 750ms sleep entirely since it serves no purpose. Closes #16
1 parent 88df622 commit b8a2659

5 files changed

Lines changed: 166 additions & 7 deletions

File tree

internal/king/export_test.go

Lines changed: 35 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,10 @@
11
package king
22

3-
import "github.com/quic-go/quic-go"
3+
import (
4+
"context"
5+
6+
"github.com/quic-go/quic-go"
7+
)
48

59
// SetQUICConn sets a QUIC connection for a service ID (test helper).
610
func (tunnelSrv *TunnelServer) SetQUICConn(serviceID string, conn *quic.Conn) {
@@ -35,3 +39,33 @@ func (tunnelSrv *TunnelServer) SetServiceAuth(serviceID string, auth ServiceAuth
3539

3640
tunnelSrv.services[serviceID] = auth
3741
}
42+
43+
// RemoveQUICConn removes a QUIC connection and signals drainNotify (test helper).
44+
func (tunnelSrv *TunnelServer) RemoveQUICConn(serviceID string) {
45+
tunnelSrv.mu.Lock()
46+
47+
delete(tunnelSrv.quicConns, serviceID)
48+
49+
empty := len(tunnelSrv.quicConns) == 0
50+
tunnelSrv.mu.Unlock()
51+
52+
if empty {
53+
select {
54+
case tunnelSrv.drainNotify <- struct{}{}:
55+
default:
56+
}
57+
}
58+
}
59+
60+
// QUICConnCount returns the number of QUIC connections (test helper).
61+
func (tunnelSrv *TunnelServer) QUICConnCount() int {
62+
tunnelSrv.mu.RLock()
63+
defer tunnelSrv.mu.RUnlock()
64+
65+
return len(tunnelSrv.quicConns)
66+
}
67+
68+
// WaitForDrain exposes the private waitForDrain method (test helper).
69+
func (tunnelSrv *TunnelServer) WaitForDrain(ctx context.Context) {
70+
tunnelSrv.waitForDrain(ctx)
71+
}

internal/king/king.go

Lines changed: 21 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -24,9 +24,9 @@ import (
2424
const (
2525
defaultCouncilHost = "http://localhost:8080"
2626
kvPairCount = 2
27-
finalSyncTimeout = 2 * time.Second
28-
shutdownGracePause = 1 * time.Second
29-
certValidityYears = 10 * 365 * 24 * time.Hour
27+
finalSyncTimeout = 2 * time.Second
28+
quicDrainTimeout = 5 * time.Second
29+
certValidityYears = 10 * 365 * 24 * time.Hour
3030
)
3131

3232
var (
@@ -289,7 +289,24 @@ func performShutdown(
289289

290290
syncerInstance.sync(finalCtx)
291291

292-
time.Sleep(shutdownGracePause)
292+
drainCtx, drainCancel := context.WithTimeout(
293+
context.WithoutCancel(ctx), quicDrainTimeout,
294+
)
295+
defer drainCancel()
296+
297+
var drainWg sync.WaitGroup
298+
299+
for _, tunnelSrv := range tunnels {
300+
drainWg.Add(1)
301+
302+
go func(srv *TunnelServer) {
303+
defer drainWg.Done()
304+
305+
srv.waitForDrain(drainCtx)
306+
}(tunnelSrv)
307+
}
308+
309+
drainWg.Wait()
293310

294311
for _, tunnelSrv := range tunnels {
295312
tunnelSrv.close()

internal/king/king_test.go

Lines changed: 80 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,10 @@
11
package king_test
22

33
import (
4+
"context"
45
"sync"
56
"testing"
7+
"time"
68

79
"github.com/firecow/burrow/internal/king"
810
"github.com/firecow/burrow/internal/state"
@@ -894,6 +896,84 @@ func TestOnStateChanged_ClearsServicesOnEmpty(t *testing.T) {
894896
}
895897
}
896898

899+
// --- WaitForDrain ---
900+
901+
func TestWaitForDrain_EmptyImmediate(t *testing.T) {
902+
t.Parallel()
903+
904+
tunnelSrv := king.NewTunnelServer(testBindPort, nil)
905+
906+
done := make(chan struct{})
907+
908+
go func() {
909+
tunnelSrv.WaitForDrain(t.Context())
910+
close(done)
911+
}()
912+
913+
select {
914+
case <-done:
915+
case <-time.After(time.Second):
916+
t.Fatal("waitForDrain should return immediately when no connections")
917+
}
918+
}
919+
920+
func TestWaitForDrain_WaitsForRemoval(t *testing.T) {
921+
t.Parallel()
922+
923+
tunnelSrv := king.NewTunnelServer(testBindPort, nil)
924+
tunnelSrv.SetQUICConn("svc-1", nil)
925+
926+
done := make(chan struct{})
927+
928+
go func() {
929+
tunnelSrv.WaitForDrain(t.Context())
930+
close(done)
931+
}()
932+
933+
select {
934+
case <-done:
935+
t.Fatal("waitForDrain should not return while connections exist")
936+
case <-time.After(50 * time.Millisecond):
937+
}
938+
939+
tunnelSrv.RemoveQUICConn("svc-1")
940+
941+
select {
942+
case <-done:
943+
case <-time.After(time.Second):
944+
t.Fatal("waitForDrain should return after connection removed")
945+
}
946+
}
947+
948+
func TestWaitForDrain_RespectsContextTimeout(t *testing.T) {
949+
t.Parallel()
950+
951+
tunnelSrv := king.NewTunnelServer(testBindPort, nil)
952+
tunnelSrv.SetQUICConn("svc-1", nil)
953+
954+
ctx, cancel := context.WithTimeout(
955+
t.Context(), 100*time.Millisecond,
956+
)
957+
defer cancel()
958+
959+
done := make(chan struct{})
960+
961+
go func() {
962+
tunnelSrv.WaitForDrain(ctx)
963+
close(done)
964+
}()
965+
966+
select {
967+
case <-done:
968+
case <-time.After(time.Second):
969+
t.Fatal("waitForDrain should return on context cancellation")
970+
}
971+
972+
if tunnelSrv.QUICConnCount() != 1 {
973+
t.Fatal("connection should still exist after timeout")
974+
}
975+
}
976+
897977
// --- Syncer.TriggerSync ---
898978

899979
func TestTriggerSync_SendsNotification(t *testing.T) {

internal/king/tunnel.go

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -55,6 +55,7 @@ type TunnelServer struct {
5555
services map[string]ServiceAuth
5656
tcpListeners map[int]net.Listener
5757
quicConns map[string]*quic.Conn
58+
drainNotify chan struct{}
5859
onLingConnected func()
5960
}
6061

@@ -69,6 +70,7 @@ func NewTunnelServer(
6970
services: make(map[string]ServiceAuth),
7071
tcpListeners: make(map[int]net.Listener),
7172
quicConns: make(map[string]*quic.Conn),
73+
drainNotify: make(chan struct{}, 1),
7274
onLingConnected: nil,
7375
}
7476
}
@@ -290,7 +292,35 @@ func (tunnelSrv *TunnelServer) unregisterConnections(
290292
}
291293
}
292294

295+
empty := len(tunnelSrv.quicConns) == 0
293296
tunnelSrv.mu.Unlock()
297+
298+
if empty {
299+
select {
300+
case tunnelSrv.drainNotify <- struct{}{}:
301+
default:
302+
}
303+
}
304+
}
305+
306+
func (tunnelSrv *TunnelServer) waitForDrain(
307+
ctx context.Context,
308+
) {
309+
for {
310+
tunnelSrv.mu.RLock()
311+
empty := len(tunnelSrv.quicConns) == 0
312+
tunnelSrv.mu.RUnlock()
313+
314+
if empty {
315+
return
316+
}
317+
318+
select {
319+
case <-tunnelSrv.drainNotify:
320+
case <-ctx.Done():
321+
return
322+
}
323+
}
294324
}
295325

296326
func (tunnelSrv *TunnelServer) ensureTCPListener(

internal/ling/ling.go

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,6 @@ import (
2121
const (
2222
keyValueParts = 2
2323
finalSyncTimeout = 2 * time.Second
24-
shutdownGracePeriod = 750 * time.Millisecond
2524
quicErrorCodeCloseClean = 0
2625
)
2726

@@ -373,7 +372,6 @@ func performShutdown(
373372
defer finalCancel()
374373

375374
syncerInstance.sync(finalCtx)
376-
time.Sleep(shutdownGracePeriod)
377375

378376
tunnelCli.closeAll()
379377

0 commit comments

Comments
 (0)