diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go index baa62b17..37bf9725 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go @@ -8,6 +8,7 @@ package main import ( "context" "encoding/json" + "errors" "fmt" "os" "strings" @@ -191,8 +192,32 @@ func heldCounts(s stdio.KeptState) func() (index.OldSectionsCount, error) { } } +// keptRead is a state the tools bundle reads, as the index package reads a store: Get and Keys; it writes nothing. +type keptRead struct{ s stdio.KeptState } + +func (k keptRead) Get(_ context.Context, key string) ([]byte, uint64, bool, error) { + e, err := k.s.Get(key) + if err != nil || e == nil { + return nil, 0, false, err + } + return e.Value, e.Revision, true, nil +} + +func (k keptRead) Keys(context.Context) ([]string, error) { return k.s.Keys() } + +func (keptRead) Create(context.Context, string, []byte) error { return errReadOnly } + +func (keptRead) Update(context.Context, string, []byte, uint64) error { return errReadOnly } + +func (keptRead) Delete(context.Context, string, uint64) error { return errReadOnly } + +var errReadOnly = errors.New("the tools bundle only reads this state") + func main() { - if err := stdio.Serve("", append(Tools(askTakeIn(stdio.State(index.TakeInsState), 25*time.Second), time.Now, stdio.State(index.VerdictsState), heldCounts(stdio.State(index.OldSectionsState))), catalogTools()...)); err != nil { + ask := askTakeIn(stdio.State(index.TakeInsState), 25*time.Second) + tools := append(Tools(ask, time.Now, stdio.State(index.VerdictsState), heldCounts(stdio.State(index.OldSectionsState))), catalogTools()...) + tools = append(tools, index.PendingTools(keptRead{stdio.State(index.PendingState)}, ask, time.Now)...) + if err := stdio.Serve("", tools); err != nil { fmt.Fprintf(os.Stderr, "[mesh-delivery-indexer-tools] %v\n", err) os.Exit(1) } diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go index 7a98edb7..2ef5b43e 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go @@ -16,6 +16,8 @@ type takerIn interface { TakeIn(ctx context.Context, module string) index.TakenIn TakeInRepository(ctx context.Context, repository string) index.TakenIn JudgeAgain(ctx context.Context, module, digest string) index.TakenIn + TakeInPendingByHand(ctx context.Context, key, by, why string) index.TakenIn + DropPending(ctx context.Context, key, by, why string) index.TakenIn } // takenIn answers one value of the takeins state (the tool indexer_take_in, issue 1945): a value not yet answered takes the @@ -34,7 +36,13 @@ func takenIn(ctx context.Context, ix takerIn, kv jetstream.KeyValue, e jetstream return } asked, r := "take-in", index.TakenIn{} - if a.Repository != "" { + if a.Pending != "" && a.Drop { + asked, r = "drop of a pending merge", ix.DropPending(ctx, a.Pending, a.By, a.Why) + a.Module = a.Pending + } else if a.Pending != "" { + asked, r = "hand take-in of what waits", ix.TakeInPendingByHand(ctx, a.Pending, a.By, a.Why) + a.Module = a.Pending + } else if a.Repository != "" { asked, r = "take-in of a repository", ix.TakeInRepository(ctx, a.Repository) a.Module = a.Repository } else if a.Judge { diff --git a/modules/mesh-delivery-indexer/internal/index/candidates.go b/modules/mesh-delivery-indexer/internal/index/candidates.go index 85d1c415..75b6ea90 100644 --- a/modules/mesh-delivery-indexer/internal/index/candidates.go +++ b/modules/mesh-delivery-indexer/internal/index/candidates.go @@ -121,6 +121,12 @@ type Indexer struct { ownMu sync.Mutex // guards ownFP ownFP string // the fingerprint of the build this program runs, asked at the start pass (learnOwnBuild) judgePending atomic.Bool // the start's judge pass failed: judged again at the next message (verdicts.go) + // Pending is the state of merges held for the indexer their revision names (pending.go, novox/hq ADR 0633 §1); nil + // takes every merge in at once, as before the take-in step. + Pending Store + // SayTakeIn publishes a take-in event on the indexer's own stream take-in (ADR 0633 §5); nil says none. + SayTakeIn func(ctx context.Context, t delivery.TakeIn) error + pendingMu sync.Mutex // one take-in step at a time: a merge, a group's end, the start and a hand act (pending.go) } func (ix *Indexer) logf(format string, args ...any) { diff --git a/modules/mesh-delivery-indexer/internal/index/overlay.go b/modules/mesh-delivery-indexer/internal/index/overlay.go new file mode 100644 index 00000000..6405f0a4 --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/overlay.go @@ -0,0 +1,110 @@ +package index + +import ( + "context" + "maps" + "slices" + "sync" +) + +// overlayStore is a store whose writes stay in memory over the store it reads through (novox/hq ADR 0633 §1): a merge +// applied to it is computed whole, its every fingerprint and verdict, and nothing reaches the bus. A key it wrote reads +// as written, a key it deleted as absent, every other as the store under it holds it. Its compare-and-set holds against +// what it reads: a key it wrote has a revision of its own, counted above every revision under it. A nil store under +// it holds nothing. +type overlayStore struct { + under Store + mu sync.Mutex + vals map[string]overlaid + next uint64 +} + +type overlaid struct { + value []byte + revision uint64 + deleted bool +} + +func newOverlay(under Store) *overlayStore { + return &overlayStore{under: under, vals: map[string]overlaid{}, next: 1 << 62} +} + +func (o *overlayStore) Get(ctx context.Context, key string) ([]byte, uint64, bool, error) { + o.mu.Lock() + v, ok := o.vals[key] + o.mu.Unlock() + if ok { + if v.deleted { + return nil, v.revision, false, nil + } + return slices.Clone(v.value), v.revision, true, nil + } + if o.under == nil { + return nil, 0, false, nil + } + return o.under.Get(ctx, key) +} + +// put checks the revision read against what the key holds now and writes it. +func (o *overlayStore) put(ctx context.Context, key string, value []byte, revision uint64, create, del bool) error { + _, rev, found, err := o.Get(ctx, key) + if err != nil { + return err + } + o.mu.Lock() + defer o.mu.Unlock() + switch { + case create && found, !create && (!found || rev != revision): + return ErrConflict + } + o.next++ + o.vals[key] = overlaid{value: slices.Clone(value), revision: o.next, deleted: del} + return nil +} + +func (o *overlayStore) Create(ctx context.Context, key string, value []byte) error { + return o.put(ctx, key, value, 0, true, false) +} + +func (o *overlayStore) Update(ctx context.Context, key string, value []byte, revision uint64) error { + return o.put(ctx, key, value, revision, false, false) +} + +func (o *overlayStore) Delete(ctx context.Context, key string, revision uint64) error { + return o.put(ctx, key, nil, revision, false, true) +} + +func (o *overlayStore) Keys(ctx context.Context) ([]string, error) { + var keys []string + if o.under != nil { + var err error + if keys, err = o.under.Keys(ctx); err != nil { + return nil, err + } + } + o.mu.Lock() + defer o.mu.Unlock() + keys = slices.DeleteFunc(keys, func(k string) bool { _, ok := o.vals[k]; return ok }) + for k, v := range o.vals { + if !v.deleted { + keys = append(keys, k) + } + } + slices.Sort(keys) + return keys, nil +} + +// written is what the overlay holds of key: its value, and whether it wrote or deleted it at all. +func (o *overlayStore) written(key string) (value []byte, deleted, ok bool) { + o.mu.Lock() + defer o.mu.Unlock() + v, ok := o.vals[key] + return slices.Clone(v.value), v.deleted, ok +} + +// changed is every key the overlay wrote or deleted, in order. +func (o *overlayStore) changed() []string { + o.mu.Lock() + defer o.mu.Unlock() + return slices.Sorted(maps.Keys(o.vals)) +} diff --git a/modules/mesh-delivery-indexer/internal/index/pending.go b/modules/mesh-delivery-indexer/internal/index/pending.go new file mode 100644 index 00000000..1fdb7977 --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/pending.go @@ -0,0 +1,837 @@ +package index + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "maps" + "slices" + "strings" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/process" + + "git.novox.be/novox/mesh-delivery/sdk/graph" +) + +// The take-in step (novox/hq ADR 0633): a source revision is taken in only by the indexer it names. A merged message is +// kept in the indexer's own state pending, one key per merge, until the take-in that ends it; a revision naming another +// build of the indexer writes only what building and moving that indexer needs, and the rest stays pending for it. +const ( + // PendingState is the indexer's state of held merges: the bucket mesh-delivery-indexer_pending, keyed + // delivery.PendingKey, so the key order is the merge order. An old `state` entry (the indexer is foundation, ADR + // 0595 §11), history 1, reached by the indexer alone. + PendingState = "pending" + PendingBucket = delivery.IndexerModule + "_" + PendingState + // PendingTool reads it (ADR 0633 §5); TakeInPendingTool and DropPendingTool steer it. + PendingTool = "indexer_pending" + TakeInPendingTool = "indexer_take_in_pending" + DropPendingTool = "indexer_drop_pending" +) + +// errClaimLost says another take-in claimed a pending merge first: this one took nothing in. +var errClaimLost = errors.New("another take-in claimed the merges first, so this one took nothing in") + +// NoBuild opens what a revision waits for when it gives the indexer no fingerprint: the reason follows. No indexer +// runs such a build, so the revision waits until a later merge gives it one or a hand take-in (indexer_take_in_pending). +const NoBuild = "no build: " + +// ErrOwnBuildUnknown is what a merged message answers while the indexer cannot learn the build it runs (ADR 0633 §1): +// the message is left unacknowledged, so the bus delivers it again, and nothing is written. +var ErrOwnBuildUnknown = errors.New("the build of the indexer this program runs is not known") + +// Pending is one value of the pending state: the merged message whole, as the forge bridge encoded it, the group it +// was merged in (Ended once the group's end was heard), since when it is held, the indexer build its revision waits +// for (absent while it waits for nothing yet, or behind a group being gathered), the build of the indexer that last +// wrote it, the wait it was last said under on the take-in stream (Said, SaidAt), and Taking, set by a take-in that +// claimed it: a second indexer that finds it claimed takes nothing in (ADR 0633 §1: never two take-ins of one merge). +type Pending struct { + V int `json:"v"` + Forge json.RawMessage `json:"forge"` + Group string `json:"group,omitempty"` + Ended bool `json:"ended,omitempty"` + Since time.Time `json:"since"` + WaitsFor string `json:"waits-for,omitempty"` + LastWriter string `json:"last-writer"` + Said string `json:"said,omitempty"` + SaidKind string `json:"said-kind,omitempty"` + SaidAt time.Time `json:"said-at,omitzero"` + Taking string `json:"taking,omitempty"` +} + +// PendingEntry is one key of the pending state as read: its key, revision, value and the merged message decoded. +type PendingEntry struct { + Key string + Revision uint64 + Value Pending + Forge delivery.Forge +} + +// DecodePending reads one pending value and its merged message. +func DecodePending(data []byte) (Pending, delivery.Forge, error) { + var p Pending + if err := json.Unmarshal(data, &p); err != nil { + return p, delivery.Forge{}, fmt.Errorf("not a pending value: %w", err) + } + if p.V != 1 { + return p, delivery.Forge{}, fmt.Errorf("a pending value of version %d, not 1", p.V) + } + f, err := delivery.DecodeForge(p.Forge) + if err != nil { + return p, f, fmt.Errorf("its merged message: %w", err) + } + if f.Kind != delivery.ForgeMerged { + return p, f, fmt.Errorf("a pending value holds a %s message, not a merged one", f.Kind) + } + return p, f, nil +} + +func encodePending(p Pending) ([]byte, error) { + p.V = 1 + return json.Marshal(p) +} + +// ReadPending is every value the pending state holds, in key order (the merge order); a key that does not read is +// said through logf and left. +func ReadPending(ctx context.Context, st Store, logf func(string, ...any)) ([]PendingEntry, error) { + keys, err := st.Keys(ctx) + if err != nil { + return nil, fmt.Errorf("listing %s: %w", PendingState, err) + } + slices.Sort(keys) + var out []PendingEntry + for _, key := range keys { + value, rev, found, err := st.Get(ctx, key) + if err != nil { + return nil, fmt.Errorf("%s.%s: %w", PendingState, key, err) + } + if !found { + continue + } + p, f, err := DecodePending(value) + if err != nil { + if logf != nil { + logf("%s.%s is left: %v", PendingState, key, err) + } + continue + } + out = append(out, PendingEntry{Key: key, Revision: rev, Value: p, Forge: f}) + } + return out, nil +} + +// Waits is the pending state read as what waits (ADR 0633 §3, §4): Ready are the merges no group still being gathered +// holds back, in merge order, which are one revision together; Groups are the groups still being gathered, each with +// its members held so far; Behind are the merges in a gathered group's member repositories that wait behind it, by the +// group's take-in key. +type Waits struct { + Ready []PendingEntry + Groups map[string][]PendingEntry + Behind map[string][]PendingEntry +} + +// WaitsOf works out what waits from the entries, in key order. A member of a group whose end was not heard is +// gathered; from the first such member on, a later merge in one of the group's members' repositories waits behind it, +// because its tree holds that member's commit. Every other merge is ready. +func WaitsOf(entries []PendingEntry) Waits { + w := Waits{Groups: map[string][]PendingEntry{}, Behind: map[string][]PendingEntry{}} + blocked := map[string]string{} // repository, lower case, to the group key it waits behind + for _, e := range entries { + if e.Value.Group != "" && !e.Value.Ended { + w.Groups[e.Value.Group] = append(w.Groups[e.Value.Group], e) + for _, m := range e.Forge.Members { + if _, ok := blocked[strings.ToLower(m.Repository)]; !ok { + blocked[strings.ToLower(m.Repository)] = delivery.GroupTakeInKey(e.Value.Group) + } + } + continue + } + if g, ok := blocked[strings.ToLower(e.Forge.Repository)]; ok { + w.Behind[g] = append(w.Behind[g], e) + continue + } + w.Ready = append(w.Ready, e) + } + return w +} + +// mergesOf are the merges of entries, oldest first. +func mergesOf(entries []PendingEntry) []delivery.GroupMerge { + out := make([]delivery.GroupMerge, len(entries)) + for i, e := range entries { + out[i] = delivery.GroupMerge{Repository: e.Forge.Repository, Number: e.Forge.Number, Merge: e.Forge.Merge} + } + return out +} + +func sinceOf(entries []PendingEntry) time.Time { + var at time.Time + for _, e := range entries { + if at.IsZero() || e.Value.Since.Before(at) { + at = e.Value.Since + } + } + return at +} + +// groupWait is the take-in body of a group being gathered. +func groupWait(kind, group string, held []PendingEntry) delivery.TakeIn { + t := delivery.TakeIn{Kind: kind, Key: delivery.GroupTakeInKey(group), What: delivery.TakeInGroup, Group: group, + Merges: mergesOf(held), Since: sinceOf(held)} + if len(held) > 0 { + t.Members = held[len(held)-1].Forge.Members + } + return t +} + +// revisionWait is the take-in body of a revision: its key is its oldest merge's. +func revisionWait(kind string, ready []PendingEntry, waitsFor string) delivery.TakeIn { + return delivery.TakeIn{Kind: kind, Key: ready[0].Key, What: delivery.TakeInRevision, Merges: mergesOf(ready), + WaitsFor: waitsFor, Since: sinceOf(ready)} +} + +// ownBuildNow is the build this program runs: the one learnt, else asked of the node-engine now. +func (ix *Indexer) ownBuildNow(ctx context.Context) string { + ix.ownMu.Lock() + fp := ix.ownFP + ix.ownMu.Unlock() + if fp != "" || ix.OwnBuild == nil { + return fp + } + ix.learnOwnBuild(ctx) + ix.ownMu.Lock() + defer ix.ownMu.Unlock() + return ix.ownFP +} + +// sayTakeIn publishes a take-in event and keeps what it said on the entries, so indexer_pending reads the last event of +// each wait from the state. A publish that fails is said in the log: the next evaluation says it again. +func (ix *Indexer) sayTakeIn(ctx context.Context, t delivery.TakeIn) bool { + if ix.SayTakeIn == nil { + return true + } + if err := ix.SayTakeIn(ctx, t); err != nil { + ix.logf("take-in: %s.%s is not said: %v", t.Kind, t.Key, err) + return false + } + return true +} + +// markSaid writes on each entry the wait it was said under, by compare-and-set; a lost one is left (its next +// evaluation says it again). +func (ix *Indexer) markSaid(ctx context.Context, entries []PendingEntry, t *delivery.TakeIn, change func(*Pending)) { + for _, e := range entries { + p := e.Value + if t != nil { + p.Said, p.SaidKind, p.SaidAt = t.Key, t.Kind, ix.now() + } + if change != nil { + change(&p) + } + data, err := encodePending(p) + if err != nil { + continue + } + if err := ix.Pending.Update(ctx, e.Key, data, e.Revision); err != nil { + ix.logf("take-in: %s.%s is not written again: %v", PendingState, e.Key, err) + } + } +} + +// Arrive takes one merged message of the forge bridge's stream in under ADR 0633: seq is its stream sequence. It is +// written to pending first (a redelivered message finds its key there and writes no second), and then what pending +// holds is evaluated (Evaluate): taken in whole when its revision names the build this program runs, or held for the +// indexer it names. It answers what was taken in, and a line saying what waits. A message is never written while the +// build this program runs is not known (ErrOwnBuildUnknown). +func (ix *Indexer) Arrive(ctx context.Context, seq uint64, f delivery.Forge) (Applied, string, error) { + if ix.Pending == nil || ix.Modules == nil { + a, err := ix.ApplyMessage(ctx, f) + return a, "", err + } + own := ix.ownBuildNow(ctx) + if own == "" { + return Applied{}, "", ErrOwnBuildUnknown + } + ix.pendingMu.Lock() + defer ix.pendingMu.Unlock() + raw, err := delivery.EncodeForge(f) + if err != nil { + return Applied{}, "", fmt.Errorf("%w: %v", ErrUnreadable, err) + } + key := delivery.PendingKey(seq) + data, err := encodePending(Pending{Forge: raw, Group: f.Group, Since: ix.now(), LastWriter: own}) + if err != nil { + return Applied{}, "", err + } + if err := ix.Pending.Create(ctx, key, data); err != nil && !errors.Is(err, ErrConflict) { + return Applied{}, "", fmt.Errorf("%s.%s: %w", PendingState, key, err) + } + if f.Group != "" { + entries, err := ReadPending(ctx, ix.Pending, ix.logf) + if err != nil { + return Applied{}, "", err + } + if held := WaitsOf(entries).Groups[f.Group]; len(held) > 0 { + t := groupWait(delivery.TakeInPending, f.Group, held) + if ix.sayTakeIn(ctx, t) { + ix.markSaid(ctx, held, &t, nil) + } + } + } + return ix.evaluate(ctx, own, false) +} + +// EndGroup takes a group's end in (ADR 0633 §4): once every member it names is held, the group's members are marked +// ended, the group's wait is said taken (take-in.taken under its group key), and pending is evaluated, the group now +// one revision. A member named and neither held nor taken in yet answers an error, so the end is delivered again once +// the member arrived. An end of a group taken in already (every member's merge at or behind its repository's taken +// merge) changes nothing and is said: it is never delivered again for ever. +func (ix *Indexer) EndGroup(ctx context.Context, end delivery.ForgeGroupEnd) (Applied, string, error) { + if ix.Pending == nil { + return Applied{}, "no pending state: the group's merges were taken in one by one", nil + } + own := ix.ownBuildNow(ctx) + if own == "" { + return Applied{}, "", ErrOwnBuildUnknown + } + ix.pendingMu.Lock() + defer ix.pendingMu.Unlock() + entries, err := ReadPending(ctx, ix.Pending, ix.logf) + if err != nil { + return Applied{}, "", err + } + var held []PendingEntry + for _, e := range entries { + if e.Value.Group == end.Group { + held = append(held, e) + } + } + isHeld := func(m delivery.GroupMerge) bool { + return slices.ContainsFunc(held, func(e PendingEntry) bool { + return delivery.SameRepository(e.Forge.Repository, m.Repository) && e.Forge.Number == m.Number + }) + } + var missing []string + for _, m := range end.Members { + if isHeld(m) { + continue + } + taken, err := ix.takenAlready(ctx, m) + if err != nil { + return Applied{}, "", err + } + if !taken { + missing = append(missing, fmt.Sprintf("%s#%d", m.Repository, m.Number)) + } + } + if len(missing) > 0 { + return Applied{}, "", fmt.Errorf("the end of the group %q names %s, not held yet: it waits for their merged messages", end.Group, strings.Join(missing, ", ")) + } + if len(held) == 0 { + ix.logf("take-in: the end of the group %q comes after its take-in: every member is taken in already, nothing changes", end.Group) + return Applied{}, "the group is taken in already", nil + } + if !slices.ContainsFunc(held, func(e PendingEntry) bool { return !e.Value.Ended }) { + return ix.evaluate(ctx, own, false) + } + // The group's wait ends first (take-in.taken under its group key); a revision it then is, pending for another + // indexer, is said pending under its oldest merge's key by the evaluation that follows. + t := groupWait(delivery.TakeInTaken, end.Group, held) + said := &t + if !ix.sayTakeIn(ctx, t) { + said = nil // said again by the take-in or the hold that follows (saidWaits) + } + ix.markSaid(ctx, held, said, func(p *Pending) { p.Ended = true }) + ix.logf("take-in: the group %q ended with %d members held: one revision now", end.Group, len(held)) + return ix.evaluate(ctx, own, false) +} + +// takenAlready says whether a group member's merge is at or behind its repository's taken merge. +func (ix *Indexer) takenAlready(ctx context.Context, m delivery.GroupMerge) (bool, error) { + if ix.Taken == nil { + return false, nil + } + key, err := delivery.RepositoryKey(m.Repository) + if err != nil { + return false, err + } + value, _, found, err := ix.Taken.Get(ctx, key) + if errors.Is(err, process.ErrNotYet) || (err == nil && !found) { + return false, nil + } + if err != nil { + return false, fmt.Errorf("taken.%s: %w", key, err) + } + cur, err := delivery.DecodeTaken(value) + if err != nil || cur.Merge == "" { + return false, nil + } + if cur.Merge == m.Merge { + return true, nil + } + return ix.older(ctx, m.Repository, m.Merge, cur.Merge) +} + +// TakeInAtStart is the take-in at the start (ADR 0633 §2), before the consumer on the forge bridge's stream starts: +// it asks the build this program runs until it is known (wait says when to ask again, false ends it), then evaluates +// pending, taking in the whole revisions that name this build; a group still being gathered stays, and a revision +// naming yet another build is held for that one. A claim left by an indexer that ran before is void here. +func (ix *Indexer) TakeInAtStart(ctx context.Context, wait func(context.Context) bool) (Applied, string, error) { + if ix.Pending == nil { + return Applied{}, "", nil + } + own := ix.ownBuildNow(ctx) + for own == "" { + ix.logf("take-in: the build this program runs is not known yet, so nothing pending is taken in; asked again") + if !wait(ctx) { + return Applied{}, "", ErrOwnBuildUnknown + } + own = ix.ownBuildNow(ctx) + } + ix.pendingMu.Lock() + defer ix.pendingMu.Unlock() + return ix.evaluate(ctx, own, true) +} + +// evaluate works out what pending's ready merges name, together (ADR 0633 §3), and takes them in or holds them. +func (ix *Indexer) evaluate(ctx context.Context, own string, start bool) (Applied, string, error) { + entries, err := ReadPending(ctx, ix.Pending, ix.logf) + if err != nil { + return Applied{}, "", err + } + w := WaitsOf(entries) + if len(w.Ready) == 0 { + return Applied{}, ix.waitingLine(w), nil + } + if !start { + for _, e := range w.Ready { + if e.Value.Taking != "" { + ix.logf("take-in: %s.%s is claimed by the take-in %s, so this indexer takes nothing in", PendingState, e.Key, e.Value.Taking) + return Applied{}, "claimed by another take-in", nil + } + } + } + names, dry, err := ix.computeOver(ctx, w.Ready) + if err != nil { + return Applied{}, "", err + } + if names == own { + a, err := ix.takeInWhole(ctx, w.Ready, own, "") + if errors.Is(err, errClaimLost) { + return a, "claimed by another take-in", nil + } + return a, ix.waitingLine(w), err + } + return Applied{}, ix.waitingLine(w), ix.holdFor(ctx, w.Ready, names, own, dry) +} + +// waitingLine says what waits after an evaluation, for the heard state. +func (ix *Indexer) waitingLine(w Waits) string { + var parts []string + for _, g := range slices.Sorted(maps.Keys(w.Groups)) { + parts = append(parts, fmt.Sprintf("the group %q gathered (%d held)", g, len(w.Groups[g]))) + } + for _, g := range slices.Sorted(maps.Keys(w.Behind)) { + parts = append(parts, fmt.Sprintf("%d merges behind %s", len(w.Behind[g]), g)) + } + return strings.Join(parts, "; ") +} + +// dryRun is an indexer computing a take-in without writing it, and the overlays of what it would have written. +type dryRun struct { + ix *Indexer + modules, graph, verdicts *overlayStore + candidates, taken, predefined, configs *overlayStore + oldSections *overlayStore +} + +// dry is an indexer over overlays of this one's states (overlay.go): what a take-in would write stays in memory and +// nothing reaches the bus but the manifests and the packages the revision publishes, which clause 1 writes all the +// same. It says no event, kicks nothing in the background, and knows what this one knows of the source, the toolchains +// and the previews. Every field of Indexer is named here (TestDryNamesEveryField holds it so). +func (ix *Indexer) dry() *dryRun { + over := func(st Store) (Store, *overlayStore) { + if st == nil { + return nil, nil + } + o := newOverlay(st) + return o, o + } + d := &dryRun{} + x := &Indexer{Manifests: ix.Manifests, Packages: ix.Packages, JudgeID: ix.JudgeID, Files: ix.Files, GoSum: ix.GoSum, + GoMod: ix.GoMod, Now: ix.Now, Builds: ix.Builds, Named: ix.Named, Trunk: ix.Trunk, Read: ix.Read, Contains: ix.Contains, + ReadOpen: nil, OwnBuild: nil, SayRefused: nil, SayAccepted: nil, SayTakeIn: nil, Pending: nil} + x.Logf = func(format string, args ...any) { ix.logf("take-in, computed and not written: "+format, args...) } + x.Store, d.candidates = over(ix.Store) + x.Modules, d.modules = over(ix.Modules) + x.Graph, d.graph = over(ix.Graph) + x.Verdicts, d.verdicts = over(ix.Verdicts) + x.Taken, d.taken = over(ix.Taken) + x.Predefined, d.predefined = over(ix.Predefined) + x.ConfigCandidates, d.configs = over(ix.ConfigCandidates) + x.OldSections, d.oldSections = over(ix.OldSections) + ix.mu.Lock() + x.source = sourceNow{rev: ix.source.rev, heads: maps.Clone(ix.source.heads)} + x.previews = maps.Clone(ix.previews) + x.toolchains = map[string]map[string]delivery.Toolchain{} + for l, by := range ix.toolchains { + x.toolchains[l] = maps.Clone(by) + } + x.moving = maps.Clone(ix.moving) + x.untaken = maps.Clone(ix.untaken) + ix.mu.Unlock() + ix.ownMu.Lock() + x.ownFP = ix.ownFP + ix.ownMu.Unlock() + d.ix = x + return d +} + +// computeOver applies the ready merges, in merge order, to a dry indexer, as one overlay: each later merge's trees +// govern (ADR 0633 §3). It answers the fingerprint the indexer's own module has once they are applied (none is "", +// with its reason), and the dry run, whose overlays hold what the take-in would write. A merge that can never be read +// is left out, as its take-in will drop it. +func (ix *Indexer) computeOver(ctx context.Context, ready []PendingEntry) (string, *dryRun, error) { + d := ix.dry() + for _, e := range ready { + if _, err := d.ix.ApplyMessage(ctx, e.Forge); err != nil { + if errors.Is(err, ErrUnreadable) { + continue + } + return "", d, fmt.Errorf("computing the take-in of %s (%s#%d): %w", e.Key, e.Forge.Repository, e.Forge.Number, err) + } + } + key, _ := delivery.ModulesKey(delivery.IndexerModule) + value, _, found, err := d.ix.Modules.Get(ctx, key) + if err != nil { + return "", d, err + } + if !found { + return NoBuild + "the indexer holds no record of itself", d, nil + } + m, err := delivery.DecodeModule(value) + if err != nil { + return "", d, nil + } + if m.FP2 == "" { + return NoBuild + m.None2, d, nil + } + return m.FP2, d, nil +} + +// claim marks the entries as taken in by this take-in, by compare-and-set on the revisions read; a lost one leaves +// every claim made and answers false, and nothing is taken in. +func (ix *Indexer) claim(ctx context.Context, entries []PendingEntry, own, by string) ([]PendingEntry, bool) { + id := fmt.Sprintf("%s at %s", own, ix.now().Format(time.RFC3339Nano)) + if by != "" { + id = by + ": " + id + } + claimed := make([]PendingEntry, 0, len(entries)) + for _, e := range entries { + p := e.Value + p.Taking, p.LastWriter = id, own + data, err := encodePending(p) + if err == nil { + err = ix.Pending.Update(ctx, e.Key, data, e.Revision) + } + if err != nil { + ix.logf("take-in: the claim of %s.%s is lost (%v): another take-in took it, so this one takes nothing in", PendingState, e.Key, err) + return nil, false + } + _, rev, _, err := ix.Pending.Get(ctx, e.Key) + if err != nil { + return nil, false + } + e.Revision, e.Value = rev, p + claimed = append(claimed, e) + } + return claimed, true +} + +// takeInWhole takes the entries in with this indexer, as merges are taken in (ApplyMessage), oldest first, then +// removes them from pending and says each wait they were said under taken. by names a hand act ("" for none). +func (ix *Indexer) takeInWhole(ctx context.Context, entries []PendingEntry, own, by string) (Applied, error) { + return ix.takeInWholeBy(ctx, entries, own, by, "", "") +} + +// takeInWholeBy is takeInWhole; who and why, of a hand act, are carried on the take-in.taken events. +func (ix *Indexer) takeInWholeBy(ctx context.Context, entries []PendingEntry, own, by, who, why string) (Applied, error) { + claimed, ok := ix.claim(ctx, entries, own, by) + if !ok { + return Applied{}, errClaimLost + } + var applied Applied + for _, e := range claimed { + a, err := ix.ApplyMessage(ctx, e.Forge) + applied.Written = append(applied.Written, a.Written...) + if a.Late != "" { + applied.Late = a.Late + } + if errors.Is(err, ErrUnreadable) { + ix.logf("take-in: %s (%s#%d) can never be read, dropped: %v", e.Key, e.Forge.Repository, e.Forge.Number, err) + continue + } + if err != nil { + return applied, fmt.Errorf("taking %s (%s#%d) in: %w", e.Key, e.Forge.Repository, e.Forge.Number, err) + } + } + for _, e := range claimed { + if err := ix.Pending.Delete(ctx, e.Key, e.Revision); err != nil { + ix.logf("take-in: %s.%s is taken in and not removed: %v", PendingState, e.Key, err) + } + } + 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) { + t.By, t.Why = who, why + ix.sayTakeIn(ctx, t) + } + return applied, nil +} + +func byWords(by string) string { + if by == "" { + return "" + } + return " (a hand act: " + by + ")" +} + +// saidWaits are the taken events of every wait the entries were said under that is not ended yet. +func saidWaits(entries []PendingEntry) []delivery.TakeIn { + by := map[string][]PendingEntry{} + for _, e := range entries { + if e.Value.Said != "" && e.Value.SaidKind == delivery.TakeInPending { + by[e.Value.Said] = append(by[e.Value.Said], e) + } + } + var out []delivery.TakeIn + for _, key := range slices.Sorted(maps.Keys(by)) { + held := by[key] + if strings.HasPrefix(key, "g.") { + out = append(out, groupWait(delivery.TakeInTaken, held[0].Value.Group, held)) + continue + } + t := revisionWait(delivery.TakeInTaken, held, held[0].Value.WaitsFor) + t.Key = key + out = append(out, t) + } + return out +} + +// holdFor keeps the ready merges pending for the indexer build they name and writes only what building and moving +// that indexer needs (ADR 0633 §1): the indexer's own module entry, its key of the module graph and the verdict on its +// manifest, from the dry run; the manifests and the packages are written by the dry run itself. The revision is said +// pending under its oldest merge's key, and a wait it was said under before by another key is said taken. +func (ix *Indexer) holdFor(ctx context.Context, ready []PendingEntry, names, own string, d *dryRun) error { + key, _ := delivery.ModulesKey(delivery.IndexerModule) + if err := copyKey(ctx, d.modules, ix.Modules, key); err != nil { + return fmt.Errorf("the indexer's own module entry: %w", err) + } + if gk, err := graph.Key(delivery.IndexerModule); err == nil { + if err := copyKey(ctx, d.graph, ix.Graph, gk); err != nil && !errors.Is(err, process.ErrNotYet) { + return fmt.Errorf("the indexer's key of the module graph: %w", err) + } + } + if value, _, found, err := d.ix.Modules.Get(ctx, key); err == nil && found { + if m, err := delivery.DecodeModule(value); err == nil && m.Manifest != "" { + if vk, err := delivery.VerdictsKey(m.Manifest); err == nil { + if err := copyKey(ctx, d.verdicts, ix.Verdicts, vk); err != nil && !errors.Is(err, process.ErrNotYet) { + return fmt.Errorf("the verdict on the indexer's manifest: %w", err) + } + } + } + } + t := revisionWait(delivery.TakeInPending, ready, names) + var ended, unmarked []PendingEntry + said, changed := false, false + for _, e := range ready { + switch { + case e.Value.Said == t.Key && e.Value.SaidKind == delivery.TakeInPending: + said = true + changed = changed || e.Value.WaitsFor != names + if e.Value.WaitsFor != names { + unmarked = append(unmarked, e) + } + case e.Value.Said != "" && e.Value.SaidKind == delivery.TakeInPending: + ended = append(ended, e) + unmarked = append(unmarked, e) + default: + unmarked = append(unmarked, e) + } + } + for _, old := range saidWaits(ended) { + ix.sayTakeIn(ctx, old) + } + if !said || changed { + ix.logf("take-in: %d merges, oldest %s, name the indexer build %s, not %s, which runs: they wait for it", len(ready), t.Key, names, own) + if !ix.sayTakeIn(ctx, t) { + ix.markSaid(ctx, unmarked, nil, func(p *Pending) { p.WaitsFor, p.LastWriter = names, own }) + return nil + } + unmarked = ready + } + ix.markSaid(ctx, unmarked, &t, func(p *Pending) { p.WaitsFor, p.LastWriter = names, own }) + return nil +} + +// copyKey writes what the overlay holds of key to the store, when it wrote or deleted it. +func copyKey(ctx context.Context, from *overlayStore, to Store, key string) error { + if from == nil || to == nil { + return nil + } + value, deleted, ok := from.written(key) + if !ok { + return nil + } + return write(ctx, to, key, func(cur []byte, found bool) ([]byte, bool, error) { + switch { + case deleted: + return nil, found, nil + case found && string(cur) == string(value): + return nil, false, nil + } + return value, false, nil + }) +} + +// PendingWaits is what indexer_pending answers of the pending state's entries (ADR 0633 §5): the ready merges as one +// revision, each group being gathered, and the merges behind a group, each with the modules it holds back and the +// last take-in event said of it. +func PendingWaits(entries []PendingEntry) []delivery.PendingWait { + w := WaitsOf(entries) + var out []delivery.PendingWait + if len(w.Ready) > 0 { + out = append(out, waitOf(w.Ready, delivery.TakeInRevision, "")) + } + for _, g := range slices.Sorted(maps.Keys(w.Groups)) { + out = append(out, waitOf(w.Groups[g], delivery.TakeInGroup, "")) + } + for _, g := range slices.Sorted(maps.Keys(w.Behind)) { + out = append(out, waitOf(w.Behind[g], delivery.TakeInRevision, g)) + } + slices.SortStableFunc(out, func(a, b delivery.PendingWait) int { return a.Since.Compare(b.Since) }) + return out +} + +func waitOf(entries []PendingEntry, what, behind string) delivery.PendingWait { + pw := delivery.PendingWait{Key: entries[0].Key, What: what, Merges: mergesOf(entries), Behind: behind, Since: sinceOf(entries), + WaitsFor: entries[len(entries)-1].Value.WaitsFor, Holds: []string{}} + if what == delivery.TakeInGroup { + pw.Group = entries[0].Value.Group + pw.Key = delivery.GroupTakeInKey(pw.Group) + pw.Members = entries[len(entries)-1].Forge.Members + pw.WaitsFor = "" + } + for _, e := range entries { + pw.Keys = append(pw.Keys, e.Key) + for _, raw := range e.Forge.Manifests { + if name := manifestModule(raw); name != "" && !slices.Contains(pw.Holds, name) { + pw.Holds = append(pw.Holds, name) + } + } + if e.Value.Said != "" && e.Value.SaidAt.After(pw.SaidAt) { + pw.Said, pw.SaidAt = delivery.TakeInStream+"."+e.Value.SaidKind+"."+e.Value.Said, e.Value.SaidAt + } + } + slices.Sort(pw.Holds) + return pw +} + +// TakeInPendingByHand is the hand take-in (indexer_take_in_pending, ADR 0633 §5): the wait named by its key, the +// ready revision (its m. key) or a group gathered so far (its g. key), is taken in with this indexer and its judge, +// whatever build it names, for when the named indexer can never run or the group's end never comes. by and why are +// said in the journal and on the take-in.taken event. What waited behind it is evaluated again after. +func (ix *Indexer) TakeInPendingByHand(ctx context.Context, key, by, why string) TakenIn { + r := TakenIn{Module: key} + if ix.Pending == nil { + r.Answer = "the pending state is not on the bus: nothing waits" + return r + } + own := ix.ownBuildNow(ctx) + ix.pendingMu.Lock() + defer ix.pendingMu.Unlock() + entries, err := ReadPending(ctx, ix.Pending, ix.logf) + if err != nil { + r.Answer = err.Error() + return r + } + w := WaitsOf(entries) + var take []PendingEntry + switch { + case strings.HasPrefix(key, "g."): + for g, held := range w.Groups { + if delivery.GroupTakeInKey(g) == key { + take = held + } + } + case len(w.Ready) > 0 && w.Ready[0].Key == key: + take = w.Ready + } + if len(take) == 0 { + r.Answer = fmt.Sprintf("nothing waits under %s: indexer_pending names the waits", key) + return r + } + ix.logf("take-in: a hand take-in of %s (%d merges) asked by %s: %s", key, len(take), by, why) + hand := fmt.Sprintf("%s (%s)", by, why) + if _, err := ix.takeInWholeBy(ctx, take, own, hand, by, why); err != nil { + r.Answer = err.Error() + return r + } + r.Stored = true + r.Answer = fmt.Sprintf("%d merges taken in by the running indexer %s", len(take), own) + if _, _, err := ix.evaluate(ctx, own, false); err != nil { + r.Answer += "; what waited behind it is not evaluated: " + err.Error() + } + return r +} + +// DropPending is the hand drop (indexer_drop_pending, ADR 0633 §5): the one pending merge under key, abandoned +// because a later merge replaced it, is removed and nothing of it is taken in. Its wait is said taken, naming the drop; +// the merges left under that wait are said pending again by the evaluation that follows. +func (ix *Indexer) DropPending(ctx context.Context, key, by, why string) TakenIn { + r := TakenIn{Module: key} + if ix.Pending == nil { + r.Answer = "the pending state is not on the bus: nothing waits" + return r + } + own := ix.ownBuildNow(ctx) + ix.pendingMu.Lock() + defer ix.pendingMu.Unlock() + entries, err := ReadPending(ctx, ix.Pending, ix.logf) + if err != nil { + r.Answer = err.Error() + return r + } + at := slices.IndexFunc(entries, func(e PendingEntry) bool { return e.Key == key }) + if at < 0 { + r.Answer = fmt.Sprintf("%s holds no merge under %s", PendingState, key) + return r + } + e := entries[at] + if err := ix.Pending.Delete(ctx, e.Key, e.Revision); err != nil { + r.Answer = fmt.Sprintf("%s.%s is not removed: %v", PendingState, key, err) + return r + } + ix.logf("take-in: %s (%s#%d at %s) dropped by hand, asked by %s: %s; nothing of it is taken in", key, e.Forge.Repository, e.Forge.Number, e.Forge.Merge, by, why) + if e.Value.Said != "" && e.Value.SaidKind == delivery.TakeInPending { + var same []PendingEntry + for _, o := range entries { + if o.Key != key && o.Value.Said == e.Value.Said { + same = append(same, o) + } + } + for _, t := range saidWaits([]PendingEntry{e}) { + t.By, t.Why, t.Dropped = by, why, true + ix.sayTakeIn(ctx, t) + } + ix.markSaid(ctx, same, nil, func(p *Pending) { p.Said, p.SaidKind = "", "" }) + } + r.Stored, r.Answer = true, fmt.Sprintf("%s dropped: %s#%d at %s is not taken in", key, e.Forge.Repository, e.Forge.Number, e.Forge.Merge) + if own != "" { + if _, _, err := ix.evaluate(ctx, own, false); err != nil { + r.Answer += "; what waits is not evaluated again: " + err.Error() + } + } + return r +} diff --git a/modules/mesh-delivery-indexer/internal/index/pending_test.go b/modules/mesh-delivery-indexer/internal/index/pending_test.go new file mode 100644 index 00000000..21e6870a --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/pending_test.go @@ -0,0 +1,321 @@ +package index + +import ( + "context" + "encoding/json" + "errors" + "reflect" + "slices" + "strings" + "testing" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// takeInBed is an indexer holding the catalogue's avahi and the indexer's own module, both with a fingerprint, its +// pending state and the take-in events it says; base is the catalogue merge it was built from. +type takeInBed struct { + ix *Indexer + mods *memStore + pending *memStore + said []delivery.TakeIn + base delivery.Forge + logs *strings.Builder +} + +func newTakeInBed(t *testing.T) *takeInBed { + t.Helper() + ix, mods, f := newFormIndexerUnmerged(t, mod{"avahi", catalogueManifest(t, "avahi")}, + mod{delivery.IndexerModule, renamed(t, "avahi", delivery.IndexerModule)}) + if err := ix.Apply(context.Background(), f); err != nil { + t.Fatal(err) + } + reportToolchain(t, ix, "mesh-build-go", "novox", "go", "go1.27.1") + reportToolchain(t, ix, "mesh-build-go", "ace", "go", "go1.27.1") + b := &takeInBed{ix: ix, mods: mods, pending: newMem(), base: f, logs: &strings.Builder{}} + ix.Verdicts = newMem() + ix.Pending = b.pending + ix.SayTakeIn = func(_ context.Context, t delivery.TakeIn) error { b.said = append(b.said, t); return nil } + logf := ix.Logf + ix.Logf = func(format string, args ...any) { + b.logs.WriteString(format + "\n") + if logf != nil { + logf(format, args...) + } + } + own := b.fp(t, delivery.IndexerModule) + if own == "" || b.fp(t, "avahi") == "" { + t.Fatalf("the bed's modules have no fingerprint") + } + ix.OwnBuild = func(context.Context) (string, error) { return own, nil } + return b +} + +func (b *takeInBed) fp(t *testing.T, module string) string { + t.Helper() + m, _ := moduleOf(t, b.mods, module) + return m.FP2 +} + +func (b *takeInBed) commit(t *testing.T, module string) string { + t.Helper() + m, _ := moduleOf(t, b.mods, module) + return m.Commit +} + +// merge is a catalogue merge at commit c whose trees of avahi and the indexer are given ("" keeps the base's). +func (b *takeInBed) merge(number int, c, avahi, indexer string) delivery.Forge { + f := b.base + f.Number, f.Merge, f.Head = number, c, c + f.Trees = map[string]string{} + for dir, tree := range b.base.Trees { + f.Trees[dir] = tree + } + if avahi != "" { + f.Trees["modules/avahi"] = avahi + } + if indexer != "" { + f.Trees["modules/"+delivery.IndexerModule] = indexer + } + return f +} + +func (b *takeInBed) arrive(t *testing.T, seq uint64, f delivery.Forge) { + t.Helper() + if _, _, err := b.ix.Arrive(context.Background(), seq, f); err != nil { + t.Fatalf("arrive %d: %v", seq, err) + } +} + +func (b *takeInBed) keys() []string { + k, _ := b.pending.Keys(context.Background()) + slices.Sort(k) + return k +} + +func (b *takeInBed) events() []string { + var out []string + for _, s := range b.said { + out = append(out, s.Kind+" "+s.Key) + } + return out +} + +func sha(c byte) string { return strings.Repeat(string(c), 40) } + +func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633) + ctx := context.Background() + t.Run("T6 a merge while the own build is unknown writes nothing", func(t *testing.T) { + b := newTakeInBed(t) + b.ix.OwnBuild = func(context.Context) (string, error) { return "", errors.New("the engine does not answer") } + _, _, err := b.ix.Arrive(ctx, 1, b.merge(20, sha('b'), "tree-avahi-2", "")) + if !errors.Is(err, ErrOwnBuildUnknown) || len(b.keys()) != 0 || b.commit(t, "avahi") == sha('b') { + t.Errorf("err %v, pending %v, avahi at %s: want ErrOwnBuildUnknown and nothing written", err, b.keys(), b.commit(t, "avahi")) + } + }) + 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, delivery.IndexerModule), b.fp(t, "avahi") + // A merge with no change to the indexer and nothing pending is taken in alone, as before. + b.arrive(t, 1, b.merge(20, sha('b'), "tree-avahi-2", "")) + 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()) + } + 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")) + named := b.fp(t, delivery.IndexerModule) + if named == own || b.fp(t, "avahi") != avahi1 || b.commit(t, "avahi") != sha('b') { + t.Fatalf("indexer change: indexer %s (own %s), avahi %s at %s: want only the indexer written", named, own, b.fp(t, "avahi"), b.commit(t, "avahi")) + } + if g := graphOf(t, b.ix, delivery.IndexerModule); g.FP != named { + t.Errorf("graph of the indexer names %s, want %s", g.FP, named) + } + if n := len(b.ix.Verdicts.(*memStore).vals); n > verdicts+1 { + t.Errorf("verdicts %d, want at most %d: only the new indexer manifest's", n, verdicts+1) + } + if want := []string{delivery.PendingKey(2)}; !reflect.DeepEqual(b.keys(), want) || !slices.Equal(b.events(), []string{"pending " + delivery.PendingKey(2)}) || b.said[0].WaitsFor != named { + t.Fatalf("pending %v, said %+v: want m.2 pending for %s", b.keys(), b.said, named) + } + // T5: the merge delivered again writes no second key and says nothing again. + b.arrive(t, 2, b.merge(21, sha('c'), "tree-avahi-3", "tree-indexer-2")) + if len(b.keys()) != 1 || len(b.said) != 1 { + t.Errorf("redelivered: pending %v, said %v", b.keys(), b.events()) + } + // A later merge joins: the newest governs and still names the new indexer. + b.arrive(t, 3, b.merge(22, sha('d'), "tree-avahi-4", "tree-indexer-2")) + if len(b.keys()) != 2 || b.commit(t, "avahi") != sha('b') || len(b.said) != 1 { + t.Fatalf("behind: pending %v, avahi at %s, said %v: want both pending and nothing said again", b.keys(), b.commit(t, "avahi"), b.events()) + } + // A revert brings the indexer back to the running build: everything is taken in, oldest first. + b.arrive(t, 4, b.merge(23, sha('e'), "tree-avahi-5", b.base.Trees["modules/"+delivery.IndexerModule])) + if len(b.keys()) != 0 || b.fp(t, delivery.IndexerModule) != 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, delivery.IndexerModule), b.commit(t, "avahi")) + } + if want := []string{"pending " + delivery.PendingKey(2), "taken " + delivery.PendingKey(2)}; !slices.Equal(b.events(), want) { + t.Errorf("said %v, want %v", b.events(), want) + } + }) + t.Run("T10 the named indexer takes the revision in at its start; a gathered group stays", func(t *testing.T) { + b := newTakeInBed(t) + b.arrive(t, 2, b.merge(21, sha('c'), "tree-avahi-3", "tree-indexer-2")) + g := b.merge(30, sha('f'), "", "") + g.Repository, g.Trees, g.Manifests, g.Locks, g.Group = "novox/mesh-sdk", map[string]string{}, nil, nil, "feat/x" + g.Members = []delivery.Head{{Repository: "novox/mesh-sdk", Number: 30, Head: g.Head}, {Repository: "novox/mesh-host", Number: 4, Head: sha('9')}} + b.arrive(t, 5, g) + named := b.fp(t, delivery.IndexerModule) + // The new indexer starts; its build is not known at first: a hand ask is answered meanwhile (R5), and the start waits. + next := &Indexer{} + *next = Indexer{Store: b.ix.Store, Modules: b.ix.Modules, Manifests: b.ix.Manifests, Graph: b.ix.Graph, Verdicts: b.ix.Verdicts, + Packages: b.ix.Packages, Builds: b.ix.Builds, Pending: b.pending, SayTakeIn: b.ix.SayTakeIn, Logf: b.ix.Logf, GoSum: b.ix.GoSum} + next.toolchains = b.ix.toolchains + asked := 0 + next.OwnBuild = func(context.Context) (string, error) { + if asked++; asked < 2 { + return "", errors.New("not yet") + } + return named, nil + } + if r := next.TakeInPendingByHand(ctx, "m.nothing", "lead@shanks", "a test"); r.Stored || !strings.Contains(r.Answer, "nothing waits") { + t.Errorf("a hand ask while the own build is unknown: %+v", r) + } + if _, _, err := next.TakeInAtStart(ctx, func(context.Context) bool { return true }); err != nil { + t.Fatal(err) + } + if asked < 2 || !reflect.DeepEqual(b.keys(), []string{delivery.PendingKey(5)}) || b.commit(t, "avahi") != sha('c') { + t.Errorf("start: asked %d, pending %v, avahi at %s: want the revision taken in, the group left", asked, b.keys(), b.commit(t, "avahi")) + } + }) + t.Run("T8 T9 T11 a group gathered until its end, a merge behind it, one alone, an end again", func(t *testing.T) { + b := newTakeInBed(t) + b.ix.Taken = newMem() + // The commits in the order they were made: one is in another when it comes first. + order := "a756bc" + b.ix.Contains = func(_ context.Context, _, commit, in string) (bool, error) { + return strings.IndexByte(order, commit[0]) <= strings.IndexByte(order, in[0]), nil + } + member := func(repo string, number int, c string) delivery.Forge { + f := b.merge(number, c, "", "") + f.Group = "feat/take-in" + f.Members = []delivery.Head{{Repository: catalogueRepo, Number: 40, Head: sha('1')}, {Repository: "novox/mesh-sdk", Number: 41, Head: sha('2')}} + if repo != catalogueRepo { + f.Repository, f.Trees, f.Manifests, f.Locks = repo, map[string]string{}, nil, nil + } + f.Head = map[int]string{40: sha('1'), 41: sha('2')}[number] + return f + } + b.arrive(t, 1, member(catalogueRepo, 40, sha('7'))) + if b.commit(t, "avahi") == sha('7') || !slices.Equal(b.events(), []string{"pending g.feat/take-in"}) { + t.Fatalf("first member: avahi at %s, said %v: want held, the group said pending", b.commit(t, "avahi"), b.events()) + } + // A merge in another repository is taken in alone; one in a member's repository waits behind the group. + alone := b.merge(50, sha('5'), "", "") + alone.Repository, alone.Trees, alone.Manifests, alone.Locks = "novox/hq", map[string]string{}, nil, nil + b.arrive(t, 2, alone) + b.arrive(t, 3, b.merge(51, sha('6'), "tree-avahi-9", "")) + if !reflect.DeepEqual(b.keys(), []string{delivery.PendingKey(1), delivery.PendingKey(3)}) { + t.Fatalf("pending %v: want the member and the merge behind it", b.keys()) + } + end := delivery.ForgeGroupEnd{Group: "feat/take-in", Members: []delivery.GroupMerge{{Repository: catalogueRepo, Number: 40, Merge: sha('7')}, {Repository: "novox/mesh-sdk", Number: 41, Merge: sha('b')}}} + if _, _, err := b.ix.EndGroup(ctx, end); err == nil { + t.Fatal("an end before its last member is taken in") + } + b.arrive(t, 4, member("novox/mesh-sdk", 41, sha('b'))) + if _, _, err := b.ix.EndGroup(ctx, end); err != nil { + t.Fatal(err) + } + 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) { + 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. + if _, words, err := b.ix.EndGroup(ctx, end); err != nil || !strings.Contains(words, "taken in already") { + t.Errorf("an end again: %q %v", words, err) + } + // T11 (the lead's row): a group whose revision changes the indexer: the group's wait is taken, then the revision pending. + b.said = nil + g := member(catalogueRepo, 40, sha('c')) + g.Group, g.Trees = "feat/two", b.merge(0, "", "", "tree-indexer-7").Trees + g.Members = g.Members[:1] + b.arrive(t, 9, g) + if _, _, err := b.ix.EndGroup(ctx, delivery.ForgeGroupEnd{Group: "feat/two", Members: []delivery.GroupMerge{{Repository: catalogueRepo, Number: 40, Merge: sha('c')}}}); err != nil { + t.Fatal(err) + } + if want := []string{"pending g.feat/two", "taken g.feat/two", "pending " + delivery.PendingKey(9)}; !slices.Equal(b.events(), want) { + t.Errorf("a group naming another indexer: said %v, want %v", b.events(), want) + } + }) + t.Run("T5 two take-ins of one pending revision: the one that loses the claim writes nothing", func(t *testing.T) { + b := newTakeInBed(t) + b.arrive(t, 2, b.merge(21, sha('c'), "tree-avahi-3", "tree-indexer-2")) + named := b.fp(t, delivery.IndexerModule) + b.ix.ownMu.Lock() + b.ix.ownFP = named + b.ix.ownMu.Unlock() + // The other indexer claims the merge between this one's read and its claim. + b.pending.beforeWrite = func() { + v := b.pending.vals[delivery.PendingKey(2)] + var p Pending + _ = json.Unmarshal(v, &p) + p.Taking = "the other indexer" + data, _ := json.Marshal(p) + b.pending.put(delivery.PendingKey(2), data) + } + if _, _, err := b.ix.TakeInAtStart(ctx, func(context.Context) bool { return false }); err != nil { + t.Fatal(err) + } + if b.commit(t, "avahi") == sha('c') || len(b.keys()) != 1 || !strings.Contains(b.logs.String(), "is lost") { + t.Errorf("avahi at %s, pending %v: want nothing taken in by the loser, and it said", b.commit(t, "avahi"), b.keys()) + } + }) + t.Run("T12 T13 indexer_pending names the wait; a hand take-in and a drop say what they ended", func(t *testing.T) { + b := newTakeInBed(t) + b.arrive(t, 2, b.merge(21, sha('c'), "tree-avahi-3", "tree-indexer-2")) + b.arrive(t, 3, b.merge(22, sha('d'), "tree-avahi-4", "tree-indexer-2")) + var answer delivery.PendingAnswer + for _, tool := range PendingTools(b.pending, nil, nil) { + if tool.Name == PendingTool { + out, err := tool.Run(map[string]any{}) + if err != nil { + t.Fatal(err) + } + answer = out.(delivery.PendingAnswer) + } + } + if len(answer.Waits) != 1 || answer.Waits[0].Key != delivery.PendingKey(2) || !slices.Equal(answer.Waits[0].Holds, []string{"avahi", delivery.IndexerModule}) || + answer.Waits[0].Said != "take-in.pending."+delivery.PendingKey(2) || len(answer.Waits[0].Keys) != 2 { + t.Fatalf("indexer_pending = %+v", answer) + } + b.said = nil + if r := b.ix.DropPending(ctx, delivery.PendingKey(2), "lead@shanks", "replaced by m.3"); !r.Stored { + t.Fatalf("drop: %+v", r) + } + if len(b.said) < 2 || !b.said[0].Dropped || b.said[0].Key != delivery.PendingKey(2) || b.said[0].By != "lead@shanks" || b.said[1].Key != delivery.PendingKey(3) { + t.Fatalf("drop said %+v: want m.2 taken (dropped, by) and m.3 pending", b.said) + } + b.said = nil + if r := b.ix.TakeInPendingByHand(ctx, delivery.PendingKey(3), "lead@shanks", "the named indexer can never run"); !r.Stored { + t.Fatalf("hand take-in: %+v", r) + } + if len(b.keys()) != 0 || b.commit(t, "avahi") != sha('d') || len(b.said) != 1 || b.said[0].Kind != delivery.TakeInTaken || b.said[0].By != "lead@shanks" { + t.Errorf("hand take-in: pending %v, avahi at %s, said %+v", b.keys(), b.commit(t, "avahi"), b.said) + } + }) + t.Run("the dry run names every field of the indexer", func(t *testing.T) { + // dry (pending.go) builds an indexer field by field: a field added to Indexer is named there, or this fails. + named := map[string]bool{} + for _, f := range []string{"Store", "Modules", "Manifests", "Predefined", "ConfigCandidates", "Taken", "Verdicts", "OldSections", + "JudgeID", "SayRefused", "SayAccepted", "Packages", "Graph", "Files", "GoSum", "GoMod", "Now", "Builds", "Named", "Trunk", + "Read", "Contains", "Logf", "ReadOpen", "OwnBuild", "Pending", "SayTakeIn"} { + named[f] = true + } + ty := reflect.TypeOf(Indexer{}) + for i := range ty.NumField() { + if f := ty.Field(i); f.IsExported() && !named[f.Name] { + t.Errorf("Indexer.%s is not named by dry", f.Name) + } + } + }) +} diff --git a/modules/mesh-delivery-indexer/internal/index/pending_tools.go b/modules/mesh-delivery-indexer/internal/index/pending_tools.go new file mode 100644 index 00000000..33c08f4c --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/pending_tools.go @@ -0,0 +1,69 @@ +package index + +import ( + "context" + "fmt" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// PendingTools are the indexer's tools of what waits for a take-in (novox/hq ADR 0633 §5): indexer_pending reads the +// pending state, and indexer_take_in_pending and indexer_drop_pending steer it by an ask in takeins, as indexer_take_in +// does. pending is the pending state as the tools bundle reads it; ask writes an ask and waits for the indexer's answer. +func PendingTools(pending Store, ask func(key string, value []byte) (TakeInAsk, error), now func() time.Time) []stdio.Tool { + str := func(args map[string]any, k string) string { s, _ := args[k].(string); return s } + hand := func(drop bool) func(map[string]any) (any, error) { + return func(args map[string]any) (any, error) { + key, value, err := NewPendingAsk(str(args, "key"), str(args, "why"), stdio.CallerOf(args), drop, now()) + if err != nil { + return nil, err + } + a, err := ask(key, value) + if err != nil { + return nil, err + } + return a.Result, nil + } + } + return []stdio.Tool{ + {Name: PendingTool, + Description: "What waits for a take-in (ADR 0633): merges the indexer holds in its pending state instead of taking them in. " + + "A revision (what revision, its key its oldest merge's m.) waits for the build of the indexer it names " + + "(waits-for): the running indexer wrote only what building and moving that indexer needs, and the new indexer takes " + + "it in at its start. A group (what group, key g.) is a delivery group whose members are gathered until the " + + "forge bridge says its end; a merge in one of its members' repositories waits behind it (behind). Each names its " + + "merges, since when it waits, the modules it holds back (holds) and the last take-in event said of it. Empty waits: " + + "every merge is taken in. The watchman raises take-in-waits past 60 minutes for a revision, 30 for a group; " + + "indexer_take_in_pending and indexer_drop_pending steer a wait.", + Input: map[string]any{"type": "object", "properties": map[string]any{}}, + Run: func(map[string]any) (any, error) { + var unread []string + entries, err := ReadPending(context.Background(), pending, func(format string, args ...any) { + unread = append(unread, fmt.Sprintf(format, args...)) + }) + if err != nil { + return nil, err + } + return delivery.PendingAnswer{Waits: PendingWaits(entries), Unread: unread}, nil + }}, + {Name: TakeInPendingTool, + Description: "Take a wait in by hand with the running indexer and its judge (ADR 0633 §5): the revision under its m. key, " + + "whatever build of the indexer it names, or a delivery group under its g. key with the members gathered so far. For " + + "when the named indexer can never run, or a group's end never comes. A hand act: who and why are said in the " + + "indexer's journal and on the take-in.taken event, which clears take-in-waits. indexer_pending names the keys.", + Input: map[string]any{"type": "object", "required": []string{"key", "why"}, "properties": map[string]any{ + "key": map[string]any{"type": "string", "description": "the wait's key, m. or g., as indexer_pending names it"}, + "why": map[string]any{"type": "string", "description": "why it is taken in by hand, in plain words"}}}, + Run: hand(false)}, + {Name: DropPendingTool, + Description: "Drop one pending merge by hand, abandoned because a later merge replaced it (ADR 0633 §5): it is removed from " + + "the pending state and nothing of it is taken in; what waited with it is worked out again. A hand act: who and why are " + + "said in the journal and on the take-in.taken event (dropped).", + Input: map[string]any{"type": "object", "required": []string{"key", "why"}, "properties": map[string]any{ + "key": map[string]any{"type": "string", "description": "the pending merge's key, m., as indexer_pending names its merges' wait"}, + "why": map[string]any{"type": "string", "description": "why it is dropped, in plain words"}}}, + Run: hand(true)}, + } +} diff --git a/modules/mesh-delivery-indexer/internal/index/source_test.go b/modules/mesh-delivery-indexer/internal/index/source_test.go index 8bb56ebd..31961771 100644 --- a/modules/mesh-delivery-indexer/internal/index/source_test.go +++ b/modules/mesh-delivery-indexer/internal/index/source_test.go @@ -51,6 +51,17 @@ func TestASourceRevisionStampsTheModulesOfTheRepositoriesHeadItNames(t *testing. } func TestAMergeHeardAfterTheSourceNamesItsHeadIsStampedWhenWritten(t *testing.T) { // 1344/AC1 (the source first) + // A source revision older than the one held is left (merged here from a test of its own, for the test budget). + t.Run("an older source revision is left", func(t *testing.T) { + ix, mods := modulesIndexer(t, rec(t, "build-1", "a", "novox/mesh-delivery", "modules/a", "main", "", "2026-10-19T09:00:00Z", aInputs)) + ctx := context.Background() + _ = ix.Apply(ctx, mergedMsg("novox/mesh-delivery", map[string]string{"modules/a": "newtree"})) + _ = ix.ApplySource(ctx, 9, sourceOf("novox/mesh-delivery", merge)) + _ = ix.ApplySource(ctx, 7, sourceOf("novox/mesh-delivery", merge)) + if m, _ := moduleOf(t, mods, "a"); m.Src != 9 { + t.Errorf("module a Src = %d, want 9", m.Src) + } + }) ix, mods := modulesIndexer(t, rec(t, "build-1", "a", "novox/mesh-delivery", "modules/a", "main", "", "2026-10-19T09:00:00Z", aInputs)) ctx := context.Background() if err := ix.ApplySource(ctx, 7, sourceOf("novox/mesh-delivery", merge)); err != nil { @@ -93,14 +104,3 @@ func TestAMergeThatNoSourceRevisionNamesIsWrittenAsToday(t *testing.T) { // 1344 t.Errorf("module a = %+v, want the new merge with Src 0", m) } } - -func TestASourceRevisionOlderThanTheOneHeldIsLeft(t *testing.T) { - ix, mods := modulesIndexer(t, rec(t, "build-1", "a", "novox/mesh-delivery", "modules/a", "main", "", "2026-10-19T09:00:00Z", aInputs)) - ctx := context.Background() - _ = ix.Apply(ctx, mergedMsg("novox/mesh-delivery", map[string]string{"modules/a": "newtree"})) - _ = ix.ApplySource(ctx, 9, sourceOf("novox/mesh-delivery", merge)) - _ = ix.ApplySource(ctx, 7, sourceOf("novox/mesh-delivery", merge)) - if m, _ := moduleOf(t, mods, "a"); m.Src != 9 { - t.Errorf("module a Src = %d, want 9", m.Src) - } -} diff --git a/modules/mesh-delivery-indexer/internal/index/takein.go b/modules/mesh-delivery-indexer/internal/index/takein.go index 579ba62b..f1982e05 100644 --- a/modules/mesh-delivery-indexer/internal/index/takein.go +++ b/modules/mesh-delivery-indexer/internal/index/takein.go @@ -33,7 +33,11 @@ type TakeInAsk struct { // Judge asks the manifests to be judged again now (indexer_judge_again) instead of the module to be taken in. Judge bool `json:"judge,omitempty"` // Digest names, with Judge, the one manifest digest of the module to judge (indexer_judge with a digest). - Digest string `json:"digest,omitempty"` + Digest string `json:"digest,omitempty"` + // Pending asks the wait under this take-in key to be taken in by hand (indexer_take_in_pending), or with Drop the + // one pending merge under it to be dropped (indexer_drop_pending), instead of a module (pending.go, ADR 0633 §5). + Pending string `json:"pending,omitempty"` + Drop bool `json:"drop,omitempty"` By string `json:"by,omitempty"` At time.Time `json:"at"` Result *TakenIn `json:"result,omitempty"` @@ -52,8 +56,8 @@ func DecodeTakeIn(data []byte) (TakeInAsk, error) { if a.V != 1 { return a, fmt.Errorf("a takeins value of version %d, not 1", a.V) } - if a.Module == "" && a.Repository == "" { - return a, errors.New("a takeins value naming no module or repository") + if a.Module == "" && a.Repository == "" && a.Pending == "" { + return a, errors.New("a takeins value naming no module, repository or pending wait") } return a, nil } @@ -82,3 +86,25 @@ func NewTakeInRepository(repository, why, by string, at time.Time) (string, []by data, err := json.Marshal(TakeInAsk{V: 1, Repository: repository, Why: strings.TrimSpace(why), By: by, At: at.UTC()}) return "repository." + key, data, err } + +// NewPendingAsk is the value indexer_take_in_pending (drop false) or indexer_drop_pending (drop true) writes: its key +// in takeins, `pending..`, and the value. key is a wait's key as indexer_pending names it, for a +// drop one pending merge's. +func NewPendingAsk(key, why, by string, drop bool, at time.Time) (string, []byte, error) { + key = strings.TrimSpace(key) + if !strings.HasPrefix(key, "m.") && !(strings.HasPrefix(key, "g.") && !drop) { + if drop { + return "", nil, errors.New("name the pending merge to drop by its key, m., as indexer_pending names it") + } + return "", nil, errors.New("name the wait to take in by its key, m. or g., as indexer_pending names it") + } + if strings.TrimSpace(why) == "" { + return "", nil, errors.New("say why (why): a hand act on what waits is said in the journal and on its take-in event") + } + act := "take-in" + if drop { + act = "drop" + } + data, err := json.Marshal(TakeInAsk{V: 1, Pending: key, Drop: drop, Why: strings.TrimSpace(why), By: by, At: at.UTC()}) + return "pending." + act + "." + key, data, err +} diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index 4911041b..d8f68d3b 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -16,6 +16,8 @@ import ( "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-delivery/sdk/graph" + + "mesh-delivery-indexer/internal/index" ) // The controller's manifest reader is mesh-controller internal/catalogue (Manifest, manifest.go: Consumes :521, @@ -28,6 +30,12 @@ type state struct { History int `json:"history,omitempty"` } +type streamDecl struct { + Name string `json:"name"` + Events []string `json:"events"` + MaxAgeDays int `json:"max-age-days,omitempty"` +} + type artifact struct { Name string `json:"name"` Kind string `json:"kind"` @@ -51,6 +59,7 @@ type manifest struct { Emits []string `json:"emits"` Consumes []string `json:"consumes"` State []state `json:"state"` + Streams []streamDecl `json:"streams"` Reads []string `json:"reads"` Invokes []string `json:"invokes"` OwnSecrets map[string]string `json:"own-secrets"` @@ -99,7 +108,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build } // 1126 note 76: heard keeps the last 64 FORGE messages the indexer handled, for indexer_heard. if want := []state{{Name: "candidates", History: 10}, {Name: "heard", History: 64}, {Name: "manifests", History: 1}, {Name: "modules", History: 20}, {Name: "predefined", History: 5}, {Name: "config-candidates", History: 5}, {Name: "taken", History: 1}, {Name: "takeins", History: 1}, {Name: "verdicts", History: 5}, {Name: "oldsections", History: 1}, {Name: delivery.PackagesState, History: delivery.PackagesHistory}, - {Name: graph.State, History: graph.History}}; !reflect.DeepEqual(m.State, want) { + {Name: graph.State, History: graph.History}, {Name: index.PendingState, History: 1}}; !reflect.DeepEqual(m.State, want) { t.Errorf("state = %+v, want %+v", m.State, want) } if want := []string{"mesh-build-go.toolchains", "mesh-build-node.toolchains", "mesh-delivery-assigner.placement", "mesh-delivery-builder.build-log", "mesh-delivery-checker.previews", "mesh-delivery-gate.source"}; !reflect.DeepEqual(sorted(m.Reads), want) { @@ -125,7 +134,9 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build // 1945: indexer_take_in, served by its tools bundle through the takeins state; the process serves none. The same bundle // reads the verdicts state (indexer_verdicts) and asks the indexer to judge again through the takeins state // (indexer_judge; issue 774, clause 3). 1966/AC11: indexer_old_sections reads the count of the old sections (state oldsections). - if want := []string{"indexer_take_in", "indexer_old_sections", "indexer_verdicts", "indexer_judge"}; !reflect.DeepEqual(m.Tools, want) { + // 1585/T12, T13 (ADR 0633 §5): indexer_pending reads the pending state; indexer_take_in_pending and indexer_drop_pending + // ask through takeins. + if want := []string{"indexer_take_in", "indexer_old_sections", "indexer_verdicts", "indexer_judge", index.PendingTool, index.TakeInPendingTool, index.DropPendingTool}; !reflect.DeepEqual(m.Tools, want) { t.Errorf("tools = %v, want %v", m.Tools, want) } // ADR 0604 §2: it holds the seat mesh-catalog, serving its four verbs from its tools bundle (which stays without a @@ -134,8 +145,13 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build !slices.Equal(m.Claims[0].Serves, []string{"modules", "module", "provides", "dependents"}) { t.Errorf("claims = %+v, want mesh-catalog with its four verbs", m.Claims) } - // No streams, no bus, no other seat (ADR 0437 §4); it says no event. - for _, k := range []string{"streams", "logs", "provides", "requires", "uses", "defines-seats", "seat", "seats", "bus"} { + // 1585/T11 (ADR 0633 §1, §5): its one stream, take-in, an old streams entry (the indexer is foundation, ADR 0595 §11), + // carries the take-in events. + if want := []streamDecl{{Name: delivery.TakeInStream, Events: []string{"take-in.**"}, MaxAgeDays: 7}}; !reflect.DeepEqual(m.Streams, want) { + t.Errorf("streams = %+v, want %+v", m.Streams, want) + } + // No other bus section, no other seat (ADR 0437 §4). + for _, k := range []string{"logs", "provides", "requires", "uses", "defines-seats", "seat", "seats", "bus"} { if _, ok := top[k]; ok { t.Errorf("the manifest declares %q: the indexer has none", k) } diff --git a/modules/mesh-delivery-indexer/module.json b/modules/mesh-delivery-indexer/module.json index 3fa9e8a7..fafd9bcc 100644 --- a/modules/mesh-delivery-indexer/module.json +++ b/modules/mesh-delivery-indexer/module.json @@ -18,7 +18,10 @@ "indexer_take_in", "indexer_old_sections", "indexer_verdicts", - "indexer_judge" + "indexer_judge", + "indexer_pending", + "indexer_take_in_pending", + "indexer_drop_pending" ], "consumes": [ "mesh-build-go.toolchain.reported", @@ -38,7 +41,11 @@ { "name": "verdicts", "history": 5 }, { "name": "oldsections", "history": 1 }, { "name": "packages", "history": 5 }, - { "name": "graph", "history": 5 } + { "name": "graph", "history": 5 }, + { "name": "pending", "history": 1 } + ], + "streams": [ + { "name": "take-in", "events": ["take-in.**"], "max-age-days": 7 } ], "emits": [ "manifest-refused",