Fix DRSM nil pointer crashes in distributed deployments (#197)

* drsm: fix nil pointer crashes in change stream handler

Add defensive nil checks to prevent crashes when processing MongoDB
change stream events in distributed deployments.

Fixes three crash scenarios:
1. Empty owner field - skip to prevent resource leaks
2. Nil chunk pointer - chunk not yet in global table (out-of-order events)
3. Nil pod pointer - pod not yet registered locally

Skip invalid updates with warning logs rather than crashing. Eventual
consistency maintained by periodic checkAllChunks() resync (3 seconds).

Tested with multiple instances during pod failures and
network partitions.

Signed-off-by: Edvin Lindqvist <[email protected]>

* simplified owner check and centralized podChunks verification/initialization

Co-authored-by: Arrobo, Gabriel <[email protected]>
Signed-off-by: Edvin Lindqvist <[email protected]>

* Stepped up version

Signed-off-by: Edvin Lindqvist <[email protected]>

---------

Signed-off-by: Edvin Lindqvist <[email protected]>
Co-authored-by: Edvin Lindqvist <[email protected]>
Co-authored-by: Arrobo, Gabriel <[email protected]>

Author: 3 people

VERSION

@@ -1 +1 @@
-1.5.7-dev
+1.5.7

drsm/updates.go

@@ -110,6 +110,12 @@ func (d *Drsm) handleDbUpdates() {
 	}
 }
 
+func (d *Drsm) ensurePodChunksInitialized(podD *podData) {
+	if podD.podChunks == nil {
+		podD.podChunks = make(map[int32]*chunk)
+	}
+}
+
 func iterateChangeStream(d *Drsm, routineCtx context.Context, stream *mongo.ChangeStream) {
 	logger.DrsmLog.Debugf("iterate change stream for podData: %v", d)
 
@@ -154,15 +160,32 @@ func iterateChangeStream(d *Drsm, routineCtx context.Context, stream *mongo.Chan
 				// update on chunkId..
 				// looks like chunk owner getting change
 				owner := s.Update.UpdFields.PodId
+				if owner == "" {
+					logger.DrsmLog.Warnf("stream(Update): missing owner in update for doc %s, operation: %+v", s.DId.Id, s.Update)
+					continue
+				}
 				c := getChunkIdFromDocId(s.DId.Id)
 				d.globalChunkTblMutex.Lock()
 				cp := d.globalChunkTbl[c]
 				d.globalChunkTblMutex.Unlock()
+				if cp == nil {
+					logger.DrsmLog.Warnf("stream(Update): chunk %d not found in global table for owner %s - will be corrected by periodic resync", c, owner)
+					// Without a chunk reference there is nothing to update; skip to avoid panic.
+					// The periodic checkAllChunks() will resync state from MongoDB.
+					continue
+				}
 				// TODO update IP address as well.
 				cp.Owner.PodName = owner
 				cp.Owner.PodIp = s.Update.UpdFields.PodIp
 				cp.Owner.PodInstance = s.Update.UpdFields.PodInstance
-				podD := d.podMap[owner]
+				podD, found := d.podMap[owner]
+				if !found {
+					logger.DrsmLog.Warnf("stream(Update): pod %s not in local map for chunk %d update - will be corrected when keepalive arrives or during periodic resync", owner, c)
+					// Wait for proper pod initialization via keepalive. Eventual consistency will be maintained by periodic resync and proper keepalive events.
+					continue
+				}
+				// Defensive: should never happen if addPod() was called, but prevents panic
+				d.ensurePodChunksInitialized(podD)
 				podD.podChunks[c] = cp // add chunk to pod
 				logger.DrsmLog.Infof("stream(Update): pod to chunk map %v", podD.podChunks)
 			}
@@ -270,7 +293,7 @@ func (d *Drsm) addChunk(full *FullStream) {
 func (d *Drsm) addPod(full *FullStream) *podData {
 	podI := PodId{PodName: full.PodId, PodInstance: full.PodInstance, PodIp: full.PodIp}
 	pod := &podData{PodId: podI}
-	pod.podChunks = make(map[int32]*chunk)
+	d.ensurePodChunksInitialized(pod)
 	d.podMap[full.PodId] = pod
 	logger.DrsmLog.Infof("keepalive insert d.podMaps %v", d.podMap)
 	return pod