Merge pull request 'Raise the assigner's asks on the operator channel, lift a hold once every node runs it healthy, and keep its cause (issue 1906)' (#40) from fix/r6-hold-asks into main

This commit was merged in pull request #40.
This commit is contained in:
2026-10-25 13:27:29 +00:00
15 changed files with 1048 additions and 50 deletions
@@ -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.putJSON(t, provider.AssignmentsBucket, "ace.web", []byte(`{"node":"ace","module":"web","build":"b-1"}`))
b.start(t) b.start(t)
hkey, _ := delivery.HeldKey("web", fNew) 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) t.Fatalf("held = %+v, %v", h, err)
} }
akey, _ := delivery.HoldAskKey("web", fNew) akey, _ := delivery.HoldAskKey("web", fNew)
a, err := delivery.DecodeAssignerAsk(akey, b.get(t, delivery.AssignerAsksBucket, akey)) 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) 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 return last
} }
// answer says the teller wrote the operator's answer to the ask at key. // 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) { b.answerTo(t, key, option, "r1") } 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). // 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) { 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") 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")
}
}
}
@@ -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 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 answeredAt map[string]uint64
asks *composeAsks 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) { func (p *passer) pass(ctx context.Context) {
@@ -88,7 +97,7 @@ func (p *passer) wants(ctx context.Context) {
for name := range names { for name := range names {
fp := trunk[name].FP2 fp := trunk[name].FP2
name := name 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) }, Composed: func(node, fp string) (delivery.ComposedDigest, bool) { return p.w.composed(node, name, fp) },
Restored: func(node string) bool { Restored: func(node string) bool {
key, err := delivery.RestoreAskKey(node, name, fp) key, err := delivery.RestoreAskKey(node, name, fp)
@@ -103,6 +112,7 @@ func (p *passer) wants(ctx context.Context) {
grant, _ := delivery.ControllerGrantKey(name, fp) grant, _ := delivery.ControllerGrantKey(name, fp)
hold, _ := delivery.HoldAskKey(name, fp) hold, _ := delivery.HoldAskKey(name, fp)
m.Released = p.answeredWith(hold, "release") 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) 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 { if m.Released {
p.clearHeld(name, fp) 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) }) p.put(ctx, delivery.AssignerWantsBucket, key, func() ([]byte, error) { return delivery.EncodeHeld(held) })
} }
} }
if out.Held != nil || m.Held { if out.Lifted {
p.holdAsk(ctx, name, fp) 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) p.heldCondition(ctx, name, fp, out.Held, len(out.Restore) > 0)
} }
for _, node := range out.Restore { 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 // 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 // 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. // 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) key, err := delivery.HoldAskKey(module, fp)
if err != nil || p.w.has(delivery.AssignerAsksBucket, key) { if err != nil || p.w.has(delivery.AssignerAsksBucket, key) {
return 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) }) 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. // 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 { func (p *passer) heldHold(module, fp string, fresh *delivery.Held) delivery.Held {
if fresh != nil { if fresh != nil {
@@ -245,6 +305,7 @@ func (p *passer) heldCondition(ctx context.Context, module, fp string, fresh *de
return return
} }
p.said[c.Key+".raised"] = true 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. // 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) }) 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 // answeredWith says whether the teller wrote an answer to the ask at key as it stands now and the option chosen is option. The
// operator's, given in the view or on the phone, and stays under its key (ADR 0539 §14, §16, §19). // 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 { func (p *passer) answeredWith(key, option string) bool {
a, ok := p.w.answer(key) 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. // clearHeld clears, once, the condition of a hold the operator released.
@@ -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")
}
}
@@ -326,6 +326,13 @@ func (w *world) nodes(manifests map[string]catalogue.Manifest, assigned map[stri
return n 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. // has says whether a bucket holds a key now.
func (w *world) has(bucket, key string) bool { func (w *world) has(bucket, key string) bool {
w.mu.Lock() w.mu.Lock()
@@ -10,14 +10,17 @@ import (
"git.novox.be/novox/mesh-sdk/go/delivery" "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 // HoldAsk is the ask about a module's fingerprint the assigner holds, at its key (delivery.HoldAskKey): keep it held or
// it held or release it. Releasing writes the held fingerprint's want again by the rules (ADR 0539 §19). // release it. Releasing writes the held fingerprint's want again by the rules (ADR 0539 §19). node and words are the held
func HoldAsk(key, module, fp string, now time.Time) delivery.AssignerAsk { // record's: the machine whose health check failed and the hold's words (issue 1906). asking numbers the writing: 0 the first, a
return delivery.AssignerAsk{Ask: asks.Ask{ // larger number an ask written again after a hold lifted by itself, so the router takes it as new.
ID: delivery.AskingID(key, 0), 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)), 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 "+ 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. Releasing it lets the wants move to it again.", module, short(fp)), "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{ Options: []asks.Option{
{ID: "keep", Label: "Keep it held", Does: "the version stays held", Level: asks.Acknowledge}, {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}, {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 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). // The options of the record ask (ADR 0539 §14) and of the backup ask (§16).
const ( const (
OptionYes = "yes" OptionYes = "yes"
@@ -17,13 +17,25 @@ func TestTheHoldAskIsValid(t *testing.T) {
t.Fatal(err) t.Fatal(err)
} }
now := time.Now() 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 { if err := a.Ask.Check(now); err != nil {
t.Errorf("the router refuses the ask: %v", err) t.Errorf("the router refuses the ask: %v", err)
} }
if _, err := delivery.EncodeAssignerAsk(key, a); err != nil { if _, err := delivery.EncodeAssignerAsk(key, a); err != nil {
t.Errorf("the ask cannot be written: %v", err) 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 { if _, err := delivery.ViewAnswerKey(key); err != nil {
t.Errorf("the view may not answer it: %v", err) 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 // 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. // the view, and name the move and the node.
func TestTheRecordAndBackupAsksAreValid(t *testing.T) { func TestTheRecordAndBackupAsksAreValid(t *testing.T) {
@@ -4,6 +4,8 @@ import (
"fmt" "fmt"
"slices" "slices"
"sort" "sort"
"strings"
"time"
"git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/delivery" "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 // 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. // as any module is, losing the writes made since (ADR 0539 §19). Nil: none did.
Restored func(node string) bool 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) { func (m Module) composition(node, fp string) (delivery.ComposedDigest, bool) {
@@ -138,6 +144,9 @@ type Outcome struct {
Wants map[string]delivery.Want 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 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 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 is the moves a person must allow before the want names them: each is asked once (ADR 0539 §14).
Record []RecordNeed Record []RecordNeed
// Restore is the nodes of a module holding data whose held fingerprint failed its health check there and that are not set // 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 { 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{}} out := Outcome{Wants: map[string]delivery.Want{}, Rules: map[string]Holding{}}
held := m.Held 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)) { 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 { 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 held = true
break break
} }
@@ -252,6 +264,43 @@ func Wants(m Module, n Nodes, wants map[string]delivery.Want, lastHealthy map[st
return out 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. // normal is a want without its version, as a decoded and a composed one are compared.
func normal(w delivery.Want) delivery.Want { func normal(w delivery.Want) delivery.Want {
w.V = 0 w.V = 0
@@ -4,6 +4,7 @@ import (
"slices" "slices"
"strings" "strings"
"testing" "testing"
"time"
"git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/delivery"
@@ -110,8 +111,10 @@ func TestAnOfflineCanaryIsPassedOver(t *testing.T) {
func TestAFailedHealthHoldsAndSetsBack(t *testing.T) { func TestAFailedHealthHoldsAndSetsBack(t *testing.T) {
m, n := ready() m, n := ready()
n.Haves["ace"]["mail"] = delivery.Have{FP: fNew, Phase: delivery.PhaseRunning, Health: delivery.HealthFailed, Detail: "connection refused"} 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}) 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) 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 { 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)
}
}
}
@@ -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())
}
@@ -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)
}
}
@@ -545,3 +545,99 @@ func TestTellerFillsAnAskFromTheViewsAnswerOnce(t *testing.T) { // 1297/B10 AC
t.Error("the merge's ask was answered from a view key") 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())
}
}
@@ -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). // 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) failed := make(chan error, 2)
assignerKV, viewKV := optionalBucket(ctx, js, delivery.AssignerAsksBucket, logger), optionalBucket(ctx, js, delivery.ViewAnswersBucket, logger) 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 { if assignerKV != nil && viewKV != nil {
t.Assigner, t.View = kvGate{ctx: ctx, kv: assignerKV}, kvGate{ctx: ctx, kv: viewKV}
go func() { go func() {
failed <- watchAsks(ctx, viewKV, func(key string, _ uint64, _ []byte) { t.OnViewAnswer(key) }, func(string) {}, nil) 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() { go func() {
if err := watchAsks(ctx, gate, t.OnAsk, t.OnAskGone, t.OnCurrent); err != nil { 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 { func optionalBucket(ctx context.Context, js jetstream.JetStream, name string, logger *log.Logger) jetstream.KeyValue {
kv, err := js.KeyValue(ctx, name) kv, err := js.KeyValue(ctx, name)
if err != nil { 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 nil
} }
return kv return kv
@@ -340,6 +340,9 @@ func (t *Teller) Answer(o asks.Option, w asks.Warrant) error {
if srec == nil { if srec == nil {
return fmt.Errorf("the teller holds no gate key for the ask %s", w.Ask) 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) raw, found, err := t.Gate.Get(s.Key)
if err != nil { if err != nil {
return fmt.Errorf("the gate's ask %s could not be read: %w", s.Key, err) 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 { if err != nil {
return err return err
} }
// Compare-and-set: created when the key has no answer; updated at the revision read when it holds an earlier asking's return t.writeAnswer(s.Key, w.Ask, body, w.Says())
// answer (the gate ignores that one); left when it holds this asking's. }
akey := delivery.AnswerKey(s.Key)
// 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) cur, crev, found, err := t.Answers.Get(akey)
if err != nil { 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 { if found && answeredAsking(cur) == askID {
t.logf("%s: an answer is there already, and is not written again", s.Key) t.logf("%s: an answer is there already, and is not written again", key)
return nil return nil
} }
written, err := t.Answers.Write(akey, body, crev, found) written, err := t.Answers.Write(akey, body, crev, found)
if err != nil { 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 { 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 return nil
} }
t.logf("%s: %s", s.Key, w.Says()) t.logf("%s: %s", key, words)
return nil return nil
} }
@@ -423,10 +431,10 @@ func (t *Teller) saidOnce(h asker.Handled, w asks.Warrant, err error) {
t.clear(w.Ask) t.clear(w.Ask)
case h == asker.Refused || (h == asker.Acted && err != nil): 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.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: case h == asker.Recorded && w.Outcome == asks.OutcomeRefused:
t.logf("the router refused the ask %s: %s", w.Ask, w.Words) 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: case err != nil:
t.logf("the warrant of %s: %v", w.Ask, err) 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. // conditionKey is the key of the condition about an ask.
func conditionKey(id string) string { return delivery.TellerModule + ".merge-ask." + id } 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) { func (t *Teller) raise(askID, headline, why string) {
if t.Say == nil { if t.Say == nil {
return 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) t.logf("the condition of %s could not be kept: %v", id, err)
} }
needs := "to ask for the Approve again" k := t.kindOf(askID)
c := asks.Condition{Key: conditionKey(id), Severity: asks.SeverityWarning, Subject: asks.Scope{Scope: "merge", ID: id}, State: asks.StateRaised, needs := k.needs
Summary: "The merge ask " + id + ": " + why, Headline: headline, Needs: needs, c := asks.Condition{Key: conditionKey(id), Severity: asks.SeverityWarning, Subject: asks.Scope{Scope: k.scope, ID: id}, State: asks.StateRaised,
Explanation: asks.NeedsYou + needs + ". The merge ask " + id + ", " + why + ". Nothing was merged.", Summary: k.noun + id + ": " + why, Headline: headline, Needs: needs,
Resolved: "A good answer to the merge ask was written."} Explanation: asks.NeedsYou + needs + ". " + k.noun + id + ", " + why + ". " + k.nothing,
Resolved: k.resolved}
if err := t.Say(c); err != nil { if err := t.Say(c); err != nil {
t.logf("the condition of %s could not be said: %v", id, err) t.logf("the condition of %s could not be said: %v", id, err)
_ = t.Store.Delete(condKey(id)) _ = t.Store.Delete(condKey(id))
@@ -477,7 +510,7 @@ func (t *Teller) clear(askID string) {
if err != nil || e == nil { if err != nil || e == nil {
return 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 { if err := t.Say(c); err != nil {
t.logf("the condition of %s could not be cleared: %v", id, err) t.logf("the condition of %s could not be cleared: %v", id, err)
return return
@@ -578,7 +611,7 @@ func (t *Teller) roomRefused(w asks.Warrant) {
return 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.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 return
} }
if t.keepWaiting(ws, rec) != nil { if t.keepWaiting(ws, rec) != nil {
@@ -675,6 +708,11 @@ func (t *Teller) askWaiting(n int) {
i++ i++
continue 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) a, why := t.stillAsked(x)
if why != "" { if why != "" {
t.logf("%s: the ask %s, waiting for room, is not asked again: %s", x.Key, x.ID, 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 return
} }
ws[i].Asked = true 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) 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 { if n--; n <= 0 {
return return
@@ -701,6 +742,9 @@ func (t *Teller) stillAsked(x waitingAsk) (asks.Ask, string) {
if x.Key == "" { if x.Key == "" {
return asks.Ask{}, "the teller holds no gate key for it" 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) raw, found, err := t.Gate.Get(x.Key)
if err != nil { if err != nil {
return asks.Ask{}, "the gate's ask could not be read: " + err.Error() 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, "" 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 // 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. // publishes it under the bus message id given.
func (t *Teller) askAgain(a asks.Ask, msgID string) error { func (t *Teller) askAgain(a asks.Ask, msgID string) error {
@@ -25,7 +25,8 @@ func (t *Teller) OnViewAnswer(key string) {
t.fillFromView(askKey) 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) { func (t *Teller) OnAssignerAsk(key string) {
t.mu.Lock() t.mu.Lock()
defer t.mu.Unlock() defer t.mu.Unlock()
@@ -34,6 +35,7 @@ func (t *Teller) OnAssignerAsk(key string) {
return return
} }
t.fillFromView(key) 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 // 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) t.logf("%s: another writer answered first, and the view's answer is not written", askKey)
default: default:
t.logf("%s: %s", askKey, words) 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
} }
} }
@@ -10,8 +10,9 @@ import (
const fpHeld = "src2:2222222222222222222222222222222222222222222222222222222222222222" 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 // holdAsk is a key of the assigner's asks: the ask about a held fingerprint, shaped as the assigner composes it
// none yet; the value is `{"v":1,"ask":<the router's ask>}`). // (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":<the router's ask>}`).
func holdAsk(t *testing.T) (key string, value []byte, ask asks.Ask) { func holdAsk(t *testing.T) (key string, value []byte, ask asks.Ask) {
t.Helper() t.Helper()
key, err := delivery.HoldAskKey("plex", fpHeld) 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, 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", 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}) value, err = delivery.EncodeAssignerAsk(key, delivery.AssignerAsk{Ask: ask})
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)