From 661d728d9111e3c901370149e98918bb0d65cb17 Mon Sep 17 00:00:00 2001 From: Edvin Lindqvist Date: Thu, 20 Aug 2026 10:35:41 +0200 Subject: [PATCH] Fix the write/write race on chunk.Owner Two goroutines write the owner of the same chunk with no synchronisation, and the second one is triggered by the first: - claimChunk records this pod once the claim succeeds (claim.go:46-47); - iterateChangeStream records the owner from the update that the claim just made, since it is watching that collection (updates.go:183-185). The same fields are read from two more places: scanChunk checks whether it still owns the chunk (scan.go:18), and FindOwnerInt32ID hands the owner to the caller. That last path leaks the race past the package boundary. GetOwner returned &c.Owner, so although FindOwnerInt32ID holds globalChunkTblMutex while calling it, the caller dereferences the pointer after that lock has been released, while both writers are still running. The visible effect is a torn PodId: a PodName belonging to one pod paired with another pod's PodIp, which points the caller at the wrong pod. The chunk_test.go assertion reproduces exactly that with the new locks removed, without needing the race detector: torn owner: PodName "pod-x" with PodIp "10.0.0.22", want "10.0.0.11" Owner is now guarded by a mutex on the chunk itself rather than on Drsm, because these are chunk methods and GetOwner has no Drsm to reach a shared lock through. The mutex is a leaf: none of the four helpers calls anything else, so the only nesting is globalChunkTblMutex then ownerMutex in FindOwnerInt32ID, and nothing takes them the other way round. GetOwner keeps its signature and returns a pointer to a copy, so callers holding the result are unaffected by later writes and no consumer has to change. setOwnerAddress deliberately leaves PodInstance alone, which is what claimChunk has always done - the change-stream update that the claim triggers is what fills it in. That is preserved rather than fixed here. Co-Authored-By: Claude Opus 5 (1M context) Signed-off-by: Edvin Lindqvist --- drsm/chunk.go | 31 ++++++++++++- drsm/chunk_test.go | 109 +++++++++++++++++++++++++++++++++++++++++++++ drsm/claim.go | 3 +- drsm/drsm.go | 2 + drsm/scan.go | 2 +- drsm/updates.go | 8 ++-- 6 files changed, 148 insertions(+), 7 deletions(-) create mode 100644 drsm/chunk_test.go diff --git a/drsm/chunk.go b/drsm/chunk.go index 3a8eb94..ad971fb 100644 --- a/drsm/chunk.go +++ b/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) { diff --git a/drsm/chunk_test.go b/drsm/chunk_test.go new file mode 100644 index 0000000..5109004 --- /dev/null +++ b/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) + } +} diff --git a/drsm/claim.go b/drsm/claim.go index fa3363a..a69fd38 100644 --- a/drsm/claim.go +++ b/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 diff --git a/drsm/drsm.go b/drsm/drsm.go index 9ac3522..1924b8e 100644 --- a/drsm/drsm.go +++ b/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 { diff --git a/drsm/scan.go b/drsm/scan.go index d86dc39..5d74484 100644 --- a/drsm/scan.go +++ b/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 } diff --git a/drsm/updates.go b/drsm/updates.go index 44f3aa2..dcb843b 100644 --- a/drsm/updates.go +++ b/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)