Bound gateway informer startup and honor forwarding path MTUs

Cancel obsolete unsynced generations, bound required-cache synchronization, preserve adjacent-hop MTU caps, and use the active generation in route summaries.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Copilot-Session: ab9b78ee-42b8-4841-8d74-cd78d002d48b

Author: phealy

cmd/unbounded-net-node/gateway_informer_lifecycle.go

@@ -33,23 +33,26 @@ type gatewayInformerGeneration struct {
 }
 
 type gatewayInformerLifecycle struct {
-	client dynamic.Interface
-	resync time.Duration
-
-	mu         sync.RWMutex
-	required   bool
-	starting   bool
-	retiring   bool
-	generation uint64
-	active     *gatewayInformerGeneration
-	onEvent    func()
-	lastErr    error
+	client      dynamic.Interface
+	resync      time.Duration
+	syncTimeout time.Duration
+
+	mu          sync.RWMutex
+	required    bool
+	starting    bool
+	retiring    bool
+	generation  uint64
+	active      *gatewayInformerGeneration
+	onEvent     func()
+	lastErr     error
+	startCancel context.CancelFunc
 }
 
 func newGatewayInformerLifecycle(client dynamic.Interface, resync time.Duration) *gatewayInformerLifecycle {
 	return &gatewayInformerLifecycle{
-		client: client,
-		resync: resync,
+		client:      client,
+		resync:      resync,
+		syncTimeout: 30 * time.Second,
 	}
 }
 
