Compose each machine once per pass, so take-ins of done builds stop colliding and every new build gets its composition and credential (issue 1945) #460
@@ -114,28 +114,55 @@ func inList(list []string, s string) bool {
|
||||
return false
|
||||
}
|
||||
|
||||
// doneKeyTakeOnce is one take-in; a variable so a test may stand in for it. doneKeysSay says a line of the take-ins.
|
||||
var (
|
||||
doneKeyTakeOnce = takeDoneKeyOnce
|
||||
doneKeysSay = func(s string) { fmt.Println(s) }
|
||||
)
|
||||
|
||||
// doneKeysMu keeps one take-in of a module's done key at a time in this process, per module.
|
||||
var doneKeysMu sync.Map
|
||||
|
||||
// takeDoneKey is a woken module's take-in on the live stores and buckets, said. A take-in that fails (its compositions
|
||||
// or its grant not written) is tried again every doneKeyRetry while it fails, loudly: printed each time, and an urgent
|
||||
// condition raised by the first until it has worked, as the chain's own moves are (layBuildsOutSoon). A key not done
|
||||
// yet is no failure: its next write wakes this again.
|
||||
// yet is no failure: it is said once for each reason it waits for and looked at again every doneKeyRetry.
|
||||
func takeDoneKey(ctx context.Context, open *stores, module string) {
|
||||
mu, _ := doneKeysMu.LoadOrStore(module, &sync.Mutex{})
|
||||
mu.(*sync.Mutex).Lock()
|
||||
if !mu.(*sync.Mutex).TryLock() {
|
||||
return // a take-in of the module already follows it, and looks again within doneKeyRetry
|
||||
}
|
||||
defer mu.(*sync.Mutex).Unlock()
|
||||
failed := false
|
||||
saidWait := ""
|
||||
for {
|
||||
err := takeDoneKeyOnce(ctx, open, module)
|
||||
err := doneKeyTakeOnce(ctx, open, module)
|
||||
o := doneKeyObservation(module)
|
||||
if err == nil || errors.Is(err, errKeyWaits) {
|
||||
if err == nil {
|
||||
if keeper := conditionsFrom; keeper != nil && failed {
|
||||
_, _ = keeper.Clear(ctx, o.Key(), "composed and granted")
|
||||
}
|
||||
return
|
||||
}
|
||||
fmt.Printf("done keys: %s: %v; tried again in %s\n", module, err, doneKeyRetry)
|
||||
if errors.Is(err, errKeyWaits) {
|
||||
// A take-in that waits is said once for each reason, and tried again every doneKeyRetry: the key write that
|
||||
// would wake it may never come (issue 1945, live on 2026-10-25: mesh-host's take-in waited unsaid).
|
||||
if why := err.Error(); why != saidWait {
|
||||
doneKeysSay(fmt.Sprintf("done keys: %s waits: %s; looked at again every %s", module,
|
||||
strings.TrimPrefix(why, errKeyWaits.Error()+": "), doneKeyRetry))
|
||||
saidWait = why
|
||||
}
|
||||
} else {
|
||||
doneKeysSay(fmt.Sprintf("done keys: %s: %v; tried again in %s", module, err, doneKeyRetry))
|
||||
}
|
||||
if errors.Is(err, errKeyWaits) {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-time.After(doneKeyRetry):
|
||||
}
|
||||
continue
|
||||
}
|
||||
if keeper := conditionsFrom; keeper != nil && !failed {
|
||||
o.Summary += err.Error()
|
||||
if _, oerr := keeper.Observe(ctx, o); oerr == nil {
|
||||
@@ -295,7 +322,8 @@ func takeDoneKeyAtStart(ctx context.Context, open *stores, module string) {
|
||||
mu.(*sync.Mutex).Lock()
|
||||
err := takeDoneKeyOnce(ctx, open, module)
|
||||
mu.(*sync.Mutex).Unlock()
|
||||
if err != nil && !errors.Is(err, errKeyWaits) {
|
||||
if err != nil {
|
||||
// A failure or a wait is followed by takeDoneKey, which says it and looks again (issue 1945).
|
||||
inBackground(func() { takeDoneKey(ctx, open, module) })
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,86 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// A take-in that waits is said once for each reason it waits for, and tried again every doneKeyRetry until it is taken
|
||||
// in: never left for a key write that may not come (issue 1945, live on 2026-10-25: mesh-host's take-in at 26bd waited,
|
||||
// said nothing, and nothing woke it until another write).
|
||||
func TestAWaitingTakeInIsSaidOnceAndTriedAgain(t *testing.T) {
|
||||
was, wasRetry, wasSay := doneKeyTakeOnce, doneKeyRetry, doneKeysSay
|
||||
t.Cleanup(func() { doneKeyTakeOnce, doneKeyRetry, doneKeysSay = was, wasRetry, wasSay })
|
||||
doneKeyRetry = 5 * time.Millisecond
|
||||
var mu sync.Mutex
|
||||
var said []string
|
||||
doneKeysSay = func(s string) { mu.Lock(); said = append(said, s); mu.Unlock() }
|
||||
calls := 0
|
||||
doneKeyTakeOnce = func(context.Context, *stores, string) error {
|
||||
calls++
|
||||
switch {
|
||||
case calls <= 2:
|
||||
return fmt.Errorf("%w: the artifact store could not hold mesh-host's artifact", errKeyWaits)
|
||||
case calls == 3:
|
||||
return fmt.Errorf("%w: mesh-host's key cannot be read", errKeyWaits)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
done := make(chan struct{})
|
||||
go func() { takeDoneKey(context.Background(), nil, "mesh-host"); close(done) }()
|
||||
select {
|
||||
case <-done:
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("a waiting take-in was not tried again until it was taken in")
|
||||
}
|
||||
if calls != 4 {
|
||||
t.Fatalf("tried %d times, want 4", calls)
|
||||
}
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
waits := 0
|
||||
for _, s := range said {
|
||||
if strings.Contains(s, "mesh-host waits") {
|
||||
waits++
|
||||
}
|
||||
}
|
||||
if waits != 2 {
|
||||
t.Fatalf("said %q: want each of the two reasons once", said)
|
||||
}
|
||||
}
|
||||
|
||||
// A wake while a take-in of the module already follows it adds no second one: the follower looks again itself.
|
||||
func TestAWakeWhileATakeInFollowsAddsNone(t *testing.T) {
|
||||
was, wasRetry, wasSay := doneKeyTakeOnce, doneKeyRetry, doneKeysSay
|
||||
t.Cleanup(func() { doneKeyTakeOnce, doneKeyRetry, doneKeysSay = was, wasRetry, wasSay })
|
||||
doneKeyRetry = 20 * time.Millisecond
|
||||
doneKeysSay = func(string) {}
|
||||
var mu sync.Mutex
|
||||
calls := 0
|
||||
release := make(chan struct{})
|
||||
doneKeyTakeOnce = func(context.Context, *stores, string) error {
|
||||
mu.Lock()
|
||||
calls++
|
||||
mu.Unlock()
|
||||
select {
|
||||
case <-release:
|
||||
return nil
|
||||
default:
|
||||
return fmt.Errorf("%w: the key says building", errKeyWaits)
|
||||
}
|
||||
}
|
||||
go takeDoneKey(context.Background(), nil, "board")
|
||||
time.Sleep(5 * time.Millisecond)
|
||||
returned := make(chan struct{})
|
||||
go func() { takeDoneKey(context.Background(), nil, "board"); close(returned) }()
|
||||
select {
|
||||
case <-returned:
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("a second wake waited for the follower")
|
||||
}
|
||||
close(release)
|
||||
}
|
||||
Reference in New Issue
Block a user