Subtask A: a record ask is kept while its node waits on it or its yes is unwritten, and an ask written again is asked again under a new id (issue 1966)
The assigner took record.novox.nats away as stale in the same second the operator's yes was written (the yes counts only while the ask is there), and rewrote it under the id the router already kept, so the teller never asked it again. The stale test compared a module's ask with the whole node's composition; it now reads the module's own have, and restore and record asks are numbered by their writing's time as hold asks are. Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
@@ -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")
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user