@@ -75,7 +78,11 @@ func (l *gatewayInformerLifecycle) setRequired(required bool) {
 	defer l.mu.Unlock()
 
 	l.required = required
+
 	l.retiring = !required && l.active != nil
+	if !required && l.startCancel != nil {
+		l.startCancel()
+	}
 }
 
 func (l *gatewayInformerLifecycle) readiness() (bool, string) {
@@ -155,6 +162,20 @@ func (l *gatewayInformerLifecycle) start(ctx context.Context) (gatewayInformerSn
 		return snapshot, false, nil
 	}
 
+	if !l.required {
+		l.mu.Unlock()
+
+		return gatewayInformerSnapshot{}, false, nil
+	}
+
+	if l.starting {
+		l.mu.Unlock()
+
+		return gatewayInformerSnapshot{}, false, fmt.Errorf("gateway informer caches are already starting")
+	}
+
+	runCtx, cancel := context.WithCancel(ctx)
+	l.startCancel = cancel
 	l.starting = true
 	l.lastErr = nil
 	l.generation++
@@ -168,12 +189,12 @@ func (l *gatewayInformerLifecycle) start(ctx context.Context) (gatewayInformerSn
 
 	cleanups, err := addGatewayInformerEventHandlers(pools, nodes, poolPeerings, l.requestUpdate)
 	if err != nil {
+		cancel()
 		l.recordStartFailure(err)
 
 		return gatewayInformerSnapshot{}, false, err
 	}
 
-	runCtx, cancel := context.WithCancel(ctx)
 	factory.Start(runCtx.Done())
 
 	snapshot := gatewayInformerSnapshot{
@@ -183,7 +204,11 @@ func (l *gatewayInformerLifecycle) start(ctx context.Context) (gatewayInformerSn
 		nodes:        nodes,
 		poolPeerings: poolPeerings,
 	}
-	if !cache.WaitForCacheSync(runCtx.Done(), pools.HasSynced, nodes.HasSynced, poolPeerings.HasSynced) {
+
+	syncCtx, stopSync := context.WithTimeout(runCtx, l.syncTimeout)
+	defer stopSync()
+
+	if !cache.WaitForCacheSync(syncCtx.Done(), pools.HasSynced, nodes.HasSynced, poolPeerings.HasSynced) {
 		cancel()
 		runGatewayInformerCleanups(cleanups)
 
@@ -204,7 +229,18 @@ func (l *gatewayInformerLifecycle) start(ctx context.Context) (gatewayInformerSn
 	}
 
 	l.mu.Lock()
+	if !l.required || runCtx.Err() != nil {
+		l.starting = false
+		l.startCancel = nil
+		l.mu.Unlock()
+		cancel()
+		runGatewayInformerCleanups(cleanups)
+
+		return gatewayInformerSnapshot{}, false, fmt.Errorf("gateway informer generation %d start canceled", generation)
+	}
+
 	l.starting = false
+	l.startCancel = nil
 	l.lastErr = nil
 	l.active = activeGeneration
 	l.retiring = !l.required
@@ -218,6 +254,7 @@ func (l *gatewayInformerLifecycle) recordStartFailure(err error) {
 	defer l.mu.Unlock()
 
 	l.starting = false
+	l.startCancel = nil
 	l.lastErr = err
 }
 
@@ -244,7 +281,11 @@ func (l *gatewayInformerLifecycle) close() {
 	active := l.active
 	l.active = nil
 	l.required = false
+
 	l.retiring = false
+	if l.startCancel != nil {
+		l.startCancel()
+	}
 	l.mu.Unlock()
 
 	if active != nil {

cmd/unbounded-net-node/gateway_informer_lifecycle_test.go

@@ -5,6 +5,7 @@ package main
 
 import (
 	"context"
+	"errors"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -15,10 +16,76 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	dynamicfake "k8s.io/client-go/dynamic/fake"
+	k8stesting "k8s.io/client-go/testing"
 
 	unboundednetv1alpha1 "github.com/Azure/unbounded/api/net/v1alpha1"
 )
 
+func TestGatewayInformerFailedListCanBeAbandoned(t *testing.T) {
+	for _, mode := range []string{"assignment removed", "sync timeout", "close"} {
+		t.Run(mode, func(t *testing.T) {
+			client := newGatewayLifecycleClient()
+			listed := make(chan struct{}, 1)
+
+			client.PrependReactor("list", "*", func(k8stesting.Action) (bool, runtime.Object, error) {
+				select {
+				case listed <- struct{}{}:
+				default:
+				}
+
+				return true, nil, errors.New("list unavailable")
+			})
+			lifecycle := newGatewayInformerLifecycle(client, 0)
+			t.Cleanup(lifecycle.close)
+
+			if mode == "sync timeout" {
+				lifecycle.syncTimeout = 50 * time.Millisecond
+			}
+
+			done := make(chan error, 1)
+
+			go func() {
+				_, _, err := lifecycle.prepare(t.Context(), true)
+				done <- err
+			}()
+
+			select {
+			case <-listed:
+			case <-time.After(2 * time.Second):
+				t.Fatal("informers did not attempt LIST")
+			}
+
+			switch mode {
+			case "assignment removed":
+				lifecycle.setRequired(false)
+			case "close":
+				lifecycle.close()
+			}
+
+			select {
+			case err := <-done:
+				if err == nil {
+					t.Fatal("failed LIST unexpectedly produced synced caches")
+				}
+			case <-time.After(2 * time.Second):
+				t.Fatal("cache sync blocked reconciliation")
+			}
+
+			if lifecycle.snapshot().available {
+				t.Fatal("unsynced generation was published")
+			}
+
+			if mode != "sync timeout" {
+				if _, retire, err := lifecycle.prepare(t.Context(), false); err != nil || retire {
+					t.Fatalf("failed to abandon caches: retire=%v err=%v", retire, err)
+				}
+			} else if ready, _ := lifecycle.readiness(); ready {
+				t.Fatal("timed-out required caches reported ready")
+			}
+		})
+	}
+}
+
 func TestGatewayInformerLifecycleLateEventHandler(t *testing.T) {
 	client := newGatewayLifecycleClient()
 	lifecycle := newGatewayInformerLifecycle(client, 0)

cmd/unbounded-net-node/reconciliation_helpers.go

@@ -6,6 +6,7 @@ package main
 import (
 	"encoding/json"
 	"fmt"
+	"slices"
 	"strconv"
 	"strings"
 	"time"
@@ -276,6 +277,10 @@ func gatewayPoolPeeringTunnelMTU(
 	remotePool = strings.TrimSpace(remotePool)
 
 	for _, peering := range poolPeerings {
+		if !unboundednetv1alpha1.SpecEnabled(peering.Spec.Enabled) {
+			continue
+		}
+
 		hasLocal := false
 		hasRemote := false
 
@@ -294,36 +299,51 @@ func gatewayPoolPeeringTunnelMTU(
 }
 
 func learnedGatewayTransitTunnelMTU(
-	peerPool string,
 	learnedRoutes map[string]unboundednetv1alpha1.GatewayNodeRoute,
+	siteMTUs, poolMTUs map[string]int,
+	assignments []unboundednetv1alpha1.SiteGatewayPoolAssignment,
 	poolPeerings []unboundednetv1alpha1.GatewayPoolPeering,
 ) int {
 	result := 0
-	peerPool = strings.TrimSpace(peerPool)
 
 	for _, route := range learnedRoutes {
 		for _, path := range route.Paths {
-			pathPools := map[string]bool{peerPool: true}
+			for index, hop := range path {
+				switch hop.Type {
+				case "Site":
+					result = resolveTunnelMTU(result, siteMTUs[hop.Name])
+				case "GatewayPool":
+					result = resolveTunnelMTU(result, poolMTUs[hop.Name])
+				}
 
-			for _, hop := range path {
-				if hop.Type == "GatewayPool" {
-					if poolName := strings.TrimSpace(hop.Name); poolName != "" {
-						pathPools[poolName] = true
-					}
+				if index == 0 {
+					continue
 				}
-			}
 
-			for _, peering := range poolPeerings {
-				matchedPools := 0
+				// Paths run from origin to advertiser. Only adjacent hops
+				// identify a forwarding edge; co-occurrence is not enough.
+				previous := path[index-1]
+				if previous.Type == "GatewayPool" && hop.Type == "GatewayPool" {
+					result = resolveTunnelMTU(result, gatewayPoolPeeringTunnelMTU([]string{previous.Name}, hop.Name, poolPeerings))
 
-				for _, poolName := range peering.Spec.GatewayPools {
-					if pathPools[strings.TrimSpace(poolName)] {
-						matchedPools++
-					}
+					continue
+				}
+
+				site, pool := previous, hop
+				if site.Type == "GatewayPool" && pool.Type == "Site" {
+					site, pool = pool, site
 				}
 
-				if matchedPools >= 2 {
-					result = resolveTunnelMTU(result, tunnelMTUFromSpec(peering.Spec.TunnelMTU))
+				if site.Type != "Site" || pool.Type != "GatewayPool" {
+					continue
+				}
+
+				for _, assignment := range assignments {
+					if unboundednetv1alpha1.SpecEnabled(assignment.Spec.Enabled) &&
+						slices.Contains(assignment.Spec.Sites, site.Name) &&
+						slices.Contains(assignment.Spec.GatewayPools, pool.Name) {
+						result = resolveTunnelMTU(result, tunnelMTUFromSpec(assignment.Spec.TunnelMTU))
+					}
 				}
 			}
 		}

cmd/unbounded-net-node/reconciliation_helpers_test.go

@@ -250,6 +250,49 @@ func TestMergeLowestTunnelMTU(t *testing.T) {
 	}
 }
 
+func TestLearnedGatewayTransitPathCaps(t *testing.T) {
+	path := []unboundednetv1alpha1.GatewayNodePathHop{
+		{Type: "Site", Name: "site-b"},
+		{Type: "GatewayPool", Name: "pool-h"},
+		{Type: "GatewayPool", Name: "pool-j"},
+		{Type: "GatewayPool", Name: "pool-g"},
+	}
+	routes := map[string]unboundednetv1alpha1.GatewayNodeRoute{
+		"10.0.0.0/24": {Paths: [][]unboundednetv1alpha1.GatewayNodePathHop{path}},
+	}
+
+	for _, tc := range []struct {
+		name        string
+		sites       map[string]int
+		pools       map[string]int
+		assignments []unboundednetv1alpha1.SiteGatewayPoolAssignment
+		peerings    []unboundednetv1alpha1.GatewayPoolPeering
+		want        int
+	}{
+		{name: "origin Site", sites: map[string]int{"site-b": 1200, "unrelated": 900}, want: 1200},
+		{name: "downstream pool", pools: map[string]int{"pool-h": 1250, "unrelated": 900}, want: 1250},
+		{name: "assignment edge", assignments: []unboundednetv1alpha1.SiteGatewayPoolAssignment{
+			{Spec: unboundednetv1alpha1.SiteGatewayPoolAssignmentSpec{Sites: []string{"site-b"}, GatewayPools: []string{"pool-h"}, TunnelMTU: ptrInt32(1100)}},
+		}, want: 1100},
+		{name: "assignment nonadjacent", assignments: []unboundednetv1alpha1.SiteGatewayPoolAssignment{
+			{Spec: unboundednetv1alpha1.SiteGatewayPoolAssignmentSpec{Sites: []string{"site-b"}, GatewayPools: []string{"pool-g"}, TunnelMTU: ptrInt32(1100)}},
+		}},
+		{name: "adjacent peering", peerings: []unboundednetv1alpha1.GatewayPoolPeering{
+			{Spec: unboundednetv1alpha1.GatewayPoolPeeringSpec{GatewayPools: []string{"pool-h", "pool-j"}, TunnelMTU: ptrInt32(1300)}},
+		}, want: 1300},
+		{name: "nonadjacent peering", peerings: []unboundednetv1alpha1.GatewayPoolPeering{
+			{Spec: unboundednetv1alpha1.GatewayPoolPeeringSpec{GatewayPools: []string{"pool-h", "pool-g"}, TunnelMTU: ptrInt32(1000)}},
+		}},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			got := learnedGatewayTransitTunnelMTU(routes, tc.sites, tc.pools, tc.assignments, tc.peerings)
+			if got != tc.want {
+				t.Fatalf("transit MTU=%d want=%d", got, tc.want)
+			}
+		})
+	}
+}
+
 func TestForwardingAwareTunnelMTU(t *testing.T) {
 	siteMTUs := map[string]int{
 		"site-a": 1450,
@@ -296,11 +339,11 @@ func TestForwardingAwareTunnelMTU(t *testing.T) {
 			}},
 		},
 	}
-	if got := learnedGatewayTransitTunnelMTU("pool-g", learnedRoutes, []unboundednetv1alpha1.GatewayPoolPeering{peering}); got != 1200 {
+	if got := learnedGatewayTransitTunnelMTU(learnedRoutes, nil, nil, nil, []unboundednetv1alpha1.GatewayPoolPeering{peering}); got != 1200 {
 		t.Fatalf("worker learned gateway transit MTU = %d, want 1200", got)
 	}
 
-	if got := learnedGatewayTransitTunnelMTU("pool-g", nil, []unboundednetv1alpha1.GatewayPoolPeering{peering}); got != 0 {
+	if got := learnedGatewayTransitTunnelMTU(nil, nil, nil, nil, []unboundednetv1alpha1.GatewayPoolPeering{peering}); got != 0 {
 		t.Fatalf("gateway peering without a learned forwarding path constrained worker MTU: %d", got)
 	}
 

cmd/unbounded-net-node/site_watch_reconcile.go

@@ -283,6 +283,7 @@ func watchSiteAndConfigureWireGuard(ctx context.Context, clientset kubernetes.In
 			gatewaySnapshot        gatewayInformerSnapshot
 			retireGatewayInformers bool
 			gatewayStateStable     bool
+			err                    error
 		)
 
 		for range 3 {
@@ -1871,9 +1872,9 @@ func updateWireGuardFromSlices(ctx context.Context, dynamicClient dynamic.Interf
 				excludedNodeCIDRSites := networkPeeredSites
 				routedCidrs, routeDistances, learnedRoutes := routedCIDRsForGatewayPeer(peerGatewayNode, fallbackRoutedCIDRs, mySiteName, localGatewayPools, excludedNodeCIDRSites, now, staleAfter)
 
-				transitTunnelMTU := learnedGatewayTransitTunnelMTU(pool.Name, learnedRoutes, poolPeerings)
+				transitTunnelMTU := learnedGatewayTransitTunnelMTU(learnedRoutes, siteTunnelMTUs, poolTunnelMTUs, assignments, poolPeerings)
 				if isGatewayNode {
-					transitTunnelMTU = gatewayPoolPeeringTunnelMTU(localGatewayPools, pool.Name, poolPeerings)
+					transitTunnelMTU = resolveTunnelMTU(transitTunnelMTU, gatewayPoolPeeringTunnelMTU(localGatewayPools, pool.Name, poolPeerings))
 				}
 
 				peeringTunnelProtocol := poolPeeringTunnelProtocols[pool.Name]