From 216eb6242ef0d6ef4416cd0662d2c316e76ff16a Mon Sep 17 00:00:00 2001 From: Peter Sprygada Date: Wed, 7 Oct 2026 15:07:36 -0400 Subject: [PATCH] fix: Stop the gateway rewriting unchanged rules on every pod attach Every gateway node reconciled on any BGPAdvertisement, BGPVRFInstance or BGPRouter event in its namespace, status writes and its own VIP advertisements included. Each pass re-applied every rule: a fresh quota reservation, a rebuilt Maglev table, and a Put of every vip_table and vip_addr_table entry. Pod churn anywhere in the cluster therefore rewrote every VIP on every gateway node. The engine now applies a rule only when it is new or differs from the active one, compared field by field including each backend's uSID. An unchanged rule still reports an applied status, so its advertisement and Programmed condition are untouched. Skipping unchanged rules removes the every-pass rewrite that used to repair a failed update or removal, so the engine now tracks those itself. A failed apply releases the key's quota and can leave its datapath entries half-written, and a failed removal leaves the rule active with its quota already released. Either marks the key dirty, and a dirty key is applied on the next pass even when its desired rule equals the active one. A new rule that fails to apply never becomes active, so it is retried as before. The NetworkGateway controller's watches now drop events that cannot change what a pass does: - BGPAdvertisements without both a VRFID and a Function, which the backend resolver ignores, except a delete or spec edit of the gateway's own VIP advertisements, so a removed one is still restored. - Status-only updates to all three kinds. Every field the resolver reads is in spec. Unchanged rules are no longer rewritten after an external wipe of the pinned maps. A process restart, which empties the engine's active set, still rebuilds them. Fixes #719 Co-Authored-By: Claude Opus 5.5 --- .../controller/networkgateway_controller.go | 57 +++++++ .../networkgateway_predicate_test.go | 88 +++++++++++ internal/gateway/diff.go | 30 +++- internal/gateway/diff_test.go | 62 +++++++- internal/gateway/engine.go | 53 +++++-- internal/gateway/engine_test.go | 147 ++++++++++++++++++ internal/gateway/quota.go | 4 +- 7 files changed, 417 insertions(+), 24 deletions(-) create mode 100644 internal/controller/networkgateway_predicate_test.go diff --git a/internal/controller/networkgateway_controller.go b/internal/controller/networkgateway_controller.go index f99b9be3..8627d161 100644 --- a/internal/controller/networkgateway_controller.go +++ b/internal/controller/networkgateway_controller.go @@ -22,10 +22,13 @@ import ( "k8s.io/apimachinery/pkg/util/validation" "k8s.io/client-go/util/retry" ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/builder" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" + "sigs.k8s.io/controller-runtime/pkg/event" "sigs.k8s.io/controller-runtime/pkg/handler" "sigs.k8s.io/controller-runtime/pkg/log" + "sigs.k8s.io/controller-runtime/pkg/predicate" ctrlreconcile "sigs.k8s.io/controller-runtime/pkg/reconcile" "go.datum.net/galactic/internal/gateway" @@ -962,25 +965,79 @@ func (r *NetworkGatewayReconciler) SetupWithManager(mgr ctrl.Manager) error { // reconcile and the rule would stay broken until an unrelated event. // Each broadcasts to every NetworkGateway in the namespace, since any // of them could be the one waiting on this object. + // + // Backend resolution reads only these objects' specs, so status-only + // updates are filtered out, and advertisements that can never locate a + // backend (the gateway's own VIP advertisements among them) are + // filtered out entirely. Every pod attach still writes one that passes; + // the engine then skips every rule whose resolved state is unchanged. Watches(&bgpv1alpha1.BGPRouter{}, handler.EnqueueRequestsFromMapFunc( func(ctx context.Context, obj client.Object) []ctrlreconcile.Request { return broadcastToGatewayRequests(ctx, r.Client, obj.GetNamespace(), "BGPRouter", obj.GetName()) }), + builder.WithPredicates(predicate.GenerationChangedPredicate{}), ). Watches(&bgpv1alpha1.BGPAdvertisement{}, handler.EnqueueRequestsFromMapFunc( func(ctx context.Context, obj client.Object) []ctrlreconcile.Request { return broadcastToGatewayRequests(ctx, r.Client, obj.GetNamespace(), "BGPAdvertisement", obj.GetName()) }), + builder.WithPredicates(backendAdvertisementPredicate()), ). Watches(&bgpv1alpha1.BGPVRFInstance{}, handler.EnqueueRequestsFromMapFunc( func(ctx context.Context, obj client.Object) []ctrlreconcile.Request { return broadcastToGatewayRequests(ctx, r.Client, obj.GetNamespace(), "BGPVRFInstance", obj.GetName()) }), + builder.WithPredicates(predicate.GenerationChangedPredicate{}), ). Named("networkgateway"). Complete(r) } +// backendAdvertisementPredicate passes only BGPAdvertisement events a pass can +// act on. +// +// buildBackendSIDIndex ignores an advertisement without both VRFID and +// Function, so a create or delete of one is dropped. An update passes when its +// spec changed and either side carries both, so an advertisement gaining or +// losing them still triggers a pass. +// +// The gateway's own VIP advertisements carry networkRuleLabel and neither +// field. Their creates come from this reconciler and are dropped, but a delete +// or a spec edit passes, so the next pass restores what was lost. +func backendAdvertisementPredicate() predicate.Predicate { + return predicate.Funcs{ + CreateFunc: func(e event.CreateEvent) bool { return locatesBackend(e.Object) }, + DeleteFunc: func(e event.DeleteEvent) bool { + return locatesBackend(e.Object) || isRuleAdvertisement(e.Object) + }, + UpdateFunc: func(e event.UpdateEvent) bool { + if e.ObjectOld == nil || e.ObjectNew == nil { + return false + } + if e.ObjectOld.GetGeneration() == e.ObjectNew.GetGeneration() { + return false + } + return locatesBackend(e.ObjectOld) || locatesBackend(e.ObjectNew) || + isRuleAdvertisement(e.ObjectOld) || isRuleAdvertisement(e.ObjectNew) + }, + GenericFunc: func(e event.GenericEvent) bool { return locatesBackend(e.Object) }, + } +} + +// locatesBackend reports whether obj is a BGPAdvertisement buildBackendSIDIndex +// would consider: one carrying both a VRFID and a Function. +func locatesBackend(obj client.Object) bool { + adv, ok := obj.(*bgpv1alpha1.BGPAdvertisement) + return ok && adv.Spec.VRFID != nil && adv.Spec.Function != nil +} + +// isRuleAdvertisement reports whether obj is a gateway VIP advertisement made +// for a NetworkRule. +func isRuleAdvertisement(obj client.Object) bool { + _, ok := obj.GetLabels()[networkRuleLabel] + return ok +} + // ruleToGatewayRequests maps a NetworkRule change to every NetworkGateway in // its namespace. A NetworkRule carries no gatewayRef, and every gateway node // advertises every VIP identically (anycast), so every gateway node in the diff --git a/internal/controller/networkgateway_predicate_test.go b/internal/controller/networkgateway_predicate_test.go new file mode 100644 index 00000000..9e130399 --- /dev/null +++ b/internal/controller/networkgateway_predicate_test.go @@ -0,0 +1,88 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +package controller + +import ( + "testing" + + "sigs.k8s.io/controller-runtime/pkg/event" + + bgpv1alpha1 "go.datum.net/network/api/v1alpha1" +) + +func testAdvertisement(generation int64, srv6 bool) *bgpv1alpha1.BGPAdvertisement { + adv := &bgpv1alpha1.BGPAdvertisement{} + adv.Name = "adv" + adv.Namespace = "ns" + adv.Generation = generation + if srv6 { + adv.Spec.VRFID = ptr(int32(7)) + adv.Spec.Function = ptr(bgpv1alpha1.SRv6FunctionEndDT6) + } + return adv +} + +func TestBackendAdvertisementPredicate(t *testing.T) { + p := backendAdvertisementPredicate() + + t.Run("create or delete without VRFID and Function is dropped", func(t *testing.T) { + adv := testAdvertisement(1, false) + if p.Create(event.CreateEvent{Object: adv}) { + t.Error("Create passed an advertisement that cannot locate a backend") + } + if p.Delete(event.DeleteEvent{Object: adv}) { + t.Error("Delete passed an advertisement that cannot locate a backend") + } + }) + + t.Run("create or delete with VRFID and Function passes", func(t *testing.T) { + adv := testAdvertisement(1, true) + if !p.Create(event.CreateEvent{Object: adv}) { + t.Error("Create dropped a backend advertisement") + } + if !p.Delete(event.DeleteEvent{Object: adv}) { + t.Error("Delete dropped a backend advertisement") + } + }) + + t.Run("gateway VIP advertisement heals on delete or edit but not create", func(t *testing.T) { + vipAdv := func(generation int64) *bgpv1alpha1.BGPAdvertisement { + adv := testAdvertisement(generation, false) + adv.Labels = map[string]string{networkRuleLabel: "rule"} + return adv + } + if p.Create(event.CreateEvent{Object: vipAdv(1)}) { + t.Error("Create passed the gateway's own VIP advertisement") + } + if !p.Delete(event.DeleteEvent{Object: vipAdv(1)}) { + t.Error("Delete dropped a gateway VIP advertisement") + } + if !p.Update(event.UpdateEvent{ObjectOld: vipAdv(1), ObjectNew: vipAdv(2)}) { + t.Error("Update dropped a spec edit to a gateway VIP advertisement") + } + if p.Update(event.UpdateEvent{ObjectOld: vipAdv(1), ObjectNew: vipAdv(1)}) { + t.Error("Update passed a status-only change to a gateway VIP advertisement") + } + }) + + tests := []struct { + name string + old, new *bgpv1alpha1.BGPAdvertisement + want bool + }{ + {"status-only update", testAdvertisement(1, true), testAdvertisement(1, true), false}, + {"spec change on a backend advertisement", testAdvertisement(1, true), testAdvertisement(2, true), true}, + {"spec change on a VIP advertisement", testAdvertisement(1, false), testAdvertisement(2, false), false}, + {"VRFID and Function removed", testAdvertisement(1, true), testAdvertisement(2, false), true}, + {"VRFID and Function added", testAdvertisement(1, false), testAdvertisement(2, true), true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := p.Update(event.UpdateEvent{ObjectOld: tt.old, ObjectNew: tt.new}); got != tt.want { + t.Errorf("Update = %v, want %v", got, tt.want) + } + }) + } +} diff --git a/internal/gateway/diff.go b/internal/gateway/diff.go index bfe9164a..915b0591 100644 --- a/internal/gateway/diff.go +++ b/internal/gateway/diff.go @@ -4,19 +4,26 @@ package gateway -import "sort" +import ( + "slices" + "sort" +) // diffRuleKeys compares the currently active rule keys against the desired state // and returns which need applying and which removing. // -// A key present in both is always returned in toApply: the engine applies -// unconditionally rather than diffing individual field changes. The two slices -// partition the key space, not changed against unchanged. +// A key is in toApply when it is not active yet or its desired rule differs from +// the active one. An unchanged rule is left out, so an event that changes nothing +// a rule depends on, such as a pod attach elsewhere in the namespace, writes +// nothing to the datapath. // // Both are sorted, for deterministic log output and test assertions, the inputs // being maps whose iteration order is randomized. func diffRuleKeys(active map[string]DesiredRule, desired map[string]DesiredRule) (toApply, toRemove []string) { - for key := range desired { + for key, rule := range desired { + if cur, ok := active[key]; ok && rulesEqual(cur, rule) { + continue + } toApply = append(toApply, key) } for key := range active { @@ -28,3 +35,16 @@ func diffRuleKeys(active map[string]DesiredRule, desired map[string]DesiredRule) sort.Strings(toRemove) return toApply, toRemove } + +// rulesEqual reports whether a and b would program identical datapath state. +// VIPAddresses and Backends compare in order: the controller builds both in +// spec order, so a reordered spec reapplies once with the same result. +func rulesEqual(a, b DesiredRule) bool { + return a.Key == b.Key && + a.VPCRef == b.VPCRef && + a.VPCAttachmentRef == b.VPCAttachmentRef && + a.Protocol == b.Protocol && + a.Port == b.Port && + slices.Equal(a.VIPAddresses, b.VIPAddresses) && + slices.Equal(a.Backends, b.Backends) +} diff --git a/internal/gateway/diff_test.go b/internal/gateway/diff_test.go index 1a3beb6b..ddb0aace 100644 --- a/internal/gateway/diff_test.go +++ b/internal/gateway/diff_test.go @@ -5,29 +5,33 @@ package gateway import ( + "net/netip" "reflect" "testing" ) const ( - testKeyA = "ns/a" - testKeyB = "ns/b" - testKeyKeep = "ns/keep" + testKeyA = "ns/a" + testKeyB = "ns/b" + testKeyKeep = "ns/keep" + testKeyChanged = "ns/changed" ) func TestDiffRuleKeys_ApplyAndRemove(t *testing.T) { active := map[string]DesiredRule{ - testKeyKeep: {Key: testKeyKeep}, - testKeyA: {Key: testKeyA}, + testKeyKeep: {Key: testKeyKeep, Port: 80}, + testKeyChanged: {Key: testKeyChanged, Port: 80}, + testKeyA: {Key: testKeyA}, } desired := map[string]DesiredRule{ - testKeyKeep: {Key: testKeyKeep}, - testKeyB: {Key: testKeyB}, + testKeyKeep: {Key: testKeyKeep, Port: 80}, + testKeyChanged: {Key: testKeyChanged, Port: 443}, + testKeyB: {Key: testKeyB}, } toApply, toRemove := diffRuleKeys(active, desired) - if want := []string{testKeyB, testKeyKeep}; !reflect.DeepEqual(toApply, want) { + if want := []string{testKeyB, testKeyChanged}; !reflect.DeepEqual(toApply, want) { t.Errorf("toApply = %v, want %v", toApply, want) } if want := []string{testKeyA}; !reflect.DeepEqual(toRemove, want) { @@ -63,3 +67,45 @@ func TestDiffRuleKeys_AllRemovedNoDesired(t *testing.T) { t.Errorf("toRemove = %v, want %v", toRemove, want) } } + +func TestRulesEqual(t *testing.T) { + vip := netip.MustParseAddr("2001:db8::1") + usidA := netip.MustParseAddr("fc00:0:1::") + usidB := netip.MustParseAddr("fc00:0:2::") + backend := func(usid netip.Addr) DesiredBackend { + return DesiredBackend{Address: netip.MustParseAddr("2001:db8:1::10"), Port: 8080, USID: usid} + } + base := DesiredRule{ + Key: testKeyA, VPCRef: "vpc", VPCAttachmentRef: "att", + VIPAddresses: []netip.Addr{vip}, Protocol: "udp", Port: 80, + Backends: []DesiredBackend{backend(usidA)}, + } + clone := func(mut func(*DesiredRule)) DesiredRule { + r := base + r.VIPAddresses = append([]netip.Addr(nil), base.VIPAddresses...) + r.Backends = append([]DesiredBackend(nil), base.Backends...) + mut(&r) + return r + } + + tests := []struct { + name string + b DesiredRule + want bool + }{ + {"identical copy", clone(func(*DesiredRule) {}), true}, + {"backend uSID moved", clone(func(r *DesiredRule) { r.Backends[0] = backend(usidB) }), false}, + {"backend added", clone(func(r *DesiredRule) { r.Backends = append(r.Backends, backend(usidB)) }), false}, + {"VIP removed", clone(func(r *DesiredRule) { r.VIPAddresses = nil }), false}, + {"port changed", clone(func(r *DesiredRule) { r.Port = 443 }), false}, + {"protocol changed", clone(func(r *DesiredRule) { r.Protocol = "" }), false}, + {"VPC changed", clone(func(r *DesiredRule) { r.VPCRef = "other" }), false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := rulesEqual(base, tt.b); got != tt.want { + t.Errorf("rulesEqual = %v, want %v", got, tt.want) + } + }) + } +} diff --git a/internal/gateway/engine.go b/internal/gateway/engine.go index 793e4cd0..94fb09f3 100644 --- a/internal/gateway/engine.go +++ b/internal/gateway/engine.go @@ -8,6 +8,7 @@ import ( "context" "errors" "fmt" + "slices" "sync" ) @@ -19,6 +20,11 @@ type Engine struct { mu sync.Mutex active map[string]DesiredRule + // dirty holds active keys whose last apply or removal failed. Their + // datapath and quota state no longer match their active rule, so the + // next Reconcile applies them even when the desired rule equals it. + dirty map[string]struct{} + datapath Datapath quota QuotaEnforcer telemetry TelemetryEmitter @@ -30,44 +36,71 @@ type Engine struct { func NewEngine(datapath Datapath, quota QuotaEnforcer, telemetry TelemetryEmitter) *Engine { return &Engine{ active: make(map[string]DesiredRule), + dirty: make(map[string]struct{}), datapath: datapath, quota: quota, telemetry: telemetry, } } -// Reconcile converges live state toward desired: every rule in desired is -// re-applied, and every rule still active but no longer in desired is torn -// down. The caller must already have withdrawn a rule's BGP route before it -// disappears from desired. +// Reconcile converges live state toward desired: every rule in desired that is +// new or differs from its active version is applied, and every rule still +// active but no longer in desired is torn down. The caller must already have +// withdrawn a rule's BGP route before it disappears from desired. // -// It applies everything in desired and removes everything not in it, rather -// than diffing field by field, so a partial previous failure, such as a crash -// between two rules, self-heals on the next call instead of requiring the -// caller to track what succeeded. +// A rule equal to its active version is skipped: no quota reservation, no +// datapath write, no telemetry. It still gets an applied status, so the result +// carries one status per desired rule. A partial previous failure still +// self-heals on the next call: a new rule only becomes active once applied, and +// an active rule whose update or removal failed is marked dirty and applied +// again whatever its desired state. func (e *Engine) Reconcile(ctx context.Context, desired EngineState) (EngineStatus, error) { e.mu.Lock() defer e.mu.Unlock() toApply, toRemove := diffRuleKeys(e.active, desired.Rules) + for key := range e.dirty { + if _, ok := desired.Rules[key]; ok && !slices.Contains(toApply, key) { + toApply = append(toApply, key) + } + } + slices.Sort(toApply) - var statuses []RuleStatus + applying := make(map[string]struct{}, len(toApply)) + for _, key := range toApply { + applying[key] = struct{}{} + } + statuses := make([]RuleStatus, 0, len(desired.Rules)+len(toRemove)) + for key := range desired.Rules { + if _, ok := applying[key]; !ok { + statuses = append(statuses, RuleStatus{Key: key, Applied: true}) + } + } for _, key := range toApply { rule := desired.Rules[key] if err := e.applyRuleLocked(ctx, rule); err != nil { + // A failed apply releases the key's quota and may leave its + // datapath entries half-written, so an active rule no longer + // matches what is programmed. + if _, ok := e.active[key]; ok { + e.dirty[key] = struct{}{} + } statuses = append(statuses, RuleStatus{Key: key, Applied: false, Error: err.Error()}) continue } e.active[key] = rule + delete(e.dirty, key) statuses = append(statuses, RuleStatus{Key: key, Applied: true}) } for _, key := range toRemove { if err := e.removeRuleLocked(ctx, key); err != nil { + e.dirty[key] = struct{}{} statuses = append(statuses, RuleStatus{Key: key, Applied: true, Error: err.Error()}) continue } delete(e.active, key) + delete(e.dirty, key) } healthy := true @@ -107,9 +140,11 @@ func (e *Engine) Stop(ctx context.Context) error { if firstErr == nil { firstErr = fmt.Errorf("stop: remove rule %s: %w", key, err) } + e.dirty[key] = struct{}{} continue } delete(e.active, key) + delete(e.dirty, key) } return firstErr } diff --git a/internal/gateway/engine_test.go b/internal/gateway/engine_test.go index e0f5f0ce..d6114f20 100644 --- a/internal/gateway/engine_test.go +++ b/internal/gateway/engine_test.go @@ -7,6 +7,8 @@ package gateway import ( "context" "errors" + "net/netip" + "slices" "strings" "testing" ) @@ -15,6 +17,7 @@ import ( // convergence logic without a kernel or root privileges. type fakeDatapath struct { applied map[string]DesiredRule + applyCalls []string removed []string applyErr error removeErr error @@ -27,6 +30,7 @@ func newFakeDatapath() *fakeDatapath { } func (f *fakeDatapath) ApplyRule(_ context.Context, rule DesiredRule) error { + f.applyCalls = append(f.applyCalls, rule.Key) if f.applyErr != nil { return f.applyErr } @@ -56,6 +60,7 @@ var _ Datapath = (*fakeDatapath)(nil) // fakeQuota is a controllable QuotaEnforcer. type fakeQuota struct { + reserved int deny bool checkErr error releaseErr error @@ -63,6 +68,7 @@ type fakeQuota struct { } func (f *fakeQuota) CheckAndReserve(context.Context, DesiredRule) (bool, error) { + f.reserved++ if f.checkErr != nil { return false, f.checkErr } @@ -151,6 +157,147 @@ func TestEngine_ReconcileIsIdempotentAcrossRepeatedCalls(t *testing.T) { } } +func TestEngine_ReconcileSkipsUnchangedRule(t *testing.T) { + dp := newFakeDatapath() + quota := &fakeQuota{} + tel := &fakeTelemetry{} + e := NewEngine(dp, quota, tel) + ctx := context.Background() + desired := EngineState{Rules: map[string]DesiredRule{testKeyA: {Key: testKeyA, Port: 80}}} + + for i := range 3 { + status, err := e.Reconcile(ctx, desired) + if err != nil { + t.Fatalf("Reconcile %d: %v", i, err) + } + if !status.Healthy || len(status.Rules) != 1 || !status.Rules[0].Applied || status.Rules[0].Key != testKeyA { + t.Errorf("Reconcile %d status = %+v, want one healthy applied entry for %s", i, status, testKeyA) + } + } + if len(dp.applyCalls) != 1 { + t.Errorf("ApplyRule calls = %v, want exactly one for an unchanged rule", dp.applyCalls) + } + if quota.reserved != 1 { + t.Errorf("quota reservations = %d, want 1", quota.reserved) + } + if len(tel.applied) != 1 { + t.Errorf("RuleApplied events = %v, want 1", tel.applied) + } +} + +func TestEngine_ReconcileReappliesOnlyChangedRule(t *testing.T) { + dp := newFakeDatapath() + e := NewEngine(dp, &fakeQuota{}, &fakeTelemetry{}) + ctx := context.Background() + backend := func(usid string) []DesiredBackend { + return []DesiredBackend{{ + Address: netip.MustParseAddr("2001:db8:1::10"), Port: 8080, USID: netip.MustParseAddr(usid), + }} + } + ruleA := DesiredRule{Key: testKeyA, Port: 80, Backends: backend("fc00:0:1::")} + ruleB := DesiredRule{Key: testKeyB, Port: 443, Backends: backend("fc00:0:1::")} + + first := EngineState{Rules: map[string]DesiredRule{testKeyA: ruleA, testKeyB: ruleB}} + if _, err := e.Reconcile(ctx, first); err != nil { + t.Fatalf("first Reconcile: %v", err) + } + dp.applyCalls = nil + + moved := ruleA + moved.Backends = backend("fc00:0:2::") + second := EngineState{Rules: map[string]DesiredRule{testKeyA: moved, testKeyB: ruleB}} + if _, err := e.Reconcile(ctx, second); err != nil { + t.Fatalf("second Reconcile: %v", err) + } + if want := []string{testKeyA}; !slices.Equal(dp.applyCalls, want) { + t.Errorf("ApplyRule calls = %v, want %v", dp.applyCalls, want) + } + if got := dp.applied[testKeyA].Backends[0].USID; got != netip.MustParseAddr("fc00:0:2::") { + t.Errorf("applied backend uSID = %s, want fc00:0:2::", got) + } +} + +func TestEngine_ReconcileRetriesFailedRule(t *testing.T) { + dp := newFakeDatapath() + e := NewEngine(dp, &fakeQuota{}, &fakeTelemetry{}) + ctx := context.Background() + desired := EngineState{Rules: map[string]DesiredRule{testKeyA: {Key: testKeyA}}} + + dp.applyErr = errors.New("simulated apply failure") + if _, err := e.Reconcile(ctx, desired); err != nil { + t.Fatalf("first Reconcile: %v", err) + } + dp.applyErr = nil + if _, err := e.Reconcile(ctx, desired); err != nil { + t.Fatalf("second Reconcile: %v", err) + } + if len(dp.applyCalls) != 2 { + t.Errorf("ApplyRule calls = %v, want a retry after the failed apply", dp.applyCalls) + } + if _, ok := dp.applied[testKeyA]; !ok { + t.Error("rule not applied after the retry") + } +} + +func TestEngine_ReconcileRetriesActiveRuleAfterFailedUpdate(t *testing.T) { + dp := newFakeDatapath() + e := NewEngine(dp, &fakeQuota{}, &fakeTelemetry{}) + ctx := context.Background() + rule := DesiredRule{Key: testKeyA, Port: 80} + changed := DesiredRule{Key: testKeyA, Port: 443} + + if _, err := e.Reconcile(ctx, EngineState{Rules: map[string]DesiredRule{testKeyA: rule}}); err != nil { + t.Fatalf("first Reconcile: %v", err) + } + dp.applyErr = errors.New("simulated apply failure") + if _, err := e.Reconcile(ctx, EngineState{Rules: map[string]DesiredRule{testKeyA: changed}}); err != nil { + t.Fatalf("failed update Reconcile: %v", err) + } + dp.applyErr = nil + dp.applyCalls = nil + + // Reverting to the still-active rule must reapply it: the failed update + // released its quota and may have left the datapath half-written. + if _, err := e.Reconcile(ctx, EngineState{Rules: map[string]DesiredRule{testKeyA: rule}}); err != nil { + t.Fatalf("revert Reconcile: %v", err) + } + if want := []string{testKeyA}; !slices.Equal(dp.applyCalls, want) { + t.Errorf("ApplyRule calls after revert = %v, want %v", dp.applyCalls, want) + } + + dp.applyCalls = nil + if _, err := e.Reconcile(ctx, EngineState{Rules: map[string]DesiredRule{testKeyA: rule}}); err != nil { + t.Fatalf("steady Reconcile: %v", err) + } + if len(dp.applyCalls) != 0 { + t.Errorf("ApplyRule calls once clean = %v, want none", dp.applyCalls) + } +} + +func TestEngine_ReconcileReappliesRuleWhoseRemovalFailed(t *testing.T) { + dp := newFakeDatapath() + e := NewEngine(dp, &fakeQuota{}, &fakeTelemetry{}) + ctx := context.Background() + desired := EngineState{Rules: map[string]DesiredRule{testKeyA: {Key: testKeyA, Port: 80}}} + + if _, err := e.Reconcile(ctx, desired); err != nil { + t.Fatalf("first Reconcile: %v", err) + } + dp.removeErr = errors.New("simulated teardown failure") + if _, err := e.Reconcile(ctx, EngineState{Rules: map[string]DesiredRule{}}); err != nil { + t.Fatalf("failed removal Reconcile: %v", err) + } + dp.removeErr = nil + dp.applyCalls = nil + + if _, err := e.Reconcile(ctx, desired); err != nil { + t.Fatalf("return Reconcile: %v", err) + } + if want := []string{testKeyA}; !slices.Equal(dp.applyCalls, want) { + t.Errorf("ApplyRule calls = %v, want %v after a failed removal", dp.applyCalls, want) + } +} + func TestEngine_ReconcileQuotaDeniedSkipsApply(t *testing.T) { dp := newFakeDatapath() telem := &fakeTelemetry{} diff --git a/internal/gateway/quota.go b/internal/gateway/quota.go index 0d35cc45..ac318029 100644 --- a/internal/gateway/quota.go +++ b/internal/gateway/quota.go @@ -73,8 +73,8 @@ func NewNodeQuotaEnforcer(maxRulesPerTenant, maxRuleTableEntries int) *NodeQuota // CheckAndReserve reports whether rule fits within both limits and, if so, // reserves its map footprint. Idempotent for a key already reserved: // re-checking, or changing, an active rule's VIP count never double-counts it. -// That is required because the engine calls this for every desired rule on -// every pass, not only for new ones. +// That is required because the engine calls this again whenever an active +// rule changes or is retried after a failure, not only for new rules. func (e *NodeQuotaEnforcer) CheckAndReserve(_ context.Context, rule DesiredRule) (bool, error) { e.mu.Lock() defer e.mu.Unlock()