use sync.Once - Events should return the same event channel

Signed-off-by: Davanum Srinivas <[email protected]>

Author: dims

store-client/pkg/client/mongodb_client.go

@@ -19,6 +19,7 @@ import (
 	"fmt"
 	"log/slog"
 	"strings"
+	"sync"
 
 	"github.com/nvidia/nvsentinel/data-models/pkg/model"
 	"github.com/nvidia/nvsentinel/store-client/pkg/config"
@@ -887,26 +888,33 @@ type mongoSessionContext struct {
 }
 
 type mongoChangeStreamWatcher struct {
-	watcher *mongoWatcher.ChangeStreamWatcher
+	watcher   *mongoWatcher.ChangeStreamWatcher
+	eventChan chan Event
+	initOnce  sync.Once
 }
 
 func (w *mongoChangeStreamWatcher) Start(ctx context.Context) {
 	w.watcher.Start(ctx)
 }
 
+// Events returns the event channel for this watcher.
+// The returned channel is initialized once and cached, so multiple calls
+// return the same channel. Safe for concurrent use.
 func (w *mongoChangeStreamWatcher) Events() <-chan Event {
-	bsonChan := w.watcher.Events()
-	eventChan := make(chan Event)
+	w.initOnce.Do(func() {
+		w.eventChan = make(chan Event)
+		bsonChan := w.watcher.Events()
 
-	go func() {
-		defer close(eventChan)
+		go func() {
+			defer close(w.eventChan)
 
-		for rawEvent := range bsonChan {
-			eventChan <- &mongoEvent{rawEvent: rawEvent}
-		}
-	}()
+			for rawEvent := range bsonChan {
+				w.eventChan <- &mongoEvent{rawEvent: rawEvent}
+			}
+		}()
+	})
 
-	return eventChan
+	return w.eventChan
 }
 
 func (w *mongoChangeStreamWatcher) MarkProcessed(ctx context.Context, token []byte) error {

store-client/pkg/datastore/providers/mongodb/adapter.go

@@ -20,6 +20,7 @@ import (
 	"log/slog"
 	"os"
 	"path/filepath"
+	"sync"
 
 	"github.com/nvidia/nvsentinel/store-client/pkg/client"
 	"github.com/nvidia/nvsentinel/store-client/pkg/config"
@@ -164,7 +165,9 @@ func (a *AdaptedMongoStore) CreateChangeStreamWatcher(ctx context.Context, clien
 
 // AdaptedChangeStreamWatcher adapts our existing change stream watcher to the new interface
 type AdaptedChangeStreamWatcher struct {
-	watcher client.ChangeStreamWatcher
+	watcher   client.ChangeStreamWatcher
+	eventChan chan datastore.EventWithToken
+	initOnce  sync.Once
 }
 
 // NewAdaptedChangeStreamWatcher creates a new adapted change stream watcher
@@ -173,33 +176,36 @@ func NewAdaptedChangeStreamWatcher(watcher client.ChangeStreamWatcher) datastore
 }
 
 // Events returns the events channel
+// CRITICAL FIX: Only create the channel and goroutine ONCE to prevent event loss
 func (a *AdaptedChangeStreamWatcher) Events() <-chan datastore.EventWithToken {
-	eventChan := make(chan datastore.EventWithToken)
+	a.initOnce.Do(func() {
+		a.eventChan = make(chan datastore.EventWithToken)
 
-	go func() {
-		defer close(eventChan)
+		go func() {
+			defer close(a.eventChan)
 
-		for event := range a.watcher.Events() {
-			// Convert from our existing Event interface to the new EventWithToken
-			eventMap := make(map[string]interface{})
+			for event := range a.watcher.Events() {
+				// Convert from our existing Event interface to the new EventWithToken
+				eventMap := make(map[string]interface{})
 
-			// Extract the event data
-			if err := event.UnmarshalDocument(&eventMap); err != nil {
-				slog.Error("Failed to unmarshal event", "error", err)
-				continue
-			}
+				// Extract the event data
+				if err := event.UnmarshalDocument(&eventMap); err != nil {
+					slog.Error("Failed to unmarshal event", "error", err)
+					continue
+				}
 
-			// Create EventWithToken
-			eventWithToken := datastore.EventWithToken{
-				Event:       eventMap,
-				ResumeToken: []byte(""), // We'll need to extract the resume token properly
-			}
+				// Create EventWithToken
+				eventWithToken := datastore.EventWithToken{
+					Event:       eventMap,
+					ResumeToken: []byte(""), // We'll need to extract the resume token properly
+				}
 
-			eventChan <- eventWithToken
-		}
-	}()
+				a.eventChan <- eventWithToken
+			}
+		}()
+	})
 
-	return eventChan
+	return a.eventChan
 }
 
 // Start starts the watcher

tests/helpers/health_events_analyzer.go

@@ -18,7 +18,6 @@ import (
 	"context"
 	"math/rand"
 	"testing"
-	"time"
 
 	pb "github.com/nvidia/nvsentinel/data-models/pkg/protos"
 	"github.com/stretchr/testify/require"