Merge feat/47-G-fix11-b5a into feat/47-G-fix11-b5d

# Conflicts:
#	modules/mesh-delivery-indexer/internal/index/pending_test.go
#	modules/mesh-delivery-indexer/internal/index/takein.go
#	modules/mesh-delivery-indexer/manifest_test.go
#	modules/mesh-delivery-indexer/module.json
This commit is contained in:
2026-11-12 05:55:24 +01:00
22 changed files with 652 additions and 1008 deletions
@@ -62,10 +62,11 @@ func ParseCandidatesKey(key string) (string, error) {
return group, nil
}
var tryGroupOK = regexp.MustCompile(`^try-[0-9]{1,20}$`)
var tryGroupOK = regexp.MustCompile(`^(try-[0-9]{1,20}|run-[A-Za-z0-9_-]{1,64})$`)
// IsTryGroup says a candidates key is the group of a whole try, try-<digits>, which is also the try's id: the indexer makes the
// candidate of a try's group from its previews, and the checker judges it by its try, never as a pull request's check.
// IsTryGroup says a candidates key is the group of a whole try, try-<digits>, which is also the try's id, or of a run on
// request, run-<id> (design 47 G A-160): the indexer makes the candidate of a run from the runner's run-started, and the
// checker judges neither as a pull request's check.
func IsTryGroup(key string) bool { return tryGroupOK.MatchString(key) }
// NoneNoPreview is why a candidate's touched module has no fingerprint of the new form yet: its merge preview is not
@@ -315,7 +315,6 @@ func main() {
_, err := acts.Record(meshhandactsclient.Request{What: what, Why: why, Cause: index.FingerprintsAgainTool, Verb: "fingerprints-again"})
return err
}, 10*time.Second)...)
tools = append(tools, index.TryCandidateTools(keptRead{stdio.State("candidates")}, keptRead{stdio.State(index.TryRecoveryState)}, asks, time.Now)...)
tools = append(tools, configReadersTool(stdio.NewWatchedStates(), trunk.State("modules")), busStateTool(trunk.State), requirementsTool(keptSource),
packageUsersTool(trunk.State), seatDefinitionTool(keptSource), outOfStepTool(keptSource))
if err := stdio.Serve("", tools); err != nil {
@@ -33,9 +33,10 @@ import (
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshdeliveryclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/git"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdeliverychecker"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein"
@@ -1136,86 +1137,98 @@ func TestIndexerWritesAGoModuleAgainWhenTheBuildModuleReportsItsToolchain(t *tes
}
}
// TestIndexerGivesACandidateItsFingerprintWhenTheCheckerCreatesThePreviewsBucket: the indexer starts with the stream PREVIEWS
// and its durable consumer (made by the controller from the checker's declaration and the indexer's consumes) and no bucket
// mesh-delivery-checker_previews; a pull request of mesh-catalog touching avahi makes the candidate with the reason
// 'no merge preview is computed'; the checker then makes its bucket, writes the preview of that head onto the trunk head
// the source names, and says so on its stream; the candidate's avahi gets, within 20 s and with no restart, the reason
// computed from the preview (the toolchain is not reported here, so 'toolchain not yet reported').
func TestIndexerGivesACandidateItsFingerprintWhenTheCheckerCreatesThePreviewsBucket(t *testing.T) { // 1445/AC4
// TestIndexerBuildsARunsCandidateAndDropsItWhenEveryMembersPipelineEnded (design 47 G §64, §67 test 64, §190; A-160, A-174):
// the indexer starts with the runner's streams run-started and pipeline-ended and its durables on them, its state runs and a
// registration of avahi; the runner says run-started of run 9 with avahi at a commit on a branch no registration watches;
// within 20 s candidates.run-9 holds one pull of number 0 at that commit, avahi's manifest read through the seat git; the
// indexer is restarted, the runner says avahi's pipeline of run 9 ended, and the candidate and the run's key go.
func TestIndexerBuildsARunsCandidateAndDropsItWhenEveryMembersPipelineEnded(t *testing.T) {
manifest, err := os.ReadFile("../../internal/index/testdata/catalogue-a67f0196-avahi.module.json")
if err != nil {
t.Fatal(err)
}
const goSum = "3b5028f0c1e9bde04dd766abd63eb5931bc516bae695cf46f00abcd8510b0381"
const trunk, head = "5555555555555555555555555555555555555555", "0123456789abcdef0123456789abcdef01234567"
js, ctx, candidates, _ := startIndexer(t, func(ctx context.Context, builds buildLog) {
builds.end(t, "novox/mesh-catalog", "avahi", "modules/avahi")
}, func(ctx context.Context, js jetstream.JetStream) {
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: PreviewsStream, Subjects: []string{previewWrittenSubject}}); err != nil {
t.Fatal(err)
}
if _, err := js.CreateOrUpdateConsumer(ctx, PreviewsStream, jetstream.ConsumerConfig{Durable: "n1_mesh-delivery-indexer", AckPolicy: jetstream.AckExplicitPolicy}); err != nil {
t.Fatal(err)
}
seedBuildModules(t, ctx, js, nil)
// The seat mesh-delivery's verb source, as the gate serves it (design 47 G §115): main names the catalogue's trunk head.
answer := func(m *nats.Msg) {
raw, _ := json.Marshal(map[string]any{"result": meshdelivery.SourceAnswer{Branch: "main", Group: "g", Revision: 5,
Repositories: []meshdelivery.SourceHead{{Repository: "novox/mesh-catalog", Head: trunk, Trunk: "main"}}}, "node": "n1"})
_ = m.Respond(raw)
}
for _, subject := range []string{"mesh.seat.mesh-delivery.call.source.>", "mesh.seat.mesh-delivery.tool.source"} {
if _, err := js.Conn().Subscribe(subject, answer); err != nil {
const commit = "0123456789abcdef0123456789abcdef01234567"
var runs jetstream.KeyValue
js, ctx, candidates, _, restart := startIndexerRun(t, func(context.Context, buildLog) {}, func(ctx context.Context, js jetstream.JetStream) {
for stream, subject := range map[string]string{meshpipelinerunner.StreamRunStarted: meshpipelinerunner.SubjectRunStartedAll,
meshpipelinerunner.StreamPipelineEnded: meshpipelinerunner.SubjectPipelineEndedAll} {
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: stream, Subjects: []string{subject}}); err != nil {
t.Fatal(err)
}
if _, err := js.CreateOrUpdateConsumer(ctx, stream, jetstream.ConsumerConfig{Durable: "n1_mesh-delivery-indexer", AckPolicy: jetstream.AckExplicitPolicy,
FilterSubjects: []string{subject}}); err != nil {
t.Fatal(err)
}
}
var err error
if runs, err = js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.RunsBucket, History: 1}); err != nil {
t.Fatal(err)
}
regs, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.RegistrationsBucket, History: 1})
if err != nil {
t.Fatal(err)
}
reg, _ := json.Marshal(meshcatalog.Registration{V: 1, Module: "avahi", Origin: "novox/mesh-catalog", Path: "modules/avahi", Branches: []string{"main"}})
if _, err := regs.Create(ctx, index.RegistrationKey("avahi"), reg); err != nil {
t.Fatal(err)
}
seedBuildModules(t, ctx, js, nil)
// The seat git as the forge module serves it: avahi's directory and its manifest at the run's commit.
reply := func(m *nats.Msg, result any) {
raw, _ := json.Marshal(map[string]any{"result": result, "node": "n0"})
_ = m.Respond(raw)
}
if _, err := js.Conn().Subscribe("mesh.seat.git.tool.trees", func(m *nats.Msg) {
reply(m, git.TreesAnswer{Root: "r0", Commit: commit, Trees: map[string]string{"modules/avahi": "newtree"}})
}); err != nil {
t.Fatal(err)
}
if _, err := js.Conn().Subscribe("mesh.seat.git.tool.file", func(m *nats.Msg) {
var a git.FileInput
_ = json.Unmarshal(m.Data, &a)
switch a.Path {
case "modules/avahi/module.json":
reply(m, git.FileAnswer{Path: a.Path, Ref: a.Ref, Content: string(manifest)})
case "modules/avahi/go.sum":
reply(m, git.FileAnswer{Path: a.Path, Ref: a.Ref, Content: goSum})
default:
reply(m, git.FileAnswer{Path: a.Path, Ref: a.Ref, Absent: true})
}
}); err != nil {
t.Fatal(err)
}
})
pull, err := forgemsg.EncodeForge(forgemsg.Forge{Kind: forgemsg.ForgePull, Repository: "novox/mesh-catalog", Branch: "feat/x", Number: 301, Head: head, Action: forgemsg.PullUpdated,
Dirs: []string{"modules/avahi"}, DirsKnown: true, Manifests: map[string]string{"modules/avahi": string(manifest)}})
if err != nil {
started, _ := json.Marshal(meshpipelinerunner.RunStarted{V: 1, Run: "9", By: "lead@n1", Members: []meshpipelinerunner.Member{{Module: "avahi", Commit: commit, Branch: "feature/1966-x"}}})
if _, err := js.Publish(ctx, meshpipelinerunner.SubjectRunStarted("avahi", "feature-1966-x", commit), started); err != nil {
t.Fatal(err)
}
if _, err := js.Publish(ctx, "mesh.stream.mesh-delivery-forge-bridge.event.forge.novox.mesh-catalog.pull", pull); err != nil {
t.Fatal(err)
}
none2 := func(want string) string {
var got string
for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
if e, err := candidates.Get(ctx, "feat/x"); err == nil {
if c, err := index.DecodeCandidate(e.Value()); err == nil && len(c.Pulls) == 1 {
got = c.Pulls[0].Fps["avahi"].None2
if got == want {
break
}
}
key, _ := meshcatalogclient.CandidatesKey("run-9")
var c meshcatalog.CandidatesByGroupValue
for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
if e, err := candidates.Get(ctx, key); err == nil {
if c, err = index.DecodeCandidate(e.Value()); err == nil {
break
}
}
return got
}
if got := none2("no merge preview is computed"); got != "no merge preview is computed" {
t.Fatalf("none2 = %q within 20 s of the pull request, want 'no merge preview is computed'", got)
if len(c.Pulls) != 1 || c.Pulls[0].Number != 0 || c.Pulls[0].Head != commit || c.Pulls[0].Fps["avahi"].Manifest != meshcatalogclient.ManifestDigest(manifest) {
t.Fatalf("candidates.run-9 = %+v within 20 s of run-started, want avahi at the run's commit with its manifest", c)
}
// The checker's bucket is made only now, as when the checker is assigned after the indexer started.
kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.PreviewsBucket, History: index.PreviewsHistory})
if err != nil {
restart()
ended, _ := json.Marshal(meshpipelinerunner.PipelineEnded{V: 1, Pipeline: "feature-1966-x.0.avahi", Module: "avahi", Branch: "feature/1966-x", Commit: commit,
Outcome: "passed", Steps: []meshpipelinerunner.StepResult{}, Run: "9"})
if _, err := js.Publish(ctx, meshpipelinerunner.SubjectPipelineEnded("avahi", "feature-1966-x", commit), ended); err != nil {
t.Fatal(err)
}
pv := meshdeliverychecker.PreviewsByRepositoryHeadTrunkValue{Repository: "novox/mesh-catalog", Head: head, Trunk: trunk, Merge: "6666666666666666666666666666666666666666",
Modules: map[string]meshdeliverychecker.PreviewsByRepositoryHeadTrunkValueModulesValue{"modules/avahi": {Tree: "newtree", Manifest: meshcatalogclient.ManifestDigest(manifest), Locks: map[string]string{"go.sum": goSum}}}}
key, _ := index.PreviewKey(pv.Repository, pv.Head, pv.Trunk)
data, err := index.EncodePreview(pv)
if err != nil {
t.Fatal(err)
gone := false
for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline) && !gone; time.Sleep(50 * time.Millisecond) {
_, cerr := candidates.Get(ctx, key)
_, rerr := runs.Get(ctx, "run-9")
gone = errors.Is(cerr, jetstream.ErrKeyNotFound) && errors.Is(rerr, jetstream.ErrKeyNotFound)
}
if _, err := kv.Create(ctx, key, data); err != nil {
t.Fatal(err)
}
if _, err := js.Publish(ctx, previewWrittenSubject, []byte(`{"key":"`+key+`"}`)); err != nil {
t.Fatal(err)
}
if got := none2(sdkstep.ToolchainNoReport); got != sdkstep.ToolchainNoReport {
t.Errorf("none2 = %q within 20 s of the preview, want %q (computed from the preview)", got, sdkstep.ToolchainNoReport)
if !gone {
t.Error("the run's candidate or its runs key stays 20 s after its member's pipeline ended across a restart")
}
}
@@ -407,7 +407,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
// The manifests a take-in wrote and no judge pass judged yet (issue 1966): a state this delivery brings, bound when it
// appears; until then a take-in's new manifests are kept for their judging in its memory alone.
unjudged := process.AwaitBucket(ctx, js, conn, index.UnjudgedBucket, "a take-in's new manifests are kept for their judging in its memory alone", logger.Printf)
tryRecovery := process.AwaitBucket(ctx, js, conn, index.TryRecoveryBucket, "how the recovery of the open tries went is not kept", logger.Printf)
// The runs on request the indexer holds a candidate for (A-174): a state this delivery brings, bound when it appears.
runs := process.AwaitBucket(ctx, js, conn, index.RunsBucket, "no run on request's candidate is written or dropped", logger.Printf)
findings := process.AwaitBucket(ctx, js, conn, findingsBucket, "no finding of the indexer is written", logger.Printf)
// The indexer's findings are the seat mesh-conditions' generated finding, written through mesh-sdk's held finding writer
// (design 47 G §689, A-29): it holds what is written before the state is bound, and starts once the state is.
@@ -447,7 +448,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
Predefined: awaitedStore{b: predefined}, ConfigCandidates: awaitedStore{b: configCandidates}, Logf: logger.Printf,
Named: namedVia(calls.Call), Trunk: trunkVia(calls.Call), Contains: containsVia(calls.Call), Read: readVia(calls.Call),
ReadOpen: openVia(calls.Call), Packages: awaitedStore{b: packages, js: js, name: index.PackagesBucket}, Files: filesVia(calls.Call), GoSum: goSumVia(calls.Call), GoMod: goModVia(calls.Call),
OwnBuild: ownBuildVia(calls.Call, cred.Node), OpenTries: openTriesVia(calls.Call), OwnBuildState: awaitedStore{b: ownBuildState}, Heard: heardStore, Placement: placementVia(calls.CallKept), RunningWhole: runningWholeVia(js), TryRecovery: awaitedStore{b: tryRecovery}, Graph: graph,
OwnBuild: ownBuildVia(calls.Call, cred.Node), OwnBuildState: awaitedStore{b: ownBuildState}, Heard: heardStore, Placement: placementVia(calls.CallKept), RunningWhole: runningWholeVia(js), Runs: awaitedStore{b: runs}, Graph: graph,
SeatDefinitions: awaitedStore{b: seatDefs}, ReadFile: readFileVia(calls.Call), SaySeatDefinition: saySeats,
Revisions: revisions, EntryIndex: awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.EntryIndexBucket, "no entry is found by its fingerprint", logger.Printf)}, // Message a kept call: a read, called again through a transport fault (ADR 0668 §1)
// Each catalog revision may bring another holder of a build seat: the build modules are found again (buildmodules.go).
@@ -670,17 +671,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
buildModules.hearWith(func(stream, module string) {
hear(stream, module+"'s toolchain reports are not heard until the indexer starts again", func(m jetstream.Msg) { tools.Reported(m) })
})
// The consumer on the checker's stream exists before the open tries are listed and is heard after (ADR 0638 §3): an event
// said between the call and the hearing waits in it, and a try that started or ended meanwhile is found either way.
var checksCons jetstream.Consumer
_ = notHeard.Bind(PreviewsStream, "the checker's previews and tries are not heard until the indexer starts again", func() error {
c, err := cred.EventConsumer(ctx, js, PreviewsStream)
if err != nil {
return fmt.Errorf("the consumer %s_%s on %s does not exist yet: %w", cred.Node, cred.Module, PreviewsStream, err)
}
checksCons = c
return nil
})
// The versions each package's series holds (DLV-I5): the registry seats' published, on the
// indexer's durable consumer of the shared events stream. Where it is not there yet, no series moves until the indexer
// starts again, said.
@@ -744,6 +734,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
return nil
})
hearNamed(ctx, js, cred, notHeard, named, logger.Printf)
hearRuns(ctx, js, cred, notHeard, ix, logger.Printf)
// The checker's verdicts carry a pull request's out-of-step files (DLV-I20): on the indexer's durable consumer of the
// checker's stream asks. Where it is not there yet, no pull request's out-of-step files are kept until the indexer
// starts again, said.
@@ -762,17 +753,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
} else if words != "" {
logger.Printf("take-in at start: %s", words)
}
// The open tries are found again through the checker's tool: a try event missed while the indexer was down would leave a
// try with no candidate, or a candidate with no try. Not fatal: the events come, and a preview written writes the candidates.
// A recovery that fails is said (journal, and indexer_try_candidates) and the events are heard all the same.
_ = ix.RecoverTries(ctx)
if checksCons != nil {
pc, err := process.HearEvents(checksCons, func(m jetstream.Msg) { previews.heardChecks(m) }, logger.Printf)
if err != nil {
return err
}
defer pc.Stop()
}
cons, err := cred.EventConsumer(ctx, js, forgeStream)
if err != nil {
// Essential: the forge's merges are what the indexer indexes.
@@ -6,7 +6,6 @@ import (
"encoding/hex"
"encoding/json"
"fmt"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-checker/checkermsg"
"path"
"slices"
"strings"
@@ -261,31 +260,3 @@ func openVia(call toolCaller) func(ctx context.Context) ([]forgemsg.OpenPull, er
return out, nil
}
}
// openTriesTool is the checker's tool that lists the tries that wait or run (mesh-delivery-checker checker_tries).
const openTriesTool = "mesh-delivery-checker.checker_tries"
// openTriesVia lists the open tries through the checker's tool, never its state (ADR 0596).
func openTriesVia(call toolCaller) func(ctx context.Context) ([]checkermsg.Open, error) {
return func(context.Context) ([]checkermsg.Open, error) {
raw, err := call(openTriesTool, map[string]any{})
if err != nil {
return nil, err
}
// No try open is an answer: the checker's list is empty, or null in an older checker (issue 1921). An answer without the
// key is refused below, never read as none.
var o map[string]json.RawMessage
if err := json.Unmarshal(raw, &o); err != nil {
return nil, fmt.Errorf("the answer of %s is not {tries}: %v", openTriesTool, err)
}
list, ok := o["tries"]
if !ok {
return nil, fmt.Errorf("the answer of %s is not {tries}: no tries key", openTriesTool)
}
var tries []checkermsg.Open
if err := json.Unmarshal(list, &tries); err != nil {
return nil, fmt.Errorf("the answer of %s is not {tries}: %v", openTriesTool, err)
}
return tries, nil
}
}
@@ -4,7 +4,9 @@ import (
"context"
"encoding/json"
"errors"
"fmt"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/git"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner"
"mesh-delivery-indexer/internal/index"
"slices"
"strings"
@@ -127,23 +129,30 @@ func TestTheTrunkIsTheDirectoriesTheForgeListsTrees(t *testing.T) { // 1900/AC3
if _, err := ownBuildVia(none, "n1")(context.Background()); err == nil {
t.Error("an engine that says nothing runs gave a build")
}
// Issue 1921: the checker's list of open tries; none open is an answer, whether [] or null.
// Design 47 G §64, A-160, A-174: the runner's run-started and pipeline-ended reach the indexer; one that is no such body,
// or a run that can never be taken in, is terminated, any other failure is left for redelivery, and a taken one is acked.
for _, row := range []struct {
name, answer string
want int
refused bool
name, body string
ended bool
err error
acked, termd int
started, endedOf string
}{
{"one open", `{"tries":[{"id":"try-1","state":"running","whole":true,"members":[{"repository":"novox/a","head":"h","trunk":"t"}]}]}`, 1, false},
{"an empty list", `{"tries":[]}`, 0, false},
{"null, as an older checker answers none", `{"tries":null}`, 0, false},
{"no tries key", `{}`, 0, true},
{"not json", `nope`, 0, true},
{"tries of another shape", `{"tries":{"a":1}}`, 0, true},
{name: "a run started", body: `{"v":1,"run":"7","members":[{"module":"a","commit":"c"}],"by":"s"}`, acked: 1, started: "7"},
{name: "a run's pipeline ended", body: `{"v":1,"pipeline":"p","module":"a","branch":"b","commit":"c","outcome":"passed","steps":[],"run":"7"}`, ended: true, acked: 1, endedOf: "7"},
{name: "not json", body: `nope`, termd: 1},
{name: "a run never taken in", body: `{"v":1,"run":"7","members":[]}`, err: fmt.Errorf("%w: no member", index.ErrUnreadable), termd: 1, started: "7"},
{name: "a bus fault", body: `{"v":1,"run":"7","members":[{"module":"a","commit":"c"}]}`, err: errors.New("the bus did not answer"), started: "7"},
} {
call := func(string, any) (json.RawMessage, error) { return json.RawMessage(row.answer), nil }
got, err := openTriesVia(call)(context.Background())
if (err != nil) != row.refused || len(got) != row.want {
t.Errorf("%s: %d tries, error %v; want %d, refused %v", row.name, len(got), err, row.want, row.refused)
ix := &fakeRuns{err: row.err}
m := &fakeMsg{subject: "x", data: []byte(row.body)}
if row.ended {
heardPipelineEnded(context.Background(), ix, m, t.Logf)
} else {
heardRunStarted(context.Background(), ix, m, t.Logf)
}
if m.acked != row.acked || m.termd != row.termd || ix.started != row.started || ix.ended != row.endedOf {
t.Errorf("%s: acked %d, terminated %d, started %q, ended %q", row.name, m.acked, m.termd, ix.started, ix.ended)
}
}
}
@@ -177,3 +186,19 @@ func TestReadViaReadsManifestsAndLocks(t *testing.T) {
t.Fatalf("read = %+v", got)
}
}
// fakeRuns records the run events handed to it and answers err.
type fakeRuns struct {
err error
started, ended string
}
func (f *fakeRuns) StartRun(_ context.Context, e meshpipelinerunner.RunStarted) error {
f.started = e.Run
return f.err
}
func (f *fakeRuns) EndRunPipeline(_ context.Context, e meshpipelinerunner.PipelineEnded) error {
f.ended = e.Run
return f.err
}
@@ -4,30 +4,19 @@ import (
"context"
"errors"
"log"
"strings"
"sync"
"git.novox.be/novox/mesh-sdk/go/process"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-checker/checkermsg"
"mesh-delivery-indexer/internal/index"
)
// PreviewsStream is the stream the checker carries its events on, as the bus names it (declared by the
// checker as the stream checks, heard by this module through consumes; novox/hq ADR 0484 clause 3).
const PreviewsStream = "CHECKS"
// previewWrittenSubject is the subject of the checker's preview.written event.
const previewWrittenSubject = "mesh.stream." + index.CheckerModule + ".event.preview.written"
// previewWatch reads the merge previews the checker writes (mesh-delivery-checker_previews) and follows them: every change
// writes the candidates' fingerprints again (index.Indexer.ApplyPreview). The checker's bucket is made when it is assigned,
// after the indexer started, and the indexer holds no timer (ADR 0437 §8): it starts with the bucket absent, and the
// checker says each time it wrote a preview (preview.written, handled by Written); the bucket is then watched and settled
// as the others, with no restart.
// after the indexer started, and the indexer holds no timer (ADR 0437 §8) and hears no preview.written (design 47 G A-160): a
// bucket absent at the start is not watched until the indexer starts again.
type previewWatch struct {
ctx context.Context
js bucketOpener
@@ -53,16 +42,16 @@ func (w *previewWatch) ensure() error {
kv, err := w.js.KeyValue(w.ctx, index.PreviewsBucket)
switch {
case errors.Is(err, jetstream.ErrBucketNotFound):
w.logger.Printf("the bucket %s does not exist yet: it is watched when the checker says it wrote a preview", index.PreviewsBucket)
w.logger.Printf("the bucket %s does not exist yet: it is watched from the indexer's next start", index.PreviewsBucket)
return nil
case err != nil:
// Refused or not answered (its grant not on the bus yet, issue 1945): said, and opened again at the checker's next preview.
w.logger.Printf("the bucket %s cannot be read (%v): it is watched when the checker says it wrote a preview", index.PreviewsBucket, err)
w.logger.Printf("the bucket %s cannot be read (%v): it is watched from the indexer's next start", index.PreviewsBucket, err)
return nil
}
watcher, err := settle(w.ctx, kv, func(e jetstream.KeyValueEntry) { w.ix.PutPreview(e.Key(), e.Value(), !opOf(e)) })
if err != nil {
process.NotWatched(w.logger.Printf, index.PreviewsBucket, err, "it is watched when the checker says it wrote a preview")
process.NotWatched(w.logger.Printf, index.PreviewsBucket, err, "it is watched from the indexer's next start")
return nil
}
w.watched = true
@@ -73,64 +62,5 @@ func (w *previewWatch) ensure() error {
}
})
}()
if err := w.ix.RecomputeCandidates(w.ctx); err != nil {
return err
}
return w.ix.WriteTryCandidates(w.ctx)
}
// Written takes one preview.written event of the checker in: its bucket is watched if it is not. A subject that is not the
// checker's is terminated; a failure stays unacknowledged, and the server delivers it again.
func (w *previewWatch) Written(m delivered) {
if !strings.HasPrefix(m.Subject(), previewWrittenSubject) {
w.logger.Printf("%s: not the checker's preview.written, dropped", m.Subject())
_ = m.Term()
return
}
if err := w.ensure(); err != nil {
w.logger.Printf("%s: left for redelivery: %v", m.Subject(), err)
return
}
_ = m.Ack()
}
// tryEventSubject is the subject of an event of the checker's stream checks about a try.
func tryEventSubject(event string) string {
return "mesh.stream." + index.CheckerModule + ".event." + event
}
// heardChecks takes one event of the checker's stream in: a preview written (Written), or a whole try's start or end, from
// which the candidate of its group is written or deleted (index.Indexer.StartTry, EndTry). A failure stays unacknowledged, and
// the server delivers it again; a body that is no try event is terminated.
func (w *previewWatch) heardChecks(m delivered) {
var event string
switch m.Subject() {
case tryEventSubject(checkermsg.Started):
event = checkermsg.Started
case tryEventSubject(checkermsg.Ended):
event = checkermsg.Ended
default:
w.Written(m)
return
}
e, err := checkermsg.Decode(event, m.Data())
if err != nil {
w.logger.Printf("%s: not a try event, dropped: %v", m.Subject(), err)
_ = m.Term()
return
}
if event == checkermsg.Started {
var ms []index.TryMember
for _, x := range e.Members {
ms = append(ms, index.TryMember{Repository: x.Repository, Head: x.Head, Trunk: x.Trunk})
}
err = w.ix.StartTry(w.ctx, e.ID, ms)
} else {
err = w.ix.EndTry(w.ctx, e.ID)
}
if err != nil {
w.logger.Printf("%s of %s: left for redelivery: %v", event, e.ID, err)
return
}
_ = m.Ack()
return w.ix.RecomputeCandidates(w.ctx)
}
@@ -0,0 +1,77 @@
package main
import (
"context"
"encoding/json"
"errors"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner"
"git.novox.be/novox/mesh-sdk/go/process"
"mesh-delivery-indexer/internal/index"
)
// runTaker is what the indexer does with the runner's events of a run on request (index.Indexer.StartRun, EndRunPipeline).
type runTaker interface {
StartRun(ctx context.Context, e meshpipelinerunner.RunStarted) error
EndRunPipeline(ctx context.Context, e meshpipelinerunner.PipelineEnded) error
}
// hearRuns hears the runner's run-started and pipeline-ended (design 47 G §64, §190; A-160, A-174), each on its own stream, on
// the indexer's durable: a run's candidate is written on its start and dropped once every member's pipeline has ended. A stream
// the bus does not hold yet is the finding family-not-heard, and no run's candidate is written or dropped meanwhile.
func hearRuns(ctx context.Context, js jetstream.JetStream, cred process.Account, notHeard binder, ix runTaker, logf func(string, ...any)) {
for _, s := range []struct {
stream string
each func(jetstream.Msg)
}{
{meshpipelinerunner.StreamRunStarted, func(m jetstream.Msg) { heardRunStarted(ctx, ix, m, logf) }},
{meshpipelinerunner.StreamPipelineEnded, func(m jetstream.Msg) { heardPipelineEnded(ctx, ix, m, logf) }},
} {
_ = notHeard.Bind(s.stream, "no run on request's candidate is written or dropped until the indexer starts again", func() error {
cc, err := cred.HearSeatEvents(ctx, js, s.stream, s.each, logf)
if err != nil {
return err
}
go func() { <-ctx.Done(); cc.Stop() }()
return nil
})
}
}
// heardRunStarted takes one run-started in. A body that is no run-started, or a run that can never be taken in, is said and
// terminated; any other failure stays unacknowledged, and the server delivers it again.
func heardRunStarted(ctx context.Context, ix runTaker, m delivered, logf func(string, ...any)) {
var e meshpipelinerunner.RunStarted
if err := json.Unmarshal(m.Data(), &e); err != nil {
logf("%s: not a run-started, dropped: %v", m.Subject(), err)
_ = m.Term()
return
}
settleRun(m, ix.StartRun(ctx, e), "run-started of "+e.Run, logf)
}
// heardPipelineEnded takes one pipeline-ended in: of a run's pipeline, its module is kept as ended.
func heardPipelineEnded(ctx context.Context, ix runTaker, m delivered, logf func(string, ...any)) {
var e meshpipelinerunner.PipelineEnded
if err := json.Unmarshal(m.Data(), &e); err != nil {
logf("%s: not a pipeline-ended, dropped: %v", m.Subject(), err)
_ = m.Term()
return
}
settleRun(m, ix.EndRunPipeline(ctx, e), "pipeline-ended of "+e.Module+" in the run "+e.Run, logf)
}
func settleRun(m delivered, err error, what string, logf func(string, ...any)) {
switch {
case err == nil:
_ = m.Ack()
case errors.Is(err, index.ErrUnreadable):
logf("%s: never taken in, dropped: %v", what, err)
_ = m.Term()
default:
logf("%s: left for redelivery: %v", what, err)
}
}
@@ -22,8 +22,6 @@ type takerIn interface {
TakeInPendingByHand(ctx context.Context, key, by, why string) index.TakenIn
DropPending(ctx context.Context, key, by, why string, abandon bool) index.TakenIn
TakeInMerge(ctx context.Context, repository, merge string, seq uint64, by, why string) index.TakenIn
DropTryCandidate(ctx context.Context, id, by, why string) index.TakenIn
RecoverTriesByHand(ctx context.Context, by, why string) index.TakenIn
FingerprintsAgain(ctx context.Context, module string) index.TakenIn
RecomputeByHand(ctx context.Context, by, why string) index.TakenIn
DropCandidateGraph(ctx context.Context, group, heads, by, why string) index.TakenIn
@@ -64,12 +62,6 @@ func takenIn(ctx context.Context, ix takerIn, kv jetstream.KeyValue, e jetstream
a.Module = r.Module
} else if a.Fingerprints {
asked, r = "computing of fingerprints again", ix.FingerprintsAgain(ctx, a.Module)
} else if a.RecoverTries {
asked, r = "recovery of the tries", ix.RecoverTriesByHand(ctx, a.By, a.Why)
a.Module = "tries"
} else if a.TryCandidate != "" && a.Drop {
asked, r = "drop of a try's candidate", ix.DropTryCandidate(ctx, a.TryCandidate, a.By, a.Why)
a.Module = a.TryCandidate
} else if a.Merge != "" {
asked, r = "hand take-in of a merge", ix.TakeInMerge(ctx, a.MergeRepository, a.Merge, a.Sequence, a.By, a.Why)
a.Module = r.Module
+1 -3
View File
@@ -15,6 +15,7 @@ require (
git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdeliverychecker v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/nodebuildgo v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/npmpackageregistry v0.0.0-00010101000000-000000000000
@@ -41,7 +42,6 @@ require (
replace git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun => ../mesh-delivery-builder/buildrun
require (
git.novox.be/novox/mesh-delivery/modules/mesh-delivery-checker/checkermsg v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk v0.0.0-00010101000000-000000000000
@@ -53,8 +53,6 @@ replace git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk => ../mesh-de
replace git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein => ./takein
replace git.novox.be/novox/mesh-delivery/modules/mesh-delivery-checker/checkermsg => ../mesh-delivery-checker/checkermsg
replace git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg => ../mesh-delivery-forge-bridge/forgemsg
replace git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds => ../mesh-delivery-builder/builds
@@ -15,7 +15,6 @@ import (
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/nodebuildgo"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-checker/checkermsg"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshdeliveryclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
@@ -219,14 +218,10 @@ type Indexer struct {
Pending Store
// SayTakeIn publishes a take-in event (take-in.pending, take-in.taken; ADR 0633 §5); nil says none.
SayTakeIn func(ctx context.Context, t takein.TakeIn) error
// OpenTries lists the tries that wait or run, from the checker's tool (checker_tries); nil recovers none (trycandidates.go).
OpenTries func(ctx context.Context) ([]checkermsg.Open, error)
// TryRecovery keeps how the last recovery of the open tries went (trycandidates.go); nil keeps none.
TryRecovery Store
tryMu sync.Mutex // one try event or recovery at a time (trycandidates.go)
recoveryFailed atomic.Bool // the last recovery of the open tries failed (trycandidates.go)
tries map[string][]TryMember // the whole tries that wait or run, by try id (trycandidates.go); guarded by mu
pendingMu sync.Mutex // one take-in step at a time: a merge, a group's end, the start and a hand act (pending.go)
// Runs is the state runs: the runs on request the indexer holds a candidate for (runcandidates.go, A-174); nil keeps none.
Runs Store
runMu sync.Mutex // one run event at a time (runcandidates.go)
pendingMu sync.Mutex // one take-in step at a time: a merge, a group's end, the start and a hand act (pending.go)
}
func (ix *Indexer) logf(format string, args ...any) {
@@ -388,7 +383,7 @@ func (ix *Indexer) applyOpen(ctx context.Context, f forgemsg.Forge) error {
}
for _, key := range keys {
if meshcatalogclient.IsTryGroup(key) {
continue // a try's candidate is no pull request's: the checker's try owns it (trycandidates.go)
continue // a run's candidate is no pull request's: its run owns it (runcandidates.go)
}
err := write(ctx, ix.Store, key, func(value []byte, found bool) ([]byte, bool, error) {
if !found {
@@ -237,7 +237,7 @@ func (ix *Indexer) configCandidatesAfterMerge(ctx context.Context, f forgemsg.Fo
var errs []error
for _, g := range groups {
if meshcatalogclient.IsTryGroup(g) {
continue // a try's candidate holds no pull request's config declarations (trycandidates.go)
continue // a run's candidate holds no pull request's config declarations (runcandidates.go)
}
raw, _, found, err := ix.Store.Get(ctx, g)
if err != nil || !found {
@@ -1131,7 +1131,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
for _, f := range []string{"Store", "Modules", "Manifests", "Predefined", "ConfigCandidates", "Taken", "Verdicts", "OldSections", "Unjudged",
"JudgeID", "SayRefused", "SayAccepted", "Packages", "Graph", "Files", "GoSum", "GoMod", "Now", "Builds", "Named", "Trunk",
"Read", "Contains", "Logf", "ReadOpen", "OwnBuild", "Message", "Pending", "SayTakeIn",
"Requirements", "Findings", "SeatDefinitions", "ReadFile", "SaySeatDefinition", "OpenTries", "TryRecovery", "OwnBuildState", "Heard", "Placement", "RunningWhole", "Revisions", "SayCatalogRevision", "SayFill", "EntryIndex", "NamedSets", "Registry", "SeedModules", "Versions", "SayVersionReleased", "SayVersionsFilled", "Inputs", "BuildEdges", "Pins", "PinsPending"} {
"Requirements", "Findings", "SeatDefinitions", "ReadFile", "SaySeatDefinition", "Runs", "OwnBuildState", "Heard", "Placement", "RunningWhole", "Revisions", "SayCatalogRevision", "SayFill", "EntryIndex", "NamedSets", "Registry", "SeedModules", "Versions", "SayVersionReleased", "SayVersionsFilled", "Inputs", "BuildEdges", "Pins", "PinsPending"} {
named[f] = true
}
ty := reflect.TypeOf(Indexer{})
@@ -4,12 +4,8 @@ import (
"context"
"encoding/json"
"fmt"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient"
"strings"
"time"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein"
stdio "git.novox.be/novox/mesh-sdk/go"
)
@@ -115,94 +111,3 @@ func PendingTools(pending, ownBuild Store, ask func(key string, value []byte) (T
Run: merge},
}
}
// TryCandidateTools are the tools of a group try's candidate (issue 1921, trycandidates.go): indexer_try_candidates reads the
// candidates the indexer holds for tries (the try, its heads, each member's modules and whether each has its fingerprint),
// and indexer_drop_try_candidate drops one by an ask in takeins. candidates is the candidates state as the tools bundle reads it.
func TryCandidateTools(candidates, recovery Store, ask func(key string, value []byte) (TakeInAsk, error), now func() time.Time) []stdio.Tool {
return []stdio.Tool{
{Name: TryCandidatesTool,
Description: "The candidates the indexer holds for group tries (a try of several repositories at exact commits, mesh-delivery-checker try): " +
"per try its heads hash, each member's repository and commit, the modules the try touches and, per module, its manifest digest, its " +
"fingerprint or why it has none yet (a merge preview not yet written). The checker's manifest validator and the node-engines' " +
"compose-candidate read this candidate. Read-only; an empty answer means no whole try waits or runs.",
Input: map[string]any{"type": "object", "properties": map[string]any{}},
Run: func(map[string]any) (any, error) {
ctx := context.Background()
keys, err := candidates.Keys(ctx)
if err != nil {
return nil, err
}
var out []meshcatalog.CandidatesByGroupValue
for _, k := range keys {
if !meshcatalogclient.IsTryGroup(k) {
continue
}
v, _, found, err := candidates.Get(ctx, k)
if err != nil || !found {
continue
}
if c, err := DecodeCandidate(v); err == nil {
out = append(out, c)
}
}
// How the recovery of the open tries went: loud when it is not done (since when, why).
said := map[string]any{"done": false, "why": "no recovery has been written yet"}
if v, _, found, err := recovery.Get(ctx, TryRecoveryKey); err == nil && found {
var r TryRecovery
if json.Unmarshal(v, &r) == nil {
said = map[string]any{"done": r.Done}
if !r.Done {
said["why"], said["since"] = r.Why, r.Since
said["says"] = "recovery not done, since " + r.Since.Format(time.RFC3339) + ": " + r.Why
}
}
}
return map[string]any{"tries": out, "recovery": said}, nil
}},
{Name: RecoverTriesTool,
Description: "Run the recovery of the open tries now: the indexer lists them through the checker's checker_tries, writes the candidate of each " +
"whole try that waits or runs and deletes the candidate of a try no longer listed. It runs by itself at the indexer's start; when it failed " +
"(indexer_try_candidates says recovery not done, since when and why) this runs it again. A hand act: who and why are said in the journal.",
Input: map[string]any{"type": "object", "required": []string{"why"}, "properties": map[string]any{
"why": map[string]any{"type": "string", "description": "why it is run now, in plain words"}}},
Run: func(args map[string]any) (any, error) {
why, _ := args["why"].(string)
if strings.TrimSpace(why) == "" {
return nil, fmt.Errorf("say why: a hand act is said in the journal")
}
data, err := json.Marshal(TakeInAsk{V: 1, RecoverTries: true, Why: strings.TrimSpace(why), By: stdio.CallerOf(args), At: now().UTC()})
if err != nil {
return nil, err
}
a, err := ask("tries.recover."+now().UTC().Format("20060102T150405.000000000"), data)
if err != nil {
return nil, err
}
return a.Result, nil
}},
{Name: DropTryCandidateTool,
Description: "Drop a group try's candidate by hand (the indexer deletes it itself when the try ends): for a candidate left for a try that " +
"is gone, or to make the indexer write it again from the previews at the try's next change. A hand act: who and why are said in the journal.",
Input: map[string]any{"type": "object", "required": []string{"try", "why"}, "properties": map[string]any{
"try": map[string]any{"type": "string", "description": "the try's id, try-<digits>"},
"why": map[string]any{"type": "string", "description": "why it is dropped, in plain words"}}},
Run: func(args map[string]any) (any, error) {
id, _ := args["try"].(string)
why, _ := args["why"].(string)
id = strings.TrimSpace(id)
if !meshcatalogclient.IsTryGroup(id) || strings.TrimSpace(why) == "" {
return nil, fmt.Errorf("name the try (try-<digits>) and say why")
}
data, err := json.Marshal(TakeInAsk{V: 1, TryCandidate: id, Drop: true, Why: strings.TrimSpace(why), By: stdio.CallerOf(args), At: now().UTC()})
if err != nil {
return nil, err
}
a, err := ask("try-candidate.drop."+id, data)
if err != nil {
return nil, err
}
return a.Result, nil
}},
}
}
@@ -77,10 +77,7 @@ func (ix *Indexer) ApplyPreview(ctx context.Context, key string, value []byte, d
if !ix.PutPreview(key, value, deleted) {
return nil
}
if err := ix.RecomputeCandidates(ctx); err != nil {
return err
}
return ix.WriteTryCandidates(ctx) // a try that waited for this preview gets its candidate
return ix.RecomputeCandidates(ctx)
}
// RecomputeCandidates writes again every candidate whose pull requests' fingerprints, from the previews held, are no longer
@@ -7,15 +7,11 @@ import (
"fmt"
"maps"
"os"
"runtime"
"slices"
"strings"
"testing"
"time"
sdkstep "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildstep"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-checker/checkermsg"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdeliverychecker"
@@ -229,221 +225,6 @@ func TestAPreviewHeardBeforeItsPullRequestGivesTheFingerprintWhenTheCandidateIsW
if got := fpOf(t, cands); got.Fp2 != "" || got.None2 != meshcatalogclient.NoneNoPreview {
t.Errorf("a new head: %+v, want no preview", got)
}
// Issue 1921: a whole group try has no pull request, yet the indexer writes its candidate from the checker's try events and
// the previews its fetch wrote (number 0, the heads hash of its members), keeps it apart from the pull requests' candidates,
// and deletes it when the try ends. Its manifests are judged with the seat definitions its members carry, in one locked
// pass, whether the contracts module's manifest changed or only its seat.json did (the wait is bounded: a lock held twice
// would show as a failure here, not as a stuck run).
for _, row := range []struct {
name string
unchanged bool // the contracts module's module.json is the trunk's: only its definitions changed
}{
{name: "a changed manifest that carries definitions"},
{name: "an unchanged manifest that carries definitions", unchanged: true},
} {
t.Run("a group try's candidate: "+row.name, func(t *testing.T) {
ix, mods, cands := previewBed(t, true)
ctx := context.Background()
ix.Verdicts = newMem()
var changed map[string]any
if err := json.Unmarshal([]byte(catalogueManifest(t, "avahi")), &changed); err != nil {
t.Fatal(err)
}
changed["build"] = map[string]any{"artifacts": []any{map[string]any{"name": "go-seats", "kind": "package", "language": "go", "paths": []any{"go"}, "generated-from": "defs"}}}
raw, _ := json.Marshal(changed)
pv := avahiPreview(t, trunkNow, h1)
pm := meshdeliverychecker.PreviewsByRepositoryHeadTrunkValueModulesValue{Tree: "tree-avahi", Manifest: meshcatalogclient.ManifestDigest(raw), Locks: map[string]string{"go.sum": avahiGoSum}}
if row.unchanged {
m, found := moduleOf(t, mods, "avahi")
_ = found
pm.Manifest = m.Manifest
key, _ := meshcatalogclient.ManifestsKey(m.Manifest)
data, _ := meshcatalogclient.EncodeManifest(meshcatalog.ManifestsByDigestValue{Module: "avahi", Manifest: string(raw)})
if _, _, _, err := ix.Manifests.Get(ctx, key); err != nil {
t.Fatal(err)
}
_ = ix.Manifests.(*memStore).Update(ctx, key, data, ix.Manifests.(*memStore).revs[key])
}
pv.Modules["modules/avahi"] = pm
ix.Read = func(_ context.Context, _, _, dir string) (RepoRead, error) {
return RepoRead{Manifests: map[string]string{dir: string(raw)}}, nil
}
listed := 0
ix.Files = func(_ context.Context, _, commit string, dirs []string) (map[string]map[string]string, error) {
listed++
if commit != h1 || !slices.Equal(dirs, []string{"modules/avahi/defs/seats"}) {
t.Errorf("listed %v at %s, want the member's head", dirs, commit)
}
return map[string]map[string]string{dirs[0]: {"node-engine/seat.json": "1"}}, nil
}
ix.ReadFile = func(_ context.Context, _, _, p string) ([]byte, error) {
switch p {
case "modules/avahi/defs/seats/node-engine/seat.json":
return []byte(`{"format":1,"seat":"node-engine","scope":"node","grain":"seat-list","decision":"d","domain":"Core","verbs":[{"name":"try-verb"}]}`), nil
}
return nil, nil
}
started := `{"id":"try-7"}`
_ = started
members := []TryMember{{Repository: catalogueRepo, Head: h1, Trunk: trunkNow}}
held := func() []string { k, _ := cands.Keys(ctx); return k }
if err := ix.StartTry(ctx, "try-7", members); err != nil || len(held()) != 0 {
t.Fatalf("a try before its preview: %v, candidates %v", err, held())
}
done := make(chan struct{})
go func() { defer close(done); applyPreview(t, ix, pv) }()
select {
case <-done:
case <-time.After(10 * time.Second):
panic("writing the try's candidate hangs: the seat lock is held twice (a locked pass cannot be left by t.Fatal: the lock stays held)")
}
if listed == 0 {
t.Error("the member's definitions were not read")
}
key, _ := meshcatalogclient.CandidatesKey("try-7")
rawc, _, found, _ := cands.Get(ctx, key)
if !found {
t.Fatalf("no candidate for the try after its preview: %v", held())
}
c, err := DecodeCandidate(rawc)
pull := meshcatalog.CandidatesByGroupValuePullsItem{}
if len(c.Pulls) == 1 {
pull = c.Pulls[0]
}
if row.unchanged {
if err != nil || c.Group != "try-7" || len(pull.Modules) != 0 {
t.Fatalf("candidate = %+v, %v, want the try's group with no module touched", c, err)
}
} else if err != nil || c.Group != "try-7" || pull.Number != 0 || pull.Head != h1 || pull.Fps["avahi"].Manifest == "" { // no fingerprint asserted: this mesh-sdk pin does not read generated-from yet
t.Fatalf("candidate = %+v, %v, want one pull of number 0 with avahi's manifest", c, err)
}
if err := ix.Apply(ctx, forgemsg.Forge{Kind: forgemsg.ForgeOpen}); err != nil || len(held()) != 1 {
t.Errorf("the open message left %v (%v), want the try's candidate alone", held(), err)
}
// The tools: one reads the candidates held for tries, one drops by an ask the indexer answers.
asks := func(key string, value []byte) (TakeInAsk, error) {
a, err := DecodeTakeIn(value)
r := ix.DropTryCandidate(ctx, a.TryCandidate, a.By, a.Why)
a.Result = &r
return a, err
}
recovery := newMem()
ix.TryRecovery = recovery
tools := TryCandidateTools(cands, recovery, asks, time.Now)
if out, err := tools[0].Run(map[string]any{}); err != nil || len(out.(map[string]any)["tries"].([]meshcatalog.CandidatesByGroupValue)) != 1 {
t.Errorf("%s = %v, %v, want the try's candidate", tools[0].Name, out, err)
}
if _, err := tools[2].Run(map[string]any{"try": "feat/x", "why": "w"}); err == nil {
t.Error("a branch was dropped as a try")
}
if out, err := tools[2].Run(map[string]any{"try": "try-7", "why": "by hand"}); err != nil || !out.(*TakenIn).Stored || len(held()) != 0 {
t.Errorf("drop = %v, %v, candidates %v", out, err, held())
}
// The start recovered from the checker's tool: an open whole try is written again, one the checker no longer
// lists is deleted, and a try narrowed to a level is none.
ix.OpenTries = func(context.Context) ([]checkermsg.Open, error) {
return []checkermsg.Open{{ID: "try-7", Whole: true, Members: []checkermsg.Member{{Repository: catalogueRepo, Head: h1, Trunk: trunkNow}}},
{ID: "try-8", Whole: false, Members: []checkermsg.Member{{Repository: catalogueRepo, Head: h2, Trunk: trunkNow}}}}, nil
}
if err := ix.RecoverTries(ctx); err != nil || len(held()) != 1 {
t.Fatalf("recovered: %v, candidates %v, want try-7's alone", err, held())
}
ix.OpenTries = func(context.Context) ([]checkermsg.Open, error) { return nil, nil }
if err := ix.RecoverTries(ctx); err != nil || len(held()) != 0 {
t.Errorf("recovered with no open try: %v, candidates %v, want none", err, held())
}
// A recovery whose call fails is loud and deletes nothing: the candidate stays, indexer_try_candidates says recovery not
// done since when and why, and the next try event runs the recovery again, which clears it; the tool runs it by hand.
if err := ix.StartTry(ctx, "try-7", members); err != nil || len(held()) != 1 {
t.Fatalf("started: %v, candidates %v", err, held())
}
ix.OpenTries = func(context.Context) ([]checkermsg.Open, error) {
return nil, errors.New("checker_tries does not answer")
}
if err := ix.RecoverTries(ctx); err == nil || len(held()) != 1 {
t.Fatalf("a failed call: %v, candidates %v, want the error and the candidate kept", err, held())
}
if out, err := tools[0].Run(map[string]any{}); err != nil || !strings.Contains(out.(map[string]any)["recovery"].(map[string]any)["says"].(string), "recovery not done, since") ||
!strings.Contains(out.(map[string]any)["recovery"].(map[string]any)["says"].(string), "checker_tries does not answer") {
t.Errorf("%s = %v, %v, want recovery not done with since and why", tools[0].Name, out, err)
}
ix.OpenTries = func(context.Context) ([]checkermsg.Open, error) { return nil, nil }
if err := ix.StartTry(ctx, "try-11", members); err != nil || len(held()) != 1 || held()[0] != "try-11" {
t.Fatalf("a try event after a failed recovery: %v, candidates %v, want the recovery run (try-7 not listed, gone) and try-11 written", err, held())
}
if out, _ := tools[0].Run(map[string]any{}); out.(map[string]any)["recovery"].(map[string]any)["done"] != true {
t.Errorf("recovery still said not done after a success: %v", out)
}
ix.OpenTries = func(context.Context) ([]checkermsg.Open, error) { return nil, errors.New("down") }
recoverAsk := func(key string, value []byte) (TakeInAsk, error) {
a, err := DecodeTakeIn(value)
r := ix.RecoverTriesByHand(ctx, a.By, a.Why)
a.Result = &r
return a, err
}
byHand := TryCandidateTools(cands, recovery, recoverAsk, time.Now)
if out, err := byHand[1].Run(map[string]any{"why": "the checker was replaced"}); err != nil || out.(*TakenIn).Stored || !strings.Contains(out.(*TakenIn).Answer, "recovery not done") {
t.Errorf("%s with the checker down = %v, %v", byHand[1].Name, out, err)
}
ix.OpenTries = func(context.Context) ([]checkermsg.Open, error) { return nil, nil }
if out, err := byHand[1].Run(map[string]any{"why": "the checker is back"}); err != nil || !out.(*TakenIn).Stored || len(held()) != 0 {
t.Errorf("%s = %v, %v, candidates %v, want it recovered (try-11 not listed)", byHand[1].Name, out, err, held())
}
if err := ix.StartTry(ctx, "try-7", members); err != nil {
t.Fatal(err)
}
// ADR 0638 §3, §4: the consumer exists before the call, so what happens between the call and the hearing is heard after
// it. A try the call listed and whose start is then heard gets one candidate; a try started after the list gets one; a
// try that ended during the call, listed as open, leaves none once its end is heard.
ix.OpenTries = func(context.Context) ([]checkermsg.Open, error) {
return []checkermsg.Open{{ID: "try-7", Whole: true, Members: []checkermsg.Member{{Repository: catalogueRepo, Head: h1, Trunk: trunkNow}}},
{ID: "try-9", Whole: true, Members: []checkermsg.Member{{Repository: catalogueRepo, Head: h1, Trunk: trunkNow}}}}, nil
}
if err := ix.RecoverTries(ctx); err != nil {
t.Fatal(err)
}
for _, step := range []struct {
what string
do func() error
want int
}{
{"the start heard after the call, of a try the call listed", func() error { return ix.StartTry(ctx, "try-7", members) }, 2},
{"a start after the list", func() error { return ix.StartTry(ctx, "try-10", members) }, 3},
{"the end heard of a try ended during the call", func() error { return ix.EndTry(ctx, "try-9") }, 2},
} {
if err := step.do(); err != nil || len(held()) != step.want {
t.Fatalf("%s: %v, candidates %v, want %d", step.what, err, held(), step.want)
}
}
for _, id := range []string{"try-10"} {
if err := ix.EndTry(ctx, id); err != nil {
t.Fatal(err)
}
}
if err := ix.StartTry(ctx, "try-7", members); err != nil || len(held()) != 1 {
t.Fatalf("started again: %v, candidates %v", err, held())
}
if err := ix.EndTry(ctx, "try-7"); err != nil || len(held()) != 0 {
t.Errorf("an ended try: %v, candidates %v, want none", err, held())
}
// The hand recovery never overlaps a try event: a start heard while the call to the checker is out writes its candidate
// before or after the sweep, never between the call and the sweep (which would delete it, the try not being listed).
startDone := make(chan error, 1)
ix.OpenTries = func(context.Context) ([]checkermsg.Open, error) {
go func() { startDone <- ix.StartTry(ctx, "try-12", members) }()
for i := 0; i < 2000 && len(held()) == 0; i++ {
runtime.Gosched() // lets the start write if nothing holds it back
}
return nil, nil
}
if r := ix.RecoverTriesByHand(ctx, "lead", "a test"); !r.Stored {
t.Fatalf("the hand recovery = %+v", r)
}
if err := <-startDone; err != nil || len(held()) != 1 || held()[0] != "try-12" {
t.Errorf("a start heard during the hand recovery: %v, candidates %v, want try-12 kept", err, held())
}
})
}
// Issue 1921: a head whose tree lacks the directory asked is asked once, however many passes read it, until the head changes;
// any other failure (the forge does not answer) is asked again at every pass and said once per head.
for _, row := range []struct {
@@ -0,0 +1,296 @@
package index
// **A run on request's candidate** (design 47 G §64, §190, §321; A-160, A-174): the runner says `run-started` with the run's
// members, each a module at a named commit on a branch; the indexer builds the run's candidate from it, `candidates.run-<id>`,
// and drops it once every member's pipeline of that run has ended (`pipeline-ended` naming the run). Each member maps to its
// repository and directory by its registration, and its manifest and fingerprint come from the same take-in the indexer
// computes for a watched branch (its files read through the seat git), over the revision view of the member's branch when its
// registration watches that branch, else over the trunk's (A-93). No checker preview is heard and no checker tool is asked.
//
// What the indexer knows of each run is kept in its state runs (key run-<id>: the members and the modules whose pipeline
// ended), written on run-started and on each of the run's pipeline-ended and removed with the candidate, so a restart between
// two members' ends loses nothing (A-174). A pipeline-ended heard before its run-started is kept all the same: the run-started
// that follows finds it.
import (
"context"
"encoding/json"
"fmt"
"slices"
"strings"
"time"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshdeliveryclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/trunk"
"git.novox.be/novox/mesh-sdk/go/keys"
)
// RunsState is the indexer's state of the runs on request it holds a candidate for (A-174): the bucket
// mesh-delivery-indexer_runs, one key per run, run-<id>.
const (
RunsState = "runs"
RunsBucket = Module + "_" + RunsState
)
// Run is one value of the state runs: the run's members as run-started named them, and the modules whose pipeline of the
// run has ended.
type Run struct {
V int `json:"v"`
Run string `json:"run"`
Members []meshpipelinerunner.Member `json:"members,omitempty"`
Ended []string `json:"ended,omitempty"`
At time.Time `json:"at"`
}
// RunGroup is the candidate's group, and the runs key, of a run: run-<id> (A-160).
func RunGroup(run string) (string, error) {
g := "run-" + strings.TrimSpace(run)
if strings.TrimSpace(run) == "" || !meshcatalogclient.IsTryGroup(g) {
return "", fmt.Errorf("%w: %q is no run's id", ErrUnreadable, run)
}
return g, nil
}
// done says every member's pipeline of the run has ended; a run whose run-started is not heard yet is not done.
func (r Run) done() bool {
if len(r.Members) == 0 {
return false
}
for _, m := range r.Members {
if !slices.Contains(r.Ended, m.Module) {
return false
}
}
return true
}
// StartRun takes a run on request in, as the runner said it started: its members are kept in the state runs and its
// candidate is written; a run every member of which has ended already (its ends heard first) is dropped instead.
func (ix *Indexer) StartRun(ctx context.Context, e meshpipelinerunner.RunStarted) error {
group, err := RunGroup(e.Run)
if err != nil {
return err
}
if len(e.Members) == 0 {
return fmt.Errorf("%w: the run %s names no member", ErrUnreadable, e.Run)
}
ix.runMu.Lock()
defer ix.runMu.Unlock()
r, err := ix.updateRun(ctx, group, func(r *Run) { r.Members = slices.Clone(e.Members) })
if err != nil {
return err
}
if r.done() {
return ix.dropRun(ctx, group)
}
return ix.writeRunCandidate(ctx, group, r.Members)
}
// EndRunPipeline takes one pipeline-ended in: of a run's pipeline, its module is kept as ended, and once every member's has
// ended the candidate and the run's key are removed. A pipeline of no run is none of this.
func (ix *Indexer) EndRunPipeline(ctx context.Context, e meshpipelinerunner.PipelineEnded) error {
if e.Run == "" {
return nil
}
group, err := RunGroup(e.Run)
if err != nil {
return err
}
ix.runMu.Lock()
defer ix.runMu.Unlock()
r, err := ix.updateRun(ctx, group, func(r *Run) {
if e.Module != "" && !slices.Contains(r.Ended, e.Module) {
r.Ended = append(r.Ended, e.Module)
slices.Sort(r.Ended)
}
})
if err != nil {
return err
}
if !r.done() {
return nil
}
return ix.dropRun(ctx, group)
}
// updateRun reads the run's key, changes it and writes it back, made when absent.
func (ix *Indexer) updateRun(ctx context.Context, group string, change func(*Run)) (Run, error) {
if ix.Runs == nil {
return Run{}, fmt.Errorf("the state %s is not bound, so the run %s is not kept", RunsState, group)
}
var out Run
err := write(ctx, ix.Runs, group, func(value []byte, found bool) ([]byte, bool, error) {
r := Run{V: 1, Run: strings.TrimPrefix(group, "run-")}
if found {
if err := json.Unmarshal(value, &r); err != nil {
return nil, false, fmt.Errorf("%s.%s holds what is not a run: %w", RunsState, group, err)
}
}
change(&r)
r.At = ix.now().UTC()
out = r
data, err := json.Marshal(r)
return data, false, err
})
return out, err
}
// dropRun removes the run's candidate, then its key.
func (ix *Indexer) dropRun(ctx context.Context, group string) error {
if ix.Store != nil {
key, err := meshcatalogclient.CandidatesKey(group)
if err != nil {
return err
}
if err := write(ctx, ix.Store, key, func(_ []byte, found bool) ([]byte, bool, error) { return nil, found, nil }); err != nil {
return fmt.Errorf("the candidate of %s is not dropped: %w", group, err)
}
}
if err := write(ctx, ix.Runs, group, func(_ []byte, found bool) ([]byte, bool, error) { return nil, found, nil }); err != nil {
return fmt.Errorf("the run %s is not removed: %w", group, err)
}
ix.logf("the run %s: every member's pipeline ended, its candidate is dropped", group)
return nil
}
// RunBase is the branch whose revision view a member's take-in starts from (A-174): the member's own branch when its
// registration watches it, else the trunk.
func RunBase(reg meshcatalog.Registration, branch string) string {
if branch != "" && slices.Contains(reg.Branches, branch) {
return branch
}
return TrunkBranch
}
// runMember is one member placed: its registration's repository and directory, and the branch its take-in starts from.
type runMember struct {
meshpipelinerunner.Member
repository, dir, base string
}
// writeRunCandidate writes the run's candidate: one pull of number 0 per member, its module's manifest and fingerprint as
// the take-in of its commit over its base's revision view computes them, its manifests judged with the seat definitions
// the members carry, as a revision's are.
func (ix *Indexer) writeRunCandidate(ctx context.Context, group string, members []meshpipelinerunner.Member) error {
if ix.Store == nil || ix.Read == nil {
return nil
}
if ix.Registry == nil || ix.Registry.Registrations == nil || ix.Revisions == nil {
return fmt.Errorf("the run %s: the indexer holds no registrations or no revisions, so its members cannot be taken in", group)
}
var placed []runMember
for _, m := range members {
reg, found, err := ix.Registry.Registration(ctx, m.Module)
if err != nil {
return fmt.Errorf("the run %s: the registration of %s: %w", group, m.Module, err)
}
if !found {
return fmt.Errorf("%w: the run %s: %s has no registration, so its repository is not known", ErrUnreadable, group, m.Module)
}
dir := strings.Trim(reg.Path, "/")
if dir == meshcatalogclient.RootPath {
dir = ""
}
placed = append(placed, runMember{Member: m, repository: trunk.RepositoryName(reg.Origin), dir: dir, base: RunBase(reg, m.Branch)})
}
slices.SortFunc(placed, func(a, b runMember) int {
return strings.Compare(a.repository+"\x00"+a.Module, b.repository+"\x00"+b.Module)
})
heads := make([]meshdeliveryclient.Head, len(placed))
for i, m := range placed {
heads[i] = meshdeliveryclient.Head{Repository: m.repository, Head: m.Commit}
}
hash, err := meshdeliveryclient.HeadsHash(heads)
if err != nil {
return err
}
pulls := make([]meshcatalog.CandidatesByGroupValuePullsItem, len(placed))
var entries []PendingEntry
over := map[string]string{}
// One take-in per base branch: the members on one base are taken in together, so one member sees another's.
bases := map[string]*dryRun{}
for i, m := range placed {
d, ok := bases[m.base]
if !ok {
view, nodes, _, err := BranchView(ctx, ix.Revisions, m.base)
if err != nil {
return fmt.Errorf("the run %s: the revision view of %s: %w", group, m.base, err)
}
d = ix.dry()
d.ix.Modules, d.ix.Graph, d.ix.Taken = view, nodes, newOverlay(nil)
d.ix.Logf = func(format string, args ...any) {
ix.logf("the run "+group+", computed and not written: "+format, args...)
}
bases[m.base] = d
}
read, err := ix.Read(ctx, m.repository, m.Commit, m.dir)
if err != nil {
return fmt.Errorf("the run %s: reading %s of %s at %.12s: %w", group, m.Module, m.repository, m.Commit, err)
}
raw := read.Manifests[m.dir]
if raw == "" {
return fmt.Errorf("%w: the run %s: %s at %.12s holds no manifest at %q", ErrUnreadable, group, m.repository, m.Commit, m.dir)
}
if name := manifestModule(raw); name != m.Module {
return fmt.Errorf("%w: the run %s: the manifest at %q of %s at %.12s names %q, not %s", ErrUnreadable, group, m.dir, m.repository, m.Commit, name, m.Module)
}
f := forgemsg.Forge{V: keys.Version, Kind: forgemsg.ForgeMerged, Repository: m.repository, Merge: m.Commit, Root: read.Root,
Trees: map[string]string{m.dir: read.Trees[m.dir]}, Manifests: map[string]string{m.dir: raw}}
if l := read.Locks[m.dir]; len(l) > 0 {
f.Locks = map[string]map[string]string{m.dir: l}
}
if p := read.Packages[m.dir]; len(p) > 0 {
f.Packages = map[string][]string{m.dir: p}
}
if _, err := d.ix.applyManifests(ctx, f, false); err != nil {
return fmt.Errorf("the run %s: taking %s in: %w", group, m.Module, err)
}
key, err := meshcatalogclient.ModulesKey(m.Module)
if err != nil {
return err
}
value, _, found, err := d.ix.Modules.Get(ctx, key)
if err != nil {
return err
}
if !found {
return fmt.Errorf("the run %s: the take-in of %s at %.12s wrote no module (the journal says why)", group, m.Module, m.Commit)
}
mv, err := meshcatalogclient.DecodeModule(value)
if err != nil {
return err
}
pulls[i] = meshcatalog.CandidatesByGroupValuePullsItem{Repository: m.repository, Head: m.Commit, Modules: []string{m.Module},
Fps: map[string]meshcatalog.CandidatesByGroupValuePullsItemFpsValue{m.Module: {Manifest: mv.Manifest, Fp2: mv.Fp2, None2: mv.None2,
Packages: mv.Packages}}}
over[m.Module] = mv.Manifest
entries = append(entries, PendingEntry{Forge: forgemsg.Forge{Repository: m.repository, Merge: m.Commit, Manifests: map[string]string{m.dir: raw}}})
}
seats, err := ix.revisionSeats(ctx, entries)
if err != nil {
return fmt.Errorf("the run %s: reading the seat definitions of its members: %w", group, err)
}
ix.judgeDigestsWith(ctx, seats, over, "the run "+group)
c := meshcatalog.CandidatesByGroupValue{V: keys.Version, Group: group, Heads: hash, Pulls: pulls}
c.Fpset = meshcatalogclient.FingerprintSet(c.Pulls)
if err := ix.refuseDisagreement(c); err != nil {
return err
}
data, err := EncodeCandidate(c)
if err != nil {
return err
}
key, err := meshcatalogclient.CandidatesKey(group)
if err != nil {
return err
}
if err := write(ctx, ix.Store, key, func(_ []byte, _ bool) ([]byte, bool, error) { return data, false, nil }); err != nil {
return err
}
ix.logf("the run %s: its candidate of %d members is written", group, len(placed))
return nil
}
@@ -0,0 +1,130 @@
package index
import (
"context"
"encoding/json"
"errors"
"testing"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner"
)
// Design 47 G §64, §190, A-160, A-174: a run on request's candidate is built from the runner's run-started, each member taken
// in from its registration's repository at its commit (read through the seat git), and dropped once every member's pipeline of
// the run has ended; what the indexer knows of the run is kept in its state runs, so a restart between two ends loses nothing.
func TestARunsCandidateIsBuiltOnItsStartAndDroppedWhenEveryMembersPipelineEnded(t *testing.T) {
const c1, c2 = "1111111111111111111111111111111111111111", "2222222222222222222222222222222222222222"
bed := func(t *testing.T) (*Indexer, *memStore, *memStore, *int) {
t.Helper()
ix, _, _ := manifestsIndexer(t)
cands, runs := newMem(), newMem()
ix.Store, ix.Runs, ix.Revisions = cands, runs, newMemObjects(0)
ix.Registry = &Registry{Registrations: MemoryStore()}
for _, r := range []meshcatalog.Registration{
{Module: "avahi", Origin: catalogueRepo, Path: "modules/avahi", Branches: []string{"main"}},
{Module: "netcheck", Origin: "novox/mesh-networking", Path: "modules/netcheck", Branches: []string{"main", "beta"}},
} {
data, _ := json.Marshal(r)
if err := ix.Registry.Registrations.Create(context.Background(), RegistrationKey(r.Module), data); err != nil {
t.Fatal(err)
}
}
reads := 0
ix.Read = func(_ context.Context, repository, commit, dir string) (RepoRead, error) {
reads++
module := map[string]string{"modules/avahi": "avahi", "modules/netcheck": "netcheck"}[dir]
return RepoRead{Root: "r", Trees: map[string]string{dir: "tree-" + commit[:4]}, Manifests: map[string]string{dir: catalogueManifest(t, module)},
Locks: map[string]map[string]string{dir: {"go.sum": avahiGoSum}}}, nil
}
return ix, cands, runs, &reads
}
ctx := context.Background()
started := meshpipelinerunner.RunStarted{V: 1, Run: "7", By: "lead@shanks", Members: []meshpipelinerunner.Member{
{Module: "avahi", Commit: c1, Branch: "feature/1966-x"}, {Module: "netcheck", Commit: c2, Branch: "beta"}}}
ended := func(module string) meshpipelinerunner.PipelineEnded {
return meshpipelinerunner.PipelineEnded{V: 1, Pipeline: "p", Module: module, Branch: "b", Commit: "c", Outcome: "passed", Run: "7"}
}
key, _ := meshcatalogclient.CandidatesKey("run-7")
t.Run("built on its start, one pull per member, and dropped with the run after a restart", func(t *testing.T) {
ix, cands, runs, reads := bed(t)
if err := ix.StartRun(ctx, started); err != nil {
t.Fatal(err)
}
raw, _, found, _ := cands.Get(ctx, key)
c, err := DecodeCandidate(raw)
if !found || err != nil || c.Group != "run-7" || len(c.Pulls) != 2 || *reads != 2 {
t.Fatalf("candidate %+v (%v, found %v, %d reads), want run-7 with two pulls read through the seat git", c, err, found, *reads)
}
for _, p := range c.Pulls {
m := p.Modules[0]
if p.Number != 0 || len(p.Modules) != 1 || p.Fps[m].Manifest == "" || (p.Fps[m].Fp2 == "" && p.Fps[m].None2 == "") ||
(m == "avahi") != (p.Repository == catalogueRepo && p.Head == c1) {
t.Errorf("pull %+v, want number 0, its registration's repository at its commit, a manifest and a fingerprint or why none", p)
}
}
if err := ix.EndRunPipeline(ctx, ended("avahi")); err != nil {
t.Fatal(err)
}
// A restart: a fresh indexer over the same states.
again := &Indexer{Store: cands, Runs: runs, Revisions: ix.Revisions, Registry: ix.Registry, Read: ix.Read}
if _, _, found, _ := cands.Get(ctx, key); !found {
t.Fatal("the candidate went before its last member's pipeline ended")
}
if err := again.EndRunPipeline(ctx, ended("netcheck")); err != nil {
t.Fatal(err)
}
if _, _, found, _ := cands.Get(ctx, key); found {
t.Error("the candidate stays after every member's pipeline ended")
}
if k, _ := runs.Keys(ctx); len(k) != 0 {
t.Errorf("runs holds %v after the run's end, want none", k)
}
})
t.Run("ends heard before the start drop it at the start", func(t *testing.T) {
ix, cands, runs, _ := bed(t)
for _, m := range []string{"avahi", "netcheck"} {
if err := ix.EndRunPipeline(ctx, ended(m)); err != nil {
t.Fatal(err)
}
}
if err := ix.StartRun(ctx, started); err != nil {
t.Fatal(err)
}
k1, _ := cands.Keys(ctx)
k2, _ := runs.Keys(ctx)
if len(k1)+len(k2) != 0 {
t.Errorf("candidates %v, runs %v, want none", k1, k2)
}
})
t.Run("a member with no registration is never taken in; a pipeline of no run is none of this", func(t *testing.T) {
ix, cands, _, _ := bed(t)
lost := started
lost.Members = []meshpipelinerunner.Member{{Module: "gone", Commit: c1}}
if err := ix.StartRun(ctx, lost); !errors.Is(err, ErrUnreadable) {
t.Errorf("a member with no registration: %v, want unreadable", err)
}
if err := ix.EndRunPipeline(ctx, meshpipelinerunner.PipelineEnded{Module: "avahi"}); err != nil {
t.Error(err)
}
if k, _ := cands.Keys(ctx); len(k) != 0 {
t.Errorf("candidates %v, want none", k)
}
})
t.Run("a member's take-in starts from its own branch only when its registration watches it", func(t *testing.T) {
reg := meshcatalog.Registration{Branches: []string{"main", "beta"}}
for branch, want := range map[string]string{"beta": "beta", "main": TrunkBranch, "feature/1966-x": TrunkBranch, "": TrunkBranch} {
if got := RunBase(reg, branch); got != want {
t.Errorf("RunBase(%q) = %q, want %q", branch, got, want)
}
}
})
t.Run("run-<id> is a run's group beside try-<digits>, never a pull request's", func(t *testing.T) {
for g, want := range map[string]bool{"run-7": true, "run-a1-b": true, "try-3": true, "run-": false, "run-a.b": false, "feat/x": false} {
if meshcatalogclient.IsTryGroup(g) != want {
t.Errorf("IsTryGroup(%q) = %v, want %v", g, !want, want)
}
}
})
}
@@ -57,13 +57,9 @@ type TakeInAsk struct {
Merge string `json:"merge,omitempty"`
MergeRepository string `json:"merge-repository,omitempty"`
Sequence uint64 `json:"sequence,omitempty"`
// TryCandidate asks the candidate of this try's group to be dropped (indexer_drop_try_candidate), trycandidates.go.
TryCandidate string `json:"try-candidate,omitempty"`
// Fingerprints asks the fingerprints of Module ("all": every module) to be computed again now, from the published
// packages as they are read (indexer_fingerprints_again, fingerprints.go).
Fingerprints bool `json:"fingerprints,omitempty"`
// RecoverTries asks the open tries' candidates to be recovered now (indexer_recover_tries).
RecoverTries bool `json:"recover-tries,omitempty"`
// Recompute asks the catalog graph to be computed again now and compared with the newest kept revision (indexer_recompute).
Recompute bool `json:"recompute,omitempty"`
// CandidateGroup and CandidateHeads, with Drop, ask that candidate's catalog half to be dropped (indexer_drop_candidate_graph).
@@ -98,8 +94,8 @@ func DecodeTakeIn(data []byte) (TakeInAsk, error) {
if a.V != 1 {
return a, fmt.Errorf("a takeins value of version %d, not 1", a.V)
}
if a.Module == "" && a.Repository == "" && a.Pending == "" && a.Merge == "" && a.TryCandidate == "" && !a.RecoverTries && !a.Fingerprints && !a.CatchUp && !a.Recompute && a.CandidateGroup == "" && a.NamedFingerprint == "" && a.UnpinRevision == "" && a.Registry == nil {
return a, errors.New("a takeins value naming no module, repository, pending wait, merge or try")
if a.Module == "" && a.Repository == "" && a.Pending == "" && a.Merge == "" && !a.Fingerprints && !a.CatchUp && !a.Recompute && a.CandidateGroup == "" && a.NamedFingerprint == "" && a.UnpinRevision == "" && a.Registry == nil {
return a, errors.New("a takeins value naming no module, repository, pending wait or merge")
}
return a, nil
}
@@ -1,437 +0,0 @@
package index
// **A group try's candidate** (issue 1921; novox/hq ADR 0593 §7): a try of several repositories at exact commits has no pull
// request, so the forge bridge sends the indexer no message for it; yet the node-engine's compose-candidate and the checker's
// manifest validator read the indexer's candidate of a group. The indexer therefore watches the checker's state tries (the
// checker keeps it; this module reads it) and, for a try that is not narrowed to a level or to some tests, writes
// `candidates.<try id>`: one pull per member (number 0), its modules and manifests taken from the merge preview the try's
// fetch wrote, its manifests kept and judged with the seat definitions the members carry, as a take-in judges a revision's.
// The fingerprints come with the previews (RecomputeCandidates). The candidate is deleted when the try ends. `try-<digits>`
// is the reserved group name of a try: the bridge's open message never takes it out, and the checker never checks it.
import (
"context"
"encoding/json"
"errors"
"fmt"
"path"
"slices"
"strings"
"time"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshdeliveryclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdeliverychecker"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg"
"git.novox.be/novox/mesh-sdk/go/keys"
)
// TryCandidateTools: the tools that read, drop and recover a try's candidate.
const (
TryCandidatesTool = "indexer_try_candidates"
DropTryCandidateTool = "indexer_drop_try_candidate"
RecoverTriesTool = "indexer_recover_tries"
// TryRecoveryState is the indexer's state of how the last recovery of the open tries went: the bucket
// mesh-delivery-indexer_try-recovery, one key (TryRecoveryKey).
TryRecoveryState = "try-recovery"
TryRecoveryBucket = Module + "_" + TryRecoveryState
TryRecoveryKey = "state"
)
// TryRecovery is the value of the try-recovery state: whether the candidates of the open tries were recovered at the
// indexer's start or by hand, and when it began to fail, why.
type TryRecovery struct {
V int `json:"v"`
Done bool `json:"done"`
Since time.Time `json:"since,omitzero"`
Why string `json:"why,omitempty"`
At time.Time `json:"at"`
}
// TryMember is one repository of a try: its exact commit and the trunk commit it is merged onto.
type TryMember struct{ Repository, Head, Trunk string }
// StartTry takes a whole try in, as the checker said it started (checkermsg.Started): its members at their commits are kept, and
// its candidate is written once every member's preview is held. The checker's state is never read (ADR 0596).
func (ix *Indexer) StartTry(ctx context.Context, id string, members []TryMember) error {
ix.tryMu.Lock()
defer ix.tryMu.Unlock()
ix.retryRecovery(ctx)
return ix.startTry(ctx, id, members)
}
// retryRecovery runs the recovery again on a try event when the last one failed: the failure stays said until one succeeds.
func (ix *Indexer) retryRecovery(ctx context.Context) {
if ix.recoveryFailed.Load() && ix.OpenTries != nil {
_ = ix.recoverNow(ctx)
}
}
func (ix *Indexer) startTry(ctx context.Context, id string, members []TryMember) error {
if !meshcatalogclient.IsTryGroup(id) || len(members) == 0 {
return fmt.Errorf("%q with %d members is no try to start", id, len(members))
}
ix.mu.Lock()
if ix.tries == nil {
ix.tries = map[string][]TryMember{}
}
ix.tries[id] = slices.Clone(members)
ix.mu.Unlock()
return ix.writeTryCandidate(ctx, id)
}
// EndTry takes a try out, as the checker said it ended (checkermsg.Ended), and deletes its candidate.
func (ix *Indexer) EndTry(ctx context.Context, id string) error {
ix.tryMu.Lock()
defer ix.tryMu.Unlock()
ix.retryRecovery(ctx)
return ix.endTry(ctx, id)
}
func (ix *Indexer) endTry(ctx context.Context, id string) error {
ix.mu.Lock()
delete(ix.tries, id)
ix.mu.Unlock()
if !meshcatalogclient.IsTryGroup(id) {
return nil
}
return ix.dropTryCandidate(ctx, id)
}
// RecoverTries is, at the indexer's start, what the events it missed would have said: the candidates of the tries that wait
// or run now are written, and the candidate of a try the checker no longer lists is deleted. The checker's tool lists them
// (OpenTries); its bucket is not read. Nothing to recover from when no part is wired to ask.
//
// A recovery that fails is said loudly and kept (the try-recovery state, read by indexer_try_candidates) until one succeeds:
// the events are still heard after it, and indexer_recover_tries runs it again by hand.
//
// A recovery and a try event never overlap (tryMu): a sweep that deleted the candidate a start event wrote after the call to the
// checker would leave a running try with none.
func (ix *Indexer) RecoverTries(ctx context.Context) error {
ix.tryMu.Lock()
defer ix.tryMu.Unlock()
return ix.recoverNow(ctx)
}
func (ix *Indexer) recoverNow(ctx context.Context) error {
if ix.OpenTries == nil || ix.Store == nil {
return nil
}
err := ix.recoverTries(ctx)
ix.recoveryFailed.Store(err != nil)
ix.noteRecovery(ctx, err)
return err
}
// noteRecovery writes how a recovery went: failing keeps the time it began to fail, and the journal says it each time.
func (ix *Indexer) noteRecovery(ctx context.Context, err error) {
now := ix.now()
r := TryRecovery{V: 1, Done: err == nil, At: now.UTC()}
if err != nil {
r.Why, r.Since = err.Error(), now.UTC()
ix.logf("RECOVERY OF THE TRIES NOT DONE: %v (the try candidates may be stale or missing until indexer_recover_tries or the next start succeeds)", err)
}
if ix.TryRecovery == nil {
return
}
for range writeTries {
cur, rev, found, gerr := ix.TryRecovery.Get(ctx, TryRecoveryKey)
if gerr != nil {
ix.logf("the try-recovery state cannot be read: %v", gerr)
return
}
if err != nil && found {
var was TryRecovery
if json.Unmarshal(cur, &was) == nil && !was.Done && !was.Since.IsZero() {
r.Since = was.Since
}
}
data, _ := json.Marshal(r)
var werr error
if found {
werr = ix.TryRecovery.Update(ctx, TryRecoveryKey, data, rev)
} else {
werr = ix.TryRecovery.Create(ctx, TryRecoveryKey, data)
}
if werr == nil {
return
}
if !errors.Is(werr, ErrConflict) {
ix.logf("the try-recovery state cannot be written: %v", werr)
return
}
}
}
// RecoverTriesByHand runs the recovery now (indexer_recover_tries), a hand act said in the journal.
func (ix *Indexer) RecoverTriesByHand(ctx context.Context, by, why string) TakenIn {
r := TakenIn{Module: "tries"}
ix.logf("recovery of the tries asked by %s (%s)", by, why)
if ix.OpenTries == nil {
r.Answer = "the indexer is wired to list no open try: nothing to recover from"
return r
}
if err := ix.RecoverTries(ctx); err != nil {
r.Answer = "recovery not done: " + err.Error()
return r
}
r.Stored, r.Answer = true, "the candidates of the open tries are recovered"
return r
}
func (ix *Indexer) recoverTries(ctx context.Context) error {
open, err := ix.OpenTries(ctx)
if err != nil {
return fmt.Errorf("the open tries cannot be listed, so a missed try event is not recovered: %w", err)
}
live := map[string]bool{}
for _, o := range open {
if !o.Whole || len(o.Members) == 0 {
continue
}
live[o.ID] = true
var ms []TryMember
for _, m := range o.Members {
ms = append(ms, TryMember{m.Repository, m.Head, m.Trunk})
}
if err := ix.startTry(ctx, o.ID, ms); err != nil {
return err
}
}
keys, err := ix.Store.Keys(ctx)
if err != nil {
return err
}
for _, k := range keys {
if meshcatalogclient.IsTryGroup(k) && !live[k] {
if err := ix.endTry(ctx, k); err != nil {
return err
}
}
}
return nil
}
// WriteTryCandidates writes the candidate of every live try that has none yet and whose previews are all there: after a
// preview was written, the try that waited for it.
func (ix *Indexer) WriteTryCandidates(ctx context.Context) error {
ix.mu.Lock()
ids := make([]string, 0, len(ix.tries))
for id := range ix.tries {
ids = append(ids, id)
}
ix.mu.Unlock()
slices.Sort(ids)
for _, id := range ids {
if err := ix.writeTryCandidate(ctx, id); err != nil {
return err
}
}
return nil
}
func (ix *Indexer) dropTryCandidate(ctx context.Context, id string) error {
if ix.Store == nil {
return nil
}
key, err := meshcatalogclient.CandidatesKey(id)
if err != nil {
return err
}
return write(ctx, ix.Store, key, func(_ []byte, found bool) ([]byte, bool, error) { return nil, found, nil })
}
// DropTryCandidate deletes a try's candidate by hand (indexer_drop_try_candidate); a live try gets it again at its next
// change.
func (ix *Indexer) DropTryCandidate(ctx context.Context, id, by, why string) TakenIn {
r := TakenIn{Module: id}
switch {
case !meshcatalogclient.IsTryGroup(id):
r.Answer = id + " is no try's group, try-<digits>: nothing dropped"
default:
if err := ix.dropTryCandidate(ctx, id); err != nil {
r.Answer = "the candidate of " + id + " could not be dropped: " + err.Error()
} else {
r.Stored, r.Answer = true, "the candidate of "+id+" is dropped by "+by+" ("+why+")"
ix.logf("%s", r.Answer)
}
}
return r
}
// writeTryCandidate writes the candidate of a live try when it has none at its members' heads and every member's merge
// preview is held.
func (ix *Indexer) writeTryCandidate(ctx context.Context, id string) error {
if ix.Store == nil || ix.Read == nil {
return nil
}
ix.mu.Lock()
members := slices.Clone(ix.tries[id])
ix.mu.Unlock()
if len(members) == 0 {
return nil
}
key, err := meshcatalogclient.CandidatesKey(id)
if err != nil {
return err
}
slices.SortFunc(members, func(a, b TryMember) int { return strings.Compare(a.Repository, b.Repository) })
heads := make([]meshdeliveryclient.Head, len(members))
for i, m := range members {
heads[i] = meshdeliveryclient.Head{Repository: m.Repository, Head: m.Head}
}
hash, err := meshdeliveryclient.HeadsHash(heads)
if err != nil {
return err
}
if value, _, found, err := ix.Store.Get(ctx, key); err == nil && found {
if c, err := DecodeCandidate(value); err == nil && c.Heads == hash {
return nil // fingerprints follow the previews (RecomputeCandidates)
}
}
previews := make([]meshdeliverychecker.PreviewsByRepositoryHeadTrunkValue, len(members))
for i, m := range members {
pk, err := PreviewKey(m.Repository, m.Head, m.Trunk)
if err != nil {
return fmt.Errorf("the try %s: %w", id, err)
}
ix.mu.Lock()
pv, ok := ix.previews[pk]
ix.mu.Unlock()
if !ok {
return nil // its fetch has not written the preview yet: this is called again when it does
}
previews[i] = pv
}
held, err := ix.heldModules(ctx)
if err != nil {
return err
}
pulls := make([]meshcatalog.CandidatesByGroupValuePullsItem, len(members))
var entries []PendingEntry
over := map[string]string{}
for i, m := range members {
pull := meshcatalog.CandidatesByGroupValuePullsItem{Repository: heads[i].Repository, Head: heads[i].Head, Modules: []string{}}
// defs are every module directory's manifest of the member at its head, the unchanged ones too: a contracts module whose
// seat.json changed while its module.json did not carries definitions all the same (ADR 0636 §4).
defs := map[string]string{}
for _, dir := range slices.Sorted(mapKeys(previews[i].Modules)) {
pm := previews[i].Modules[dir]
name, unchanged := "", false
for n, h := range held {
if strings.EqualFold(h.Repository, m.Repository) && h.Path == forgeDir(dir) {
name, unchanged = n, h.Manifest == pm.Manifest
break
}
}
if unchanged {
if raw := ix.keptManifest(ctx, pm.Manifest); raw != "" {
defs[dir] = raw
}
continue // the manifest the trunk holds: the try does not touch the module
}
rr, err := ix.Read(ctx, m.Repository, m.Head, forgeDir(dir))
if err != nil {
return fmt.Errorf("the try %s: reading the manifest of %s at %.12s: %w", id, dir, m.Head, err)
}
raw := rr.Manifests[forgeDir(dir)]
if raw == "" {
ix.logf("the try %s: %s holds no manifest at %.12s, so no module of it is taken", id, dir, m.Head)
continue
}
module := manifestModule(raw)
if module == "" {
module = name
}
if module == "" {
module = path.Base(strings.Trim(dir, "/."))
}
digest, err := ix.keepManifest(ctx, module, raw)
if err != nil {
return err
}
if digest != pm.Manifest {
ix.logf("the try %s: the manifest of %s read at %.12s is not the one its preview names, left", id, dir, m.Head)
continue
}
if pull.Fps == nil {
pull.Fps = map[string]meshcatalog.CandidatesByGroupValuePullsItemFpsValue{}
}
pull.Modules = append(pull.Modules, module)
pull.Fps[module] = meshcatalog.CandidatesByGroupValuePullsItemFpsValue{Manifest: digest, None2: meshcatalogclient.NoneNoPreview}
defs[dir] = raw
over[module] = digest
}
pulls[i] = pull
entries = append(entries, PendingEntry{Forge: forgemsg.Forge{Repository: m.Repository, Merge: m.Head, Manifests: defs}})
}
// The manifests are judged with the seat definitions the members carry (seatsof.go, ADR 0636 §4), in one pass that takes the
// seats. Their verdicts are the verdicts state's, keyed by the manifest's digest, as a pull request's candidate's are
// (judgeDigests in ApplyMessage): nothing reads them but the engines' composition of that digest and the module
// value of a trunk module with the same digest.
seats, err := ix.revisionSeats(ctx, entries)
if err != nil {
return fmt.Errorf("the try %s: reading the seat definitions of its members: %w", id, err)
}
ix.judgeDigestsWith(ctx, seats, over, "the try "+id)
c := meshcatalog.CandidatesByGroupValue{V: keys.Version, Group: id, Heads: hash, Pulls: pulls}
c.Fpset = meshcatalogclient.FingerprintSet(c.Pulls)
ix.withPreviews(ctx, &c)
if err := ix.refuseDisagreement(c); err != nil {
return err
}
data, err := EncodeCandidate(c)
if err != nil {
return err
}
return write(ctx, ix.Store, key, func(_ []byte, _ bool) ([]byte, bool, error) { return data, false, nil })
}
// keptManifest is the text of the manifest the indexer keeps under a digest, or empty.
func (ix *Indexer) keptManifest(ctx context.Context, digest string) string {
if ix.Manifests == nil || digest == "" {
return ""
}
key, err := meshcatalogclient.ManifestsKey(digest)
if err != nil {
return ""
}
value, _, found, err := ix.Manifests.Get(ctx, key)
if err != nil || !found {
return ""
}
m, err := meshcatalogclient.DecodeManifest(value)
if err != nil {
return ""
}
return m.Manifest
}
func mapKeys[V any](m map[string]V) func(func(string) bool) {
return func(yield func(string) bool) {
for k := range m {
if !yield(k) {
return
}
}
}
}
// forgeDir is a preview's directory as the forge reader spells it: the root as "".
func forgeDir(dir string) string {
if dir == "." || dir == "/" {
return ""
}
return strings.Trim(dir, "/")
}
// heldAt says the trunk holds a module of the repository in that directory.
func heldAt(held map[string]meshcatalog.ModulesByModuleValue, repository, dir string) bool {
for _, h := range held {
if strings.EqualFold(h.Repository, repository) && h.Path == forgeDir(dir) {
return true
}
}
return false
}
@@ -127,7 +127,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
t.Errorf("uses = %v, invokes = %v, want the seat operator-channel used and seat:mesh-module-assigner.assignments invoked", m.Uses, m.Invokes)
}
// 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: "predefined", History: 5}, {Name: "config-candidates", History: 5}, {Name: "taken", History: 1}, {Name: "takeins", History: 1}, {Name: "verdicts", History: 5}, {Name: "oldsections", History: 1}, {Name: "try-recovery", History: 1}, {Name: "own-build", History: 1}, {Name: index.PackagesState, History: index.PackagesHistory},
if want := []state{{Name: "candidates", History: 10}, {Name: "heard", History: 64}, {Name: "manifests", History: 1}, {Name: "predefined", History: 5}, {Name: "config-candidates", History: 5}, {Name: "taken", History: 1}, {Name: "takeins", History: 1}, {Name: "verdicts", History: 5}, {Name: "oldsections", History: 1}, {Name: index.RunsState, History: 1}, {Name: "own-build", History: 1}, {Name: index.PackagesState, History: index.PackagesHistory},
{Name: views.MarksState, History: 1}, {Name: index.FindingsState, History: 5}, {Name: views.StatusState, History: 1},
{Name: index.RequirementsState, History: 5}, {Name: index.SeatDefinitionsState, History: 5},
{Name: index.PendingState, History: 1}, {Name: index.UnjudgedState, History: 1}, {Name: index.EntryIndexState, History: 1, MaxMiB: 8}, {Name: index.NamedState, History: 1, MaxMiB: 4}, // design 47 G §695
@@ -166,8 +166,8 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
// "Reads FORGE" is a consumes of the indexer's stream: the broker grants only buckets by reads (mesh-control
// internal/broker/state.go:62-93).
// The registry seats' published fill each (§413: no consume of version.numbered) package's series (DLV-I5, design 51 *The versions*).
if want := []string{"go-package-registry.published", "mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported", "mesh-builds.package-published", "mesh-builds.toolchain-agreed", "mesh-conditions.findings-asked", "mesh-delivery-checker.preview.written", "mesh-delivery-checker.try.ended", "mesh-delivery-checker.try.started", "mesh-delivery-checker.verdict", "mesh-delivery-forge-bridge.forge.**",
"mesh-delivery.freed", "mesh-delivery.heads-replaced", "mesh-delivery.pinned", "mesh-delivery.pins-stated", "mesh-delivery.source-moved", "mesh-wants.version-delivered", "mesh-wants.wait-ended", "mesh-wants.wanted", "node-build-go.step-ended", "node-engine.running", "npm-package-registry.published", "operator-channel.decided", "seat:mesh-module-assigner.views.assignments"}; !reflect.DeepEqual(sorted(m.Consumes), want) {
if want := []string{"go-package-registry.published", "mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported", "mesh-builds.package-published", "mesh-builds.toolchain-agreed", "mesh-conditions.findings-asked", "mesh-delivery-checker.verdict", "mesh-delivery-forge-bridge.forge.**",
"mesh-delivery.freed", "mesh-delivery.heads-replaced", "mesh-delivery.pinned", "mesh-delivery.pins-stated", "mesh-delivery.source-moved", "mesh-pipeline-runner.pipeline-ended", "mesh-pipeline-runner.run-started", "mesh-wants.version-delivered", "mesh-wants.wait-ended", "mesh-wants.wanted", "node-build-go.step-ended", "node-engine.running", "npm-package-registry.published", "operator-channel.decided", "seat:mesh-module-assigner.views.assignments"}; !reflect.DeepEqual(sorted(m.Consumes), want) {
t.Errorf("consumes = %v, want %v", m.Consumes, want)
}
if b, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 || b != "${dir:state}/broker" {
@@ -179,7 +179,7 @@ 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{"seat:git.trees", "seat:git.file", "seat:git.contains", "seat:git.open-pulls", "seat:git.branch", "mesh-delivery-checker.checker_tries", "mesh-delivery-forge-bridge.bridge_merged_message", "seat:node-engine.composition", "seat:mesh-delivery.source", "seat:mesh-module-assigner.assignments", "seat:mesh-module-assigner.placement", "seat:mesh-hand-acts.hand-act", "seat:mesh-watchdogs.presence", "seat:node-build-go.steps", "seat:node-build-node.steps", "seat:node-build-image.steps", "mesh-delivery-builder.builder_versions"}; !reflect.DeepEqual(m.Invokes, want) {
if want := []string{"seat:git.trees", "seat:git.file", "seat:git.contains", "seat:git.open-pulls", "seat:git.branch", "mesh-delivery-forge-bridge.bridge_merged_message", "seat:node-engine.composition", "seat:mesh-delivery.source", "seat:mesh-module-assigner.assignments", "seat:mesh-module-assigner.placement", "seat:mesh-hand-acts.hand-act", "seat:mesh-watchdogs.presence", "seat:node-build-go.steps", "seat:node-build-node.steps", "seat:node-build-image.steps", "mesh-delivery-builder.builder_versions"}; !reflect.DeepEqual(m.Invokes, want) {
t.Errorf("invokes = %v, want %v", m.Invokes, want)
}
// 1945: indexer_take_in, served by its tools bundle through the takeins state; the process serves none. The same bundle
@@ -198,12 +198,12 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
}
want = append(want, "indexer_first_views", "indexer_first_view_again", index.ConfigReadersTool, index.BusStateTool, index.RequirementsTool, index.PackageUsersTool, index.SayAcceptedTool, // DLV-F3r, DLV-I10, DLV-I11, DLV-I7, DLV-I8, DLV-I9c
index.SeatDefinitionTool, index.SeatDefinitionsAgainTool, index.OutOfStepTool, // DLV-I19
index.PendingTool, index.TakeInPendingTool, index.DropPendingTool, index.TryCandidatesTool, index.DropTryCandidateTool, index.RecoverTriesTool, index.TakeInMergeTool, index.FingerprintsAgainTool, // issue 1966; 1585/T12, T13 (ADR 0633 §5)
index.PendingTool, index.TakeInPendingTool, index.DropPendingTool, index.TakeInMergeTool, index.FingerprintsAgainTool, // issue 1966; 1585/T12, T13 (ADR 0633 §5)
// milestone 47/G W4.12 (design 47, the indexer's tools): the catalog graph revisions read and steered.
index.CatalogRevisionTool, index.CatalogRevisionsTool, index.CandidateGraphsTool, index.RecomputeTool, index.DropCandidateGraphTool,
index.EntryIndexTool, index.NamedDropTool, // design 47 G §675
index.ProposalsTool, index.ProposalDropTool, index.SeedAgainTool, index.InputsTool, index.PinsTool, index.UnpinTool, index.ConsumerTool) // design 47 G §667, §8, §579
if !reflect.DeepEqual(m.Tools, want) || len(want) != 50 || slices.Contains(m.Tools, "modules") {
if !reflect.DeepEqual(m.Tools, want) || len(want) != 47 || slices.Contains(m.Tools, "modules") {
t.Errorf("tools = %v, want %v", m.Tools, want)
}
// Design 47 G §2, §9: every version of every registration is kept with no age limit, 64 MiB, on the old streams section.
+3 -8
View File
@@ -58,9 +58,6 @@
"indexer_pending",
"indexer_take_in_pending",
"indexer_drop_pending",
"indexer_try_candidates",
"indexer_drop_try_candidate",
"indexer_recover_tries",
"indexer_take_in_merge",
"indexer_fingerprints_again",
"indexer_catalog_revision",
@@ -84,9 +81,6 @@
"mesh-build-node.toolchain.reported",
"mesh-conditions.findings-asked",
"seat:mesh-module-assigner.views.assignments",
"mesh-delivery-checker.preview.written",
"mesh-delivery-checker.try.ended",
"mesh-delivery-checker.try.started",
"mesh-builds.package-published",
"mesh-builds.toolchain-agreed",
"mesh-wants.version-delivered",
@@ -100,6 +94,8 @@
"mesh-delivery.heads-replaced",
"mesh-delivery.pins-stated",
"mesh-wants.wanted",
"mesh-pipeline-runner.run-started",
"mesh-pipeline-runner.pipeline-ended",
"node-engine.running",
"operator-channel.decided",
"npm-package-registry.published"
@@ -114,7 +110,7 @@
{ "name": "takeins", "history": 1 },
{ "name": "verdicts", "history": 5 },
{ "name": "oldsections", "history": 1 },
{ "name": "try-recovery", "history": 1 },
{ "name": "runs", "history": 1 },
{ "name": "own-build", "history": 1 },
{ "name": "packages", "history": 5 },
{ "name": "view-marks", "history": 1 },
@@ -168,7 +164,6 @@
"seat:git.contains",
"seat:git.open-pulls",
"seat:git.branch",
"mesh-delivery-checker.checker_tries",
"mesh-delivery-forge-bridge.bridge_merged_message",
"seat:node-engine.composition",
"seat:mesh-delivery.source",