From 4370038bf4b26b2b623bc2f2a14702a45e0dad2a Mon Sep 17 00:00:00 2001 From: Edvin Lindqvist Date: Thu, 20 Aug 2026 09:58:38 +0200 Subject: [PATCH] Fix concurrent map access on the shared drsm tables An AMF or SMF pod aborts with an unrecoverable "fatal error: concurrent map writes" when a peer pod goes down during a rolling update. The stack is entirely inside drsm: runtime.mapassign_fast32 drsm.(*chunk).scanChunk drsm/scan.go:23 drsm.(*chunk).claimChunk.gowrap1 podDownDetected starts one claimChunk goroutine per reclaimed chunk, and each successful claim starts a scanChunk goroutine that writes d.scanChunks. A pod that owned two or more chunks therefore races itself. Drsm declares a single mutex, globalChunkTblMutex, and it guards only globalChunkTbl. scanChunks is not the only unguarded table, so this covers all four: - scanChunks written in scan.go, read in ReleaseInt32ID - localChunkTbl written in scan.go, ranged in AllocateInt32ID - podMap written by addPod from both the change stream and the checkAllChunks ticker, read by podDownDetected - podChunks written by addChunk from those same two goroutines, ranged by podDownDetected localChunkTbl matters because scan.go writes it on the line after the reported crash, so guarding only scanChunks moves the abort to "concurrent map iteration and map write". podMap needs no pod-down event at all: addChunk reaches addPod from the change-stream goroutine and from the 3-second checkAllChunks ticker, so two goroutines insert into it concurrently in steady state. The two scan tables already have a lock - the package-level mutex held by AllocateInt32ID and ReleaseInt32ID - and the scan goroutines simply did not take it, so startScan and completeScan now do. The critical sections cover only the map operations; resourceValidCb is still invoked without the lock held, so a callback that re-enters drsm cannot deadlock. podMap and the podChunks reachable through it get a new podMapMutex, and every access moves behind a helper that holds it. No path holds two locks at once: addChunk releases podMapMutex before taking globalChunkTblMutex, and the delete handler signals podDown after releasing it, since podDownDetected acquires the same lock. Two behaviour changes fall out of the refactor: - podDownDetected dereferenced d.podMap[p] without checking it was present, which would panic for an unknown pod. podChunkIds returns no ids instead. - addChunk logs podChunks before writing globalChunkTbl rather than after, because that log has to read the map under the lock. drsm had no tests. drsm_test.go adds three that drive the real call paths with no MongoDB connection; the two concurrency tests report 10 to 20 data races per run with the new locks removed, and pass with them in place. A separate write/write race on chunk.Owner is left alone here: claimChunk sets it on a successful claim while the change-stream handler sets the same fields from the update that claim triggered. It is a different defect from the tables and gets its own change. Co-Authored-By: Claude Opus 5 (1M context) Signed-off-by: Edvin Lindqvist --- drsm/claim.go | 6 +- drsm/drsm.go | 2 + drsm/drsm_test.go | 195 ++++++++++++++++++++++++++++++++++++++++++++++ drsm/scan.go | 27 +++++-- drsm/updates.go | 91 +++++++++++++++++----- 5 files changed, 292 insertions(+), 29 deletions(-) create mode 100644 drsm/drsm_test.go diff --git a/drsm/claim.go b/drsm/claim.go index fa3363a..9ab4f7d 100644 --- a/drsm/claim.go +++ b/drsm/claim.go @@ -15,14 +15,14 @@ func (d *Drsm) podDownDetected() { for p := range d.podDown { logger.DrsmLog.Infof("pod Down detected %v", p) // Given Pod find out current Chunks owned by this POD - pd := d.podMap[p] - for k := range pd.podChunks { + ids, owner := d.podChunkIds(p) + for _, k := range ids { d.globalChunkTblMutex.Lock() c, found := d.globalChunkTbl[k] d.globalChunkTblMutex.Unlock() logger.DrsmLog.Debugf("found: %v chunk: %v", found, c) if found { - go c.claimChunk(d, pd.PodId.PodName) + go c.claimChunk(d, owner) } } } diff --git a/drsm/drsm.go b/drsm/drsm.go index 9ac3522..1254e8b 100644 --- a/drsm/drsm.go +++ b/drsm/drsm.go @@ -56,6 +56,8 @@ type Drsm struct { resourceValidCb func(int32) bool mongo *MongoDBLibrary.MongoClient globalChunkTblMutex sync.Mutex + // podMapMutex guards podMap and every podData.podChunks reachable through it. + podMapMutex sync.Mutex } func (d *Drsm) DeletePod(podInstance string) { diff --git a/drsm/drsm_test.go b/drsm/drsm_test.go new file mode 100644 index 0000000..63aa480 --- /dev/null +++ b/drsm/drsm_test.go @@ -0,0 +1,195 @@ +// SPDX-FileCopyrightText: 2026 Forsway Scandinavia AB +// +// SPDX-License-Identifier: Apache-2.0 + +package drsm + +import ( + "fmt" + "sync" + "testing" +) + +const ( + testPod = "amf-test-pod" + otherPod = "amf-other-pod" + iterations = 200 +) + +// newTestDrsm builds a Drsm with only the in-memory state initialised. Every path exercised +// here stays inside those maps, so no MongoDB connection is needed. +func newTestDrsm() *Drsm { + return &Drsm{ + mode: ResourceClient, + clientId: PodId{PodName: testPod}, + localChunkTbl: make(map[int32]*chunk), + globalChunkTbl: make(map[int32]*chunk), + podMap: make(map[string]*podData), + scanChunks: make(map[int32]*chunk), + podDown: make(chan string, 10), + } +} + +func newTestChunk(id int32, freeIds int32) *chunk { + c := &chunk{Id: id, Owner: PodId{PodName: testPod}, AllocIds: make(map[int32]bool)} + for i := int32(0); i < freeIds; i++ { + c.FreeIds = append(c.FreeIds, i) + } + return c +} + +// TestScanTablesConcurrentAccess drives the scan goroutines started by claimChunk against +// the AllocateInt32ID/ReleaseInt32ID API paths. scanChunks and localChunkTbl are reachable +// from both, so every access has to hold mutex. +func TestScanTablesConcurrentAccess(t *testing.T) { + d := newTestDrsm() + + allocatable := newTestChunk(1, 64) + d.localChunkTbl[allocatable.Id] = allocatable + + scanning := newTestChunk(2, 0) + d.startScan(scanning) + + completing := newTestChunk(3, 0) + + var wg sync.WaitGroup + wg.Add(4) + + // Allocation ranges localChunkTbl; releasing the id back keeps FreeIds from running + // out, which would otherwise reach GetNewChunk and MongoDB. + go func() { + defer wg.Done() + for range iterations { + id, err := d.AllocateInt32ID() + if err != nil { + t.Errorf("AllocateInt32ID: %v", err) + return + } + if err := d.ReleaseInt32ID(id); err != nil { + t.Errorf("ReleaseInt32ID(%d): %v", id, err) + return + } + } + }() + + // Releasing an id of a chunk that is still being scanned misses localChunkTbl and then + // reads scanChunks. + go func() { + defer wg.Done() + for i := range iterations { + id := scanning.Id<<10 | int32(i%1024) + if err := d.ReleaseInt32ID(id); err != nil { + t.Errorf("ReleaseInt32ID(%d): %v", id, err) + return + } + } + }() + + // A chunk being published as scanning: writes scanChunks. + go func() { + defer wg.Done() + for range iterations { + d.startScan(scanning) + } + }() + + // A completed scan: writes localChunkTbl and deletes from scanChunks. + go func() { + defer wg.Done() + for range iterations { + d.completeScan(completing) + d.startScan(completing) + } + }() + + wg.Wait() + + if _, found := d.scanChunks[scanning.Id]; !found { + t.Errorf("chunk %d should still be in scanChunks", scanning.Id) + } + if _, found := d.localChunkTbl[completing.Id]; !found { + t.Errorf("chunk %d should have been moved to localChunkTbl", completing.Id) + } +} + +// TestPodMapConcurrentAccess drives the three goroutines that reach podMap: the change +// stream (addChunk, ensurePod, recordChunkOwner), the checkAllChunks ticker (addChunk) and +// the pod-down handler (podChunkIds). +func TestPodMapConcurrentAccess(t *testing.T) { + d := newTestDrsm() + d.ensurePod(&FullStream{PodId: testPod}) + + addChunks := func(first int32) { + for i := range int32(iterations) { + id := first + i + d.addChunk(&FullStream{Id: fmt.Sprintf("chunkid-%d", id), PodId: testPod}) + } + } + + var wg sync.WaitGroup + wg.Add(5) + + // The change stream and the periodic resync both call addChunk, on disjoint chunk ids. + go func() { + defer wg.Done() + addChunks(1) + }() + go func() { + defer wg.Done() + addChunks(iterations + 1) + }() + + // A chunk changing owner writes podChunks through recordChunkOwner. + go func() { + defer wg.Done() + for i := range int32(iterations) { + id := 2*iterations + 1 + i + if !d.recordChunkOwner(testPod, id, newTestChunk(id, 0)) { + t.Errorf("recordChunkOwner(%d): pod %s should be known", id, testPod) + return + } + } + }() + + // The pod-down handler ranges podChunks, and keepalives insert into podMap. + go func() { + defer wg.Done() + for range iterations { + d.podChunkIds(testPod) + d.podDownCandidate(testPod) + } + }() + go func() { + defer wg.Done() + for range iterations { + d.ensurePod(&FullStream{PodId: otherPod}) + } + }() + + wg.Wait() + + ids, owner := d.podChunkIds(testPod) + if len(ids) != 3*iterations { + t.Errorf("podChunks holds %d chunks, want %d", len(ids), 3*iterations) + } + if owner != testPod { + t.Errorf("owner is %q, want %q", owner, testPod) + } + if len(d.globalChunkTbl) != 2*iterations { + t.Errorf("globalChunkTbl holds %d chunks, want %d", len(d.globalChunkTbl), 2*iterations) + } +} + +// TestPodChunkIdsUnknownPod covers the pod-down path for a pod that is no longer in podMap. +// Dereferencing the missing entry used to panic. +func TestPodChunkIdsUnknownPod(t *testing.T) { + d := newTestDrsm() + + ids, owner := d.podChunkIds("pod-that-never-registered") + if ids != nil { + t.Errorf("ids is %v, want nil", ids) + } + if owner != "" { + t.Errorf("owner is %q, want empty", owner) + } +} diff --git a/drsm/scan.go b/drsm/scan.go index d86dc39..f29cfb7 100644 --- a/drsm/scan.go +++ b/drsm/scan.go @@ -19,8 +19,7 @@ func (c *chunk) scanChunk(d *Drsm) { logger.DrsmLog.Infoln("do not perform scan task if Chunk is not owned by us") return } - c.State = Scanning - d.scanChunks[c.Id] = c + d.startScan(c) var i int32 for i = 0; i < 1000; i++ { c.ScanIds = append(c.ScanIds, i) @@ -47,9 +46,7 @@ func (c *chunk) scanChunk(d *Drsm) { } } else { // mark as owned. and remove from scan list and add to local table - c.State = Owned - d.localChunkTbl[c.Id] = c - delete(d.scanChunks, c.Id) + d.completeScan(c) logger.DrsmLog.Debugf("scan complete for Chunk %v", c.Id) return } @@ -61,3 +58,23 @@ func (c *chunk) scanChunk(d *Drsm) { } } } + +// startScan publishes c as being scanned. scanChunks and localChunkTbl are shared with +// AllocateInt32ID and ReleaseInt32ID, which hold mutex, so the scan goroutines started by +// claimChunk have to hold it as well. +func (d *Drsm) startScan(c *chunk) { + mutex.Lock() + defer mutex.Unlock() + c.State = Scanning + d.scanChunks[c.Id] = c +} + +// completeScan moves c out of the scan table and into the local table once every id in the +// chunk has been scanned. +func (d *Drsm) completeScan(c *chunk) { + mutex.Lock() + defer mutex.Unlock() + c.State = Owned + d.localChunkTbl[c.Id] = c + delete(d.scanChunks, c.Id) +} diff --git a/drsm/updates.go b/drsm/updates.go index 44f3aa2..810b061 100644 --- a/drsm/updates.go +++ b/drsm/updates.go @@ -111,6 +111,8 @@ func (d *Drsm) handleDbUpdates() { } } +// ensurePodChunksInitialized allocates podD.podChunks on first use. Callers must hold +// podMapMutex. func (d *Drsm) ensurePodChunksInitialized(podD *podData) { if podD.podChunks == nil { podD.podChunks = make(map[int32]*chunk) @@ -148,12 +150,7 @@ func iterateChangeStream(d *Drsm, routineCtx context.Context, stream *mongo.Chan switch full.Type { case "keepalive": // logger.DrsmLog.Debugf("insert keepalive document") - pod, found := d.podMap[full.PodId] - if !found { - d.addPod(full) - } else { - logger.DrsmLog.Debugln("keepalive insert document: found existing podId", pod) - } + d.ensurePod(full) case "chunk": // logger.DrsmLog.Debugln("insert chunk document") d.addChunk(full) @@ -183,25 +180,18 @@ func iterateChangeStream(d *Drsm, routineCtx context.Context, stream *mongo.Chan cp.Owner.PodName = owner cp.Owner.PodIp = s.Update.UpdFields.PodIp cp.Owner.PodInstance = s.Update.UpdFields.PodInstance - podD, found := d.podMap[owner] - if !found { + if !d.recordChunkOwner(owner, c, cp) { 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) } case "delete": logger.DrsmLog.Debugln("delete operations") if !isChunkDoc(s.DId.Id) { // not chunk type doc. So its POD doc. // delete only gets document id - pod, found := d.podMap[s.DId.Id] - if pod != nil { - logger.DrsmLog.Infof("Stream(Delete): Pod %v and found %v. Chunks owned by crashed pod = %v", pod, found, pod.podChunks) + if d.podDownCandidate(s.DId.Id) { d.podDown <- s.DId.Id } } @@ -272,10 +262,6 @@ func (d *Drsm) checkAllChunks() { } func (d *Drsm) addChunk(full *FullStream) { - pod, found := d.podMap[full.PodId] - if !found { - pod = d.addPod(full) - } did := full.Id if did == "" { did = full.ChunkId @@ -286,16 +272,79 @@ func (d *Drsm) addChunk(full *FullStream) { c := &chunk{Id: cid, Owner: o} c.resourceValidCb = d.resourceValidCb + d.podMapMutex.Lock() + pod, found := d.podMap[full.PodId] + if !found { + pod = d.addPodLocked(full) + } pod.podChunks[cid] = c + logger.DrsmLog.Debugf("chunk id %v, podChunks %v", cid, pod.podChunks) + // Released before globalChunkTblMutex is taken: no path holds both locks at once. + d.podMapMutex.Unlock() d.globalChunkTblMutex.Lock() d.globalChunkTbl[cid] = c d.globalChunkTblMutex.Unlock() +} - logger.DrsmLog.Debugf("chunk id %v, podChunks %v", cid, pod.podChunks) +// ensurePod adds the pod described by full to podMap unless it is already known. +func (d *Drsm) ensurePod(full *FullStream) { + d.podMapMutex.Lock() + defer d.podMapMutex.Unlock() + if pod, found := d.podMap[full.PodId]; found { + logger.DrsmLog.Debugln("keepalive insert document: found existing podId", pod) + return + } + d.addPodLocked(full) +} + +// recordChunkOwner records cp against the pod that now owns it. It reports whether that pod +// is known locally. +func (d *Drsm) recordChunkOwner(owner string, chunkId int32, cp *chunk) bool { + d.podMapMutex.Lock() + defer d.podMapMutex.Unlock() + podD, found := d.podMap[owner] + if !found { + return false + } + // Defensive: should never happen if the pod went through addPodLocked, but prevents panic + d.ensurePodChunksInitialized(podD) + podD.podChunks[chunkId] = cp // add chunk to pod + logger.DrsmLog.Infof("stream(Update): pod to chunk map %v", podD.podChunks) + return true +} + +// podDownCandidate reports whether podName is still known locally, logging the chunks it +// owned. The caller signals podDown outside the lock: podDownDetected takes podMapMutex, +// so signalling while holding it would deadlock once the channel buffer is full. +func (d *Drsm) podDownCandidate(podName string) bool { + d.podMapMutex.Lock() + defer d.podMapMutex.Unlock() + pod, found := d.podMap[podName] + if found { + logger.DrsmLog.Infof("Stream(Delete): Pod %v and found %v. Chunks owned by crashed pod = %v", pod, found, pod.podChunks) + } + return found +} + +// podChunkIds returns the ids of the chunks currently recorded against podName, together +// with the owner name recorded for that pod, which claimChunk needs to build its filter. +func (d *Drsm) podChunkIds(podName string) ([]int32, string) { + d.podMapMutex.Lock() + defer d.podMapMutex.Unlock() + pd, found := d.podMap[podName] + if !found { + return nil, "" + } + ids := make([]int32, 0, len(pd.podChunks)) + for k := range pd.podChunks { + ids = append(ids, k) + } + return ids, pd.PodId.PodName } -func (d *Drsm) addPod(full *FullStream) *podData { +// addPodLocked adds the pod described by full to podMap. Callers must hold podMapMutex. +func (d *Drsm) addPodLocked(full *FullStream) *podData { podI := PodId{PodName: full.PodId, PodInstance: full.PodInstance, PodIp: full.PodIp} pod := &podData{PodId: podI} d.ensurePodChunksInitialized(pod)