Indexer: stop holding the catalogue's placement, it is the seat mesh-placement's now (issue 1126)
Deletes the placement and candidate writes, the reread tool and its forge reads, the two buckets and the placement-changed event. D1a and D1b are one commit: the index package, the named reads, the tools and the manifest do not compile apart.
This commit is contained in:
@@ -49,25 +49,6 @@ 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
|
||||
}},
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -273,12 +273,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
// Its taken state (ADR 0556 clause 1) is bound when it appears, as every state a delivery adds (issue 1790): until then
|
||||
// a merge's taken key is said and not written, and the rest of the indexer runs.
|
||||
taken := process.AwaitBucket(ctx, js, conn, delivery.TakenBucket, "no repository's taken key is written", logger.Printf)
|
||||
// Its placement state (ADR 0549 §2) is bound when it appears, as every state a delivery adds (issue 1790): until then a
|
||||
// catalogue merge's placement is said and not written, and the rest of the indexer runs.
|
||||
placement := process.AwaitBucket(ctx, js, conn, delivery.PlacementBucket, "no catalogue merge's placement is written", logger.Printf)
|
||||
// Its placement candidates (issue 1126, ADR 0549 §4) are bound the same way: until the state appears a catalogue pull
|
||||
// request's placement is said and not written, so no node answers it and its check names every placed node as silent.
|
||||
placementCandidates := process.AwaitBucket(ctx, js, conn, delivery.PlacementCandidatesBucket, "no catalogue pull request's placement candidate is written", logger.Printf)
|
||||
// The controller's build records only name the directory of a module whose key names none (fillPaths): without them
|
||||
// such a module waits for its path, every other is indexed (issue 1945).
|
||||
builds := process.OptionalBucket(ctx, js, delivery.BuildsBucket, "a module whose key names no directory is not given one from the controller's build records until the indexer starts again", logger.Printf)
|
||||
@@ -310,7 +304,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}, ReadPlacement: placementVia(calls.Call), ReadOpen: openVia(calls.Call), Say: func(event string, body any) error { return calls.Emit(event, body) }}
|
||||
ReadOpen: openVia(calls.Call)}
|
||||
// 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.
|
||||
|
||||
@@ -171,12 +171,10 @@ func readVia(call toolCaller) func(ctx context.Context, repository, sha, dir str
|
||||
}
|
||||
}
|
||||
|
||||
// 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.
|
||||
// The forge's tool for the open pull requests, and the trunk branch they are open into.
|
||||
const (
|
||||
listBranchesTool = "gitea.gitea_list_branches"
|
||||
openPullsTool = "gitea.gitea_open_pulls"
|
||||
placementTrunk = "main"
|
||||
openPullsTool = "gitea.gitea_open_pulls"
|
||||
trunkBranch = "main"
|
||||
)
|
||||
|
||||
// openVia asks the forge which pull requests into the trunk are open now (gitea_open_pulls), every repository's.
|
||||
@@ -199,109 +197,10 @@ func openVia(call toolCaller) func(ctx context.Context) ([]delivery.OpenPull, er
|
||||
}
|
||||
out := []delivery.OpenPull{}
|
||||
for _, p := range *o.Pulls {
|
||||
if p.Base == placementTrunk && p.Number > 0 {
|
||||
if p.Base == trunkBranch && p.Number > 0 {
|
||||
out = append(out, delivery.OpenPull{Repository: p.Owner + "/" + p.Repo, Number: p.Number})
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
}
|
||||
|
||||
// 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,7 +14,6 @@ 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
|
||||
@@ -32,15 +31,9 @@ func takenIn(ctx context.Context, ix takerIn, kv jetstream.KeyValue, e jetstream
|
||||
if !a.Answered.IsZero() {
|
||||
return
|
||||
}
|
||||
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)
|
||||
}
|
||||
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
|
||||
|
||||
@@ -65,20 +65,10 @@ type Indexer struct {
|
||||
// Contains says whether a commit is the second or an ancestor of it (gitea.gitea_contains).
|
||||
Contains func(ctx context.Context, repository, commit, in string) (bool, error)
|
||||
Logf func(format string, args ...any)
|
||||
// Placement is the placement bucket (delivery.PlacementBucket, ADR 0549 §2); nil writes no placement.
|
||||
Placement Store
|
||||
// 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)
|
||||
// ReadOpen asks the forge which pull requests into the trunk are open now (gitea_open_pulls). A catalogue pull message
|
||||
// that finds the candidates of another pull request asks it and deletes those of a pull request no longer open, so a
|
||||
// closed message the indexer missed heals at the next one (issue 1126); nil asks none.
|
||||
ReadOpen func(ctx context.Context) ([]delivery.OpenPull, error)
|
||||
// Say says one of the indexer's events (placement-changed); nil says none.
|
||||
Say func(event string, body any) error
|
||||
|
||||
mu sync.Mutex // guards source, toolchains and previews, which the watches and the forge messages all reach
|
||||
source sourceNow
|
||||
@@ -134,14 +124,6 @@ func (ix *Indexer) ApplyMessage(ctx context.Context, f delivery.Forge) (Applied,
|
||||
}
|
||||
ix.kickRetry(ctx) // a registered module the forge could not read is read again on a merge, in the background (issue 1945)
|
||||
}
|
||||
if applied.Late == "" { // a late merged message would write an older placement back
|
||||
if err := ix.applyPlacement(ctx, f); err != nil {
|
||||
return applied, err
|
||||
}
|
||||
}
|
||||
if err := ix.applyPlacementCandidates(ctx, f); err != nil {
|
||||
return applied, err
|
||||
}
|
||||
var digests map[string]string
|
||||
var held map[string]delivery.Module
|
||||
if f.Kind == delivery.ForgePull {
|
||||
@@ -196,14 +178,6 @@ func (ix *Indexer) applyOpen(ctx context.Context, f delivery.Forge) error {
|
||||
for _, p := range f.Open {
|
||||
named[p] = true
|
||||
}
|
||||
if ix.PlacementCandidates != nil {
|
||||
// A pull request the forge no longer has open keeps no placement candidate (issue 1126).
|
||||
if err := ix.dropPlacementCandidates(ctx, func(_ string, c delivery.PlacementCandidate) bool {
|
||||
return !named[delivery.OpenPull{Repository: c.Pull.Repository, Number: c.Pull.Number}]
|
||||
}, "its pull request is not open"); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
keys, err := ix.Store.Keys(ctx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("listing the candidates: %w", err)
|
||||
|
||||
@@ -43,8 +43,6 @@ func withManifests(f delivery.Forge, manifests map[string]string, locks map[stri
|
||||
}
|
||||
|
||||
func TestAMergedManifestIsKeptByItsDigestAndNamedByTheModule(t *testing.T) { // 1449/AC1
|
||||
// A merge's placement files are written as its manifests are (issue 1126): folded here to hold the count.
|
||||
placementIsWrittenAtAMergeOnlyWhereItChanged(t)
|
||||
avahi := catalogueManifest(t, "avahi")
|
||||
ix, mods, manifests := manifestsIndexer(t, rec(t, "build-1", "avahi", catalogueRepo, "modules/avahi", "main", "", "2026-10-19T09:00:00Z", aInputs))
|
||||
f := withManifests(mergedMsg(catalogueRepo, map[string]string{"modules/avahi": "newtree"}), map[string]string{"modules/avahi": avahi}, nil)
|
||||
|
||||
@@ -21,8 +21,8 @@ import (
|
||||
// that cannot say for another reason (a timeout) writes nothing and is an error: the server delivers the message again.
|
||||
//
|
||||
// - A merged message asks once, before it writes anything: is its merge older than the repository's taken merge (the
|
||||
// last merge applied whole, taken.go)? Then the whole message is late: no module is written or deleted, no placement
|
||||
// is written, the taken key stays, and the log says so.
|
||||
// last merge applied whole, taken.go)? Then the whole message is late: no module is written or deleted, the taken key
|
||||
// stays, and the log says so.
|
||||
// - A take-in of a held module asks once for that module: is the source's head older than the module's record
|
||||
// (takeInHeld)?
|
||||
|
||||
@@ -92,7 +92,7 @@ func (ix *Indexer) lateMerge(ctx context.Context, w *mergeWrite) error {
|
||||
w.asked = cur.Merge
|
||||
if old {
|
||||
w.late = cur.Merge
|
||||
ix.logf("%s at %.12s: the merged message is older than taken.%s at %.12s, so no module is written or deleted and no placement written (a late or repeated message; a merge never moves a module backwards, issue 1126)",
|
||||
ix.logf("%s at %.12s: the merged message is older than taken.%s at %.12s, so no module is written or deleted (a late or repeated message; a merge never moves a module backwards, issue 1126)",
|
||||
w.repository, w.merge, key, cur.Merge)
|
||||
}
|
||||
return nil
|
||||
|
||||
@@ -1,123 +0,0 @@
|
||||
package index
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"maps"
|
||||
"slices"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
)
|
||||
|
||||
// Placement of modules on nodes (novox/hq ADR 0549 §2, issue 1126): at each merge of the catalogue the forge bridge
|
||||
// carries every placement file of the merge commit, and the indexer writes each node's key `placement.<node>` in its
|
||||
// state `placement` only when it differs (the revision alone is no change), deletes the key of a node whose file is
|
||||
// gone, and says `placement-changed` for each node it wrote or deleted, naming the modules added, removed or changed.
|
||||
// A merge that changes no placement writes nothing and says nothing.
|
||||
//
|
||||
// A file that does not read as a placement, or names another node than its file, is said in the log and its key is left
|
||||
// as it was: the catalogue's own check refuses such a file before it merges (ADR 0549 §1), so it is a fault of that
|
||||
// check, and a node keeps composing from the placement it had rather than from none.
|
||||
|
||||
// applyPlacement takes the placement files of a merged message in; a message that read none (another repository, an
|
||||
// older bridge) changes nothing.
|
||||
func (ix *Indexer) applyPlacement(ctx context.Context, f delivery.Forge) error {
|
||||
if ix.Placement == nil || f.Kind != delivery.ForgeMerged || !f.PlacementsRead {
|
||||
return nil
|
||||
}
|
||||
keys, err := ix.Placement.Keys(ctx)
|
||||
if errors.Is(err, process.ErrNotYet) {
|
||||
// The state is not on the bus yet: said, and the merge's candidates are still taken out (issue 1790's binding
|
||||
// writes the next catalogue merge's placement once it appears).
|
||||
ix.logf("%s at %s: its placement files are not written: %v", f.Repository, f.Merge, err)
|
||||
return nil
|
||||
}
|
||||
if err != nil {
|
||||
return fmt.Errorf("reading the placement keys: %w", err)
|
||||
}
|
||||
gone := map[string]bool{}
|
||||
for _, key := range keys {
|
||||
if node, err := delivery.ParsePlacementKey(key); err == nil {
|
||||
gone[node] = true
|
||||
}
|
||||
}
|
||||
var refused []string
|
||||
for _, node := range slices.Sorted(maps.Keys(f.Placements)) {
|
||||
delete(gone, node)
|
||||
next, err := delivery.DecodePlacement([]byte(f.Placements[node]))
|
||||
if err == nil && next.Node != node {
|
||||
err = fmt.Errorf("the file of %s names the node %q", node, next.Node)
|
||||
}
|
||||
if err != nil {
|
||||
refused = append(refused, node)
|
||||
ix.logf("%s at %s: the placement file of %s is refused and placement.%s is left as it was (the catalogue's check should have refused it): %v",
|
||||
f.Repository, f.Merge, node, node, err)
|
||||
continue
|
||||
}
|
||||
next.Revision = f.Merge
|
||||
if err := ix.writePlacement(ctx, node, &next); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
for _, node := range slices.Sorted(maps.Keys(gone)) {
|
||||
if err := ix.writePlacement(ctx, node, nil); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if len(refused) > 0 {
|
||||
// Said, and the merge's candidates are still taken out: a refused file holds nobody's delivery.
|
||||
ix.logf("%s at %s: the placement files of %v do not read; their keys are left as they were", f.Repository, f.Merge, refused)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// writePlacement writes one node's key, or deletes it for next nil, only when it differs, and says what changed.
|
||||
func (ix *Indexer) writePlacement(ctx context.Context, node string, next *delivery.Placement) error {
|
||||
key, err := delivery.PlacementKey(node)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
var change *delivery.PlacementChange
|
||||
err = write(ctx, ix.Placement, key, func(value []byte, found bool) ([]byte, bool, error) {
|
||||
change = nil
|
||||
var before delivery.Placement
|
||||
if found {
|
||||
cur, err := delivery.DecodePlacement(value)
|
||||
if err != nil {
|
||||
ix.logf("placement.%s holds what is not a placement, written again: %v", node, err)
|
||||
} else {
|
||||
before = cur
|
||||
}
|
||||
}
|
||||
if next == nil {
|
||||
if !found {
|
||||
return nil, false, nil
|
||||
}
|
||||
c := delivery.ChangeOf(before, delivery.Placement{Node: node})
|
||||
change = &c
|
||||
return nil, true, nil
|
||||
}
|
||||
if found && delivery.SamePlacement(before, *next) {
|
||||
return nil, false, nil
|
||||
}
|
||||
data, err := delivery.EncodePlacement(*next)
|
||||
if err != nil {
|
||||
return nil, false, err
|
||||
}
|
||||
c := delivery.ChangeOf(before, *next)
|
||||
change = &c
|
||||
return data, false, nil
|
||||
})
|
||||
if err != nil || change == nil {
|
||||
return err
|
||||
}
|
||||
if ix.Say != nil {
|
||||
if err := ix.Say(delivery.PlacementChangedEvent, *change); err != nil {
|
||||
// The key is written: a reader that watches it hears the change; the event is said loudly as not said.
|
||||
ix.logf("placement.%s is written, and %s could not be said: %v", node, delivery.PlacementChangedEvent, err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -1,352 +0,0 @@
|
||||
package index
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"maps"
|
||||
"slices"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
)
|
||||
|
||||
// Placement candidates (issue 1126; novox/hq ADR 0549 §4: a placement change passes the composition check before any
|
||||
// node sees it). An updated pull request of the catalogue that touches placement/ carries every placement file at its
|
||||
// head (the forge bridge, PlacementsRead). For each node whose file at the head differs from the trunk's key
|
||||
// `placement.<node>` (SamePlacement, the revision left out), or whose file the head removes, the indexer writes the key
|
||||
// `<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, 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
|
||||
// candidate that is not one.
|
||||
|
||||
// applyPlacementCandidates takes one forge message into the placement candidates.
|
||||
func (ix *Indexer) applyPlacementCandidates(ctx context.Context, f delivery.Forge) error {
|
||||
if ix.PlacementCandidates == nil || !delivery.SameRepository(f.Repository, delivery.CatalogueRepository) || trunkLike[f.Branch] {
|
||||
return nil
|
||||
}
|
||||
if f.Kind == delivery.ForgeMerged || f.Action == delivery.PullClosed {
|
||||
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 %s", f.Repository, f.Number, map[bool]string{true: "merged", false: "closed"}[f.Kind == delivery.ForgeMerged]))
|
||||
}
|
||||
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
|
||||
}
|
||||
ix.dropCandidatesOfClosedPulls(ctx, f)
|
||||
trunk, err := ix.trunkPlacements(ctx)
|
||||
if err != nil {
|
||||
ix.logf("%s#%d at %s: its placement candidates are not written: %v", f.Repository, f.Number, f.Head, err)
|
||||
return nil
|
||||
}
|
||||
nodes := map[string]bool{}
|
||||
for node := range f.Placements {
|
||||
nodes[node] = true
|
||||
}
|
||||
for node := range trunk {
|
||||
nodes[node] = true
|
||||
}
|
||||
pull := delivery.Head{Repository: f.Repository, Number: f.Number, Head: f.Head}
|
||||
for _, node := range slices.Sorted(maps.Keys(nodes)) {
|
||||
key, err := delivery.PlacementCandidateKey(f.Branch, node)
|
||||
if err != nil {
|
||||
return fmt.Errorf("%w: %s#%d: %v", ErrUnreadable, f.Repository, f.Number, err)
|
||||
}
|
||||
var next *delivery.PlacementCandidate // nil: no key for this node
|
||||
raw, atHead := f.Placements[node]
|
||||
before, inTrunk := trunk[node]
|
||||
switch {
|
||||
case atHead:
|
||||
p, err := delivery.DecodePlacement([]byte(raw))
|
||||
if err == nil && p.Node != node {
|
||||
err = fmt.Errorf("the file of %s names the node %q", node, p.Node)
|
||||
}
|
||||
if err != nil {
|
||||
ix.logf("%s#%d at %s: the placement file of %s is refused, so %s has no placement candidate (the catalogue's check refuses the pull request): %v",
|
||||
f.Repository, f.Number, f.Head, node, node, err)
|
||||
break
|
||||
}
|
||||
if inTrunk && delivery.SamePlacement(before, p) {
|
||||
break
|
||||
}
|
||||
next = &delivery.PlacementCandidate{Group: f.Branch, Pull: pull, Placement: &p}
|
||||
case inTrunk:
|
||||
next = &delivery.PlacementCandidate{Group: f.Branch, Pull: pull} // the head removes the node's file
|
||||
}
|
||||
if err := ix.writePlacementCandidate(ctx, key, next); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
// A node the head and the trunk both no longer name keeps no key of this group.
|
||||
return ix.dropPlacementCandidates(ctx, func(node string, c delivery.PlacementCandidate) bool {
|
||||
return c.Group == f.Branch && delivery.SameRepository(c.Pull.Repository, f.Repository) && c.Pull.Number == f.Number && !nodes[node]
|
||||
}, fmt.Sprintf("%s#%d at %s", f.Repository, f.Number, f.Head))
|
||||
}
|
||||
|
||||
// dropCandidatesOfClosedPulls heals a closed or merged message the indexer never heard (issue 1126: closed probe 532 kept
|
||||
// its candidates, and ace answered them for 22 minutes before the open 534). When a catalogue pull message arrives and a
|
||||
// candidate of another pull request exists, the forge is asked which pull requests are open, and a candidate of one that is
|
||||
// not is deleted. A forge that cannot say leaves them as they are and is said in the log; indexer_placement_reread is the
|
||||
// same by hand.
|
||||
func (ix *Indexer) dropCandidatesOfClosedPulls(ctx context.Context, f delivery.Forge) {
|
||||
if ix.ReadOpen == nil {
|
||||
return
|
||||
}
|
||||
keys, err := ix.PlacementCandidates.Keys(ctx)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
other := false
|
||||
for _, key := range keys {
|
||||
value, _, found, err := ix.PlacementCandidates.Get(ctx, key)
|
||||
if err != nil || !found {
|
||||
continue
|
||||
}
|
||||
if c, err := delivery.DecodePlacementCandidate(value); err != nil || c.Pull.Number != f.Number || !delivery.SameRepository(c.Pull.Repository, f.Repository) {
|
||||
other = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !other {
|
||||
return
|
||||
}
|
||||
open, err := ix.ReadOpen(ctx)
|
||||
if err != nil {
|
||||
ix.logf("%s#%d: the open pull requests could not be read, so no stale placement candidate is deleted: %v", f.Repository, f.Number, err)
|
||||
return
|
||||
}
|
||||
isOpen := map[delivery.OpenPull]bool{}
|
||||
for _, p := range open {
|
||||
isOpen[delivery.OpenPull{Repository: p.Repository, Number: p.Number}] = true
|
||||
}
|
||||
isOpen[delivery.OpenPull{Repository: f.Repository, Number: f.Number}] = true // the message in hand is an open one
|
||||
if err := ix.dropPlacementCandidates(ctx, func(_ string, c delivery.PlacementCandidate) bool {
|
||||
for p := range isOpen {
|
||||
if p.Number == c.Pull.Number && delivery.SameRepository(p.Repository, c.Pull.Repository) {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}, "its pull request is not open (the forge says so at the next pull message)"); err != nil {
|
||||
ix.logf("%s#%d: stale placement candidates are not deleted: %v", f.Repository, f.Number, err)
|
||||
}
|
||||
}
|
||||
|
||||
// trunkPlacements is every node's placement in the trunk's state placement, by node.
|
||||
func (ix *Indexer) trunkPlacements(ctx context.Context) (map[string]delivery.Placement, error) {
|
||||
out := map[string]delivery.Placement{}
|
||||
if ix.Placement == nil {
|
||||
return nil, errors.New("the indexer reads no placement state")
|
||||
}
|
||||
keys, err := ix.Placement.Keys(ctx)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("reading the placement keys: %w", err)
|
||||
}
|
||||
for _, key := range keys {
|
||||
node, err := delivery.ParsePlacementKey(key)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
value, _, found, err := ix.Placement.Get(ctx, key)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("reading %s: %w", key, err)
|
||||
}
|
||||
if !found {
|
||||
continue
|
||||
}
|
||||
p, err := delivery.DecodePlacement(value)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("%s holds what is not a placement: %w", key, err)
|
||||
}
|
||||
out[node] = p
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// writePlacementCandidate writes one candidate key, or deletes it for next nil, only when it changes.
|
||||
func (ix *Indexer) writePlacementCandidate(ctx context.Context, key string, next *delivery.PlacementCandidate) error {
|
||||
var data []byte
|
||||
if next != nil {
|
||||
var err error
|
||||
if data, err = delivery.EncodePlacementCandidate(*next); err != nil {
|
||||
return fmt.Errorf("%w: %s.%s: %v", ErrUnreadable, delivery.PlacementCandidatesState, key, err)
|
||||
}
|
||||
}
|
||||
err := write(ctx, ix.PlacementCandidates, key, func(value []byte, found bool) ([]byte, bool, error) {
|
||||
switch {
|
||||
case next == nil:
|
||||
return nil, found, nil
|
||||
case found && string(value) == string(data):
|
||||
return nil, false, nil
|
||||
}
|
||||
return data, false, nil
|
||||
})
|
||||
if errors.Is(err, process.ErrNotYet) {
|
||||
ix.logf("%s.%s is not written: %v", delivery.PlacementCandidatesState, key, err)
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
// dropPlacementCandidates deletes every candidate key whose node and value gone says, and says each with why.
|
||||
func (ix *Indexer) dropPlacementCandidates(ctx context.Context, gone func(node string, c delivery.PlacementCandidate) bool, why string) error {
|
||||
keys, err := ix.PlacementCandidates.Keys(ctx)
|
||||
if errors.Is(err, process.ErrNotYet) {
|
||||
return nil
|
||||
}
|
||||
if err != nil {
|
||||
return fmt.Errorf("reading the placement candidate keys: %w", err)
|
||||
}
|
||||
for _, key := range keys {
|
||||
_, node, err := delivery.ParsePlacementCandidateKey(key)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
dropped := false
|
||||
err = write(ctx, ix.PlacementCandidates, key, func(value []byte, found bool) ([]byte, bool, error) {
|
||||
if !found {
|
||||
return nil, false, nil
|
||||
}
|
||||
c, err := delivery.DecodePlacementCandidate(value)
|
||||
if err != nil {
|
||||
ix.logf("%s.%s holds what is not a placement candidate, deleted: %v", delivery.PlacementCandidatesState, key, err)
|
||||
return nil, true, nil
|
||||
}
|
||||
dropped = gone(node, c)
|
||||
return nil, dropped, nil
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if dropped {
|
||||
ix.logf("%s.%s is deleted: %s", delivery.PlacementCandidatesState, key, why)
|
||||
}
|
||||
}
|
||||
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
|
||||
}
|
||||
@@ -1,219 +0,0 @@
|
||||
package index
|
||||
|
||||
import (
|
||||
"context"
|
||||
"reflect"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
// ADR 0549 §2 (issue 1126): a merge changing one node's file writes that key alone and says one event; a merge with no
|
||||
// placement change writes nothing and says nothing; a node whose file is gone loses its key and says its modules
|
||||
// removed; a file that does not read leaves its key; a message that read no placement changes none.
|
||||
func placementIsWrittenAtAMergeOnlyWhereItChanged(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
ix, _, logs := newIndexer(t)
|
||||
place := newMem()
|
||||
ix.Placement = place
|
||||
var said []delivery.PlacementChange
|
||||
ix.Say = func(event string, body any) error {
|
||||
if event != delivery.PlacementChangedEvent {
|
||||
t.Errorf("said %s", event)
|
||||
}
|
||||
said = append(said, body.(delivery.PlacementChange))
|
||||
return nil
|
||||
}
|
||||
merged := func(merge string, files map[string]string) delivery.Forge {
|
||||
return delivery.Forge{V: delivery.Version, Kind: delivery.ForgeMerged, Repository: delivery.CatalogueRepository, Branch: "main",
|
||||
Number: 7, Head: "h", Merge: merge, Root: "r", Trees: map[string]string{"": "t"}, Placements: files, PlacementsRead: true}
|
||||
}
|
||||
ace := `{"v":1,"node":"ace","modules":[{"module":"plex"}]}`
|
||||
g14 := `{"v":1,"node":"g14","modules":[{"module":"zsh"}]}`
|
||||
written := func(node string) (delivery.Placement, uint64) {
|
||||
t.Helper()
|
||||
v, ok := place.vals["placement."+node]
|
||||
if !ok {
|
||||
return delivery.Placement{}, 0
|
||||
}
|
||||
p, err := delivery.DecodePlacement(v)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return p, place.revs["placement."+node]
|
||||
}
|
||||
|
||||
if err := ix.Apply(ctx, merged("m1", map[string]string{"ace": ace, "g14": g14})); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if p, _ := written("ace"); p.Revision != "m1" || !reflect.DeepEqual(p.Assigned(), []string{"plex"}) || len(said) != 2 {
|
||||
t.Fatalf("first merge: %+v, said %v", p, said)
|
||||
}
|
||||
_, aceRev := written("ace")
|
||||
|
||||
said = nil
|
||||
if err := ix.Apply(ctx, merged("m2", map[string]string{"ace": ace, "g14": g14})); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if p, rev := written("ace"); rev != aceRev || p.Revision != "m1" || len(said) != 0 {
|
||||
t.Errorf("a merge with no placement change wrote %+v (revision %d, was %d) and said %v", p, rev, aceRev, said)
|
||||
}
|
||||
|
||||
said = nil
|
||||
if err := ix.Apply(ctx, merged("m3", map[string]string{"ace": ace, "g14": `{"v":1,"node":"g14","modules":[{"module":"zsh"},{"module":"xterm"}]}`})); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, rev := written("ace"); rev != aceRev || len(said) != 1 || said[0].Node != "g14" || said[0].Revision != "m3" ||
|
||||
!reflect.DeepEqual(said[0].Added, []string{"xterm"}) {
|
||||
t.Errorf("one file changed: ace's revision %d (was %d), said %+v", rev, aceRev, said)
|
||||
}
|
||||
|
||||
said = nil
|
||||
if err := ix.Apply(ctx, merged("m4", map[string]string{"ace": ace, "g14": `{"v":1,"node":"ace","modules":[]}`})); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if p, _ := written("g14"); len(p.Modules) != 2 || len(said) != 0 || !strings.Contains(logs.String(), "the placement file of %s is refused") {
|
||||
t.Errorf("a file naming another node replaced the key: %+v, said %v, logs %s", p, said, logs)
|
||||
}
|
||||
|
||||
said = nil
|
||||
if err := ix.Apply(ctx, merged("m5", map[string]string{"ace": ace})); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, rev := written("g14"); rev != 0 || len(said) != 1 || !reflect.DeepEqual(said[0].Removed, []string{"xterm", "zsh"}) {
|
||||
t.Errorf("a file removed: g14's key at revision %d, said %+v", rev, said)
|
||||
}
|
||||
|
||||
said = nil
|
||||
unread := merged("m6", nil)
|
||||
unread.PlacementsRead = false
|
||||
if err := ix.Apply(ctx, unread); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if p, _ := written("ace"); p.Revision != "m1" || len(said) != 0 {
|
||||
t.Errorf("a message that read no placement changed it: %+v, %v", p, said)
|
||||
}
|
||||
|
||||
// Issue 1126 (ADR 0549 §4), placement candidates: the trunk now places ace alone. A catalogue pull request writes
|
||||
// `<group>.<node>` for each node whose file at its head differs from the trunk's, or that the head removes, and only
|
||||
// when the value changes; a pull message that read no placement, or a pull request changing none, writes nothing; a
|
||||
// file that does not read gets no key; a merge, a close or an open message not naming it deletes its keys.
|
||||
cands := newMem()
|
||||
ix.PlacementCandidates = cands
|
||||
said = nil
|
||||
pullOf := func(branch string, number int, head string, files map[string]string) delivery.Forge {
|
||||
return delivery.Forge{V: delivery.Version, Kind: delivery.ForgePull, Repository: delivery.CatalogueRepository, Branch: branch,
|
||||
Number: number, Head: head, Action: delivery.PullUpdated, Placements: files, PlacementsRead: files != nil}
|
||||
}
|
||||
candidate := func(key string) (delivery.PlacementCandidate, uint64, bool) {
|
||||
t.Helper()
|
||||
v, ok := cands.vals[key]
|
||||
if !ok {
|
||||
return delivery.PlacementCandidate{}, 0, false
|
||||
}
|
||||
c, err := delivery.DecodePlacementCandidate(v)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return c, cands.revs[key], true
|
||||
}
|
||||
keys := func() string { k, _ := cands.Keys(ctx); slices.Sort(k); return strings.Join(k, ",") }
|
||||
g14New := `{"v":1,"node":"g14","modules":[{"module":"zsh"}]}`
|
||||
g14P := `{"v":1,"node":"g14","modules":[{"module":"htop"}]}`
|
||||
g14Q := `{"v":1,"node":"g14","modules":[{"module":"mc"}]}`
|
||||
for _, row := range []struct {
|
||||
name string
|
||||
f delivery.Forge
|
||||
keys string
|
||||
open []int // the pull requests the forge says are open; nil: the indexer asks none
|
||||
}{
|
||||
{"a pull request adding g14", pullOf("feat/p", 9, h1, map[string]string{"ace": ace, "g14": g14New}), "feat/p.g14", nil},
|
||||
{"the same head again", pullOf("feat/p", 9, h1, map[string]string{"ace": ace, "g14": g14New}), "feat/p.g14", nil},
|
||||
{"a pull message that read no placement", pullOf("feat/p", 9, h2, nil), "feat/p.g14", nil},
|
||||
{"a pull request changing no placement", pullOf("feat/q", 10, h1, map[string]string{"ace": ace}), "feat/p.g14", nil},
|
||||
{"a new head removing ace", pullOf("feat/p", 9, h2, map[string]string{"g14": g14New}), "feat/p.ace,feat/p.g14", nil},
|
||||
{"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", nil},
|
||||
{"the head adding g14 again", pullOf("feat/p", 9, h2, map[string]string{"g14": g14New}), "feat/p.ace,feat/p.g14", nil},
|
||||
{"a head the forge says changes no placement file", func() delivery.Forge { f := pullOf("feat/p", 9, h3, nil); f.DirsKnown = true; return f }(), "", nil},
|
||||
{"the head adding g14 and removing ace again", pullOf("feat/p", 9, h2, map[string]string{"g14": g14New}), "feat/p.ace,feat/p.g14", nil},
|
||||
{"an open message not naming it", delivery.Forge{V: delivery.Version, Kind: delivery.ForgeOpen, Open: []delivery.OpenPull{{Repository: delivery.CatalogueRepository, Number: 10}}}, "", nil},
|
||||
{"the pull request heard again", pullOf("feat/p", 9, h2, map[string]string{"g14": g14New}), "feat/p.ace,feat/p.g14", nil},
|
||||
{"its merge", func() delivery.Forge {
|
||||
m := merged("m7", map[string]string{"g14": g14New})
|
||||
m.Branch, m.Number = "feat/p", 9
|
||||
return m
|
||||
}(), "", nil},
|
||||
// A closed message the indexer never heard heals at the next pull message (issue 1126, closed probe 532).
|
||||
{"pull 9 changes g14, both open", pullOf("feat/p", 9, h2, map[string]string{"g14": g14P}), "feat/p.g14", []int{9, 10}},
|
||||
{"pull 10 changes g14, both open", pullOf("feat/q", 10, h1, map[string]string{"g14": g14Q}), "feat/p.g14,feat/q.g14", []int{9, 10}},
|
||||
{"pull 9 closed unheard, pull 10 heard again", pullOf("feat/q", 10, h2, map[string]string{"g14": g14Q}), "feat/q.g14", []int{10}},
|
||||
{"pull 10 closed", func() delivery.Forge { f := pullOf("feat/q", 10, h2, nil); f.Action = delivery.PullClosed; return f }(), "", []int{10}},
|
||||
} {
|
||||
ix.ReadOpen = nil
|
||||
if row.open != nil {
|
||||
ix.ReadOpen = func(context.Context) ([]delivery.OpenPull, error) {
|
||||
var out []delivery.OpenPull
|
||||
for _, n := range row.open {
|
||||
out = append(out, delivery.OpenPull{Repository: delivery.CatalogueRepository, Number: n})
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
}
|
||||
_, revBefore, _ := candidate("feat/p.g14")
|
||||
if err := ix.Apply(ctx, row.f); err != nil {
|
||||
t.Fatalf("%s: %v", row.name, err)
|
||||
}
|
||||
if got := keys(); got != row.keys {
|
||||
t.Errorf("%s: keys %q, want %q", row.name, got, row.keys)
|
||||
}
|
||||
c, rev, ok := candidate("feat/p.g14")
|
||||
switch row.name {
|
||||
case "a pull request adding g14":
|
||||
if !ok || c.Group != "feat/p" || c.Pull != (delivery.Head{Repository: delivery.CatalogueRepository, Number: 9, Head: h1}) || c.Placement == nil ||
|
||||
!reflect.DeepEqual(c.Placement.Assigned(), []string{"zsh"}) {
|
||||
t.Errorf("%s: %+v", row.name, c)
|
||||
}
|
||||
case "the same head again", "a pull message that read no placement", "a pull request changing no placement":
|
||||
if rev != revBefore {
|
||||
t.Errorf("%s: feat/p.g14 written again (revision %d, was %d)", row.name, rev, revBefore)
|
||||
}
|
||||
case "a new head removing ace":
|
||||
if a, _, _ := candidate("feat/p.ace"); a.Placement != nil || a.Pull.Head != h2 || c.Pull.Head != h2 {
|
||||
t.Errorf("%s: ace %+v, g14 at %s", row.name, a, c.Pull.Head)
|
||||
}
|
||||
}
|
||||
}
|
||||
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,10 +17,6 @@ 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
|
||||
@@ -36,19 +32,10 @@ type TakeInAsk struct {
|
||||
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
|
||||
}
|
||||
func (a TakeInAsk) Done() bool { return !a.Answered.IsZero() && a.Result != nil }
|
||||
|
||||
// DecodeTakeIn reads a takeins value.
|
||||
func DecodeTakeIn(data []byte) (TakeInAsk, error) {
|
||||
@@ -59,7 +46,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 == "" && !a.Placement {
|
||||
if a.Module == "" {
|
||||
return a, errors.New("a takeins value naming no module")
|
||||
}
|
||||
return a, nil
|
||||
|
||||
@@ -81,12 +81,12 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
|
||||
if m.Module != "mesh-delivery-indexer" {
|
||||
t.Errorf("module = %q", m.Module)
|
||||
}
|
||||
// 1126 (ADR 0549 §2): the indexer's placement state, and the event it says when a node's placement changed.
|
||||
if want := []string{"placement-changed"}; !reflect.DeepEqual(m.Emits, want) {
|
||||
t.Errorf("emits = %v, want %v", m.Emits, want)
|
||||
// 1126: placement is the seat mesh-placement's (ADR 0567): the indexer holds no placement state and says no event.
|
||||
if len(m.Emits) != 0 {
|
||||
t.Errorf("emits = %v, want none", m.Emits)
|
||||
}
|
||||
// 1126 note 76: heard keeps the last 64 FORGE messages the indexer handled, for indexer_heard.
|
||||
if want := []state{{Name: "candidates", History: 10}, {Name: "heard", History: 64}, {Name: "manifests", History: 1}, {Name: "modules", History: 20}, {Name: "placement", History: 10}, {Name: "placement-candidates", History: 5}, {Name: "taken", History: 1}, {Name: "takeins", History: 1}}; !reflect.DeepEqual(m.State, want) {
|
||||
if want := []state{{Name: "candidates", History: 10}, {Name: "heard", History: 64}, {Name: "manifests", History: 1}, {Name: "modules", History: 20}, {Name: "taken", History: 1}, {Name: "takeins", 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-controller.assignments", "mesh-controller.builds", "mesh-delivery-checker.previews", "mesh-delivery-gate.source"}; !reflect.DeepEqual(sorted(m.Reads), want) {
|
||||
@@ -106,15 +106,14 @@ 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).
|
||||
// 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) {
|
||||
if want := []string{"gitea.gitea_commit_trees", "gitea.gitea_contains", "gitea.gitea_get_file", "gitea.gitea_open_pulls"}; !reflect.DeepEqual(m.Invokes, want) {
|
||||
t.Errorf("invokes = %v, want %v", m.Invokes, 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) {
|
||||
// 1945: 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) {
|
||||
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.
|
||||
// No streams, no bus, no seat (ADR 0437 §4); it says no event.
|
||||
for _, k := range []string{"streams", "logs", "claims", "provides", "requires", "uses", "defines-seats", "seat", "seats", "bus"} {
|
||||
if _, ok := top[k]; ok {
|
||||
t.Errorf("the manifest declares %q: the indexer has none", k)
|
||||
|
||||
@@ -3,11 +3,7 @@
|
||||
"version": "1",
|
||||
"slug": "indexer",
|
||||
"tools": [
|
||||
"indexer_take_in",
|
||||
"indexer_placement_reread"
|
||||
],
|
||||
"emits": [
|
||||
"placement-changed"
|
||||
"indexer_take_in"
|
||||
],
|
||||
"consumes": [
|
||||
"mesh-build-go.toolchain.reported",
|
||||
@@ -20,8 +16,6 @@
|
||||
{ "name": "heard", "history": 64 },
|
||||
{ "name": "manifests", "history": 1 },
|
||||
{ "name": "modules", "history": 20 },
|
||||
{ "name": "placement", "history": 10 },
|
||||
{ "name": "placement-candidates", "history": 5 },
|
||||
{ "name": "taken", "history": 1 },
|
||||
{ "name": "takeins", "history": 1 }
|
||||
],
|
||||
@@ -37,7 +31,6 @@
|
||||
"gitea.gitea_commit_trees",
|
||||
"gitea.gitea_contains",
|
||||
"gitea.gitea_get_file",
|
||||
"gitea.gitea_list_branches",
|
||||
"gitea.gitea_open_pulls"
|
||||
],
|
||||
"own-secrets": {
|
||||
|
||||
Reference in New Issue
Block a user