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)