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)"
|
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) {
|
func replaceGiven(ctx context.Context, open *stores, node string, started map[string]time.Time) {
|
||||||
if _, busy := replacing.LoadOrStore(node, true); busy {
|
if _, busy := replacing.LoadOrStore(node, true); busy {
|
||||||
|
markGivenAgain(node, started)
|
||||||
return
|
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)
|
replaced, err := open.inventory.ReplaceGivenAfterStart(ctx, node, started)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Printf("a value given by hand on %s could not be replaced after its module started: %v", node, err)
|
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