diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go index 7891870..9ab48e5 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go @@ -341,12 +341,14 @@ func TestAHeldFingerprintIsAskedAbout(t *testing.T) { b.putJSON(t, provider.AssignmentsBucket, "ace.web", []byte(`{"node":"ace","module":"web","build":"b-1"}`)) b.start(t) hkey, _ := delivery.HeldKey("web", fNew) - if h, err := delivery.DecodeHeld(b.get(t, delivery.AssignerWantsBucket, hkey)); err != nil || h.Node != "ace" { + // 1906/H3: the held record names the machine and keeps the health, when and the node's words. + if h, err := delivery.DecodeHeld(b.get(t, delivery.AssignerWantsBucket, hkey)); err != nil || h.Node != "ace" || + !strings.HasPrefix(h.Words, "health failed, phase running, seen at ") || !strings.HasSuffix(h.Words, ": connection refused") { t.Fatalf("held = %+v, %v", h, err) } akey, _ := delivery.HoldAskKey("web", fNew) a, err := delivery.DecodeAssignerAsk(akey, b.get(t, delivery.AssignerAsksBucket, akey)) - if err != nil || a.Ask.ID != delivery.AskingID(akey, 0) || len(a.Ask.Options) != 2 { + if err != nil || a.Asking <= 0 || a.Ask.ID != delivery.AskingID(akey, a.Asking) || len(a.Ask.Options) != 2 { t.Fatalf("the ask = %+v, %v", a, err) } } @@ -580,8 +582,15 @@ func (b *bed) wantNow(t *testing.T, node, module, what string, ok func(delivery. 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) { b.answerTo(t, key, option, "r1") } +// answer says the teller wrote the operator's answer to the ask that stands at key (its warrant names that asking's id). +func (b *bed) answer(t *testing.T, key, option string) { + t.Helper() + a, err := delivery.DecodeAssignerAsk(key, b.get(t, delivery.AssignerAsksBucket, key)) + if err != nil { + t.Fatal(err) + } + b.answerTo(t, key, option, a.Ask.ID) +} // answerTo is answer to the ask with this id (the router's warrant names it). func (b *bed) answerTo(t *testing.T, key, option, askID string) { @@ -773,3 +782,81 @@ func TestAUserListRevisionOfTheBusIsWrittenAtOnce(t *testing.T) { t.Errorf("the user-list revision was asked") } } + +// 1906/H2, whole on the bus: given a fingerprint held after one failed check, when every node that runs the module then runs +// it healthy, the assigner lifts the hold by itself: the held record and its ask are taken away, the held condition is +// cleared saying why, and the want stays at the fingerprint. +func TestAHoldIsLiftedOnceEveryNodeRunsItHealthy(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) + 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) + hkey, _ := delivery.HeldKey("web", fNew) + akey, _ := delivery.HoldAskKey("web", fNew) + b.get(t, delivery.AssignerWantsBucket, hkey) + b.get(t, delivery.AssignerAsksBucket, akey) + healthy, _ := delivery.EncodeHave(delivery.Have{FP: fNew, Phase: delivery.PhaseRunning, Health: delivery.HealthOK}) + b.putJSON(t, delivery.HaveBucket, "ace.web", healthy) + for deadline := time.Now().Add(15 * time.Second); time.Now().Before(deadline); time.Sleep(100 * time.Millisecond) { + if b.hasKey(delivery.AssignerWantsBucket, hkey) || b.hasKey(delivery.AssignerAsksBucket, akey) { + continue + } + for _, c := range b.conditionsSaid(t) { + if c.Subject.ID == "web" && c.State == asks.StateCleared && strings.Contains(c.Resolved, "lifted its hold by itself") { + if w := b.wantNow(t, "ace", "web", "the want at the lifted fingerprint", func(w delivery.Want) bool { return w.FP == fNew }); w.Rule != 0 { + t.Errorf("the want = %+v", w) + } + return + } + } + } + t.Fatalf("the hold was not lifted: held %v, ask %v", b.hasKey(delivery.AssignerWantsBucket, hkey), b.hasKey(delivery.AssignerAsksBucket, akey)) +} + +// 1906/H2, whole on the bus: a hold lifted by itself and written again after the assigner was started again is asked under a +// new id, so the teller raises it; the answer to the first hold does not carry over (1906/H4). +func TestAHoldWrittenAgainAfterALiftAndARestartIsAskedAgain(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) + failed, _ := delivery.EncodeHave(delivery.Have{FP: fNew, Phase: delivery.PhaseRunning, Health: delivery.HealthFailed, Detail: "connection refused"}) + healthy, _ := delivery.EncodeHave(delivery.Have{FP: fNew, Phase: delivery.PhaseRunning, Health: delivery.HealthOK}) + b.putJSON(t, delivery.HaveBucket, "ace.web", failed) + b.putWant(t, "ace", "web", delivery.Want{FP: fNew, Composed: "sha256:" + strings.Repeat("b", 64)}) + _, stop := b.startStoppable(t) + hkey, _ := delivery.HeldKey("web", fNew) + akey, _ := delivery.HoldAskKey("web", fNew) + first, err := delivery.DecodeAssignerAsk(akey, b.get(t, delivery.AssignerAsksBucket, akey)) + if err != nil { + t.Fatal(err) + } + b.putJSON(t, delivery.HaveBucket, "ace.web", healthy) + for deadline := time.Now().Add(15 * time.Second); b.hasKey(delivery.AssignerWantsBucket, hkey); time.Sleep(100 * time.Millisecond) { + if time.Now().After(deadline) { + t.Fatal("the hold was not lifted") + } + } + stop() + time.Sleep(1100 * time.Millisecond) // the next hold is written in another second + b.answerTo(t, akey, "keep", first.Ask.ID) // an answer to the first hold, which the later hold must not take + b.putJSON(t, delivery.HaveBucket, "ace.web", failed) + b.start(t) + again, err := delivery.DecodeAssignerAsk(akey, b.get(t, delivery.AssignerAsksBucket, akey)) + if err != nil { + t.Fatal(err) + } + if again.Ask.ID == first.Ask.ID { + t.Fatalf("the hold written again is asked under the first id %s", first.Ask.ID) + } + b.putJSON(t, delivery.HaveBucket, "ace.web", healthy) + for deadline := time.Now().Add(15 * time.Second); b.hasKey(delivery.AssignerWantsBucket, hkey); time.Sleep(100 * time.Millisecond) { + if time.Now().After(deadline) { + t.Fatal("the stale \"keep\" of the first hold kept the second one held") + } + } +} 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 c736c38..0beb8f0 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go @@ -28,6 +28,15 @@ type passer struct { // answeredAt is, by the id of a backup ask, how many values the have had when its answer was written into the want. answeredAt map[string]uint64 asks *composeAsks + // now is the clock; nil is time.Now. + now func() time.Time +} + +func (p *passer) clock() time.Time { + if p.now != nil { + return p.now() + } + return time.Now() } func (p *passer) pass(ctx context.Context) { @@ -88,7 +97,7 @@ func (p *passer) wants(ctx context.Context) { for name := range names { fp := trunk[name].FP2 name := name - m := assign.Module{Name: name, FP: fp, Manifest: manifests[name], Nodes: nodesOf[name], + m := assign.Module{Name: name, FP: fp, Manifest: manifests[name], Nodes: nodesOf[name], Now: time.Now(), 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) @@ -103,6 +112,7 @@ func (p *passer) wants(ctx context.Context) { grant, _ := delivery.ControllerGrantKey(name, fp) hold, _ := delivery.HoldAskKey(name, fp) m.Released = p.answeredWith(hold, "release") + m.Kept = p.answeredWith(hold, "keep") 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) @@ -120,8 +130,11 @@ func (p *passer) wants(ctx context.Context) { p.put(ctx, delivery.AssignerWantsBucket, key, func() ([]byte, error) { return delivery.EncodeHeld(held) }) } } - if out.Held != nil || m.Held { - p.holdAsk(ctx, name, fp) + if out.Lifted { + p.lift(ctx, name, fp) + } + if out.Held != nil || (m.Held && !out.Lifted) { + p.holdAsk(ctx, name, fp, out.Held) p.heldCondition(ctx, name, fp, out.Held, len(out.Restore) > 0) } for _, node := range out.Restore { @@ -210,15 +223,62 @@ func (p *passer) put(ctx context.Context, bucket, key string, encode func() ([]b // holdAsk writes, once, the ask about a held fingerprint (ADR 0528 clause 6, ADR 0437 clause 10): the first ask that needs // no proof of who answered, which the view may answer and the teller fills. In release 5 an answer changes nothing on a // node (wants are only compared, ADR 0429 §7), and the options say so. -func (p *passer) holdAsk(ctx context.Context, module, fp string) { +func (p *passer) holdAsk(ctx context.Context, module, fp string, fresh *delivery.Held) { key, err := delivery.HoldAskKey(module, fp) if err != nil || p.w.has(delivery.AssignerAsksBucket, key) { return } - a := assign.HoldAsk(key, module, fp, time.Now()) + // The asking is the hold's time in seconds: a hold of the same fingerprint written again after a lift is another asking, under + // an id of its own, also after a restart, since the router drops a second ask under an id it keeps (issue 1906). + now := p.clock() + asking := int(now.Unix()) + h := p.heldHold(module, fp, fresh) + a := assign.HoldAsk(key, module, fp, h.Node, h.Words, asking, now) p.put(ctx, delivery.AssignerAsksBucket, key, func() ([]byte, error) { return delivery.EncodeAssignerAsk(key, a) }) } +// lift takes away a hold the assigner lifted by itself, since every node that runs the module runs the held fingerprint healthy +// (issue 1906): the held record and its ask (the teller then takes the ask back from the operator), and the held condition is +// cleared saying why, once. A hold of the same fingerprint written later is said and asked again. +func (p *passer) lift(ctx context.Context, module, fp string) { + held, err := delivery.HeldKey(module, fp) + if err != nil { + return + } + ask, err := delivery.HoldAskKey(module, fp) + if err != nil { + return + } + c := assign.LiftedCondition(module, fp) + if !p.said[c.Key+".lifted"] { + if p.say != nil { + if err := p.say(c); err != nil { + // The hold stays written, so the next pass lifts it again and says so then. + p.logf("the condition of the lifted hold of %s could not be cleared, and is cleared at the next pass: %v", module, err) + return + } + } + p.logf("%s", assign.LiftedWords(module, fp)) + p.said[c.Key+".lifted"] = true + delete(p.said, c.Key+".raised") // a hold written again is said again + } + p.remove(ctx, delivery.AssignerWantsBucket, held) + p.remove(ctx, delivery.AssignerAsksBucket, ask) +} + +// remove deletes one key of the assigner's own bucket when it holds it; a delete that fails is said and made again at the next +// pass. +func (p *passer) remove(ctx context.Context, bucket, key string) { + if !p.w.has(bucket, key) { + return + } + if err := p.kvs[bucket].Delete(ctx, key); err != nil && !errors.Is(err, context.Canceled) { + p.logf("%s %s could not be taken away: %v", bucket, key, err) + return + } + p.w.put(bucket, key, nil) +} + // heldHold is the hold of fp as written: the node whose health check failed and its words. func (p *passer) heldHold(module, fp string, fresh *delivery.Held) delivery.Held { if fresh != nil { @@ -245,6 +305,7 @@ func (p *passer) heldCondition(ctx context.Context, module, fp string, fresh *de return } p.said[c.Key+".raised"] = true + delete(p.said, c.Key+".lifted") // a hold raised again and lifted again is said lifted again } // restoreAsk writes, once, the ask about a module holding data whose held fingerprint failed its health check on node. @@ -258,11 +319,21 @@ func (p *passer) restoreAsk(ctx context.Context, module, node, fp string) { 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). +// answeredWith says whether the teller wrote an answer to the ask at key as it stands now 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); it counts only +// for the asking it answered (its warrant names that asking's id), so an answer to an ask taken away, or to an earlier hold of +// the same fingerprint, does not carry over to a later one (issue 1906). func (p *passer) answeredWith(key, option string) bool { a, ok := p.w.answer(key) - return ok && a.Option == option + if !ok || a.Option != option { + return false + } + ask, err := delivery.DecodeAssignerAsk(key, p.w.value(delivery.AssignerAsksBucket, key)) + if err != nil { + return false + } + id, _, _ := strings.Cut(a.Warrant, "@") + return id == ask.Ask.ID } // clearHeld clears, once, the condition of a hold the operator released. 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 new file mode 100644 index 0000000..8b48beb --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go @@ -0,0 +1,100 @@ +package main + +import ( + "context" + "strings" + "testing" + "time" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/delivery" + + "mesh-delivery-assigner/internal/assign" +) + +// memKV is the part of a bucket a pass writes: puts and deletes. +type memKV struct { + jetstream.KeyValue + values map[string][]byte +} + +func (m *memKV) Put(_ context.Context, key string, value []byte) (uint64, error) { + m.values[key] = value + return uint64(len(m.values)), nil +} + +func (m *memKV) Delete(_ context.Context, key string, _ ...jetstream.KVDeleteOpt) error { + delete(m.values, key) + return nil +} + +// newPasser is a pass on a fresh world, as after a start, writing to the buckets given, at the clock given. +func newPasser(kvs buckets, at time.Time) *passer { + return &passer{kvs: kvs, w: newWorld(), logf: func(string, ...any) {}, said: map[string]bool{}, answeredAt: map[string]uint64{}, + now: func() time.Time { return at }} +} + +var fpTest = delivery.SourcePrefix2 + strings.Repeat("b", 64) + +// 1906/H2: given a hold lifted by itself, when the assigner is started again and the same fingerprint is held again, then the +// hold ask is written under another asking, with an id of its own, so the teller raises it (the router drops a second ask under +// an id it keeps). +func TestAHoldAskedAgainAfterALiftAndARestartHasANewID(t *testing.T) { + asksKV, wantsKV := &memKV{values: map[string][]byte{}}, &memKV{values: map[string][]byte{}} + kvs := buckets{delivery.AssignerAsksBucket: asksKV, delivery.AssignerWantsBucket: wantsKV} + key, _ := delivery.HoldAskKey("web", fpTest) + held := &delivery.Held{Node: "ace", Words: "health failed"} + first := newPasser(kvs, time.Date(2026, 10, 25, 1, 0, 0, 0, time.UTC)) + first.holdAsk(context.Background(), "web", fpTest, held) + a1, err := delivery.DecodeAssignerAsk(key, asksKV.values[key]) + if err != nil { + t.Fatal(err) + } + first.lift(context.Background(), "web", fpTest) + if _, there := asksKV.values[key]; there { + t.Fatal("the lift left the hold ask") + } + again := newPasser(kvs, time.Date(2026, 10, 25, 2, 0, 0, 0, time.UTC)) // started again: it remembers nothing + again.holdAsk(context.Background(), "web", fpTest, held) + a2, err := delivery.DecodeAssignerAsk(key, asksKV.values[key]) + if err != nil { + t.Fatal(err) + } + if a2.Ask.ID == a1.Ask.ID { + t.Errorf("the ask written again after a lift and a restart has the id %s of the first", a1.Ask.ID) + } +} + +// 1906/H4: an answer counts only for the asking it answered: an answer to an earlier hold of the same fingerprint, or to an ask +// taken away, is not taken for the hold that stands now. +func TestAnAnswerCountsOnlyForTheAskingItAnswered(t *testing.T) { + key, _ := delivery.HoldAskKey("web", fpTest) + p := newPasser(buckets{}, time.Now()) + put := func(asking int) string { + a := assign.HoldAsk(key, "web", fpTest, "ace", "health failed", asking, time.Now()) + v, err := delivery.EncodeAssignerAsk(key, a) + if err != nil { + t.Fatal(err) + } + p.w.put(delivery.AssignerAsksBucket, key, v) + return a.Ask.ID + } + first := put(100) + v, _ := delivery.EncodeAnswer(delivery.Answer{Option: "release", Warrant: first + "@2026-10-25T01:00:00Z", Words: "the operator"}) + p.w.put(delivery.AnswersBucket, key, v) + if !p.answeredWith(key, "release") { + t.Fatal("the answer to the ask that stands is not taken") + } + if p.answeredWith(key, "keep") { + t.Error("an answer was taken for an option not chosen") + } + put(200) // the same fingerprint held again, after a lift + if p.answeredWith(key, "release") { + t.Error("the answer to an earlier hold carried over to a later one") + } + p.w.put(delivery.AssignerAsksBucket, key, nil) // the ask taken away + if p.answeredWith(key, "release") { + t.Error("an answer to an ask taken away was taken") + } +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go index e5088ad..7fbb11a 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go @@ -326,6 +326,13 @@ func (w *world) nodes(manifests map[string]catalogue.Manifest, assigned map[stri return n } +// value is what a bucket holds at a key now, nil when nothing. +func (w *world) value(bucket, key string) []byte { + w.mu.Lock() + defer w.mu.Unlock() + return w.values[bucket][key] +} + // has says whether a bucket holds a key now. func (w *world) has(bucket, key string) bool { w.mu.Lock() diff --git a/modules/mesh-delivery-assigner/internal/assign/asks.go b/modules/mesh-delivery-assigner/internal/assign/asks.go index a3ca7ff..15662c9 100644 --- a/modules/mesh-delivery-assigner/internal/assign/asks.go +++ b/modules/mesh-delivery-assigner/internal/assign/asks.go @@ -10,14 +10,17 @@ import ( "git.novox.be/novox/mesh-sdk/go/delivery" ) -// 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. 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), +// HoldAsk is the ask about a module's fingerprint the assigner holds, at its key (delivery.HoldAskKey): keep it held or +// release it. Releasing writes the held fingerprint's want again by the rules (ADR 0539 §19). node and words are the held +// record's: the machine whose health check failed and the hold's words (issue 1906). asking numbers the writing: 0 the first, a +// larger number an ask written again after a hold lifted by itself, so the router takes it as new. +func HoldAsk(key, module, fp, node, words string, asking int, now time.Time) delivery.AssignerAsk { + return delivery.AssignerAsk{Asking: asking, Ask: asks.Ask{ + ID: delivery.AskingID(key, asking), 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. Releasing it lets the wants move to it again.", module, short(fp)), + Explanation: fmt.Sprintf("The health check of %s failed on %s at %s (%s), so the assigner holds that version and wants "+ + "no node to move to it. It lifts the hold by itself once every node that runs %s runs it healthy; releasing it "+ + "lets the wants move to it now.", module, node, short(fp), words, module), 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 and the wants move to it by the rules again", Level: asks.Acknowledge}, @@ -78,6 +81,13 @@ func HeldCondition(module, fp, node, words string, raised, needsYou bool) asks.C return c } +// LiftedCondition clears the condition of a held fingerprint the assigner lifted by itself, saying why (issue 1906). +func LiftedCondition(module, fp string) asks.Condition { + c := HeldCondition(module, fp, "", "", false, false) + c.Resolved = LiftedWords(module, fp) + "." + return c +} + // The options of the record ask (ADR 0539 §14) and of the backup ask (§16). const ( OptionYes = "yes" diff --git a/modules/mesh-delivery-assigner/internal/assign/asks_test.go b/modules/mesh-delivery-assigner/internal/assign/asks_test.go index f734ad2..1342b7d 100644 --- a/modules/mesh-delivery-assigner/internal/assign/asks_test.go +++ b/modules/mesh-delivery-assigner/internal/assign/asks_test.go @@ -17,13 +17,25 @@ func TestTheHoldAskIsValid(t *testing.T) { t.Fatal(err) } now := time.Now() - a := HoldAsk(key, "mail", fNew, now) + a := HoldAsk(key, "mail", fNew, "ace", "health failed, phase running: connection refused", 0, now) if err := a.Ask.Check(now); err != nil { t.Errorf("the router refuses the ask: %v", err) } if _, err := delivery.EncodeAssignerAsk(key, a); err != nil { t.Errorf("the ask cannot be written: %v", err) } + // 1906/H3: the ask names the machine whose check failed and the hold's words. + if !strings.Contains(a.Ask.Explanation, "on ace") || !strings.Contains(a.Ask.Explanation, "connection refused") { + t.Errorf("the ask does not name the node and the hold's words: %q", a.Ask.Explanation) + } + // 1906/H2: an ask written again after a lift is another asking, under an id of its own. + again := HoldAsk(key, "mail", fNew, "ace", "x", 7, now) + if again.Asking != 7 || again.Ask.ID != delivery.AskingID(key, 7) || again.Ask.ID == a.Ask.ID { + t.Errorf("the asking after a lift = %d %s", again.Asking, again.Ask.ID) + } + if _, err := delivery.EncodeAssignerAsk(key, again); err != nil { + t.Errorf("the ask asked again cannot be written: %v", err) + } if _, err := delivery.ViewAnswerKey(key); err != nil { t.Errorf("the view may not answer it: %v", err) } @@ -74,6 +86,17 @@ func TestTheHeldConditionNamesModuleNodeAndWords(t *testing.T) { } } +// 1906/H2: a hold lifted by itself clears the held condition under its key, saying why. +func TestALiftedHoldClearsItsConditionSayingWhy(t *testing.T) { + c := LiftedCondition("mail", fNew) + if _, err := asks.EncodeCondition(c); err != nil { + t.Fatalf("the condition is refused: %v", err) + } + if c.State != asks.StateCleared || c.Key != HeldCondition("mail", fNew, "ace", "x", true, false).Key || !strings.Contains(c.Resolved, "lifted its hold by itself") { + t.Errorf("condition = %+v", c) + } +} + // 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) { diff --git a/modules/mesh-delivery-assigner/internal/assign/wants.go b/modules/mesh-delivery-assigner/internal/assign/wants.go index 2f774a3..f0c83eb 100644 --- a/modules/mesh-delivery-assigner/internal/assign/wants.go +++ b/modules/mesh-delivery-assigner/internal/assign/wants.go @@ -4,6 +4,8 @@ import ( "fmt" "slices" "sort" + "strings" + "time" "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/delivery" @@ -31,6 +33,10 @@ type Module struct { // 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 + // Kept says the operator answered "keep" to the hold ask of FP: a hold kept so is not lifted by itself. + Kept bool + // Now is when this pass reads the haves: the held record names it as when the health check was seen to fail. Zero: not said. + Now time.Time } func (m Module) composition(node, fp string) (delivery.ComposedDigest, bool) { @@ -138,6 +144,9 @@ 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 + // Lifted is the hold of m.FP lifted by itself: every node that runs the module runs m.FP healthy (issue 1906). The held + // record and its ask are taken away, and the wants move by the rules again in this pass. + Lifted bool // 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 @@ -170,10 +179,13 @@ 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.Released && m.FP != "" { + if held && !m.Kept && m.FP != "" && allRunHealthy(m, n) { + out.Lifted, held = true, false + } + if !held && !out.Lifted && !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} + out.Held = &delivery.Held{Node: node, Words: HeldWords(h, m.Now)} held = true break } @@ -252,6 +264,43 @@ func Wants(m Module, n Nodes, wants map[string]delivery.Want, lastHealthy map[st return out } +// allRunHealthy says whether every node of m that runs the module (has a have of it) runs m.FP healthy, and at least one does: +// a hold whose fingerprint runs healthy wherever the module runs holds nothing back any more (issue 1906: seven modules stayed +// held on every machine after one transient failed check). A node set back to another fingerprint does not run m.FP, so its +// hold stays. +func allRunHealthy(m Module, n Nodes) bool { + ran := 0 + for _, node := range m.Nodes { + h, ok := n.Haves[node][m.Name] + if !ok { + continue + } + if !healthy(h, m.FP) { + return false + } + ran++ + } + return ran > 0 +} + +// HeldWords are the words a hold keeps of its cause (issue 1906): the health and phase the node reported, when the assigner saw +// it, and the node's own words. The node is the held record's own field. +func HeldWords(h delivery.Have, at time.Time) string { + words := fmt.Sprintf("health %s, phase %s", h.Health, h.Phase) + if !at.IsZero() { + words += ", seen at " + at.UTC().Format(time.RFC3339) + } + if d := strings.TrimSpace(h.Detail); d != "" { + return words + ": " + d + } + return words + ": the node gave no words" +} + +// LiftedWords say why a hold was lifted by itself. +func LiftedWords(module, fp string) string { + return fmt.Sprintf("every node that runs %s runs %s healthy, so the assigner lifted its hold by itself", module, short(fp)) +} + // normal is a want without its version, as a decoded and a composed one are compared. func normal(w delivery.Want) delivery.Want { w.V = 0 diff --git a/modules/mesh-delivery-assigner/internal/assign/wants_test.go b/modules/mesh-delivery-assigner/internal/assign/wants_test.go index 31ff233..18da1ed 100644 --- a/modules/mesh-delivery-assigner/internal/assign/wants_test.go +++ b/modules/mesh-delivery-assigner/internal/assign/wants_test.go @@ -4,6 +4,7 @@ import ( "slices" "strings" "testing" + "time" "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/delivery" @@ -110,8 +111,10 @@ func TestAnOfflineCanaryIsPassedOver(t *testing.T) { func TestAFailedHealthHoldsAndSetsBack(t *testing.T) { m, n := ready() n.Haves["ace"]["mail"] = delivery.Have{FP: fNew, Phase: delivery.PhaseRunning, Health: delivery.HealthFailed, Detail: "connection refused"} + m.Now = time.Date(2026, 10, 25, 1, 0, 0, 0, time.UTC) out := Wants(m, n, map[string]delivery.Want{"ace": {FP: fNew, Composed: dig(fNew)}, "g14": {FP: fOld, Composed: dig(fOld)}}, map[string]string{"ace": fOld}) - if out.Held == nil || out.Held.Node != "ace" || out.Held.Words != "connection refused" { + // 1906/H3: the held record keeps its cause in words: which machine (its node), which health, when, and the node's words. + if out.Held == nil || out.Held.Node != "ace" || out.Held.Words != "health failed, phase running, seen at 2026-10-25T01:00:00Z: connection refused" { t.Fatalf("held = %+v", out.Held) } if w := out.Wants["ace"]; w.FP != fOld || w.Composed != dig(fOld) || w.Newest != fNew || w.Rule != delivery.RuleNotHeld { @@ -520,3 +523,59 @@ func TestAHaveFromBeforeTheSwitchGetsItsWantWithNoRecordHold(t *testing.T) { } } } + +// 1906/H3: a failed check whose node gave no words is held saying so, never with empty words. +func TestAHoldOfAFailureWithoutWordsSaysSo(t *testing.T) { + m, n := ready() + n.Haves["ace"]["mail"] = delivery.Have{FP: fNew, Phase: delivery.PhaseRunning, Health: delivery.HealthFailed} + out := Wants(m, n, map[string]delivery.Want{"ace": {FP: fNew, Composed: dig(fNew)}}, map[string]string{"ace": fOld}) + if out.Held == nil || out.Held.Words != "health failed, phase running: the node gave no words" { + t.Fatalf("held = %+v", out.Held) + } +} + +// 1906/H2: given a held fingerprint that every node running the module now runs healthy, when the assigner passes, then the +// hold is lifted by itself and the wants move to it by the rules again. A node set back to another fingerprint, a node whose +// check fails, or a hold the operator answered "keep" leaves it held; a node that does not run the module at all does not count. +func TestAHoldIsLiftedOnceEveryNodeRunsItHealthy(t *testing.T) { + everywhere := func() (Module, Nodes) { + m, n := ready() + m.Held = true + n.Haves["g14"]["mail"] = delivery.Have{FP: fNew, Phase: delivery.PhaseRunning, Health: delivery.HealthOK} + return m, n + } + m, n := everywhere() + out := Wants(m, n, wantedOld(), map[string]string{"ace": fOld, "g14": fOld}) + if !out.Lifted || out.Held != nil { + t.Fatalf("lifted = %v, held = %+v: want the hold lifted", out.Lifted, out.Held) + } + for _, node := range []string{"ace", "g14"} { + if w := out.Wants[node]; w.FP != fNew || w.Rule != 0 { + t.Errorf("%s's want = %+v, want it moved to the lifted fingerprint", node, w) + } + } + + m, n = everywhere() + delete(n.Haves["g14"], "mail") + if out := Wants(m, n, wantedOld(), nil); !out.Lifted { + t.Error("a node that does not run the module kept the hold") + } + + for name, change := range map[string]func(*Module, *Nodes){ + "set back": func(_ *Module, n *Nodes) { + n.Haves["g14"]["mail"] = delivery.Have{FP: fOld, Phase: delivery.PhaseRunning, Health: delivery.HealthOK} + }, + "failing": func(_ *Module, n *Nodes) { + n.Haves["g14"]["mail"] = delivery.Have{FP: fNew, Phase: delivery.PhaseRunning, Health: delivery.HealthFailed} + }, + "kept": func(m *Module, _ *Nodes) { m.Kept = true }, + "no haves": func(_ *Module, n *Nodes) { delete(n.Haves["g14"], "mail"); delete(n.Haves["ace"], "mail") }, + "not held": func(m *Module, _ *Nodes) { m.Held = false }, + } { + m, n := everywhere() + change(&m, &n) + if out := Wants(m, n, wantedOld(), map[string]string{"ace": fOld, "g14": fOld}); out.Lifted { + t.Errorf("%s: the hold was lifted", name) + } + } +} diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/assignerasks.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/assignerasks.go new file mode 100644 index 0000000..04f1bfc --- /dev/null +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/assignerasks.go @@ -0,0 +1,213 @@ +package main + +import ( + "encoding/json" + "fmt" + "slices" + + stdio "git.novox.be/novox/mesh-sdk/go" + + "git.novox.be/novox/mesh-sdk/go/asks" + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// The assigner's asks reach the operator (issue 1906). The assigner writes its asks (the hold of a fingerprint, the restore +// of a data module, the backup before an apply and a move of a module of policy record) in its bucket asks; until now the +// teller only filled them from an answer given in the view, so an ask nobody opened the view for reached nobody: on +// 2026-10-25 seven modules stayed held on every machine and the operator was never asked. Now the teller raises each on the +// operator channel as it raises the gate's merge asks: as the asker mesh-delivery-teller, once per router id, through the same +// records, the same wait for room and the same warrants. The operator's answer, from the phone or from the view, is written +// under the ask's own key in the bucket answers, which the assigner reads (mesh-delivery-assigner pass.go answeredWith); none +// of these asks needs proof of who answered (mesh-sdk go/delivery AskNeedsProof). + +// mostAssignerOpen is how many of the assigner's asks the teller keeps open at the router at once. The router keeps at most +// asks.MostOpen open asks of one asker, and the teller asks both the gate's merge asks and the assigner's as one asker, so a burst +// of holds (seven at once on 2026-10-25) would take all the room and a merge Approve could not be asked; one place is always +// left for a merge ask. An assigner ask beyond these waits in the wait for room and is asked when one of them ends. +const mostAssignerOpen = asks.MostOpen - 1 + +// assignerOpenRecord is the key of the router ids of the assigner's asks the teller asked. +const assignerOpenRecord = "assigner.open" + +// assignerOpen is the router ids of the assigner's asks open now: those asked whose ask is neither answered nor ended. An id +// that ended is dropped from the record as it is read. +func (t *Teller) assignerOpen() ([]string, *stdio.StateEntry) { + e, err := t.Store.Get(assignerOpenRecord) + if err != nil || e == nil { + return nil, e + } + var ids []string + if err := json.Unmarshal(e.Value, &ids); err != nil { + t.logf("the record %s cannot be read: %v", assignerOpenRecord, err) + return nil, e + } + return slices.DeleteFunc(ids, func(id string) bool { + k, _, err := t.keptOf(id) + return err == nil && !open(k) + }), e +} + +// assignerRoom says whether another of the assigner's asks may be asked now. +func (t *Teller) assignerRoom() bool { + ids, _ := t.assignerOpen() + return len(ids) < mostAssignerOpen +} + +// assignerAsked keeps that the assigner's ask with this router id is asked. +func (t *Teller) assignerAsked(id string) { + ids, e := t.assignerOpen() + if !slices.Contains(ids, id) { + ids = append(ids, id) + } + var err error + if e == nil { + _, err = t.Store.Create(assignerOpenRecord, ids) + } else { + _, err = t.Store.Update(assignerOpenRecord, ids, e.Revision) + } + if err != nil { + t.logf("the record %s could not be kept: %v", assignerOpenRecord, err) + } +} + +// waitForAssignerRoom puts an assigner ask in the wait for room, once. +func (t *Teller) waitForAssignerRoom(key, id string) { + ws, rec, err := t.waiting() + if err != nil { + t.logf("%v", err) + return + } + if slices.ContainsFunc(ws, func(x waitingAsk) bool { return x.ID == id }) { + return + } + if t.keepWaiting(append(ws, waitingAsk{Key: key, ID: id}), rec) == nil { + t.logf("%s: the assigner's ask %s waits: %d of the assigner's asks are open, the most the teller keeps open so a merge ask always has room", key, id, mostAssignerOpen) + } +} + +// isAssignerKey says whether an ask key is one of the assigner's, not the gate's merge ask. +func isAssignerKey(key string) bool { + k, err := delivery.ParseAnyAskKey(key) + return err == nil && k.Kind != delivery.AskKindMerge +} + +// assignerAsk reads the assigner's ask at key: found false when the key holds none. +func (t *Teller) assignerAsk(key string) (delivery.AssignerAsk, bool, error) { + if t.Assigner == nil { + return delivery.AssignerAsk{}, false, fmt.Errorf("the teller reads no assigner asks") + } + raw, found, err := t.Assigner.Get(key) + if err != nil || !found { + return delivery.AssignerAsk{}, false, err + } + a, err := delivery.DecodeAssignerAsk(key, raw) + if err != nil { + return delivery.AssignerAsk{}, false, fmt.Errorf("the assigner's key %s holds no ask: %w", key, err) + } + return a, true, nil +} + +// raiseAssignerAsk asks the operator the assigner's ask at key, once per router id: an ask answered already (from the view or +// the phone) is not asked; an earlier asking still open is taken back first; an asking already made is only caught up with +// at the router, and one that ended is not asked again, for the router drops a second ask under an id it keeps. +func (t *Teller) raiseAssignerAsk(key string) { + a, found, err := t.assignerAsk(key) + if err != nil { + t.logf("%s: the ask is not raised: %v", key, err) + return + } + if !found { + return + } + id := a.Ask.ID + if raw, _, has, err := t.Answers.Get(delivery.AnswerKey(key)); err != nil { + t.logf("%s: the answers could not be read, so the ask is not raised: %v", key, err) + return + } else if has && answeredAsking(raw) == id { + return + } + prior, grec, err := t.seenAt(gateRecordKey(key)) + if err != nil { + t.logf("%s: %v", key, err) + return + } + if grec != nil && prior.ID != id { + t.cancelOpen(key, prior.ID) // a new asking of the same key: the earlier one is taken back + } + k, _, err := t.keptOf(id) + if err != nil { + t.logf("%s: %v", key, err) + return + } + if k != nil { + if grec == nil || prior.ID != id { + _ = t.note(id, key, 0, grec) + } + if open(k) { + t.catchUp(id) + } + return + } + if err := t.note(id, key, 0, grec); err != nil { + t.logf("%s: the record of the ask %s could not be kept, so it is not raised: %v", key, id, err) + return + } + if !t.assignerRoom() { + t.waitForAssignerRoom(key, id) + return + } + if err := t.Asker.Ask(a.Ask); err != nil { + t.logf("%s: the ask %s was not asked: %v", key, id, err) + if kept, _, _ := t.keptOf(id); open(kept) { + _ = t.Store.Delete(keptKey(id)) // kept but not published: forgotten, so the next writing of the key asks it + } + return + } + t.assignerAsked(id) + t.logf("%s: the assigner's ask %s is asked of the operator (%s)", key, id, a.Ask.Headline) +} + +// OnAssignerAskGone handles a key the assigner took away from its asks (a hold it lifted by itself): the open router ask of +// its latest asking is taken back, so the operator is not asked about what no longer holds. +func (t *Teller) OnAssignerAskGone(key string) { + t.mu.Lock() + defer t.mu.Unlock() + if !isAssignerKey(key) { + t.logf("the assigner's key %s was taken away and is no ask key of the assigner, and is left", key) + return + } + prior, grec, err := t.seenAt(gateRecordKey(key)) + if err != nil || grec == nil { + if err != nil { + t.logf("%s: %v", key, err) + } + return + } + t.cancelOpen(key, prior.ID) +} + +// answerAssigner is the act of a warrant for one of the assigner's asks: the ask must still stand at its key under the asking +// and the words the operator answered, and then the answer is written under the ask's key, once. Its options bind no act: +// the assigner reads the option chosen (asker.Client.Take has checked the option, its level and the ask's digest). +func (t *Teller) answerAssigner(key string, o asks.Option, w asks.Warrant) error { + a, found, err := t.assignerAsk(key) + if err != nil { + return err + } + if !found { + t.logf("%s: the ask %s was answered and the assigner has taken it away: the answer changes nothing", key, w.Ask) + return nil + } + if a.Ask.ID != w.Ask { + t.logf("%s: the ask %s was answered, and the assigner has asked again under %s: an answer to an earlier asking changes nothing", key, w.Ask, a.Ask.ID) + return nil + } + if a.Ask.Digest() != w.AskDigest { + return fmt.Errorf("the assigner's ask %s is no longer the ask the operator answered (its digest is %s, the warrant's %s)", key, a.Ask.Digest(), w.AskDigest) + } + body, err := delivery.EncodeAnswer(delivery.Answer{Option: o.ID, Warrant: warrantID(w), Words: w.Says()}) + if err != nil { + return err + } + return t.writeAnswer(key, w.Ask, body, w.Says()) +} diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/assignerasks_test.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/assignerasks_test.go new file mode 100644 index 0000000..1cbed7e --- /dev/null +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/assignerasks_test.go @@ -0,0 +1,209 @@ +package main + +import ( + "encoding/json" + "strings" + "testing" + + "git.novox.be/novox/mesh-sdk/go/asks" + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// 1906/H1: given the assigner's hold ask in its bucket and no answer, when the teller hears the key, then it raises the ask on +// the operator channel as itself, unchanged and once (the key heard again, as a watch replays it, asks nothing more). +func TestAnAssignerAskIsRaisedOnTheOperatorChannelOnce(t *testing.T) { + b := newBed(t) + assigner, _ := withView(b) + key, value, ask := holdAsk(t) + assigner[key] = value + b.t.OnAssignerAsk(key) + b.t.OnAssignerAsk(key) + asked := b.askedOf(asks.AskSubject(delivery.TellerModule)) + if len(asked) != 1 { + t.Fatalf("published %d asks, want one; log:\n%s", len(asked), b.logs.String()) + } + var got asks.Ask + if err := json.Unmarshal(asked[0].body, &got); err != nil { + t.Fatal(err) + } + if got.ID != ask.ID || got.Digest() != ask.Digest() { + t.Errorf("the router's ask %+v differs from the assigner's", got) + } +} + +// 1906/H1: without the view's bucket the assigner's asks are still raised. +func TestAnAssignerAskIsRaisedWithoutTheView(t *testing.T) { + b := newBed(t) + assigner := memGate{} + b.t.Assigner = assigner + key, value, _ := holdAsk(t) + assigner[key] = value + b.t.OnAssignerAsk(key) + if n := len(b.askedOf(asks.AskSubject(delivery.TellerModule))); n != 1 { + t.Fatalf("published %d asks, want one; log:\n%s", n, b.logs.String()) + } +} + +// 1906/H1: the operator's answer on the phone comes back by the same path as the view's: written under the ask's own key in +// the bucket answers, naming the asking, once; a second warrant writes nothing more and nothing is said as a fault. +func TestTheOperatorsAnswerToAnAssignerAskIsWrittenUnderItsKey(t *testing.T) { + b := newBed(t) + assigner, _ := withView(b) + key, value, ask := holdAsk(t) + assigner[key] = value + b.t.OnAssignerAsk(key) + w := ackFor(ask, "release") + b.deliver(t, w) + raw, ok := b.answers.written[delivery.AnswerKey(key)] + if !ok { + t.Fatalf("no answer under %s; log:\n%s", key, b.logs.String()) + } + if a, err := delivery.DecodeAnswer(raw); err != nil || a.Option != "release" || answeredAsking(raw) != ask.ID { + t.Errorf("answer = %+v, %v", a, err) + } + b.deliver(t, w) + if b.answers.writes != 1 || len(b.conds) != 0 { + t.Errorf("%d writes, conditions %+v", b.answers.writes, b.conds) + } +} + +// 1906/H1: an answer to an asking the assigner has since taken away or asked again changes nothing. +func TestAnAnswerToAnAssignerAskTakenAwayWritesNothing(t *testing.T) { + b := newBed(t) + assigner, _ := withView(b) + key, value, ask := holdAsk(t) + assigner[key] = value + b.t.OnAssignerAsk(key) + delete(assigner, key) + b.deliver(t, ackFor(ask, "release")) + if b.answers.writes != 0 { + t.Errorf("an answer to a withdrawn ask was written") + } +} + +// 1906/H1, H2: an ask answered in the view is taken back from the phone, and an ask the assigner took away (a hold it lifted by +// itself) is taken back too; an ask answered before it is heard is never raised. +func TestAnAssignerAskAnsweredInTheViewOrTakenAwayIsTakenBack(t *testing.T) { + cancelled := func(b *bed, id string) bool { + for _, p := range b.askedOf(asks.CancelSubject(delivery.TellerModule)) { + if string(p.body) == `{"id":"`+id+`"}` { + return true + } + } + return false + } + b := newBed(t) + assigner, view := withView(b) + key, value, ask := holdAsk(t) + assigner[key] = value + b.t.OnAssignerAsk(key) + view[key+".view"] = viewAnswer(t, "release", "") + b.t.OnViewAnswer(key + ".view") + if !cancelled(b, ask.ID) { + t.Errorf("the ask answered in the view was not taken back; log:\n%s", b.logs.String()) + } + + b = newBed(t) + assigner, _ = withView(b) + assigner[key] = value + b.t.OnAssignerAsk(key) + delete(assigner, key) + b.t.OnAssignerAskGone(key) + if !cancelled(b, ask.ID) { + t.Errorf("the ask taken away was not taken back; log:\n%s", b.logs.String()) + } + + b = newBed(t) + assigner, view = withView(b) + assigner[key] = value + view[key+".view"] = viewAnswer(t, "keep", "") + b.t.OnAssignerAsk(key) + if n := len(b.askedOf(asks.AskSubject(delivery.TellerModule))); n != 0 { + t.Errorf("an ask answered in the view was raised %d times", n) + } +} + +// 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) { + b := newBed(t) + assigner, _ := withView(b) + _, first := b.askOf(t, 1, 1) + b.askOf(t, 2, 2) + b.askOf(t, 3, 3) + key, value, ask := holdAsk(t) + assigner[key] = value + b.t.OnAssignerAsk(key) + b.refuse(t, ask.ID, roomRefusal) + b.deliver(t, warrantFor(first.Ask, "approve")) + got := b.askedIDs() + if len(got) != 5 || got[4] != ask.ID { + t.Fatalf("asks = %v, want the assigner's %s asked again", got, ask.ID) + } + + b = newBed(t) + assigner, _ = withView(b) + assigner[key] = value + b.t.OnAssignerAsk(key) + b.refuse(t, ask.ID, "the ask has no headline") + if len(b.conds) != 1 || b.conds[0].Subject.Scope != "delivery" || strings.Contains(b.conds[0].Explanation, "merge") { + t.Errorf("conditions = %+v", b.conds) + } +} + +// ackFor is the router's warrant for an option of level acknowledge, as the assigner's asks offer. +func ackFor(a asks.Ask, option string) asks.Warrant { + w := warrantFor(a, option) + w.Level, w.Label = asks.Acknowledge, option + return w +} + +// holdAskOf is the assigner's hold ask of module plex at the n-th fingerprint, shaped as holdAsk. +func holdAskOf(t *testing.T, n int) (string, []byte, asks.Ask) { + t.Helper() + key, err := delivery.HoldAskKey("plex", "src2:"+strings.Repeat(string(rune('0'+n)), 64)) + if err != nil { + t.Fatal(err) + } + _, _, ask := holdAsk(t) + ask.ID = delivery.AskID(key) + value, err := delivery.EncodeAssignerAsk(key, delivery.AssignerAsk{Ask: ask}) + if err != nil { + t.Fatal(err) + } + return key, value, ask +} + +// 1906/H5: given seven holds at once (as on 2026-10-25), when they are raised, then at most two are open at the router, a merge +// ask is still asked, and each hold answered lets the next waiting one be asked. +func TestABurstOfHoldsLeavesRoomForAMergeAsk(t *testing.T) { + b := newBed(t) + assigner, _ := withView(b) + var held []asks.Ask + for n := 0; n < 7; n++ { + key, value, ask := holdAskOf(t, n) + assigner[key] = value + b.t.OnAssignerAsk(key) + held = append(held, ask) + } + if got := b.askedIDs(); len(got) != mostAssignerOpen || got[0] != held[0].ID || got[1] != held[1].ID { + t.Fatalf("asked %v, want the first %d holds only; log:\n%s", got, mostAssignerOpen, b.logs.String()) + } + _, merge := b.askOf(t, 1, 1) + if got := b.askedIDs(); len(got) != mostAssignerOpen+1 || got[mostAssignerOpen] != merge.Ask.ID { + t.Fatalf("asked %v, want the merge ask asked beside the holds", got) + } + if len(b.dropped) != 0 || len(b.conds) != 0 { + t.Errorf("router dropped %v, conditions %+v", b.dropped, b.conds) + } + b.deliver(t, ackFor(held[0], "release")) + got := b.askedIDs() + if len(got) != mostAssignerOpen+2 || got[len(got)-1] != held[2].ID { + t.Fatalf("asked %v after a hold was answered, want the third hold asked next; log:\n%s", got, b.logs.String()) + } + // The merge ask answered frees its own place, and no hold takes it. + b.deliver(t, warrantFor(merge.Ask, "approve")) + if got := b.askedIDs(); len(got) != mostAssignerOpen+2 { + t.Errorf("asked %v after the merge ask was answered, want no further hold", got) + } +} diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go index b4c5c4b..99a2895 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go @@ -545,3 +545,99 @@ func TestTellerFillsAnAskFromTheViewsAnswerOnce(t *testing.T) { // 1297/B10 AC t.Error("the merge's ask was answered from a view key") } } + +// TestTellerRaisesAnAssignerAskAndWritesTheOperatorsAnswer starts the program against the bus: the assigner's hold ask, put in +// its bucket, is on the operator channel's queue as the teller's ask; the router's warrant for "release" is written under the +// ask's key, where the assigner reads it; a second ask the assigner takes away (a hold it lifted) is cancelled at the router +// (1906/H1, H2 on the real bus). +func TestTellerRaisesAnAssignerAskAndWritesTheOperatorsAnswer(t *testing.T) { + now = time.Now().UTC() + b := startFull(t) + assigner := b.as(t, "n1.mesh-delivery-assigner") + broker := filepath.Join(t.TempDir(), "broker") + cred, _ := json.Marshal(credential{URL: b.srv.ClientURL(), Node: "n1", Module: "mesh-delivery-teller", User: "n1.mesh-delivery-teller", Password: "pw"}) + if err := os.WriteFile(broker, cred, 0o600); err != nil { + t.Fatal(err) + } + env := map[string]string{"MESH_BROKER_FILE": broker} + logs := &syncBuffer{} + rn := startRunner(b.ctx, func(c context.Context) error { return run(c, func(k string) string { return env[k] }, logs) }) + t.Cleanup(func() { rn.finish(t, 10*time.Second) }) + + key, value, ask := holdAsk(t) + askKV, err := assigner.KeyValue(b.ctx, assignerAsks) + if err != nil { + t.Fatal(err) + } + time.Sleep(500 * time.Millisecond) // the program hears warrants and watches + if _, err := askKV.Put(b.ctx, key, value); err != nil { + t.Fatal(err) + } + seat, err := b.js.Stream(b.ctx, "SEAT_OPERATOR_CHANNEL") + if err != nil { + t.Fatal(err) + } + var raw *jetstream.RawStreamMsg + for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) { + if raw, err = seat.GetLastMsgForSubject(b.ctx, seatAsk); err == nil { + break + } + } + if raw == nil { + t.Fatalf("the assigner's ask is not on the seat's queue: %v\n%s", err, logs.String()) + } + var asked asks.Ask + if err := json.Unmarshal(raw.Data, &asked); err != nil || asked.ID != ask.ID || asked.Digest() != ask.Digest() { + t.Fatalf("the ask on the queue is %+v (%v), want the assigner's", asked, err) + } + w, _ := json.Marshal(ackFor(ask, "release")) + if err := b.ctl.Publish(seatDecided, w); err != nil { + t.Fatal(err) + } + answers, err := b.js.KeyValue(b.ctx, tellerAnswers) + if err != nil { + t.Fatal(err) + } + var e jetstream.KeyValueEntry + for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) { + if e, err = answers.Get(b.ctx, delivery.AnswerKey(key)); err == nil { + break + } + } + if e == nil { + t.Fatalf("no answer under %s: %v\n%s", key, err, logs.String()) + } + if a, err := delivery.DecodeAnswer(e.Value()); err != nil || a.Option != "release" || answeredAsking(e.Value()) != ask.ID { + t.Fatalf("answer %+v err %v", a, err) + } + + key2, err := delivery.HoldAskKey("plex", "src2:"+strings.Repeat("3", 64)) + if err != nil { + t.Fatal(err) + } + ask2 := ask + ask2.ID = delivery.AskID(key2) + value2, err := delivery.EncodeAssignerAsk(key2, delivery.AssignerAsk{Ask: ask2}) + if err != nil { + t.Fatal(err) + } + if _, err := askKV.Put(b.ctx, key2, value2); err != nil { + t.Fatal(err) + } + for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) { + if strings.Contains(logs.String(), ask2.ID+" is asked") { + break + } + } + if err := askKV.Delete(b.ctx, key2); err != nil { + t.Fatal(err) + } + for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) { + if raw, err = seat.GetLastMsgForSubject(b.ctx, seatCancel); err == nil { + break + } + } + if err != nil || string(raw.Data) != `{"id":"`+ask2.ID+`"}` { + t.Errorf("the ask taken away was not cancelled: %v %v\n%s", raw, err, logs.String()) + } +} diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go index f6b5e09..f9b9142 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go @@ -141,14 +141,20 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // there is said and leaves the teller without that path rather than without the gate's asks (a wait for it needs a timer). failed := make(chan error, 2) assignerKV, viewKV := optionalBucket(ctx, js, delivery.AssignerAsksBucket, logger), optionalBucket(ctx, js, delivery.ViewAnswersBucket, logger) + // The assigner's asks are raised on the operator channel whether or not the view's answers can be read (issue 1906). + if viewKV != nil { + t.View = kvGate{ctx: ctx, kv: viewKV} + } + if assignerKV != nil { + t.Assigner = kvGate{ctx: ctx, kv: assignerKV} + go func() { + failed <- watchAsks(ctx, assignerKV, func(key string, _ uint64, _ []byte) { t.OnAssignerAsk(key) }, t.OnAssignerAskGone, nil) + }() + } if assignerKV != nil && viewKV != nil { - t.Assigner, t.View = kvGate{ctx: ctx, kv: assignerKV}, kvGate{ctx: ctx, kv: viewKV} go func() { failed <- watchAsks(ctx, viewKV, func(key string, _ uint64, _ []byte) { t.OnViewAnswer(key) }, func(string) {}, nil) }() - go func() { - failed <- watchAsks(ctx, assignerKV, func(key string, _ uint64, _ []byte) { t.OnAssignerAsk(key) }, func(string) {}, nil) - }() } go func() { if err := watchAsks(ctx, gate, t.OnAsk, t.OnAskGone, t.OnCurrent); err != nil { @@ -170,7 +176,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro func optionalBucket(ctx context.Context, js jetstream.JetStream, name string, logger *log.Logger) jetstream.KeyValue { kv, err := js.KeyValue(ctx, name) if err != nil { - logger.Printf("the bucket %s cannot be read (%v): the view's answers fill no ask until the teller is started again with it", name, err) + logger.Printf("the bucket %s cannot be read (%v): the asks and answers it holds are not carried until the teller is started again with it", name, err) return nil } return kv diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller.go index 3e36b3f..40f1f20 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller.go @@ -340,6 +340,9 @@ func (t *Teller) Answer(o asks.Option, w asks.Warrant) error { if srec == nil { return fmt.Errorf("the teller holds no gate key for the ask %s", w.Ask) } + if isAssignerKey(s.Key) { + return t.answerAssigner(s.Key, o, w) + } raw, found, err := t.Gate.Get(s.Key) if err != nil { return fmt.Errorf("the gate's ask %s could not be read: %w", s.Key, err) @@ -367,26 +370,31 @@ func (t *Teller) Answer(o asks.Option, w asks.Warrant) error { if err != nil { return err } - // Compare-and-set: created when the key has no answer; updated at the revision read when it holds an earlier asking's - // answer (the gate ignores that one); left when it holds this asking's. - akey := delivery.AnswerKey(s.Key) + return t.writeAnswer(s.Key, w.Ask, body, w.Says()) +} + +// writeAnswer writes the answer to the asking askID under the ask's key by compare-and-set: created when the key has no answer; +// updated at the revision read when it holds an earlier asking's answer (the reader ignores that one); left when it holds this +// asking's. +func (t *Teller) writeAnswer(key, askID string, body []byte, words string) error { + akey := delivery.AnswerKey(key) cur, crev, found, err := t.Answers.Get(akey) if err != nil { - return fmt.Errorf("the answer under %s could not be read: %w", s.Key, err) + return fmt.Errorf("the answer under %s could not be read: %w", key, err) } - if found && answeredAsking(cur) == w.Ask { - t.logf("%s: an answer is there already, and is not written again", s.Key) + if found && answeredAsking(cur) == askID { + t.logf("%s: an answer is there already, and is not written again", key) return nil } written, err := t.Answers.Write(akey, body, crev, found) if err != nil { - return fmt.Errorf("the answer under %s could not be written: %w", s.Key, err) + return fmt.Errorf("the answer under %s could not be written: %w", key, err) } if !written { - t.logf("%s: another writer answered first, and the answer is not written again", s.Key) + t.logf("%s: another writer answered first, and the answer is not written again", key) return nil } - t.logf("%s: %s", s.Key, w.Says()) + t.logf("%s: %s", key, words) return nil } @@ -423,10 +431,10 @@ func (t *Teller) saidOnce(h asker.Handled, w asks.Warrant, err error) { t.clear(w.Ask) case h == asker.Refused || (h == asker.Acted && err != nil): t.logf("the warrant of %s does not hold, and no answer is written: %v", w.Ask, err) - t.raise(w.Ask, "A merge Approve did not hold", "its answer did not hold here: "+oneLine(err)) + t.raise(w.Ask, t.kindOf(w.Ask).answerHeadline, "its answer did not hold here: "+oneLine(err)) case h == asker.Recorded && w.Outcome == asks.OutcomeRefused: t.logf("the router refused the ask %s: %s", w.Ask, w.Words) - t.raise(w.Ask, "A merge ask was refused", "the router refused it: "+w.Words) + t.raise(w.Ask, t.kindOf(w.Ask).refusedHeadline, "the router refused it: "+w.Words) case err != nil: t.logf("the warrant of %s: %v", w.Ask, err) } @@ -446,6 +454,30 @@ func oneLine(err error) string { // conditionKey is the key of the condition about an ask. func conditionKey(id string) string { return delivery.TellerModule + ".merge-ask." + id } +// askWords are the words the teller's conditions use for one kind of ask: the gate's merge ask, or one of the assigner's. +type askWords struct { + scope, noun, needs, nothing, resolved string + answerHeadline, refusedHeadline, unaskedHeadline string +} + +var ( + mergeWords = askWords{scope: "merge", noun: "The merge ask ", needs: "to ask for the Approve again", nothing: "Nothing was merged.", + resolved: "A good answer to the merge ask was written.", answerHeadline: "A merge Approve did not hold", + refusedHeadline: "A merge ask was refused", unaskedHeadline: "A merge ask could not be asked"} + assignerWords = askWords{scope: "delivery", noun: "The delivery ask ", needs: "to answer it in the delivery view", + nothing: "Nothing was changed on a machine.", resolved: "A good answer to the delivery ask was written.", + answerHeadline: "An answer to a delivery ask did not hold", refusedHeadline: "A delivery ask was refused", + unaskedHeadline: "A delivery ask could not be asked"} +) + +// kindOf is the words for the ask with this router id, by the key it was asked for. +func (t *Teller) kindOf(askID string) askWords { + if isAssignerKey(t.condID(askID)) { + return assignerWords + } + return mergeWords +} + func (t *Teller) raise(askID, headline, why string) { if t.Say == nil { return @@ -457,11 +489,12 @@ func (t *Teller) raise(askID, headline, why string) { } t.logf("the condition of %s could not be kept: %v", id, err) } - needs := "to ask for the Approve again" - c := asks.Condition{Key: conditionKey(id), Severity: asks.SeverityWarning, Subject: asks.Scope{Scope: "merge", ID: id}, State: asks.StateRaised, - Summary: "The merge ask " + id + ": " + why, Headline: headline, Needs: needs, - Explanation: asks.NeedsYou + needs + ". The merge ask " + id + ", " + why + ". Nothing was merged.", - Resolved: "A good answer to the merge ask was written."} + k := t.kindOf(askID) + needs := k.needs + c := asks.Condition{Key: conditionKey(id), Severity: asks.SeverityWarning, Subject: asks.Scope{Scope: k.scope, ID: id}, State: asks.StateRaised, + Summary: k.noun + id + ": " + why, Headline: headline, Needs: needs, + Explanation: asks.NeedsYou + needs + ". " + k.noun + id + ", " + why + ". " + k.nothing, + Resolved: k.resolved} if err := t.Say(c); err != nil { t.logf("the condition of %s could not be said: %v", id, err) _ = t.Store.Delete(condKey(id)) @@ -477,7 +510,7 @@ func (t *Teller) clear(askID string) { if err != nil || e == nil { return } - c := asks.Condition{Key: conditionKey(id), Subject: asks.Scope{Scope: "merge", ID: id}, State: asks.StateCleared} + c := asks.Condition{Key: conditionKey(id), Subject: asks.Scope{Scope: t.kindOf(askID).scope, ID: id}, State: asks.StateCleared} if err := t.Say(c); err != nil { t.logf("the condition of %s could not be cleared: %v", id, err) return @@ -578,7 +611,7 @@ func (t *Teller) roomRefused(w asks.Warrant) { return } t.logf("the router refused the ask %s for want of room %d times (%s): it is asked no more, and said", w.Ask, n, w.Words) - t.raise(w.Ask, "A merge ask could not be asked", fmt.Sprintf("the router refused it %d times for want of room: %s", n, w.Words)) + t.raise(w.Ask, t.kindOf(w.Ask).unaskedHeadline, fmt.Sprintf("the router refused it %d times for want of room: %s", n, w.Words)) return } if t.keepWaiting(ws, rec) != nil { @@ -675,6 +708,11 @@ func (t *Teller) askWaiting(n int) { i++ continue } + assigner := isAssignerKey(x.Key) + if assigner && !t.assignerRoom() { + i++ // the assigner's asks hold their most open: it waits, and a merge ask after it may be asked + continue + } a, why := t.stillAsked(x) if why != "" { t.logf("%s: the ask %s, waiting for room, is not asked again: %s", x.Key, x.ID, why) @@ -687,6 +725,9 @@ func (t *Teller) askWaiting(n int) { return } ws[i].Asked = true + if assigner { + t.assignerAsked(x.ID) + } t.logf("%s: the ask %s is asked of the operator again (refused %d times for want of room)", x.Key, x.ID, x.Refusals) if n--; n <= 0 { return @@ -701,6 +742,9 @@ func (t *Teller) stillAsked(x waitingAsk) (asks.Ask, string) { if x.Key == "" { return asks.Ask{}, "the teller holds no gate key for it" } + if isAssignerKey(x.Key) { + return t.stillAskedByAssigner(x) + } raw, found, err := t.Gate.Get(x.Key) if err != nil { return asks.Ask{}, "the gate's ask could not be read: " + err.Error() @@ -724,6 +768,26 @@ func (t *Teller) stillAsked(x waitingAsk) (asks.Ask, string) { return ga.Ask, "" } +// stillAskedByAssigner is stillAsked for one of the assigner's asks. +func (t *Teller) stillAskedByAssigner(x waitingAsk) (asks.Ask, string) { + a, found, err := t.assignerAsk(x.Key) + switch { + case err != nil: + return asks.Ask{}, err.Error() + case !found: + return asks.Ask{}, "the assigner has taken it away" + case a.Ask.ID != x.ID: + return asks.Ask{}, "the assigner asks again under " + a.Ask.ID + } + if cur, _, has, err := t.Answers.Get(delivery.AnswerKey(x.Key)); err == nil && has && answeredAsking(cur) == x.ID { + return asks.Ask{}, "it is answered" + } + if err := a.Ask.Check(t.now()); err != nil { + return asks.Ask{}, "it no longer holds: " + err.Error() + } + return a.Ask, "" +} + // askAgain keeps the ask afresh as the asker would (its record of the refusal is replaced, so its warrant is taken) and // publishes it under the bus message id given. func (t *Teller) askAgain(a asks.Ask, msgID string) error { diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/viewanswers.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/viewanswers.go index 235bb4d..f41b914 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/viewanswers.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/viewanswers.go @@ -25,7 +25,8 @@ func (t *Teller) OnViewAnswer(key string) { t.fillFromView(askKey) } -// OnAssignerAsk handles a key put in the assigner's bucket asks: a view answer that came before its ask fills it now. +// OnAssignerAsk handles a key put in the assigner's bucket asks: a view answer that came before its ask fills it now, and an +// ask not answered yet is raised on the operator channel (issue 1906; assignerasks.go). func (t *Teller) OnAssignerAsk(key string) { t.mu.Lock() defer t.mu.Unlock() @@ -34,6 +35,7 @@ func (t *Teller) OnAssignerAsk(key string) { return } t.fillFromView(key) + t.raiseAssignerAsk(key) } // fillFromView writes the answer to the assigner's ask at askKey from the view's answer, when both are there and readable, the @@ -103,5 +105,6 @@ func (t *Teller) fillFromView(askKey string) { t.logf("%s: another writer answered first, and the view's answer is not written", askKey) default: t.logf("%s: %s", askKey, words) + t.cancelOpen(askKey, ask.Ask.ID) // answered in the view: the same ask on the phone is taken back } } diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/viewanswers_test.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/viewanswers_test.go index cc86e6d..d6cea8f 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/viewanswers_test.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/viewanswers_test.go @@ -10,8 +10,9 @@ import ( const fpHeld = "src2:2222222222222222222222222222222222222222222222222222222222222222" -// holdAsk is a key of the assigner's asks: the ask about a held fingerprint, as a fixture (in release 5 the assigner writes -// none yet; the value is `{"v":1,"ask":}`). +// holdAsk is a key of the assigner's asks: the ask about a held fingerprint, shaped as the assigner composes it +// (mesh-delivery-assigner internal/assign asks.go HoldAsk: two options of level acknowledge that bind no act, so the router +// takes it; the value is `{"v":1,"ask":}`). func holdAsk(t *testing.T) (key string, value []byte, ask asks.Ask) { t.Helper() key, err := delivery.HoldAskKey("plex", fpHeld) @@ -20,7 +21,7 @@ func holdAsk(t *testing.T) (key string, value []byte, ask asks.Ask) { } ask = asks.Ask{ID: delivery.AskID(key), Headline: "plex is held after a failed health check", Explanation: "ace failed the check.", Who: asks.Operator, Expires: now.Add(asks.ApproveLasts), OnExpiry: "the fingerprint stays held", - Options: []asks.Option{{ID: "release", Label: "Release", Does: "lets the move go on", Level: asks.Approve}, {ID: "keep", Label: "Keep held", Does: "keeps it held", Level: asks.Approve}}} + Options: []asks.Option{{ID: "keep", Label: "Keep it held", Does: "keeps it held", Level: asks.Acknowledge}, {ID: "release", Label: "Release it", Does: "lets the move go on", Level: asks.Acknowledge}}} value, err = delivery.EncodeAssignerAsk(key, delivery.AssignerAsk{Ask: ask}) if err != nil { t.Fatal(err)