diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go index b3982c86..4dfc33e8 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go @@ -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) }) } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go index ee81b744..58843772 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go @@ -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") @@ -1210,3 +1216,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 TestARecordAskIsNotTakenAwayBeforeItsYesIsWrittenAndIsAskedAgainWhenWrittenAgain(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 +} diff --git a/modules/mesh-delivery-assigner/internal/assign/asks.go b/modules/mesh-delivery-assigner/internal/assign/asks.go index 2587a5ae..66d2b601 100644 --- a/modules/mesh-delivery-assigner/internal/assign/asks.go +++ b/modules/mesh-delivery-assigner/internal/assign/asks.go @@ -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 { diff --git a/modules/mesh-delivery-assigner/internal/assign/asks_test.go b/modules/mesh-delivery-assigner/internal/assign/asks_test.go index 9624858b..f88004df 100644 --- a/modules/mesh-delivery-assigner/internal/assign/asks_test.go +++ b/modules/mesh-delivery-assigner/internal/assign/asks_test.go @@ -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)