fix(sdn): keep initial watch replay when sendInitialEvents is set

The resourceVersion de-dup filter dropped ADDED events whose RV was <=
the requested startingRV. During a sendInitialEvents replay the backing
WatchList intentionally re-emits existing objects as initial ADDED
events with RV <= startingRV, so the filter broke the initial-state
replay. Skip the de-dup while sendInitialEvents is set; it still applies
when resuming a plain watch from a resourceVersion. Add a regression
test covering sendInitialEvents with a non-zero starting RV.

Assisted-By: Claude <noreply@anthropic.com>
Signed-off-by: Aleksei Sviridkin <f@lex.la>
This commit is contained in:
Aleksei Sviridkin committed 2026-06-22 17:07:47 +03:00
1 parent 237a8b24e0
commit 2db614127f
2 files changed
+61 -1

No files matched your search

+6 -1
View File
@@ -662,7 +662,12 @@ func (r *REST) Watch(ctx context.Context, opts *metainternal.ListOptions) (watch
sg := policyToSecurityGroup(np)
if ev.Type == watch.Added && startingRV > 0 {
// De-dup ADDED events the client has already seen when resuming a plain
// watch from a resourceVersion. This must NOT run during a
// sendInitialEvents replay: there the backing API intentionally re-emits
// existing objects as initial ADDED events with RV <= startingRV, and
// dropping them would break the initial-state replay.
if !sendInitialEvents && ev.Type == watch.Added && startingRV > 0 {
objRV, parseErr := strconv.ParseUint(sg.ResourceVersion, 10, 64)
if parseErr == nil && objRV <= startingRV {
continue
@@ -5,6 +5,7 @@ package securitygroup
import (
"context"
"strconv"
"testing"
"time"
@@ -202,3 +203,57 @@ func TestWatchClusterWide(t *testing.T) {
t.Fatalf("cluster-wide watch did not stream the cross-namespace event: %+v", evs)
}
}
// TestWatchSendInitialEventsReplaysLowResourceVersionAdds asserts that with
// sendInitialEvents=true the resourceVersion de-dup filter does not drop
// initial ADDED events whose resourceVersion is <= the requested startingRV.
// The backing WatchList intentionally replays existing objects (RV <=
// startingRV) as the initial state; dropping them would break the replay.
func TestWatchSendInitialEventsReplaysLowResourceVersionAdds(t *testing.T) {
r := newTestREST(t)
ctx, cancel := context.WithCancel(ctxNS())
defer cancel()
sendInitialEvents := true
// A startingRV far above any RV the fake store assigns, so the ADDED event
// below has objRV <= startingRV and exercises the de-dup filter.
const startingRV = "1000000000"
w, err := r.Watch(ctx, &metainternal.ListOptions{
SendInitialEvents: &sendInitialEvents,
ResourceVersion: startingRV,
})
if err != nil {
t.Fatalf("Watch returned error: %v", err)
}
defer w.Stop()
np := markedPolicy("sg-db")
if err := r.c.Create(ctx, np); err != nil {
t.Fatalf("create marked: %v", err)
}
// Precondition: the object RV must be <= startingRV, otherwise the filter
// would not apply and the test would not exercise the regression.
objRV, perr := strconv.ParseUint(np.ResourceVersion, 10, 64)
if perr != nil {
t.Fatalf("parse object RV %q: %v", np.ResourceVersion, perr)
}
if objRV > 1000000000 {
t.Fatalf("test precondition broken: object RV %d > startingRV", objRV)
}
// The ADDED event for sg-db must survive the initial replay, not be dropped.
evs := collectEvents(t, w, 3, 2*time.Second)
found := false
for _, ev := range evs {
if ev.Type != watch.Added {
continue
}
if sg, ok := ev.Object.(*sdnv1alpha1.SecurityGroup); ok && sg.Name == "sg-db" {
found = true
break
}
}
if !found {
t.Fatalf("expected an Added event for sg-db during initial replay, got %+v", evs)
}
}