An answered ask is kept until its yes is in the want, and an ask taken away is asked again under a new asking (issue 1966) #195

Merged
mesh-tools merged 3 commits from fix/1966-record-ask-lost into main 2026-11-09 22:08:39 +00:00
6 changed files with 241 additions and 43 deletions
@@ -605,46 +605,70 @@ func (p *passer) lift(ctx context.Context, module, fp string) {
}
}
// staleAsk says whether the hold, restore or record ask at key (read as the SDK reads it) is about something the trunk has left
// behind: a module no longer on the trunk, a fingerprint that is no longer the module's trunk fingerprint, or a composition the
// controller no longer composes for the node at that fingerprint. A backup ask and an ask of another kind are never stale here.
// A module of the trunk whose fingerprint is not read yet, and a composition not read yet, are not stale: nothing known is
// staleAsk says why the hold, restore or record ask at key (read as the SDK reads it) is about something the trunk has left
// behind, or "" when it is not: a module no longer on the trunk, a fingerprint that is no longer the module's trunk fingerprint,
// or, for a record ask, a composition the module's node no longer waits on. A backup ask and an ask of another kind are never
// stale here. A module of the trunk whose fingerprint is not read yet, and a have not read yet, are not stale: nothing known is
// nothing taken away.
func (p *passer) staleAsk(key string, trunk map[string]meshcatalog.ModulesByModuleValue) bool {
func (p *passer) staleAsk(key string, trunk map[string]meshcatalog.ModulesByModuleValue) string {
k, err := asks.ParseAnyAskKey(key)
if err != nil {
return false
return ""
}
switch k.Kind {
case asks.AskKindHold, asks.AskKindRestore:
m, on := trunk[k.Module]
return !on || (m.Fp2 != "" && m.Fp2 != k.FP)
if !on {
return "its module is no longer on the trunk"
}
if m.Fp2 != "" && m.Fp2 != k.FP {
return "its fingerprint is no longer the trunk's " + m.Fp2
}
case asks.AskKindRecord:
m, on := trunk[k.Module]
if !on {
return true
return "its module is no longer on the trunk"
}
// A node that composes its own asks about its own composition, which its have names as the one it waits on (yeses): that
// ask stands while the have waits on it, whatever the controller composes (issue 774, 2026-10-31: novox's asks of
// postgres, keycloak and mongodb were taken away and made again under a new id every pass, so no answer could count).
if hkey, err := nodeengineclient.HaveKey(k.Node, k.Module); err == nil {
if h, err := nodeengineclient.DecodeHave(p.w.value(buckets.Have, hkey)); err == nil && h.Phase == nodeengineclient.PhaseWaitingYes && h.Waits == k.Composed {
return false
if m.Fp2 == "" {
return ""
}
// The ask is about the composition its module's have waits on (ADR 0547 §3): the node composes its own, and the digest of
// its whole composition (the node's own have) is not a module's. A have that is mid-move (backing up, applying, checking,
// failed or held) leaves the ask where it is, for the ask was raised for a have that waited and a have is read mid-move
// on the way to its next wait (issue 1966: nats on novox, asked, taken away and written again at each pass).
hkey, err := nodeengineclient.HaveKey(k.Node, k.Module)
if err != nil {
return ""
}
h, err := nodeengineclient.DecodeHave(p.w.value(buckets.Have, hkey))
if err != nil {
return ""
}
switch h.Phase {
case nodeengineclient.PhaseWaitingYes:
if h.Waits != k.Composed {
return "its node waits for a yes on the composition " + h.Waits + ", not this one"
}
}
// The assigner asks about a composition at more than the trunk fingerprint: a node held back by the rules is asked about the
// revised composition of the fingerprint it runs (ADR 0545), and a node's first want at the older fingerprint it runs. So a
// record ask is stale only when its digest is the composition of none of the fingerprints the node can still want (the
// trunk's, its want's, its have's), and only when every one of them is read.
for _, fp := range p.fingerprintsOf(k.Node, k.Module, m.Fp2) {
d, ok := p.w.composed(k.Node, k.Module, fp)
if !ok || d.Digest == k.Composed {
return false
case nodeengineclient.PhaseRunning:
// A yes given and not yet written into the want is the operator's answer: the ask stays until the want names it, for the
// answer counts only while its ask is there (answeredWith), and taken away first the yes was never read.
if p.answeredWith(key, assign.OptionYes) && !p.yesWritten(k) {
return ""
}
return "its node runs " + h.Fp + " and waits for no yes"
}
return len(p.fingerprintsOf(k.Node, k.Module, m.Fp2)) > 0
}
return false
return ""
}
// yesWritten says whether the want of the record ask's module on its node names the composition the ask is about as allowed.
func (p *passer) yesWritten(k asks.AskKey) bool {
key, err := mwc.WantKey(k.Node, k.Module)
if err != nil {
return false
}
w, err := mwc.DecodeWant(p.w.value(buckets.Wants, key))
return err == nil && w.Yes == k.Composed
}
// fingerprintsOf are the fingerprints a node may still want of module: the trunk's, the one its want names and the one its have
@@ -678,7 +702,7 @@ func (p *passer) staleAsks(ctx context.Context, trunk map[string]meshcatalog.Mod
return
}
for key := range p.w.read(buckets.Asks) {
if p.staleAsk(key, trunk) {
if why := p.staleAsk(key, trunk); why != "" {
// The condition is cleared first and the ask goes only once that was said: the ask is what brings this key back at the
// next pass, so a clear that failed once is made again, with no state of its own (issue 1906, note 29).
k, kerr := asks.ParseAnyAskKey(key)
@@ -686,7 +710,7 @@ func (p *passer) staleAsks(ctx context.Context, trunk map[string]meshcatalog.Mod
if held && !p.clearHeld(k.Module, k.FP) {
continue
}
p.logf("the ask %s is about a fingerprint or composition the trunk has left behind and is taken away", key)
p.logf("the ask %s is about a fingerprint or composition the trunk has left behind and is taken away: %s", key, why)
p.remove(ctx, buckets.Asks, key)
// The held record stays: it is the memory that this fingerprint failed (ADR 0408 §4), and a revert returns the trunk to
// exactly it. Only the condition, which would else ask for an answer that can no longer be given, is cleared, and it is
@@ -748,7 +772,8 @@ func (p *passer) restoreAsk(ctx context.Context, module, node, fp string) {
return
}
words := p.heldHold(module, fp, nil).Words
a := assign.RestoreAsk(key, node, module, fp, words, time.Now())
now := p.clock()
a := assign.RestoreAsk(key, node, module, fp, words, assign.AskingSeconds(now), now)
p.put(ctx, buckets.Asks, key, func() ([]byte, error) { return assign.EncodeAsk(key, a) })
}
@@ -790,7 +815,8 @@ func (p *passer) recordAsk(ctx context.Context, module, fp string, need assign.R
if err != nil || p.w.has(buckets.Asks, key) {
return
}
a := assign.RecordAsk(key, need.Node, module, fp, need, time.Now())
now := p.clock()
a := assign.RecordAsk(key, need.Node, module, fp, need, assign.AskingSeconds(now), now)
p.put(ctx, buckets.Asks, key, func() ([]byte, error) { return assign.EncodeAsk(key, a) })
}
@@ -807,9 +807,9 @@ func TestAsksOfAFingerprintTheTrunkLeftBehindAreTakenAway(t *testing.T) {
case asks.AskKindHold:
a = assign.HoldAsk(key, k.Module, k.FP, "novox", "w", 1, time.Now())
case asks.AskKindRestore:
a = assign.RestoreAsk(key, k.Node, k.Module, k.FP, "w", time.Now())
a = assign.RestoreAsk(key, k.Node, k.Module, k.FP, "w", 2, time.Now())
case asks.AskKindRecord:
a = assign.RecordAsk(key, k.Node, k.Module, fpNew, assign.RecordNeed{Node: k.Node, Digest: k.Composed}, time.Now())
a = assign.RecordAsk(key, k.Node, k.Module, fpNew, assign.RecordNeed{Node: k.Node, Digest: k.Composed}, 2, time.Now())
default:
a = assign.BackupAsk(key, k.Node, k.Module, k.FP, digOld, "w", 2, time.Now())
}
@@ -820,9 +820,15 @@ func TestAsksOfAFingerprintTheTrunkLeftBehindAreTakenAway(t *testing.T) {
asksKV.values[key] = v
p.w.put(buckets.Asks, key, v)
}
placeOn(t, p.w, "novox", "web") // a placed node composes its own: the digest its have names
placeOn(t, p.w, "novox", "web") // a placed node composes its own: its module's have waits on a composition
if !c.noDigit {
composes(t, p.w, "novox", digNew)
hkey, _ := nodeengineclient.HaveKey("novox", "web")
raw, err := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: fpOld, Phase: nodeengineclient.PhaseWaitingYes,
Health: nodeengineclient.HealthOK, Judgement: "moves-data", Detail: "d", Waits: digNew})
if err != nil {
t.Fatal(err)
}
p.w.put(buckets.Have, hkey, raw)
}
heldOld, _ := mwc.HeldKey("web", fpOld)
wantsKV.values[heldOld] = []byte("x")
@@ -920,6 +926,7 @@ func mustKey(key string, err error) string {
func TestARecordAskOfAHeldBackNodeSurvivesPassesAndItsYesMovesTheWant(t *testing.T) {
// A placed node's yes comes from its have (issue 1126): folded here to hold the count.
aPlacedNodesWantsNameTheBuildAloneAndItsYesComesFromItsHave(t)
aRecordAskIsNotTakenAwayBeforeItsYesIsWrittenAndIsAskedAgainWhenWrittenAgain(t)
aReleasesApproveAnswersTheMoveOfItsBuildAlone(t)
}
@@ -1210,3 +1217,110 @@ func mustAskID(a asks.Ask) string {
func presenceValue(online bool) ([]byte, error) {
return json.Marshal(wdc.NodePresence{V: 1, Online: online})
}
// Issue 1966 (nats on novox): a yes given for a record ask was lost because the same pass took the ask away as stale before the
// want named the yes, and an ask taken away and written again was never asked again because it carried the id the router kept.
// Given a record ask of a placed node, then (a) a have mid-move, or running with the operator's yes unwritten, leaves the ask,
// and a yes written into the want lets it go; and (b) a record, restore or hold ask written again after it was taken away has
// another id, so the teller asks it again.
func aRecordAskIsNotTakenAwayBeforeItsYesIsWrittenAndIsAskedAgainWhenWrittenAgain(t *testing.T) {
fp := meshcatalogclient.SourcePrefix2 + strings.Repeat("a", 64)
dig, other := "sha256:"+strings.Repeat("e", 64), "sha256:"+strings.Repeat("f", 64)
trunk := map[string]meshcatalog.ModulesByModuleValue{"web": {Fp2: fp, Manifest: "sha256:" + strings.Repeat("1", 64), Repository: "novox/web", Commit: strings.Repeat("1", 40), Path: meshcatalogclient.RootPath}}
rkey, _ := asks.RecordAskKey("novox", "web", dig)
haveOf := func(phase, waits string) nodeengine.HaveByNodeModuleValue {
h := nodeengine.HaveByNodeModuleValue{Fp: fp, Phase: phase, Health: nodeengineclient.HealthOK}
if phase == nodeengineclient.PhaseWaitingYes {
h.Judgement, h.Detail, h.Waits = assign.JudgedMovesData, "d", waits
}
return h
}
for _, c := range []struct {
name string
have nodeengine.HaveByNodeModuleValue
yes bool // the operator's yes is written under the ask
wanted string // the want's yes
gone bool
}{
{"waiting for a yes on it", haveOf(nodeengineclient.PhaseWaitingYes, dig), false, "", false},
{"waiting for a yes on another composition", haveOf(nodeengineclient.PhaseWaitingYes, other), false, "", true},
{"mid-move (applying)", haveOf(nodeengineclient.PhaseApplying, ""), false, "", false},
{"running, never answered", haveOf(nodeengineclient.PhaseRunning, ""), false, "", true},
{"running, the yes given and not in the want", haveOf(nodeengineclient.PhaseRunning, ""), true, "", false},
{"running, the yes written into the want", haveOf(nodeengineclient.PhaseRunning, ""), true, dig, true},
} {
t.Run(c.name, func(t *testing.T) {
asksKV, wantsKV := &memKV{values: map[string][]byte{}}, &memKV{values: map[string][]byte{}}
p := newPasser(kvSet{buckets.Asks: asksKV, buckets.Wants: wantsKV}, time.Now())
placeOn(t, p.w, "novox", "web")
composes(t, p.w, "novox", "sha256:"+strings.Repeat("9", 64)) // the whole node's composition is never a module's
a := assign.RecordAsk(rkey, "novox", "web", fp, assign.RecordNeed{Node: "novox", Digest: dig}, 2, time.Now())
v, err := assign.EncodeAsk(rkey, a)
if err != nil {
t.Fatal(err)
}
asksKV.values[rkey] = v
p.w.put(buckets.Asks, rkey, v)
hkey, _ := nodeengineclient.HaveKey("novox", "web")
raw, err := nodeengineclient.EncodeHave(c.have)
if err != nil {
t.Fatal(err)
}
p.w.put(buckets.Have, hkey, raw)
if c.yes {
body, _ := mdc.EncodeAnswer(mdc.Answer{Option: assign.OptionYes, Warrant: mustAskID(a) + "@2026-11-09T10:34:48Z", Words: "yes"})
p.w.put(buckets.Answers, rkey, body)
}
if c.wanted != "" {
wkey, _ := mwc.WantKey("novox", "web")
w, _ := mwc.EncodeWant(mwc.Want{Fp: fp, Self: true, Yes: c.wanted})
p.w.put(buckets.Wants, wkey, w)
}
p.staleAsks(context.Background(), trunk)
if _, there := asksKV.values[rkey]; there == c.gone {
t.Errorf("the ask is left in the bucket = %v, want %v", there, !c.gone)
}
})
}
t.Run("a key written again after it was taken away has another id", func(t *testing.T) {
rec := func(p *passer) string {
p.recordAsk(context.Background(), "web", fp, assign.RecordNeed{Node: "novox", Digest: dig})
return readAskID(t, p, rkey)
}
rest, _ := asks.RestoreAskKey("novox", "web", fp)
restore := func(p *passer) string {
p.restoreAsk(context.Background(), "web", "novox", fp)
return readAskID(t, p, rest)
}
hold, _ := asks.HoldAskKey("web", fp)
holdIt := func(p *passer) string {
p.holdAsk(context.Background(), "web", fp, &mwc.Held{Node: "novox", Words: "w"})
return readAskID(t, p, hold)
}
for name, c := range map[string]struct {
write func(*passer) string
key string
}{"record": {rec, rkey}, "restore": {restore, rest}, "hold": {holdIt, hold}} {
at := time.Now()
asksKV := &memKV{values: map[string][]byte{}}
p := newPasser(kvSet{buckets.Asks: asksKV, buckets.Wants: &memKV{values: map[string][]byte{}}}, at)
first := c.write(p)
p.remove(context.Background(), buckets.Asks, c.key)
p.now = func() time.Time { return at.Add(time.Minute) }
if again := c.write(p); again == first {
t.Errorf("%s: the ask written again carries the id %s the router kept, so it is never asked again", name, first)
}
}
})
}
// readAskID is the router id of the ask at key in the pass's asks.
func readAskID(t *testing.T, p *passer, key string) string {
t.Helper()
d, err := decodeAsk(key, p.w.value(buckets.Asks, key))
if err != nil {
t.Fatalf("%s: %v", key, err)
}
return d.Ask.ID
}
@@ -45,11 +45,13 @@ const (
)
// RestoreAsk is the ask about a module holding data whose fingerprint failed its health check on a node (ADR 0429 §6): it is
// not set back by itself, since a set-back would lose the writes made since. words is the check's words. Restore sets the
// not set back by itself, since a set-back would lose the writes made since. words is the check's words. asking is the writing's
// number (AskingSeconds): an ask taken away as stale and written again is another asking under an id of its own, since the
// router drops a second ask under an id it keeps and the teller took the first back (issue 1966). Restore sets the
// node back to the last healthy fingerprint and loses the newer writes; leaving it stopped keeps the data as it is.
func RestoreAsk(key, node, module, fp, words string, now time.Time) asks.Ask {
return assignerAsk(0, oc.Ask{
ID: meshdeliveryclient.AskingID(key, 0),
func RestoreAsk(key, node, module, fp, words string, asking int, now time.Time) asks.Ask {
return assignerAsk(asking, oc.Ask{
ID: meshdeliveryclient.AskingID(key, asking),
Headline: fmt.Sprintf("%s failed its health check on %s", module, node),
Explanation: fmt.Sprintf("%s holds data, and its health check failed on %s at %s: %s. Setting it back to the version that ran healthy "+
"would lose the writes made since; leaving it stopped keeps the data as it is.", module, node, short(fp), words),
@@ -160,15 +162,17 @@ const (
// RecordAsk is the ask about a move of a module of policy record on node to the composition with the digest, judged by the
// controller to recreate a container, to move data, or not to be judgeable (ADR 0539 §14). The want names the composition only
// after the answer yes. The way the move was judged is in the ask.
func RecordAsk(key, node, module, fp string, need RecordNeed, now time.Time) asks.Ask {
// after the answer yes. The way the move was judged is in the ask. asking is the writing's number (AskingSeconds), so a key taken
// away as stale and written again is asked again under a new id (issue 1966: nats on novox was asked once, taken back and never
// asked again, for the rewritten key carried the id the router already kept).
func RecordAsk(key, node, module, fp string, need RecordNeed, asking int, now time.Time) asks.Ask {
what := map[Judgement]string{
JudgedRecreates: "recreates a container",
JudgedMovesData: "moves the module's data",
JudgedCannot: "cannot be judged: " + need.Words,
}[need.Judgement]
return assignerAsk(0, oc.Ask{
ID: meshdeliveryclient.AskingID(key, 0),
return assignerAsk(asking, oc.Ask{
ID: meshdeliveryclient.AskingID(key, asking),
Headline: fmt.Sprintf("%s waits for your yes on %s", module, node),
Explanation: fmt.Sprintf("%s is recorded: a person takes each of its moves. The next one on %s (version %s) %s. "+
"The node is told to apply it only after you say yes.", module, node, short(fp), what),
@@ -183,6 +187,10 @@ func RecordAsk(key, node, module, fp string, need RecordNeed, now time.Time) ask
})
}
// AskingSeconds is the number of an asking written at now: its time in seconds, as a hold's and an assignments' ask are numbered
// (never below 2, which AskingID needs for an id of its own). A key written again after it was taken away is another asking.
func AskingSeconds(now time.Time) int { return max(int(now.Unix()), 2) }
// NextBackupAsking is the number of the asking that follows the one numbered after (a negative number: none yet). Numbers 0 and 1
// give one id (meshdeliveryclient.AskingID), so the first asking is 0 and the next 2.
func NextBackupAsking(after int) int {
@@ -81,7 +81,7 @@ func theRestoreAskIsValid(t *testing.T) {
t.Fatal(err)
}
now := time.Now()
a := RestoreAsk(key, "ace", "mail", fNew, "connection refused", now)
a := RestoreAsk(key, "ace", "mail", fNew, "connection refused", 2, now)
if err := occ.Check(mustAsk(t, a), now); err != nil {
t.Errorf("the router refuses the ask: %v", err)
}
@@ -186,7 +186,7 @@ func TestTheRecordAndBackupAsksAreValid(t *testing.T) {
if err != nil {
t.Fatal(err)
}
r := RecordAsk(rkey, "ace", "mail", fNew, RecordNeed{Node: "ace", Digest: "sha256:" + strings.Repeat("b", 64), Judgement: JudgedCannot, Words: "the composition it runs is gone"}, now)
r := RecordAsk(rkey, "ace", "mail", fNew, RecordNeed{Node: "ace", Digest: "sha256:" + strings.Repeat("b", 64), Judgement: JudgedCannot, Words: "the composition it runs is gone"}, 2, now)
bkey, err := asks.BackupAskKey("ace", "mail", fNew)
if err != nil {
t.Fatal(err)
@@ -168,6 +168,10 @@ func (t *Teller) raiseAssignerAsk(key string) {
}
if open(k) {
t.catchUp(id)
} else {
// Said, never silent (issue 1966: a record ask taken back when its key went was written again under the same id on
// 2026-11-09 and nothing said why the operator was never asked): the assigner numbers each writing (asks.Ask.Asking).
t.logf("%s: the ask %s ended (answered, lapsed or taken back) and the assigner wrote the key again under the same id: not asked again, the router would drop it; a writing under a new asking is asked", key, id)
}
return
}
@@ -128,6 +128,7 @@ func TestAnAssignerAskAnsweredInTheViewOrTakenAwayIsTakenBack(t *testing.T) {
// 1906/H1: an assigner ask refused for want of room waits and is asked again when another of the teller's asks ends, as a merge
// ask does; a refusal for its words is said as a condition about a delivery ask, not a merge.
func TestAnAssignerAskWaitsForRoomAndARefusalIsSaidAsADeliveryAsk(t *testing.T) {
anAssignerAskWrittenAgainAfterBeingTakenAwayIsAskedAgainUnderANewAsking(t)
b := newBed(t)
assigner, _ := withView(b)
_, first := b.askOf(t, 1, 1)
@@ -209,3 +210,48 @@ func TestABurstOfHoldsLeavesRoomForAMergeAsk(t *testing.T) {
t.Errorf("asked %v after the merge ask was answered, want no further hold", got)
}
}
// Issue 1966 (nats on novox): an assigner ask taken back when its key went is asked again when the assigner writes the key
// again under a new asking, whatever its kind; written again under the id the router kept it is not (the router would drop it)
// and the teller says so, never silent.
func anAssignerAskWrittenAgainAfterBeingTakenAwayIsAskedAgainUnderANewAsking(t *testing.T) {
digest := "sha256:" + strings.Repeat("e", 64)
fp := "src2:" + strings.Repeat("a", 64)
recordKey, _ := asks.RecordAskKey("novox", "nats", digest)
restoreKey, _ := asks.RestoreAskKey("novox", "nats", fp)
holdKey, _ := asks.HoldAskKey("nats", fp)
for kind, key := range map[string]string{"record": recordKey, "restore": restoreKey, "hold": holdKey} {
t.Run(kind, func(t *testing.T) {
write := func(assigner memGate, asking int) oc.Ask {
ask := oc.Ask{ID: meshdeliveryclient.AskingID(key, asking), Headline: "nats waits", Explanation: "e", Who: occ.Operator,
Expires: occ.FormatTime(now.Add(occ.ApproveLasts)), OnExpiry: "stays",
Options: []oc.Option{{ID: "yes", Label: "Yes", Does: "d", Level: occ.Acknowledge}, {ID: "no", Label: "No", Does: "d", Level: occ.Acknowledge}}}
value, err := asks.EncodeAsk(key, asks.Ask{Ask: rawAsk(t, ask), Asking: asking})
if err != nil {
t.Fatal(err)
}
assigner[key] = value
return ask
}
b := newBed(t)
assigner, _ := withView(b)
first := write(assigner, 1000)
b.t.OnAssignerAsk(key)
delete(assigner, key)
b.t.OnAssignerAskGone(key) // the assigner took it away as stale: the router ask is taken back
second := write(assigner, 1060)
b.t.OnAssignerAsk(key)
if got := b.askedIDs(); len(got) != 2 || got[0] != first.ID || got[1] != second.ID || first.ID == second.ID {
t.Fatalf("asked %v, want %s then %s; log:\n%s", got, first.ID, second.ID, b.logs.String())
}
// Written again under the id the router kept, after it was taken back: not asked, and said.
delete(assigner, key)
b.t.OnAssignerAskGone(key)
write(assigner, 1060)
b.t.OnAssignerAsk(key)
if got := b.askedIDs(); len(got) != 2 || !strings.Contains(b.logs.String(), "wrote the key again under the same id") {
t.Errorf("asked %v after the same id was written again; log:\n%s", got, b.logs.String())
}
})
}
}