Harden auto-standby startup and reconnects

Author: sjmiller609

lib/autostandby/conntrack_events_linux.go

@@ -19,6 +19,7 @@ import (
 type conntrackStream struct {
 	fd        int
 	closeOnce sync.Once
+	done      chan struct{}
 	events    chan ConnectionEvent
 	errs      chan error
 }
@@ -52,6 +53,7 @@ func (s *ConntrackSource) OpenStream(ctx context.Context) (ConnectionStream, err
 
 	stream := &conntrackStream{
 		fd:     fd,
+		done:   make(chan struct{}),
 		events: make(chan ConnectionEvent, 256),
 		errs:   make(chan error, 16),
 	}
@@ -66,6 +68,7 @@ func (s *conntrackStream) Errors() <-chan error { return s.errs }
 func (s *conntrackStream) Close() error {
 	var err error
 	s.closeOnce.Do(func() {
+		close(s.done)
 		err = unix.Close(s.fd)
 	})
 	return err
@@ -76,8 +79,11 @@ func (s *conntrackStream) run(ctx context.Context) {
 	defer close(s.errs)
 
 	go func() {
-		<-ctx.Done()
-		_ = s.Close()
+		select {
+		case <-ctx.Done():
+			_ = s.Close()
+		case <-s.done:
+		}
 	}()
 
 	buf := make([]byte, 1<<20)

lib/autostandby/conntrack_events_linux_test.go

@@ -13,6 +13,28 @@ import (
 	"golang.org/x/sys/unix"
 )
 
+func TestConntrackStreamCloseSignalsDone(t *testing.T) {
+	t.Parallel()
+
+	fd, err := unix.Socket(unix.AF_UNIX, unix.SOCK_DGRAM, 0)
+	require.NoError(t, err)
+
+	stream := &conntrackStream{
+		fd:     fd,
+		done:   make(chan struct{}),
+		events: make(chan ConnectionEvent),
+		errs:   make(chan error),
+	}
+
+	require.NoError(t, stream.Close())
+
+	select {
+	case <-stream.done:
+	default:
+		t.Fatal("expected Close to signal stream shutdown")
+	}
+}
+
 func TestConnectionEventFromNetlinkMessageParsesIPv4TCPEvent(t *testing.T) {
 	t.Parallel()
 

lib/autostandby/controller.go

@@ -186,7 +186,8 @@ func (c *Controller) Run(ctx context.Context) error {
 	}
 
 	if err := c.startupResync(ctx); err != nil {
-		return err
+		c.recordControllerError("startup_resync")
+		c.log.Warn("auto-standby startup resync failed", "error", err)
 	}
 
 	instanceEvents, unsubscribe, err := c.store.SubscribeInstanceEvents()
@@ -399,6 +400,8 @@ func (c *Controller) startupResync(ctx context.Context) error {
 	}
 	conns, err := c.source.ListConnections(ctx)
 	if err != nil {
+		c.setObserverError(err)
+		c.recordObserverError("startup_resync")
 		c.recordStartupResync(start, "error")
 		recordSpanError(span, err)
 		return err

lib/autostandby/controller_test.go

@@ -17,6 +17,7 @@ type fakeInstanceStore struct {
 	persistedRuntime map[string]*Runtime
 	events           chan InstanceEvent
 	standbyErr       error
+	listErr          error
 }
 
 func newFakeInstanceStore(instances []Instance) *fakeInstanceStore {
@@ -28,6 +29,9 @@ func newFakeInstanceStore(instances []Instance) *fakeInstanceStore {
 }
 
 func (f *fakeInstanceStore) ListInstances(context.Context) ([]Instance, error) {
+	if f.listErr != nil {
+		return nil, f.listErr
+	}
 	out := make([]Instance, 0, len(f.instances))
 	for _, inst := range f.instances {
 		out = append(out, cloneInstance(inst))
@@ -56,13 +60,21 @@ func (f *fakeInstanceStore) SubscribeInstanceEvents() (<-chan InstanceEvent, fun
 
 type fakeConnectionSource struct {
 	connections []Connection
+	listErr     error
+	openErr     error
 }
 
 func (f *fakeConnectionSource) ListConnections(context.Context) ([]Connection, error) {
+	if f.listErr != nil {
+		return nil, f.listErr
+	}
 	return append([]Connection(nil), f.connections...), nil
 }
 
 func (f *fakeConnectionSource) OpenStream(context.Context) (ConnectionStream, error) {
+	if f.openErr != nil {
+		return nil, f.openErr
+	}
 	return &fakeConnectionStream{
 		events: make(chan ConnectionEvent),
 		errs:   make(chan error),
@@ -405,6 +417,44 @@ func TestStatusReportsObserverError(t *testing.T) {
 	require.Equal(t, ReasonObserverError, status.Reason)
 }
 
+func TestRunDegradesWhenStartupResyncFails(t *testing.T) {
+	t.Parallel()
+
+	store := newFakeInstanceStore([]Instance{{
+		ID:             "inst-err",
+		Name:           "inst-err",
+		State:          StateRunning,
+		NetworkEnabled: true,
+		IP:             "192.168.100.60",
+		AutoStandby:    &Policy{Enabled: true, IdleTimeout: "1m"},
+	}})
+	source := &fakeConnectionSource{
+		listErr: errors.New("conntrack permission denied"),
+	}
+	controller := NewController(store, source, ControllerOptions{})
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	done := make(chan error, 1)
+	go func() {
+		done <- controller.Run(ctx)
+	}()
+
+	select {
+	case err := <-done:
+		require.NoError(t, err, "controller should wait for cancellation instead of exiting on startup resync failure")
+	case <-time.After(50 * time.Millisecond):
+	}
+
+	cancel()
+	require.NoError(t, <-done)
+
+	status := controller.Describe(store.instances[0])
+	require.Equal(t, StatusError, status.Status)
+	require.Equal(t, ReasonObserverError, status.Reason)
+}
+
 func mustAddr(raw string) netip.Addr {
 	return netip.MustParseAddr(raw)
 }