diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/holds.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/holds.go new file mode 100644 index 00000000..c4f594ef --- /dev/null +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/holds.go @@ -0,0 +1,72 @@ +package main + +import ( + "fmt" + "strings" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/handacts" + + "git.novox.be/novox/mesh-delivery/sdk/enginereads" + + "mesh-delivery-indexer/internal/index" +) + +// holdsTool reads the indexer's mesh-wide holds (novox/hq ADR 0632 §2), asks one again now, or moves one by a person's +// word: the asks go through the takeins state, as indexer_take_in's do, and the indexer answers there. +func holdsTool(read stdio.KeptState, ask asker, now func() time.Time, acts handacts.Recorder) stdio.Tool { + str := func(args map[string]any, k string) string { s, _ := args[k].(string); return strings.TrimSpace(s) } + return stdio.Tool{Name: index.HoldsTool, + Description: "The modules whose new manifest the indexer holds from the whole mesh (ADR 0632 §2): a manifest every node-engine " + + "reads (the store holder's first) that needs a reader capability read mesh-wide (" + strings.Join(index.MeshWideReads, ", ") + + ") which not every counted node-engine states, present or away (ADR 0624 §1). Per module: what it needs, which nodes " + + "lack what (or why no node is counted yet), the manifest the modules state keeps and the one held, the merge that " + + "brought it and since when. The hold moves by itself once every counted node-engine states it (on manifest-reads-changed, " + + "the placement's change or a merge). With module and ask_again, the engines of the nodes that lack it are asked again " + + "and the hold is weighed now; with module, release and why, it moves now by your word whatever the engines state (a hand " + + "act, recorded through mesh-hand-acts: an engine that lacks the reading may lose the store). indexer_engine_reads reads " + + "what each node's engine states.", + Input: map[string]any{"type": "object", "properties": map[string]any{ + "module": map[string]any{"type": "string", "description": "one module; every hold when left out"}, + "ask_again": map[string]any{"type": "string", "description": "\"true\": ask the lacking nodes' engines again and weigh the hold now"}, + "release": map[string]any{"type": "string", "description": "\"true\": move the hold now by your word, whatever the engines state"}, + "why": map[string]any{"type": "string", "description": "why it is asked again or moved, in plain words"}}}, + Run: func(args map[string]any) (any, error) { + module := str(args, "module") + release := str(args, "release") == "true" + if release || str(args, "ask_again") == "true" { + key, value, err := index.NewHoldAgain(module, str(args, "why"), stdio.CallerOf(args), release, now()) + if err != nil { + return nil, err + } + if release { + if _, err := acts.Record(handacts.Request{What: "moved the mesh-wide hold of " + module + " by hand", + Why: str(args, "why"), Cause: index.HoldsTool, Verb: "hold-release"}); err != nil { + return nil, fmt.Errorf("the hand act was not recorded through mesh-hand-acts, so the hold was not moved: %w", err) + } + } + a, err := ask(key, value) + if err != nil { + return nil, err + } + return a.Result, nil + } + keys, err := read.Keys() + if err != nil { + return nil, fmt.Errorf("the holds state cannot be listed: %w", err) + } + return index.ListHolds(keys, func(key string) ([]byte, error) { + e, err := read.Get(key) + if err != nil || e == nil { + return nil, err + } + return e.Value, nil + }, module) + }} +} + +// engineReadsTool lists, asks again and drops what each counted node's engine reads, as the indexer keeps it. +func engineReadsTool(now func() time.Time) stdio.Tool { + return enginereads.Tool(index.EngineReadsTool, "the indexer", enginereads.KeptState(stdio.State(index.EngineReadsState)), true, now) +} 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 2fc39c4d..d1d30ab8 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 @@ -250,7 +250,15 @@ func heldCounts(s stdio.KeptState) func() (index.OldSectionsCount, error) { } 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)), handacts.Recorder{Ask: stdio.Ask}), append(append(catalogTools(), viewTools(os.Getenv, keptSource, time.Now)...), configReadersTool(stdio.NewWatchedStates()), busStateTool(keptSource), requirementsTool(keptSource), packageUsersTool(keptSource), seatDefinitionTool(keptSource), outOfStepTool(keptSource))...)); err != nil { + asks := askTakeIn(stdio.State(index.TakeInsState), 25*time.Second) + acts := handacts.Recorder{Ask: stdio.Ask} + tools := Tools(asks, time.Now, stdio.State(index.VerdictsState), heldCounts(stdio.State(index.OldSectionsState)), acts) + tools = append(tools, catalogTools()...) + tools = append(tools, viewTools(os.Getenv, keptSource, time.Now)...) + tools = append(tools, configReadersTool(stdio.NewWatchedStates()), busStateTool(keptSource), requirementsTool(keptSource), + packageUsersTool(keptSource), seatDefinitionTool(keptSource), outOfStepTool(keptSource), + holdsTool(stdio.State(index.HoldsState), asks, time.Now, acts), engineReadsTool(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/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go index b04f2850..987baddd 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -388,6 +388,20 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // Only the toolchain reports of nodes the catalogue places the build module on count (issue 1940, ADR 0549 §3): the // placement is settled first, so the toolchains settle through the gate already knowing it. gate := buildrun.NewToolchainGate(sortedModules(), ix.PutToolchain, logger.Printf) + // What each counted node's engine reads (ADR 0624 §1), for the mesh-wide hold of a manifest every node reads (ADR 0632 + // §2): asked through the node-engine seat's verb as the indexer's own account, for every node the placement names, + // kept in the state engine-reads once it is bound; each change weighs the holds again. + reads := newEngineReads(calls.CallOn, gate, func() { ix.KickHolds(ctx) }, logger.Printf) + gate.OnPlaced = reads.placed + ix.Counted, ix.AskEngines = reads.counted, reads.ask + // The holds, new with ADR 0632 (issue 1790's kind), bound when they appear: until then a merge that would hold a + // manifest is left for redelivery, said, rather than write what a counted engine cannot read. + holds := process.AwaitBucket(ctx, js, conn, index.HoldsBucket, "a merge whose manifest every node reads, needing what not every counted node-engine states, is left for redelivery", logger.Printf) + ix.Holds = awaitedStore{holds} + holds.Ready(func(jetstream.KeyValue) { ix.KickHolds(ctx) }) + process.AwaitBucket(ctx, js, conn, index.EngineReadsBucket, "what each node's engine reads is kept in memory only", logger.Printf).Ready(func(kv jetstream.KeyValue) { + reads.bind(ctx, kv, errs, logger.Printf) + }) // The placement is the assigner's: read through its first view and its views family, never its bucket (DLV-F1r). placementErrs := make(chan error) followPlacement(ctx, js, cred, calls.Call, gate, func(ctx context.Context) { @@ -496,7 +510,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // The versions each package's series holds (DLV-I5): the registry seats' published and the builder's numbered, on the // indexer's durable consumer of the shared events stream. Where it is not there yet, no series moves until the indexer // starts again, said. - hear(process.EventsStream, "no package's series moves until the indexer starts again", func(m jetstream.Msg) { + hear(process.EventsStream, "no package's series moves and no node's manifest-reads-changed is heard until the indexer starts again", func(m jetstream.Msg) { + if reads.heard(m, logger.Printf) { // the node-engine seat's manifest-reads-changed (ADR 0618 §4): its node asked again + return + } if !heardVersion(ctx, ix, m, logger.Printf) { _ = m.Ack() // nothing else the indexer consumes rides the shared stream } diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/reads.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/reads.go new file mode 100644 index 00000000..f8de008b --- /dev/null +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/reads.go @@ -0,0 +1,113 @@ +package main + +import ( + "context" + "slices" + "sync" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/process" + + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun" + "git.novox.be/novox/mesh-delivery/sdk/enginereads" + + "mesh-delivery-indexer/internal/index" +) + +// engineReads is what each counted node's engine states it reads, for the indexer's mesh-wide hold (index.Holds; novox/hq +// ADR 0632 §2, ADR 0624 §1), in the shape the builder keeps it: asked through the node-engine seat's verb manifest-reads +// for every node the placement names, on the indexer's start, on the seat's event manifest-reads-changed for that node and +// when a person asks again (indexer_engine_reads, indexer_holds); kept in the indexer's state engine-reads, one key per +// node, so a node that is away keeps its last statement. A node the placement no longer names is forgotten. Until the +// placement is read no node is counted, and every hold stays. +type engineReads struct { + keeper *enginereads.Keeper + gate *buildrun.ToolchainGate + mu sync.Mutex + known bool // the placement is read + nodes []string +} + +// newEngineReads asks through call (process.Conn.CallOn) and tells changed after each statement changed. +func newEngineReads(call enginereads.Caller, gate *buildrun.ToolchainGate, changed func(), logf func(string, ...any)) *engineReads { + return &engineReads{gate: gate, keeper: &enginereads.Keeper{Call: call, Logf: logf, Changed: changed}} +} + +// bind keeps the statements in kv and hears it: every node's last statement at the start, a person's ask again and drop +// later. Every counted node is asked again once bound, so what was asked before is kept in it too. +func (r *engineReads) bind(ctx context.Context, kv jetstream.KeyValue, errs chan<- error, logf func(string, ...any)) { + r.keeper.Put = func(node string, v []byte) error { _, err := kv.Put(ctx, node, v); return err } + r.keeper.Delete = func(node string) error { return kv.Delete(ctx, node) } + each := func(e jetstream.KeyValueEntry) { + r.keeper.Heard(e.Key(), e.Value(), e.Operation() != jetstream.KeyValuePut) + } + w, err := settle(ctx, kv, each) + if err != nil { + process.NotWatched(logf, index.EngineReadsBucket, err, index.EngineReadsTool+" is not answered until the indexer starts again") + return + } + go func() { errs <- follow(ctx, w, each) }() + r.mu.Lock() + nodes := slices.Clone(r.nodes) + r.mu.Unlock() + r.ask(nodes, "the state engine-reads is bound") +} + +// placed takes the placement in, once it is read and after each change of it: a node it names newly is asked, one it no +// longer names is forgotten, key and all. +func (r *engineReads) placed() { + nodes := r.gate.Nodes() + r.mu.Lock() + before, first := r.nodes, !r.known + r.nodes, r.known = nodes, true + r.mu.Unlock() + for _, node := range nodes { + if first || !slices.Contains(before, node) { + r.keeper.AskNode(node, "counted: the placement names it") + } + } + for _, node := range before { + if !slices.Contains(nodes, node) { + r.keeper.Forget(node) + } + } + if r.keeper.Changed != nil { + r.keeper.Changed() + } +} + +// heard takes an event of the indexer's consumer on EVENTS: manifest-reads-changed asks its node again, and answers true. +func (r *engineReads) heard(m jetstream.Msg, logf func(string, ...any)) bool { + node, ours, err := enginereads.NodeOf(m.Subject(), m.Data()) + if !ours { + return false + } + if ackErr := m.Ack(); ackErr != nil { + logf("acknowledging %s: %v", m.Subject(), ackErr) + } + if err != nil { + logf("%s holds what is not %s, and %s is asked again all the same: %v", m.Subject(), enginereads.Event, node, err) + } + r.keeper.AskNode(node, enginereads.Event) + return true +} + +// counted is index.Indexer.Counted: what each counted node states (nil, the baseline, for one with no statement), and +// whether the placement is read. +func (r *engineReads) counted() (map[string][]string, bool) { + r.mu.Lock() + nodes, known := slices.Clone(r.nodes), r.known + r.mu.Unlock() + if !known { + return nil, false + } + return r.keeper.Reads(nodes), true +} + +// ask asks these nodes' engines again (index.Indexer.AskEngines). +func (r *engineReads) ask(nodes []string, why string) { + for _, node := range nodes { + r.keeper.AskNode(node, why) + } +} 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 054cb76e..043643d0 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go @@ -18,6 +18,7 @@ type takerIn interface { JudgeAgain(ctx context.Context, module, digest string) index.TakenIn SayAcceptedAgain(ctx context.Context, module string) index.TakenIn SeatDefinitionsAgain(ctx context.Context, module string, pull int, head string) index.TakenIn + HoldAgain(ctx context.Context, module string, release bool, by 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 @@ -39,6 +40,12 @@ func takenIn(ctx context.Context, ix takerIn, kv jetstream.KeyValue, e jetstream if a.Repository != "" { asked, r = "take-in of a repository", ix.TakeInRepository(ctx, a.Repository) a.Module = a.Repository + } else if a.Hold != "" { + asked = "weighing of its hold again" + if a.Hold == index.HoldRelease { + asked = "move of its hold by hand" + } + r = ix.HoldAgain(ctx, a.Module, a.Hold == index.HoldRelease, a.By) } else if a.SeatDefinitions { asked, r = "reading of its seat definitions again", ix.SeatDefinitionsAgain(ctx, a.Module, a.Pull, a.Head) } else if a.Say { diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/views.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/views.go index d7cd5dc8..1747c4c9 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/views.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/views.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "fmt" + "sync/atomic" "time" "github.com/nats-io/nats.go/jetstream" @@ -150,6 +151,7 @@ func followPlacement(ctx context.Context, js jetstream.JetStream, cred process.A logf("views: the finding of the first view %s is not written: %v", s.Tool, err) } } + var settled atomic.Bool // the first view answered: the gate's nodes are the counted ones from now on f := &views.Follower{Owner: delivery.AssignerModule, Tool: placementTool, State: delivery.PlacementState, Ask: views.AskVia(call, placementTool), Logf: logf, Raise: func(s views.Status) { finding(s, true) }, Clear: func(s views.Status) { finding(s, false) }, @@ -157,6 +159,9 @@ func followPlacement(ctx context.Context, js jetstream.JetStream, cred process.A if gate.Placement(key, value, deleted, at) { changed(ctx) } + if settled.Load() && gate.OnPlaced != nil { + gate.OnPlaced() // the counted nodes may have changed (ADR 0624 §1) + } }} status.Ready(func(kv jetstream.KeyValue) { f.Said = views.KeepStatus(ctx, kv, logf) @@ -172,6 +177,10 @@ func followPlacement(ctx context.Context, js jetstream.JetStream, cred process.A if gate.Settled() { changed(ctx) } + settled.Store(true) + if gate.OnPlaced != nil { + gate.OnPlaced() // the placement is read: its nodes are the counted ones + } close(answered) }() // The assigner's views family, on the indexer's durable consumer, once the first view is asked; not heard, it is the diff --git a/modules/mesh-delivery-indexer/internal/index/candidates.go b/modules/mesh-delivery-indexer/internal/index/candidates.go index 53ce5221..2e474abb 100644 --- a/modules/mesh-delivery-indexer/internal/index/candidates.go +++ b/modules/mesh-delivery-indexer/internal/index/candidates.go @@ -115,6 +115,16 @@ type Indexer struct { // that finds the candidates of another pull request asks it and deletes those of a pull request no longer open, so a // closed message the indexer missed heals at the next one (issue 1126); nil asks none. ReadOpen func(ctx context.Context) ([]delivery.OpenPull, error) + // Holds is the holds state (holds.go; novox/hq ADR 0632 §2): a module's new entry whose manifest needs a reader + // capability read mesh-wide, waiting for every counted node-engine to state it. Nil, or Counted nil, holds nothing. + Holds Store + // Counted is what each counted node's engine states it reads (ADR 0624 §1; nil, the baseline, for a node with no + // statement), and whether the counted nodes are known (the placement is read). + Counted func() (map[string][]string, bool) + // AskEngines asks these nodes' engines what they read again (the node-engine seat's verb manifest-reads); nil asks none. + AskEngines func(nodes []string, why string) + holding atomic.Bool // a MoveHolds runs in the background (KickHolds) + holdAgain atomic.Bool // a kick arrived while it ran mu sync.Mutex // guards source, toolchains and previews, which the watches and the forge messages all reach source sourceNow diff --git a/modules/mesh-delivery-indexer/internal/index/holds.go b/modules/mesh-delivery-indexer/internal/index/holds.go new file mode 100644 index 00000000..c53365fc --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/holds.go @@ -0,0 +1,470 @@ +package index + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "maps" + "reflect" + "slices" + "sort" + "strings" + "time" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/process" +) + +// The indexer's mesh-wide hold (novox/hq ADR 0632 §2, ADR 0624 §1). Every node-engine reads the store holder's manifest +// wherever the store runs, to find the store (mesh-host internal/compose/frombus), and an engine before the reading +// form:seat-reach finds it only through a module serving the retired provision artifact-store. A manifest that needs a +// reader capability read mesh-wide (MeshWideReads) is therefore not written to a module's entry in the modules state +// while some counted node-engine does not state it: the entry keeps its previous manifest, fingerprint and inputs, and +// the new one waits in the state holds, one key per module, until every counted node-engine states what it needs. Then +// it is written as its merge would have written it, judged and said. The assigner's hold of ADR 0614 §4 holds a want on +// one node; this one holds the entry itself, which every node reads. +// +// "Counted" is ADR 0624 §1's set: every node the placement names, present or away, each read as its engine last stated +// through the node-engine seat's verb manifest-reads (a node with no statement read as the baseline; the indexer's state +// engine-reads, sdk/enginereads). Until the placement is read no node is counted and every hold stays. The hold moves +// on the event manifest-reads-changed, the placement's change, a person's ask again, and a merge; never on a timer. +const ( + // HoldsState is the indexer's state of held entries: the bucket mesh-delivery-indexer_holds, keyed as the modules state. + HoldsState = "holds" + HoldsBucket = delivery.IndexerModule + "_" + HoldsState + // HoldsTool reads the holds, asks one again now, or moves one by a person's word. + HoldsTool = "indexer_holds" + // EngineReadsState is the indexer's state of what each counted node's engine states it reads (sdk/enginereads). + EngineReadsState = "engine-reads" + EngineReadsBucket = delivery.IndexerModule + "_" + EngineReadsState + // EngineReadsTool lists, asks again and drops those statements. + EngineReadsTool = "indexer_engine_reads" +) + +// MeshWideReads are the reader capabilities whose need holds a module's new entry from the whole mesh, not only from the +// nodes it is placed on: form:seat-reach, which catalogue.ManifestNeeds marks for the store holder's own manifest (a +// claim of container-image-registry at the mesh's scope) and for a manifest naming a seat's reach (ADR 0632 §1, §2). +var MeshWideReads = []string{catalogue.ReadForm + catalogue.ReferenceFormSeatReach} + +// HoldVersion is the version of a holds value this indexer writes and reads. +const HoldVersion = 1 + +// Hold is one module's held entry: what the new manifest needs of MeshWideReads, which counted nodes lack which of it +// (or why no node is counted), the manifest the modules state keeps and the one held, the merge that brought it, since +// when, and the entry written once it moves. +type Hold struct { + V int `json:"v"` + Module string `json:"module"` + Needs []string `json:"needs"` + Lacking map[string][]string `json:"lacking,omitempty"` + Unknown string `json:"unknown,omitempty"` + Kept string `json:"kept"` + Manifest string `json:"manifest"` + Repository string `json:"repository"` + Commit string `json:"commit"` + Since time.Time `json:"since"` + Next json.RawMessage `json:"next"` +} + +// EncodeHold is h at HoldVersion. +func EncodeHold(h Hold) ([]byte, error) { + if h.Module == "" || len(h.Next) == 0 { + return nil, errors.New("a hold names its module and the entry it holds") + } + h.V = HoldVersion + return json.Marshal(h) +} + +// DecodeHold reads a holds value; it refuses another version and one holding no entry. +func DecodeHold(raw []byte) (Hold, error) { + var h Hold + if err := json.Unmarshal(raw, &h); err != nil { + return h, err + } + if h.V != HoldVersion { + return h, fmt.Errorf("a holds value of version %d, not %d", h.V, HoldVersion) + } + if h.Module == "" || len(h.Next) == 0 { + return h, errors.New("a holds value naming no module or no entry") + } + return h, nil +} + +// meshWide is what of needs MeshWideReads names, sorted. +func meshWide(needs []string) []string { + var out []string + for _, n := range needs { + if slices.Contains(MeshWideReads, n) { + out = append(out, n) + } + } + sort.Strings(out) + return slices.Compact(out) +} + +// noneCounted is why a hold stays while no node is counted. +const noneCounted = "no node is counted yet (the placement is not read), so the hold stays until it is" + +// lacking is, of the counted nodes, each whose engine lacks some of needs, with what it lacks; or why no node is counted. +func (ix *Indexer) lacking(needs []string) (map[string][]string, string) { + counted, known := ix.Counted() + if !known || len(counted) == 0 { + return nil, noneCounted + } + out := map[string][]string{} + for node, stated := range counted { + if miss := catalogue.MissingManifestReads(needs, stated); len(miss) > 0 { + out[node] = miss + } + } + if len(out) == 0 { + return nil, "" + } + return out, "" +} + +// lackingWords says who lacks what, or why no node is counted. +func lackingWords(lacking map[string][]string, unknown string) string { + if unknown != "" { + return unknown + } + var parts []string + for _, node := range slices.Sorted(maps.Keys(lacking)) { + parts = append(parts, node+" lacks "+strings.Join(lacking[node], ", ")) + } + return strings.Join(parts, "; ") +} + +// holdBack decides, for one module of a merge, whether its new entry next waits (true: kept in the holds state, the +// modules state left at cur) or is written now (false: any hold of it dropped, the merge having moved it). It holds only +// a changed manifest of a module the modules state holds an entry of: a new module has no previous entry to keep, and is +// held on each node by the assigner (ADR 0614 §4). An error is a hold that could not be kept: the merge is left, and +// delivered again, rather than writing what a counted engine cannot read. +func (ix *Indexer) holdBack(ctx context.Context, key string, next, cur delivery.Module, has bool) (bool, error) { + if ix.Holds == nil || ix.Counted == nil { + return false, nil + } + needs := meshWide(next.Needs) + var lacking map[string][]string + var unknown string + if len(needs) > 0 && has && cur.Manifest != "" && next.Manifest != "" && cur.Manifest != next.Manifest { + lacking, unknown = ix.lacking(needs) + } + if len(lacking) == 0 && unknown == "" { + return false, ix.dropHold(ctx, key, "its merge "+next.Commit+" brought a manifest every counted node-engine reads") + } + raw, err := delivery.EncodeModule(next) + if err != nil { + return false, err + } + h := Hold{Module: key, Needs: needs, Lacking: lacking, Unknown: unknown, Kept: cur.Manifest, Manifest: next.Manifest, + Repository: next.Repository, Commit: next.Commit, Since: ix.now().UTC(), Next: raw} + err = write(ctx, ix.Holds, key, func(value []byte, found bool) ([]byte, bool, error) { + h := h + if found { + if old, err := DecodeHold(value); err == nil && old.Manifest == h.Manifest { + // The same manifest held again by a later merge: its first merge and time stand, as a module's commit does + // (writeOverlay), unless the entry itself moved. + if prev, err := delivery.DecodeModule(old.Next); err == nil && sameButWhere(prev, next) { + h.Next, h.Commit = old.Next, old.Commit + } + h.Since = old.Since + } + } + data, err := EncodeHold(h) + return data, false, err + }) + if err != nil { + return false, fmt.Errorf("%s: its new manifest %s is held from the mesh, and the hold cannot be kept: %w", key, next.Manifest, err) + } + ix.logf("modules.%s keeps manifest %s: the new manifest %s of %s needs %s, which not every counted node-engine states (%s; ADR 0632 §2); it moves once each does", + key, cur.Manifest, next.Manifest, next.Commit, strings.Join(needs, ", "), lackingWords(lacking, unknown)) + return true, nil +} + +// dropHold deletes the hold of key, if there is one, saying why. +func (ix *Indexer) dropHold(ctx context.Context, key, why string) error { + if ix.Holds == nil { + return nil + } + dropped := false + err := write(ctx, ix.Holds, key, func(_ []byte, found bool) ([]byte, bool, error) { + dropped = found + return nil, found, nil + }) + if errors.Is(err, process.ErrNotYet) { + return nil // a holds state not on the bus yet holds nothing to drop + } + if err != nil { + return fmt.Errorf("holds.%s: %w", key, err) + } + if dropped { + ix.logf("holds.%s dropped: %s", key, why) + } + return nil +} + +// KickHolds moves every hold whose needs every counted node-engine now states, in the background, once at a time: on a +// node's statement changed, the placement's change and the holds state's binding. +func (ix *Indexer) KickHolds(ctx context.Context) { + if ix.Holds == nil || ix.Counted == nil { + return + } + if ix.holding.Swap(true) { + ix.holdAgain.Store(true) + return + } + go func() { + for { + ix.holdAgain.Store(false) + if err := ix.MoveHolds(ctx); err != nil { + ix.logf("moving the holds: %v", err) + } + if !ix.holdAgain.Load() { + ix.holding.Store(false) + if !ix.holdAgain.Load() || ix.holding.Swap(true) { + return + } + } + } + }() +} + +// MoveHolds weighs every hold against what the counted node-engines state now: one every node reads is written, and +// one still held is kept with who lacks what now. +func (ix *Indexer) MoveHolds(ctx context.Context) error { + if ix.Holds == nil || ix.Counted == nil { + return nil + } + keys, err := ix.Holds.Keys(ctx) + if errors.Is(err, process.ErrNotYet) { + return nil // weighed once the state is bound (main.go) + } + if err != nil { + return fmt.Errorf("listing the holds: %w", err) + } + var errs []error + for _, key := range slices.Sorted(slices.Values(keys)) { + if _, _, err := ix.moveHold(ctx, key, ""); err != nil { + errs = append(errs, err) + } + } + return errors.Join(errs...) +} + +// moveHold weighs one hold; with by, a person's word moves it whatever the engines state. It answers whether it moved, +// and what stands. +func (ix *Indexer) moveHold(ctx context.Context, key, by string) (bool, string, error) { + moved := false + for try := 0; try < writeTries; try++ { + value, rev, found, err := ix.Holds.Get(ctx, key) + if err != nil { + return moved, "", fmt.Errorf("holds.%s: %w", key, err) + } + if !found { + if moved { + return true, "moved", nil + } + return false, "no hold of " + key + " is kept", nil + } + h, err := DecodeHold(value) + if err != nil { + ix.logf("holds.%s holds what is not a hold, left: %v", key, err) + return moved, "", fmt.Errorf("holds.%s: %w", key, err) + } + next, err := delivery.DecodeModule(h.Next) + if err != nil { + return moved, "", fmt.Errorf("holds.%s: its entry: %w", key, err) + } + lacking, unknown := ix.lacking(h.Needs) + if by == "" && (len(lacking) > 0 || unknown != "") { + if reflect.DeepEqual(lacking, h.Lacking) && unknown == h.Unknown { + return false, lackingWords(lacking, unknown), nil + } + h.Lacking, h.Unknown = lacking, unknown + data, err := EncodeHold(h) + if err != nil { + return false, "", err + } + if err := ix.Holds.Update(ctx, key, data, rev); errors.Is(err, ErrConflict) { + continue + } else if err != nil { + return false, "", fmt.Errorf("holds.%s: %w", key, err) + } + return false, lackingWords(lacking, unknown), nil + } + ix.recompute.Lock() + // Under the lock a merge writes its modules by, read again: a merge since replaced or dropped the hold. + if _, now, still, err := ix.Holds.Get(ctx, key); err != nil || !still || now != rev { + ix.recompute.Unlock() + if err != nil { + return moved, "", fmt.Errorf("holds.%s: %w", key, err) + } + continue + } + curRaw, _, has, err := ix.Modules.Get(ctx, key) + if err != nil { + ix.recompute.Unlock() + return moved, "", fmt.Errorf("modules.%s: %w", key, err) + } + cur, derr := delivery.DecodeModule(curRaw) + if !has || derr != nil || !delivery.SameRepository(cur.Repository, next.Repository) { + err := ix.Holds.Delete(ctx, key, rev) + ix.recompute.Unlock() + if errors.Is(err, ErrConflict) { + continue + } + ix.logf("holds.%s dropped: the modules state no longer holds the module of %s", key, next.Repository) + return moved, "dropped: the modules state no longer holds the module of " + next.Repository, err + } + next.Src = ix.srcOf(next.Repository, next.Commit) + w := &mergeWrite{repository: next.Repository, merge: next.Commit} + err = ix.recomputeLocked(ctx, map[string]delivery.Module{key: next}, w) + if err == nil { + err = ix.Holds.Delete(ctx, key, rev) + } + ix.recompute.Unlock() + if errors.Is(err, ErrConflict) { + moved = true + continue // a merge held a newer manifest meanwhile: weighed in turn + } + if err != nil { + return moved, "", fmt.Errorf("moving the hold of %s: %w", key, err) + } + moved = true + why := "every counted node-engine states " + strings.Join(h.Needs, ", ") + if by != "" { + why = "moved by " + by + "'s word, " + cmpOr(lackingWords(lacking, unknown), "every counted node-engine stating "+strings.Join(h.Needs, ", ")) + } + ix.logf("modules.%s moves from manifest %s to %s of %s, held since %s: %s", key, h.Kept, h.Manifest, h.Commit, h.Since.Format(time.RFC3339), why) + if err := ix.RecomputeCandidates(ctx); err != nil { + ix.logf("the candidates after the hold of %s moved: %v", key, err) + } + if ix.Verdicts != nil { + if _, err := ix.JudgeAll(ctx, w.fresh); err != nil { + ix.logf("judge: the manifests are not judged after the hold of %s moved: %v", key, err) + } + } + return true, why, nil + } + return moved, "", fmt.Errorf("holds.%s: it kept changing while it was weighed", key) +} + +func cmpOr(a, b string) string { + if a != "" { + return a + } + return b +} + +// HoldAgain answers a person's ask of one hold (the tool indexer_holds): again asks the engines of the counted nodes that +// lack what it needs once more and weighs it now; by moves it by that person's word, whatever the engines state. +func (ix *Indexer) HoldAgain(ctx context.Context, module string, release bool, by string) TakenIn { + r := TakenIn{Module: module} + key, err := delivery.ModulesKey(module) + if err != nil { + r.Answer = err.Error() + return r + } + if ix.Holds == nil || ix.Counted == nil { + r.Answer = "the indexer keeps no holds: its holds state or the engines' statements are not bound" + return r + } + value, _, found, err := ix.Holds.Get(ctx, key) + switch { + case err != nil: + r.Answer = "the holds state cannot be read: " + err.Error() + return r + case !found: + r.Answer = "no hold of " + module + " is kept: its entry in the modules state is the newest manifest merged" + return r + } + h, err := DecodeHold(value) + if err != nil { + r.Answer = "its hold cannot be read: " + err.Error() + return r + } + r.Repository, r.Commit = h.Repository, h.Commit + if !release { + lacking, _ := ix.lacking(h.Needs) + if ix.AskEngines != nil && len(lacking) > 0 { + ix.AskEngines(slices.Sorted(maps.Keys(lacking)), "the hold of "+module+" asked again") + } + } + who := "" + if release { + who = cmpOr(by, "a person") + } + moved, words, err := ix.moveHold(ctx, key, who) + switch { + case err != nil: + r.Answer = "the hold was not weighed: " + err.Error() + case moved: + r.Stored = true + r.Answer = "moved: modules." + key + " now names manifest " + h.Manifest + " (" + words + ")" + default: + r.Held = true + r.Answer = "held at manifest " + h.Kept + ": " + words + "; the engines asked again answer in the background, and the hold moves once each states " + strings.Join(h.Needs, ", ") + } + return r +} + +// ListHolds is every hold the state holds, by module, or one module's; a value that does not read is named with why. +func ListHolds(keys []string, get func(key string) ([]byte, error), module string) (map[string]any, error) { + if module != "" { + key, err := delivery.ModulesKey(module) + if err != nil { + return nil, err + } + keys = []string{key} + } + holds, unread := map[string]Hold{}, map[string]string{} + for _, key := range slices.Sorted(slices.Values(keys)) { + raw, err := get(key) + if err != nil { + return nil, err + } + if raw == nil { + continue + } + h, err := DecodeHold(raw) + if err != nil { + unread[key] = err.Error() + continue + } + h.Next = nil // the entry whole is the modules state's form; its digest and commit say which + holds[key] = h + } + return map[string]any{"holds": holds, "unread": unread}, nil +} + +// NewHoldAgain is the takeins value indexer_holds writes to ask one module's hold again (release false) or to move it by +// a person's word (release true, which takes why: it is a hand act). +func NewHoldAgain(module, why, by string, release bool, at time.Time) (string, []byte, error) { + key, data, err := NewTakeIn(module, why, by, at) + if err != nil { + return "", nil, err + } + var a TakeInAsk + if err := json.Unmarshal(data, &a); err != nil { + return "", nil, err + } + if release && a.Why == "" { + return "", nil, errors.New("say why the hold is moved by hand (why): it is a hand act") + } + a.Hold = HoldAskAgain + if release { + a.Hold = HoldRelease + } + data, err = json.Marshal(a) + return holdPrefix + key, data, err +} + +// The asks of a hold a takeins value names (TakeInAsk.Hold), and the prefix of its key. +const ( + HoldAskAgain = "again" + HoldRelease = "release" + holdPrefix = "hold." +) diff --git a/modules/mesh-delivery-indexer/internal/index/holds_test.go b/modules/mesh-delivery-indexer/internal/index/holds_test.go new file mode 100644 index 00000000..5d2ba8cc --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/holds_test.go @@ -0,0 +1,71 @@ +package index + +import ( + "context" + "slices" + "testing" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// ADR 0632 §2, ADR 0624 §1: the store holder's new manifest, which every node-engine reads to find the store, keeps the +// entry at the old manifest while one counted node states only the baseline reads (the event manifest-reads-changed not +// yet heard of it, and the hold asked again by a person), and moves once every counted node states form:seat-reach. +func TestTheStoreHoldersNewManifestWaitsForEveryCountedEngine(t *testing.T) { + ctx := context.Background() + const dir = "modules/distribution" + old := `{"module":"distribution","version":"1","listens":[{"name":"registry","port":5000,"from":"mesh"}]}` + next := `{"module":"distribution","version":"1","claims":[{"name":"container-image-registry","scope":"mesh"}],` + + `"listens":[{"name":"registry","port":5000,"from":"mesh"}]}` + if needs, _ := catalogue.ManifestNeeds([]byte(next)); !slices.Contains(needs, "form:seat-reach") { + t.Fatalf("the new manifest needs %v, want form:seat-reach (mesh-sdk ecdbf1a)", needs) + } + ix, mods, _ := manifestsIndexer(t) + holds := newMem() + stated := map[string][]string{"n1": catalogue.ManifestReads(), "n2": nil} // n2: an engine before the statement, the baseline + var asked []string + ix.Holds, ix.Counted = holds, func() (map[string][]string, bool) { return stated, true } + ix.AskEngines = func(nodes []string, _ string) { asked = append(asked, nodes...) } + + first := withManifests(mergedMsg(catalogueRepo, map[string]string{dir: "tree1"}), map[string]string{dir: old}, nil) + if err := ix.Apply(ctx, first); err != nil { + t.Fatal(err) + } + kept, _ := moduleOf(t, mods, "distribution") + if kept.Manifest != delivery.ManifestDigest([]byte(old)) || len(holds.vals) != 0 { + t.Fatalf("after the first merge modules.distribution names %s with %d holds, want the old manifest and none", kept.Manifest, len(holds.vals)) + } + second := withManifests(mergedMsg(catalogueRepo, map[string]string{dir: "tree2"}), map[string]string{dir: next}, nil) + second.Merge = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb" + if err := ix.Apply(ctx, second); err != nil { + t.Fatal(err) + } + stays := func(when string) { + t.Helper() + m, _ := moduleOf(t, mods, "distribution") + if m.Manifest != kept.Manifest || m.FP2 != kept.FP2 || m.Commit != kept.Commit || m.Inputs2 == nil || m.Inputs2.Tree != "tree1" { + t.Fatalf("%s modules.distribution = manifest %s fp2 %q commit %s, want the old entry kept whole (%s, %q, %s)", + when, m.Manifest, m.FP2, m.Commit, kept.Manifest, kept.FP2, kept.Commit) + } + h, err := DecodeHold(holds.vals["distribution"]) + if err != nil || h.Manifest != delivery.ManifestDigest([]byte(next)) || h.Kept != kept.Manifest || !slices.Equal(h.Lacking["n2"], []string{"form:seat-reach"}) || len(h.Lacking) != 1 { + t.Fatalf("%s holds.distribution = %+v (%v), want the new manifest held, n2 alone lacking form:seat-reach", when, h, err) + } + } + stays("with n2 stating only the baseline,") + if r := ix.HoldAgain(ctx, "distribution", false, "op@shanks"); !r.Held || r.Stored || !slices.Equal(asked, []string{"n2"}) { + t.Fatalf("asked again: %+v, engines asked %v; want it held and n2's engine asked again", r, asked) + } + stays("asked again with n2 still stating only the baseline,") + + stated["n2"] = catalogue.ManifestReads() // n2's engine states the reading (manifest-reads-changed) + if err := ix.MoveHolds(ctx); err != nil { + t.Fatal(err) + } + m, _ := moduleOf(t, mods, "distribution") + if m.Manifest != delivery.ManifestDigest([]byte(next)) || m.Commit != second.Merge || m.Inputs2 == nil || m.Inputs2.Tree != "tree2" || len(holds.vals) != 0 { + t.Fatalf("once every counted engine states form:seat-reach modules.distribution = manifest %s commit %s, %d holds; want the new entry and none", + m.Manifest, m.Commit, len(holds.vals)) + } +} diff --git a/modules/mesh-delivery-indexer/internal/index/modules.go b/modules/mesh-delivery-indexer/internal/index/modules.go index 3c8307e2..3f7c0120 100644 --- a/modules/mesh-delivery-indexer/internal/index/modules.go +++ b/modules/mesh-delivery-indexer/internal/index/modules.go @@ -118,6 +118,15 @@ func (ix *Indexer) applyMoving(ctx context.Context, f delivery.Forge, drop bool, next.None2 = in.None2 + ": the forge could not say them: " + namedErr[dir] } } + // A manifest every node reads, needing what some counted node-engine does not state, waits in the holds state; the + // module keeps its entry until each states it (holds.go, ADR 0632 §2). + waits, err := ix.holdBack(ctx, key, next, cur, has) + if err != nil { + return w, err + } + if waits { + continue + } overlay[key] = next } // A module of this repository the merge commit holds no module.json for is gone from the trunk, and so is its key diff --git a/modules/mesh-delivery-indexer/internal/index/takein.go b/modules/mesh-delivery-indexer/internal/index/takein.go index 5cb8e06c..8c26ed02 100644 --- a/modules/mesh-delivery-indexer/internal/index/takein.go +++ b/modules/mesh-delivery-indexer/internal/index/takein.go @@ -38,6 +38,9 @@ type TakeInAsk struct { Say bool `json:"say,omitempty"` // SeatDefinitions asks a contracts module's seat definitions to be read again now (indexer_seat_definitions_again, // DLV-I19): its trunk's, or with Pull and Head a pull request's head. + // Hold asks the module's mesh-wide hold (holds.go): HoldAskAgain weighs it again now, HoldRelease moves it by a + // person's word (indexer_holds). + Hold string `json:"hold,omitempty"` SeatDefinitions bool `json:"seat-definitions,omitempty"` Pull int `json:"pull,omitempty"` Head string `json:"head,omitempty"` diff --git a/modules/mesh-delivery-indexer/internal/index/toolchains.go b/modules/mesh-delivery-indexer/internal/index/toolchains.go index b19fe0c2..83d1f488 100644 --- a/modules/mesh-delivery-indexer/internal/index/toolchains.go +++ b/modules/mesh-delivery-indexer/internal/index/toolchains.go @@ -152,6 +152,14 @@ func (ix *Indexer) recomputeModules(ctx context.Context, overlay map[string]deli } ix.recompute.Lock() defer ix.recompute.Unlock() + return ix.recomputeLocked(ctx, overlay, w) +} + +// recomputeLocked is recomputeModules with ix.recompute held by the caller (a hold that moves, holds.go). +func (ix *Indexer) recomputeLocked(ctx context.Context, overlay map[string]delivery.Module, w *mergeWrite) error { + if ix.Manifests == nil { + return ix.writeOverlay(ctx, overlay, nil, w) + } keys, err := ix.Modules.Keys(ctx) if err != nil { return fmt.Errorf("listing the modules: %w", err) diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index e4ff7b19..108cd0bf 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -114,7 +114,8 @@ 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}, {Name: views.MarksState, History: 1}, {Name: health.FindingsState, History: 5}, {Name: views.StatusState, History: 1}, - {Name: index.RequirementsState, History: 5}, {Name: index.SeatDefinitionsState, History: 5}}; !reflect.DeepEqual(m.State, want) { // DLV-I7, DLV-I17 + {Name: index.RequirementsState, History: 5}, {Name: index.SeatDefinitionsState, History: 5}, + {Name: index.HoldsState, History: 5}, {Name: index.EngineReadsState, History: 1}}; !reflect.DeepEqual(m.State, want) { // DLV-I7, DLV-I17, ADR 0632 §2 t.Errorf("state = %+v, want %+v", m.State, want) } // DLV-F1r (ADR 0615 §1): the assigner's placement is read through its first view and views family, never its state. @@ -125,7 +126,8 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build // internal/broker/state.go:62-93). // The registry seats' published and the builder's numbered fill each package's series (DLV-I5, design 51 *The versions*). if want := []string{"go-package-registry.published", "mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported", "mesh-delivery-assigner.views.placement", - "mesh-delivery-builder.version.numbered", "mesh-delivery-checker.preview.written", "mesh-delivery-checker.verdict", "mesh-delivery-forge-bridge.forge.**", "npm-package-registry.published"}; !reflect.DeepEqual(sorted(m.Consumes), want) { + "mesh-delivery-builder.version.numbered", "mesh-delivery-checker.preview.written", "mesh-delivery-checker.verdict", "mesh-delivery-forge-bridge.forge.**", + "node-engine.manifest-reads-changed", "npm-package-registry.published"}; !reflect.DeepEqual(sorted(m.Consumes), want) { // ADR 0632 §2: the counted engines' reads t.Errorf("consumes = %v, want %v", m.Consumes, want) } if b, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 || b != "${dir:state}/broker" { @@ -137,7 +139,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build // Three invokes: the file of a registered module's manifest and lock files (registered.go), the forge's trees (the named paths' objects, since the indexer is the one reader of a manifest's paths: // ADR 0499 clauses 2 and 3, ADR 0495 clause 1; and the module directories of the trunk, issue 1900) and whether a // commit is in another (issue 1900). - if want := []string{"mesh-delivery-assigner.placement", "gitea.gitea_commit_trees", "gitea.gitea_contains", "gitea.gitea_get_file", "gitea.gitea_open_pulls", "seat:node-engine.composition", "seat:mesh-hand-acts.hand-act"}; !reflect.DeepEqual(m.Invokes, want) { + if want := []string{"mesh-delivery-assigner.placement", "gitea.gitea_commit_trees", "gitea.gitea_contains", "gitea.gitea_get_file", "gitea.gitea_open_pulls", "seat:node-engine.composition", "seat:node-engine.manifest-reads", "seat:mesh-hand-acts.hand-act"}; !reflect.DeepEqual(m.Invokes, want) { t.Errorf("invokes = %v, want %v", m.Invokes, want) } // 1945: indexer_take_in, served by its tools bundle through the takeins state; the process serves none. The same bundle @@ -153,8 +155,9 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build } } want = append(want, "indexer_first_views", "indexer_first_view_again", index.ConfigReadersTool, index.BusStateTool, index.RequirementsTool, index.PackageUsersTool, index.SayAcceptedTool, // DLV-F3r, DLV-I10, DLV-I11, DLV-I7, DLV-I8, DLV-I9c - index.SeatDefinitionTool, index.SeatDefinitionsAgainTool, index.OutOfStepTool) // DLV-I19 - if !reflect.DeepEqual(m.Tools, want) || len(want) != 29 { + index.SeatDefinitionTool, index.SeatDefinitionsAgainTool, index.OutOfStepTool, // DLV-I19 + index.HoldsTool, index.EngineReadsTool) // ADR 0632 §2: the mesh-wide holds and the engines' reads they weigh + if !reflect.DeepEqual(m.Tools, want) || len(want) != 31 { t.Errorf("tools = %v, want %v", m.Tools, want) } // Every read tool of the indexer a reader of this repository invokes is one it serves (the checker's seventh first diff --git a/modules/mesh-delivery-indexer/module.json b/modules/mesh-delivery-indexer/module.json index dc97a71f..739ae752 100644 --- a/modules/mesh-delivery-indexer/module.json +++ b/modules/mesh-delivery-indexer/module.json @@ -44,7 +44,9 @@ "indexer_say_accepted", "indexer_seat_definition", "indexer_seat_definitions_again", - "indexer_out_of_step" + "indexer_out_of_step", + "indexer_holds", + "indexer_engine_reads" ], "consumes": [ "go-package-registry.published", @@ -55,6 +57,7 @@ "mesh-delivery-checker.preview.written", "mesh-delivery-checker.verdict", "mesh-delivery-forge-bridge.forge.**", + "node-engine.manifest-reads-changed", "npm-package-registry.published" ], "state": [ @@ -74,7 +77,9 @@ { "name": "findings", "history": 5 }, { "name": "first-views", "history": 1 }, { "name": "requirements", "history": 5 }, - { "name": "seat-definitions", "history": 5 } + { "name": "seat-definitions", "history": 5 }, + { "name": "holds", "history": 5 }, + { "name": "engine-reads", "history": 1 } ], "streams": [ { "name": "indexer-views", "events": ["views.**"], "max-age-days": 7, "max-mib": 256 }, @@ -98,6 +103,7 @@ "gitea.gitea_get_file", "gitea.gitea_open_pulls", "seat:node-engine.composition", + "seat:node-engine.manifest-reads", "seat:mesh-hand-acts.hand-act" ], "own-secrets": {