Show placement candidates and each node's answer in indexer_placement, add indexer_placement_reread, and drop a group's candidates when its head changes no placement file, so a missed message or a revert is recoverable by tool (issue 1126, ADR 0559 §3, §7, §8)
This commit is contained in:
@@ -47,6 +47,25 @@ func Tools(ask asker, now func() time.Time) []stdio.Tool {
|
||||
}
|
||||
return a.Result, nil
|
||||
}},
|
||||
{Name: index.PlacementRereadTool,
|
||||
Description: "Read the catalogue's placement files again now (novox/hq ADR 0559 §8): the indexer asks the forge for the trunk's head and " +
|
||||
"every open catalogue pull request's, and writes its states placement (placement.<node>) and placement-candidates (<group>.<node>) " +
|
||||
"as a merge and a pull request's head would, only where a key differs; a candidate key whose file no longer differs, or whose pull " +
|
||||
"request is closed, is deleted. The recovery for a forge message the indexer missed. Answers the heads read, the keys written or " +
|
||||
"deleted in each state, or why nothing was (the forge could not say). indexer_placement (mesh-delivery-view) reads the result.",
|
||||
Input: map[string]any{"type": "object", "properties": map[string]any{
|
||||
"why": map[string]any{"type": "string", "description": "why it is read again now, in plain words"}}},
|
||||
Run: func(args map[string]any) (any, error) {
|
||||
key, value, err := index.NewPlacementReread(str(args, "why"), stdio.CallerOf(args), now())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
a, err := ask(key, value)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return a.Reread, nil
|
||||
}},
|
||||
}
|
||||
}
|
||||
|
||||
@@ -64,7 +83,7 @@ func askTakeIn(s stdio.KeptState, bound time.Duration) asker {
|
||||
return false
|
||||
}
|
||||
a, err := index.DecodeTakeIn(v)
|
||||
if err != nil || a.Answered.IsZero() || a.Result == nil {
|
||||
if err != nil || !a.Done() {
|
||||
return false
|
||||
}
|
||||
select {
|
||||
|
||||
@@ -355,7 +355,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
defer calls.Close()
|
||||
ix := &index.Indexer{Store: kvStore{candidates}, Modules: kvStore{modules}, Manifests: kvStore{manifests}, Taken: awaitedStore{taken}, Builds: known, Logf: logger.Printf,
|
||||
Named: namedVia(calls.Call), Trunk: trunkVia(calls.Call), Contains: containsVia(calls.Call), Read: readVia(calls.Call),
|
||||
Placement: awaitedStore{placement}, PlacementCandidates: awaitedStore{placementCandidates}, Say: func(event string, body any) error { return calls.Emit(event, body) }}
|
||||
Placement: awaitedStore{placement}, PlacementCandidates: awaitedStore{placementCandidates}, ReadPlacement: placementVia(calls.Call), Say: func(event string, body any) error { return calls.Emit(event, body) }}
|
||||
// The gate's source bucket (mesh-delivery-gate_source) comes with the gate (release 3e, novox/hq ADR 0450 §1): its key
|
||||
// main is the mesh's source revision, which stamps the modules (issue 1344). Where it does not exist yet the indexer does
|
||||
// not wait for it (that needs a timer, ADR 0437 §8): it says so, and is started again with the gate.
|
||||
|
||||
@@ -170,3 +170,110 @@ func readVia(call toolCaller) func(ctx context.Context, repository, sha, dir str
|
||||
return out, nil
|
||||
}
|
||||
}
|
||||
|
||||
// The forge's tools the placement reread asks (indexer_placement_reread, novox/hq ADR 0559 §8): a repository's branches
|
||||
// with their commits, and every open pull request with its branch and head.
|
||||
const (
|
||||
listBranchesTool = "gitea.gitea_list_branches"
|
||||
openPullsTool = "gitea.gitea_open_pulls"
|
||||
placementTrunk = "main"
|
||||
)
|
||||
|
||||
// placementVia reads the catalogue's placement now through call: the trunk's head (gitea_list_branches), every placement
|
||||
// file at it, and each open pull request of the catalogue into the trunk (gitea_open_pulls) with every placement file at
|
||||
// its head, each read whole or not at all (gitea_commit_trees with listed, gitea_get_file).
|
||||
func placementVia(call toolCaller) func(ctx context.Context) (index.PlacementRead, error) {
|
||||
owner, repo, _ := strings.Cut(delivery.CatalogueRepository, "/")
|
||||
filesAt := func(sha string) (map[string]string, error) {
|
||||
raw, err := call(commitTreesTool, map[string]any{"owner": owner, "repo": repo, "sha": sha, "listed": []string{delivery.PlacementDir}})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var a struct {
|
||||
Listed map[string]map[string]string `json:"listed"`
|
||||
}
|
||||
if err := json.Unmarshal(raw, &a); err != nil || a.Listed == nil {
|
||||
return nil, fmt.Errorf("the answer of %s lists no directory: %v", commitTreesTool, err)
|
||||
}
|
||||
var out map[string]string
|
||||
for name := range a.Listed[delivery.PlacementDir] {
|
||||
file := delivery.PlacementDir + "/" + name
|
||||
node, ok := delivery.PlacementFileNode(file)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
raw, err := call(getFileTool, map[string]string{"owner": owner, "repo": repo, "path": file, "ref": sha})
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("%s at %s: %w", file, sha, err)
|
||||
}
|
||||
var f struct {
|
||||
File *struct {
|
||||
Content *string `json:"content"`
|
||||
} `json:"file"`
|
||||
}
|
||||
if err := json.Unmarshal(raw, &f); err != nil || f.File == nil || f.File.Content == nil {
|
||||
return nil, fmt.Errorf("the answer of %s for %s is not {file: {content}}: %v", getFileTool, file, err)
|
||||
}
|
||||
if out == nil {
|
||||
out = map[string]string{}
|
||||
}
|
||||
out[node] = *f.File.Content
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
return func(context.Context) (index.PlacementRead, error) {
|
||||
raw, err := call(listBranchesTool, map[string]string{"owner": owner, "repo": repo})
|
||||
if err != nil {
|
||||
return index.PlacementRead{}, err
|
||||
}
|
||||
var b struct {
|
||||
Branches []struct {
|
||||
Name string `json:"name"`
|
||||
Commit string `json:"commit"`
|
||||
} `json:"branches"`
|
||||
}
|
||||
if err := json.Unmarshal(raw, &b); err != nil {
|
||||
return index.PlacementRead{}, fmt.Errorf("the answer of %s is not {branches}: %w", listBranchesTool, err)
|
||||
}
|
||||
var out index.PlacementRead
|
||||
for _, br := range b.Branches {
|
||||
if br.Name == placementTrunk {
|
||||
out.Trunk = br.Commit
|
||||
}
|
||||
}
|
||||
if out.Trunk == "" {
|
||||
return index.PlacementRead{}, fmt.Errorf("%s names no branch %s of %s", listBranchesTool, placementTrunk, delivery.CatalogueRepository)
|
||||
}
|
||||
if out.Files, err = filesAt(out.Trunk); err != nil {
|
||||
return index.PlacementRead{}, err
|
||||
}
|
||||
raw, err = call(openPullsTool, map[string]any{})
|
||||
if err != nil {
|
||||
return index.PlacementRead{}, err
|
||||
}
|
||||
var o struct {
|
||||
Pulls *[]struct {
|
||||
Owner string `json:"owner"`
|
||||
Repo string `json:"repo"`
|
||||
Number int `json:"number"`
|
||||
Base string `json:"base"`
|
||||
Head string `json:"head"`
|
||||
HeadSHA string `json:"head_sha"`
|
||||
} `json:"pulls"`
|
||||
}
|
||||
if err := json.Unmarshal(raw, &o); err != nil || o.Pulls == nil {
|
||||
return index.PlacementRead{}, fmt.Errorf("the answer of %s is not {pulls}: %v", openPullsTool, err)
|
||||
}
|
||||
for _, p := range *o.Pulls {
|
||||
if !delivery.SameRepository(p.Owner+"/"+p.Repo, delivery.CatalogueRepository) || p.Base != placementTrunk || p.Number <= 0 || p.Head == "" || p.HeadSHA == "" {
|
||||
continue
|
||||
}
|
||||
files, err := filesAt(p.HeadSHA)
|
||||
if err != nil {
|
||||
return index.PlacementRead{}, fmt.Errorf("%s#%d: %w", delivery.CatalogueRepository, p.Number, err)
|
||||
}
|
||||
out.Pulls = append(out.Pulls, index.PullPlacement{Branch: p.Head, Number: p.Number, Head: p.HeadSHA, Files: files})
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
}
|
||||
|
||||
@@ -14,6 +14,7 @@ import (
|
||||
// takerIn is the part of the indexer a takeins value asks through.
|
||||
type takerIn interface {
|
||||
TakeIn(ctx context.Context, module string) index.TakenIn
|
||||
RereadPlacement(ctx context.Context) index.PlacementReread
|
||||
}
|
||||
|
||||
// takenIn answers one value of the takeins state (the tool indexer_take_in, issue 1945): a value not yet answered takes the
|
||||
@@ -31,9 +32,15 @@ func takenIn(ctx context.Context, ix takerIn, kv jetstream.KeyValue, e jetstream
|
||||
if !a.Answered.IsZero() {
|
||||
return
|
||||
}
|
||||
r := ix.TakeIn(ctx, a.Module)
|
||||
a.Result, a.Answered = &r, time.Now().UTC()
|
||||
logf("%s: a take-in asked by %s (%s): %s", a.Module, a.By, a.Why, r.Answer)
|
||||
if a.Placement {
|
||||
r := ix.RereadPlacement(ctx)
|
||||
a.Reread, a.Answered = &r, time.Now().UTC()
|
||||
logf("a placement reread asked by %s (%s): %s", a.By, a.Why, r.Answer)
|
||||
} else {
|
||||
r := ix.TakeIn(ctx, a.Module)
|
||||
a.Result, a.Answered = &r, time.Now().UTC()
|
||||
logf("%s: a take-in asked by %s (%s): %s", a.Module, a.By, a.Why, r.Answer)
|
||||
}
|
||||
data, err := json.Marshal(a)
|
||||
if err != nil {
|
||||
return
|
||||
|
||||
@@ -70,6 +70,9 @@ type Indexer struct {
|
||||
// PlacementCandidates is the placement candidates bucket (delivery.PlacementCandidatesBucket, issue 1126); nil writes
|
||||
// no candidate (placement_candidates.go).
|
||||
PlacementCandidates Store
|
||||
// ReadPlacement asks the forge for the catalogue's placement now, at the trunk's head and each open pull request's
|
||||
// (indexer_placement_reread, ADR 0559 §8); nil rereads none.
|
||||
ReadPlacement func(ctx context.Context) (PlacementRead, error)
|
||||
// Say says one of the indexer's events (placement-changed); nil says none.
|
||||
Say func(event string, body any) error
|
||||
// Drained says whether the indexer has applied every message of the forge stream (the consumer has none pending and none
|
||||
|
||||
@@ -18,7 +18,8 @@ import (
|
||||
// `<group>.<node>` of its state `placement-candidates`; a node whose file is the trunk's has none, and a key the head no
|
||||
// longer needs is deleted. The key is written only when its value changes, so its write is the node's event to compose
|
||||
// in shadow and answer. When the pull request closes or merges, or an open message no longer names it, its keys are
|
||||
// deleted. A pull message that read no placement (a forge resend, a pull request touching none) leaves them as they are.
|
||||
// deleted, and so are they when a head the forge says the changes of (DirsKnown) carries no placement: it touches none.
|
||||
// A pull message that says nothing of its head (a resend of the open pull requests) leaves them as they are.
|
||||
//
|
||||
// A file at the head that does not read as a placement, or names another node, is said in the log and that node gets no
|
||||
// key (one it had is deleted): the catalogue's own check refuses the pull request (ADR 0549 §1), and a node answers no
|
||||
@@ -34,7 +35,17 @@ func (ix *Indexer) applyPlacementCandidates(ctx context.Context, f delivery.Forg
|
||||
return c.Group == f.Branch && delivery.SameRepository(c.Pull.Repository, f.Repository) && c.Pull.Number == f.Number
|
||||
}, fmt.Sprintf("%s#%d %s", f.Repository, f.Number, map[bool]string{true: "merged", false: "closed"}[f.Kind == delivery.ForgeMerged]))
|
||||
}
|
||||
if f.Kind != delivery.ForgePull || !f.PlacementsRead {
|
||||
if f.Kind != delivery.ForgePull {
|
||||
return nil
|
||||
}
|
||||
if !f.PlacementsRead {
|
||||
if f.DirsKnown {
|
||||
// The forge said what this head changes (its directories and paths) and the bridge read no placement: the head
|
||||
// changes no placement file, so a key an earlier head left goes (ADR 0559 §3).
|
||||
return ix.dropPlacementCandidates(ctx, func(_ string, c delivery.PlacementCandidate) bool {
|
||||
return c.Group == f.Branch && delivery.SameRepository(c.Pull.Repository, f.Repository) && c.Pull.Number == f.Number
|
||||
}, fmt.Sprintf("%s#%d at %s changes no placement file", f.Repository, f.Number, f.Head))
|
||||
}
|
||||
return nil
|
||||
}
|
||||
trunk, err := ix.trunkPlacements(ctx)
|
||||
@@ -178,3 +189,114 @@ func (ix *Indexer) dropPlacementCandidates(ctx context.Context, gone func(node s
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// PlacementRead is the catalogue's placement as the forge has it now (indexer_placement_reread, ADR 0559 §8): the trunk's
|
||||
// head and every placement file at it, by node, and each open pull request into the trunk with its own.
|
||||
type PlacementRead struct {
|
||||
Trunk string
|
||||
Files map[string]string
|
||||
Pulls []PullPlacement
|
||||
}
|
||||
|
||||
// PullPlacement is one open catalogue pull request and every placement file at its head, by node.
|
||||
type PullPlacement struct {
|
||||
Branch string
|
||||
Number int
|
||||
Head string
|
||||
Files map[string]string
|
||||
}
|
||||
|
||||
// PlacementReread is the answer of indexer_placement_reread: the heads read, the keys written or deleted in each state,
|
||||
// and in words what happened.
|
||||
type PlacementReread struct {
|
||||
Trunk string `json:"trunk,omitempty"`
|
||||
Pulls []string `json:"pulls,omitempty"`
|
||||
Placement []string `json:"placement"`
|
||||
Candidates []string `json:"candidates"`
|
||||
Answer string `json:"answer"`
|
||||
}
|
||||
|
||||
// RereadPlacement reads the catalogue's placement now, at the trunk's head and at the head of each open catalogue pull
|
||||
// request, and writes placement and placement-candidates as a merged and a pull message would (ADR 0549 §2, ADR 0559
|
||||
// §3), only where a key differs; a candidate key of a pull request that is no longer open, or whose file no longer
|
||||
// differs at its head, is deleted. It is the recovery for a forge message the indexer missed (ADR 0559 §8).
|
||||
func (ix *Indexer) RereadPlacement(ctx context.Context) PlacementReread {
|
||||
out := PlacementReread{Placement: []string{}, Candidates: []string{}}
|
||||
if ix.ReadPlacement == nil || ix.Placement == nil || ix.PlacementCandidates == nil {
|
||||
out.Answer = "the indexer reads no placement here: it has no forge reader or no placement state"
|
||||
return out
|
||||
}
|
||||
read, err := ix.ReadPlacement(ctx)
|
||||
if err != nil {
|
||||
out.Answer = "the forge could not say the catalogue's placement, nothing is written: " + err.Error()
|
||||
return out
|
||||
}
|
||||
out.Trunk = read.Trunk
|
||||
before := revisions(ctx, ix.Placement)
|
||||
beforeC := revisions(ctx, ix.PlacementCandidates)
|
||||
if err := ix.applyPlacement(ctx, delivery.Forge{Kind: delivery.ForgeMerged, Repository: delivery.CatalogueRepository, Merge: read.Trunk,
|
||||
Placements: read.Files, PlacementsRead: true}); err != nil {
|
||||
out.Answer = "the trunk's placement could not be written: " + err.Error()
|
||||
return out
|
||||
}
|
||||
open := map[int]bool{}
|
||||
var failed []string
|
||||
for _, p := range read.Pulls {
|
||||
open[p.Number] = true
|
||||
out.Pulls = append(out.Pulls, fmt.Sprintf("%s#%d at %s", delivery.CatalogueRepository, p.Number, p.Head))
|
||||
f := delivery.Forge{Kind: delivery.ForgePull, Repository: delivery.CatalogueRepository, Branch: p.Branch, Number: p.Number, Head: p.Head,
|
||||
Action: delivery.PullUpdated, Placements: p.Files, PlacementsRead: true}
|
||||
if err := ix.applyPlacementCandidates(ctx, f); err != nil {
|
||||
failed = append(failed, fmt.Sprintf("#%d: %v", p.Number, err))
|
||||
}
|
||||
}
|
||||
if err := ix.dropPlacementCandidates(ctx, func(_ string, c delivery.PlacementCandidate) bool {
|
||||
return delivery.SameRepository(c.Pull.Repository, delivery.CatalogueRepository) && !open[c.Pull.Number]
|
||||
}, "its pull request is not open (indexer_placement_reread)"); err != nil {
|
||||
failed = append(failed, err.Error())
|
||||
}
|
||||
out.Placement = changed(before, revisions(ctx, ix.Placement))
|
||||
out.Candidates = changed(beforeC, revisions(ctx, ix.PlacementCandidates))
|
||||
switch {
|
||||
case len(failed) > 0:
|
||||
out.Answer = "read at the trunk's head " + read.Trunk + "; some candidates are not written: " + fmt.Sprint(failed)
|
||||
case len(out.Placement)+len(out.Candidates) == 0:
|
||||
out.Answer = "read at the trunk's head " + read.Trunk + " and every open catalogue pull request's: every key was as the forge has it, nothing written"
|
||||
default:
|
||||
out.Answer = fmt.Sprintf("read at the trunk's head %s and every open catalogue pull request's: %d placement keys and %d candidate keys written or deleted",
|
||||
read.Trunk, len(out.Placement), len(out.Candidates))
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// revisions is every key of a store with its revision; a state not on the bus yet has none.
|
||||
func revisions(ctx context.Context, st Store) map[string]uint64 {
|
||||
out := map[string]uint64{}
|
||||
keys, err := st.Keys(ctx)
|
||||
if err != nil {
|
||||
return out
|
||||
}
|
||||
for _, key := range keys {
|
||||
if _, rev, found, err := st.Get(ctx, key); err == nil && found {
|
||||
out[key] = rev
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// changed is every key written, created or deleted between two revisions maps, sorted.
|
||||
func changed(before, after map[string]uint64) []string {
|
||||
out := []string{}
|
||||
for key, rev := range after {
|
||||
if before[key] != rev {
|
||||
out = append(out, key)
|
||||
}
|
||||
}
|
||||
for key := range before {
|
||||
if _, ok := after[key]; !ok {
|
||||
out = append(out, key)
|
||||
}
|
||||
}
|
||||
slices.Sort(out)
|
||||
return out
|
||||
}
|
||||
|
||||
@@ -133,6 +133,8 @@ func placementIsWrittenAtAMergeOnlyWhereItChanged(t *testing.T) {
|
||||
{"a new head removing ace", pullOf("feat/p", 9, h2, map[string]string{"g14": g14New}), "feat/p.ace,feat/p.g14"},
|
||||
{"a head whose g14 file does not read", pullOf("feat/p", 9, h3, map[string]string{"g14": `{"v":1,"node":"ace","modules":[]}`}), "feat/p.ace"},
|
||||
{"the head adding g14 again", pullOf("feat/p", 9, h2, map[string]string{"g14": g14New}), "feat/p.ace,feat/p.g14"},
|
||||
{"a head the forge says changes no placement file", func() delivery.Forge { f := pullOf("feat/p", 9, h3, nil); f.DirsKnown = true; return f }(), ""},
|
||||
{"the head adding g14 and removing ace again", pullOf("feat/p", 9, h2, map[string]string{"g14": g14New}), "feat/p.ace,feat/p.g14"},
|
||||
{"an open message not naming it", delivery.Forge{V: delivery.Version, Kind: delivery.ForgeOpen, Open: []delivery.OpenPull{{Repository: delivery.CatalogueRepository, Number: 10}}}, ""},
|
||||
{"the pull request heard again", pullOf("feat/p", 9, h2, map[string]string{"g14": g14New}), "feat/p.ace,feat/p.g14"},
|
||||
{"its merge", func() delivery.Forge {
|
||||
@@ -168,4 +170,32 @@ func placementIsWrittenAtAMergeOnlyWhereItChanged(t *testing.T) {
|
||||
if p, _ := written("g14"); len(p.Modules) != 1 || p.Revision != "m7" {
|
||||
t.Errorf("the merge's trunk placement of g14: %+v", p)
|
||||
}
|
||||
|
||||
// ADR 0559 §8, indexer_placement_reread: a reread with nothing changed writes nothing; one after a missed merge and a
|
||||
// missed head writes those keys; one after a missed revert deletes the candidate; a forge that cannot say writes nothing.
|
||||
g14Other := `{"v":1,"node":"g14","modules":[{"module":"zsh"},{"module":"htop"}]}`
|
||||
for _, row := range []struct {
|
||||
name string
|
||||
read PlacementRead
|
||||
err error
|
||||
placement, cands []string
|
||||
}{
|
||||
{"nothing changed", PlacementRead{Trunk: "m7", Files: map[string]string{"g14": g14New}}, nil, []string{}, []string{}},
|
||||
{"a missed merge and a missed head", PlacementRead{Trunk: "m8", Files: map[string]string{"g14": g14New, "ace": ace},
|
||||
Pulls: []PullPlacement{{Branch: "feat/r", Number: 11, Head: h1, Files: map[string]string{"g14": g14Other, "ace": ace}}}}, nil,
|
||||
[]string{"placement.ace"}, []string{"feat/r.g14"}},
|
||||
{"a missed revert", PlacementRead{Trunk: "m8", Files: map[string]string{"g14": g14New, "ace": ace},
|
||||
Pulls: []PullPlacement{{Branch: "feat/r", Number: 11, Head: h2, Files: map[string]string{"g14": g14New, "ace": ace}}}}, nil,
|
||||
[]string{}, []string{"feat/r.g14"}},
|
||||
{"a forge that cannot say", PlacementRead{}, context.DeadlineExceeded, []string{}, []string{}},
|
||||
} {
|
||||
ix.ReadPlacement = func(context.Context) (PlacementRead, error) { return row.read, row.err }
|
||||
r := ix.RereadPlacement(ctx)
|
||||
if !reflect.DeepEqual(r.Placement, row.placement) || !reflect.DeepEqual(r.Candidates, row.cands) || r.Answer == "" {
|
||||
t.Errorf("%s: %+v", row.name, r)
|
||||
}
|
||||
}
|
||||
if keys() != "" {
|
||||
t.Errorf("after the reread of a revert, candidates %q", keys())
|
||||
}
|
||||
}
|
||||
|
||||
@@ -17,6 +17,10 @@ import (
|
||||
// of the indexer's states stay in mesh-delivery-view.
|
||||
const (
|
||||
TakeInTool = "indexer_take_in"
|
||||
// PlacementRereadTool reads the catalogue's placement again now (ADR 0559 §8); it asks through the same state, under
|
||||
// PlacementRereadKey, a key no module has (a module's name holds no dot).
|
||||
PlacementRereadTool = "indexer_placement_reread"
|
||||
PlacementRereadKey = "placement.reread"
|
||||
// TakeInsState is the indexer's state of take-in asks: the bucket mesh-delivery-indexer_takeins, keyed as the modules state.
|
||||
TakeInsState = "takeins"
|
||||
TakeInsBucket = delivery.IndexerModule + "_" + TakeInsState
|
||||
@@ -26,12 +30,24 @@ const (
|
||||
// indexer read it, its answer.
|
||||
type TakeInAsk struct {
|
||||
V int `json:"v"`
|
||||
Module string `json:"module"`
|
||||
Module string `json:"module,omitempty"`
|
||||
Why string `json:"why,omitempty"`
|
||||
By string `json:"by,omitempty"`
|
||||
At time.Time `json:"at"`
|
||||
Result *TakenIn `json:"result,omitempty"`
|
||||
Answered time.Time `json:"answered,omitzero"`
|
||||
// Placement asks indexer_placement_reread instead of a module's take-in; Reread is its answer.
|
||||
Placement bool `json:"placement,omitempty"`
|
||||
Reread *PlacementReread `json:"reread,omitempty"`
|
||||
}
|
||||
|
||||
// Done says the ask is answered.
|
||||
func (a TakeInAsk) Done() bool { return !a.Answered.IsZero() && (a.Result != nil || a.Reread != nil) }
|
||||
|
||||
// NewPlacementReread is the value indexer_placement_reread writes: its key and the value.
|
||||
func NewPlacementReread(why, by string, at time.Time) (string, []byte, error) {
|
||||
data, err := json.Marshal(TakeInAsk{V: 1, Placement: true, Why: strings.TrimSpace(why), By: by, At: at.UTC()})
|
||||
return PlacementRereadKey, data, err
|
||||
}
|
||||
|
||||
// DecodeTakeIn reads a takeins value.
|
||||
@@ -43,7 +59,7 @@ 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 == "" {
|
||||
if a.Module == "" && !a.Placement {
|
||||
return a, errors.New("a takeins value naming no module")
|
||||
}
|
||||
return a, nil
|
||||
|
||||
@@ -105,11 +105,12 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
|
||||
// Three invokes: the file of a registered module's manifest and lock files (registered.go), the forge's trees (the named paths' objects, since the indexer is the one reader of a manifest's paths:
|
||||
// ADR 0499 clauses 2 and 3, ADR 0495 clause 1; and the module directories of the trunk, issue 1900) and whether a
|
||||
// commit is in another (issue 1900).
|
||||
if want := []string{"gitea.gitea_commit_trees", "gitea.gitea_contains", "gitea.gitea_get_file"}; !reflect.DeepEqual(m.Invokes, want) {
|
||||
// ADR 0559 §8: the placement reread also asks the catalogue's branches and open pull requests.
|
||||
if want := []string{"gitea.gitea_commit_trees", "gitea.gitea_contains", "gitea.gitea_get_file", "gitea.gitea_list_branches", "gitea.gitea_open_pulls"}; !reflect.DeepEqual(m.Invokes, want) {
|
||||
t.Errorf("invokes = %v, want %v", m.Invokes, want)
|
||||
}
|
||||
// 1945: one tool, indexer_take_in, served by its tools bundle through the takeins state; the process serves none.
|
||||
if want := []string{"indexer_take_in"}; !reflect.DeepEqual(m.Tools, want) {
|
||||
// 1945: indexer_take_in, and indexer_placement_reread (ADR 0559 §8), served by its tools bundle through the takeins state; the process serves none.
|
||||
if want := []string{"indexer_take_in", "indexer_placement_reread"}; !reflect.DeepEqual(m.Tools, want) {
|
||||
t.Errorf("tools = %v, want %v", m.Tools, want)
|
||||
}
|
||||
// No streams, no bus, no seat (ADR 0437 §4); its one event is placement-changed (ADR 0549 §2), checked above.
|
||||
|
||||
@@ -3,7 +3,8 @@
|
||||
"version": "1",
|
||||
"slug": "indexer",
|
||||
"tools": [
|
||||
"indexer_take_in"
|
||||
"indexer_take_in",
|
||||
"indexer_placement_reread"
|
||||
],
|
||||
"emits": [
|
||||
"placement-changed"
|
||||
@@ -34,7 +35,9 @@
|
||||
"invokes": [
|
||||
"gitea.gitea_commit_trees",
|
||||
"gitea.gitea_contains",
|
||||
"gitea.gitea_get_file"
|
||||
"gitea.gitea_get_file",
|
||||
"gitea.gitea_list_branches",
|
||||
"gitea.gitea_open_pulls"
|
||||
],
|
||||
"own-secrets": {
|
||||
"broker": "${dir:state}/broker"
|
||||
|
||||
@@ -65,7 +65,9 @@ func Tools(r view.Reader, u view.UsageReader) []stdio.Tool {
|
||||
{Name: view.PlacementReadTool,
|
||||
Description: "Where modules run, as the indexer read it from the catalogue's placement files at the last merge (placement.<node>, novox/hq ADR 0549): " +
|
||||
"each node's modules in order with their host ports, bindings and settings placement, its pins and the seats its modules hold, and the catalogue revision. " +
|
||||
"With node, that node's; with module, every node placing it; with neither, every node's. Keys that do not read are listed as unreadable. Read-only.",
|
||||
"With node, that node's; with module, every node placing it; with neither, every node's. Keys that do not read are listed as unreadable. " +
|
||||
"Beside it, candidates: each open catalogue pull request's placement of a node that differs from the trunk's (placement-candidates <group>.<node>, issue 1126), " +
|
||||
"narrowed the same way, with the node's answer from its <node>.node have (answered: same, differs, refused, stale for an earlier head, none when it has not answered). Read-only.",
|
||||
Input: map[string]any{"type": "object", "properties": nodeModule},
|
||||
Run: func(args map[string]any) (any, error) {
|
||||
node, module := nodeModuleOf(args)
|
||||
@@ -202,8 +204,9 @@ func read(state stdio.KeptState) view.StateRead {
|
||||
func main() {
|
||||
r := view.Reader{State: read(stdio.State("mesh-delivery-indexer.modules")), Builds: read(stdio.State("mesh-delivery-builder.builds")), CandidatesState: read(stdio.State("mesh-delivery-indexer.candidates")),
|
||||
WantsState: read(stdio.State("mesh-delivery-assigner.wants")), DifferencesState: read(stdio.State("mesh-delivery-assigner.differences")), HavesState: read(stdio.State("node-engine.have")),
|
||||
ReachState: read(stdio.State("reach")),
|
||||
PlacementState: read(stdio.State("mesh-delivery-indexer.placement"))}
|
||||
ReachState: read(stdio.State("reach")),
|
||||
PlacementState: read(stdio.State("mesh-delivery-indexer.placement")),
|
||||
PlacementCandidatesState: read(stdio.State("mesh-delivery-indexer.placement-candidates"))}
|
||||
u := view.UsageReader{Haves: r.HavesState, Resolved: read(stdio.State("mesh-config.resolved")), Tries: read(stdio.State("mesh-delivery-checker.tries")),
|
||||
Steps: map[string]view.StateRead{}, Pauses: map[string]view.StateRead{}, Backlog: backlogOf(socket())}
|
||||
for _, lang := range delivery.AskLanguages {
|
||||
|
||||
@@ -131,6 +131,50 @@ func TestModulesToolAnswersEveryKeyAsStored(t *testing.T) {
|
||||
if _, err := run(t, p, view.PlacementReadTool, map[string]any{"node": "novox"}); err == nil || !strings.Contains(err.Error(), "no placement of novox") {
|
||||
t.Errorf("a node with no key: %v", err)
|
||||
}
|
||||
if c := out.(map[string]any)["candidates"].([]view.CandidateRow); len(c) != 0 {
|
||||
t.Errorf("with no candidate open: %v", c)
|
||||
}
|
||||
|
||||
// ADR 0559 §7 (issue 1126): beside the trunk's, each open candidate key and the node's answer from its <node>.node:
|
||||
// answered at the key's head and revision, at an earlier head (stale), or not at all (none); a node the trunk does not
|
||||
// place yet is shown by its candidate alone.
|
||||
const h1, h2 = "1111111111111111111111111111111111111111", "2222222222222222222222222222222222222222"
|
||||
cand := func(node, head string, placed bool) string {
|
||||
pl := `,"placement":{"v":1,"node":"` + node + `","modules":[{"module":"plex"}]}`
|
||||
if !placed {
|
||||
pl = ""
|
||||
}
|
||||
return `{"v":1,"group":"feat/p","pull":{"repository":"novox/mesh-catalog","number":9,"head":"` + head + `"}` + pl + `}`
|
||||
}
|
||||
cands := fake{"feat/p.ace": cand("ace", h2, true), "feat/p.g14": cand("g14", h2, false), "feat/p.novox": cand("novox", h2, true)}
|
||||
revs := map[string]uint64{"feat/p.ace": 4, "feat/p.g14": 5, "feat/p.novox": 6}
|
||||
p.PlacementCandidatesState = view.StateRead{Keys: cands.reader().State.Keys, GetEntry: func(key string) (json.RawMessage, uint64, error) {
|
||||
raw, _ := cands.reader().State.Get(key)
|
||||
return raw, revs[key], nil
|
||||
}}
|
||||
p.HavesState = fake{
|
||||
"ace.node": `{"v":1,"composed":"x","candidates":{"feat/p":{"head":"` + h2 + `","revision":4,"verdict":"same","same":["plex"]}}}`,
|
||||
"g14.node": `{"v":1,"composed":"x","candidates":{"feat/p":{"head":"` + h1 + `","revision":2,"verdict":"same"}}}`,
|
||||
}.reader().State
|
||||
out, err = run(t, p, view.PlacementReadTool, map[string]any{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
answered := map[string]string{}
|
||||
for _, c := range out.(map[string]any)["candidates"].([]view.CandidateRow) {
|
||||
answered[c.Node] = c.Answered
|
||||
}
|
||||
if want := map[string]string{"ace": "same", "g14": view.AnswerStale, "novox": view.AnswerNone}; fmt.Sprint(answered) != fmt.Sprint(want) {
|
||||
t.Errorf("answers %v, want %v", answered, want)
|
||||
}
|
||||
out, err = run(t, p, view.PlacementReadTool, map[string]any{"node": "novox"})
|
||||
if c := out.(map[string]any)["candidates"].([]view.CandidateRow); err != nil || len(c) != 1 || c[0].Pull.Head != h2 || len(out.(map[string]any)["placement"].(map[string]json.RawMessage)) != 0 {
|
||||
t.Errorf("a node placed by its candidate alone: %v, %v", out, err)
|
||||
}
|
||||
out, _ = run(t, p, view.PlacementReadTool, map[string]any{"module": "plex"})
|
||||
if c := out.(map[string]any)["candidates"].([]view.CandidateRow); len(c) != 2 {
|
||||
t.Errorf("the candidates placing plex: %v (g14's removes its file)", c)
|
||||
}
|
||||
}
|
||||
|
||||
const (
|
||||
|
||||
@@ -3,7 +3,7 @@ module mesh-delivery-view
|
||||
go 1.26.0
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026211913-68d65e6bbf61
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011211-82f10c3bc9fa
|
||||
github.com/nats-io/nats-server/v2 v2.11.17
|
||||
github.com/nats-io/nats.go v1.54.0
|
||||
)
|
||||
|
||||
@@ -1,5 +1,9 @@
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026211913-68d65e6bbf61 h1:/c6ncICWjq55nBf8TlDM0ROe/PuH+qZIdnuVbtg8UuM=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026211913-68d65e6bbf61/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027010747-7e99c354ac7f h1:aWqn4HHTPUp/+28h7b8uFqQptAYvQFs05CALMHN7SkI=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027010747-7e99c354ac7f/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011211-82f10c3bc9fa h1:tT6sl3vY0iVAXw+mqy3iGKUBUjVBRWvQvKUOTmTyWwk=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011211-82f10c3bc9fa/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
|
||||
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
|
||||
|
||||
@@ -43,6 +43,8 @@ type Reader struct {
|
||||
HavesState StateRead // the node-engine's haves
|
||||
ReachState StateRead // the view's own reach (ADR 0556)
|
||||
PlacementState StateRead // the indexer's placement
|
||||
// PlacementCandidatesState is the indexer's placement candidates (issue 1126); unset reads none.
|
||||
PlacementCandidatesState StateRead
|
||||
}
|
||||
|
||||
// ReachReadTool reads the view's reach: per repository, how far each module reached and the reached head (ADR 0556).
|
||||
@@ -652,11 +654,19 @@ func (r Reader) Haves(node, module string) (KeptAnswer, error) {
|
||||
// Placement is the indexer's placement as stored (ADR 0549 §2): one node's key with node, every node placing a module
|
||||
// with module (each node's key whole, so its pins, ports and bindings are beside the module), and every node's key with
|
||||
// neither. A key that does not read as a placement is listed as unreadable, never dropped.
|
||||
//
|
||||
// Beside it, candidates (issue 1126, ADR 0549 §4): each open catalogue pull request's placement of a node that differs from
|
||||
// the trunk's (the indexer's placement-candidates `<group>.<node>`), narrowed the same way, each with the node's answer from
|
||||
// its `<node>.node` have and whether that answer is for the candidate as it stands (its head and key revision).
|
||||
func (r Reader) Placement(node, module string) (map[string]any, error) {
|
||||
node, module = strings.TrimSpace(node), strings.TrimSpace(module)
|
||||
if r.PlacementState.Keys == nil || r.PlacementState.Get == nil {
|
||||
return nil, fmt.Errorf("the placement state cannot be read here")
|
||||
}
|
||||
candidates, err := r.PlacementCandidates(node, module)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if node != "" {
|
||||
key, err := delivery.PlacementKey(node)
|
||||
if err != nil {
|
||||
@@ -666,10 +676,14 @@ func (r Reader) Placement(node, module string) (map[string]any, error) {
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if len(raw) == 0 {
|
||||
if len(raw) == 0 && len(candidates) == 0 {
|
||||
return nil, fmt.Errorf("no placement of %s: the catalogue holds no placement file for it, or the indexer has read none yet", node)
|
||||
}
|
||||
return map[string]any{"placement": map[string]json.RawMessage{node: raw}}, nil
|
||||
out := map[string]json.RawMessage{}
|
||||
if len(raw) > 0 {
|
||||
out[node] = raw
|
||||
}
|
||||
return map[string]any{"placement": out, "candidates": candidates}, nil
|
||||
}
|
||||
keys, err := r.PlacementState.Keys()
|
||||
if err != nil {
|
||||
@@ -698,5 +712,102 @@ func (r Reader) Placement(node, module string) (map[string]any, error) {
|
||||
}
|
||||
out[n] = values[i]
|
||||
}
|
||||
return map[string]any{"placement": out, "unreadable": unreadable}, nil
|
||||
return map[string]any{"placement": out, "unreadable": unreadable, "candidates": candidates}, nil
|
||||
}
|
||||
|
||||
// The states of a node's answer to a placement candidate, as the view and the assigner's check read it.
|
||||
const (
|
||||
// AnswerNone: the node's have holds no answer for the candidate's group (a node asleep, or not composing yet).
|
||||
AnswerNone = "none"
|
||||
// AnswerStale: the node answered an earlier head or revision of the candidate.
|
||||
AnswerStale = "stale"
|
||||
)
|
||||
|
||||
// CandidateRow is one placement candidate and the node's answer to it.
|
||||
type CandidateRow struct {
|
||||
Group string `json:"group"`
|
||||
Node string `json:"node"`
|
||||
Pull delivery.Head `json:"pull"`
|
||||
Revision uint64 `json:"revision"`
|
||||
Placement json.RawMessage `json:"placement,omitempty"` // the candidate value as stored
|
||||
Removes bool `json:"removes,omitempty"` // the head removes the node's file
|
||||
Answered string `json:"answered"` // same, differs, refused, stale or none
|
||||
Answer *delivery.CandidateAnswer `json:"answer,omitempty"`
|
||||
}
|
||||
|
||||
// PlacementCandidates is every placement candidate, narrowed by node and by a module its placement names, with each
|
||||
// node's answer. A key that does not read is a row naming only its key's group and node, answered none, never dropped.
|
||||
func (r Reader) PlacementCandidates(node, module string) ([]CandidateRow, error) {
|
||||
st := r.PlacementCandidatesState
|
||||
if st.Keys == nil || st.GetEntry == nil {
|
||||
return []CandidateRow{}, nil
|
||||
}
|
||||
keys, err := st.Keys()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("reading the placement candidates: %w", err)
|
||||
}
|
||||
sort.Strings(keys)
|
||||
answers := map[string]map[string]delivery.CandidateAnswer{} // node, then group
|
||||
rows := []CandidateRow{}
|
||||
for _, key := range keys {
|
||||
group, n, err := delivery.ParsePlacementCandidateKey(key)
|
||||
if err != nil || (node != "" && n != node) {
|
||||
continue
|
||||
}
|
||||
raw, rev, err := st.GetEntry(key)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("reading %s: %w", key, err)
|
||||
}
|
||||
if len(raw) == 0 {
|
||||
continue
|
||||
}
|
||||
row := CandidateRow{Group: group, Node: n, Revision: rev, Placement: raw, Answered: AnswerNone}
|
||||
c, err := delivery.DecodePlacementCandidate(raw)
|
||||
if err == nil {
|
||||
row.Pull, row.Removes = c.Pull, c.Placement == nil
|
||||
if module != "" && (c.Placement == nil || !slices.Contains(c.Placement.Assigned(), module)) {
|
||||
continue
|
||||
}
|
||||
} else if module != "" {
|
||||
continue
|
||||
}
|
||||
if _, read := answers[n]; !read {
|
||||
answers[n] = r.candidateAnswers(n)
|
||||
}
|
||||
if a, ok := answers[n][group]; ok && err == nil {
|
||||
row.Answer = &a
|
||||
row.Answered = AnswerOf(c, rev, a)
|
||||
}
|
||||
rows = append(rows, row)
|
||||
}
|
||||
return rows, nil
|
||||
}
|
||||
|
||||
// AnswerOf is the state of a node's answer a to the candidate c at key revision rev: its verdict when it answers that
|
||||
// head and revision, AnswerStale otherwise.
|
||||
func AnswerOf(c delivery.PlacementCandidate, rev uint64, a delivery.CandidateAnswer) string {
|
||||
if a.Head != c.Pull.Head || a.Revision != rev {
|
||||
return AnswerStale
|
||||
}
|
||||
return a.Verdict
|
||||
}
|
||||
|
||||
// candidateAnswers is a node's answers to its candidates, by group, from its `<node>.node` have; none when it has none.
|
||||
func (r Reader) candidateAnswers(node string) map[string]delivery.CandidateAnswer {
|
||||
if r.HavesState.Get == nil {
|
||||
return nil
|
||||
}
|
||||
key, err := delivery.HaveNodeKey(node)
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
raw, err := r.HavesState.Get(key)
|
||||
if err != nil || len(raw) == 0 {
|
||||
return nil
|
||||
}
|
||||
h, err := delivery.DecodeHaveNode(raw)
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
return h.Candidates
|
||||
}
|
||||
|
||||
@@ -87,7 +87,7 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7
|
||||
if m.Module != "mesh-delivery-view" {
|
||||
t.Errorf("module = %q", m.Module)
|
||||
}
|
||||
want := []string{"mesh-build-go.pauses", "mesh-build-go.steps", "mesh-build-image.pauses", "mesh-build-image.steps", "mesh-build-node.pauses", "mesh-build-node.steps", "mesh-config.resolved", "mesh-controller.assignments", "mesh-controller.builds", "mesh-controller.composed", "mesh-delivery-assigner.asks", "mesh-delivery-assigner.differences", "mesh-delivery-assigner.wants", "mesh-delivery-builder.builds", "mesh-delivery-checker.tries", "mesh-delivery-gate.asks", "mesh-delivery-gate.source", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.modules", "mesh-delivery-indexer.placement", "mesh-delivery-indexer.taken", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "mesh-delivery-watchman.stuck", "node-engine.have"} // build_usage reads the build modules' steps and pauses, mesh-config's capacities and the checker's tries; 1168/AC1, 1241/AC4, 1345/AC3, 1297/B8, B9
|
||||
want := []string{"mesh-build-go.pauses", "mesh-build-go.steps", "mesh-build-image.pauses", "mesh-build-image.steps", "mesh-build-node.pauses", "mesh-build-node.steps", "mesh-config.resolved", "mesh-controller.assignments", "mesh-controller.builds", "mesh-controller.composed", "mesh-delivery-assigner.asks", "mesh-delivery-assigner.differences", "mesh-delivery-assigner.wants", "mesh-delivery-builder.builds", "mesh-delivery-checker.tries", "mesh-delivery-gate.asks", "mesh-delivery-gate.source", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.modules", "mesh-delivery-indexer.placement", "mesh-delivery-indexer.placement-candidates", "mesh-delivery-indexer.taken", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "mesh-delivery-watchman.stuck", "node-engine.have"} // indexer_placement reads the placement candidates and every <node>.node (ADR 0559 §7); build_usage reads the build modules' steps and pauses, mesh-config's capacities and the checker's tries; 1168/AC1, 1241/AC4, 1345/AC3, 1297/B8, B9
|
||||
got := slices.Clone(m.Reads)
|
||||
slices.Sort(got)
|
||||
if !reflect.DeepEqual(got, want) {
|
||||
|
||||
@@ -49,6 +49,7 @@
|
||||
"mesh-delivery-indexer.candidates",
|
||||
"mesh-delivery-indexer.modules",
|
||||
"mesh-delivery-indexer.placement",
|
||||
"mesh-delivery-indexer.placement-candidates",
|
||||
"mesh-delivery-indexer.taken",
|
||||
"mesh-delivery-teller.answers",
|
||||
"mesh-delivery-watchman.presence",
|
||||
|
||||
Reference in New Issue
Block a user