subnet/etcd: unify the two subnet watch loops (#2539)

* subnet/etcd: unify the two subnet watch loops

watchSubnets and watchSubnet were near-identical copies of one reconnect
loop. That is why the compaction fix landed in the first (#2471) months
before the second (#2524): nothing connected them, so fixing one did not
even suggest the other.

Both now call a shared runWatch that owns the reconnect loop, backoff,
compaction recovery and revision bookkeeping. What genuinely differs
stays in a subnetWatch struct: the key, start revision, result channel
and resync callback.

The two response converters are unified too. Their apparent differences
were artifacts of copied code: the parser already sets EnableIPv4,
compaction is reported by the watch response rather than event parsing,
and cursor-only or empty results misrepresent parse failures as
snapshots. The shared converter skips malformed events, while TTL lookup
failures retry from the same revision instead of advancing past an
event.

watchSubnet picks up the resync retry loop watchSubnets already had.
Retry waits and result delivery honor context cancellation, so a blocked
receiver cannot keep the watch alive. Each per-attempt watch context is
also canceled before reconnecting rather than accumulating deferred
cleanup for the lifetime of the loop.

* subnet/etcd: give the two watch functions the same argument order

watchSubnets took (ctx, chan, since) while watchSubnet took (ctx, since,
sn, sn6, chan), so two adjacent functions in the same interface
disagreed about where the revision and the channel belong.

Both are now (ctx, since, [subnet], chan), which is watchSubnet's
existing shape, so only watchSubnets moves. Mechanical: the interface,
the mock, one caller in local_manager.go and two in the tests.

Author: phinze

pkg/subnet/etcd/local_manager.go

@@ -352,7 +352,7 @@ func (m *LocalManager) WatchLeases(ctx context.Context, receiver chan []lease.Le
 		return err
 	}
 
-	err = m.registry.watchSubnets(ctx, receiver, nextIndex)
+	err = m.registry.watchSubnets(ctx, nextIndex, receiver)
 	if err != nil {
 		return err
 	}
@@ -406,10 +406,6 @@ func (m *LocalManager) CompleteLease(ctx context.Context, myLease *lease.Lease,
 	}
 }
 
-func isIndexTooSmall(err error) bool {
-	return err == rpctypes.ErrGRPCCompacted
-}
-
 func isSubnetConfigCompat(config *subnet.Config, sn ip.IP4Net) bool {
 	if sn.IP < config.SubnetMin || sn.IP > config.SubnetMax {
 		return false

pkg/subnet/etcd/mock_registry.go

@@ -197,7 +197,7 @@ func (msr *MockSubnetRegistry) deleteSubnet(ctx context.Context, sn ip.IP4Net, s
 	return nil
 }
 
-func (msr *MockSubnetRegistry) watchSubnets(ctx context.Context, leaseWatchChan chan []lease.LeaseWatchResult, since int64) error {
+func (msr *MockSubnetRegistry) watchSubnets(ctx context.Context, since int64, leaseWatchChan chan []lease.LeaseWatchResult) error {
 	log.Infof("watchSubnets started with since= [ %d]", since)
 	for {
 		msr.mux.Lock()