Code for requeue event once even loop exit, registry cleanup an… (#1281)

* Code added for requeue event once even loop exit, registry cleanup and orphaned registry detection

* Unit and integration tests added for requeueEvent, registry cleanup and orphaned registry detection

* Apply gofumpt formatting

* Removing parallel run of VSS registry tests

* Code added to close dead WebSocket before removal in getOrCreateWebSocket

* Code added to replace infinite retry with errorCount threshold in eventLoop

* Code modified to provide threshold for connection retry and recreate shared websocket after reconnect failure threshold is reached

* Requeue event, registry cleanup and orphan entry code added in VDS+VSS along with testcases

* Test added for vaultDynamicSecret in event_watcher_registry_test.go

* Comments updated

* Fixed WebSocket stop-notification ordering, atomicity, and delivery; eliminated subscriber double-delete race; narrowed the Client interface; deduplicated subscriber code; closed orphan detection gaps; and cleaned up tests.

Author: harshit-nema

controllers/event_watcher_registry_test.go

@@ -1065,6 +1065,49 @@ func (r *VaultDynamicSecretReconciler) ensureEventWatcher(
 	logger := log.FromContext(ctx).WithName("ensureEventWatcher")
 	name := client.ObjectKeyFromObject(o)
 
+	// onStop is shared across all Subscriber structs for this resource so that
+	// only the first OnStop call deletes the registry entry. Without this, a
+	// second or third independent closure could delete the entry that a
+	// reconciler already re-created after the first OnStop fired.
+	var onStopOnce sync.Once
+	onStop := func() {
+		onStopOnce.Do(func() {
+			r.eventWatcherRegistry.Delete(name)
+		})
+	}
+
+	// newDVSSubscriber is a local factory that stamps out a *vault.Subscriber
+	// with all fields common to every subscription for this resource. Only
+	// vaultNS and vaultPath vary between the engine-events subscriber (Step 4)
+	// and the lease-events subscribers (Steps 1 and 5).
+	//
+	// NOTE: lease subscribers intentionally pass vaultNS="" — lease events are
+	// routed by lease ID alone, so setting a namespace would produce a
+	// "<namespace>/<leaseID>" key that never matches the "<leaseID>" lookup.
+	newDVSSubscriber := func(vaultNS, vaultPath string) *vault.Subscriber {
+		return &vault.Subscriber{
+			ResourceKey:       name,
+			VaultNS:           vaultNS,
+			VaultPath:         vaultPath,
+			ResourceType:      vault.ResourceTypeVaultDynamicSecret,
+			ReconcileCh:       r.SourceCh,
+			PendingVaultIndex: &r.pendingVaultIndex,
+			// OnStop cleans up the registry entry when the WebSocket dies.
+			// The log is emitted by notifySubscribersOfStop via ws.logger,
+			// which is always valid (unlike the reconcile-context logger
+			// captured here, which goes stale after Reconcile returns).
+			OnStop: onStop,
+			NewObject: func() client.Object {
+				return &secretsv1beta1.VaultDynamicSecret{
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace: name.Namespace,
+						Name:      name.Name,
+					},
+				}
+			},
+		}
+	}
+
 	currentLeaseID := o.Status.SecretLease.ID
 	meta, hasMeta := r.eventWatcherRegistry.Get(name)
 
@@ -1103,31 +1146,40 @@ func (r *VaultDynamicSecretReconciler) ensureEventWatcher(
 
 		// Lease events are mount-type-independent; still subscribe if applicable.
 		if !o.Spec.AllowStaticCreds && currentLeaseID != "" {
-			if !(hasMeta && meta.LastLeaseID == currentLeaseID && meta.LastClientID == c.ID()) {
-				leaseSubscriber := &vault.Subscriber{
-					ResourceKey:       name,
-					VaultPath:         currentLeaseID,
-					ResourceType:      "VaultDynamicSecret",
-					ReconcileCh:       r.SourceCh,
-					PendingVaultIndex: &r.pendingVaultIndex,
-				}
-				if err := c.SubscribeToEvents(ctx, vault.EventTypeLease, leaseSubscriber); err != nil {
-					// Surface this too, instead of only logging it, so a compound
-					// failure (mount type AND lease subscribe both broken) is visible.
-					resultErr = errors.Join(resultErr, fmt.Errorf("failed to subscribe to lease events: %w", err))
-				} else {
-					// Record what we did establish so the next reconcile can detect
-					// "lease-only subscription already active" and skip re-subscribing
-					// on every cycle while the mount-type lookup keeps failing.
-					// LastEventType is left empty so Step 2's staleness check still
-					// forces a full re-subscribe once GetMountType succeeds.
-					r.eventWatcherRegistry.Register(name, &eventWatcherMeta{
-						LastClientID:   c.ID(),
-						LastGeneration: o.GetGeneration(),
-						LastLeaseID:    currentLeaseID,
-						LastEventType:  "",
-					})
-				}
+			// Skip re-subscription only when the existing lease watcher is
+			// confirmed alive. A metadata-only match is not sufficient: if the
+			// lease WebSocket died (e.g. reconnect threshold exceeded) but
+			// OnStop has not yet cleaned the registry, the entry looks valid
+			// but the resource is actually unsubscribed. The liveness check
+			// catches this orphaned state and falls through to re-subscribe.
+			leaseAlive := hasMeta &&
+				meta.LastLeaseID == currentLeaseID &&
+				meta.LastClientID == c.ID() &&
+				c.IsWebSocketHealthy(vault.EventTypeLease)
+			if leaseAlive {
+				return resultErr
+			}
+			// Dead or missing — clear any stale entry before re-subscribing.
+			if hasMeta {
+				r.eventWatcherRegistry.Delete(name)
+			}
+			leaseSubscriber := newDVSSubscriber("", currentLeaseID)
+			if err := c.SubscribeToEvents(ctx, vault.EventTypeLease, leaseSubscriber); err != nil {
+				// Surface this too, instead of only logging it, so a compound
+				// failure (mount type AND lease subscribe both broken) is visible.
+				resultErr = errors.Join(resultErr, fmt.Errorf("failed to subscribe to lease events: %w", err))
+			} else {
+				// Record what we did establish so the next reconcile can detect
+				// "lease-only subscription already active" and skip re-subscribing
+				// on every cycle while the mount-type lookup keeps failing.
+				// LastEventType is left empty so Step 2's staleness check still
+				// forces a full re-subscribe once GetMountType succeeds.
+				r.eventWatcherRegistry.Register(name, &eventWatcherMeta{
+					LastClientID:   c.ID(),
+					LastGeneration: o.GetGeneration(),
+					LastLeaseID:    currentLeaseID,
+					LastEventType:  "",
+				})
 			}
 		}
 		return resultErr
@@ -1141,9 +1193,19 @@ func (r *VaultDynamicSecretReconciler) ensureEventWatcher(
 		meta.LastClientID == c.ID() &&
 		meta.LastLeaseID == currentLeaseID &&
 		meta.LastEventType == eventType {
-		logger.V(consts.LogLevelDebug).Info("Event subscription already active",
+		// Orphaned entry detection: verify the WebSocket is actually alive.
+		// This is a safety net for cases where OnStop did not fire (e.g. operator
+		// restart) or there was a race between WebSocket death and OnStop cleanup.
+		if c.IsWebSocketHealthy(eventType) {
+			logger.V(consts.LogLevelDebug).Info("Event subscription already active",
+				"namespace", o.Namespace, "name", o.Name)
+			return nil
+		}
+		// WebSocket is dead or missing — orphaned registry entry detected.
+		logger.Info("Detected orphaned registry entry (WebSocket is dead or missing), cleaning up",
 			"namespace", o.Namespace, "name", o.Name)
-		return nil
+		r.eventWatcherRegistry.Delete(name)
+		hasMeta = false
 	}
 
 	// Step 3: tear down the stale subscription now that we have a valid type.
@@ -1155,14 +1217,7 @@ func (r *VaultDynamicSecretReconciler) ensureEventWatcher(
 
 	// Step 4: subscribe to engine events.
 	vaultPath := buildVaultEventKey(o)
-	subscriber := &vault.Subscriber{
-		ResourceKey:       name,
-		VaultNS:           o.Spec.Namespace,
-		VaultPath:         vaultPath,
-		ResourceType:      "VaultDynamicSecret",
-		ReconcileCh:       r.SourceCh,
-		PendingVaultIndex: &r.pendingVaultIndex,
-	}
+	subscriber := newDVSSubscriber(o.Spec.Namespace, vaultPath)
 	if err := c.SubscribeToEvents(ctx, eventType, subscriber); err != nil {
 		return fmt.Errorf("failed to subscribe to %s events: %w", eventType, err)
 	}
@@ -1176,13 +1231,7 @@ func (r *VaultDynamicSecretReconciler) ensureEventWatcher(
 	// unWatchEventsWithLeaseID), so setting VaultNS here would key the subscriber
 	// as "<namespace>/<leaseID>" and never match the "<leaseID>" lookup.
 	if !o.Spec.AllowStaticCreds && o.Status.SecretLease.ID != "" {
-		leaseSubscriber := &vault.Subscriber{
-			ResourceKey:       name,
-			VaultPath:         currentLeaseID,
-			ResourceType:      "VaultDynamicSecret",
-			ReconcileCh:       r.SourceCh,
-			PendingVaultIndex: &r.pendingVaultIndex,
-		}
+		leaseSubscriber := newDVSSubscriber("", currentLeaseID)
 		if err := c.SubscribeToEvents(ctx, vault.EventTypeLease, leaseSubscriber); err != nil {
 			// Non-fatal: the database/LDAP subscription above still provides
 			// event-driven updates. Surface the failure as a warning event so

controllers/vaultdynamicsecret_controller.go

@@ -2443,14 +2443,22 @@ type mockEnsureClient struct {
 	mountTypeErr    error
 	subscribed      []vault.EventType
 	seen            []vault.EventType // from UnsubscribeFromEvents
+	// webSocketHealthy is returned by IsWebSocketHealthy.
+	webSocketHealthy bool
+	// onSubscribe is an optional hook called on every SubscribeToEvents call,
+	// allowing tests to capture Subscriber fields such as OnStop.
+	onSubscribe func(vault.EventType, *vault.Subscriber)
 }
 
 func (m *mockEnsureClient) GetMountType(_ context.Context, _ string) (string, error) {
 	return m.mountTypeResult, m.mountTypeErr
 }
 
-func (m *mockEnsureClient) SubscribeToEvents(_ context.Context, et vault.EventType, _ *vault.Subscriber) error {
+func (m *mockEnsureClient) SubscribeToEvents(_ context.Context, et vault.EventType, sub *vault.Subscriber) error {
 	m.subscribed = append(m.subscribed, et)
+	if m.onSubscribe != nil {
+		m.onSubscribe(et, sub)
+	}
 	return nil
 }
 
@@ -2461,6 +2469,10 @@ func (m *mockEnsureClient) UnsubscribeFromEvents(et vault.EventType, _ vault.Sub
 
 func (m *mockEnsureClient) ID() string { return "test-client" }
 
+func (m *mockEnsureClient) IsWebSocketHealthy(_ vault.EventType) bool {
+	return m.webSocketHealthy
+}
+
 // Test_ensureEventWatcher_GetMountTypeError_ReusesPriorEventType verifies
 // that when GetMountType fails but a prior LastEventType is stored, that prior
 // type is reused — keeping the subscription on the correct event stream instead
@@ -2694,3 +2706,100 @@ func TestVaultDynamicSecretReconciler_syncSecret_vaultIndex(t *testing.T) {
 		})
 	}
 }
+
+// Test_ensureEventWatcher_OrphanedEntry_NilWebSocket verifies that when the
+// registry has a matching entry but IsWebSocketHealthy returns false (e.g. after
+// an operator restart), ensureEventWatcher detects the orphaned entry, clears it,
+// and re-subscribes rather than returning nil.
+func Test_ensureEventWatcher_OrphanedEntry_NilWebSocket(t *testing.T) {
+	ch := make(chan event.GenericEvent, 10)
+	r := &VaultDynamicSecretReconciler{
+		eventWatcherRegistry: newEventWatcherRegistry(),
+		SourceCh:             ch,
+		Recorder:             record.NewFakeRecorder(10),
+	}
+
+	o := &secretsv1beta1.VaultDynamicSecret{
+		ObjectMeta: metav1.ObjectMeta{Namespace: "default", Name: "db-secret", Generation: 1},
+		Spec: secretsv1beta1.VaultDynamicSecretSpec{
+			Mount: "database",
+			Path:  "creds/my-role",
+		},
+	}
+	key := client.ObjectKeyFromObject(o)
+
+	// Pre-populate registry with fully-matching metadata so the staleness check
+	// would normally return nil — the only thing that should trigger re-subscribe
+	// is the nil WebSocket.
+	r.eventWatcherRegistry.Register(key, &eventWatcherMeta{
+		LastClientID:   "test-client",
+		LastGeneration: 1,
+		LastLeaseID:    "",
+		LastEventType:  vault.EventTypeDatabase,
+	})
+
+	// IsWebSocketHealthy returns false — simulates operator restart with no live WebSocket.
+	m := &mockEnsureClient{
+		mountTypeResult:  "database",
+		webSocketHealthy: false,
+	}
+
+	err := r.ensureEventWatcher(context.Background(), o, m)
+
+	require.NoError(t, err)
+	assert.Contains(t, m.subscribed, vault.EventTypeDatabase,
+		"must re-subscribe when WebSocket is nil (orphaned entry)")
+
+	// Registry must be refreshed with new metadata after re-subscription.
+	meta, ok := r.eventWatcherRegistry.Get(key)
+	require.True(t, ok, "registry must have an entry after re-subscription")
+	assert.Equal(t, vault.EventTypeDatabase, meta.LastEventType)
+}
+
+// Test_VDS_ensureEventWatcher_OnStop_CleansRegistry verifies that the OnStop
+// callback set on the VDS engine-events Subscriber deletes the registry entry
+// when invoked, so the next reconcile falls through to re-subscribe instead of
+// returning early because it sees a stale registry entry with matching metadata.
+func Test_VDS_ensureEventWatcher_OnStop_CleansRegistry(t *testing.T) {
+	ch := make(chan event.GenericEvent, 10)
+	r := &VaultDynamicSecretReconciler{
+		eventWatcherRegistry: newEventWatcherRegistry(),
+		SourceCh:             ch,
+		Recorder:             record.NewFakeRecorder(10),
+	}
+
+	o := &secretsv1beta1.VaultDynamicSecret{
+		ObjectMeta: metav1.ObjectMeta{Namespace: "default", Name: "db-secret", Generation: 1},
+		Spec: secretsv1beta1.VaultDynamicSecretSpec{
+			Mount: "database",
+			Path:  "creds/my-role",
+		},
+	}
+	key := client.ObjectKeyFromObject(o)
+
+	// captureOnStop intercepts the first OnStop set on any SubscribeToEvents call.
+	var capturedOnStop func()
+	m := &mockEnsureClient{
+		mountTypeResult: "database",
+		onSubscribe: func(_ vault.EventType, sub *vault.Subscriber) {
+			if capturedOnStop == nil && sub.OnStop != nil {
+				capturedOnStop = sub.OnStop
+			}
+		},
+	}
+
+	err := r.ensureEventWatcher(context.Background(), o, m)
+	require.NoError(t, err)
+
+	// Registry must have an entry after subscription.
+	_, ok := r.eventWatcherRegistry.Get(key)
+	require.True(t, ok, "registry must have entry after ensureEventWatcher")
+
+	// Simulate WebSocket death by invoking the captured OnStop callback.
+	require.NotNil(t, capturedOnStop, "OnStop must be set on the engine-events subscriber")
+	capturedOnStop()
+
+	// Registry entry must be gone — next reconcile will re-subscribe.
+	_, ok = r.eventWatcherRegistry.Get(key)
+	assert.False(t, ok, "registry entry must be deleted after OnStop fires")
+}

controllers/vaultdynamicsecret_controller_test.go

@@ -373,19 +373,46 @@ func (r *VaultStaticSecretReconciler) ensureEventWatcher(ctx context.Context, o
 	logger := log.FromContext(ctx).WithName("ensureEventWatcher")
 	name := client.ObjectKeyFromObject(o)
 
+	// onStop is shared across all Subscriber structs for this resource so that
+	// only the first OnStop call deletes the registry entry. A single subscriber
+	// is registered today, but the Once guard makes the pattern consistent with
+	// VaultDynamicSecret and safe if additional subscribers are added in future.
+	var onStopOnce sync.Once
+	onStop := func() {
+		onStopOnce.Do(func() {
+			r.eventWatcherRegistry.Delete(name)
+		})
+	}
+
+	// vssEventType is the single event type VSS subscribes to. Declaring it once
+	// here keeps the orphan check and the subscribe call in sync — a future change
+	// to the VSS event type only needs to be made in one place.
+	const vssEventType = vault.EventTypeKV
+
 	meta, ok := r.eventWatcherRegistry.Get(name)
 	if ok {
-		// The subscription is active, and if the VSS object has not been updated,
-		// and the client ID is the same, just return
-		if meta.LastGeneration == o.GetGeneration() && meta.LastClientID == c.ID() {
-			logger.V(consts.LogLevelDebug).Info("Event subscription already active",
+		// Check if the WebSocket is actually healthy (Orphaned Entry Detection)
+		// This is a safety net in case OnStop callback didn't run or there was a race condition
+		if c.IsWebSocketHealthy(vssEventType) {
+			// WebSocket is healthy, check if metadata matches
+			if meta.LastGeneration == o.GetGeneration() && meta.LastClientID == c.ID() {
+				// The subscription is active, and if the VSS object has not been updated,
+				// and the client ID is the same, just return
+				logger.V(consts.LogLevelDebug).Info("Event subscription already active",
+					"namespace", o.Namespace, "name", o.Name)
+				return nil
+			}
+			// The subscription exists but metadata or vault client has changed, unsubscribe first
+			logger.V(consts.LogLevelDebug).Info("Unsubscribing due to metadata or client change",
+				"namespace", o.Namespace, "name", o.Name)
+			r.unWatchEvents(o, c)
+		} else {
+			// WebSocket is dead or missing - orphaned registry entry detected
+			logger.Info("Detected orphaned registry entry (WebSocket is dead or missing), cleaning up",
 				"namespace", o.Namespace, "name", o.Name)
-			return nil
+			r.eventWatcherRegistry.Delete(name)
+			// Fall through to create a new WebSocket subscription
 		}
-		// The subscription exists but metadata or vault client has changed, unsubscribe first
-		logger.V(consts.LogLevelDebug).Info("Unsubscribing due to metadata or client change",
-			"namespace", o.Namespace, "name", o.Name)
-		r.unWatchEvents(o, c)
 	}
 
 	// Build the vault path for subscription
@@ -396,12 +423,22 @@ func (r *VaultStaticSecretReconciler) ensureEventWatcher(ctx context.Context, o
 		ResourceKey:       name,
 		VaultNS:           o.Spec.Namespace,
 		VaultPath:         vaultPath,
-		ResourceType:      "VaultStaticSecret",
+		ResourceType:      vault.ResourceTypeVaultStaticSecret,
 		ReconcileCh:       r.SourceCh,
 		PendingVaultIndex: &r.pendingVaultIndex,
+		// OnStop callback cleans up registry when WebSocket dies
+		OnStop: onStop,
+		NewObject: func() client.Object {
+			return &secretsv1beta1.VaultStaticSecret{
+				ObjectMeta: metav1.ObjectMeta{
+					Namespace: name.Namespace,
+					Name:      name.Name,
+				},
+			}
+		},
 	}
 
-	if err := c.SubscribeToEvents(ctx, vault.EventTypeKV, subscriber); err != nil {
+	if err := c.SubscribeToEvents(ctx, vssEventType, subscriber); err != nil {
 		return fmt.Errorf("failed to subscribe to events: %w", err)
 	}