Take again a replacement of given values asked while one ran on the node (issue 1945)
mesh/delivery merged: source.main moved to novox/mesh-controller@af2b0844a1fc
mesh/check pass 3b4cb69e84dd on af2b0844a1fc | unit 963 run, 171 kept, 2 skipped | integration 1622 run, 232 kept, 7 skipped | over bound: integration…
mesh/delivery merged: source.main moved to novox/mesh-controller@af2b0844a1fc
mesh/check pass 3b4cb69e84dd on af2b0844a1fc | unit 963 run, 171 kept, 2 skipped | integration 1622 run, 232 kept, 7 skipped | over bound: integration…
This commit is contained in:
@@ -53,12 +53,61 @@ func startedOn(h sdkdelivery.Have, written time.Time) time.Time {
|
||||
|
||||
const givenWhy = "a value given by hand lives only until its module's first good start under the mesh (novox/hq ADR 0228)"
|
||||
|
||||
// replaceGiven replaces what a module's good start makes due, composes the machines, and says so.
|
||||
// replaceGiven replaces what a module's good start makes due, composes the machines, and says so. One at a time for a
|
||||
// node: a call that finds one running leaves what it was asked for as a mark, and the one running takes it once it is
|
||||
// done, so a start heard meanwhile is never lost (issue 1945, the kind of the dropped wake).
|
||||
func replaceGiven(ctx context.Context, open *stores, node string, started map[string]time.Time) {
|
||||
if _, busy := replacing.LoadOrStore(node, true); busy {
|
||||
markGivenAgain(node, started)
|
||||
return
|
||||
}
|
||||
defer replacing.Delete(node)
|
||||
for {
|
||||
replaceGivenOnce(ctx, open, node, started)
|
||||
replacing.Delete(node)
|
||||
// Read after letting go: a call meanwhile only marked the node.
|
||||
pending := takeGivenMark(node)
|
||||
if pending == nil || ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
if _, busy := replacing.LoadOrStore(node, true); busy {
|
||||
markGivenAgain(node, pending) // the one running now takes it
|
||||
return
|
||||
}
|
||||
started = pending
|
||||
}
|
||||
}
|
||||
|
||||
// givenAgain is, by node, the starts asked for while a replacement ran there, merged.
|
||||
var givenAgain = struct {
|
||||
sync.Mutex
|
||||
by map[string]map[string]time.Time
|
||||
}{by: map[string]map[string]time.Time{}}
|
||||
|
||||
func markGivenAgain(node string, started map[string]time.Time) {
|
||||
givenAgain.Lock()
|
||||
defer givenAgain.Unlock()
|
||||
m := givenAgain.by[node]
|
||||
if m == nil {
|
||||
m = map[string]time.Time{}
|
||||
givenAgain.by[node] = m
|
||||
}
|
||||
for module, at := range started {
|
||||
if at.After(m[module]) {
|
||||
m[module] = at
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func takeGivenMark(node string) map[string]time.Time {
|
||||
givenAgain.Lock()
|
||||
defer givenAgain.Unlock()
|
||||
m := givenAgain.by[node]
|
||||
delete(givenAgain.by, node)
|
||||
return m
|
||||
}
|
||||
|
||||
// replaceGivenOnce is one replacement on a node; a variable so a test can stand in for it.
|
||||
var replaceGivenOnce = func(ctx context.Context, open *stores, node string, started map[string]time.Time) {
|
||||
replaced, err := open.inventory.ReplaceGivenAfterStart(ctx, node, started)
|
||||
if err != nil {
|
||||
log.Printf("a value given by hand on %s could not be replaced after its module started: %v", node, err)
|
||||
|
||||
@@ -0,0 +1,34 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// A start heard while a replacement of given values runs on the node is not lost: it is marked, and the replacement
|
||||
// running takes it once it is done (issue 1945, the kind of the dropped wake: replaceGiven returned at once when busy).
|
||||
func TestAStartHeardDuringAReplacementIsTaken(t *testing.T) {
|
||||
was := replaceGivenOnce
|
||||
t.Cleanup(func() { replaceGivenOnce = was })
|
||||
var mu sync.Mutex
|
||||
var asked []map[string]time.Time
|
||||
at := time.Date(2026, 10, 25, 13, 0, 0, 0, time.UTC)
|
||||
replaceGivenOnce = func(ctx context.Context, open *stores, node string, started map[string]time.Time) {
|
||||
mu.Lock()
|
||||
asked = append(asked, started)
|
||||
first := len(asked) == 1
|
||||
mu.Unlock()
|
||||
if first {
|
||||
// Another module starts on the node while this one is being replaced.
|
||||
replaceGiven(ctx, open, node, map[string]time.Time{"redis": at})
|
||||
}
|
||||
}
|
||||
replaceGiven(context.Background(), nil, "ace", map[string]time.Time{"postgres": at})
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
if len(asked) != 2 || !asked[1]["redis"].Equal(at) {
|
||||
t.Fatalf("asked %v: the start heard during the replacement was lost", asked)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user