Track multiple sources per vanflow record during purge.

cmd/network-observer/internal/collector/collector.go

@@ -318,11 +318,7 @@ func (c *Collector) handleStoreDelete(e store.Entry) {
 }
 
 func (c *Collector) purge(source store.SourceRef) int {
-	matching := c.Records.Index(store.SourceIndex, store.Entry{Metadata: store.Metadata{Source: source}})
-	for _, record := range matching {
-		c.Records.Delete(record.Record.Identity())
-	}
-	return len(matching)
+	return c.Records.RemoveSource(source)
 }
 
 func (c *Collector) discoveryHandler(ctx context.Context) func(eventsource.Info) {

cmd/network-observer/internal/collector/processes.go

@@ -155,7 +155,7 @@ func (m *processManager) run(ctx context.Context) func() error {
 								)
 								for _, procID := range procIDs {
 									if procEntry, ok := m.stor.Get(procID); ok {
-										if procEntry.Source == m.source {
+										if procEntry.HasSource(m.source) {
 											toDelete = procID
 										} else {
 											replacedBy = procID
@@ -171,17 +171,17 @@ func (m *processManager) run(ctx context.Context) func() error {
 										slog.String("host", host),
 										slog.String("replaced_by", replacedBy),
 									)
-									m.stor.Delete(toDelete)
+									m.stor.DetachSource(toDelete, m.source)
 								}
 							}
 						}
 						for _, proc := range processes {
-							if proc.Source == m.source {
+							if proc.HasSource(m.source) {
 								_, ok := actualProcessHosts[proc.Record.Identity()]
 								if ok {
 									continue
 								}
-								if _, deleted := m.stor.Delete(proc.Record.Identity()); deleted {
+								if _, detached := m.stor.DetachSource(proc.Record.Identity(), m.source); detached {
 									m.logger.Info("Deleting site server process with no connectors",
 										slog.String("id", proc.Record.Identity()),
 										slog.String("site_id", siteID),

internal/flow/status.go

@@ -385,11 +385,7 @@ func (s *StatusSync) notify() {
 }
 
 func (s *StatusSync) purge(source store.SourceRef) int {
-	matching := s.records.Index(store.SourceIndex, store.Entry{Metadata: store.Metadata{Source: source}})
-	for _, record := range matching {
-		s.records.Delete(record.Record.Identity())
-	}
-	return len(matching)
+	return s.records.RemoveSource(source)
 }
 
 func (s *StatusSync) handleDiscovery(source eventsource.Info) {

pkg/vanflow/store/store.go

@@ -1,6 +1,7 @@
 package store
 
 import (
+	"slices"
 	"time"
 
 	"github.com/skupperproject/skupper/pkg/vanflow"
@@ -16,7 +17,49 @@ type Entry struct {
 type Metadata struct {
 	LastUpdate time.Time
 
-	Source SourceRef
+	// Sources is the set of sources that have asserted this record.
+	Sources []SourceRef
+}
+
+func sourceRefEqual(a, b SourceRef) bool {
+	return a.ID == b.ID && a.Version == b.Version
+}
+
+func (m Metadata) HasSource(source SourceRef) bool {
+	for _, existing := range m.Sources {
+		if sourceRefEqual(existing, source) {
+			return true
+		}
+	}
+	return false
+}
+
+func (m *Metadata) AddSource(source SourceRef) bool {
+	for _, existing := range m.Sources {
+		if sourceRefEqual(existing, source) {
+			return false
+		}
+	}
+	m.Sources = append(slices.Clone(m.Sources), source)
+	return true
+}
+
+func (m *Metadata) RemoveSource(source SourceRef) bool {
+	i := slices.IndexFunc(m.Sources, func(existing SourceRef) bool {
+		return sourceRefEqual(existing, source)
+	})
+	if i < 0 {
+		return false
+	}
+	m.Sources = slices.Delete(slices.Clone(m.Sources), i, i+1)
+	return true
+}
+
+func newMetadata(source SourceRef) Metadata {
+	return Metadata{
+		LastUpdate: time.Now(),
+		Sources:    []SourceRef{source},
+	}
 }
 
 // SourceRef identifies a record source
@@ -41,4 +84,13 @@ type Interface interface {
 	IndexValues(index string) []string
 
 	Replace([]Entry)
+
+	// RemoveSource detaches all records from the given source, deleting any
+	// that are no longer asserted by any source.
+	RemoveSource(source SourceRef) int
+
+	// DetachSource removes source from the record identified by id. The record
+	// is deleted only when no sources remain. Returns the entry state before
+	// the operation and whether the source was detached.
+	DetachSource(id string, source SourceRef) (Entry, bool)
 }

pkg/vanflow/store/syncmap.go

@@ -32,6 +32,8 @@ type SyncMapStoreConfig struct {
 func NewSyncMapStore(cfg SyncMapStoreConfig) Interface {
 	if cfg.Indexers == nil {
 		cfg.Indexers = defaultIndexers()
+	} else if cfg.Indexers[SourceIndex] == nil {
+		cfg.Indexers[SourceIndex] = SourceIndexer
 	}
 	return &syncMapStore{
 		indexers:      cfg.Indexers,
@@ -42,30 +44,48 @@ func NewSyncMapStore(cfg SyncMapStoreConfig) Interface {
 	}
 }
 
-func (m *syncMapStore) Add(record vanflow.Record, source SourceRef) bool {
+func cloneEntry(entry Entry) Entry {
+	entry.Sources = append([]SourceRef(nil), entry.Sources...)
+	return entry
+}
 
-	entry, ok := func() (Entry, bool) {
-		key := record.Identity()
+func (m *syncMapStore) Add(record vanflow.Record, source SourceRef) bool {
+	var entry Entry
+	var added bool
+	var prev Entry
+	var sourceAdded bool
 
-		entry := Entry{
-			Metadata: Metadata{LastUpdate: time.Now(), Source: source},
-			Record:   record,
-		}
+	key := record.Identity()
+	entry = Entry{
+		Metadata: newMetadata(source),
+		Record:   record,
+	}
 
-		m.mu.Lock()
-		defer m.mu.Unlock()
-		if _, exists := m.items[key]; exists {
-			return entry, false
+	m.mu.Lock()
+	if curr, exists := m.items[key]; exists {
+		entry = curr
+		prev = cloneEntry(curr)
+		if curr.AddSource(source) {
+			curr.LastUpdate = time.Now()
+			m.items[key] = curr
+			m.reindex(key, &prev, curr)
+			entry = curr
+			sourceAdded = true
 		}
+	} else {
 		m.items[key] = entry
 		m.reindex(key, nil, entry)
-		return entry, true
-	}()
+		added = true
+	}
+	m.mu.Unlock()
 
-	if ok && m.eventHandlers.OnAdd != nil {
+	if added && m.eventHandlers.OnAdd != nil {
 		m.eventHandlers.OnAdd(entry)
 	}
-	return ok
+	if sourceAdded && m.eventHandlers.OnChange != nil {
+		m.eventHandlers.OnChange(prev, entry)
+	}
+	return added
 }
 
 func (m *syncMapStore) Update(record vanflow.Record) bool {
@@ -154,10 +174,19 @@ func (m *syncMapStore) Patch(record vanflow.Record, source SourceRef) {
 			}
 		}
 		if !changed {
+			prev = cloneEntry(curr)
+			if curr.AddSource(source) {
+				next = curr
+				next.LastUpdate = time.Now()
+				m.items[key] = next
+				m.reindex(key, &prev, next)
+				return prev, next, ok, nil
+			}
 			return prev, next, noChange, nil
 		}
 
-		prev = curr
+		prev = cloneEntry(curr)
+		curr.AddSource(source)
 		next = curr
 		patched, err := encoding.Decode(currAttrs)
 		next.Record = patched.(vanflow.Record)
@@ -231,6 +260,97 @@ func (m *syncMapStore) IndexValues(index string) []string {
 	return values
 }
 
+func (m *syncMapStore) detachSourceLocked(key string, source SourceRef) (prev Entry, next Entry, detached bool, deleted bool) {
+	curr, exists := m.items[key]
+	if !exists {
+		return prev, next, false, false
+	}
+	prev = cloneEntry(curr)
+	if !curr.RemoveSource(source) {
+		return prev, next, false, false
+	}
+	if len(curr.Sources) == 0 {
+		delete(m.items, key)
+		m.unindex(key, prev)
+		return prev, next, true, true
+	}
+	curr.LastUpdate = time.Now()
+	m.items[key] = curr
+	m.reindex(key, &prev, curr)
+	return prev, curr, true, false
+}
+
+func (m *syncMapStore) DetachSource(id string, source SourceRef) (Entry, bool) {
+	m.mu.Lock()
+	prev, next, detached, deleted := m.detachSourceLocked(id, source)
+	m.mu.Unlock()
+
+	if !detached {
+		return prev, false
+	}
+	if deleted {
+		if m.eventHandlers.OnDelete != nil {
+			m.eventHandlers.OnDelete(prev)
+		}
+		return prev, true
+	}
+	if m.eventHandlers.OnChange != nil {
+		m.eventHandlers.OnChange(prev, next)
+	}
+	return prev, true
+}
+
+func (m *syncMapStore) RemoveSource(source SourceRef) int {
+	var deleted []Entry
+	var changed []struct {
+		prev Entry
+		next Entry
+	}
+	count := 0
+
+	m.mu.Lock()
+	indexer := m.indexers[SourceIndex]
+	if indexer != nil {
+		idx := m.indices[SourceIndex]
+		if idx != nil {
+			keys := make(keySet)
+			for _, indexVal := range indexer(Entry{Metadata: Metadata{Sources: []SourceRef{source}}}) {
+				for key := range idx[indexVal] {
+					keys.Add(key)
+				}
+			}
+			for key := range keys {
+				prev, next, detached, recordDeleted := m.detachSourceLocked(key, source)
+				if !detached {
+					continue
+				}
+				count++
+				if recordDeleted {
+					deleted = append(deleted, prev)
+					continue
+				}
+				changed = append(changed, struct {
+					prev Entry
+					next Entry
+				}{prev: prev, next: next})
+			}
+		}
+	}
+	m.mu.Unlock()
+
+	for _, entry := range deleted {
+		if m.eventHandlers.OnDelete != nil {
+			m.eventHandlers.OnDelete(entry)
+		}
+	}
+	for _, update := range changed {
+		if m.eventHandlers.OnChange != nil {
+			m.eventHandlers.OnChange(update.prev, update.next)
+		}
+	}
+	return count
+}
+
 func (m *syncMapStore) Replace(items []Entry) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
@@ -308,8 +428,16 @@ const (
 	TypeIndex   = "ByType"
 )
 
+func sourceIndexKey(source SourceRef) string {
+	return fmt.Sprintf("%s/%s", source.Version, source.ID)
+}
+
 func SourceIndexer(e Entry) []string {
-	return []string{fmt.Sprintf("%s/%s", e.Source.Version, e.Metadata.Source.ID)}
+	keys := make([]string, 0, len(e.Sources))
+	for _, source := range e.Sources {
+		keys = append(keys, sourceIndexKey(source))
+	}
+	return keys
 }
 func TypeIndexer(e Entry) []string {
 	return []string{e.Record.GetTypeMeta().String()}