+autoremoval of stale discovery records

api/pkg/discovery/service.go

@@ -3,6 +3,7 @@ package discovery
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"strings"
 	"sync"
 	"time"
@@ -17,6 +18,8 @@ import (
 	"go.uber.org/zap"
 )
 
+const DefaultRegistryStaleCleanupInterval = time.Duration(DefaultHealthIntervalSec) * time.Second
+
 type RegistryOption func(*RegistryService)
 
 func WithRegistryKVTTL(ttl time.Duration) RegistryOption {
@@ -28,15 +31,28 @@ func WithRegistryKVTTL(ttl time.Duration) RegistryOption {
 	}
 }
 
+// WithRegistryStaleCleanupInterval configures periodic stale-entry eviction.
+func WithRegistryStaleCleanupInterval(interval time.Duration) RegistryOption {
+	return func(s *RegistryService) {
+		if s == nil {
+			return
+		}
+		s.staleCleanupInterval = interval
+	}
+}
+
 type RegistryService struct {
-	logger    mlogger.Logger
-	registry  *Registry
-	producer  msg.Producer
-	sender    string
-	consumers []consumerHandler
-	kv        *KVStore
-	kvWatcher nats.KeyWatcher
-	kvOptions []KVStoreOption
+	logger               mlogger.Logger
+	registry             *Registry
+	producer             msg.Producer
+	sender               string
+	consumers            []consumerHandler
+	kv                   *KVStore
+	kvWatcher            nats.KeyWatcher
+	kvOptions            []KVStoreOption
+	staleCleanupInterval time.Duration
+	staleCleanupStop     chan struct{}
+	staleCleanupDone     chan struct{}
 
 	startOnce sync.Once
 	stopOnce  sync.Once
@@ -83,10 +99,11 @@ func NewRegistryService(logger mlogger.Logger, msgBroker mb.Broker, producer msg
 	}
 
 	svc := &RegistryService{
-		logger:   logger,
-		registry: registry,
-		producer: producer,
-		sender:   sender,
+		logger:               logger,
+		registry:             registry,
+		producer:             producer,
+		sender:               sender,
+		staleCleanupInterval: DefaultRegistryStaleCleanupInterval,
 	}
 	for _, opt := range opts {
 		if opt != nil {
@@ -128,6 +145,7 @@ func (s *RegistryService) Start() {
 			}()
 		}
 		s.startKVWatch()
+		s.startStaleCleanup()
 	})
 }
 
@@ -144,6 +162,12 @@ func (s *RegistryService) Stop() {
 		if s.kvWatcher != nil {
 			_ = s.kvWatcher.Stop()
 		}
+		if s.staleCleanupStop != nil {
+			close(s.staleCleanupStop)
+		}
+		if s.staleCleanupDone != nil {
+			<-s.staleCleanupDone
+		}
 		s.logInfo("Discovery registry service stopped")
 	})
 }
@@ -305,6 +329,56 @@ func (s *RegistryService) startKVWatch() {
 	go s.consumeKVUpdates(watcher)
 }
 
+func (s *RegistryService) startStaleCleanup() {
+	if s == nil || s.registry == nil || s.staleCleanupInterval <= 0 {
+		return
+	}
+
+	s.staleCleanupStop = make(chan struct{})
+	s.staleCleanupDone = make(chan struct{})
+
+	go func() {
+		defer close(s.staleCleanupDone)
+
+		ticker := time.NewTicker(s.staleCleanupInterval)
+		defer ticker.Stop()
+
+		s.pruneStaleEntries(time.Now())
+		for {
+			select {
+			case <-ticker.C:
+				s.pruneStaleEntries(time.Now())
+			case <-s.staleCleanupStop:
+				return
+			}
+		}
+	}()
+}
+
+func (s *RegistryService) pruneStaleEntries(now time.Time) {
+	if s == nil || s.registry == nil {
+		return
+	}
+
+	removedKeys := s.registry.DeleteStale(now)
+	if len(removedKeys) == 0 {
+		return
+	}
+
+	for _, key := range removedKeys {
+		if strings.TrimSpace(key) == "" {
+			continue
+		}
+		s.logInfo("Discovery registry stale entry removed", zap.String("key", key))
+		if s.kv == nil {
+			continue
+		}
+		if err := s.kv.Delete(key); err != nil && !errors.Is(err, nats.ErrKeyNotFound) {
+			s.logWarn("Failed to delete stale discovery KV entry", zap.String("key", key), zap.Error(err))
+		}
+	}
+}
+
 func (s *RegistryService) consumeKVUpdates(watcher nats.KeyWatcher) {
 	if s == nil || watcher == nil {
 		return