diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go index d381868..e5852de 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go @@ -348,7 +348,7 @@ func TestAHeldFingerprintIsAskedAbout(t *testing.T) { } 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) } } @@ -582,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) { @@ -809,3 +816,47 @@ func TestAHoldIsLiftedOnceEveryNodeRunsItHealthy(t *testing.T) { } t.Fatalf("the hold was not lifted: held %v, ask %v", b.hasKey(delivery.AssignerWantsBucket, hkey), b.hasKey(delivery.AssignerAsksBucket, akey)) } + +// 1906/H2, whole on the bus: a hold lifted by itself and written again after the assigner was started again is asked under a +// new id, so the teller raises it; the answer to the first hold does not carry over (1906/H4). +func TestAHoldWrittenAgainAfterALiftAndARestartIsAskedAgain(t *testing.T) { + b := startBed(t, "") + fNew := delivery.SourcePrefix2 + strings.Repeat("b", 64) + b.fixture(t, "web", fNew, `{"module":"web","version":"1"}`) + b.compose(t, "ace", "web", fNew, 'b', delivery.JudgedNeither) + failed, _ := delivery.EncodeHave(delivery.Have{FP: fNew, Phase: delivery.PhaseRunning, Health: delivery.HealthFailed, Detail: "connection refused"}) + healthy, _ := delivery.EncodeHave(delivery.Have{FP: fNew, Phase: delivery.PhaseRunning, Health: delivery.HealthOK}) + b.putJSON(t, delivery.HaveBucket, "ace.web", failed) + b.putWant(t, "ace", "web", delivery.Want{FP: fNew, Composed: "sha256:" + strings.Repeat("b", 64)}) + _, stop := b.startStoppable(t) + hkey, _ := delivery.HeldKey("web", fNew) + akey, _ := delivery.HoldAskKey("web", fNew) + first, err := delivery.DecodeAssignerAsk(akey, b.get(t, delivery.AssignerAsksBucket, akey)) + if err != nil { + t.Fatal(err) + } + b.putJSON(t, delivery.HaveBucket, "ace.web", healthy) + for deadline := time.Now().Add(15 * time.Second); b.hasKey(delivery.AssignerWantsBucket, hkey); time.Sleep(100 * time.Millisecond) { + if time.Now().After(deadline) { + t.Fatal("the hold was not lifted") + } + } + stop() + time.Sleep(1100 * time.Millisecond) // the next hold is written in another second + b.answerTo(t, akey, "keep", first.Ask.ID) // an answer to the first hold, which the later hold must not take + b.putJSON(t, delivery.HaveBucket, "ace.web", failed) + b.start(t) + again, err := delivery.DecodeAssignerAsk(akey, b.get(t, delivery.AssignerAsksBucket, akey)) + if err != nil { + t.Fatal(err) + } + if again.Ask.ID == first.Ask.ID { + t.Fatalf("the hold written again is asked under the first id %s", first.Ask.ID) + } + b.putJSON(t, delivery.HaveBucket, "ace.web", healthy) + for deadline := time.Now().Add(15 * time.Second); b.hasKey(delivery.AssignerWantsBucket, hkey); time.Sleep(100 * time.Millisecond) { + if time.Now().After(deadline) { + t.Fatal("the stale \"keep\" of the first hold kept the second one held") + } + } +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go index b591b66..0beb8f0 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go @@ -28,9 +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 - // lifted is, by hold ask key, the holds this process lifted by itself: a hold of the same fingerprint written again is - // asked under another asking, since the router drops a second ask under an id it keeps (issue 1906). - lifted map[string]bool + // 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) { @@ -222,10 +228,10 @@ func (p *passer) holdAsk(ctx context.Context, module, fp string, fresh *delivery if err != nil || p.w.has(delivery.AssignerAsksBucket, key) { return } - asking, now := 0, time.Now() - if p.lifted[key] { - asking = int(now.Unix()) // another asking than any before it, also one of an earlier process - } + // 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) }) @@ -256,10 +262,6 @@ func (p *passer) lift(ctx context.Context, module, fp string) { p.said[c.Key+".lifted"] = true delete(p.said, c.Key+".raised") // a hold written again is said again } - if p.lifted == nil { - p.lifted = map[string]bool{} - } - p.lifted[ask] = true p.remove(ctx, delivery.AssignerWantsBucket, held) p.remove(ctx, delivery.AssignerAsksBucket, ask) } @@ -317,11 +319,21 @@ func (p *passer) restoreAsk(ctx context.Context, module, node, fp string) { p.put(ctx, delivery.AssignerAsksBucket, key, func() ([]byte, error) { return delivery.EncodeAssignerAsk(key, a) }) } -// answeredWith says whether the teller wrote an answer to the ask at key and the option chosen is option. The answer is the -// operator's, given in the view or on the phone, and stays under its key (ADR 0539 §14, §16, §19). +// answeredWith says whether the teller wrote an answer to the ask at key as it stands now and the option chosen is option. The +// answer is the operator's, given in the view or on the phone, and stays under its key (ADR 0539 §14, §16, §19); it counts only +// for the asking it answered (its warrant names that asking's id), so an answer to an ask taken away, or to an earlier hold of +// the same fingerprint, does not carry over to a later one (issue 1906). func (p *passer) answeredWith(key, option string) bool { a, ok := p.w.answer(key) - return ok && a.Option == option + if !ok || a.Option != option { + return false + } + ask, err := delivery.DecodeAssignerAsk(key, p.w.value(delivery.AssignerAsksBucket, key)) + if err != nil { + return false + } + id, _, _ := strings.Cut(a.Warrant, "@") + return id == ask.Ask.ID } // clearHeld clears, once, the condition of a hold the operator released. diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go new file mode 100644 index 0000000..8b48beb --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go @@ -0,0 +1,100 @@ +package main + +import ( + "context" + "strings" + "testing" + "time" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/delivery" + + "mesh-delivery-assigner/internal/assign" +) + +// memKV is the part of a bucket a pass writes: puts and deletes. +type memKV struct { + jetstream.KeyValue + values map[string][]byte +} + +func (m *memKV) Put(_ context.Context, key string, value []byte) (uint64, error) { + m.values[key] = value + return uint64(len(m.values)), nil +} + +func (m *memKV) Delete(_ context.Context, key string, _ ...jetstream.KVDeleteOpt) error { + delete(m.values, key) + return nil +} + +// newPasser is a pass on a fresh world, as after a start, writing to the buckets given, at the clock given. +func newPasser(kvs buckets, at time.Time) *passer { + return &passer{kvs: kvs, w: newWorld(), logf: func(string, ...any) {}, said: map[string]bool{}, answeredAt: map[string]uint64{}, + now: func() time.Time { return at }} +} + +var fpTest = delivery.SourcePrefix2 + strings.Repeat("b", 64) + +// 1906/H2: given a hold lifted by itself, when the assigner is started again and the same fingerprint is held again, then the +// hold ask is written under another asking, with an id of its own, so the teller raises it (the router drops a second ask under +// an id it keeps). +func TestAHoldAskedAgainAfterALiftAndARestartHasANewID(t *testing.T) { + asksKV, wantsKV := &memKV{values: map[string][]byte{}}, &memKV{values: map[string][]byte{}} + kvs := buckets{delivery.AssignerAsksBucket: asksKV, delivery.AssignerWantsBucket: wantsKV} + key, _ := delivery.HoldAskKey("web", fpTest) + held := &delivery.Held{Node: "ace", Words: "health failed"} + first := newPasser(kvs, time.Date(2026, 10, 25, 1, 0, 0, 0, time.UTC)) + first.holdAsk(context.Background(), "web", fpTest, held) + a1, err := delivery.DecodeAssignerAsk(key, asksKV.values[key]) + if err != nil { + t.Fatal(err) + } + first.lift(context.Background(), "web", fpTest) + if _, there := asksKV.values[key]; there { + t.Fatal("the lift left the hold ask") + } + again := newPasser(kvs, time.Date(2026, 10, 25, 2, 0, 0, 0, time.UTC)) // started again: it remembers nothing + again.holdAsk(context.Background(), "web", fpTest, held) + a2, err := delivery.DecodeAssignerAsk(key, asksKV.values[key]) + if err != nil { + t.Fatal(err) + } + if a2.Ask.ID == a1.Ask.ID { + t.Errorf("the ask written again after a lift and a restart has the id %s of the first", a1.Ask.ID) + } +} + +// 1906/H4: an answer counts only for the asking it answered: an answer to an earlier hold of the same fingerprint, or to an ask +// taken away, is not taken for the hold that stands now. +func TestAnAnswerCountsOnlyForTheAskingItAnswered(t *testing.T) { + key, _ := delivery.HoldAskKey("web", fpTest) + p := newPasser(buckets{}, time.Now()) + put := func(asking int) string { + a := assign.HoldAsk(key, "web", fpTest, "ace", "health failed", asking, time.Now()) + v, err := delivery.EncodeAssignerAsk(key, a) + if err != nil { + t.Fatal(err) + } + p.w.put(delivery.AssignerAsksBucket, key, v) + return a.Ask.ID + } + first := put(100) + v, _ := delivery.EncodeAnswer(delivery.Answer{Option: "release", Warrant: first + "@2026-10-25T01:00:00Z", Words: "the operator"}) + p.w.put(delivery.AnswersBucket, key, v) + if !p.answeredWith(key, "release") { + t.Fatal("the answer to the ask that stands is not taken") + } + if p.answeredWith(key, "keep") { + t.Error("an answer was taken for an option not chosen") + } + put(200) // the same fingerprint held again, after a lift + if p.answeredWith(key, "release") { + t.Error("the answer to an earlier hold carried over to a later one") + } + p.w.put(delivery.AssignerAsksBucket, key, nil) // the ask taken away + if p.answeredWith(key, "release") { + t.Error("an answer to an ask taken away was taken") + } +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go index e5088ad..7fbb11a 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go @@ -326,6 +326,13 @@ func (w *world) nodes(manifests map[string]catalogue.Manifest, assigned map[stri return n } +// value is what a bucket holds at a key now, nil when nothing. +func (w *world) value(bucket, key string) []byte { + w.mu.Lock() + defer w.mu.Unlock() + return w.values[bucket][key] +} + // has says whether a bucket holds a key now. func (w *world) has(bucket, key string) bool { w.mu.Lock() diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/assignerasks.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/assignerasks.go index 1997787..04f1bfc 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/assignerasks.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/assignerasks.go @@ -1,7 +1,11 @@ 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" @@ -16,6 +20,71 @@ import ( // 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) @@ -83,6 +152,10 @@ func (t *Teller) raiseAssignerAsk(key string) { 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) { @@ -90,6 +163,7 @@ func (t *Teller) raiseAssignerAsk(key string) { } return } + t.assignerAsked(id) t.logf("%s: the assigner's ask %s is asked of the operator (%s)", key, id, a.Ask.Headline) } diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/assignerasks_test.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/assignerasks_test.go index 8134392..1cbed7e 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/assignerasks_test.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/assignerasks_test.go @@ -157,3 +157,53 @@ func ackFor(a asks.Ask, option string) asks.Warrant { w.Level, w.Label = asks.Acknowledge, option return w } + +// holdAskOf is the assigner's hold ask of module plex at the n-th fingerprint, shaped as holdAsk. +func holdAskOf(t *testing.T, n int) (string, []byte, asks.Ask) { + t.Helper() + key, err := delivery.HoldAskKey("plex", "src2:"+strings.Repeat(string(rune('0'+n)), 64)) + if err != nil { + t.Fatal(err) + } + _, _, ask := holdAsk(t) + ask.ID = delivery.AskID(key) + value, err := delivery.EncodeAssignerAsk(key, delivery.AssignerAsk{Ask: ask}) + if err != nil { + t.Fatal(err) + } + return key, value, ask +} + +// 1906/H5: given seven holds at once (as on 2026-10-25), when they are raised, then at most two are open at the router, a merge +// ask is still asked, and each hold answered lets the next waiting one be asked. +func TestABurstOfHoldsLeavesRoomForAMergeAsk(t *testing.T) { + b := newBed(t) + assigner, _ := withView(b) + var held []asks.Ask + for n := 0; n < 7; n++ { + key, value, ask := holdAskOf(t, n) + assigner[key] = value + b.t.OnAssignerAsk(key) + held = append(held, ask) + } + if got := b.askedIDs(); len(got) != mostAssignerOpen || got[0] != held[0].ID || got[1] != held[1].ID { + t.Fatalf("asked %v, want the first %d holds only; log:\n%s", got, mostAssignerOpen, b.logs.String()) + } + _, merge := b.askOf(t, 1, 1) + if got := b.askedIDs(); len(got) != mostAssignerOpen+1 || got[mostAssignerOpen] != merge.Ask.ID { + t.Fatalf("asked %v, want the merge ask asked beside the holds", got) + } + if len(b.dropped) != 0 || len(b.conds) != 0 { + t.Errorf("router dropped %v, conditions %+v", b.dropped, b.conds) + } + b.deliver(t, ackFor(held[0], "release")) + got := b.askedIDs() + if len(got) != mostAssignerOpen+2 || got[len(got)-1] != held[2].ID { + t.Fatalf("asked %v after a hold was answered, want the third hold asked next; log:\n%s", got, b.logs.String()) + } + // The merge ask answered frees its own place, and no hold takes it. + b.deliver(t, warrantFor(merge.Ask, "approve")) + if got := b.askedIDs(); len(got) != mostAssignerOpen+2 { + t.Errorf("asked %v after the merge ask was answered, want no further hold", got) + } +} diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller.go index d4468b3..40f1f20 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller.go @@ -708,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) @@ -720,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