Fix regression notifying workloads when entry removed (spiffe#3923)

PR spiffe#2305 fixed spurious notifications of workloads when nothing had
changed but unfortunately introduced a regression wherein a workload is
no longer notified by the cache when an entry for that workload is
removed.

The bug is caused by false sharing of the selRem temporary selector set.
Previously selRem was used to build the selectors for entries being
removed and the the contents of selRem were merged into the single
notification. When multiple notification sets were introduced, selRem
was added as a notification set. Unfortunately, selRem is cleared while
processing entries, causing the notification set to be empty.

Existing unit-tests did not catch this because the update that removes
the existing entry did not have additional entries to be processed (that
would cause selRem to be cleared).

This PR fixes the bug by allocating a new selector set to be appended to
the notification set instead of using selRem. It also cleans up some
selector set usage and adds some additional logic to the unit-test so
this condition can be caught in the future.

Fixes: spiffe#3922

Signed-off-by: Andrew Harding <[email protected]>

Author: azdagron

pkg/agent/manager/cache/cache.go

@@ -271,10 +271,10 @@ func (c *Cache) UpdateEntries(update *UpdateEntries, checkSVID func(*common.Regi
 			// built a set of selectors for the record being removed, drop the
 			// record for each selector index, and add the entry selectors to
 			// the notify set.
-			clearSelectorSet(selRem)
-			selRem.Merge(record.entry.Selectors...)
-			c.delSelectorIndicesRecord(selRem, record)
-			notifySets = append(notifySets, selRem)
+			notifySet, notifySetDone := allocSelectorSet(record.entry.Selectors...)
+			defer notifySetDone()
+			c.delSelectorIndicesRecord(notifySet, record)
+			notifySets = append(notifySets, notifySet)
 			delete(c.records, id)
 			// Remove stale entry since, registration entry is no longer on cache.
 			delete(c.staleEntries, id)
@@ -324,17 +324,15 @@ func (c *Cache) UpdateEntries(update *UpdateEntries, checkSVID func(*common.Regi
 		// are notified.
 		if selectorsChanged {
 			if existingEntry != nil {
-				notifySet, selSetDone := allocSelectorSet()
-				defer selSetDone()
-				notifySet.Merge(existingEntry.Selectors...)
+				notifySet, notifySetDone := allocSelectorSet(existingEntry.Selectors...)
+				defer notifySetDone()
 				notifySets = append(notifySets, notifySet)
 			}
 		}
 
 		if federatedBundlesChanged || selectorsChanged {
-			notifySet, selSetDone := allocSelectorSet()
-			defer selSetDone()
-			notifySet.Merge(newEntry.Selectors...)
+			notifySet, notifySetDone := allocSelectorSet(newEntry.Selectors...)
+			defer notifySetDone()
 			notifySets = append(notifySets, notifySet)
 		}
 
@@ -385,8 +383,8 @@ func (c *Cache) UpdateSVIDs(update *UpdateSVIDs) {
 	defer c.mu.Unlock()
 
 	// Allocate a set of selectors that
-	notifySet, selSetDone := allocSelectorSet()
-	defer selSetDone()
+	notifySet, notifySetDone := allocSelectorSet()
+	defer notifySetDone()
 
 	// Add/update records for registration entries in the update
 	for entryID, svid := range update.X509SVIDs {

pkg/agent/manager/cache/cache_test.go

@@ -400,6 +400,8 @@ func TestSubcriberNotifiedWhenEntryDropped(t *testing.T) {
 	assertAnyWorkloadUpdate(t, subB)
 
 	foo := makeRegistrationEntry("FOO", "A")
+	bar := makeRegistrationEntry("BAR", "B")
+
 	updateEntries := &UpdateEntries{
 		Bundles:             makeBundles(bundleV1),
 		RegistrationEntries: makeRegistrationEntries(foo),
@@ -408,19 +410,35 @@ func TestSubcriberNotifiedWhenEntryDropped(t *testing.T) {
 	cache.UpdateSVIDs(&UpdateSVIDs{
 		X509SVIDs: makeX509SVIDs(foo),
 	})
+
 	// make sure subA gets notified with FOO but not subB
 	assertWorkloadUpdateEqual(t, subA, &WorkloadUpdate{
 		Bundle:     bundleV1,
 		Identities: []Identity{{Entry: foo}},
 	})
 	assertNoWorkloadUpdate(t, subB)
 
-	updateEntries.RegistrationEntries = nil
+	// Swap out FOO for BAR
+	updateEntries.RegistrationEntries = makeRegistrationEntries(bar)
 	cache.UpdateEntries(updateEntries, nil)
+	cache.UpdateSVIDs(&UpdateSVIDs{
+		X509SVIDs: makeX509SVIDs(bar),
+	})
 	assertWorkloadUpdateEqual(t, subA, &WorkloadUpdate{
 		Bundle: bundleV1,
 	})
-	assertNoWorkloadUpdate(t, subB)
+	assertWorkloadUpdateEqual(t, subB, &WorkloadUpdate{
+		Bundle:     bundleV1,
+		Identities: []Identity{{Entry: bar}},
+	})
+
+	// Drop both
+	updateEntries.RegistrationEntries = nil
+	cache.UpdateEntries(updateEntries, nil)
+	assertNoWorkloadUpdate(t, subA)
+	assertWorkloadUpdateEqual(t, subB, &WorkloadUpdate{
+		Bundle: bundleV1,
+	})
 
 	// Make sure trying to update SVIDs of removed entry does not notify
 	cache.UpdateSVIDs(&UpdateSVIDs{

pkg/agent/manager/cache/lru_cache.go

@@ -273,10 +273,10 @@ func (c *LRUCache) UpdateEntries(update *UpdateEntries, checkSVID func(*common.R
 			// built a set of selectors for the record being removed, drop the
 			// record for each selector index, and add the entry selectors to
 			// the notify set.
-			clearSelectorSet(selRem)
-			selRem.Merge(record.entry.Selectors...)
-			c.delSelectorIndicesRecord(selRem, record)
-			notifySets = append(notifySets, selRem)
+			notifySet, notifySetDone := allocSelectorSet(record.entry.Selectors...)
+			defer notifySetDone()
+			c.delSelectorIndicesRecord(notifySet, record)
+			notifySets = append(notifySets, notifySet)
 			delete(c.records, id)
 			delete(c.svids, id)
 			// Remove stale entry since, registration entry is no longer on cache.
@@ -329,17 +329,15 @@ func (c *LRUCache) UpdateEntries(update *UpdateEntries, checkSVID func(*common.R
 		// are notified.
 		if selectorsChanged {
 			if existingEntry != nil {
-				notifySet, selSetDone := allocSelectorSet()
-				defer selSetDone()
-				notifySet.Merge(existingEntry.Selectors...)
+				notifySet, notifySetDone := allocSelectorSet(existingEntry.Selectors...)
+				defer notifySetDone()
 				notifySets = append(notifySets, notifySet)
 			}
 		}
 
 		if federatedBundlesChanged || selectorsChanged {
-			notifySet, selSetDone := allocSelectorSet()
-			defer selSetDone()
-			notifySet.Merge(newEntry.Selectors...)
+			notifySet, notifySetDone := allocSelectorSet(newEntry.Selectors...)
+			defer notifySetDone()
 			notifySets = append(notifySets, notifySet)
 		}
 
@@ -427,8 +425,8 @@ func (c *LRUCache) UpdateSVIDs(update *UpdateSVIDs) {
 	defer c.mu.Unlock()
 
 	// Allocate a set of selectors that
-	notifySet, selSetDone := allocSelectorSet()
-	defer selSetDone()
+	notifySet, notifySetDone := allocSelectorSet()
+	defer notifySetDone()
 
 	// Add/update records for registration entries in the update
 	for entryID, svid := range update.X509SVIDs {

pkg/agent/manager/cache/lru_cache_test.go

@@ -387,6 +387,8 @@ func TestLRUCacheSubscriberNotifiedWhenEntryDropped(t *testing.T) {
 	assertAnyWorkloadUpdate(t, subB)
 
 	foo := makeRegistrationEntry("FOO", "A")
+	bar := makeRegistrationEntry("BAR", "B")
+
 	updateEntries := &UpdateEntries{
 		Bundles:             makeBundles(bundleV1),
 		RegistrationEntries: makeRegistrationEntries(foo),
@@ -395,19 +397,35 @@ func TestLRUCacheSubscriberNotifiedWhenEntryDropped(t *testing.T) {
 	cache.UpdateSVIDs(&UpdateSVIDs{
 		X509SVIDs: makeX509SVIDs(foo),
 	})
+
 	// make sure subA gets notified with FOO but not subB
 	assertWorkloadUpdateEqual(t, subA, &WorkloadUpdate{
 		Bundle:     bundleV1,
 		Identities: []Identity{{Entry: foo}},
 	})
 	assertNoWorkloadUpdate(t, subB)
 
-	updateEntries.RegistrationEntries = nil
+	// Swap out FOO for BAR
+	updateEntries.RegistrationEntries = makeRegistrationEntries(bar)
 	cache.UpdateEntries(updateEntries, nil)
+	cache.UpdateSVIDs(&UpdateSVIDs{
+		X509SVIDs: makeX509SVIDs(bar),
+	})
 	assertWorkloadUpdateEqual(t, subA, &WorkloadUpdate{
 		Bundle: bundleV1,
 	})
-	assertNoWorkloadUpdate(t, subB)
+	assertWorkloadUpdateEqual(t, subB, &WorkloadUpdate{
+		Bundle:     bundleV1,
+		Identities: []Identity{{Entry: bar}},
+	})
+
+	// Drop both
+	updateEntries.RegistrationEntries = nil
+	cache.UpdateEntries(updateEntries, nil)
+	assertNoWorkloadUpdate(t, subA)
+	assertWorkloadUpdateEqual(t, subB, &WorkloadUpdate{
+		Bundle: bundleV1,
+	})
 
 	// Make sure trying to update SVIDs of removed entry does not notify
 	cache.UpdateSVIDs(&UpdateSVIDs{