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
@@ -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")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
// 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()
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
|
||||
// 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).
|
||||
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
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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":<the router's 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":<the router's 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)
|
||||
|
||||
Reference in New Issue
Block a user