diff --git a/cmd/mesh-controller/build_source.go b/cmd/mesh-controller/build_source.go index 85f105c6..1339df15 100644 --- a/cmd/mesh-controller/build_source.go +++ b/cmd/mesh-controller/build_source.go @@ -9,6 +9,7 @@ import ( "regexp" "slices" "strings" + "sync" "time" "git.novox.be/novox/mesh-sdk/go/catalogue" @@ -107,7 +108,10 @@ func sourceFormsRead(ctx context.Context, now func() time.Time, f := &sourceForms{bases: map[string]string{}} var failedAt time.Time // when the last read failed; zero when it did not var failure error + var mu sync.Mutex // the forms are shared by take-ins side by side (sharedSourceForms): one read at a time f.read = func(ctx context.Context, seat string, again bool) (string, error) { + mu.Lock() + defer mu.Unlock() if b, known := f.bases[seat]; known { return b, nil } diff --git a/cmd/mesh-controller/key_takein.go b/cmd/mesh-controller/key_takein.go index 3f00bbc2..ddfac56e 100644 --- a/cmd/mesh-controller/key_takein.go +++ b/cmd/mesh-controller/key_takein.go @@ -13,6 +13,7 @@ import ( "path" "sort" "strings" + "sync" "time" "git.novox.be/novox/mesh-sdk/go/buildstep" @@ -122,7 +123,7 @@ var predefinedAtTrunk = func(ctx context.Context, inv *inventory.Inventory) (pre } // formsOfSources reads how the sources are spelled on this mesh; a variable so a test names the forge's address. -var formsOfSources = newSourceForms +var formsOfSources = sharedSourceForms // errNoCatalogueRepository says no module of the mesh is built from the catalogue's repository on the forge: not a wait, // since no read of the forge makes one appear (issue 1902); the plan fails with these words. @@ -632,3 +633,40 @@ func handCoversEvery(m catalogue.Manifest, k builderkeys.Key, hand map[string]st } return nil } + +// sharedSourceForms is the one reading of how the sources are spelled (the forge's address above all) that every take-in +// of the inventory shares, made again once it is older than sourceFormsKept, so a forge that moved is read within that +// bound. A fresh reading per take-in read the whole mesh once each, about 120 at once at 16:08 on 2026-10-25, each with +// pools of its own (issues 1945 and 1277). A base that could not be read is not kept: the reading reads it again. +func sharedSourceForms(ctx context.Context, inv *inventory.Inventory) *sourceForms { + sharedFormsMu.Lock() + defer sharedFormsMu.Unlock() + now := sourceFormsNow() + if kept, ok := sharedForms.Load(inv); ok { + if k := kept.(keptForms); now.Sub(k.at) < sourceFormsKept { + return k.forms + } + } + // Read with no caller's deadline: the reading outlives the take-in that made it. + forms := sourceFormsFor(context.WithoutCancel(ctx), inv) + sharedForms.Store(inv, keptForms{forms: forms, at: now}) + return forms +} + +// keptForms is a shared reading of the sources and when it was made. +type keptForms struct { + forms *sourceForms + at time.Time +} + +// sourceFormsFor makes a fresh reading of the sources; a variable so a test counts them. +var sourceFormsFor = newSourceForms + +var ( + sharedForms sync.Map // *inventory.Inventory -> keptForms + sharedFormsMu sync.Mutex + sourceFormsNow = time.Now +) + +// sourceFormsKept is how long the take-ins share one reading of the sources before it is made again. +const sourceFormsKept = time.Minute diff --git a/cmd/mesh-controller/shared_forms_test.go b/cmd/mesh-controller/shared_forms_test.go new file mode 100644 index 00000000..5afb5558 --- /dev/null +++ b/cmd/mesh-controller/shared_forms_test.go @@ -0,0 +1,52 @@ +package main + +import ( + "context" + "sync" + "sync/atomic" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + "github.com/novox/mesh-controller/internal/inventory" +) + +// The take-ins share one reading of how the sources are spelled (the forge's address): a burst of 120 take-ins reads the +// mesh once, not once each (issues 1945 and 1277: each take-in's fresh read opened pools of its own at 16:08 on +// 2026-10-25). A reading older than sourceFormsKept is made again, so a moved forge is read. +func TestTakeInsShareOneReadingOfTheSources(t *testing.T) { + was, wasNow := sourceFormsFor, sourceFormsNow + t.Cleanup(func() { sourceFormsFor, sourceFormsNow = was, wasNow; sharedForms = sync.Map{} }) + var made, worldReads atomic.Int64 + sourceFormsFor = func(ctx context.Context, _ *inventory.Inventory) *sourceForms { + made.Add(1) + return sourceFormsRead(ctx, time.Now, func(context.Context) (catalogue.World, error) { + worldReads.Add(1) + return catalogue.World{}, nil + }) + } + clock := time.Date(2026, 10, 25, 16, 8, 0, 0, time.UTC) + var clockMu sync.Mutex + sourceFormsNow = func() time.Time { clockMu.Lock(); defer clockMu.Unlock(); return clock } + inv := &inventory.Inventory{} + var wg sync.WaitGroup + for range 120 { + wg.Add(1) + go func() { + defer wg.Done() + f := formsOfSources(context.Background(), inv) + _, _ = f.read(context.Background(), gitSeat, true) + }() + } + wg.Wait() + if n := made.Load(); n != 1 { + t.Fatalf("made %d readings of the sources for 120 take-ins, want one", n) + } + clockMu.Lock() + clock = clock.Add(sourceFormsKept + time.Second) + clockMu.Unlock() + formsOfSources(context.Background(), inv) + if n := made.Load(); n != 2 { + t.Fatalf("a reading older than %s was kept (%d made)", sourceFormsKept, n) + } +}