Fix the write/write race on chunk.Owner

drsm/chunk.go

@@ -14,8 +14,37 @@ import (
 	"go.mongodb.org/mongo-driver/v2/bson"
 )
 
+// GetOwner returns a copy of the pod recorded as owning c. Returning &c.Owner would let the
+// caller read the fields after FindOwnerInt32ID has dropped its lock, while claimChunk and
+// the change-stream handler are still writing them.
 func (c *chunk) GetOwner() *PodId {
-	return &c.Owner
+	c.ownerMutex.Lock()
+	defer c.ownerMutex.Unlock()
+	owner := c.Owner
+	return &owner
+}
+
+// setOwner replaces the recorded owner of c.
+func (c *chunk) setOwner(owner PodId) {
+	c.ownerMutex.Lock()
+	defer c.ownerMutex.Unlock()
+	c.Owner = owner
+}
+
+// setOwnerAddress records this pod as the owner. PodInstance is left as it was, which is what
+// claimChunk has always done - the change-stream update that the claim triggers supplies it.
+func (c *chunk) setOwnerAddress(podName, podIp string) {
+	c.ownerMutex.Lock()
+	defer c.ownerMutex.Unlock()
+	c.Owner.PodName = podName
+	c.Owner.PodIp = podIp
+}
+
+// ownerPodName returns the name of the pod recorded as owning c.
+func (c *chunk) ownerPodName() string {
+	c.ownerMutex.Lock()
+	defer c.ownerMutex.Unlock()
+	return c.Owner.PodName
 }
 
 func (d *Drsm) GetNewChunk() (*chunk, error) {

drsm/chunk_test.go

@@ -0,0 +1,109 @@
+// SPDX-FileCopyrightText: 2026 Forsway Scandinavia AB
+//
+// SPDX-License-Identifier: Apache-2.0
+
+package drsm
+
+import (
+	"sync"
+	"testing"
+)
+
+const (
+	podXName   = "pod-x"
+	podXIp     = "10.0.0.11"
+	podYName   = "pod-y"
+	podYIp     = "10.0.0.22"
+	ownerLoops = 2000
+)
+
+// ownerPairs is the set of consistent owner records a reader may legitimately observe. A
+// PodName paired with the other pod's PodIp means a read caught a half-finished write.
+var ownerPairs = map[string]string{podXName: podXIp, podYName: podYIp}
+
+// TestChunkOwnerConcurrentAccess drives the two writers of chunk.Owner against its two
+// readers: claimChunk records this pod on a successful claim, the change-stream handler
+// records the owner from the update that the claim triggered, scanChunk checks whether it
+// still owns the chunk, and FindOwnerInt32ID hands the owner to the caller.
+func TestChunkOwnerConcurrentAccess(t *testing.T) {
+	c := &chunk{Id: 1, Owner: PodId{PodName: podXName, PodIp: podXIp, PodInstance: podXName + "-1"}}
+
+	var wg sync.WaitGroup
+	wg.Add(4)
+
+	// The change-stream handler replaces the whole record.
+	go func() {
+		defer wg.Done()
+		for i := range ownerLoops {
+			if i%2 == 0 {
+				c.setOwner(PodId{PodName: podXName, PodIp: podXIp, PodInstance: podXName + "-1"})
+			} else {
+				c.setOwner(PodId{PodName: podYName, PodIp: podYIp, PodInstance: podYName + "-1"})
+			}
+		}
+	}()
+
+	// claimChunk sets only the name and the address.
+	go func() {
+		defer wg.Done()
+		for i := range ownerLoops {
+			if i%2 == 0 {
+				c.setOwnerAddress(podXName, podXIp)
+			} else {
+				c.setOwnerAddress(podYName, podYIp)
+			}
+		}
+	}()
+
+	// FindOwnerInt32ID's reader must never see a torn pair.
+	go func() {
+		defer wg.Done()
+		for range ownerLoops {
+			owner := c.GetOwner()
+			want, known := ownerPairs[owner.PodName]
+			if !known {
+				t.Errorf("GetOwner returned unknown PodName %q", owner.PodName)
+				return
+			}
+			if owner.PodIp != want {
+				t.Errorf("torn owner: PodName %q with PodIp %q, want %q", owner.PodName, owner.PodIp, want)
+				return
+			}
+		}
+	}()
+
+	// scanChunk's ownership check.
+	go func() {
+		defer wg.Done()
+		for range ownerLoops {
+			if _, known := ownerPairs[c.ownerPodName()]; !known {
+				t.Errorf("ownerPodName returned unknown pod %q", c.ownerPodName())
+				return
+			}
+		}
+	}()
+
+	wg.Wait()
+}
+
+// TestGetOwnerReturnsSnapshot covers the reason GetOwner copies: callers read the fields
+// after FindOwnerInt32ID has released globalChunkTblMutex, so they must not hold a pointer
+// into the chunk.
+func TestGetOwnerReturnsSnapshot(t *testing.T) {
+	c := &chunk{Id: 1, Owner: PodId{PodName: podXName, PodIp: podXIp}}
+
+	owner := c.GetOwner()
+	owner.PodName = "scribbled"
+	if got := c.ownerPodName(); got != podXName {
+		t.Errorf("writing to the returned PodId changed the chunk: owner is %q, want %q", got, podXName)
+	}
+
+	before := c.GetOwner()
+	c.setOwner(PodId{PodName: podYName, PodIp: podYIp})
+	if before.PodName != podXName {
+		t.Errorf("a later setOwner mutated an earlier snapshot: %q, want %q", before.PodName, podXName)
+	}
+	if got := c.ownerPodName(); got != podYName {
+		t.Errorf("setOwner did not take effect: %q, want %q", got, podYName)
+	}
+}

drsm/claim.go

@@ -43,8 +43,7 @@ func (c *chunk) claimChunk(d *Drsm, curOwner string) {
 	if updated == nil {
 		// TODO : don't add to local pool yet. We can add it only if scan is done.
 		logger.DrsmLog.Infof("claimChunk %v success", c.Id)
-		c.Owner.PodName = d.clientId.PodName
-		c.Owner.PodIp = d.clientId.PodIp
+		c.setOwnerAddress(d.clientId.PodName, d.clientId.PodIp)
 		go c.scanChunk(d)
 	} else {
 		// no problem, some other POD successfully claimed this chunk

drsm/drsm.go

@@ -32,6 +32,8 @@ type chunk struct {
 	ScanIds         []int32
 	stopScan        chan bool
 	resourceValidCb func(int32) bool
+	// ownerMutex guards Owner, which claimChunk and the change-stream handler both write.
+	ownerMutex sync.Mutex
 }
 
 type podData struct {

drsm/scan.go

@@ -15,7 +15,7 @@ func (c *chunk) scanChunk(d *Drsm) {
 		return
 	}
 
-	if c.Owner.PodName != d.clientId.PodName {
+	if c.ownerPodName() != d.clientId.PodName {
 		logger.DrsmLog.Infoln("do not perform scan task if Chunk is not owned by us")
 		return
 	}

drsm/updates.go

@@ -180,9 +180,11 @@ func iterateChangeStream(d *Drsm, routineCtx context.Context, stream *mongo.Chan
 					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
+				cp.setOwner(PodId{
+					PodName:     owner,
+					PodIp:       s.Update.UpdFields.PodIp,
+					PodInstance: s.Update.UpdFields.PodInstance,
+				})
 				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)