Give each caller of a pass its own nodes' errors and write the grants whenever asked; a wake during a take-in makes it look once more (issue 1945)
mesh/delivery closed: a newer head of the same pull request
mesh/check pass 8114cee2af47 on 0fff3efd4e0b | unit 959 run, 172 kept, 2 skipped | integration 1615 run, 236 kept, 7 skipped | over bound: integration…

This commit is contained in:
2026-10-25 12:44:28 +01:00
parent baa8f58473
commit 8114cee2af
5 changed files with 175 additions and 33 deletions
+15 -4
View File
@@ -120,6 +120,9 @@ var (
doneKeysSay = func(s string) { fmt.Println(s) }
)
// doneKeysLookAgain marks a module woken while a take-in of it ran: that take-in goes round once more.
var doneKeysLookAgain sync.Map
// doneKeysMu keeps one take-in of a module's done key at a time in this process, per module.
var doneKeysMu sync.Map
@@ -130,17 +133,25 @@ var doneKeysMu sync.Map
func takeDoneKey(ctx context.Context, open *stores, module string) {
mu, _ := doneKeysMu.LoadOrStore(module, &sync.Mutex{})
if !mu.(*sync.Mutex).TryLock() {
return // a take-in of the module already follows it, and looks again within doneKeyRetry
// A take-in of the module already follows it: it is told to look again once it is done, so a newer key written
// meanwhile is never lost (the review of issue 1945).
doneKeysLookAgain.Store(module, true)
return
}
defer mu.(*sync.Mutex).Unlock()
failed := false
saidWait := ""
for {
doneKeysLookAgain.Delete(module) // what this look reads covers every wake before it
err := doneKeyTakeOnce(ctx, open, module)
o := doneKeyObservation(module)
if err == nil {
if keeper := conditionsFrom; keeper != nil && failed {
_, _ = keeper.Clear(ctx, o.Key(), "composed and granted")
failed = false
}
if _, again := doneKeysLookAgain.LoadAndDelete(module); again {
continue // a wake came while this one ran: once more, for the key written meanwhile
}
return
}
@@ -357,9 +368,9 @@ func composerFor(open *stores) *composer {
if c, ok := doneKeysComposers.Load(open); ok {
return c.(*composer)
}
c := newComposer(func(ctx context.Context, nodes []string) error {
_, err := composeModules(ctx, open, nodes)
return err
c := newComposer(func(ctx context.Context, nodes []string) map[string]error {
_, failed := composeModulesByNode(ctx, open, nodes)
return failed
}, func(ctx context.Context) error {
bus, err := aBus()
if err != nil {