novox/mesh-delivery
10
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request 'A take-in wait ends whoever learnt of it: taken under every key, and the watchman's seed ends what the indexer no longer holds (issue 1966)' (#202) from fix/1966-take-in-ends-every-wait into main

Browse Source
This commit was merged in pull request #202.
This commit is contained in:
mesh-tools
2026-11-10 22:50:33 +00:00
parent a6bba2af80 e82da39d29
commit 532e4d740f
5 changed files with 160 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-indexer/internal/index/pending.go
+35 -1
View File
@@ -1086,13 +1086,47 @@ func (ix *Indexer) takeInWholeBy(ctx context.Context, entries []PendingEntry, ow
}
ix.forget(claimed...)
ix.logf("take-in: %d merges taken in as one revision by %s%s, oldest %s", len(claimed), own, byWords(by), claimed[0].Key)
for _, t := range saidWaits(claimed) {
for _, t := range takenWaits(claimed) {
t.By, t.Why = who, why
ix.sayTakeIn(ctx, t)
}
return applied, nil
}
// takenWaits are the taken events of a take-in: every wait the entries were said under (saidWaits), every wait
// indexer_pending names the claimed entries by now (WaitsOf: a revision by its oldest merge's key, a group by its g. key) and
// the key of every claimed entry, since indexer_pending named it by that key earlier when it waited behind a group. So a wait
// the watchman learned of without the indexer marking it (seeded from indexer_pending, or said by the own-build-unknown
// path) ends too (design 47, Taken in by the indexer it names). No key is said twice.
func takenWaits(entries []PendingEntry) []takein.TakeIn {
out := saidWaits(entries)
said := map[string]bool{}
for _, t := range out {
said[t.Key] = true
}
w := WaitsOf(entries)
var named []takein.TakeIn
if len(w.Ready) > 0 {
named = append(named, revisionWait(takein.TakeInTaken, w.Ready, w.Ready[len(w.Ready)-1].Value.WaitsFor))
}
for _, g := range slices.Sorted(maps.Keys(w.Groups)) {
named = append(named, groupWait(takein.TakeInTaken, g, w.Groups[g]))
}
for _, g := range slices.Sorted(maps.Keys(w.Behind)) {
named = append(named, revisionWait(takein.TakeInTaken, w.Behind[g], w.Behind[g][len(w.Behind[g])-1].Value.WaitsFor))
}
for _, e := range entries {
named = append(named, revisionWait(takein.TakeInTaken, []PendingEntry{e}, e.Value.WaitsFor))
}
for _, t := range named {
if !said[t.Key] {
said[t.Key] = true
out = append(out, t)
}
}
return out
}
func byWords(by string) string {
if by == "" {
return ""
modules/mesh-delivery-indexer/internal/index/pending_test.go
+33 -5
View File
@@ -205,6 +205,33 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
t.Errorf("no pending state: err %v, avahi at %s: want the merge taken in at once", err, b.commit(t, "avahi"))
}
})
t.Run("1966/AC195-2 AC195-3 a merge said pending with its build unknown and taken in on redelivery ends that wait under its key; a revision wait is said once", func(t *testing.T) {
b := newTakeInBed(t)
ownBuild := b.ix.OwnBuild
b.ix.ownFP, b.ix.ownFixed = "", false
b.ix.OwnBuild = func(context.Context) (string, error) { return "", errors.New("the engine does not answer") }
b.ix.FixOwnBuild(ctx)
b.said = nil
if _, _, err := b.ix.Arrive(ctx, 7, b.merge(30, sha('b'), "tree-avahi-2", "")); !errors.Is(err, ErrOwnBuildUnknown) {
t.Fatal(err)
}
b.ix.OwnBuild, b.ix.ownFP, b.ix.ownFixed = ownBuild, "", false
b.ix.FixOwnBuild(ctx)
b.arrive(t, 7, b.merge(30, sha('b'), "tree-avahi-2", "")) // the stream delivers it again
if want := []string{"pending " + takein.PendingKey(7), "taken " + takein.PendingKey(7)}; !slices.Equal(b.events(), want) || len(b.keys()) != 0 {
t.Fatalf("said %v, pending %v: want %v and nothing pending", b.events(), b.keys(), want)
}
// Two merges said pending under the oldest's key (waiting for a named indexer), then taken in together: one taken, under
// that key, and no entry's key said twice.
b.said = nil
b.arrive(t, 8, b.merge(31, sha('c'), "tree-avahi-3", "tree-indexer-2"))
b.arrive(t, 9, b.merge(32, sha('d'), "tree-avahi-4", "tree-indexer-2"))
b.said = nil
b.arrive(t, 10, b.merge(33, sha('e'), "tree-avahi-5", b.base.Trees["modules/"+Module]))
if want := []string{"taken " + takein.PendingKey(8), "taken " + takein.PendingKey(9), "taken " + takein.PendingKey(10)}; !slices.Equal(b.events(), want) || len(b.keys()) != 0 {
t.Errorf("said %v, pending %v: want %v, the wait's key (m.8, said pending) once and each other entry's own", b.events(), b.keys(), want)
}
})
t.Run("T7 and T3 a merge alone, an indexer change, a merge behind it and a revert", func(t *testing.T) {
b := newTakeInBed(t)
own, avahi0 := b.fp(t, Module), b.fp(t, "avahi")
@@ -225,9 +252,10 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
t.Fatalf("taken in at the bind: %d verdicts (were %d), want the new manifest's", n, judgedBefore)
}
// A merge with no change to the indexer and nothing pending is taken in alone, as before.
if avahi1 := b.fp(t, "avahi"); avahi1 == avahi0 || len(b.keys()) != 0 || len(b.said) != 0 {
t.Fatalf("alone: avahi %s (was %s), pending %v, said %v: want taken in", avahi1, avahi0, b.keys(), b.events())
if avahi1 := b.fp(t, "avahi"); avahi1 == avahi0 || len(b.keys()) != 0 || !slices.Equal(b.events(), []string{"taken " + takein.PendingKey(1)}) {
t.Fatalf("alone: avahi %s (was %s), pending %v, said %v: want taken in, said taken under its key (1966/AC195-1)", avahi1, avahi0, b.keys(), b.events())
}
b.said = nil
avahi1, verdicts := b.fp(t, "avahi"), len(b.ix.Verdicts.(*memStore).vals)
// T4: a change to the indexer writes only its entry (with its graph key and verdict); the rest is pending.
b.arrive(t, 2, b.merge(21, sha('c'), "tree-avahi-3", "tree-indexer-2"))
@@ -259,7 +287,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
if len(b.keys()) != 0 || b.fp(t, Module) != own || b.commit(t, "avahi") != sha('e') {
t.Fatalf("revert: pending %v, indexer %s, avahi at %s: want all taken in", b.keys(), b.fp(t, Module), b.commit(t, "avahi"))
}
if want := []string{"pending " + takein.PendingKey(2), "taken " + takein.PendingKey(2)}; !slices.Equal(b.events(), want) {
if want := []string{"pending " + takein.PendingKey(2), "taken " + takein.PendingKey(2), "taken " + takein.PendingKey(3), "taken " + takein.PendingKey(4)}; !slices.Equal(b.events(), want) {
t.Errorf("said %v, want %v", b.events(), want)
}
})
@@ -459,7 +487,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
if len(b.keys()) != 0 || b.commit(t, "avahi") != sha('6') {
t.Fatalf("end: pending %v, avahi at %s: want the group and the merge behind it taken in", b.keys(), b.commit(t, "avahi"))
}
if want := []string{"pending g.feat/take-in", "pending g.feat/take-in", "taken g.feat/take-in"}; !slices.Equal(b.events(), want) {
if want := []string{"pending g.feat/take-in", "taken " + takein.PendingKey(2), "pending g.feat/take-in", "taken g.feat/take-in", "taken " + takein.PendingKey(1), "taken " + takein.PendingKey(3), "taken " + takein.PendingKey(4)}; !slices.Equal(b.events(), want) {
t.Errorf("said %v, want %v", b.events(), want)
}
// R1: the end published again after the take-in changes nothing and is not delivered again.
@@ -510,7 +538,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
if err := b.ix.PendingAgain(ctx); err != nil {
t.Fatal(err)
}
if len(b.keys()) != 0 || b.commit(t, "avahi") != sha('6') || !slices.Equal(b.events(), []string{"taken g.feat/take-in"}) {
if len(b.keys()) != 0 || b.commit(t, "avahi") != sha('6') || !slices.Equal(b.events(), []string{"taken g.feat/take-in", "taken " + takein.PendingKey(2)}) {
t.Errorf("all taken: pending %v, avahi at %s, said %v: want the group ended and the merge behind it taken in", b.keys(), b.commit(t, "avahi"), b.events())
}
held := func() *takeInBed {
modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go
+35 -2
View File
@@ -13,6 +13,8 @@ import (
"git.novox.be/novox/mesh-sdk/go/process"
"git.novox.be/novox/mesh-sdk/go/views"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein"
"mesh-delivery-watchman/internal/watch"
"github.com/nats-io/nats.go/jetstream"
@@ -77,6 +79,34 @@ func TestRunNeedsItsCredential(t *testing.T) { // 1135/AC6
if err == nil || !strings.Contains(err.Error(), "MESH_BROKER_FILE") {
t.Fatalf("run = %v, want a refusal naming MESH_BROKER_FILE", err)
}
t.Run("1966/AC195-5: a seed indexer_pending could not answer, or answered unreadably, ends nothing", func(t *testing.T) {
for name, call := range map[string]func(string, any) (json.RawMessage, error){
"cannot be asked": func(string, any) (json.RawMessage, error) { return nil, errors.New("no answer") },
"unreadable": func(string, any) (json.RawMessage, error) { return json.RawMessage(`{"waits":`), nil },
} {
out := &findingsOut{findings: map[string]bool{}}
w := watch.New(out, time.Now)
const key = "m.00000000000000000042"
old := watch.Finding{Scope: watch.ScopeDelivery, ID: key, Kind: watch.KindTakeInWaits, Severity: watch.Warning, Source: "take-in",
Since: time.Now().Add(-3 * time.Hour), Summary: "held", Said: "held", Headline: "held", Explanation: "held", Needs: "a take-in"}
value, err := watch.EncodeFinding(old)
if err != nil {
t.Fatal(err)
}
out.findings[old.Key()] = true
w.SeedFinding(old.Key(), value)
body, err := takein.EncodeTakeIn(takein.TakeIn{Kind: takein.TakeInPending, Key: "m.00000000000000000043", What: takein.TakeInRevision, Since: time.Now(), WaitsFor: "src2:x",
Merges: []takein.Merge{{Repository: "novox/a", Number: 1, Merge: "c0ffee"}}})
if err != nil {
t.Fatal(err)
}
w.TakeIn(takein.TakeInPendingEvent, body)
seedTakeIn(w, call, func(string, ...any) {})
if !out.findings[old.Key()] || out.pulseDeletes != 0 {
t.Errorf("%s: finding stands %v, signals ended %d: want the previous run's finding and the held wait kept", name, out.findings[old.Key()], out.pulseDeletes)
}
}
})
t.Run("A2j: a want stood since the assigner's bus wrote it, through the first view", func(t *testing.T) {
// The watchman's want wait starts at the entry's time (createdOr): through mesh-sdk's views.ProcessReader it is
// the time the owner's bus wrote the key, never the watchman's start.
@@ -113,11 +143,14 @@ type refusingBus struct{ err error }
func (r refusingBus) KeyValue(context.Context, string) (jetstream.KeyValue, error) { return nil, r.err }
// findingsOut keeps which findings stand.
type findingsOut struct{ findings map[string]bool }
type findingsOut struct {
findings map[string]bool
pulseDeletes int
}
func (o *findingsOut) PutPresence(string, bool) error { return nil }
func (o *findingsOut) PutPulse(string, []byte, time.Duration) error { return nil }
func (o *findingsOut) DeletePulse(string) error { return nil }
func (o *findingsOut) DeletePulse(string) error { o.pulseDeletes++; return nil }
func (o *findingsOut) PutStuck(string, []byte) error { return nil }
func (o *findingsOut) PutFinding(key string, _ []byte) error { o.findings[key] = true; return nil }
func (o *findingsOut) DeleteFinding(key string) error { delete(o.findings, key); return nil }
modules/mesh-delivery-watchman/internal/watch/findings_test.go
+30
View File
@@ -105,6 +105,36 @@ func TestTheWatchmansFindings(t *testing.T) {
}
w.TakeInCaughtUp()
}, map[string]string{FindingKey(ScopeDelivery, revKey, KindTakeInWaits): Warning}},
{"1966/AC195-5: a seed ends the finding and the held wait of a key indexer_pending no longer names, and keeps one it names with its own since", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
// A previous run's finding for revKey stands in the bucket (read at the start); a wait of staleKey is held from an event.
old := Finding{Scope: ScopeDelivery, ID: revKey, Kind: KindTakeInWaits, Severity: Warning, Source: "take-in", Since: t0.Add(-3 * time.Hour), Observer: Observer, Summary: "held", Said: "held", Headline: "held", Explanation: "held", Needs: "a take-in"}
value, err := EncodeFinding(old)
if err != nil {
t.Fatal(err)
}
o.findings[old.Key()] = old
w.SeedFinding(old.Key(), value)
const staleKey = "m.00000000000000000099"
w.TakeInCaughtUp()
w.TakeIn(takeInEvent(t, takein.TakeInPending, staleKey), takeInBody(t, takein.TakeInPending, staleKey, c.t))
w.SeedTakeIn(takein.PendingAnswer{Waits: []takein.PendingWait{{Key: groupKey, What: takein.TakeInGroup, Keys: []string{groupKey},
Merges: []takein.Merge{{Repository: "novox/a", Number: 1, Merge: "c0ffee"}}, Since: c.t.Add(-10 * time.Minute)}}})
if _, ok := o.findings[old.Key()]; ok || w.signals[signalKey("take-in", staleKey)].Finding.ID != "" || w.signals[signalKey("take-in", revKey)].Finding.ID != "" {
t.Fatalf("a wait the answer does not name stands: findings %v, signals %v", o.findings, w.signals)
}
if o.ttls[signalKey("take-in", groupKey)] != TakeInGroupBound-10*time.Minute {
t.Fatalf("the named wait: ttl %v, want the rest of its bound from its own since", o.ttls[signalKey("take-in", groupKey)])
}
}, map[string]string{}},
{"1966/AC195-4: a taken event for a key the watchman holds no wait for changes no finding and no signal", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
w.TakeInCaughtUp()
w.TakeIn(takeInEvent(t, takein.TakeInPending, groupKey), takeInBody(t, takein.TakeInPending, groupKey, t0))
ttls, pulses := len(o.ttls), len(o.pulses)
w.TakeIn(takeInEvent(t, takein.TakeInTaken, revKey), takeInBody(t, takein.TakeInTaken, revKey, t0))
if o.findingWrites != 0 || len(o.ttls) != ttls || len(o.pulses) != pulses || o.ttls[signalKey("take-in", groupKey)] != TakeInGroupBound {
t.Fatalf("a taken for an unheld key: findings written %d, ttls %v, pulses %v", o.findingWrites, o.ttls, o.pulses)
}
}, map[string]string{}},
{"ADR 0633 §5 (1585/T21): a replayed pending followed by its taken never raises", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
w.TakeIn(takeInEvent(t, takein.TakeInPending, revKey), takeInBody(t, takein.TakeInPending, revKey, t0.Add(-2*time.Hour)))
w.TakeIn(takeInEvent(t, takein.TakeInTaken, revKey), takeInBody(t, takein.TakeInTaken, revKey, t0.Add(-2*time.Hour)))
modules/mesh-delivery-watchman/internal/watch/takein.go
+27 -1
View File
@@ -52,10 +52,36 @@ func (w *Watchman) TakeIn(event string, body []byte) {
// and group the indexer holds now, as indexer_pending answers them (ADR 0633 §5): the consumer is durable, so the
// pending events it acknowledged before a restart are not heard again, and a wait within its bound would otherwise be
// lost. Each wait's bound runs from its own since, as a pending event's does; the stream's events that follow replace
// or end it.
// or end it. A wait or finding the answer does not name is ended.
func (w *Watchman) SeedTakeIn(a takein.PendingAnswer) {
w.mu.Lock()
defer w.mu.Unlock()
// A wait or finding of a key the answer does not name is ended as a take-in.taken would end it: the finding a previous
// run wrote (read at the start, before the seed), a wait held from an event, a signal standing (design 47, Taken in by
// the indexer it names). Only an answered seed gets here: one that could not ask ends nothing.
named := map[string]bool{}
for _, wait := range a.Waits {
named[wait.Key] = true
}
stale := map[string]bool{}
for _, value := range w.findings {
if f, err := DecodeFinding(value); err == nil && f.Scope == ScopeDelivery && f.Kind == KindTakeInWaits && !named[f.ID] {
stale[f.ID] = true
}
}
for _, due := range w.signals {
if due.Finding.Kind == KindTakeInWaits && due.Finding.Scope == ScopeDelivery && !named[due.Finding.ID] {
stale[due.Finding.ID] = true
}
}
for key := range w.takeInHeld {
if !named[key] {
stale[key] = true
}
}
for _, key := range slices.Sorted(maps.Keys(stale)) {
w.takeIn(takein.TakeInTaken, key, takein.TakeIn{Kind: takein.TakeInTaken, Key: key})
}
for _, wait := range a.Waits {
w.takeIn(takein.TakeInPending, wait.Key, takein.TakeIn{Kind: takein.TakeInPending, Key: wait.Key, What: wait.What,
Merges: wait.Merges, Group: wait.Group, Members: wait.Members, WaitsFor: wait.WaitsFor, Since: wait.Since})