Fix the races on chunk.FreeIds and chunk.ScanIds

drsm/chunk.go

@@ -66,6 +66,44 @@ func (d *Drsm) GetNewChunk() (*chunk, error) {
 	return c, nil
 }
 
+// appendScanIds seeds the ids still to be scanned. FreeIds, ScanIds and AllocIds are shared
+// with AllocateInt32ID and ReleaseInt32ID, which hold mutex, so the scan goroutine started by
+// claimChunk has to hold it too. The slice is built first so the lock covers only the append.
+func (c *chunk) appendScanIds(n int32) {
+	ids := make([]int32, 0, n)
+	for i := int32(0); i < n; i++ {
+		ids = append(ids, i)
+	}
+	mutex.Lock()
+	defer mutex.Unlock()
+	c.ScanIds = append(c.ScanIds, ids...)
+}
+
+// nextScanId takes the next id to scan, reporting false once every id has been scanned.
+func (c *chunk) nextScanId() (int32, bool) {
+	mutex.Lock()
+	defer mutex.Unlock()
+	if len(c.ScanIds) == 0 {
+		return 0, false
+	}
+	id := c.ScanIds[len(c.ScanIds)-1]
+	c.ScanIds = c.ScanIds[:len(c.ScanIds)-1]
+	return id, true
+}
+
+// recordScanResult files a scanned id as free or as already in use. The caller invokes
+// resourceValidCb outside the lock on purpose: it is supplied by the NF and may re-enter
+// drsm, which would deadlock on a non-reentrant mutex.
+func (c *chunk) recordScanResult(id int32, free bool) {
+	mutex.Lock()
+	defer mutex.Unlock()
+	if free {
+		c.FreeIds = append(c.FreeIds, id)
+	} else {
+		c.AllocIds[id] = true // Id is in use
+	}
+}
+
 func (c *chunk) AllocateIntID() (int32, error) {
 	if len(c.FreeIds) == 0 {
 		err := fmt.Errorf("freeIds in chunk 0")

drsm/scan.go

@@ -21,10 +21,7 @@ func (c *chunk) scanChunk(d *Drsm) {
 	}
 	c.State = Scanning
 	d.scanChunks[c.Id] = c
-	var i int32
-	for i = 0; i < 1000; i++ {
-		c.ScanIds = append(c.ScanIds, i)
-	}
+	c.appendScanIds(1000)
 
 	ticker := time.NewTicker(5000 * time.Millisecond)
 	defer ticker.Stop()
@@ -35,16 +32,10 @@ func (c *chunk) scanChunk(d *Drsm) {
 			// TODO : find candidate and then scan that Id.
 			// once all Ids are scanned then we can start using this block
 			if c.resourceValidCb != nil {
-				if len(c.ScanIds) != 0 {
-					id := c.ScanIds[len(c.ScanIds)-1]
-					c.ScanIds = c.ScanIds[:len(c.ScanIds)-1]
+				if id, ok := c.nextScanId(); ok {
 					rid := c.Id<<10 | id
 					res := c.resourceValidCb(rid)
-					if res {
-						c.FreeIds = append(c.FreeIds, id)
-					} else {
-						c.AllocIds[id] = true // Id is in use
-					}
+					c.recordScanResult(id, res)
 				} else {
 					// mark as owned. and remove from scan list and add to local table
 					c.State = Owned

drsm/scan_test.go

@@ -0,0 +1,109 @@
+// SPDX-FileCopyrightText: 2026 Forsway Scandinavia AB
+//
+// SPDX-License-Identifier: Apache-2.0
+
+package drsm
+
+import (
+	"sync"
+	"testing"
+)
+
+const (
+	scanSeed   = 1024
+	scanRounds = 2000
+)
+
+func newScanTestChunk(id int32) *chunk {
+	return &chunk{Id: id, State: Scanning, AllocIds: make(map[int32]bool)}
+}
+
+// TestScanFieldsConcurrentAccess drives the scan goroutine's handling of ScanIds, FreeIds and
+// AllocIds against the two API paths that touch the same fields. api.go calls AllocateIntID
+// and ReleaseIntID with mutex held, so the test does the same, which is what makes the scan
+// goroutine's accesses the unsynchronised side.
+//
+// Releasing an id that belongs to a chunk still being scanned is a supported path:
+// ReleaseInt32ID looks the chunk up in scanChunks precisely so that can happen, and
+// ReleaseIntID then walks ScanIds while the scan goroutine is popping from it.
+func TestScanFieldsConcurrentAccess(t *testing.T) {
+	c := newScanTestChunk(1)
+	c.appendScanIds(scanSeed)
+
+	var (
+		seenMu sync.Mutex
+		seen   = make(map[int32]bool)
+	)
+
+	var wg sync.WaitGroup
+	wg.Add(3)
+
+	// The scan goroutine: pop an id, consult the callback outside the lock, record the result.
+	go func() {
+		defer wg.Done()
+		for {
+			id, ok := c.nextScanId()
+			if !ok {
+				return
+			}
+			seenMu.Lock()
+			if seen[id] {
+				t.Errorf("id %d was handed out twice by nextScanId", id)
+				seenMu.Unlock()
+				return
+			}
+			seen[id] = true
+			seenMu.Unlock()
+
+			c.recordScanResult(id, id%3 != 0)
+		}
+	}()
+
+	// ReleaseInt32ID's inner call. With State == Scanning this also walks ScanIds.
+	go func() {
+		defer wg.Done()
+		for i := range int32(scanRounds) {
+			mutex.Lock()
+			c.ReleaseIntID(i % scanSeed)
+			mutex.Unlock()
+		}
+	}()
+
+	// AllocateInt32ID's inner call.
+	go func() {
+		defer wg.Done()
+		for range scanRounds {
+			mutex.Lock()
+			_, _ = c.AllocateIntID()
+			mutex.Unlock()
+		}
+	}()
+
+	wg.Wait()
+
+	// appendScanIds seeds distinct ids and both mutators only remove, so nothing may be left.
+	if _, ok := c.nextScanId(); ok {
+		t.Error("ScanIds should be drained once the scan goroutine has finished")
+	}
+}
+
+// TestAppendScanIdsSeedsDistinctIds pins the seeding behaviour the drain assertion relies on.
+func TestAppendScanIdsSeedsDistinctIds(t *testing.T) {
+	c := newScanTestChunk(2)
+	c.appendScanIds(8)
+
+	got := make(map[int32]bool)
+	for {
+		id, ok := c.nextScanId()
+		if !ok {
+			break
+		}
+		if got[id] {
+			t.Fatalf("id %d seeded more than once", id)
+		}
+		got[id] = true
+	}
+	if len(got) != 8 {
+		t.Errorf("seeded %d ids, want 8", len(got))
+	}
+}