diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main.go index d570a958..8fb3566a 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main.go @@ -105,7 +105,7 @@ func TryTools(tr Trying) []stdio.Tool { {Name: check.TryTool, Description: "Try a delivery group, or one pushed head, before its pull requests are opened (try a group, group try, try several repositories): " + "name the members, each a repository and the exact commit asked (never the branch's head now, so a push after the ask changes nothing of what is tried), " + - "or one repository and its head for a try of one member. The checker judges it by the code a group's pull-request check runs, so the two cannot drift: " + + "or one repository and its head for a try of one member (the seat verb try of mesh-delivery-checker takes members from the registry milestone on; until it is delivered, ask a group through this tool). The checker judges it by the code a group's pull-request check runs, so the two cannot drift: " + "each member merged onto its trunk's head, the unit and the integration level of every member through the build modules, the manifest validator over " + "the members' manifests, the module graph, and the composition on every placed node. Every failure is named (each member, level, test, manifest and node), not " + "the first. A try owes the forge nothing (no status, no Approve asked) and writes no verdict of a pull request; what its passing levels prove is kept, so a pull request's " + 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 37bf9725..754f3b94 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 @@ -217,6 +217,7 @@ func main() { ask := askTakeIn(stdio.State(index.TakeInsState), 25*time.Second) tools := append(Tools(ask, time.Now, stdio.State(index.VerdictsState), heldCounts(stdio.State(index.OldSectionsState))), catalogTools()...) tools = append(tools, index.PendingTools(keptRead{stdio.State(index.PendingState)}, ask, time.Now)...) + tools = append(tools, index.TryCandidateTools(keptRead{stdio.State("candidates")}, ask, time.Now)...) if err := stdio.Serve("", tools); err != nil { fmt.Fprintf(os.Stderr, "[mesh-delivery-indexer-tools] %v\n", err) os.Exit(1) diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/previews.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/previews.go index 21f0b645..8c15b996 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/previews.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/previews.go @@ -39,7 +39,7 @@ type previewWatch struct { } func newPreviewWatch(ctx context.Context, js bucketOpener, ix *index.Indexer, logger *log.Logger) *previewWatch { - return &previewWatch{ctx: ctx, js: js, ix: ix, logger: logger, errs: make(chan error, 1)} + return &previewWatch{ctx: ctx, js: js, ix: ix, logger: logger, errs: make(chan error, 2)} } // ensure watches the previews bucket if it is not watched and exists, settling what it holds, then writes the candidates' @@ -73,7 +73,34 @@ func (w *previewWatch) ensure() error { } }) }() - return w.ix.RecomputeCandidates(w.ctx) + w.watchTries() + if err := w.ix.RecomputeCandidates(w.ctx); err != nil { + return err + } + return w.ix.WriteTryCandidates(w.ctx) +} + +// watchTries follows the checker's tries (index.TriesBucket) beside its previews: a whole try that waits or runs gets its +// candidate, once its previews are there, and loses it when it ends (index.Indexer.ApplyTry). A bucket not there is watched at +// the checker's next preview.written, as the previews are. +func (w *previewWatch) watchTries() { + kv, err := w.js.KeyValue(w.ctx, index.TriesBucket) + if err != nil { + w.logger.Printf("the bucket %s cannot be read (%v): a group try's candidate is written when the checker says it wrote a preview", index.TriesBucket, err) + return + } + tw, err := settle(w.ctx, kv, func(e jetstream.KeyValueEntry) { w.ix.PutTry(e.Key(), e.Value(), !opOf(e)) }) + if err != nil { + process.NotWatched(w.logger.Printf, index.TriesBucket, err, "it is watched when the checker says it wrote a preview") + return + } + go func() { + w.errs <- follow(w.ctx, tw, func(e jetstream.KeyValueEntry) { + if err := w.ix.ApplyTry(w.ctx, e.Key(), e.Value(), !opOf(e)); err != nil { + w.logger.Printf("%s.%s could not be taken into the candidates: %v", index.TriesBucket, e.Key(), err) + } + }) + }() } // Written takes one preview.written event of the checker in: its bucket is watched if it is not. A subject that is not the 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 cfa942d2..7a40aabb 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go @@ -19,6 +19,7 @@ type takerIn interface { TakeInPendingByHand(ctx context.Context, key, by, why string) index.TakenIn DropPending(ctx context.Context, key, by, why string) index.TakenIn TakeInMerge(ctx context.Context, repository, merge string, seq uint64, by, why string) index.TakenIn + DropTryCandidate(ctx context.Context, id, by, why string) index.TakenIn } // takenIn answers one value of the takeins state (the tool indexer_take_in, issue 1945): a value not yet answered takes the @@ -37,7 +38,10 @@ func takenIn(ctx context.Context, ix takerIn, kv jetstream.KeyValue, e jetstream return } asked, r := "take-in", index.TakenIn{} - if a.Merge != "" { + if a.TryCandidate != "" && a.Drop { + asked, r = "drop of a try's candidate", ix.DropTryCandidate(ctx, a.TryCandidate, a.By, a.Why) + a.Module = a.TryCandidate + } else if a.Merge != "" { asked, r = "hand take-in of a merge", ix.TakeInMerge(ctx, a.MergeRepository, a.Merge, a.Sequence, a.By, a.Why) a.Module = r.Module } else if a.Pending != "" && a.Drop { diff --git a/modules/mesh-delivery-indexer/internal/index/candidates.go b/modules/mesh-delivery-indexer/internal/index/candidates.go index 72acca96..c04f53bd 100644 --- a/modules/mesh-delivery-indexer/internal/index/candidates.go +++ b/modules/mesh-delivery-indexer/internal/index/candidates.go @@ -137,7 +137,8 @@ type Indexer struct { Pending Store // SayTakeIn publishes a take-in event on the indexer's own stream take-in (ADR 0633 §5); nil says none. SayTakeIn func(ctx context.Context, t takein.TakeIn) error - pendingMu sync.Mutex // one take-in step at a time: a merge, a group's end, the start and a hand act (pending.go) + tries map[string][]TryMember // the whole tries that wait or run, by try id (trycandidates.go); guarded by mu + pendingMu sync.Mutex // one take-in step at a time: a merge, a group's end, the start and a hand act (pending.go) } func (ix *Indexer) logf(format string, args ...any) { @@ -283,6 +284,9 @@ func (ix *Indexer) applyOpen(ctx context.Context, f forgemsg.Forge) error { return fmt.Errorf("listing the candidates: %w", err) } for _, key := range keys { + if IsTryGroup(key) { + continue // a try's candidate is no pull request's: the checker's try owns it (trycandidates.go) + } err := write(ctx, ix.Store, key, func(value []byte, found bool) ([]byte, bool, error) { if !found { return nil, false, nil diff --git a/modules/mesh-delivery-indexer/internal/index/pending_tools.go b/modules/mesh-delivery-indexer/internal/index/pending_tools.go index db6cd499..b49a883f 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending_tools.go +++ b/modules/mesh-delivery-indexer/internal/index/pending_tools.go @@ -2,9 +2,13 @@ package index import ( "context" + "encoding/json" "fmt" + "strings" "time" + "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein" stdio "git.novox.be/novox/mesh-sdk/go" ) @@ -92,3 +96,61 @@ func PendingTools(pending Store, ask func(key string, value []byte) (TakeInAsk, Run: merge}, } } + +// TryCandidateTools are the tools of a group try's candidate (issue 1921, trycandidates.go): indexer_try_candidates reads the +// candidates the indexer holds for tries (the try, its heads, each member's modules and whether each has its fingerprint), +// and indexer_drop_try_candidate drops one by an ask in takeins. candidates is the candidates state as the tools bundle reads it. +func TryCandidateTools(candidates Store, ask func(key string, value []byte) (TakeInAsk, error), now func() time.Time) []stdio.Tool { + return []stdio.Tool{ + {Name: TryCandidatesTool, + Description: "The candidates the indexer holds for group tries (a try of several repositories at exact commits, mesh-delivery-checker try): " + + "per try its heads hash, each member's repository and commit, the modules the try touches and, per module, its manifest digest, its " + + "fingerprint or why it has none yet (a merge preview not yet written). The checker's manifest validator and the node-engines' " + + "compose-candidate read this candidate. Read-only; an empty answer means no whole try waits or runs.", + Input: map[string]any{"type": "object", "properties": map[string]any{}}, + Run: func(map[string]any) (any, error) { + ctx := context.Background() + keys, err := candidates.Keys(ctx) + if err != nil { + return nil, err + } + var out []delivery.Candidate + for _, k := range keys { + if !IsTryGroup(k) { + continue + } + v, _, found, err := candidates.Get(ctx, k) + if err != nil || !found { + continue + } + if c, err := delivery.DecodeCandidate(v); err == nil { + out = append(out, c) + } + } + return map[string]any{"tries": out}, nil + }}, + {Name: DropTryCandidateTool, + Description: "Drop a group try's candidate by hand (the indexer deletes it itself when the try ends): for a candidate left for a try that " + + "is gone, or to make the indexer write it again from the previews at the try's next change. A hand act: who and why are said in the journal.", + Input: map[string]any{"type": "object", "required": []string{"try", "why"}, "properties": map[string]any{ + "try": map[string]any{"type": "string", "description": "the try's id, try-"}, + "why": map[string]any{"type": "string", "description": "why it is dropped, in plain words"}}}, + Run: func(args map[string]any) (any, error) { + id, _ := args["try"].(string) + why, _ := args["why"].(string) + id = strings.TrimSpace(id) + if !IsTryGroup(id) || strings.TrimSpace(why) == "" { + return nil, fmt.Errorf("name the try (try-) and say why") + } + data, err := json.Marshal(TakeInAsk{V: 1, TryCandidate: id, Drop: true, Why: strings.TrimSpace(why), By: stdio.CallerOf(args), At: now().UTC()}) + if err != nil { + return nil, err + } + a, err := ask("try-candidate.drop."+id, data) + if err != nil { + return nil, err + } + return a.Result, nil + }}, + } +} diff --git a/modules/mesh-delivery-indexer/internal/index/previews.go b/modules/mesh-delivery-indexer/internal/index/previews.go index bb367fb6..18bf869d 100644 --- a/modules/mesh-delivery-indexer/internal/index/previews.go +++ b/modules/mesh-delivery-indexer/internal/index/previews.go @@ -72,7 +72,10 @@ func (ix *Indexer) ApplyPreview(ctx context.Context, key string, value []byte, d if !ix.PutPreview(key, value, deleted) { return nil } - return ix.RecomputeCandidates(ctx) + if err := ix.RecomputeCandidates(ctx); err != nil { + return err + } + return ix.WriteTryCandidates(ctx) // a try that waited for this preview gets its candidate } // RecomputeCandidates writes again every candidate whose pull requests' fingerprints, from the previews held, are no longer diff --git a/modules/mesh-delivery-indexer/internal/index/previews_test.go b/modules/mesh-delivery-indexer/internal/index/previews_test.go index 282ad860..e7ced5db 100644 --- a/modules/mesh-delivery-indexer/internal/index/previews_test.go +++ b/modules/mesh-delivery-indexer/internal/index/previews_test.go @@ -6,6 +6,7 @@ import ( "os" "strings" "testing" + "time" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg" "git.novox.be/novox/mesh-sdk/go/delivery" @@ -182,6 +183,67 @@ func TestAPreviewHeardBeforeItsPullRequestGivesTheFingerprintWhenTheCandidateIsW if got := fpOf(t, cands); got.FP2 != "" || got.None2 != NoneNoPreview { t.Errorf("a new head: %+v, want no preview", got) } + // Issue 1921: a whole group try has no pull request, yet the indexer writes its candidate from the checker's try and the + // previews its fetch wrote (number 0, the heads hash of its members), keeps it apart from the pull requests' candidates, + // and deletes it when the try ends. A try narrowed to a level or to tests has none. + t.Run("a group try's candidate", func(t *testing.T) { + ix, mods, cands := previewBed(t, true) + ctx := context.Background() + changed := catalogueManifest(t, "avahi") // not the manifest the trunk holds, so the try touches avahi + pv := avahiPreview(t, trunkNow, h1) + pv.Modules["modules/avahi"] = delivery.PreviewModule{Tree: "tree-avahi", Manifest: delivery.ManifestDigest([]byte(changed)), Locks: map[string]string{"go.sum": avahiGoSum}} + ix.Read = func(_ context.Context, _, _, dir string) (RepoRead, error) { + return RepoRead{Manifests: map[string]string{dir: changed}}, nil + } + tryValue := func(state, extra string) []byte { + return []byte(`{"id":"try-7","state":"` + state + `",` + extra + `"members":[{"repository":"` + catalogueRepo + `","head":"` + h1 + `","trunk":"` + trunkNow + `"}]}`) + } + held := func() []string { k, _ := cands.Keys(ctx); return k } + if err := ix.ApplyTry(ctx, "try-7", tryValue("running", `"level":"unit",`), false); err != nil || len(held()) != 0 { + t.Fatalf("a narrowed try: %v, candidates %v", err, held()) + } + if err := ix.ApplyTry(ctx, "try-7", tryValue("running", ""), false); err != nil || len(held()) != 0 { + t.Fatalf("a try before its preview: %v, candidates %v", err, held()) + } + applyPreview(t, ix, pv) + key, _ := delivery.CandidatesKey("try-7") + raw, _, found, _ := cands.Get(ctx, key) + if !found { + t.Fatalf("no candidate for the try after its preview: %v", held()) + } + c, err := delivery.DecodeCandidate(raw) + _, _ = moduleOf(t, mods, "avahi") + if err != nil || c.Group != "try-7" || len(c.Pulls) != 1 || c.Pulls[0].Number != 0 || c.Pulls[0].Head.Head != h1 || + c.Pulls[0].Fps["avahi"].FP2 == "" || c.Pulls[0].Fps["avahi"].Manifest == "" { + t.Fatalf("candidate = %+v, %v, want one pull of number 0 with avahi's fingerprint", c, err) + } + if err := ix.Apply(ctx, forgemsg.Forge{Kind: forgemsg.ForgeOpen}); err != nil || len(held()) != 1 { + t.Errorf("the open message left %v (%v), want the try's candidate alone", held(), err) + } + // The tools: one reads the candidates held for tries, one drops by an ask the indexer answers. + asks := func(key string, value []byte) (TakeInAsk, error) { + a, err := DecodeTakeIn(value) + r := ix.DropTryCandidate(ctx, a.TryCandidate, a.By, a.Why) + a.Result = &r + return a, err + } + tools := TryCandidateTools(cands, asks, time.Now) + if out, err := tools[0].Run(map[string]any{}); err != nil || len(out.(map[string]any)["tries"].([]delivery.Candidate)) != 1 { + t.Errorf("%s = %v, %v, want the try's candidate", tools[0].Name, out, err) + } + if _, err := tools[1].Run(map[string]any{"try": "feat/x", "why": "w"}); err == nil { + t.Error("a branch was dropped as a try") + } + if out, err := tools[1].Run(map[string]any{"try": "try-7", "why": "by hand"}); err != nil || !out.(*TakenIn).Stored || len(held()) != 0 { + t.Errorf("drop = %v, %v, candidates %v", out, err, held()) + } + if err := ix.ApplyTry(ctx, "try-7", tryValue("running", ""), false); err != nil || len(held()) != 1 { + t.Fatalf("the try's next change: %v, candidates %v, want it written again", err, held()) + } + if err := ix.ApplyTry(ctx, "try-7", tryValue("pass", ""), false); err != nil || len(held()) != 0 { + t.Errorf("an ended try: %v, candidates %v, want none", err, held()) + } + }) } func TestAPreviewNamingNoManifestOfTheModulesDirectoryGivesAReason(t *testing.T) { diff --git a/modules/mesh-delivery-indexer/internal/index/takein.go b/modules/mesh-delivery-indexer/internal/index/takein.go index e6cb17c6..55ec09a2 100644 --- a/modules/mesh-delivery-indexer/internal/index/takein.go +++ b/modules/mesh-delivery-indexer/internal/index/takein.go @@ -40,14 +40,16 @@ type TakeInAsk struct { Pending string `json:"pending,omitempty"` // Merge asks the merge of MergeRepository at this commit to be taken in by hand (indexer_take_in_merge): its merged message // is asked of the forge bridge and run through the take-in as a delivered one, under its stream sequence Sequence. - Merge string `json:"merge,omitempty"` - MergeRepository string `json:"merge-repository,omitempty"` - Sequence uint64 `json:"sequence,omitempty"` - Drop bool `json:"drop,omitempty"` - By string `json:"by,omitempty"` - At time.Time `json:"at"` - Result *TakenIn `json:"result,omitempty"` - Answered time.Time `json:"answered,omitzero"` + Merge string `json:"merge,omitempty"` + MergeRepository string `json:"merge-repository,omitempty"` + Sequence uint64 `json:"sequence,omitempty"` + // TryCandidate asks the candidate of this try's group to be dropped (indexer_drop_try_candidate), trycandidates.go. + TryCandidate string `json:"try-candidate,omitempty"` + Drop bool `json:"drop,omitempty"` + By string `json:"by,omitempty"` + At time.Time `json:"at"` + Result *TakenIn `json:"result,omitempty"` + Answered time.Time `json:"answered,omitzero"` } // Done says the ask is answered. @@ -62,8 +64,8 @@ func DecodeTakeIn(data []byte) (TakeInAsk, error) { if a.V != 1 { return a, fmt.Errorf("a takeins value of version %d, not 1", a.V) } - if a.Module == "" && a.Repository == "" && a.Pending == "" && a.Merge == "" { - return a, errors.New("a takeins value naming no module, repository, pending wait or merge") + if a.Module == "" && a.Repository == "" && a.Pending == "" && a.Merge == "" && a.TryCandidate == "" { + return a, errors.New("a takeins value naming no module, repository, pending wait, merge or try") } return a, nil } diff --git a/modules/mesh-delivery-indexer/internal/index/trycandidates.go b/modules/mesh-delivery-indexer/internal/index/trycandidates.go new file mode 100644 index 00000000..6f44914e --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/trycandidates.go @@ -0,0 +1,302 @@ +package index + +// **A group try's candidate** (issue 1921; novox/hq ADR 0593 §7): a try of several repositories at exact commits has no pull +// request, so the forge bridge sends the indexer no message for it; yet the node-engine's compose-candidate and the checker's +// manifest validator read the indexer's candidate of a group. The indexer therefore watches the checker's state tries (the +// checker keeps it; this module reads it) and, for a try that is not narrowed to a level or to some tests, writes +// `candidates.`: one pull per member (number 0), its modules and manifests taken from the merge preview the try's +// fetch wrote, its manifests kept and judged with the seat definitions the members carry, as a take-in judges a revision's. +// The fingerprints come with the previews (RecomputeCandidates). The candidate is deleted when the try ends. `try-` +// is the reserved group name of a try: the bridge's open message never takes it out, and the checker never checks it. + +import ( + "context" + "encoding/json" + "fmt" + "path" + "regexp" + "slices" + "strings" + + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg" + "git.novox.be/novox/mesh-sdk/go/delivery" + + "mesh-delivery-indexer/judge" +) + +// TriesBucket is the checker's state of tries, which the indexer reads. +const TriesBucket = delivery.CheckerModule + "_tries" + +// TryCandidateTools: the tools that read and drop a try's candidate. +const ( + TryCandidatesTool = "indexer_try_candidates" + DropTryCandidateTool = "indexer_drop_try_candidate" +) + +var tryGroupOK = regexp.MustCompile(`^try-[0-9]{1,20}$`) + +// IsTryGroup says a candidates key is the group of a try. +func IsTryGroup(key string) bool { return tryGroupOK.MatchString(key) } + +// TryMember is one repository of a try: its exact commit and the trunk commit it is merged onto. +type TryMember struct{ Repository, Head, Trunk string } + +// tryValue is what the indexer reads of a value of the checker's tries. +type tryValue struct { + ID string `json:"id"` + State string `json:"state"` + Level string `json:"level"` + Tests string `json:"tests"` + Members []struct { + Repository string `json:"repository"` + Head string `json:"head"` + Trunk string `json:"trunk"` + } `json:"members"` + Repository, Head, Trunk string +} + +// liveTry reads a tries value: its members, and whether it is a whole try that waits or runs. +func liveTry(value []byte) (id string, members []TryMember, live bool) { + var t tryValue + if json.Unmarshal(value, &t) != nil || !IsTryGroup(t.ID) { + return "", nil, false + } + for _, m := range t.Members { + members = append(members, TryMember{m.Repository, m.Head, m.Trunk}) + } + if len(members) == 0 && t.Repository != "" { + members = []TryMember{{t.Repository, t.Head, t.Trunk}} + } + live = (t.State == "queued" || t.State == "running") && t.Level == "" && t.Tests == "" && len(members) > 0 + return t.ID, members, live +} + +// PutTry takes a change of the tries state in without writing a candidate; it answers the try's id, and whether it is a +// whole try that waits or runs now, and whether it was one before. +func (ix *Indexer) PutTry(key string, value []byte, deleted bool) (id string, live, was bool) { + ix.mu.Lock() + defer ix.mu.Unlock() + _, was = ix.tries[key] + if deleted { + delete(ix.tries, key) + return key, false, was + } + _, members, live := liveTry(value) + if !live { + delete(ix.tries, key) + return key, false, was + } + if ix.tries == nil { + ix.tries = map[string][]TryMember{} + } + ix.tries[key] = members + return key, true, was +} + +// ApplyTry takes one change of the tries in: a whole try that waits or runs gets its candidate (once its previews are +// there), and a try that ended or went loses it. +func (ix *Indexer) ApplyTry(ctx context.Context, key string, value []byte, deleted bool) error { + id, live, was := ix.PutTry(key, value, deleted) + switch { + case live: + return ix.writeTryCandidate(ctx, id) + case was || deleted || IsTryGroup(key): + return ix.dropTryCandidate(ctx, key) + } + return nil +} + +// WriteTryCandidates writes the candidate of every live try that has none yet and whose previews are all there: after a +// preview was written, the try that waited for it. +func (ix *Indexer) WriteTryCandidates(ctx context.Context) error { + ix.mu.Lock() + ids := make([]string, 0, len(ix.tries)) + for id := range ix.tries { + ids = append(ids, id) + } + ix.mu.Unlock() + slices.Sort(ids) + for _, id := range ids { + if err := ix.writeTryCandidate(ctx, id); err != nil { + return err + } + } + return nil +} + +func (ix *Indexer) dropTryCandidate(ctx context.Context, id string) error { + if ix.Store == nil { + return nil + } + key, err := delivery.CandidatesKey(id) + if err != nil { + return err + } + return write(ctx, ix.Store, key, func(_ []byte, found bool) ([]byte, bool, error) { return nil, found, nil }) +} + +// DropTryCandidate deletes a try's candidate by hand (indexer_drop_try_candidate); a live try gets it again at its next +// change. +func (ix *Indexer) DropTryCandidate(ctx context.Context, id, by, why string) TakenIn { + r := TakenIn{Module: id} + switch { + case !IsTryGroup(id): + r.Answer = id + " is no try's group, try-: nothing dropped" + default: + if err := ix.dropTryCandidate(ctx, id); err != nil { + r.Answer = "the candidate of " + id + " could not be dropped: " + err.Error() + } else { + r.Stored, r.Answer = true, "the candidate of "+id+" is dropped by "+by+" ("+why+")" + ix.logf("%s", r.Answer) + } + } + return r +} + +// writeTryCandidate writes the candidate of a live try when it has none at its members' heads and every member's merge +// preview is held. +func (ix *Indexer) writeTryCandidate(ctx context.Context, id string) error { + if ix.Store == nil || ix.Read == nil { + return nil + } + ix.mu.Lock() + members := slices.Clone(ix.tries[id]) + ix.mu.Unlock() + if len(members) == 0 { + return nil + } + key, err := delivery.CandidatesKey(id) + if err != nil { + return err + } + slices.SortFunc(members, func(a, b TryMember) int { return strings.Compare(a.Repository, b.Repository) }) + heads := make([]delivery.Head, len(members)) + for i, m := range members { + heads[i] = delivery.Head{Repository: m.Repository, Head: m.Head} + } + hash, err := delivery.HeadsHash(heads) + if err != nil { + return err + } + if value, _, found, err := ix.Store.Get(ctx, key); err == nil && found { + if c, err := delivery.DecodeCandidate(value); err == nil && c.HeadsHash == hash { + return nil // fingerprints follow the previews (RecomputeCandidates) + } + } + previews := make([]delivery.Preview, len(members)) + for i, m := range members { + pk, err := delivery.PreviewKey(m.Repository, m.Head, m.Trunk) + if err != nil { + return fmt.Errorf("the try %s: %w", id, err) + } + ix.mu.Lock() + pv, ok := ix.previews[pk] + ix.mu.Unlock() + if !ok { + return nil // its fetch has not written the preview yet: this is called again when it does + } + previews[i] = pv + } + held, err := ix.heldModules(ctx) + if err != nil { + return err + } + pulls := make([]delivery.CandidatePull, len(members)) + var entries []PendingEntry + over := map[string]string{} + for i, m := range members { + pull := delivery.CandidatePull{Head: heads[i], Modules: []string{}} + raws := map[string]string{} + for _, dir := range slices.Sorted(mapKeys(previews[i].Modules)) { + pm := previews[i].Modules[dir] + name := "" + for n, h := range held { + if delivery.SameRepository(h.Repository, m.Repository) && h.Path == forgeDir(dir) { + name = n + if h.Manifest == pm.Manifest { + name = "" // the manifest the trunk holds: the try does not touch the module + } + break + } + } + if name == "" && heldAt(held, m.Repository, dir) { + continue + } + rr, err := ix.Read(ctx, m.Repository, m.Head, forgeDir(dir)) + if err != nil { + return fmt.Errorf("the try %s: reading the manifest of %s at %.12s: %w", id, dir, m.Head, err) + } + raw := rr.Manifests[forgeDir(dir)] + if raw == "" { + ix.logf("the try %s: %s holds no manifest at %.12s, so no module of it is taken", id, dir, m.Head) + continue + } + module := manifestModule(raw) + if module == "" { + module = path.Base(strings.Trim(dir, "/.")) + } + digest, err := ix.keepManifest(ctx, module, raw) + if err != nil { + return err + } + if digest != pm.Manifest { + ix.logf("the try %s: the manifest of %s read at %.12s is not the one its preview names, left", id, dir, m.Head) + continue + } + if pull.Fps == nil { + pull.Fps = map[string]delivery.CandidateModule{} + } + pull.Modules = append(pull.Modules, module) + pull.Fps[module] = delivery.CandidateModule{Manifest: digest, None2: NoneNoPreview} + raws[dir] = raw + over[module] = digest + } + pulls[i] = pull + entries = append(entries, PendingEntry{Forge: forgemsg.Forge{Repository: m.Repository, Merge: m.Head, Manifests: raws}}) + } + // The manifests are judged with the seat definitions the members carry (seatsof.go, ADR 0636 §4). + seats, err := ix.revisionSeats(ctx, entries) + if err != nil { + return fmt.Errorf("the try %s: reading the seat definitions of its members: %w", id, err) + } + judge.WithSeats(seats, func() { ix.judgeDigests(ctx, over, "the try "+id) }) + c := delivery.Candidate{V: delivery.Version, Group: id, HeadsHash: hash, Pulls: pulls} + c.FPSet = delivery.FingerprintSet(c.Pulls) + ix.withPreviews(ctx, &c) + if err := ix.refuseDisagreement(c); err != nil { + return err + } + data, err := delivery.EncodeCandidate(c) + if err != nil { + return err + } + return write(ctx, ix.Store, key, func(_ []byte, _ bool) ([]byte, bool, error) { return data, false, nil }) +} + +func mapKeys[V any](m map[string]V) func(func(string) bool) { + return func(yield func(string) bool) { + for k := range m { + if !yield(k) { + return + } + } + } +} + +// forgeDir is a preview's directory as the forge reader spells it: the root as "". +func forgeDir(dir string) string { + if dir == "." || dir == "/" { + return "" + } + return strings.Trim(dir, "/") +} + +// heldAt says the trunk holds a module of the repository in that directory. +func heldAt(held map[string]delivery.Module, repository, dir string) bool { + for _, h := range held { + if delivery.SameRepository(h.Repository, repository) && h.Path == forgeDir(dir) { + return true + } + } + return false +} diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index f04f6d04..d372e294 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -112,7 +112,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build {Name: graph.State, History: graph.History}, {Name: index.PendingState, History: 1}}; !reflect.DeepEqual(m.State, want) { t.Errorf("state = %+v, want %+v", m.State, want) } - if want := []string{"mesh-build-go.toolchains", "mesh-build-node.toolchains", "mesh-delivery-assigner.placement", "mesh-delivery-builder.build-log", "mesh-delivery-checker.previews", "mesh-delivery-gate.source"}; !reflect.DeepEqual(sorted(m.Reads), want) { + if want := []string{"mesh-build-go.toolchains", "mesh-build-node.toolchains", "mesh-delivery-assigner.placement", "mesh-delivery-builder.build-log", "mesh-delivery-checker.previews", "mesh-delivery-checker.tries", "mesh-delivery-gate.source"}; !reflect.DeepEqual(sorted(m.Reads), want) { t.Errorf("reads = %v, want %v", m.Reads, want) } // "Reads FORGE" is a consumes of the indexer's stream: the broker grants only buckets by reads (mesh-control @@ -137,7 +137,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build // (indexer_judge; issue 774, clause 3). 1966/AC11: indexer_old_sections reads the count of the old sections (state oldsections). // 1585/T12, T13 (ADR 0633 §5): indexer_pending reads the pending state; indexer_take_in_pending and indexer_drop_pending // ask through takeins; indexer_take_in_merge takes one merge in, its message asked of the forge bridge (issue 1585). - if want := []string{"indexer_take_in", "indexer_old_sections", "indexer_verdicts", "indexer_judge", index.PendingTool, index.TakeInPendingTool, index.DropPendingTool, index.TakeInMergeTool}; !reflect.DeepEqual(m.Tools, want) { + if want := []string{"indexer_take_in", "indexer_old_sections", "indexer_verdicts", "indexer_judge", index.PendingTool, index.TakeInPendingTool, index.DropPendingTool, index.TryCandidatesTool, index.DropTryCandidateTool, index.TakeInMergeTool}; !reflect.DeepEqual(m.Tools, want) { t.Errorf("tools = %v, want %v", m.Tools, want) } // ADR 0604 §2: it holds the seat mesh-catalog, serving its four verbs from its tools bundle (which stays without a diff --git a/modules/mesh-delivery-indexer/module.json b/modules/mesh-delivery-indexer/module.json index 3ae6276e..a468f962 100644 --- a/modules/mesh-delivery-indexer/module.json +++ b/modules/mesh-delivery-indexer/module.json @@ -22,6 +22,8 @@ "indexer_pending", "indexer_take_in_pending", "indexer_drop_pending", + "indexer_try_candidates", + "indexer_drop_try_candidate", "indexer_take_in_merge" ], "consumes": [ @@ -57,6 +59,7 @@ "mesh-delivery-assigner.placement", "mesh-delivery-builder.build-log", "mesh-delivery-checker.previews", + "mesh-delivery-checker.tries", "mesh-delivery-gate.source" ], "invokes": [