CI-only: fix/go-module-caller-save-race at 77be9e71 (do not merge)

internal/indexer/go_module_source_calls.go

@@ -38,16 +38,33 @@ func cloneGoModuleSourceEdges(edges []*graph.Edge) []*graph.Edge {
 }
 
 // A successful replacement can change bytes without changing the stat.
-// Keep source facts, but make recovery validate them with the digest.
+// Keep source facts, but make recovery validate them with the digest. The
+// epoch also tells a recovery that read before this save, even of a file not
+// cached yet, not to trust its stat when it stores its result.
 func (idx *Indexer) invalidateGoModuleSourceStamp(rel string) {
 	idx.goModuleSourceMu.Lock()
 	defer idx.goModuleSourceMu.Unlock()
+	idx.goModuleSourceEpoch++
 	if cached, ok := idx.goModuleSourceCache[rel]; ok {
 		cached.size, cached.modified = 0, 0
 		idx.goModuleSourceCache[rel] = cached
 	}
 }
 
+// storeGoModuleSource keeps the facts but drops stat trust when a save was
+// invalidated after epoch was read, since the bytes may predate that save.
+func (idx *Indexer) storeGoModuleSource(rel string, entry goModuleSourceCacheEntry, epoch uint64) {
+	idx.goModuleSourceMu.Lock()
+	defer idx.goModuleSourceMu.Unlock()
+	if idx.goModuleSourceEpoch != epoch {
+		entry.size, entry.modified = 0, 0
+	}
+	if idx.goModuleSourceCache == nil {
+		idx.goModuleSourceCache = make(map[string]goModuleSourceCacheEntry)
+	}
+	idx.goModuleSourceCache[rel] = entry
+}
+
 func (mi *MultiIndexer) finishGoModuleCalls(cr *resolver.CrossRepoResolver) {
 	for _, file := range cr.TakeGoModuleSourceFiles() {
 		mi.restoreGoModuleSourceFile(cr, file)
@@ -70,6 +87,7 @@ func (mi *MultiIndexer) restoreGoModuleSourceFile(cr *resolver.CrossRepoResolver
 	}
 	idx.goModuleSourceMu.Lock()
 	cached, hit := idx.goModuleSourceCache[rel]
+	epoch := idx.goModuleSourceEpoch
 	idx.goModuleSourceMu.Unlock()
 	if hit && cached.size == info.Size() && cached.modified == info.ModTime().UnixNano() {
 		cr.RestoreGoModuleSourceCalls(cloneGoModuleSourceEdges(cached.edges))
@@ -97,9 +115,7 @@ func (mi *MultiIndexer) restoreGoModuleSourceFile(cr *resolver.CrossRepoResolver
 	digest := sha256.Sum256(src)
 	if hit && cached.digest == digest {
 		cached.size, cached.modified = info.Size(), info.ModTime().UnixNano()
-		idx.goModuleSourceMu.Lock()
-		idx.goModuleSourceCache[rel] = cached
-		idx.goModuleSourceMu.Unlock()
+		idx.storeGoModuleSource(rel, cached, epoch)
 		cr.RestoreGoModuleSourceCalls(cloneGoModuleSourceEdges(cached.edges))
 		return
 	}
@@ -120,12 +136,7 @@ func (mi *MultiIndexer) restoreGoModuleSourceFile(cr *resolver.CrossRepoResolver
 	}
 	if err == nil && !skipped {
 		idx.applyRepoPrefix(result.Nodes, result.Edges)
-		idx.goModuleSourceMu.Lock()
-		if idx.goModuleSourceCache == nil {
-			idx.goModuleSourceCache = make(map[string]goModuleSourceCacheEntry)
-		}
-		idx.goModuleSourceCache[rel] = goModuleSourceCacheEntry{digest: digest, edges: cloneGoModuleSourceEdges(result.Edges), size: info.Size(), modified: info.ModTime().UnixNano()}
-		idx.goModuleSourceMu.Unlock()
+		idx.storeGoModuleSource(rel, goModuleSourceCacheEntry{digest: digest, edges: cloneGoModuleSourceEdges(result.Edges), size: info.Size(), modified: info.ModTime().UnixNano()}, epoch)
 		cr.RestoreGoModuleSourceCalls(result.Edges)
 	}
 	result.ReleaseTree()

internal/indexer/go_module_source_store_test.go

@@ -0,0 +1,61 @@
+package indexer
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// Given a caller recovery that reads the file while a save of that file is
+// invalidated, when the recovery stores its result, then the stat is not
+// trusted, the next recovery reads again, and the one after reuses the facts.
+// "cached" stores through the equal-digest path; "first" stores through the
+// extract path for a file that had no cache entry when the save landed.
+func TestGoModuleSourceStoreDropsTrustAfterConcurrentSave(t *testing.T) {
+	for _, mode := range []string{"cached", "first"} {
+		t.Run(mode, func(t *testing.T) {
+			mi, _, repos := goModuleLifecycleFixture(t, "memory")
+			main := filepath.Join(repos[0].Path, "main.go")
+			writeFile(t, filepath.Join(repos[1].Path, "lib.go"), "package library\nfunc Other(){}\n")
+			for _, entry := range repos[:2] {
+				_, err := mi.TrackRepoCtx(t.Context(), entry)
+				require.NoError(t, err)
+			}
+			mi.runCrossRepoResolveFiles([]string{"b/lib.go"})
+			idx := mi.GetIndexer("a")
+			require.Contains(t, idx.goModuleSourceCache, "main.go")
+			if mode == "cached" {
+				// A new stat with equal bytes makes the recovery verify by reading.
+				touched := time.Now().Add(-time.Hour)
+				require.NoError(t, os.Chtimes(main, touched, touched))
+			} else {
+				delete(idx.goModuleSourceCache, "main.go")
+			}
+			original := readGoModuleCallerFile
+			reads := 0
+			readGoModuleCallerFile = func(path string) ([]byte, error) {
+				reads++
+				src, err := original(path)
+				if reads == 1 {
+					idx.invalidateGoModuleSourceStamp("main.go") // a save lands during the read
+				}
+				return src, err
+			}
+			t.Cleanup(func() { readGoModuleCallerFile = original })
+			cr := mi.newCrossRepoResolver()
+			mi.restoreGoModuleSourceFile(cr, "a/main.go")
+			require.Equal(t, 1, reads)
+			stored := idx.goModuleSourceCache["main.go"]
+			assert.Zero(t, stored.size, "a result read before a known save must not be trusted by stat")
+			assert.Zero(t, stored.modified, "a result read before a known save must not be trusted by stat")
+			mi.restoreGoModuleSourceFile(cr, "a/main.go")
+			assert.Equal(t, 2, reads, "the next recovery verifies content again")
+			mi.restoreGoModuleSourceFile(cr, "a/main.go")
+			assert.Equal(t, 2, reads, "an undisturbed verification restores stat trust")
+		})
+	}
+}

internal/indexer/indexer.go

@@ -506,6 +506,7 @@ type Indexer struct {
 
 	goModuleSourceMu    sync.Mutex
 	goModuleSourceCache map[string]goModuleSourceCacheEntry
+	goModuleSourceEpoch uint64 // bumped by each known save; guarded by goModuleSourceMu
 
 	// deferResolve, when set, makes IndexCtx skip the cross-cutting passes
 	// (per-repo ResolveAll / semantic enrichment / contract extraction +