fix(discov): prevent etcd key leak when Watch DELETE races with lease expiry

The publisher's Watch-based self-healing (added after v1.8) re-puts
the key with p.lease whenever a DELETE event is observed. When the
DELETE is caused by lease expiry and races with the KeepAlive channel
close, the publisher may Put with an expired or zero LeaseID, which
makes the key permanent in etcd (no TTL) and leaks forever.

This fix adds a TimeToLive check before re-putting: if the lease is
already expired or gone, the publisher skips the Put and immediately
restarts the keepalive flow instead, avoiding orphaned keys.

Author: antengye

core/discov/internal/etcdclient.go

@@ -19,5 +19,6 @@ type EtcdClient interface {
 	KeepAlive(ctx context.Context, id clientv3.LeaseID) (<-chan *clientv3.LeaseKeepAliveResponse, error)
 	Put(ctx context.Context, key, val string, opts ...clientv3.OpOption) (*clientv3.PutResponse, error)
 	Revoke(ctx context.Context, id clientv3.LeaseID) (*clientv3.LeaseRevokeResponse, error)
+	TimeToLive(ctx context.Context, id clientv3.LeaseID, opts ...clientv3.LeaseOption) (*clientv3.LeaseTimeToLiveResponse, error)
 	Watch(ctx context.Context, key string, opts ...clientv3.OpOption) clientv3.WatchChan
 }

core/discov/internal/etcdclient_mock.go

@@ -151,6 +151,24 @@ func (p *Publisher) keepAliveAsync(cli internal.EtcdClient) error {
 					if evt.Type == clientv3.EventTypeDelete {
 						logc.Infof(cli.Ctx(), "etcd publisher watch: %s, event: %v",
 							evt.Kv.Key, evt.Type)
+
+						// Make sure the lease is still valid before re-putting the key.
+						// Otherwise the Put may happen with an already-expired or zero
+						// LeaseID (e.g. when the DELETE event is caused by lease expiry
+						// and races with KeepAlive channel close), which makes the key
+						// permanent in etcd (no TTL) and leaks forever.
+						ttlResp, ttlErr := cli.TimeToLive(cli.Ctx(), p.lease)
+						if ttlErr != nil || ttlResp == nil || ttlResp.TTL <= 0 {
+							logc.Errorf(cli.Ctx(),
+								"etcd publisher lease expired, skip re-put and restart keepalive: leaseID=%d, err=%v",
+								p.lease, ttlErr)
+							p.revoke(cli)
+							if err := p.doKeepAlive(); err != nil {
+								logc.Errorf(cli.Ctx(), "etcd publisher KeepAlive: %v", err)
+							}
+							return
+						}
+
 						_, err := cli.Put(cli.Ctx(), p.fullKey, p.value, clientv3.WithLease(p.lease))
 						if err != nil {
 							logc.Errorf(cli.Ctx(), "etcd publisher re-put key: %v", err)