test(server): replace distance lock timing threshold with a deterministic check

cmd/server/distance_lock_contention_test.go

@@ -1,131 +1,256 @@
 package main
 
 import (
+	"runtime"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
 )
 
-// TestComputeAnalyticsDistanceLockHoldDuration asserts that
-// computeAnalyticsDistance does NOT hold s.mu.RLock() for the entire
-// compute — otherwise readers serialize writers (which need s.mu.Lock for
-// ingest / buildDistanceIndex), turning a 3s analytics call into 15s under
-// heavy ingest (issue #1239).
+// TestComputeAnalyticsDistanceReleasesLockBeforeCompute asserts the #1239
+// invariant: computeAnalyticsDistance holds s.mu.RLock() only while it
+// snapshots distHops/distPaths (and builds the region match set), and has
+// released it before the post-snapshot work starts. Held across the compute,
+// the read lock serializes every ingest writer (s.mu.Lock) behind analytics
+// readers, which turned a 3s analytics call into 15s under heavy ingest.
 //
-// Methodology: run N reader goroutines calling computeAnalyticsDistance
-// continuously, while the test goroutine measures how long it takes to
-// complete W bare mu.Lock()/mu.Unlock() cycles. Each writer cycle must
-// wait for ALL currently-holding RLocks to release. Pre-fix, every reader
-// holds RLock for the entire compute (~ms), so each writer cycle waits
-// behind an active reader → avg cycle hundreds of microseconds to
-// milliseconds. Post-fix, readers hold RLock only long enough to grab
-// slice headers (microseconds), so writer cycles complete unimpeded.
-func TestComputeAnalyticsDistanceLockHoldDuration(t *testing.T) {
-	if testing.Short() {
-		t.Skip("skipping concurrency timing test in -short mode")
-	}
-
+// The test checks the lock state directly instead of timing writers. It
+// parks the compute at a known point after the snapshot, asks the mutex
+// whether a writer can get in, and then lets the compute finish while the
+// writer still holds the lock:
+//
+//  1. The test enters tx.decodedOnce.Do first and blocks inside it, so the
+//     compute's tx.ParsedDecoded() call has to wait. With a region and an
+//     area set, that call is in the area filter, the first step after
+//     s.mu.RUnlock().
+//  2. It waits until the compute goroutine's stack shows it parked there.
+//     That proves the compute has passed the snapshot and cannot move on.
+//  3. s.mu.TryLock() must then succeed. No other goroutine touches this
+//     store's mu (NewPacketStore starts none), so a failure can only mean
+//     the compute still holds the read lock.
+//  4. Still holding the write lock, the test points distHops/distPaths at
+//     fresh slices and releases the park. Any s.mu.RLock() the compute takes
+//     from here on blocks behind the test, so the compute either returns
+//     (pass) or shows up parked in RLock (fail). The result must come from
+//     the snapshot alone; under -race this also covers the post-snapshot
+//     reads.
+//
+// A regression that still holds the read lock at the park point fails at
+// step 3; one that takes it again later in the compute fails at step 4. Both
+// fail on any machine, core count or race mode, because nothing is measured.
+// The deadlines only turn a broken setup into a failure instead of a hang.
+//
+// Scope: only the region+area path is driven. The default (region="",
+// area="") path has no call the test can block. It shares the lock
+// acquisition with this path, so a change to that shared code is caught,
+// but a lock taken on the default or area-only path alone is not. Step 4
+// assigns fresh slices; it does not model ingest and eviction compacting
+// distHops/distPaths in place.
+//
+// This replaces a writer-latency threshold (150µs, then 5ms) that overlapped
+// both healthy CI runs and a deliberate regression on fast machines. The
+// measurement lives on as BenchmarkComputeAnalyticsDistanceWriterCycle.
+func TestComputeAnalyticsDistanceReleasesLockBeforeCompute(t *testing.T) {
 	db := setupTestDB(t)
 	defer db.Close()
 	store := NewPacketStore(db, nil)
 
-	// Populate distHops/distPaths with enough records that compute takes
-	// a measurable amount of time (~ms). With region="", compute never
-	// dereferences distHopRecord.tx, so dummy zero-value records suffice.
+	const (
+		region   = "SEAM"
+		area     = "SEAM-AREA"
+		observer = "obs-seam"
+	)
+	obs := []*StoreObs{{ObserverID: observer}}
+	// Both transmissions pass the region filter; only inArea passes the
+	// area filter, so the result also shows the area filter really ran.
+	inArea := &StoreTx{DecodedJSON: `{"public_key":"aa"}`, Observations: obs}
+	outOfArea := &StoreTx{DecodedJSON: `{"public_key":"zz"}`, Observations: obs}
+
+	hop := func(from string, dist float64, tx *StoreTx) distHopRecord {
+		return distHopRecord{
+			FromName: from, FromPk: from, ToName: "B", ToPk: "bb",
+			Dist: dist, Type: "R↔R", Hash: "h-" + from,
+			Timestamp: "2024-01-01T00:00:00Z", HourBucket: "2024-01-01-00",
+			tx: tx,
+		}
+	}
+	store.mu.Lock()
+	store.distHops = []distHopRecord{hop("aa", 10, inArea), hop("zz", 99, outOfArea)}
+	store.distPaths = []distPathRecord{{Hash: "p", TotalDist: 10, HopCount: 1, tx: inArea}}
+	store.mu.Unlock()
+
+	store.config = &Config{Areas: map[string]AreaEntry{area: {Label: "seam"}}}
+	store.regionObsMu.Lock()
+	store.regionObsCache = map[string]map[string]bool{region: {observer: true}}
+	store.regionObsCacheTime = time.Now()
+	store.regionObsMu.Unlock()
+	store.areaNodeMu.Lock()
+	store.areaNodeCache[area] = map[string]bool{"aa": true}
+	store.areaNodeCacheTimes[area] = time.Now()
+	store.areaNodeMu.Unlock()
+
+	// Step 1: occupy inArea's decode so the compute blocks on it.
+	held := make(chan struct{})
+	release := make(chan struct{})
+	var releaseOnce sync.Once
+	unblock := func() { releaseOnce.Do(func() { close(release) }) }
+	defer unblock()
+	go inArea.decodedOnce.Do(func() {
+		inArea.parsedDecoded = map[string]interface{}{"public_key": "aa"}
+		close(held)
+		<-release
+	})
+	<-held
+
+	done := make(chan map[string]interface{}, 1)
+	go func() { done <- store.computeAnalyticsDistance(region, area) }()
+
+	// Step 2: the compute is past the snapshot and parked in the area filter.
+	distanceWaitFor(t, "computeAnalyticsDistance to park in tx.ParsedDecoded", func() bool {
+		return distanceComputeWaiting(".(*StoreTx).ParsedDecoded(", "sync.(*Once).doSlow(")
+	})
+
+	// Step 3: the invariant.
+	if !store.mu.TryLock() {
+		unblock()
+		<-done
+		t.Fatal("s.mu.TryLock() failed while computeAnalyticsDistance was parked after its snapshot: " +
+			"the main RLock is still held across the compute, so ingest writers queue behind analytics readers (issue #1239)")
+	}
+	// Step 4: new data while the compute is mid-flight, then let it finish
+	// with the write lock still held.
+	store.distHops = []distHopRecord{hop("xx", 500, inArea)}
+	store.distPaths = nil
+	unblock()
+	var r map[string]interface{}
+	distanceWaitFor(t, "computeAnalyticsDistance to return or block in s.mu.RLock() "+
+		"(a timeout here means it is stuck on s.mu some other way, e.g. Lock())", func() bool {
+		select {
+		case r = <-done:
+			return true
+		default:
+			return distanceComputeWaiting("[sync.RWMutex.RLock")
+		}
+	})
+	store.mu.Unlock()
+	if r == nil {
+		<-done
+		t.Fatal("computeAnalyticsDistance blocked in s.mu.RLock() after its snapshot: " +
+			"it takes the main lock again for the compute, so ingest writers queue behind analytics readers (issue #1239)")
+	}
+
+	topHops, _ := r["topHops"].([]map[string]interface{})
+	if len(topHops) != 1 || topHops[0]["fromPk"] != "aa" {
+		t.Fatalf("topHops = %v, want exactly the in-area snapshot hop (fromPk aa)", topHops)
+	}
+	topPaths, _ := r["topPaths"].([]map[string]interface{})
+	if len(topPaths) != 1 {
+		t.Fatalf("topPaths = %v, want the one snapshot path", topPaths)
+	}
+}
+
+// distanceComputeWaiting reports whether a goroutine started by
+// TestComputeAnalyticsDistanceReleasesLockBeforeCompute and inside
+// computeAnalyticsDistance has every marker in its runtime.Stack dump. The
+// markers are frames (sync.(*Once).doSlow under tx.ParsedDecoded) or the
+// wait reason in the goroutine header ([sync.RWMutex.RLock]). Requiring the
+// creator keeps a compute leaked by another test from matching.
+//
+// Maintenance: the match depends on the runtime.Stack text format, those
+// runtime frame names and wait reasons, and the method names used here.
+// Re-verify it when changing the Go version or renaming those methods.
+func distanceComputeWaiting(markers ...string) bool {
+	buf := make([]byte, 1<<16)
+	for {
+		n := runtime.Stack(buf, true)
+		if n < len(buf) {
+			buf = buf[:n]
+			break
+		}
+		buf = make([]byte, 2*len(buf))
+	}
+	for _, g := range strings.Split(string(buf), "\n\n") {
+		if !strings.Contains(g, ".(*PacketStore).computeAnalyticsDistance(") ||
+			!strings.Contains(g, ".TestComputeAnalyticsDistanceReleasesLockBeforeCompute in goroutine ") {
+			continue
+		}
+		all := true
+		for _, m := range markers {
+			all = all && strings.Contains(g, m)
+		}
+		if all {
+			return true
+		}
+	}
+	return false
+}
+
+// distanceWaitFor yields until cond holds. The deadline only turns a broken
+// test into a failure instead of a hang; it is not part of the coordination.
+func distanceWaitFor(t *testing.T, what string, cond func() bool) {
+	t.Helper()
+	deadline := time.Now().Add(10 * time.Second)
+	for !cond() {
+		if time.Now().After(deadline) {
+			t.Fatalf("timed out waiting for %s", what)
+		}
+		runtime.Gosched()
+	}
+}
+
+// BenchmarkComputeAnalyticsDistanceWriterCycle times one bare
+// s.mu.Lock/Unlock (ns/op) while 8 goroutines run computeAnalyticsDistance
+// on 20k hops, the setup the lock test above used to time and gate on. The
+// number depends on core count, scheduler and -race, so it is only for
+// comparing builds on one machine and is never asserted.
+func BenchmarkComputeAnalyticsDistanceWriterCycle(b *testing.B) {
+	store := NewPacketStore(nil, nil) // the default path never touches the DB
+
 	const N = 20000
 	hops := make([]distHopRecord, N)
-	for i := 0; i < N; i++ {
+	for i := range hops {
 		hops[i] = distHopRecord{
-			FromName:   "A",
-			FromPk:     "aa",
-			ToName:     "B",
-			ToPk:       "bb",
-			Dist:       float64(i%500) + 0.5,
-			Type:       []string{"R↔R", "C↔R", "C↔C"}[i%3],
-			Hash:       "h",
-			Timestamp:  "2024-01-01T00:00:00Z",
-			HourBucket: "2024-01-01-00",
+			FromName: "A", FromPk: "aa", ToName: "B", ToPk: "bb",
+			Dist: float64(i%500) + 0.5, Type: []string{"R↔R", "C↔R", "C↔C"}[i%3],
+			Hash: "h", Timestamp: "2024-01-01T00:00:00Z", HourBucket: "2024-01-01-00",
 		}
 	}
 	paths := make([]distPathRecord, 200)
 	for i := range paths {
-		paths[i] = distPathRecord{
-			Hash:      "p",
-			TotalDist: float64(i),
-			HopCount:  3,
-			Timestamp: "2024-01-01T00:00:00Z",
-			Hops: []distHopDetail{
-				{FromName: "A", FromPk: "aa", ToName: "B", ToPk: "bb", Dist: 1},
-			},
-		}
+		paths[i] = distPathRecord{Hash: "p", TotalDist: float64(i), HopCount: 3, Timestamp: "2024-01-01T00:00:00Z"}
 	}
 	store.mu.Lock()
 	store.distHops = hops
 	store.distPaths = paths
 	store.mu.Unlock()
 
-	// Sanity: result is non-empty.
-	r := store.computeAnalyticsDistance("", "")
-	if r == nil {
-		t.Fatal("expected non-nil result")
-	}
-	if _, ok := r["topHops"]; !ok {
-		t.Fatal("expected topHops in result")
-	}
-
-	// Background readers churn computeAnalyticsDistance.
 	const Readers = 8
 	var stop atomic.Bool
-	var readerErrs atomic.Int64
-	var wg sync.WaitGroup
+	var wg, ready sync.WaitGroup
 	wg.Add(Readers)
+	ready.Add(Readers)
 	for i := 0; i < Readers; i++ {
 		go func() {
 			defer wg.Done()
+			store.computeAnalyticsDistance("", "")
+			ready.Done()
 			for !stop.Load() {
-				rr := store.computeAnalyticsDistance("", "")
-				if rr == nil {
-					readerErrs.Add(1)
-				}
-				if _, ok := rr["topHops"]; !ok {
-					readerErrs.Add(1)
-				}
+				store.computeAnalyticsDistance("", "")
 			}
 		}()
 	}
+	ready.Wait() // time only while every reader is in its compute loop
+	defer func() {
+		stop.Store(true)
+		wg.Wait()
+	}()
 
-	// Let readers ramp up.
-	time.Sleep(50 * time.Millisecond)
-
-	// Measure writer (mu.Lock/Unlock) throughput.
-	const WriterCycles = 200
-	start := time.Now()
-	for i := 0; i < WriterCycles; i++ {
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
 		store.mu.Lock()
 		store.mu.Unlock()
 	}
-	elapsed := time.Since(start)
-
-	stop.Store(true)
-	wg.Wait()
-
-	if readerErrs.Load() > 0 {
-		t.Fatalf("readers returned empty/invalid results: %d", readerErrs.Load())
-	}
-
-	avgMicros := elapsed.Microseconds() / int64(WriterCycles)
-	t.Logf("avg writer Lock/Unlock cycle: %dµs over %d cycles (total %v) with %d concurrent readers, %d hops, %d paths",
-		avgMicros, WriterCycles, elapsed, Readers, N, len(paths))
-
-	// If readers hold the main RLock for their entire compute, every
-	// writer Lock cycle waits for an active reader to release: avg cycle
-	// >> 100µs at this data scale. After the refactor, readers hold the
-	// main RLock only long enough to snapshot slice headers (<1µs), so
-	// writer cycles complete in tens of microseconds.
-	const MaxAvgMicros = 150
-	if avgMicros > MaxAvgMicros {
-		t.Fatalf("avg writer Lock/Unlock cycle %dµs exceeds %dµs threshold — computeAnalyticsDistance is holding the main RLock for too long and blocking writers (issue #1239)",
-			avgMicros, MaxAvgMicros)
-	}
+	b.StopTimer()
 }