The assigner acts on answers: a release writes the held want again, a backup answer goes into the node's want, a record move waits for its yes, a restore answer sets a data module back (issue 1906 D4)
This commit is contained in:
@@ -22,6 +22,8 @@ import (
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
"git.novox.be/novox/mesh-sdk/go/provider"
|
||||
|
||||
"mesh-delivery-assigner/internal/assign"
|
||||
)
|
||||
|
||||
// bed is an embedded bus with JetStream, every bucket the assigner reads and keeps, and the stream the mesh's events
|
||||
@@ -515,3 +517,183 @@ func TestAWantWaitsForItsComposition(t *testing.T) {
|
||||
t.Fatalf("the node's own want = %+v, %v", w, err)
|
||||
}
|
||||
}
|
||||
|
||||
// fixture is a module web (or the manifest given) at fp on the node ace, which is online: built, granted and assigned.
|
||||
func (b *bed) fixture(t *testing.T, name, fp, manifest string) {
|
||||
t.Helper()
|
||||
digest := b.manifest(t, name, manifest)
|
||||
mod, _ := delivery.EncodeModule(delivery.Module{FP2: fp, Manifest: digest, Repository: "novox/" + name, Commit: strings.Repeat("1", 40), Path: delivery.RootPath})
|
||||
b.putJSON(t, delivery.ModulesBucket, name, mod)
|
||||
built, _ := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: delivery.BuildDone, By: "builder.novox.1", Step: "pushed", Manifest: digest})
|
||||
bkey, _ := delivery.BuilderBuildKey(name, fp)
|
||||
b.putJSON(t, delivery.BuilderBuildsBucket, bkey, built)
|
||||
gkey, _ := delivery.ControllerGrantKey(name, fp)
|
||||
grant, _ := delivery.EncodeControllerGrant(delivery.ControllerGrant{})
|
||||
b.putJSON(t, delivery.ControllerGrantsBucket, gkey, grant)
|
||||
on, _ := delivery.EncodePresence(delivery.Presence{Online: true})
|
||||
b.putJSON(t, delivery.PresenceBucket, "ace", on)
|
||||
b.putJSON(t, provider.AssignmentsBucket, "ace."+name, []byte(`{"node":"ace","module":"`+name+`","build":"b-1"}`))
|
||||
}
|
||||
|
||||
func (b *bed) putWant(t *testing.T, node, module string, w delivery.Want) {
|
||||
t.Helper()
|
||||
key, _ := delivery.WantKey(node, module)
|
||||
v, err := delivery.EncodeWant(w)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
b.putJSON(t, delivery.AssignerWantsBucket, key, v)
|
||||
}
|
||||
|
||||
// wantNow waits until the want of module on node satisfies ok and answers it.
|
||||
func (b *bed) wantNow(t *testing.T, node, module, what string, ok func(delivery.Want) bool) delivery.Want {
|
||||
t.Helper()
|
||||
key, _ := delivery.WantKey(node, module)
|
||||
var last delivery.Want
|
||||
for deadline := time.Now().Add(15 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
|
||||
if e, err := b.buckets[delivery.AssignerWantsBucket].Get(b.ctx, key); err == nil {
|
||||
if w, err := delivery.DecodeWant(e.Value()); err == nil {
|
||||
last = w
|
||||
if ok(w) {
|
||||
return w
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
t.Fatalf("never saw: %s; the want is %+v", what, last)
|
||||
return last
|
||||
}
|
||||
|
||||
// answer says the teller wrote the operator's answer to the ask at key.
|
||||
func (b *bed) answer(t *testing.T, key, option string) {
|
||||
t.Helper()
|
||||
v, err := delivery.EncodeAnswer(delivery.Answer{Option: option, Warrant: "r1@2026-10-24T12:00:00Z", Words: "the operator"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
b.putJSON(t, delivery.AnswersBucket, key, v)
|
||||
}
|
||||
|
||||
func (b *bed) hasKey(bucket, key string) bool {
|
||||
_, err := b.buckets[bucket].Get(b.ctx, key)
|
||||
return err == nil
|
||||
}
|
||||
|
||||
// 1906/D4 (ADR 0539 §19), whole on the bus: the operator's release of a held fingerprint writes its want again, by the rules,
|
||||
// and clears the condition.
|
||||
func TestAReleaseAnswerWritesTheWantAgain(t *testing.T) {
|
||||
b := startBed(t, "")
|
||||
fNew, fOld := delivery.SourcePrefix2+strings.Repeat("b", 64), delivery.SourcePrefix2+strings.Repeat("a", 64)
|
||||
b.fixture(t, "web", fNew, `{"module":"web","version":"1"}`)
|
||||
b.compose(t, "ace", "web", fNew, 'b', delivery.JudgedNeither)
|
||||
b.compose(t, "ace", "web", fOld, 'a', delivery.JudgedNeither)
|
||||
healthy, _ := delivery.EncodeHave(delivery.Have{FP: fOld, Phase: delivery.PhaseRunning, Health: delivery.HealthOK})
|
||||
b.putJSON(t, delivery.HaveBucket, "ace.web", healthy) // it ran the older fingerprint healthy before
|
||||
failed, _ := delivery.EncodeHave(delivery.Have{FP: fNew, Phase: delivery.PhaseRunning, Health: delivery.HealthFailed, Detail: "connection refused"})
|
||||
b.putJSON(t, delivery.HaveBucket, "ace.web", failed)
|
||||
b.putWant(t, "ace", "web", delivery.Want{FP: fNew, Composed: "sha256:" + strings.Repeat("b", 64)})
|
||||
b.start(t)
|
||||
w := b.wantNow(t, "ace", "web", "set back to the older fingerprint", func(w delivery.Want) bool { return w.FP == fOld })
|
||||
if w.Newest != fNew {
|
||||
t.Fatalf("the set-back want = %+v", w)
|
||||
}
|
||||
hold, _ := delivery.HoldAskKey("web", fNew)
|
||||
b.get(t, delivery.AssignerAsksBucket, hold)
|
||||
b.answer(t, hold, "release")
|
||||
w = b.wantNow(t, "ace", "web", "the held fingerprint's want written again", func(w delivery.Want) bool { return w.FP == fNew })
|
||||
if w.Composed != "sha256:"+strings.Repeat("b", 64) || w.Rule != 0 {
|
||||
t.Errorf("the want = %+v", w)
|
||||
}
|
||||
for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(100 * time.Millisecond) {
|
||||
cleared := false
|
||||
for _, c := range b.conditionsSaid(t) {
|
||||
cleared = cleared || (c.Subject.ID == "web" && c.State == asks.StateCleared)
|
||||
}
|
||||
if cleared {
|
||||
return
|
||||
}
|
||||
}
|
||||
t.Error("the condition of the held fingerprint was not cleared")
|
||||
}
|
||||
|
||||
// 1906/D4 (ADR 0539 §16), whole on the bus: a module held on a node because its backup failed is asked about, and the answer
|
||||
// is written into the node's want.
|
||||
func TestABackupAnswerIsWrittenIntoTheNodesWant(t *testing.T) {
|
||||
b := startBed(t, "")
|
||||
fNew := delivery.SourcePrefix2 + strings.Repeat("b", 64)
|
||||
b.fixture(t, "web", fNew, `{"module":"web","version":"1"}`)
|
||||
b.compose(t, "ace", "web", fNew, 'b', delivery.JudgedNeither)
|
||||
b.putWant(t, "ace", "web", delivery.Want{FP: fNew, Composed: "sha256:" + strings.Repeat("b", 64)})
|
||||
held, _ := delivery.EncodeHave(delivery.Have{FP: fNew, Phase: delivery.PhaseHeld, Health: delivery.HealthWaits, Detail: "no space left"})
|
||||
b.putJSON(t, delivery.HaveBucket, "ace.web", held)
|
||||
b.start(t)
|
||||
akey, _ := delivery.BackupAskKey("ace", "web", fNew)
|
||||
a, err := delivery.DecodeAssignerAsk(akey, b.get(t, delivery.AssignerAsksBucket, akey))
|
||||
if err != nil || !strings.Contains(a.Ask.Explanation, "no space left") {
|
||||
t.Fatalf("the backup ask = %+v, %v", a, err)
|
||||
}
|
||||
b.answer(t, akey, delivery.BackupSkip)
|
||||
w := b.wantNow(t, "ace", "web", "the answer in the want", func(w delivery.Want) bool { return w.Backup == delivery.BackupSkip })
|
||||
if w.FP != fNew || w.Composed != "sha256:"+strings.Repeat("b", 64) {
|
||||
t.Errorf("the want = %+v", w)
|
||||
}
|
||||
time.Sleep(time.Second)
|
||||
if w := b.wantNow(t, "ace", "web", "the answer kept", func(delivery.Want) bool { return true }); w.Backup != delivery.BackupSkip {
|
||||
t.Errorf("a later pass erased the answer: %+v", w)
|
||||
}
|
||||
}
|
||||
|
||||
const recorded = `{"module":"web","version":"1","upgrade":{"policy":"record","why":"a person takes each build"}}`
|
||||
|
||||
// 1906/D4 (ADR 0539 §14), whole on the bus: a move of a module of policy record waits for its ask's yes when judged to recreate
|
||||
// a container, to move data or not to be judgeable, at a new fingerprint and at a new revision of the same alike; one judged
|
||||
// neither is written at once.
|
||||
func TestARecordMoveWaitsForTheYesAndANeitherMoveDoesNot(t *testing.T) {
|
||||
b := startBed(t, "")
|
||||
fNew, fOld := delivery.SourcePrefix2+strings.Repeat("b", 64), delivery.SourcePrefix2+strings.Repeat("a", 64)
|
||||
b.fixture(t, "web", fNew, recorded)
|
||||
old := "sha256:" + strings.Repeat("a", 64)
|
||||
b.putWant(t, "ace", "web", delivery.Want{FP: fOld, Composed: old})
|
||||
nw := b.compose(t, "ace", "web", fNew, 'b', delivery.JudgedRecreates)
|
||||
b.start(t)
|
||||
akey, _ := delivery.RecordAskKey("ace", "web", nw)
|
||||
a, err := delivery.DecodeAssignerAsk(akey, b.get(t, delivery.AssignerAsksBucket, akey))
|
||||
if err != nil || !strings.Contains(a.Ask.Explanation, "recreates a container") {
|
||||
t.Fatalf("the record ask = %+v, %v", a, err)
|
||||
}
|
||||
time.Sleep(time.Second)
|
||||
if w := b.wantNow(t, "ace", "web", "the old want", func(delivery.Want) bool { return true }); w.FP != fOld {
|
||||
t.Fatalf("the move was written before the yes: %+v", w)
|
||||
}
|
||||
b.answer(t, akey, assign.OptionYes)
|
||||
b.wantNow(t, "ace", "web", "the move after the yes", func(w delivery.Want) bool { return w.FP == fNew && w.Composed == nw })
|
||||
// a new revision at the same fingerprint, judged to move data, waits as well
|
||||
rev := b.compose(t, "ace", "web", fNew, 'c', delivery.JudgedMovesData)
|
||||
rkey, _ := delivery.RecordAskKey("ace", "web", rev)
|
||||
b.get(t, delivery.AssignerAsksBucket, rkey)
|
||||
time.Sleep(time.Second)
|
||||
if w := b.wantNow(t, "ace", "web", "the want", func(delivery.Want) bool { return true }); w.Composed != nw {
|
||||
t.Fatalf("the revision was written before the yes: %+v", w)
|
||||
}
|
||||
// and a revision judged neither is written at once
|
||||
free := b.compose(t, "ace", "web", fNew, 'd', delivery.JudgedNeither)
|
||||
b.wantNow(t, "ace", "web", "a move judged neither written at once", func(w delivery.Want) bool { return w.Composed == free })
|
||||
}
|
||||
|
||||
// 1906/D4 (ADR 0539 §7): a revision of the bus module's composition at the same fingerprint that is judged neither (its user
|
||||
// list only) is written at once, and no ask is made; the bus's new fingerprint is not written.
|
||||
func TestAUserListRevisionOfTheBusIsWrittenAtOnce(t *testing.T) {
|
||||
b := startBed(t, "")
|
||||
fNew, fOld := delivery.SourcePrefix2+strings.Repeat("b", 64), delivery.SourcePrefix2+strings.Repeat("a", 64)
|
||||
b.fixture(t, "bus", fNew, `{"module":"bus","version":"1","provides":[{"name":"mesh-bus"}]}`)
|
||||
b.putWant(t, "ace", "bus", delivery.Want{FP: fOld, Composed: "sha256:" + strings.Repeat("a", 64)})
|
||||
b.compose(t, "ace", "bus", fNew, 'b', delivery.JudgedNeither)
|
||||
b.start(t)
|
||||
time.Sleep(time.Second)
|
||||
if w := b.wantNow(t, "ace", "bus", "the want", func(delivery.Want) bool { return true }); w.FP != fOld {
|
||||
t.Fatalf("the assigner moved the bus to a new fingerprint: %+v", w)
|
||||
}
|
||||
b.putWant(t, "ace", "bus", delivery.Want{FP: fNew, Composed: "sha256:" + strings.Repeat("b", 64)})
|
||||
free := b.compose(t, "ace", "bus", fNew, 'e', delivery.JudgedNeither)
|
||||
b.wantNow(t, "ace", "bus", "the user-list revision written at once", func(w delivery.Want) bool { return w.Composed == free })
|
||||
}
|
||||
|
||||
@@ -42,6 +42,7 @@ var readBuckets = []string{
|
||||
provider.AssignmentsBucket,
|
||||
delivery.ControllerGrantsBucket,
|
||||
delivery.ControllerComposedBucket,
|
||||
delivery.AnswersBucket,
|
||||
delivery.BuilderBuildsBucket,
|
||||
delivery.CandidatesBucket,
|
||||
delivery.ManifestsBucket,
|
||||
@@ -134,6 +135,7 @@ var watched = []string{
|
||||
provider.AssignmentsBucket,
|
||||
delivery.ControllerGrantsBucket,
|
||||
delivery.ControllerComposedBucket,
|
||||
delivery.AnswersBucket,
|
||||
delivery.BuilderBuildsBucket,
|
||||
delivery.CandidatesBucket,
|
||||
delivery.ModulesBucket,
|
||||
|
||||
@@ -30,6 +30,7 @@ func (p *passer) pass(ctx context.Context) {
|
||||
p.compositions(ctx)
|
||||
p.wants(ctx)
|
||||
p.nodeWants(ctx)
|
||||
p.backups(ctx)
|
||||
}
|
||||
|
||||
// nodeWants writes each node's own want, `want.<node>.node`, naming the digest of the node's own composition, once the
|
||||
@@ -84,11 +85,24 @@ func (p *passer) wants(ctx context.Context) {
|
||||
fp := trunk[name].FP2
|
||||
name := name
|
||||
m := assign.Module{Name: name, FP: fp, Manifest: manifests[name], Nodes: nodesOf[name],
|
||||
Composed: func(node, fp string) (delivery.ComposedDigest, bool) { return p.w.composed(node, name, fp) }}
|
||||
Composed: func(node, fp string) (delivery.ComposedDigest, bool) { return p.w.composed(node, name, fp) },
|
||||
Restored: func(node string) bool {
|
||||
key, err := delivery.RestoreAskKey(node, name, fp)
|
||||
return err == nil && p.answeredWith(key, assign.OptionRestore)
|
||||
},
|
||||
Allowed: func(node, digest string) bool {
|
||||
key, err := delivery.RecordAskKey(node, name, digest)
|
||||
return err == nil && p.answeredWith(key, assign.OptionYes)
|
||||
}}
|
||||
if fp != "" {
|
||||
held, _ := delivery.HeldKey(name, fp)
|
||||
grant, _ := delivery.ControllerGrantKey(name, fp)
|
||||
m.Built, m.Held, m.Granted = p.w.built(name, fp), p.w.has(delivery.AssignerWantsBucket, held), p.w.has(delivery.ControllerGrantsBucket, grant)
|
||||
hold, _ := delivery.HoldAskKey(name, fp)
|
||||
m.Released = p.answeredWith(hold, "release")
|
||||
m.Built, m.Held, m.Granted = p.w.built(name, fp), p.w.has(delivery.AssignerWantsBucket, held) && !m.Released, p.w.has(delivery.ControllerGrantsBucket, grant)
|
||||
if m.Released {
|
||||
p.clearHeld(name, fp)
|
||||
}
|
||||
}
|
||||
last := map[string]string{}
|
||||
for _, node := range m.Nodes {
|
||||
@@ -109,6 +123,9 @@ func (p *passer) wants(ctx context.Context) {
|
||||
for _, node := range out.Restore {
|
||||
p.restoreAsk(ctx, name, node, fp)
|
||||
}
|
||||
for _, need := range out.Record {
|
||||
p.recordAsk(ctx, name, fp, need)
|
||||
}
|
||||
for node, want := range out.Wants {
|
||||
key, err := delivery.WantKey(node, name)
|
||||
if err != nil {
|
||||
@@ -227,3 +244,100 @@ func (p *passer) restoreAsk(ctx context.Context, module, node, fp string) {
|
||||
a := assign.RestoreAsk(key, node, module, fp, words, time.Now())
|
||||
p.put(ctx, delivery.AssignerAsksBucket, key, func() ([]byte, error) { return delivery.EncodeAssignerAsk(key, a) })
|
||||
}
|
||||
|
||||
// answeredWith says whether the teller wrote an answer to the ask at key and the option chosen is option. The answer is the
|
||||
// operator's, given in the view or on the phone, and stays under its key (ADR 0539 §14, §16, §19).
|
||||
func (p *passer) answeredWith(key, option string) bool {
|
||||
a, ok := p.w.answer(key)
|
||||
return ok && a.Option == option
|
||||
}
|
||||
|
||||
// clearHeld clears, once, the condition of a hold the operator released.
|
||||
func (p *passer) clearHeld(module, fp string) {
|
||||
c := assign.HeldCondition(module, fp, "", "", false, false)
|
||||
if p.say == nil || p.said[c.Key+".cleared"] {
|
||||
return
|
||||
}
|
||||
if err := p.say(c); err != nil {
|
||||
p.logf("the condition of the released %s could not be cleared, and is cleared at the next pass: %v", module, err)
|
||||
return
|
||||
}
|
||||
p.said[c.Key+".cleared"] = true
|
||||
}
|
||||
|
||||
// recordAsk writes, once, the ask about a move of a module of policy record that waits for the operator's yes.
|
||||
func (p *passer) recordAsk(ctx context.Context, module, fp string, need assign.RecordNeed) {
|
||||
key, err := delivery.RecordAskKey(need.Node, module, need.Digest)
|
||||
if err != nil || p.w.has(delivery.AssignerAsksBucket, key) {
|
||||
return
|
||||
}
|
||||
a := assign.RecordAsk(key, need.Node, module, fp, need, time.Now())
|
||||
p.put(ctx, delivery.AssignerAsksBucket, key, func() ([]byte, error) { return delivery.EncodeAssignerAsk(key, a) })
|
||||
}
|
||||
|
||||
// backups sees to the modules held on a node because the backup before an apply failed (the have's phase held, ADR 0539 §16):
|
||||
// it asks, once for each asking, whether to back up again or go on without a backup, raises the condition naming node, module and
|
||||
// the backup's words, and writes the answer into the node's want for the module, which the node agent reads from it and from
|
||||
// nothing else. A backup that fails again after the answer again is asked again, as the second asking.
|
||||
func (p *passer) backups(ctx context.Context) {
|
||||
wanted := p.w.wantsOf()
|
||||
for key, value := range p.w.read(delivery.HaveBucket) {
|
||||
node, module, isNode, err := delivery.ParseHaveKey(key)
|
||||
if err != nil || isNode {
|
||||
continue
|
||||
}
|
||||
have, err := delivery.DecodeHave(value)
|
||||
if err != nil || have.Phase != delivery.PhaseHeld {
|
||||
continue
|
||||
}
|
||||
want, ok := wanted[module][node]
|
||||
if !ok || want.Removed {
|
||||
continue
|
||||
}
|
||||
akey, err := delivery.BackupAskKey(node, module, want.FP)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
answer, answered := p.w.answer(akey)
|
||||
switch {
|
||||
case answered && (answer.Option == delivery.BackupAgain || answer.Option == delivery.BackupSkip):
|
||||
if want.Backup != answer.Option {
|
||||
want.Backup = answer.Option
|
||||
wkey, err := delivery.WantKey(node, module)
|
||||
if err == nil {
|
||||
w := want
|
||||
p.put(ctx, delivery.AssignerWantsBucket, wkey, func() ([]byte, error) { return delivery.EncodeWant(w) })
|
||||
}
|
||||
continue
|
||||
}
|
||||
if answer.Option == delivery.BackupAgain {
|
||||
// the node tried again and the backup failed again: the second asking, once
|
||||
p.backupAsk(ctx, akey, node, module, want.FP, have.Detail, 2)
|
||||
}
|
||||
case !answered:
|
||||
p.backupAsk(ctx, akey, node, module, want.FP, have.Detail, 0)
|
||||
c := assign.BackupCondition(node, module, want.FP, have.Detail, true)
|
||||
if p.say != nil && !p.said[c.Key] {
|
||||
if err := p.say(c); err != nil {
|
||||
p.logf("the condition of the failed backup of %s on %s could not be said: %v", module, node, err)
|
||||
} else {
|
||||
p.said[c.Key] = true
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// backupAsk writes the backup ask at its asking, unless that asking stands already.
|
||||
func (p *passer) backupAsk(ctx context.Context, key, node, module, fp, words string, asking int) {
|
||||
if p.w.has(delivery.AssignerAsksBucket, key) {
|
||||
if asking == 0 {
|
||||
return
|
||||
}
|
||||
if a, err := delivery.DecodeAssignerAsk(key, p.w.read(delivery.AssignerAsksBucket)[key]); err == nil && a.Ask.ID == delivery.AskingID(key, asking) {
|
||||
return
|
||||
}
|
||||
}
|
||||
a := assign.BackupAsk(key, node, module, fp, words, asking, time.Now())
|
||||
p.put(ctx, delivery.AssignerAsksBucket, key, func() ([]byte, error) { return delivery.EncodeAssignerAsk(key, a) })
|
||||
}
|
||||
|
||||
@@ -374,3 +374,15 @@ func (w *world) nodeComposed() map[string]delivery.ComposedDigest {
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// answer is the teller's answer to the ask at key (mesh-delivery-teller_answers, the same key), and whether there is one.
|
||||
func (w *world) answer(key string) (delivery.Answer, bool) {
|
||||
w.mu.Lock()
|
||||
value, ok := w.values[delivery.AnswersBucket][key]
|
||||
w.mu.Unlock()
|
||||
if !ok {
|
||||
return delivery.Answer{}, false
|
||||
}
|
||||
a, err := delivery.DecodeAnswer(value)
|
||||
return a, err == nil
|
||||
}
|
||||
|
||||
@@ -10,17 +10,16 @@ import (
|
||||
)
|
||||
|
||||
// HoldAsk is the ask about a module's fingerprint the assigner holds, at its key (delivery.HoldAskKey), first asking: keep
|
||||
// it held or release it. Both options acknowledge and bind nothing: in release 5 the wants are only compared, so no answer
|
||||
// changes what a node runs (ADR 0429 §7); the node agent acts on the hold from release 6.
|
||||
// it held or release it. Releasing writes the held fingerprint's want again by the rules (ADR 0539 §19).
|
||||
func HoldAsk(key, module, fp string, now time.Time) delivery.AssignerAsk {
|
||||
return delivery.AssignerAsk{Ask: asks.Ask{
|
||||
ID: delivery.AskingID(key, 0),
|
||||
Headline: fmt.Sprintf("%s is held at %s", module, short(fp)),
|
||||
Explanation: fmt.Sprintf("A node's health check of %s failed at %s, so the assigner holds that version and wants "+
|
||||
"no node to move to it. Until release 6 this changes nothing a node runs: the controller still sends.", module, short(fp)),
|
||||
"no node to move to it. Releasing it lets the wants move to it again.", module, short(fp)),
|
||||
Options: []asks.Option{
|
||||
{ID: "keep", Label: "Keep it held", Does: "the version stays held", Level: asks.Acknowledge},
|
||||
{ID: "release", Label: "Release it", Does: "the version is released when wants are applied, from release 6", Level: asks.Acknowledge},
|
||||
{ID: "release", Label: "Release it", Does: "the version is released and the wants move to it by the rules again", Level: asks.Acknowledge},
|
||||
},
|
||||
Who: asks.Operator,
|
||||
Expires: now.Add(7 * 24 * time.Hour),
|
||||
@@ -77,3 +76,71 @@ func HeldCondition(module, fp, node, words string, raised, needsYou bool) asks.C
|
||||
}
|
||||
return c
|
||||
}
|
||||
|
||||
// The options of the record ask (ADR 0539 §14) and of the backup ask (§16).
|
||||
const (
|
||||
OptionYes = "yes"
|
||||
OptionNo = "no"
|
||||
OptionBackupAgain = delivery.BackupAgain
|
||||
OptionBackupSkip = delivery.BackupSkip
|
||||
)
|
||||
|
||||
// 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) delivery.AssignerAsk {
|
||||
what := map[delivery.Judgement]string{
|
||||
delivery.JudgedRecreates: "recreates a container",
|
||||
delivery.JudgedMovesData: "moves the module's data",
|
||||
delivery.JudgedCannot: "cannot be judged: " + need.Words,
|
||||
}[need.Judgement]
|
||||
return delivery.AssignerAsk{Ask: asks.Ask{
|
||||
ID: delivery.AskingID(key, 0),
|
||||
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),
|
||||
Options: []asks.Option{
|
||||
{ID: OptionYes, Label: "Yes, move it", Does: "the node is told to apply the new composition of " + module, Level: asks.Acknowledge},
|
||||
{ID: OptionNo, Label: "No, not now", Does: "the node keeps what it runs", Level: asks.Acknowledge},
|
||||
},
|
||||
Who: asks.Operator,
|
||||
Expires: now.Add(7 * 24 * time.Hour),
|
||||
OnExpiry: "the node keeps what it runs",
|
||||
About: key,
|
||||
}}
|
||||
}
|
||||
|
||||
// BackupAsk is the ask about a backup of a module on node that failed before an apply (ADR 0539 §16): back up again, or go on
|
||||
// without a backup. asking is the number of the asking (0 the first, 2 the second: delivery.AskingID), so a backup that fails again after the answer again is asked
|
||||
// again. words are the backup's own words, from the node's have.
|
||||
func BackupAsk(key, node, module, fp, words string, asking int, now time.Time) delivery.AssignerAsk {
|
||||
return delivery.AssignerAsk{Ask: asks.Ask{
|
||||
ID: delivery.AskingID(key, asking),
|
||||
Headline: fmt.Sprintf("The backup of %s failed on %s", module, node),
|
||||
Explanation: fmt.Sprintf("The backup before applying %s at %s on %s failed: %s. %s keeps running what it has until you answer.",
|
||||
module, short(fp), node, words, module),
|
||||
Options: []asks.Option{
|
||||
{ID: OptionBackupAgain, Label: "Back up again", Does: "the node tries the backup again before it applies", Level: asks.Acknowledge},
|
||||
{ID: OptionBackupSkip, Label: "Go on without a backup", Does: "the node applies the new version without a backup", Level: asks.Acknowledge},
|
||||
},
|
||||
Who: asks.Operator,
|
||||
Expires: now.Add(7 * 24 * time.Hour),
|
||||
OnExpiry: "the module keeps running what it has",
|
||||
About: key,
|
||||
}}
|
||||
}
|
||||
|
||||
// BackupCondition is the condition raised when a backup before an apply fails and the module is held on the node (ADR 0539 §16),
|
||||
// naming the node, the module and the backup's words; cleared under the same key.
|
||||
func BackupCondition(node, module, fp, words string, raised bool) asks.Condition {
|
||||
key := delivery.AssignerModule + ".backup." + node + "." + module + "." + strings.TrimPrefix(fp, delivery.SourcePrefix2)
|
||||
subject := asks.Scope{Scope: "module", ID: module, Machine: node}
|
||||
if !raised {
|
||||
return asks.Condition{Key: key, Subject: subject, State: asks.StateCleared}
|
||||
}
|
||||
headline := fmt.Sprintf("The backup of %s failed on %s", module, node)
|
||||
needs := "answer whether to back up again or go on without a backup"
|
||||
return asks.Condition{Key: key, Severity: asks.SeverityWarning, Subject: subject, State: asks.StateRaised,
|
||||
Summary: headline, Headline: headline, Needs: needs,
|
||||
Explanation: asks.NeedsYou + needs + ". The backup before applying " + module + " at " + short(fp) + " on " + node + " failed: " + words + "."}
|
||||
}
|
||||
|
||||
@@ -73,3 +73,43 @@ func TestTheHeldConditionNamesModuleNodeAndWords(t *testing.T) {
|
||||
t.Errorf("cleared = %+v", got)
|
||||
}
|
||||
}
|
||||
|
||||
// 1906/D4: the record ask and the backup ask are valid asks of the router and of mesh-sdk's layout at their keys, answerable in
|
||||
// the view, and name the move and the node.
|
||||
func TestTheRecordAndBackupAsksAreValid(t *testing.T) {
|
||||
now := time.Now()
|
||||
rkey, err := delivery.RecordAskKey("ace", "mail", "sha256:"+strings.Repeat("b", 64))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
r := RecordAsk(rkey, "ace", "mail", fNew, RecordNeed{Node: "ace", Digest: "sha256:" + strings.Repeat("b", 64), Judgement: delivery.JudgedCannot, Words: "the composition it runs is gone"}, now)
|
||||
bkey, err := delivery.BackupAskKey("ace", "mail", fNew)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
b := BackupAsk(bkey, "ace", "mail", fNew, "no space left", 0, now)
|
||||
for key, a := range map[string]delivery.AssignerAsk{rkey: r, bkey: b} {
|
||||
if err := a.Ask.Check(now); err != nil {
|
||||
t.Errorf("%s: the router refuses the ask: %v", key, err)
|
||||
}
|
||||
if _, err := delivery.EncodeAssignerAsk(key, a); err != nil {
|
||||
t.Errorf("%s: cannot be written: %v", key, err)
|
||||
}
|
||||
if _, err := delivery.ViewAnswerKey(key); err != nil {
|
||||
t.Errorf("%s: the view may not answer it: %v", key, err)
|
||||
}
|
||||
}
|
||||
if !strings.Contains(r.Ask.Explanation, "gone") || r.Ask.Options[0].ID != OptionYes {
|
||||
t.Errorf("the record ask = %+v", r.Ask)
|
||||
}
|
||||
if !strings.Contains(b.Ask.Explanation, "no space left") || b.Ask.Options[0].ID != delivery.BackupAgain || b.Ask.Options[1].ID != delivery.BackupSkip {
|
||||
t.Errorf("the backup ask = %+v", b.Ask)
|
||||
}
|
||||
if again := BackupAsk(bkey, "ace", "mail", fNew, "x", 2, now); again.Ask.ID == b.Ask.ID {
|
||||
t.Errorf("asked again under the same id")
|
||||
}
|
||||
c := BackupCondition("ace", "mail", fNew, "no space left", true)
|
||||
if _, err := asks.EncodeCondition(c); err != nil {
|
||||
t.Errorf("the backup condition is refused: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -23,6 +23,14 @@ type Module struct {
|
||||
// Composed answers the controller's composition of the module for a node at a fingerprint (mesh-controller_composed): its
|
||||
// digest and the judgement of the move to it, and whether it exists. Nil: none exists (ADR 0539 §1, §6).
|
||||
Composed func(node, fp string) (delivery.ComposedDigest, bool)
|
||||
// Allowed says whether the operator answered yes to the record ask about the move of this module on node to the composition
|
||||
// with this digest (ADR 0539 §14). Nil: none did.
|
||||
Allowed func(node, digest string) bool
|
||||
// Released is a hold on FP the operator released (ADR 0539 §19): FP is not held again by a health check that failed.
|
||||
Released bool
|
||||
// Restored says whether the operator answered "restore" to the restore ask of this data module on node: it is then set back
|
||||
// as any module is, losing the writes made since (ADR 0539 §19). Nil: none did.
|
||||
Restored func(node string) bool
|
||||
}
|
||||
|
||||
func (m Module) composition(node, fp string) (delivery.ComposedDigest, bool) {
|
||||
@@ -130,6 +138,8 @@ type Outcome struct {
|
||||
Wants map[string]delivery.Want
|
||||
// Held is the hold of m.FP to write, when a node's health check failed on it and it is not held yet.
|
||||
Held *delivery.Held
|
||||
// Record is the moves a person must allow before the want names them: each is asked once (ADR 0539 §14).
|
||||
Record []RecordNeed
|
||||
// Restore is the nodes of a module holding data whose held fingerprint failed its health check there and that are not set
|
||||
// back: the operator is asked whether to restore each (ADR 0539 §19).
|
||||
Restore []string
|
||||
@@ -137,6 +147,15 @@ type Outcome struct {
|
||||
Rules map[string]Holding
|
||||
}
|
||||
|
||||
// RecordNeed is a move of a module of policy record on a node that waits for the operator's yes: the composition's digest, the
|
||||
// controller's judgement of the move and its words.
|
||||
type RecordNeed struct {
|
||||
Node string
|
||||
Digest string
|
||||
Judgement delivery.Judgement
|
||||
Words string
|
||||
}
|
||||
|
||||
// Holding is a rule holding a node back, and why.
|
||||
type Holding struct {
|
||||
Rule int
|
||||
@@ -151,7 +170,7 @@ type Holding struct {
|
||||
func Wants(m Module, n Nodes, wants map[string]delivery.Want, lastHealthy map[string]string) Outcome {
|
||||
out := Outcome{Wants: map[string]delivery.Want{}, Rules: map[string]Holding{}}
|
||||
held := m.Held
|
||||
if !held && m.FP != "" {
|
||||
if !held && !m.Released && m.FP != "" {
|
||||
for _, node := range slices.Sorted(slices.Values(m.Nodes)) {
|
||||
if h, ok := n.Haves[node][m.Name]; ok && h.FP == m.FP && h.Health == delivery.HealthFailed {
|
||||
out.Held = &delivery.Held{Node: node, Words: h.Detail}
|
||||
@@ -172,11 +191,20 @@ func Wants(m Module, n Nodes, wants map[string]delivery.Want, lastHealthy map[st
|
||||
case ok:
|
||||
d, _ := m.composition(node, m.FP)
|
||||
next = delivery.Want{FP: m.FP, Composed: d.Digest}
|
||||
if had && !before.Removed && before.FP == m.FP && before.Composed == d.Digest {
|
||||
next.Backup = before.Backup // the answer to a failed backup stays while its target stays (ADR 0539 §16)
|
||||
}
|
||||
if need, wait := m.mustAsk(node, before, had, d); wait {
|
||||
if need != nil {
|
||||
out.Record = append(out.Record, *need)
|
||||
}
|
||||
continue
|
||||
}
|
||||
case held && had && before.FP == m.FP:
|
||||
out.Rules[node] = Holding{Rule: rule, Words: words}
|
||||
last := lastHealthy[node]
|
||||
switch {
|
||||
case keepsData(m.Manifest):
|
||||
case keepsData(m.Manifest) && (m.Restored == nil || !m.Restored(node)):
|
||||
next = before
|
||||
next.Words = fmt.Sprintf("%s holds data, so it is not set back from %s: the operator is asked whether to restore it", m.Name, short(m.FP))
|
||||
out.Restore = append(out.Restore, node)
|
||||
@@ -227,3 +255,23 @@ func short(fp string) string {
|
||||
}
|
||||
return part[:12]
|
||||
}
|
||||
|
||||
// mustAsk says whether the move to the composition d on node waits: for the operator's yes (need names the ask) or, for the bus,
|
||||
// for the operator's hand (ADR 0539 §14, §15). A module the node never ran, a move judged neither, a module that is not of
|
||||
// policy record and a want that stays are written as any want; so is a revision of the bus module's composition at the
|
||||
// same fingerprint judged neither (a user list only, §7).
|
||||
func (m Module) mustAsk(node string, before delivery.Want, had bool, d delivery.ComposedDigest) (*RecordNeed, bool) {
|
||||
if !had || before.Removed || (before.FP == m.FP && before.Composed == d.Digest) {
|
||||
return nil, false
|
||||
}
|
||||
if catalogue.ProvidesBus(m.Manifest) && before.FP != m.FP {
|
||||
return nil, true // the bus is never moved by the assigner; the operator writes its want (§15)
|
||||
}
|
||||
if policy, _, _ := catalogue.DerivedUpgrade(m.Manifest); policy != catalogue.PolicyRecord || d.Judgement == delivery.JudgedNeither {
|
||||
return nil, false
|
||||
}
|
||||
if m.Allowed != nil && m.Allowed(node, d.Digest) {
|
||||
return nil, false
|
||||
}
|
||||
return &RecordNeed{Node: node, Digest: d.Digest, Judgement: d.Judgement, Words: d.Words}, true
|
||||
}
|
||||
|
||||
@@ -241,3 +241,130 @@ func TestASetBackNamesTheOlderCompositionOrStays(t *testing.T) {
|
||||
t.Errorf("set-back want = %+v", w)
|
||||
}
|
||||
}
|
||||
|
||||
// movingTo is a module ready to move, whose composition at F is judged j against the one it runs, and whose digest is dig(F).
|
||||
func movingTo(j delivery.Judgement) (Module, Nodes, map[string]delivery.Want) {
|
||||
m, n := ready()
|
||||
m.Composed = func(node, fp string) (delivery.ComposedDigest, bool) {
|
||||
d := delivery.ComposedDigest{Digest: dig(fp), Judgement: j}
|
||||
if j != delivery.JudgedNeither {
|
||||
d.Against, d.Words = dig(fOld), "judged"
|
||||
}
|
||||
return d, true
|
||||
}
|
||||
return m, n, wantedOld()
|
||||
}
|
||||
|
||||
func recording(m *Module) {
|
||||
m.Manifest.Upgrade = &catalogue.UpgradePolicy{Policy: catalogue.PolicyRecord, Why: "a person takes each build"}
|
||||
}
|
||||
|
||||
// 1906/D4 (ADR 0539 §14): a module of policy record whose move is judged to recreate a container, to move data, or cannot be
|
||||
// judged, is not written until its ask is answered yes; the same at a new revision of the same fingerprint.
|
||||
func TestARecordModuleWaitsForItsAskWhenTheMoveIsNotJudgedNeither(t *testing.T) {
|
||||
for _, j := range []delivery.Judgement{delivery.JudgedRecreates, delivery.JudgedMovesData, delivery.JudgedCannot} {
|
||||
m, n, wanted := movingTo(j)
|
||||
recording(&m)
|
||||
out := Wants(m, n, wanted, nil)
|
||||
if w, ok := out.Wants["ace"]; ok {
|
||||
t.Errorf("%s: ace's want was written without the ask's yes: %+v", j, w)
|
||||
}
|
||||
var ace *RecordNeed
|
||||
for i := range out.Record {
|
||||
if out.Record[i].Node == "ace" {
|
||||
ace = &out.Record[i]
|
||||
}
|
||||
}
|
||||
if ace == nil || ace.Digest != dig(fNew) || ace.Judgement != j {
|
||||
t.Errorf("%s: the asks = %+v, want one for ace naming the digest and the judgement", j, out.Record)
|
||||
}
|
||||
// a new revision at the same fingerprint is asked too
|
||||
same := map[string]delivery.Want{"ace": {FP: fNew, Composed: dig(fOld)}}
|
||||
if out := Wants(m, n, same, nil); len(out.Record) == 0 || len(out.Wants) != 0 && out.Wants["ace"].Composed == dig(fNew) {
|
||||
t.Errorf("%s: a new revision at the same fingerprint: %+v", j, out)
|
||||
}
|
||||
m.Allowed = func(node, digest string) bool { return node == "ace" && digest == dig(fNew) }
|
||||
out = Wants(m, n, wanted, nil)
|
||||
if w := out.Wants["ace"]; w.FP != fNew || w.Composed != dig(fNew) {
|
||||
t.Errorf("%s: after the yes ace's want = %+v", j, w)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 1906/D4: a move judged neither is written at once, whatever the policy; so is a module the node never ran; and a module
|
||||
// that rolls (not record) is never asked.
|
||||
func TestAMoveJudgedNeitherOrNeverRunOrNotRecordedIsWrittenAtOnce(t *testing.T) {
|
||||
m, n, wanted := movingTo(delivery.JudgedNeither)
|
||||
recording(&m)
|
||||
if out := Wants(m, n, wanted, nil); out.Wants["ace"].FP != fNew || len(out.Record) != 0 {
|
||||
t.Errorf("neither: %+v", out)
|
||||
}
|
||||
m, n, _ = movingTo(delivery.JudgedRecreates)
|
||||
recording(&m)
|
||||
if out := Wants(m, n, nil, nil); out.Wants["ace"].FP != fNew || len(out.Record) != 0 {
|
||||
t.Errorf("a node that never ran it: %+v", out)
|
||||
}
|
||||
m, n, wanted = movingTo(delivery.JudgedRecreates)
|
||||
if out := Wants(m, n, wanted, nil); out.Wants["ace"].FP != fNew || len(out.Record) != 0 {
|
||||
t.Errorf("a rolled module: %+v", out)
|
||||
}
|
||||
}
|
||||
|
||||
// 1906/D4 (ADR 0539 §7, §15): the bus module's new fingerprint is never written by the assigner, but a revision of its
|
||||
// composition at the same fingerprint that only changes the user list (judged neither) is written at once.
|
||||
func TestTheBusIsNotMovedButItsUserListRevisionIsWrittenAtOnce(t *testing.T) {
|
||||
m, n, _ := movingTo(delivery.JudgedNeither)
|
||||
m.Manifest.Provides = []catalogue.Offer{{Name: "mesh-bus"}}
|
||||
if out := Wants(m, n, wantedOld(), nil); out.Wants["ace"].FP == fNew {
|
||||
t.Errorf("the bus moved to a new fingerprint: %+v", out.Wants["ace"])
|
||||
}
|
||||
running := map[string]delivery.Want{"ace": {FP: fNew, Composed: dig(fOld)}, "g14": {FP: fNew, Composed: dig(fOld)}}
|
||||
out := Wants(m, n, running, nil)
|
||||
if w := out.Wants["ace"]; w.FP != fNew || w.Composed != dig(fNew) {
|
||||
t.Errorf("a user-list revision of the bus: ace's want = %+v", w)
|
||||
}
|
||||
if len(out.Record) != 0 {
|
||||
t.Errorf("a user-list revision was asked: %+v", out.Record)
|
||||
}
|
||||
}
|
||||
|
||||
// 1906/D4 (ADR 0539 §16): the answer to a failed backup stays in the want while its target stays.
|
||||
func TestTheBackupAnswerStaysInTheWantWhileItsTargetStays(t *testing.T) {
|
||||
m, n := ready()
|
||||
running := map[string]delivery.Want{"ace": {FP: fNew, Composed: dig(fNew), Backup: delivery.BackupSkip}}
|
||||
if w, ok := Wants(m, n, running, nil).Wants["ace"]; ok && w.Backup == "" {
|
||||
t.Errorf("the answer was erased: %+v", w)
|
||||
}
|
||||
}
|
||||
|
||||
// 1906/D4 (ADR 0539 §19): a hold released by the operator is not held again by a health check that failed before, and the
|
||||
// wants move by the rules again.
|
||||
func TestAReleasedHoldIsNotHeldAgain(t *testing.T) {
|
||||
m, n := ready()
|
||||
n.Haves["ace"]["mail"] = delivery.Have{FP: fNew, Phase: delivery.PhaseRunning, Health: delivery.HealthFailed, Detail: "x"}
|
||||
m.Released = true
|
||||
out := Wants(m, n, map[string]delivery.Want{"ace": {FP: fOld, Composed: dig(fOld), Newest: fNew, Rule: delivery.RuleNotHeld}}, map[string]string{"ace": fOld})
|
||||
if out.Held != nil {
|
||||
t.Errorf("a released fingerprint was held again: %+v", out.Held)
|
||||
}
|
||||
if w := out.Wants["ace"]; w.FP != fNew || w.Composed != dig(fNew) || w.Rule != 0 {
|
||||
t.Errorf("ace's want = %+v, want the held fingerprint written again", w)
|
||||
}
|
||||
}
|
||||
|
||||
// 1906/D4: a data module whose restore ask the operator answered "restore" is set back like any other, losing the newer writes;
|
||||
// answered "leave stopped" (or not answered) it stays.
|
||||
func TestARestoreAnswerSetsADataModuleBack(t *testing.T) {
|
||||
m, n := ready()
|
||||
m.Manifest.Data = &catalogue.Data{Own: []catalogue.DataItem{{ID: "mail", Class: catalogue.ClassIrreplaceable}}}
|
||||
n.Haves["ace"]["mail"] = delivery.Have{FP: fNew, Phase: delivery.PhaseRunning, Health: delivery.HealthFailed}
|
||||
running := map[string]delivery.Want{"ace": {FP: fNew, Composed: dig(fNew)}}
|
||||
m.Restored = func(node string) bool { return false }
|
||||
if w, ok := Wants(m, n, running, map[string]string{"ace": fOld}).Wants["ace"]; ok && w.FP == fOld {
|
||||
t.Errorf("set back without the operator's restore: %+v", w)
|
||||
}
|
||||
m.Restored = func(node string) bool { return node == "ace" }
|
||||
if w := Wants(m, n, running, map[string]string{"ace": fOld}).Wants["ace"]; w.FP != fOld || w.Composed != dig(fOld) {
|
||||
t.Errorf("after restore ace's want = %+v", w)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -34,7 +34,7 @@ func TestManifest(t *testing.T) {
|
||||
slices.Sort(reads)
|
||||
want := []string{"mesh-controller.assignments", "mesh-controller.composed", "mesh-controller.grants",
|
||||
"mesh-delivery-builder.builds", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.manifests",
|
||||
"mesh-delivery-indexer.modules", "mesh-delivery-watchman.presence", "node-engine.have"}
|
||||
"mesh-delivery-indexer.modules", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "node-engine.have"}
|
||||
if !reflect.DeepEqual(reads, want) {
|
||||
t.Errorf("reads = %v, want exactly %v", reads, want)
|
||||
}
|
||||
|
||||
@@ -10,6 +10,7 @@
|
||||
"mesh-delivery-indexer.candidates",
|
||||
"mesh-delivery-indexer.manifests",
|
||||
"mesh-delivery-indexer.modules",
|
||||
"mesh-delivery-teller.answers",
|
||||
"mesh-delivery-watchman.presence",
|
||||
"node-engine.have"
|
||||
],
|
||||
|
||||
Reference in New Issue
Block a user