From ab736e4c6b34616811ef3344c80bc488ea0b6303 Mon Sep 17 00:00:00 2001 From: Nate Brown Date: Fri, 10 Jul 2026 10:35:17 -0500 Subject: [PATCH] Make Control safe to stop and wait on from any lifecycle state (#1794) --- cmd/nebula-service/main.go | 12 +- cmd/nebula-service/service.go | 31 +++- cmd/nebula/main.go | 5 +- control.go | 72 ++++++--- control_lifecycle_test.go | 292 ++++++++++++++++++++++++++++++++++ interface.go | 43 +++-- main.go | 11 ++ overlay/tun_android.go | 1 + overlay/tun_ios.go | 8 + service/service.go | 24 ++- 10 files changed, 441 insertions(+), 58 deletions(-) create mode 100644 control_lifecycle_test.go diff --git a/cmd/nebula-service/main.go b/cmd/nebula-service/main.go index 724c0c6a..e0b335f5 100644 --- a/cmd/nebula-service/main.go +++ b/cmd/nebula-service/main.go @@ -53,7 +53,12 @@ func main() { l := logging.NewLogger(os.Stdout) if *serviceFlag != "" { - if err := doService(configPath, configTest, Build, serviceFlag); err != nil { + if *configTest { + fmt.Println("-test is not supported with -service, run the config test without -service") + os.Exit(1) + } + + if err := doService(configPath, Build, serviceFlag); err != nil { l.Error("Service command failed", "error", err) os.Exit(1) } @@ -93,15 +98,14 @@ func main() { } if !*configTest { - wait, err := ctrl.Start() - if err != nil { + if err := ctrl.Start(); err != nil { util.LogWithContextIfNeeded("Error while running", err, l) os.Exit(1) } go ctrl.ShutdownBlock() - if err := wait(); err != nil { + if err := ctrl.Wait(); err != nil { l.Error("Nebula stopped due to fatal error", "error", err) os.Exit(2) } diff --git a/cmd/nebula-service/service.go b/cmd/nebula-service/service.go index 7c2b39c8..abe9abe0 100644 --- a/cmd/nebula-service/service.go +++ b/cmd/nebula-service/service.go @@ -3,6 +3,7 @@ package main import ( "fmt" "log" + "os" "github.com/kardianos/service" "github.com/slackhq/nebula" @@ -14,7 +15,6 @@ var logger service.Logger type program struct { configPath *string - configTest *bool build string control *nebula.Control } @@ -40,22 +40,41 @@ func (p *program) Start(s service.Service) error { } }) - p.control, err = nebula.Main(c, *p.configTest, Build, l, nil) + p.control, err = nebula.Main(c, false, Build, l, nil) if err != nil { return err } - p.control.Start() + if err := p.control.Start(); err != nil { + return err + } + + // Nebula can stop itself on a fatal packet reader error, make sure to log it if it happens. + go func() { + if err := p.control.Wait(); err != nil { + logger.Error(fmt.Sprintf("Nebula stopped due to fatal error: %v", err)) + os.Exit(2) + } + }() + return nil } func (p *program) Stop(s service.Service) error { logger.Info("Nebula service stopping.") + if p.control == nil { + return nil + } + p.control.Stop() + + // block until nebula has fully drained before reporting stopped. + // error logging is handled by Start. + _ = p.control.Wait() return nil } -func doService(configPath *string, configTest *bool, build string, serviceFlag *string) error { +func doService(configPath *string, build string, serviceFlag *string) error { if *configPath == "" { p, err := config.DefaultPath() if err != nil { @@ -73,7 +92,6 @@ func doService(configPath *string, configTest *bool, build string, serviceFlag * prg := &program{ configPath: configPath, - configTest: configTest, build: build, } @@ -105,8 +123,9 @@ func doService(configPath *string, configTest *bool, build string, serviceFlag * switch *serviceFlag { case "run": if err := s.Run(); err != nil { - // Route any errors to the system logger + // Route any errors to the system logger and report the failure logger.Error(err) + return err } default: if err := service.Control(s, *serviceFlag); err != nil { diff --git a/cmd/nebula/main.go b/cmd/nebula/main.go index 219519c2..3c786b84 100644 --- a/cmd/nebula/main.go +++ b/cmd/nebula/main.go @@ -84,8 +84,7 @@ func main() { } if !*configTest { - wait, err := ctrl.Start() - if err != nil { + if err := ctrl.Start(); err != nil { util.LogWithContextIfNeeded("Error while running", err, l) os.Exit(1) } @@ -93,7 +92,7 @@ func main() { go ctrl.ShutdownBlock() notifyReady(l) - if err := wait(); err != nil { + if err := ctrl.Wait(); err != nil { l.Error("Nebula stopped due to fatal error", "error", err) os.Exit(2) } diff --git a/control.go b/control.go index 053feab5..a79ebbfa 100644 --- a/control.go +++ b/control.go @@ -69,29 +69,29 @@ type ControlHostInfo struct { } // Start actually runs nebula, this is a nonblocking call. -// The returned function blocks until nebula has fully stopped and returns the -// first fatal reader error (if any). A nil error means nebula shut down -// gracefully; a non-nil error means a reader hit an unexpected failure that -// triggered the shutdown. -func (c *Control) Start() (func() error, error) { +// Use Wait to block until nebula has fully stopped and to learn whether a fatal reader error caused the shutdown. +func (c *Control) Start() error { c.stateLock.Lock() defer c.stateLock.Unlock() switch c.state { case StateReady: //yay! case StateStopped, StateStopping: - return nil, ErrAlreadyStopped + return ErrAlreadyStopped case StateStarted: - return nil, ErrAlreadyStarted + return ErrAlreadyStarted default: - return nil, ErrUnknownState + return ErrUnknownState } // Activate the interface err := c.f.activate() if err != nil { + // Cancel before Close so a caller returning from Wait always observes a dead Context + c.cancel() + _ = c.f.Close() c.state = StateStopped - return nil, err + return err } // Call all the delayed funcs that waited patiently for the interface to be created. @@ -114,13 +114,9 @@ func (c *Control) Start() (func() error, error) { c.f.triggerShutdown = c.Stop // Start reading packets. - out, err := c.f.run() - if err != nil { - c.state = StateStopped - return nil, err - } + c.f.run() c.state = StateStarted - return out, nil + return nil } func (c *Control) State() RunState { @@ -133,10 +129,26 @@ func (c *Control) Context() context.Context { return c.ctx } -// Stop is a non-blocking call that signals nebula to close all tunnels and shut down +// Stop tears nebula down, closing all tunnels and releasing everything it holds. +// Use Wait to block until the shutdown has completed. +// A Control that has been stopped cannot be started again, Start will return ErrAlreadyStopped. func (c *Control) Stop() { c.stateLock.Lock() - if c.state != StateStarted { + switch c.state { + case StateStarted: + // Fall through to the full teardown below + + case StateReady: + // Never started + c.cancel() + c.state = StateStopped + if err := c.f.Close(); err != nil { + c.l.Error("Close interface failed", "error", err) + } + c.stateLock.Unlock() + return + + default: c.stateLock.Unlock() // We are stopping or stopped already return @@ -145,19 +157,26 @@ func (c *Control) Stop() { c.state = StateStopping c.stateLock.Unlock() - // Stop the handshakeManager (and other services), to prevent new tunnels from - // being created while we're shutting them all down. + // Closing tunnels can be slow with a large hostmap, don't hold the lock for it c.cancel() - c.CloseAllTunnels(false) + + c.stateLock.Lock() + c.state = StateStopped if err := c.f.Close(); err != nil { c.l.Error("Close interface failed", "error", err) } - c.stateLock.Lock() - c.state = StateStopped c.stateLock.Unlock() } +// Wait blocks until nebula has fully stopped, either via Stop or an internal fatal error, +// and returns the first fatal packet reader error if there was one. +// It is safe to call from multiple goroutines and at any point in the lifecycle, +// but a Wait on a Control that is never started and never stopped will block forever. +func (c *Control) Wait() error { + return c.f.wait() +} + // ShutdownBlock will listen for and block on term and interrupt signals, calling Control.Stop() once signalled func (c *Control) ShutdownBlock() { sigChan := make(chan os.Signal, 1) @@ -170,8 +189,15 @@ func (c *Control) ShutdownBlock() { c.Stop() } -// RebindUDPServer asks the UDP listener to rebind it's listener. Mainly used on mobile clients when interfaces change +// RebindUDPServer asks the UDP listener to rebind it's listener. Mainly used on mobile clients when interfaces change. func (c *Control) RebindUDPServer() { + c.stateLock.Lock() + defer c.stateLock.Unlock() + + if c.state != StateStarted { + return + } + _ = c.f.outside.Rebind() // Trigger a lighthouse update, useful for mobile clients that should have an update interval of 0 diff --git a/control_lifecycle_test.go b/control_lifecycle_test.go new file mode 100644 index 00000000..0b5d106d --- /dev/null +++ b/control_lifecycle_test.go @@ -0,0 +1,292 @@ +package nebula + +import ( + "context" + "errors" + "io" + "net/netip" + "sync" + "testing" + "time" + + "github.com/gaissmai/bart" + "github.com/slackhq/nebula/config" + "github.com/slackhq/nebula/routing" + "github.com/slackhq/nebula/test" + "github.com/slackhq/nebula/udp" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type fakeDevice struct { + closeOnce sync.Once + closedCh chan struct{} + closed bool +} + +func newFakeDevice() *fakeDevice { + return &fakeDevice{closedCh: make(chan struct{})} +} + +// Read blocks until Close like a real tun with no traffic, then reports EOF +// the same way a closed device does +func (d *fakeDevice) Read(p []byte) (int, error) { + <-d.closedCh + return 0, io.EOF +} + +func (d *fakeDevice) Write(p []byte) (int, error) { return len(p), nil } + +func (d *fakeDevice) Close() error { + d.closeOnce.Do(func() { + d.closed = true + close(d.closedCh) + }) + return nil +} + +func (d *fakeDevice) Activate() error { return nil } +func (d *fakeDevice) Networks() []netip.Prefix { return nil } +func (d *fakeDevice) Name() string { return "fake" } +func (d *fakeDevice) RoutesFor(netip.Addr) routing.Gateways { return nil } +func (d *fakeDevice) SupportsMultiqueue() bool { return false } +func (d *fakeDevice) NewMultiQueueReader() (io.ReadWriteCloser, error) { + return nil, errors.New("unsupported") +} + +// newReadyControl hand-builds the minimum Control that Main would have +// produced right before Start, including the construction token NewInterface +// takes so waiters block until Close releases the resources +func newReadyControl(t *testing.T) (*Control, *fakeDevice, *fakeConn) { + l := test.NewLogger() + dev := newFakeDevice() + conn := &fakeConn{} + ctx, cancel := context.WithCancel(context.Background()) + + myVpnNet := netip.MustParsePrefix("10.128.0.1/16") + nt := new(bart.Lite) + nt.Insert(myVpnNet) + cs := &CertState{ + myVpnNetworks: []netip.Prefix{myVpnNet}, + myVpnNetworksTable: nt, + } + lh, err := NewLightHouseFromConfig(ctx, l, config.NewC(l), cs, nil, nil) + require.NoError(t, err) + + f := &Interface{ + ctx: ctx, + inside: dev, + outside: conn, + writers: []udp.Conn{conn}, + readers: make([]io.ReadWriteCloser, 1), + routines: 1, + hostMap: newHostMap(l), + lightHouse: lh, + l: l, + } + f.wg.Add(1) + + return &Control{ + state: StateReady, + f: f, + l: l, + ctx: ctx, + cancel: cancel, + }, dev, conn +} + +func TestControl_StopBeforeStart(t *testing.T) { + c, dev, conn := newReadyControl(t) + + // A Stop on a never started control must release everything Main acquired + c.Stop() + assert.Equal(t, StateStopped, c.State()) + assert.True(t, dev.closed, "the tun device should have been closed") + assert.True(t, conn.closed, "the udp socket should have been closed") + require.ErrorIs(t, c.ctx.Err(), context.Canceled, "the service context should have been cancelled") + + // Wait must return promptly now that the resources are released + require.NoError(t, c.Wait()) + + // A stopped control can never be started + require.ErrorIs(t, c.Start(), ErrAlreadyStopped) + + // A second Stop is a harmless no-op + c.Stop() + assert.Equal(t, StateStopped, c.State()) + require.NoError(t, c.Wait()) +} + +func TestControl_WaitBlocksUntilStop(t *testing.T) { + c, _, _ := newReadyControl(t) + + done := make(chan error, 1) + go func() { done <- c.Wait() }() + + select { + case <-done: + t.Fatal("Wait returned before Stop") + case <-time.After(50 * time.Millisecond): + } + + c.Stop() + select { + case err := <-done: + require.NoError(t, err) + case <-time.After(time.Second): + t.Fatal("Wait did not return after Stop") + } +} + +type fakeConn struct { + closed bool + rebinds int +} + +func (c *fakeConn) Rebind() error { c.rebinds++; return nil } +func (c *fakeConn) LocalAddr() (netip.AddrPort, error) { return netip.AddrPort{}, nil } +func (c *fakeConn) ListenOut(_ udp.EncReader) error { return nil } +func (c *fakeConn) WriteTo(_ []byte, _ netip.AddrPort) error { return nil } +func (c *fakeConn) ReloadConfig(_ *config.C) {} +func (c *fakeConn) SupportsMultipleReaders() bool { return true } +func (c *fakeConn) Close() error { c.closed = true; return nil } + +type multiqueueDevice struct { + *fakeDevice +} + +func (d *multiqueueDevice) SupportsMultiqueue() bool { return true } + +func TestControl_StartMultiqueueFailureReleases(t *testing.T) { + dev := &multiqueueDevice{fakeDevice: newFakeDevice()} + conn := &fakeConn{} + ctx, cancel := context.WithCancel(context.Background()) + f := &Interface{ + ctx: ctx, + inside: dev, + outside: conn, + writers: []udp.Conn{conn}, + readers: make([]io.ReadWriteCloser, 2), + routines: 2, + l: test.NewLogger(), + } + f.wg.Add(1) + + c := &Control{ + state: StateReady, + f: f, + l: test.NewLogger(), + ctx: ctx, + cancel: cancel, + } + + // The second reader fails to open, everything must be released + require.Error(t, c.Start()) + assert.Equal(t, StateStopped, c.State()) + assert.True(t, dev.closed, "the tun device should have been closed") + assert.True(t, conn.closed, "the udp socket should have been closed") + require.ErrorIs(t, c.ctx.Err(), context.Canceled) + + // And Wait must not hang on the construction token + require.NoError(t, c.Wait()) +} + +func TestInterface_CloseIsIdempotent(t *testing.T) { + dev := newFakeDevice() + f := &Interface{ + inside: dev, + l: test.NewLogger(), + } + f.wg.Add(1) + + require.NoError(t, f.Close()) + assert.True(t, dev.closed) + + // A second Close must not double release the wg token or the device + require.NoError(t, f.Close()) + require.NoError(t, f.wait()) +} + +func TestControl_FatalErrorReportsThroughWait(t *testing.T) { + c, dev, conn := newReadyControl(t) + + // Mirror what Start wires up, without needing real packet readers + c.f.triggerShutdown = c.Stop + c.state = StateStarted + + boom := errors.New("boom") + c.f.onFatal(boom) + + require.ErrorIs(t, c.Wait(), boom) + assert.Equal(t, StateStopped, c.State()) + assert.True(t, dev.closed) + assert.True(t, conn.closed) + + // A second fatal error must not fire the shutdown again or replace the first + c.f.onFatal(errors.New("later")) + require.ErrorIs(t, c.Wait(), boom) + + // Wait stays factual, a Stop after the death does not mask the error + c.Stop() + require.ErrorIs(t, c.Wait(), boom) +} + +func TestControl_ConcurrentStopAndStart(t *testing.T) { + c, _, _ := newReadyControl(t) + + var wg sync.WaitGroup + for i := 0; i < 2; i++ { + wg.Go(func() { c.Stop() }) + } + wg.Go(func() { _ = c.Start() }) + wg.Go(func() { + _ = c.Wait() + // A returned Wait must always observe the final state, no matter how + // the race resolved + assert.Equal(t, StateStopped, c.State()) + }) + wg.Wait() + + // However the race resolves, the control must end fully stopped with no + // panic and Wait must observe the final state + require.NoError(t, c.Wait()) + assert.Equal(t, StateStopped, c.State()) + require.ErrorIs(t, c.Start(), ErrAlreadyStopped) +} + +func TestControl_StartStopLifecycle(t *testing.T) { + c, dev, conn := newReadyControl(t) + + require.NoError(t, c.Start()) + assert.Equal(t, StateStarted, c.State()) + require.ErrorIs(t, c.Start(), ErrAlreadyStarted) + + // Stop must unpark the reader blocked in the device and release everything + c.Stop() + assert.Equal(t, StateStopped, c.State()) + assert.True(t, dev.closed, "the tun device should have been closed") + assert.True(t, conn.closed, "the udp socket should have been closed") + require.ErrorIs(t, c.ctx.Err(), context.Canceled) + + // The reader drained off a closed device, that is not a fatal error + require.NoError(t, c.Wait()) + require.ErrorIs(t, c.Start(), ErrAlreadyStopped) +} + +func TestControl_RebindIsGatedByState(t *testing.T) { + c, _, conn := newReadyControl(t) + + // A rebind before Start reaches nothing, the interface is not up + c.RebindUDPServer() + assert.Equal(t, 0, conn.rebinds, "rebind before start must be a no-op") + + require.NoError(t, c.Start()) + c.RebindUDPServer() + assert.Equal(t, 1, conn.rebinds, "rebind while started must reach the conn") + + // A rebind racing a completed stop must not touch the closed conn + c.Stop() + require.NoError(t, c.Wait()) + c.RebindUDPServer() + assert.Equal(t, 1, conn.rebinds, "rebind after stop must be a no-op") +} diff --git a/interface.go b/interface.go index f96e431a..c44f38b3 100644 --- a/interface.go +++ b/interface.go @@ -215,6 +215,9 @@ func NewInterface(ctx context.Context, c *InterfaceConfig) (*Interface, error) { ifce.connectionManager.intf = ifce + // Held until Close so waiting on the interface blocks until the resources are actually released + ifce.wg.Add(1) + return ifce, nil } @@ -258,17 +261,16 @@ func (f *Interface) activate() error { f.readers[i] = reader } - f.wg.Add(1) // for us to wait on Close() to return + // On error the caller owns the cleanup, Control.Start cancels the service context + // before releasing our resources so a waiter never observes a live context if err = f.inside.Activate(); err != nil { - f.wg.Done() - f.inside.Close() return err } return nil } -func (f *Interface) run() (func() error, error) { +func (f *Interface) run() { // Launch n queues to read packets from udp for i := 0; i < f.routines; i++ { f.wg.Go(func() { @@ -283,13 +285,14 @@ func (f *Interface) run() (func() error, error) { }) } - return func() error { - f.wg.Wait() - if e := f.fatalErr.Load(); e != nil { - return *e - } - return nil - }, nil +} + +func (f *Interface) wait() error { + f.wg.Wait() + if e := f.fatalErr.Load(); e != nil { + return *e + } + return nil } // onFatal stores the first fatal reader error, and calls triggerShutdown if it was the first one @@ -322,7 +325,10 @@ func (f *Interface) listenOut(i int) { f.readOutsidePackets(ViaSender{UdpAddr: fromUdpAddr}, plaintext[:0], payload, h, fwPacket, lhh, nb, i, ctCache.Get()) }) - if err != nil && !f.closed.Load() { + // An error after teardown began is shutdown noise, the closed flag covers resources + // Close releases itself and the cancelled ctx covers ones torn down by their owners + // reacting to it, like the user device pipes + if err != nil && !f.closed.Load() && f.ctx.Err() == nil { f.l.Error("Error while reading inbound packet, closing", "error", err) f.onFatal(err) } @@ -341,7 +347,8 @@ func (f *Interface) listenIn(reader io.ReadWriteCloser, i int) { for { n, err := reader.Read(packet) if err != nil { - if !f.closed.Load() { + // Same shutdown noise handling as listenOut + if !f.closed.Load() && f.ctx.Err() == nil { f.l.Error("Error while reading outbound packet, closing", "error", err, "reader", i) f.onFatal(err) } @@ -542,9 +549,15 @@ func (f *Interface) GetCertState() *CertState { return f.pki.getCertState() } +// Close releases the interface's resources: the udp sockets and the tun device. +// It is idempotent and safe to call at any point in the lifecycle, including on an interface that never activated, +// calls after the first return nil without doing anything. func (f *Interface) Close() error { + if !f.closed.CompareAndSwap(false, true) { + return nil + } + var errs []error - f.closed.Store(true) // Release the udp readers for i, u := range f.writers { @@ -560,6 +573,8 @@ func (f *Interface) Close() error { if closeErr != nil { errs = append(errs, closeErr) } + + // Release the construction token so waiters know the resources are gone f.wg.Done() return errors.Join(errs...) } diff --git a/main.go b/main.go index 7d7a0f72..d62d8dd0 100644 --- a/main.go +++ b/main.go @@ -130,6 +130,17 @@ func Main(c *config.C, configTest bool, buildVersion string, l *slog.Logger, dev udpConns := make([]udp.Conn, routines) port := c.GetInt("listen.port", 0) + // Callers get no handle to these until the Control is returned, release them on any error. + defer func() { + if reterr != nil { + for _, u := range udpConns { + if u != nil { + _ = u.Close() + } + } + } + }() + if !configTest { rawListenHost := c.GetString("listen.host", "0.0.0.0") var listenHost netip.Addr diff --git a/overlay/tun_android.go b/overlay/tun_android.go index 9cbb64be..e4080b41 100644 --- a/overlay/tun_android.go +++ b/overlay/tun_android.go @@ -40,6 +40,7 @@ func newTunFromFd(c *config.C, l *slog.Logger, deviceFd int, vpnNetworks []netip err := t.reload(c, true) if err != nil { + _ = file.Close() return nil, err } diff --git a/overlay/tun_ios.go b/overlay/tun_ios.go index 6bfcbdfb..27bf558b 100644 --- a/overlay/tun_ios.go +++ b/overlay/tun_ios.go @@ -18,6 +18,7 @@ import ( "github.com/slackhq/nebula/config" "github.com/slackhq/nebula/routing" "github.com/slackhq/nebula/util" + "golang.org/x/sys/unix" ) type tun struct { @@ -33,6 +34,12 @@ func newTun(_ *config.C, _ *slog.Logger, _ []netip.Prefix, _ bool) (*tun, error) } func newTunFromFd(c *config.C, l *slog.Logger, deviceFd int, vpnNetworks []netip.Prefix) (*tun, error) { + if err := unix.SetNonblock(deviceFd, true); err != nil { + // We own the fd from the moment it is handed to us, same as the reload error path below + _ = unix.Close(deviceFd) + return nil, fmt.Errorf("failed to set the tun fd to non-blocking mode: %w", err) + } + file := os.NewFile(uintptr(deviceFd), "/dev/tun") t := &tun{ vpnNetworks: vpnNetworks, @@ -42,6 +49,7 @@ func newTunFromFd(c *config.C, l *slog.Logger, deviceFd int, vpnNetworks []netip err := t.reload(c, true) if err != nil { + _ = file.Close() return nil, err } diff --git a/service/service.go b/service/service.go index 899e851d..6610800d 100644 --- a/service/service.go +++ b/service/service.go @@ -43,12 +43,25 @@ type Service struct { } } -func New(control *nebula.Control) (*Service, error) { - wait, err := control.Start() +func New(control *nebula.Control) (_ *Service, reterr error) { + // Check this before Start so a failure doesn't leave a running nebula + device, ok := control.Device().(*overlay.UserDevice) + if !ok { + return nil, errors.New("must be using user device") + } + + err := control.Start() if err != nil { return nil, err } + // Anything that fails after a successful Start must tear nebula back down + defer func() { + if reterr != nil { + control.Stop() + } + }() + ctx := control.Context() eg, ctx := errgroup.WithContext(ctx) s := Service{ @@ -57,11 +70,6 @@ func New(control *nebula.Control) (*Service, error) { } s.mu.listeners = map[uint16]*tcpListener{} - device, ok := control.Device().(*overlay.UserDevice) - if !ok { - return nil, errors.New("must be using user device") - } - s.ipstack = stack.New(stack.Options{ NetworkProtocols: []stack.NetworkProtocolFactory{ipv4.NewProtocol, ipv6.NewProtocol}, TransportProtocols: []stack.TransportProtocolFactory{tcp.NewProtocol, udp.NewProtocol, icmp.NewProtocol4, icmp.NewProtocol6}, @@ -147,7 +155,7 @@ func New(control *nebula.Control) (*Service, error) { // Add the nebula wait function to the group so a fatal reader error // propagates out through errgroup.Wait(). eg.Go(func() error { - return wait() + return control.Wait() }) return &s, nil