subnet/etcd: recover single-subnet watch from compaction (#2524)

watchSubnet has the same hot-loop #2471 fixed in watchSubnets: it takes a start revision it never advances, so once etcd compacts past it every reconnect fails identically at the 5s backoff cap, forever.

Detect a compacted watch response, re-read the lease at a current revision, and resume from there. If the re-read finds the lease gone, synthesize the EventRemoved the missed delete would have produced, inferred from its absence since the delete itself sits below the compaction horizon. The revoke decision stays with the caller, where CompleteLease already makes it.

Author: phinze

pkg/subnet/etcd/registry.go

@@ -431,17 +431,39 @@ func (esr *etcdSubnetRegistry) watchSubnet(ctx context.Context, since int64, sn
 			case wresp, ok := <-rch:
 				err := wresp.Err()
 				if !ok || err != nil {
+					cancel()
+					// If the watch fell behind etcd's compaction horizon, reconnecting
+					// at the same revision fails identically forever (mvcc: required
+					// revision has been compacted). Re-read the lease at a current
+					// revision and resume from there instead of hot-looping.
+					if isCompacted(wresp) {
+						log.Warningf("etcd watch for %s fell behind compaction horizon (compact rev %d), re-reading and resuming", key, wresp.CompactRevision)
+						next, rerr := esr.resyncWatchSubnet(ctx, sn, sn6, leaseWatchChan)
+						if rerr != nil {
+							log.Errorf("failed to re-read subnet lease after compaction: %v", rerr)
+							time.Sleep(exponentialBackoff)
+							exponentialBackoff = min(exponentialBackoff*2, maxBackoff)
+							break innerLoop
+						}
+						since = next
+						exponentialBackoff = initialBackoff
+						break innerLoop
+					}
 					if err != nil {
-						log.Warningf("etcd watch channel for %s closed with error %v, reconnecting...", key, err)
+						log.Warningf("etcd watch channel for %s closed with error %v, reconnecting from rev %d...", key, err, since)
 					} else {
-						log.Warningf("etcd watch channel for %s closed, reconnecting...", key)
+						log.Warningf("etcd watch channel for %s closed, reconnecting from rev %d...", key, since)
 					}
-					cancel()
 					time.Sleep(exponentialBackoff)
 					exponentialBackoff = min(exponentialBackoff*2, maxBackoff)
 					break innerLoop
 				}
 				exponentialBackoff = initialBackoff // Reset backoff on success
+				// Advance the resume revision so a future reconnect picks up where we
+				// left off rather than replaying from the original start revision.
+				if wresp.Header.Revision != 0 {
+					since = wresp.Header.Revision + 1
+				}
 				batch := make([]lease.LeaseWatchResult, 0)
 				for _, etcdEvent := range wresp.Events {
 					subnetEvent, err := parseSubnetWatchResponse(ctx, esr.cli, etcdEvent)
@@ -581,6 +603,56 @@ func (esr *etcdSubnetRegistry) resyncWatch(ctx context.Context, ch chan []lease.
 	return getNextIndex(wr.Cursor)
 }
 
+// resyncWatchSubnet re-reads a single subnet lease, emits the result on ch and
+// returns the revision to resume from. Single-subnet counterpart to resyncWatch.
+//
+// A lease deleted while the watch was compacted can't be recovered from the
+// watch stream, so synthesize the EventRemoved from its absence and leave the
+// revoke policy to the caller.
+func (esr *etcdSubnetRegistry) resyncWatchSubnet(ctx context.Context, sn ip.IP4Net, sn6 ip.IP6Net, ch chan []lease.LeaseWatchResult) (int64, error) {
+	key := path.Join(esr.etcdCfg.Prefix, "subnets", subnet.MakeSubnetKey(sn, sn6))
+	resp, err := esr.kv().Get(ctx, key)
+	if err != nil {
+		return 0, err
+	}
+
+	var wr lease.LeaseWatchResult
+	if len(resp.Kvs) == 0 {
+		wr = lease.LeaseWatchResult{
+			Events: []lease.Event{{
+				Type: lease.EventRemoved,
+				Lease: lease.Lease{
+					EnableIPv4: true,
+					Subnet:     sn,
+					EnableIPv6: !sn6.Empty(),
+					IPv6Subnet: sn6,
+				},
+			}},
+			Cursor: watchCursor{resp.Header.Revision},
+		}
+	} else {
+		ttlresp, err := esr.cli.TimeToLive(ctx, etcd.LeaseID(resp.Kvs[0].Lease))
+		if err != nil {
+			return 0, err
+		}
+		l, err := kvToIPLease(resp.Kvs[0], ttlresp.TTL)
+		if err != nil {
+			return 0, err
+		}
+		wr = lease.LeaseWatchResult{
+			Snapshot: []lease.Lease{*l},
+			Cursor:   watchCursor{resp.Header.Revision},
+		}
+	}
+
+	select {
+	case ch <- []lease.LeaseWatchResult{wr}:
+	case <-ctx.Done():
+		return 0, ctx.Err()
+	}
+	return getNextIndex(wr.Cursor)
+}
+
 // leasesWatchReset is called when incremental lease watch failed and we need to grab a snapshot
 func (esr *etcdSubnetRegistry) leasesWatchReset(ctx context.Context) (lease.LeaseWatchResult, error) {
 	wr := lease.LeaseWatchResult{}

pkg/subnet/etcd/registry_test.go

@@ -282,6 +282,117 @@ func TestWatchSubnetsRecoversFromCompaction(t *testing.T) {
 	}
 }
 
+// TestWatchSubnetRecoversFromCompaction is the single-subnet counterpart to
+// TestWatchSubnetsRecoversFromCompaction. watchSubnet had the same hot-loop:
+// it reconnected at a now-compacted revision forever and never delivered
+// anything again. After the fix it must re-read the lease and resume.
+func TestWatchSubnetRecoversFromCompaction(t *testing.T) {
+	integration.BeforeTestExternal(t)
+
+	clus := integration.NewCluster(t, &integration.ClusterConfig{Size: 1})
+	defer clus.Terminate(t)
+
+	client := clus.RandClient()
+	ctx := context.Background()
+
+	r, kvApi := newTestEtcdRegistry(t, ctx, client)
+
+	if _, err := kvApi.Put(ctx, "/coreos.com/network/config",
+		`{ "Network": "10.1.0.0/16", "Backend": { "Type": "host-gw" } }`); err != nil {
+		t.Fatal("Failed to put network config", err)
+	}
+
+	sn := ip.IP4Net{IP: ip.MustParseIP4("10.1.5.0"), PrefixLen: 24}
+	attrs := &lease.LeaseAttrs{PublicIP: ip.MustParseIP4("1.2.3.4")}
+	if _, err := r.createSubnet(ctx, sn, ip.IP6Net{}, attrs, 24*time.Hour); err != nil {
+		t.Fatal("Failed to create subnet lease", err)
+	}
+
+	// Advance the store revision, then compact past it so watching from an old
+	// revision is guaranteed to hit "required revision has been compacted".
+	var compactRev int64
+	for i := 0; i < 5; i++ {
+		resp, err := kvApi.Put(ctx, "/coreos.com/network/_bump", fmt.Sprintf("%d", i))
+		if err != nil {
+			t.Fatal("Failed to bump revision", err)
+		}
+		compactRev = resp.Header.Revision
+	}
+	if _, err := client.Compact(ctx, compactRev); err != nil {
+		t.Fatal("Failed to compact etcd", err)
+	}
+
+	receiver := make(chan []lease.LeaseWatchResult, 16)
+	go func() { _ = r.watchSubnet(ctx, 1, sn, ip.IP6Net{}, receiver) }()
+
+	// Recovery must surface a re-read snapshot carrying the watched lease.
+	if !waitForSnapshot(receiver, sn, 10*time.Second) {
+		t.Fatal("watchSubnet did not recover from compaction with a re-read snapshot")
+	}
+
+	// Renewing the lease produces a live event, proving the watch resumed at a
+	// current revision instead of staying stuck on the compacted one.
+	if _, err := r.updateSubnet(ctx, sn, ip.IP6Net{}, attrs, 24*time.Hour, 0); err != nil {
+		t.Fatal("Failed to update subnet lease", err)
+	}
+	if !waitForEvent(receiver, lease.EventAdded, sn, 10*time.Second) {
+		t.Fatal("watchSubnet did not resume delivering live events after recovery")
+	}
+}
+
+// TestWatchSubnetReportsRemovalAfterCompaction pins the design call: recovery
+// resumes the watch as it already does after an ordinary delete, but a lease
+// deleted below the compaction horizon is reported as a synthesized
+// EventRemoved rather than silently missed. That leaves the "lease revoked,
+// shut down" decision with the caller.
+func TestWatchSubnetReportsRemovalAfterCompaction(t *testing.T) {
+	integration.BeforeTestExternal(t)
+
+	clus := integration.NewCluster(t, &integration.ClusterConfig{Size: 1})
+	defer clus.Terminate(t)
+
+	client := clus.RandClient()
+	ctx := context.Background()
+
+	r, kvApi := newTestEtcdRegistry(t, ctx, client)
+
+	if _, err := kvApi.Put(ctx, "/coreos.com/network/config",
+		`{ "Network": "10.1.0.0/16", "Backend": { "Type": "host-gw" } }`); err != nil {
+		t.Fatal("Failed to put network config", err)
+	}
+
+	sn := ip.IP4Net{IP: ip.MustParseIP4("10.1.5.0"), PrefixLen: 24}
+	attrs := &lease.LeaseAttrs{PublicIP: ip.MustParseIP4("1.2.3.4")}
+	if _, err := r.createSubnet(ctx, sn, ip.IP6Net{}, attrs, 24*time.Hour); err != nil {
+		t.Fatal("Failed to create subnet lease", err)
+	}
+
+	// Delete the lease, then compact past the deletion. This is the case the
+	// watch cannot observe directly: the delete event itself is now below the
+	// compaction horizon, so only a re-read can discover it.
+	if _, err := kvApi.Delete(ctx, "/coreos.com/network/subnets/10.1.5.0-24"); err != nil {
+		t.Fatal("Failed to delete subnet lease", err)
+	}
+	var compactRev int64
+	for i := 0; i < 5; i++ {
+		resp, err := kvApi.Put(ctx, "/coreos.com/network/_bump", fmt.Sprintf("%d", i))
+		if err != nil {
+			t.Fatal("Failed to bump revision", err)
+		}
+		compactRev = resp.Header.Revision
+	}
+	if _, err := client.Compact(ctx, compactRev); err != nil {
+		t.Fatal("Failed to compact etcd", err)
+	}
+
+	receiver := make(chan []lease.LeaseWatchResult, 16)
+	go func() { _ = r.watchSubnet(ctx, 1, sn, ip.IP6Net{}, receiver) }()
+
+	if !waitForEvent(receiver, lease.EventRemoved, sn, 10*time.Second) {
+		t.Fatal("watchSubnet did not report the lease as removed after compaction")
+	}
+}
+
 // TestResyncWatchCancelWithBlockedReceiver is a regression test for the
 // ctx-aware send in resyncWatch. Before the fix, a blocked receiver caused
 // resyncWatch to hang indefinitely. After the fix it must return
@@ -370,3 +481,90 @@ func waitForEvent(receiver chan []lease.LeaseWatchResult, etype lease.EventType,
 		}
 	}
 }
+
+// TestWatchSubnetSurvivesCompactionAfterReconnect reproduces the sequence seen
+// in the field, which the tests above only approximate. They start a watch that
+// is already below the compaction horizon; here the watch is established and
+// healthy first, drifts below the horizon while still connected, and only then
+// loses its connection. That drift is what makes this hard to spot in
+// production: the watch works perfectly until something unrelated bounces etcd,
+// possibly hours later, and only then wedges.
+func TestWatchSubnetSurvivesCompactionAfterReconnect(t *testing.T) {
+	integration.BeforeTestExternal(t)
+
+	clus := integration.NewCluster(t, &integration.ClusterConfig{Size: 1})
+	defer clus.Terminate(t)
+
+	client := clus.RandClient()
+	ctx := context.Background()
+
+	r, kvApi := newTestEtcdRegistry(t, ctx, client)
+
+	if _, err := kvApi.Put(ctx, "/coreos.com/network/config",
+		`{ "Network": "10.1.0.0/16", "Backend": { "Type": "host-gw" } }`); err != nil {
+		t.Fatal("Failed to put network config", err)
+	}
+
+	sn := ip.IP4Net{IP: ip.MustParseIP4("10.1.5.0"), PrefixLen: 24}
+	attrs := &lease.LeaseAttrs{PublicIP: ip.MustParseIP4("1.2.3.4")}
+	if _, err := r.createSubnet(ctx, sn, ip.IP6Net{}, attrs, 24*time.Hour); err != nil {
+		t.Fatal("Failed to create subnet lease", err)
+	}
+
+	// Start where a real caller starts: at the revision the lease was read at,
+	// which is current, rather than at one that is already compacted.
+	_, index, err := r.getSubnet(ctx, sn, ip.IP6Net{})
+	if err != nil {
+		t.Fatal("Failed to read subnet lease", err)
+	}
+
+	receiver := make(chan []lease.LeaseWatchResult, 32)
+	go func() { _ = r.watchSubnet(ctx, index+1, sn, ip.IP6Net{}, receiver) }()
+
+	// The watch is healthy to begin with.
+	if _, err := r.updateSubnet(ctx, sn, ip.IP6Net{}, attrs, 24*time.Hour, 0); err != nil {
+		t.Fatal("Failed to update subnet lease", err)
+	}
+	if !waitForEvent(receiver, lease.EventAdded, sn, 10*time.Second) {
+		t.Fatal("watch did not deliver events before the compaction")
+	}
+
+	// Drift below the compaction horizon while still connected. Nothing breaks
+	// yet: an established watch keeps working across a compaction.
+	var compactRev int64
+	for i := 0; i < 5; i++ {
+		resp, err := kvApi.Put(ctx, "/coreos.com/network/_bump", fmt.Sprintf("%d", i))
+		if err != nil {
+			t.Fatal("Failed to bump revision", err)
+		}
+		compactRev = resp.Header.Revision
+	}
+	if _, err := client.Compact(ctx, compactRev); err != nil {
+		t.Fatal("Failed to compact etcd", err)
+	}
+
+	// Bounce etcd. This is the trigger: the watch reconnects at a revision the
+	// store no longer holds.
+	clus.Members[0].Stop(t)
+	if err := clus.Members[0].Restart(t); err != nil {
+		t.Fatal("Failed to restart etcd member", err)
+	}
+	clus.Members[0].WaitOK(t)
+
+	// Recovery surfaces a re-read snapshot rather than the individual events
+	// missed while disconnected, which is inherent to re-listing: the events are
+	// below the horizon and no longer exist to replay.
+	if !waitForSnapshot(receiver, sn, 30*time.Second) {
+		t.Fatal("watch never recovered after reconnecting below the compaction horizon")
+	}
+
+	// Only once recovery has landed is a subsequent change proof that the watch
+	// resumed at a current revision. Doing this before the snapshot would race:
+	// the change would simply be absorbed into the re-read.
+	if _, err := r.updateSubnet(ctx, sn, ip.IP6Net{}, attrs, 24*time.Hour, 0); err != nil {
+		t.Fatal("Failed to update subnet lease after restart", err)
+	}
+	if !waitForEvent(receiver, lease.EventAdded, sn, 30*time.Second) {
+		t.Fatal("watch recovered but stopped delivering live events")
+	}
+}