diff --git a/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go b/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go index 3d99c591..1f6a2e2d 100644 --- a/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go +++ b/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go @@ -15,6 +15,7 @@ import ( "strings" "sync" + "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/delivery" ) @@ -206,7 +207,7 @@ func ConvertKept(subject string, data []byte, trees Trees, files Files, kept Kep if action == delivery.PullUpdated && p.ModuleDirsSaid { f.Dirs, f.DirsKnown = p.ModuleDirs, true var err error - if f.Manifests, f.Locks, _, err = inputsRead(p.Owner, p.Repo, p.HeadSHA, p.ModuleDirs, files); err != nil { + if f.Manifests, f.Locks, f.Needs, _, err = inputsRead(p.Owner, p.Repo, p.HeadSHA, p.ModuleDirs, files); err != nil { return Message{}, fmt.Errorf("%w: %s: the manifests of %s/%s at the head %s: %v", ErrRetry, subject, p.Owner, p.Repo, p.HeadSHA, err) } } @@ -245,15 +246,18 @@ type read struct { // order the calls finish in, and the error is that of the first failing file in the order of dirs and LockFiles. After // a failure no further file is asked for, and inputsOf returns only when every call it started has returned. func inputsOf(owner, repo, ref string, dirs []string, files Files) (manifests map[string]string, locks map[string]map[string]string, err error) { - manifests, locks, _, err = inputsRead(owner, repo, ref, dirs, files) + manifests, locks, _, _, err = inputsRead(owner, repo, ref, dirs, files) return manifests, locks, err } // inputsRead is inputsOf, and also answers, on a failure too, the directories whose manifest and lock files were all // read, with their manifests and lock hashes: a merge keeps them by their tree, so its next try reads only the rest // (issue 1126). It reads no mesh package a go.sum names: the indexer alone computes what a fingerprint takes (novox/hq -// ADR 0593 §2), so the message's Packages stay empty. -func inputsRead(owner, repo, ref string, dirs []string, files Files) (manifests map[string]string, locks map[string]map[string]string, whole []string, err error) { +// ADR 0593 §2), so the message's Packages stay empty. Of a directory whose manifest names templates it reads each +// template too, in a second round, and answers the module's needs whole (catalogue.ModuleNeeds; novox/hq ADR 0632 §2): +// a template's text can need a reading its manifest does not. A template the forge does not have is no need (the judge +// refuses the manifest by name); any other failure to read one is the directory's, as a lock file's is. +func inputsRead(owner, repo, ref string, dirs []string, files Files) (manifests map[string]string, locks map[string]map[string]string, needs map[string][]string, whole []string, err error) { at := func(dir, name string) string { if dir == "" { return name @@ -267,6 +271,103 @@ func inputsRead(owner, repo, ref string, dirs []string, files Files) (manifests reads = append(reads, &read{dir: dir, path: at(dir, name)}) } } + fetch(owner, repo, ref, reads, files) + manifests = map[string]string{} + locks = map[string]map[string]string{} + broken := map[string]bool{} // a directory with a file not read (failed, or never asked after a failure) + failed := func(r *read) bool { + switch { + case r.err != nil && (r.manifest || !errors.Is(r.err, ErrNotFound)): + broken[r.dir] = true + if err == nil { + err = fmt.Errorf("%s: %w", r.path, r.err) + } + case r.err == nil && r.raw == nil && !r.asked: + broken[r.dir] = true + default: + return false + } + return true + } + for _, r := range reads { + switch { + case failed(r): + case r.err != nil: // a lock file the forge does not have + case r.manifest: + manifests[r.dir] = string(r.raw) + default: + sum := sha256.Sum256(r.raw) + if locks[r.dir] == nil { + locks[r.dir] = map[string]string{} + } + locks[r.dir][path.Base(r.path)] = hex.EncodeToString(sum[:]) + } + } + var templates []*read + if err == nil { + for _, dir := range dirs { + for _, t := range templatePaths(manifests[dir]) { + templates = append(templates, &read{dir: dir, path: at(dir, t)}) + } + } + fetch(owner, repo, ref, templates, files) + } + texts := map[string]map[string]string{} + for _, r := range templates { + if failed(r) || r.err != nil { + continue + } + if texts[r.dir] == nil { + texts[r.dir] = map[string]string{} + } + texts[r.dir][strings.TrimPrefix(r.path, at(r.dir, ""))] = string(r.raw) + } + needs = map[string][]string{} + for _, dir := range dirs { + if raw, ok := manifests[dir]; ok && !broken[dir] && len(templatePaths(raw)) > 0 { + needs[dir], _ = catalogue.ModuleNeeds([]byte(raw), texts[dir]) + } + } + for _, dir := range dirs { + if !broken[dir] { + whole = append(whole, dir) + } + } + if err != nil { + for dir := range broken { + delete(manifests, dir) + delete(locks, dir) + delete(needs, dir) + } + return manifests, locks, needs, whole, err + } + if len(manifests) == 0 { + manifests = nil + } + if len(locks) == 0 { + locks = nil + } + if len(needs) == 0 { + needs = nil + } + return manifests, locks, needs, whole, nil +} + +// templatePaths is the templates a manifest names (catalogue.TemplatePaths), none for one that does not decode. +func templatePaths(raw string) []string { + if raw == "" { + return nil + } + m, err := catalogue.Decode([]byte(raw)) + if err != nil { + return nil + } + return catalogue.TemplatePaths(m) +} + +// fetch asks the forge for each of reads at ref, at most MaxReadsInFlight at once; after a failure (a manifest, or a +// file the forge has and could not give) no further file is asked for, and it returns once every call it started has. +func fetch(owner, repo, ref string, reads []*read, files Files) { var ( wg sync.WaitGroup mu sync.Mutex @@ -295,48 +396,6 @@ func inputsRead(owner, repo, ref string, dirs []string, files Files) (manifests }(r) } wg.Wait() - manifests = map[string]string{} - locks = map[string]map[string]string{} - broken := map[string]bool{} // a directory with a file not read (failed, or never asked after a failure) - for _, r := range reads { - switch { - case r.err != nil && (r.manifest || !errors.Is(r.err, ErrNotFound)): - broken[r.dir] = true - if err == nil { - err = fmt.Errorf("%s: %w", r.path, r.err) - } - case r.err == nil && r.raw == nil && !r.asked: - broken[r.dir] = true - case r.err != nil: // a lock file the forge does not have - case r.manifest: - manifests[r.dir] = string(r.raw) - default: - sum := sha256.Sum256(r.raw) - if locks[r.dir] == nil { - locks[r.dir] = map[string]string{} - } - locks[r.dir][path.Base(r.path)] = hex.EncodeToString(sum[:]) - } - } - for _, dir := range dirs { - if !broken[dir] { - whole = append(whole, dir) - } - } - if err != nil { - for dir := range broken { - delete(manifests, dir) - delete(locks, dir) - } - return manifests, locks, whole, err - } - if len(manifests) == 0 { - manifests = nil - } - if len(locks) == 0 { - locks = nil - } - return manifests, locks, whole, nil } // merged is the merged message of a merge into the trunk: the trees of the merge commit, asked once. @@ -354,27 +413,31 @@ func merged(subject string, p pull, trees Trees, files Files, kept Kept) (Messag names = append(names, dir) } slices.Sort(names) - manifests, locks, toRead := keptInputs(names, dirs, kept) - read, readLocks, whole, err := inputsRead(p.Owner, p.Repo, p.MergeSHA, toRead, files) + manifests, locks, needs, toRead := keptInputs(names, dirs, kept) + read, readLocks, readNeeds, whole, err := inputsRead(p.Owner, p.Repo, p.MergeSHA, toRead, files) if err != nil { - keep(whole, dirs, read, readLocks, kept) // what was read whole is not asked again at the next try + keep(whole, dirs, read, readLocks, readNeeds, kept) // what was read whole is not asked again at the next try return Message{}, fmt.Errorf("%w: %s: the manifests of %s at the merge commit %s (%d of %d module directories read from the forge, the others kept by their tree): %v", ErrRetry, subject, repository, p.MergeSHA, len(toRead), len(names), err) } - keep(toRead, dirs, read, readLocks, kept) + keep(toRead, dirs, read, readLocks, readNeeds, kept) maps.Copy(manifests, read) maps.Copy(locks, readLocks) + maps.Copy(needs, readNeeds) if len(manifests) == 0 { manifests = nil } if len(locks) == 0 { locks = nil } + if len(needs) == 0 { + needs = nil + } event, err := delivery.ForgeEvent(repository, delivery.ForgeMerged) if err != nil { return Message{}, unreadable(subject, "%v", err) } f := delivery.Forge{Kind: delivery.ForgeMerged, Repository: repository, Branch: p.Head, Number: p.Number, - Head: p.HeadSHA, Merge: p.MergeSHA, Root: root, Trees: dirs, Manifests: manifests, Locks: locks} + Head: p.HeadSHA, Merge: p.MergeSHA, Root: root, Trees: dirs, Manifests: manifests, Locks: locks, Needs: needs} if repository == CatalogRepository { // The catalogue's predefined list at the merge (novox/hq ADR 0565 §2): the indexer keeps it for mesh-config. raw, err := files(p.Owner, p.Repo, PredefinedFile, p.MergeSHA) diff --git a/modules/mesh-delivery-forge-bridge/internal/bridge/files_test.go b/modules/mesh-delivery-forge-bridge/internal/bridge/files_test.go index ce1f5458..59dc8282 100644 --- a/modules/mesh-delivery-forge-bridge/internal/bridge/files_test.go +++ b/modules/mesh-delivery-forge-bridge/internal/bridge/files_test.go @@ -7,6 +7,7 @@ import ( "fmt" "os" "reflect" + "slices" "sort" "strings" "sync" @@ -122,6 +123,20 @@ func TestMergedCarriesManifestsAndLocks(t *testing.T) { // 1448/AC1 if _, err := Convert(mergedSubject, mergedBody(map[string]any{"repo": "mesh-catalog"}), treesOf(&calls, "r0", map[string]string{"modules/a": "t1"}, nil), forge.get); !errors.Is(err, ErrRetry) { t.Errorf("a forge that cannot say is %v, want a retry", err) } + // ADR 0632 §2: of a manifest naming a template, the template is read at the merge and the module's needs carried + // whole: docker's daemon file names the image seat's reach while its manifest stays the same. A manifest naming + // none carries no needs; a template the forge does not have is no need, and no retry. + forge.errs = nil + forge.have["modules/d/module.json"] = []byte(`{"module":"d","version":"1","resources":[{"id":"daemon","type":"file",` + + `"path":"/etc/d.json","template":"files/daemon.json"},{"id":"gone","type":"file","path":"/etc/g","template":"files/gone"}]}`) + forge.have["modules/d/files/daemon.json"] = []byte(`{"registry":"${seat:container-image-registry:reach}"}`) + msg, err = Convert(mergedSubject, mergedBody(nil), treesOf(&calls, "r0", map[string]string{"modules/b": "t2", "modules/d": "t4"}, nil), forge.get) + if err != nil { + t.Fatal(err) + } + if f, err = delivery.DecodeForge(msg.Data); err != nil || len(f.Needs) != 1 || !slices.Contains(f.Needs["modules/d"], "form:seat-reach") { + t.Errorf("needs = %v (%v), want modules/d's alone, with form:seat-reach from its template", f.Needs, err) + } // Issue 1126: a merge reads the forge only for the module directories whose tree changed. t.Run("a merge reads only the directories whose tree changed", mergeReadsOnlyChangedDirs) } diff --git a/modules/mesh-delivery-forge-bridge/internal/bridge/kept.go b/modules/mesh-delivery-forge-bridge/internal/bridge/kept.go index 6c687396..2a384c8b 100644 --- a/modules/mesh-delivery-forge-bridge/internal/bridge/kept.go +++ b/modules/mesh-delivery-forge-bridge/internal/bridge/kept.go @@ -13,6 +13,9 @@ package bridge type Inputs struct { Manifest string `json:"manifest"` Locks map[string]string `json:"locks,omitempty"` + // Needs are the module's needs whole when its manifest names templates (inputsRead): a value of such a manifest kept + // without them, by a bridge before ADR 0632 §2, is read again. + Needs []string `json:"needs,omitempty"` } // Kept holds Inputs by a directory's git tree hash (the bridge's state bucket inputs). Get answers false for a tree it @@ -23,8 +26,8 @@ type Kept interface { } // keptInputs splits dirs into the inputs kept for their tree hash and the directories still to read from the forge. -func keptInputs(dirs []string, trees map[string]string, kept Kept) (manifests map[string]string, locks map[string]map[string]string, read []string) { - manifests, locks = map[string]string{}, map[string]map[string]string{} +func keptInputs(dirs []string, trees map[string]string, kept Kept) (manifests map[string]string, locks map[string]map[string]string, needs map[string][]string, read []string) { + manifests, locks, needs = map[string]string{}, map[string]map[string]string{}, map[string][]string{} for _, dir := range dirs { tree := trees[dir] if kept == nil || tree == "" { @@ -32,7 +35,7 @@ func keptInputs(dirs []string, trees map[string]string, kept Kept) (manifests ma continue } in, ok := kept.Get(tree) - if !ok { + if !ok || (in.Needs == nil && len(templatePaths(in.Manifest)) > 0) { read = append(read, dir) continue } @@ -40,19 +43,22 @@ func keptInputs(dirs []string, trees map[string]string, kept Kept) (manifests ma if len(in.Locks) > 0 { locks[dir] = in.Locks } + if in.Needs != nil { + needs[dir] = in.Needs + } } - return manifests, locks, read + return manifests, locks, needs, read } // keep puts what was read of each directory under its tree hash. -func keep(dirs []string, trees map[string]string, manifests map[string]string, locks map[string]map[string]string, kept Kept) { +func keep(dirs []string, trees map[string]string, manifests map[string]string, locks map[string]map[string]string, needs map[string][]string, kept Kept) { if kept == nil { return } for _, dir := range dirs { raw, ok := manifests[dir] if tree := trees[dir]; ok && tree != "" { - kept.Put(tree, Inputs{Manifest: raw, Locks: locks[dir]}) + kept.Put(tree, Inputs{Manifest: raw, Locks: locks[dir], Needs: needs[dir]}) } } } 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 index c4f594ef..64cbe557 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/holds.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/holds.go @@ -10,6 +10,8 @@ import ( "git.novox.be/novox/mesh-delivery/sdk/enginereads" + "git.novox.be/novox/mesh-sdk/go/delivery" + "mesh-delivery-indexer/internal/index" ) @@ -25,8 +27,9 @@ func holdsTool(read stdio.KeptState, ask asker, now func() time.Time, acts handa "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.", + "act, recorded through mesh-hand-acts, which names the nodes it strands: each node whose engine lacks the reading, an " + + "away node from its return, reads the new entry without it and an engine before form:seat-reach finds no store). A hold " + + "standing past its bound is the finding " + index.HoldStandingKind + ". 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"}, @@ -41,7 +44,15 @@ func holdsTool(read stdio.KeptState, ask asker, now func() time.Time, acts handa return nil, err } if release { - if _, err := acts.Record(handacts.Request{What: "moved the mesh-wide hold of " + module + " by hand", + what := "moved the mesh-wide hold of " + module + " by hand" + if mk, err := delivery.ModulesKey(module); err == nil { + if e, err := read.Get(mk); err == nil && e != nil { + if h, err := index.DecodeHold(e.Value); err == nil { + what += ", " + index.Strands(h) + } + } + } + if _, err := acts.Record(handacts.Request{What: what, 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) } 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 987baddd..c58e719f 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -53,6 +53,37 @@ func main() { // process.ErrNotYet. type awaitedStore struct{ b *process.AwaitedBucket } +// expiringKeys is index.Bounds on a state whose keys take a time to live (key-ttl in the manifest): a key armed expires +// at its end, which its watcher hears as a purge; one standing keeps its end. +type expiringKeys struct{ b *process.AwaitedBucket } + +func (e expiringKeys) Arm(ctx context.Context, key string, ends time.Time) error { + kv := e.b.KV() + if kv == nil { + return process.ErrNotYet + } + left := time.Until(ends) + if left < time.Second { + left = time.Second // the bus's least time to live + } + value, _ := json.Marshal(map[string]any{"bounds": "the hold of " + key, "ends": ends.UTC()}) + if _, err := kv.Create(ctx, key, value, jetstream.KeyTTL(left)); err != nil && !errors.Is(err, jetstream.ErrKeyExists) { + return err + } + return nil +} + +func (e expiringKeys) Disarm(ctx context.Context, key string) error { + kv := e.b.KV() + if kv == nil { + return nil + } + if err := kv.Delete(ctx, key); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) { + return err + } + return nil +} + func (s awaitedStore) store() (kvStore, error) { kv := s.b.KV() if kv == nil { @@ -399,6 +430,24 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro 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) }) + // Each standing hold's bound, a key that expires at its end (ADR 0632 §2, the indexer's own state): its expiry weighs + // the holds again, which raises manifest-held; until the state is bound a hold past its bound is raised at its next + // weighing only, said. + holdBounds := process.AwaitBucket(ctx, js, conn, index.HoldBoundsBucket, "a hold standing past its bound is raised at its next weighing only", logger.Printf) + ix.HoldBounds = expiringKeys{holdBounds} + holdBounds.Ready(func(kv jetstream.KeyValue) { + each := func(e jetstream.KeyValueEntry) { + if e.Operation() == jetstream.KeyValuePurge { // its time to live ended + ix.HoldBoundEnded(ctx, e.Key()) + } + } + if w, err := settle(ctx, kv, each); err != nil { + process.NotWatched(logger.Printf, index.HoldBoundsBucket, err, "a hold standing past its bound is raised at its next weighing only until the indexer starts again") + } else { + go func() { errs <- follow(ctx, w, each) }() + } + ix.KickHolds(ctx) // a bound that ended while the indexer was down is raised now + }) 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) }) diff --git a/modules/mesh-delivery-indexer/internal/index/candidates.go b/modules/mesh-delivery-indexer/internal/index/candidates.go index 2e474abb..21a71951 100644 --- a/modules/mesh-delivery-indexer/internal/index/candidates.go +++ b/modules/mesh-delivery-indexer/internal/index/candidates.go @@ -125,6 +125,9 @@ type Indexer struct { 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 + // HoldBounds is the indexer's state hold-bounds (holds.go): each standing hold's bound as a key that expires at its + // end, whose expiry weighs the holds again (HoldBoundEnded). Nil raises the finding at the next weighing only. + HoldBounds Bounds 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 index c53365fc..a5931e05 100644 --- a/modules/mesh-delivery-indexer/internal/index/holds.go +++ b/modules/mesh-delivery-indexer/internal/index/holds.go @@ -48,6 +48,25 @@ const ( // 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} +// A hold standing past HoldBound is the indexer's finding HoldStandingKind about its module, which the condition keeper +// raises (ADR 0564 §2): the module's new entry is delivered nowhere while it stands, and a counted node that is away for +// long (ADR 0624 §1) keeps it so. The finding names who lacks what; it is cleared when the hold moves or is dropped. +// The bound is a key of the indexer's state hold-bounds that expires at Since+HoldBound (a key's time to live on the +// bus, as the builder's bounds): its expiry weighs the holds again, which raises the finding, and a restarted indexer +// weighs every hold at its start, so a bound that ended while it was down is raised then. No timer runs. +const ( + HoldStandingKind = "manifest-held" + HoldBound = 30 * time.Minute + HoldBoundsState = "hold-bounds" + HoldBoundsBucket = delivery.IndexerModule + "_" + HoldBoundsState +) + +// Bounds is the state hold-bounds: Arm writes key to expire at ends, a standing key keeping its end; Disarm deletes it. +type Bounds interface { + Arm(ctx context.Context, key string, ends time.Time) error + Disarm(ctx context.Context, key string) error +} + // HoldVersion is the version of a holds value this indexer writes and reads. const HoldVersion = 1 @@ -149,7 +168,9 @@ func (ix *Indexer) holdBack(ctx context.Context, key string, next, cur delivery. 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 { + // Any change of the entry is held, not only a new manifest: a template's text that newly needs a reading changes + // the module's tree and needs while its manifest stays (ADR 0632 §2, docker's daemon file). + if len(needs) > 0 && has && cur.Manifest != "" && next.Manifest != "" && !sameButWhere(cur, next) { lacking, unknown = ix.lacking(needs) } if len(lacking) == 0 && unknown == "" { @@ -162,15 +183,15 @@ func (ix *Indexer) holdBack(ctx context.Context, key string, next, cur delivery. 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 + h.Since, h.Next, h.Commit = ix.now().UTC(), raw, next.Commit 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) { + if old, err := DecodeHold(value); err == nil { + // The module is held since its first hold, whatever a later merge brings; the same entry held again keeps + // its first merge too, as a module's commit does (writeOverlay). + h.Since = old.Since + if old.Manifest == h.Manifest && sameHeld(old.Next, next) { h.Next, h.Commit = old.Next, old.Commit } - h.Since = old.Since } } data, err := EncodeHold(h) @@ -181,6 +202,7 @@ func (ix *Indexer) holdBack(ctx context.Context, key string, next, cur delivery. } 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)) + ix.holdStands(ctx, key, h) return true, nil } @@ -202,6 +224,7 @@ func (ix *Indexer) dropHold(ctx context.Context, key, why string) error { } if dropped { ix.logf("holds.%s dropped: %s", key, why) + ix.holdEnds(ctx, key) } return nil } @@ -281,6 +304,7 @@ func (ix *Indexer) moveHold(ctx context.Context, key, by string) (bool, string, lacking, unknown := ix.lacking(h.Needs) if by == "" && (len(lacking) > 0 || unknown != "") { if reflect.DeepEqual(lacking, h.Lacking) && unknown == h.Unknown { + ix.holdStands(ctx, key, h) return false, lackingWords(lacking, unknown), nil } h.Lacking, h.Unknown = lacking, unknown @@ -293,6 +317,7 @@ func (ix *Indexer) moveHold(ctx context.Context, key, by string) (bool, string, } else if err != nil { return false, "", fmt.Errorf("holds.%s: %w", key, err) } + ix.holdStands(ctx, key, h) return false, lackingWords(lacking, unknown), nil } ix.recompute.Lock() @@ -316,6 +341,9 @@ func (ix *Indexer) moveHold(ctx context.Context, key, by string) (bool, string, if errors.Is(err, ErrConflict) { continue } + if err == nil { + ix.holdEnds(ctx, key) + } 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 } @@ -334,9 +362,10 @@ func (ix *Indexer) moveHold(ctx context.Context, key, by string) (bool, string, return moved, "", fmt.Errorf("moving the hold of %s: %w", key, err) } moved = true + ix.holdEnds(ctx, key) 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, ", ")) + why = "moved by " + by + "'s word, " + stranded(lacking, unknown, 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 { @@ -352,6 +381,67 @@ func (ix *Indexer) moveHold(ctx context.Context, key, by string) (bool, string, return moved, "", fmt.Errorf("holds.%s: it kept changing while it was weighed", key) } +// sameHeld says a held entry is next but for where it came from: its first merge stands. +func sameHeld(held json.RawMessage, next delivery.Module) bool { + prev, err := delivery.DecodeModule(held) + return err == nil && sameButWhere(prev, next) +} + +// holdStands keeps the finding of h standing: raised, naming who lacks what now, once h stood HoldBound; until then its +// bound is armed in the state hold-bounds, whose expiry weighs the holds again (HoldBoundEnded). +func (ix *Indexer) holdStands(ctx context.Context, key string, h Hold) { + if ends := h.Since.Add(HoldBound); ix.now().Before(ends) { + if ix.HoldBounds != nil { + if err := ix.HoldBounds.Arm(ctx, key, ends); err != nil { + ix.logf("hold-bounds.%s is not armed, so the hold standing past %s is raised at its next weighing only: %v", key, HoldBound, err) + } + } + return + } + ix.finding(ctx, findingID("hold", key), HoldStandingKind, true, + fmt.Sprintf("modules.%s keeps manifest %s since %s: its new manifest %s of %s needs %s, and %s", + key, h.Kept, h.Since.Format(time.RFC3339), h.Manifest, cmpOr(h.Commit, "its merge"), strings.Join(h.Needs, ", "), lackingWords(h.Lacking, h.Unknown)), + "A module's new manifest is held from the whole mesh", + "Every node-engine reads "+key+"'s manifest, and one counted node-engine does not yet state the reading the new one "+ + "needs, so the indexer keeps the old entry and the new manifest is delivered nowhere (ADR 0632 §2). It moves by "+ + "itself once each counted node's engine states it; an away node keeps it held until it is back. indexer_holds "+ + "names who lacks what, asks again, or moves it by a person's word; indexer_engine_reads reads the statements.", + "The hold of "+key+" moved or was dropped.") +} + +// HoldBoundEnded takes a key of the state hold-bounds that expired: the holds are weighed again, and the one it bounds, +// still standing, is raised. +func (ix *Indexer) HoldBoundEnded(ctx context.Context, key string) { + ix.logf("hold-bounds.%s ended: the holds are weighed again", key) + ix.KickHolds(ctx) +} + +// holdEnds disarms the bound of key's hold and clears its finding. +func (ix *Indexer) holdEnds(ctx context.Context, key string) { + if ix.HoldBounds != nil { + if err := ix.HoldBounds.Disarm(ctx, key); err != nil { + ix.logf("hold-bounds.%s is not deleted; its end weighs the holds once more, which moves nothing: %v", key, err) + } + } + ix.finding(ctx, findingID("hold", key), HoldStandingKind, false, "", "", "", "") +} + +// Strands is what moving h by a person's word strands now, as its hand act and its answer say it. +func Strands(h Hold) string { return stranded(h.Lacking, h.Unknown, h.Needs) } + +// stranded says, of a hold moved by a person's word, the nodes it strands: each counted node whose engine lacks what +// the new entry needs reads it without it (an engine before form:seat-reach finds no store) until it states it. +func stranded(lacking map[string][]string, unknown string, needs []string) string { + switch { + case unknown != "": + return "while " + unknown + ": every node whose engine lacks " + strings.Join(needs, ", ") + " is stranded until it states it" + case len(lacking) == 0: + return "every counted node-engine stating " + strings.Join(needs, ", ") + ": it strands none" + } + return "and it strands " + strings.Join(slices.Sorted(maps.Keys(lacking)), ", ") + " (" + lackingWords(lacking, "") + + "): each reads the new entry without that reading until its engine states it, an away node from its return" +} + func cmpOr(a, b string) string { if a != "" { return a @@ -437,7 +527,8 @@ func ListHolds(keys []string, get func(key string) ([]byte, error), module strin 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 + return map[string]any{"holds": holds, "unread": unread, "bound": "each hold standing " + HoldBound.String() + + " past its since is the finding " + HoldStandingKind + ", its end a key of the state " + HoldBoundsState}, nil } // NewHoldAgain is the takeins value indexer_holds writes to ask one module's hold again (release false) or to move it by diff --git a/modules/mesh-delivery-indexer/internal/index/holds_test.go b/modules/mesh-delivery-indexer/internal/index/holds_test.go index 5d2ba8cc..dde34e39 100644 --- a/modules/mesh-delivery-indexer/internal/index/holds_test.go +++ b/modules/mesh-delivery-indexer/internal/index/holds_test.go @@ -3,40 +3,66 @@ package index import ( "context" "slices" + "strings" "testing" + "time" "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/health" ) -// 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) { +// fakeBounds is the state hold-bounds: the end each key was armed with, a standing key keeping its end. +type fakeBounds map[string]time.Time + +func (b fakeBounds) Arm(_ context.Context, key string, ends time.Time) error { + if _, standing := b[key]; !standing { + b[key] = ends + } + return nil +} + +func (b fakeBounds) Disarm(_ context.Context, key string) error { delete(b, key); return nil } + +// ADR 0632 §2, ADR 0624 §1, a case of TestAMergedManifestIsKeptOnceByItsDigestAndNamedByTheModule. 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. docker, whose manifest stays the same while its +// template newly names the image seat's reach (the needs the forge bridge carries), is held the same way. A hold's bound +// is a key of hold-bounds; past it the finding manifest-held names n2, cleared when the hold moves; a move by a person's +// word names the node it strands. +func storeHoldersNewManifestWaitsForEveryCountedEngine(t *testing.T) { ctx := context.Background() - const dir = "modules/distribution" + const dir, dockerDir = "modules/distribution", "modules/docker" 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) + docker := `{"module":"docker","version":"1","resources":[{"id":"daemon","type":"file","path":"/etc/docker/daemon.json","template":"files/daemon.json"}]}` + dockerNeeds, _ := catalogue.ModuleNeeds([]byte(docker), map[string]string{"files/daemon.json": `"${seat:container-image-registry:reach}"`}) + if needs, _ := catalogue.ManifestNeeds([]byte(next)); !slices.Contains(needs, "form:seat-reach") || !slices.Contains(dockerNeeds, "form:seat-reach") { + t.Fatalf("the new manifest needs %v and docker's %v, want form:seat-reach in both (mesh-sdk ecdbf1a, ModuleNeeds)", needs, dockerNeeds) } ix, mods, _ := manifestsIndexer(t) - holds := newMem() + holds, findings, bounds := newMem(), newMem(), fakeBounds{} 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 } + clock := time.Date(2026, 11, 4, 21, 0, 0, 0, time.UTC) + ix.Holds, ix.Findings, ix.HoldBounds, ix.Now = holds, findings, bounds, func() time.Time { return clock } + ix.Counted = func() (map[string][]string, bool) { return stated, true } ix.AskEngines = func(nodes []string, _ string) { asked = append(asked, nodes...) } + heldKey := health.FindingKey(health.ScopeModule, findingID("hold", "distribution"), HoldStandingKind) - first := withManifests(mergedMsg(catalogueRepo, map[string]string{dir: "tree1"}), map[string]string{dir: old}, nil) + first := withManifests(mergedMsg(catalogueRepo, map[string]string{dir: "tree1", dockerDir: "d1"}), map[string]string{dir: old, dockerDir: docker}, nil) if err := ix.Apply(ctx, first); err != nil { t.Fatal(err) } kept, _ := moduleOf(t, mods, "distribution") + keptDocker, _ := moduleOf(t, mods, "docker") 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 := withManifests(mergedMsg(catalogueRepo, map[string]string{dir: "tree2", dockerDir: "d2"}), map[string]string{dir: next, dockerDir: docker}, nil) + second.Needs = map[string][]string{dockerDir: dockerNeeds} second.Merge = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb" if err := ix.Apply(ctx, second); err != nil { t.Fatal(err) @@ -54,18 +80,36 @@ func TestTheStoreHoldersNewManifestWaitsForEveryCountedEngine(t *testing.T) { } } stays("with n2 stating only the baseline,") + if d, _ := moduleOf(t, mods, "docker"); d.Inputs2 == nil || d.Inputs2.Tree != keptDocker.Inputs2.Tree || holds.vals["docker"] == nil { + t.Fatalf("docker's template naming the seat's reach: modules.docker at tree %v, holds %v; want its old entry kept and a hold", d.Inputs2, holds.vals["docker"]) + } 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,") + if len(findings.vals) != 0 || !bounds["distribution"].Equal(clock.Add(HoldBound)) { + t.Fatalf("within its bound: findings %v, bounds %v; want none and hold-bounds.distribution ending at %v", findings.vals, bounds, clock.Add(HoldBound)) + } + clock = clock.Add(HoldBound) + if err := ix.MoveHolds(ctx); err != nil { // what the key's expiry asks (HoldBoundEnded, in the background) + t.Fatal(err) + } + stays("past its bound,") + f, err := health.DecodeFinding(findings.vals[heldKey]) + if err != nil || f.Kind != HoldStandingKind || !strings.Contains(f.Summary, "n2 lacks form:seat-reach") { + t.Fatalf("past its bound the finding %s = %+v (%v), want manifest-held naming n2", heldKey, f, err) + } + if r := ix.HoldAgain(ctx, "docker", true, "op@shanks"); !r.Stored || !strings.Contains(r.Answer, "strands n2") { + t.Fatalf("docker moved by hand: %+v, want it moved and its answer naming n2 stranded", r) + } 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)) + if m.Manifest != delivery.ManifestDigest([]byte(next)) || m.Commit != second.Merge || m.Inputs2 == nil || m.Inputs2.Tree != "tree2" || len(holds.vals) != 0 || findings.vals[heldKey] != nil || len(bounds) != 0 { + t.Fatalf("once every counted engine states form:seat-reach modules.distribution = manifest %s commit %s, %d holds, finding %s, bounds %v; want the new entry and none", + m.Manifest, m.Commit, len(holds.vals), findings.vals[heldKey], bounds) } } diff --git a/modules/mesh-delivery-indexer/internal/index/manifests_test.go b/modules/mesh-delivery-indexer/internal/index/manifests_test.go index b9817ec9..34fde8ee 100644 --- a/modules/mesh-delivery-indexer/internal/index/manifests_test.go +++ b/modules/mesh-delivery-indexer/internal/index/manifests_test.go @@ -46,6 +46,7 @@ func TestAMergedManifestIsKeptOnceByItsDigestAndNamedByTheModule(t *testing.T) { t.Run("kept by its digest and named by the module", mergedManifestIsKeptByItsDigestAndNamedByTheModule) t.Run("the same manifest applied again is written once", sameManifestAppliedAgainIsWrittenOnce) t.Run("the seat mesh-catalog's verbs answer from the states", catalogVerbsAnswerFromTheStates) + t.Run("a manifest every node reads waits for every counted engine", storeHoldersNewManifestWaitsForEveryCountedEngine) // ADR 0632 §2 } func mergedManifestIsKeptByItsDigestAndNamedByTheModule(t *testing.T) { // 1449/AC1 diff --git a/modules/mesh-delivery-indexer/internal/index/modules.go b/modules/mesh-delivery-indexer/internal/index/modules.go index 3f7c0120..957c0c2e 100644 --- a/modules/mesh-delivery-indexer/internal/index/modules.go +++ b/modules/mesh-delivery-indexer/internal/index/modules.go @@ -99,6 +99,9 @@ func (ix *Indexer) applyMoving(ctx context.Context, f delivery.Forge, drop bool, // The reader capabilities a changed manifest needs are written with it, before the judge's pass names them with its // verdict, so no reader takes a new manifest as needing nothing (ADR 0614 §4). next.Needs, _ = catalogue.ManifestNeeds([]byte(raw)) + if whole, ok := f.Needs[dir]; ok { + next.Needs = whole // its templates' text too, read by the forge bridge at the merge (ADR 0632 §2) + } if next.Manifest == "" { next.None2 = "the manifest cannot be read: it cannot be kept as a value" } else { diff --git a/modules/mesh-delivery-indexer/internal/index/verdicts.go b/modules/mesh-delivery-indexer/internal/index/verdicts.go index 970a3e76..137a2d24 100644 --- a/modules/mesh-delivery-indexer/internal/index/verdicts.go +++ b/modules/mesh-delivery-indexer/internal/index/verdicts.go @@ -470,6 +470,9 @@ func (ix *Indexer) carryVerdict(ctx context.Context, name, digest string, v deli case v.Refused(): state = delivery.VerdictRefused } + // The judge reads the manifest alone; the forms its templates' text writes came with its merge (ADR 0632 §2) and + // stand, cur being of the same manifest. + needs := withTemplateForms(needs, cur.Needs) if cur.Verdict == state && slices.Equal(cur.Problems, problems) && slices.Equal(cur.Unknown, v.Unknown) && slices.Equal(cur.Warnings, v.Warnings) && slices.Equal(cur.Needs, needs) { return nil, false, nil @@ -647,3 +650,16 @@ func NewJudgeAgain(module, digest, why, by string, at time.Time) (string, []byte data, err = json.Marshal(a) return judgeAgainPrefix + key, data, err } + +// withTemplateForms is needs with the reference forms of had added: a module's templates write forms only (catalogue +// ModuleNeeds), and a merge carried them with the manifest the judge reads again alone. +func withTemplateForms(needs, had []string) []string { + out := slices.Clone(needs) + for _, n := range had { + if strings.HasPrefix(n, catalogue.ReadForm) && !slices.Contains(out, n) { + out = append(out, n) + } + } + slices.Sort(out) + return out +} diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index 108cd0bf..86b8b248 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -30,6 +30,7 @@ import ( type state struct { Name string `json:"name"` History int `json:"history,omitempty"` + KeyTTL bool `json:"key-ttl,omitempty"` } type artifact struct { @@ -115,7 +116,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build 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}, - {Name: index.HoldsState, History: 5}, {Name: index.EngineReadsState, History: 1}}; !reflect.DeepEqual(m.State, want) { // DLV-I7, DLV-I17, ADR 0632 §2 + {Name: index.HoldsState, History: 5}, {Name: index.EngineReadsState, History: 1}, {Name: index.HoldBoundsState, History: 1, KeyTTL: true}}; !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. diff --git a/modules/mesh-delivery-indexer/module.json b/modules/mesh-delivery-indexer/module.json index 739ae752..db2e40f4 100644 --- a/modules/mesh-delivery-indexer/module.json +++ b/modules/mesh-delivery-indexer/module.json @@ -79,7 +79,8 @@ { "name": "requirements", "history": 5 }, { "name": "seat-definitions", "history": 5 }, { "name": "holds", "history": 5 }, - { "name": "engine-reads", "history": 1 } + { "name": "engine-reads", "history": 1 }, + { "name": "hold-bounds", "history": 1, "key-ttl": true } ], "streams": [ { "name": "indexer-views", "events": ["views.**"], "max-age-days": 7, "max-mib": 256 },