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

The indexer says take-in.taken for every wait indexer_pending names (issue 1966, note 195)

Browse Source 
A merge taken in with no take-in.pending said for it (read from indexer_pending at a
watchman restart, or said pending by the own-build-unknown path) ended with no taken
event, so the watchman raised take-in-waits for a wait nothing held. takeInWholeBy now
says taken for each wait WaitsOf names for the claimed entries (a revision by its oldest
merge's key, a group by its g. key) as well as for its said waits, without saying a key
twice. Drops are unchanged.

Builds: design 47 "Taken in by the indexer it names" (A wait is bounded, said and recoverable); 1966/AC195-1, AC195-2, AC195-3, AC195-4

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
jschoubben
2026-11-10 23:45:46 +01:00
parent a6bba2af80
commit c2e466e16c
3 changed files with 73 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-indexer/internal/index/pending.go
+32 -1
View File
@@ -1086,13 +1086,44 @@ 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) and every wait
// indexer_pending names the entries by (PendingWaits: a revision by its oldest merge's key, a group by its g. key) not
// already ended by one of those, 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 _, 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
+32 -4
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)}; !slices.Equal(b.events(), want) || len(b.keys()) != 0 {
t.Errorf("said %v, pending %v: want %v once", 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"))
@@ -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)}; !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/internal/watch/findings_test.go
+9
View File
@@ -105,6 +105,15 @@ func TestTheWatchmansFindings(t *testing.T) {
}
w.TakeInCaughtUp()
}, map[string]string{FindingKey(ScopeDelivery, revKey, KindTakeInWaits): Warning}},
{"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)))