The indexer writes a group try's candidate from the checker's try and its previews (issue 1921)
A whole try has no pull request, yet compose-candidate and the manifest validator read the indexer's candidate of a group. The indexer reads the checker's tries, writes candidates.<try id> (number 0 per member) from the previews, judges its manifests with the seat definitions the members carry, leaves try-* alone in the open message, deletes it when the try ends, and has a tool that reads and one that drops it. Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
This commit is contained in:
@@ -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 " +
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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-<digits>"},
|
||||
"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-<digits>) 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
|
||||
}},
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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.<try id>`: 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-<digits>`
|
||||
// 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-<digits>: 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
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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": [
|
||||
|
||||
Reference in New Issue
Block a user