diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go index 048f59d7..2d127854 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go @@ -873,16 +873,16 @@ func TestIndexerGivesACandidateItsFingerprintWhenTheCheckerCreatesThePreviewsBuc if _, err := js.CreateOrUpdateConsumer(ctx, PreviewsStream, jetstream.ConsumerConfig{Durable: "n1_mesh-delivery-indexer", AckPolicy: jetstream.AckExplicitPolicy}); err != nil { t.Fatal(err) } - source, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.SourceBucket, History: 4}) - if err != nil { - t.Fatal(err) + // 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) } - data, err := meshdeliveryclient.EncodeSource(meshdelivery.SourceByMainValue{Group: "g", Heads: map[string]string{"novox/mesh-catalog": trunk}}) - if err != nil { - t.Fatal(err) - } - if _, err := source.Put(ctx, meshdeliveryclient.SourceMainKey, data); err != nil { - t.Fatal(err) + 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 { + 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, diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go index abd61bd5..01e3af36 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -22,6 +22,7 @@ import ( "syscall" "time" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" @@ -470,22 +471,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro "indexed from it, until the indexer starts again", builds.BuildLogRead, err) } started.Store(true) - // The gate's source bucket (mesh-delivery-gate_source) comes with the gate (release 3e, novox/hq ADR 0450 §1): its key - // main is the mesh's source revision, which stamps the modules (issue 1344). Where it does not exist yet the indexer does - // not wait for it (that needs a timer, ADR 0437 §8): it says so, and is started again with the gate. - sourceErrs := make(chan error, 1) - if source := process.OptionalBucket(ctx, js, index.SourceBucket, "no source revision is read until the indexer starts again", logger.Printf); source != nil { - each := func(e jetstream.KeyValueEntry) { - if opOf(e) { - applySource(ctx, ix, e.Key(), e.Revision(), e.Value(), logger) - } - } - if sw, err := settle(ctx, source, each); err != nil { - process.NotWatched(logger.Printf, index.SourceBucket, err, "no source revision is read until the indexer starts again") - } else { - go func() { sourceErrs <- follow(ctx, sw, each) }() - } - } + // The source revision of a watched branch (design 47 G §15, §115, §362): asked of the seat mesh-delivery's verb source as a + // kept call, once at the start, before the handler of its event source-moved reads anything; a source-moved heard + // meanwhile is read after. + source := &sourceFollow{read: sourceVia(calls.CallKept), apply: ix.ApplySource, logf: logger.Printf} + go source.catchUp(ctx) // Only the toolchain reports of nodes the catalogue places the build module on count (issue 1940, ADR 0549 §3): the // assignments is settled first, so the toolchains settle through the gate already knowing it. gate := buildrun.NewToolchainGate(sortedModules(), ix.PutToolchain, logger.Printf) @@ -668,6 +658,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro }); err != nil { return err } + if err := hub.Handle(meshdelivery.SubjectSourceMovedAll, func(m jetstream.Msg) { source.heard(ctx, m) }); err != nil { + return err + } hub.OnBound(reads.atBind) // what a node's engine said before the consumer was bound (ADR 0668 §4) versions := func(m jetstream.Msg) { if !heardVersion(ctx, ix, m, logger.Printf) { @@ -753,8 +746,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return nil case err := <-errs: return err - case err := <-sourceErrs: - return err case err := <-assignmentsErrs: return err case err := <-tools.errs: diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go index e64afeeb..1450b161 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go @@ -5,7 +5,9 @@ import ( "context" "encoding/json" "errors" + "fmt" "log" + "reflect" "strings" "testing" "time" @@ -16,7 +18,6 @@ import ( "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshwatchdogs" - "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient" "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg" "git.novox.be/novox/mesh-sdk/go/process" @@ -237,24 +238,64 @@ type opener struct{ err error } func (o opener) KeyValue(context.Context, string) (jetstream.KeyValue, error) { return nil, o.err } -func TestTheSourceMainKeyIsTakenAtItsRevision(t *testing.T) { // 1344/AC1 (the watch's side: key, revision, value) - var logs bytes.Buffer - logger := log.New(&logs, "", 0) - mods := &memStore{vals: map[string][]byte{}} - ix := &index.Indexer{Store: &memStore{vals: map[string][]byte{}}, Modules: mods, Builds: index.NewBuilds()} - value, err := meshdeliveryclient.EncodeSource(meshdelivery.SourceByMainValue{Group: "g", Heads: map[string]string{"novox/mesh-delivery": head}}) - if err != nil { - t.Fatal(err) +func TestTheSourceIsReadFromTheSeatAndASourceMovedDuringTheCatchUpIsReadAfterIt(t *testing.T) { // design 47 G §15, §115, §362 + var order []string + revision := int64(4) + var logs []string + var s *sourceFollow + s = &sourceFollow{ + read: func(branch string) (meshdelivery.SourceAnswer, error) { + order = append(order, "read "+branch) + if revision == 0 { + return meshdelivery.SourceAnswer{}, fmt.Errorf("source.main is not readable") + } + if len(order) == 1 { // a source-moved is heard while the catch-up reads + s.moved(context.Background(), "main") + s.moved(context.Background(), "main") + s.moved(context.Background(), "beta") + } + return meshdelivery.SourceAnswer{Branch: branch, Revision: revision}, nil + }, + apply: func(_ context.Context, a meshdelivery.SourceAnswer) error { + order = append(order, fmt.Sprintf("apply %s %d", a.Branch, a.Revision)) + return nil + }, + logf: func(f string, a ...any) { logs = append(logs, fmt.Sprintf(f, a...)) }, } - applySource(context.Background(), ix, "other", 3, value, logger) // not the key of the mesh's source - applySource(context.Background(), ix, meshdeliveryclient.SourceMainKey, 4, []byte(`{"v":1`), logger) - if !strings.Contains(logs.String(), "source") { - t.Errorf("a source that cannot be read was not said: %q", logs.String()) + s.catchUp(context.Background()) + want := []string{"read main", "apply main 4", "read main", "apply main 4"} // once, after the catch-up; beta is left + if !reflect.DeepEqual(order, want) { + t.Errorf("order = %v, want %v", order, want) } - logs.Reset() - applySource(context.Background(), ix, meshdeliveryclient.SourceMainKey, 5, value, logger) - if logs.Len() != 0 { - t.Errorf("a good source said %q", logs.String()) + s.moved(context.Background(), "main") // after the catch-up: read at once + if len(order) != 6 { + t.Errorf("a source-moved after the catch-up was not read at once: %v", order) + } + order, revision = nil, 0 + s.moved(context.Background(), "main") + if len(order) != 1 || len(logs) != 1 || !strings.Contains(logs[0], "not readable") { + t.Errorf("a source that cannot be read: order %v, said %v, want it said and nothing applied", order, logs) + } +} + +func TestTheSeatsSourceAnswerIsReadAndItsErrorIsAnError(t *testing.T) { // design 47 G §115 + var asked any + read := sourceVia(func(key string, body any) (json.RawMessage, error) { + asked = body + if key != "seat:mesh-delivery.source" { + t.Errorf("asked %q", key) + } + if body.(meshdelivery.SourceInput).Branch == "bad" { + return json.RawMessage(`{"error":"source.bad is not readable"}`), nil + } + return json.RawMessage(`{"branch":"main","revision":7,"repositories":[{"repository":"novox/mesh-delivery","head":"` + head + `","trunk":"main"}]}`), nil + }) + a, err := read("main") + if err != nil || a.Revision != 7 || len(a.Repositories) != 1 || asked.(meshdelivery.SourceInput).Branch != "main" { + t.Errorf("source = %+v, %v", a, err) + } + if _, err := read("bad"); err == nil || !strings.Contains(err.Error(), "not readable") { + t.Errorf("an answer with an error was no error: %v", err) } } diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/source.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/source.go index f8aaaad1..10f92615 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/source.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/source.go @@ -2,26 +2,100 @@ package main import ( "context" - "log" + "encoding/json" + "fmt" + "strings" + "sync" - "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient" - - "mesh-delivery-indexer/internal/index" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery" + "github.com/nats-io/nats.go/jetstream" ) -// applySource takes one put of the gate's bucket source into the indexer: the key main, at its revision, which is the mesh's -// source revision (mesh-sdk go/delivery SourceMainKey, Source). Any other key is not the mesh's source. A value that is no -// source is said; so is a write that failed, and it is tried again only at the next revision (novox/hq ADR 0437 §8). -func applySource(ctx context.Context, ix *index.Indexer, key string, rev uint64, value []byte, logger *log.Logger) { - if key != meshdeliveryclient.SourceMainKey { +// sourceInvoke is the seat mesh-delivery's verb source as the manifest's invokes names it: a watched branch's source +// revision, asked of the seat and never read from the gate's state (design 47 G §15, §115; ADR 0596 §2). +const sourceInvoke = "seat:" + meshdelivery.Seat + "." + meshdelivery.VerbSource + +// sourceVia asks the seat for one branch's source revision through call (process.Conn.CallKept, a read) and reads its +// declared answer; an answer that carries an error is that error. +func sourceVia(call toolCaller) func(branch string) (meshdelivery.SourceAnswer, error) { + return func(branch string) (meshdelivery.SourceAnswer, error) { + raw, err := call(sourceInvoke, meshdelivery.SourceInput{Branch: branch}) + if err != nil { + return meshdelivery.SourceAnswer{}, err + } + var a meshdelivery.SourceAnswer + if err := json.Unmarshal(raw, &a); err != nil { + return a, fmt.Errorf("the answer of %s is not a source revision: %w", sourceInvoke, err) + } + if a.Error != "" { + return a, fmt.Errorf("%s: %s", sourceInvoke, a.Error) + } + return a, nil + } +} + +// sourceFollow takes the seat's source revisions into the indexer: first the catch-up of the heads the source names, then +// each source-moved the seat says (design 47 G §362). A source-moved heard before the catch-up ended is applied after it, +// once per branch. +type sourceFollow struct { + read func(branch string) (meshdelivery.SourceAnswer, error) + apply func(context.Context, meshdelivery.SourceAnswer) error + logf func(string, ...any) + + mu sync.Mutex + caught bool + waiting map[string]bool +} + +// catchUp reads the trunk branch's source revision and applies it, then the branches a source-moved named meanwhile. A +// read that fails is said; the catch-up ends all the same, and the next source-moved reads again. +func (s *sourceFollow) catchUp(ctx context.Context) { + s.take(ctx, trunkBranch) + s.mu.Lock() + s.caught = true + branches := s.waiting + s.waiting = nil + s.mu.Unlock() + for branch := range branches { + s.take(ctx, branch) + } +} + +// moved is the handler of one source-moved on branch: held until the catch-up ends, read at once after. +func (s *sourceFollow) moved(ctx context.Context, branch string) { + s.mu.Lock() + if !s.caught { + if s.waiting == nil { + s.waiting = map[string]bool{} + } + s.waiting[branch] = true + s.mu.Unlock() return } - s, err := meshdeliveryclient.DecodeSource(value) + s.mu.Unlock() + s.take(ctx, branch) +} + +// take reads one branch's source revision and applies it. A branch other than the trunk's is left until the indexer keeps +// a catalog graph per watched branch. +func (s *sourceFollow) take(ctx context.Context, branch string) { + if branch != trunkBranch { + return + } + a, err := s.read(branch) if err != nil { - logger.Printf("source.%s at revision %d is not a source, and is left: %v", key, rev, err) + s.logf("source.%s is not read, and is read again at its next source-moved: %v", branch, err) return + } + if err := s.apply(ctx, a); err != nil { + s.logf("source.%s at revision %d could not be taken into the modules: %v", branch, a.Revision, err) } - if err := ix.ApplySource(ctx, rev, s); err != nil { - logger.Printf("source.%s at revision %d could not be taken into the modules: %v", key, rev, err) +} + +// heard is the hub's handler of source-moved: the branch is the last token of the subject. The message is acknowledged at once. +func (s *sourceFollow) heard(ctx context.Context, m jetstream.Msg) { + defer func() { _ = m.Ack() }() + if branch := m.Subject()[strings.LastIndex(m.Subject(), ".")+1:]; branch != "" { + s.moved(ctx, branch) } } diff --git a/modules/mesh-delivery-indexer/internal/index/inputs_test.go b/modules/mesh-delivery-indexer/internal/index/inputs_test.go index 74763916..77adda62 100644 --- a/modules/mesh-delivery-indexer/internal/index/inputs_test.go +++ b/modules/mesh-delivery-indexer/internal/index/inputs_test.go @@ -144,7 +144,7 @@ func TestAnArtifactOfNoToolchainHasNoLanguageAndNoLock(t *testing.T) { // 1450/A func TestAManifestTheParserRefusesLeavesNoInputsAndSaysWhy(t *testing.T) { // 1450/AC5 for name, raw := range map[string]string{ "not json": "not json", - "unknown field": strings.Replace(catalogueManifest(t, "avahi"), `"version":"1.0",`, `"version":"1.0", "scope": "everything",`, 1), + "unknown field": strings.Replace(catalogueManifest(t, "avahi"), `"version": "1",`, `"version": "1", "scope": "everything",`, 1), } { in := inputsOf("avahi", raw, "t", nil, noBases) if in.Kept != nil || !strings.HasPrefix(in.None2, "the manifest cannot be read: ") || len(in.None2) <= len("the manifest cannot be read: ") { diff --git a/modules/mesh-delivery-indexer/internal/index/modules_test.go b/modules/mesh-delivery-indexer/internal/index/modules_test.go index 1d3b78fe..9a4d6fc7 100644 --- a/modules/mesh-delivery-indexer/internal/index/modules_test.go +++ b/modules/mesh-delivery-indexer/internal/index/modules_test.go @@ -14,7 +14,6 @@ import ( "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" - "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg" "git.novox.be/novox/mesh-delivery/sdk/trunk" @@ -160,7 +159,7 @@ func TestAnUnchangedModuleWritesNothing(t *testing.T) { // 1239 (unchanged value if err := ix.Apply(context.Background(), m); err != nil { return err } - if err := ix.ApplySource(context.Background(), 8, sourceOf(repo, next)); err != nil { + if err := applyS(ix, context.Background(), 8, sourceOf(repo, next)); err != nil { return err } ix.Taken = on @@ -169,7 +168,7 @@ func TestAnUnchangedModuleWritesNothing(t *testing.T) { // 1239 (unchanged value {name: "a source head whose merge was not applied is not a taken key", writes: 0, commit: map[string]string{"a": merge, "b": merge, "c": merge}, taken: "", then: func(ix *Indexer) error { // the forge stream still holds the merge: the consumer is not drained delete(ix.Taken.(*memStore).vals, takenKey(repo)) - if err := ix.ApplySource(context.Background(), 8, sourceOf(repo, next)); err != nil { + if err := applyS(ix, context.Background(), 8, sourceOf(repo, next)); err != nil { return err } return ix.FillTaken(context.Background()) @@ -180,7 +179,7 @@ func TestAnUnchangedModuleWritesNothing(t *testing.T) { // 1239 (unchanged value {name: "a source head of a drained forge stream whose merge was never heard leaves the key", writes: 0, commit: map[string]string{"a": merge, "b": merge, "c": merge}, taken: merge, then: func(ix *Indexer) error { ix.Contains = func(_ context.Context, _, commit, in string) (bool, error) { return commit == merge && in == next, nil } - if err := ix.ApplySource(context.Background(), 8, sourceOf(repo, next)); err != nil { + if err := applyS(ix, context.Background(), 8, sourceOf(repo, next)); err != nil { return err } return ix.FillTaken(context.Background()) @@ -188,7 +187,7 @@ func TestAnUnchangedModuleWritesNothing(t *testing.T) { // 1239 (unchanged value {name: "a source head of a drained forge stream fills no missing key", writes: 0, commit: map[string]string{"a": merge, "b": merge, "c": merge}, taken: "", then: func(ix *Indexer) error { delete(ix.Taken.(*memStore).vals, takenKey(repo)) - if err := ix.ApplySource(context.Background(), 8, sourceOf(repo, next)); err != nil { + if err := applyS(ix, context.Background(), 8, sourceOf(repo, next)); err != nil { return err } return ix.FillTaken(context.Background()) @@ -300,7 +299,7 @@ func TestAnUnchangedModuleWritesNothing(t *testing.T) { // 1239 (unchanged value }}, {name: "a later source revision of another repository's merge", writes: 0, commit: map[string]string{"a": merge, "b": merge, "c": merge}, taken: merge, then: func(ix *Indexer) error { - return ix.ApplySource(context.Background(), 9, meshdelivery.SourceByMainValue{Group: "g", Heads: map[string]string{repo: merge, "novox/mesh-catalog": otherMerge}}) + return applyS(ix, context.Background(), 9, srcT{Group: "g", Heads: map[string]string{repo: merge, "novox/mesh-catalog": otherMerge}}) }}, } { t.Run(tc.name, func(t *testing.T) { @@ -311,7 +310,7 @@ func TestAnUnchangedModuleWritesNothing(t *testing.T) { // 1239 (unchanged value if err := ix.Apply(ctx, mergedMsg(repo, three)); err != nil { t.Fatal(err) } - if err := ix.ApplySource(ctx, 7, sourceOf(repo, merge)); err != nil { + if err := applyS(ix, ctx, 7, sourceOf(repo, merge)); err != nil { t.Fatal(err) } writes := mods.writes @@ -379,7 +378,7 @@ func TestAMergeWritesTheModulesDirectoryAndKeepsItThroughSourceAndRecompute(t *t t.Errorf("after the merge %s = %+v (%v), want path %q", module, m, ok, want) } } - if err := ix.ApplySource(ctx, 7, sourceOf("novox/mesh-delivery", merge)); err != nil { + if err := applyS(ix, ctx, 7, sourceOf("novox/mesh-delivery", merge)); err != nil { t.Fatal(err) } if err := ix.Recompute(ctx); err != nil { diff --git a/modules/mesh-delivery-indexer/internal/index/names.go b/modules/mesh-delivery-indexer/internal/index/names.go index 6d950393..e3d860d9 100644 --- a/modules/mesh-delivery-indexer/internal/index/names.go +++ b/modules/mesh-delivery-indexer/internal/index/names.go @@ -40,7 +40,6 @@ const ( AssignmentsState = "assignments" AssignmentsBucket = AssignerModule + "_" + AssignmentsState AssignmentsHistory = 5 - SourceBucket = GateModule + "_source" PreviewsBucket = CheckerModule + "_previews" PreviewsHistory = 5 toolchainsState = "toolchains" diff --git a/modules/mesh-delivery-indexer/internal/index/previews_test.go b/modules/mesh-delivery-indexer/internal/index/previews_test.go index 26201cd4..efd893b2 100644 --- a/modules/mesh-delivery-indexer/internal/index/previews_test.go +++ b/modules/mesh-delivery-indexer/internal/index/previews_test.go @@ -17,7 +17,6 @@ import ( "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" - "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery" "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdeliverychecker" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-checker/checkermsg" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg" @@ -39,7 +38,7 @@ func previewBed(t *testing.T, report bool) (*Indexer, *memStore, *memStore) { if report { reportToolchain(t, ix, "mesh-build-go", "novox", "go", "go1.27.1") } - if err := ix.ApplySource(context.Background(), 5, meshdelivery.SourceByMainValue{Group: "g", Heads: map[string]string{catalogueRepo: trunkNow}}); err != nil { + if err := applyS(ix, context.Background(), 5, srcT{Group: "g", Heads: map[string]string{catalogueRepo: trunkNow}}); err != nil { t.Fatal(err) } return ix, mods, ix.Store.(*memStore) @@ -205,7 +204,7 @@ func TestAPreviewMadeOntoAnOldTrunkIsOutOfDate(t *testing.T) { // 1445/AC3 if got := fpOf(t, cands); got.Fp2 == "" { t.Fatalf("after the right preview: %+v", got) } - if err := ix.ApplySource(context.Background(), 6, meshdelivery.SourceByMainValue{Group: "g", Heads: map[string]string{catalogueRepo: "8888888888888888888888888888888888888888"}}); err != nil { + if err := applyS(ix, context.Background(), 6, srcT{Group: "g", Heads: map[string]string{catalogueRepo: "8888888888888888888888888888888888888888"}}); err != nil { t.Fatal(err) } if got := fpOf(t, cands); got.None2 != NonePreviewStale { diff --git a/modules/mesh-delivery-indexer/internal/index/registered_test.go b/modules/mesh-delivery-indexer/internal/index/registered_test.go index 6492329e..c2437640 100644 --- a/modules/mesh-delivery-indexer/internal/index/registered_test.go +++ b/modules/mesh-delivery-indexer/internal/index/registered_test.go @@ -9,7 +9,6 @@ import ( "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" - "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery" "git.novox.be/novox/mesh-sdk/go/keys" ) @@ -261,7 +260,7 @@ func TestAForgeThatCannotSayLeavesTheModuleUnwrittenAndAsksAgain(t *testing.T) { t.Fatal(err) } case "a new source revision": - if err := ix.ApplySource(ctx, 2, meshdelivery.SourceByMainValue{Heads: map[string]string{"jschoubben/snake": h2, "novox/other": h1}}); err != nil { + if err := applyS(ix, ctx, 2, srcT{Heads: map[string]string{"jschoubben/snake": h2, "novox/other": h1}}); err != nil { t.Fatal(err) } ix.retryWG.Wait() // read again in the background @@ -281,7 +280,7 @@ func TestAForgeThatCannotSayLeavesTheModuleUnwrittenAndAsksAgain(t *testing.T) { } // Taken in, it is read no more: a further event asks the forge nothing. n := len(asked) - if err := ix.ApplySource(ctx, 3, meshdelivery.SourceByMainValue{Heads: map[string]string{"jschoubben/snake": h2}}); err != nil { + if err := applyS(ix, ctx, 3, srcT{Heads: map[string]string{"jschoubben/snake": h2}}); err != nil { t.Fatal(err) } ix.retryWG.Wait() diff --git a/modules/mesh-delivery-indexer/internal/index/remove_test.go b/modules/mesh-delivery-indexer/internal/index/remove_test.go index 8c1b5042..e5864ca2 100644 --- a/modules/mesh-delivery-indexer/internal/index/remove_test.go +++ b/modules/mesh-delivery-indexer/internal/index/remove_test.go @@ -100,7 +100,7 @@ func TestAPathWithoutAModuleJSONIsNeverAModule(t *testing.T) { // 1900/AC2: test func TestStartingDeletesTheKeysTheTrunkDoesNotHold(t *testing.T) { // 1900/AC3 ix, mods := twoModules(t) ctx := context.Background() - if err := ix.ApplySource(ctx, 3, sourceOf(catalogueRepo, merge)); err != nil { + if err := applyS(ix, ctx, 3, sourceOf(catalogueRepo, merge)); err != nil { t.Fatal(err) } var asked string @@ -126,7 +126,7 @@ func TestStartingDeletesTheKeysTheTrunkDoesNotHold(t *testing.T) { // 1900/AC3 func TestStartingDeletesNothingWhenTheTrunkCannotBeRead(t *testing.T) { // 1900/AC3: failure is loud, never a mass delete ix, mods := twoModules(t) ctx := context.Background() - if err := ix.ApplySource(ctx, 3, sourceOf(catalogueRepo, merge)); err != nil { + if err := applyS(ix, ctx, 3, sourceOf(catalogueRepo, merge)); err != nil { t.Fatal(err) } for name, trunk := range map[string]func(context.Context, string, string) ([]string, error){ @@ -146,7 +146,7 @@ func TestAKeyWrittenAheadOfTheSourceHeadStaysAndOneWithoutAPathIsSkipped(t *test ix, mods := twoModules(t) // written at the merge commit C ctx := context.Background() const older = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb" - if err := ix.ApplySource(ctx, 3, sourceOf(catalogueRepo, older)); err != nil { + if err := applyS(ix, ctx, 3, sourceOf(catalogueRepo, older)); err != nil { t.Fatal(err) } mods.put("nopath", mustEncode(t, meshcatalog.ModulesByModuleValue{Repository: catalogueRepo, Commit: older, None2: "x", Manifest: strings.Repeat("a", 64)})) diff --git a/modules/mesh-delivery-indexer/internal/index/source.go b/modules/mesh-delivery-indexer/internal/index/source.go index 2b67be36..62123afe 100644 --- a/modules/mesh-delivery-indexer/internal/index/source.go +++ b/modules/mesh-delivery-indexer/internal/index/source.go @@ -11,8 +11,9 @@ import ( "git.novox.be/novox/mesh-delivery/sdk/trunk" ) -// The gate's source revision (novox/hq ADR 0437 §5, ADR 0450 §1): source.main names every repository's merged head, and its -// revision in the bucket is the mesh's source revision. The indexer keeps the latest it has seen. A module is stamped +// The gate's source revision of a watched branch (design 47 G §15, §115), read through the seat mesh-delivery's verb source +// and its event source-moved, never the gate's state: it names every repository's merged head, and its revision is the +// mesh's source revision of that branch. The indexer keeps the latest it has seen. A module is stamped // with it (Module.Src) when the head the source names for its repository is the merge commit the module's fingerprint // was computed at: from the modules bucket when the merge was heard first, and when the merge is heard after the source. @@ -33,12 +34,17 @@ func (ix *Indexer) srcOf(repository, merge string) uint64 { return 0 } -// ApplySource takes a revision of the gate's source.main in. An older revision than the latest seen is left. For each +// ApplySource takes the source revision the verb source answered in. An older revision than the latest seen is left. For each // repository the source names, each module built from it whose fingerprint was computed at the head the source names, and // that carries no Src yet, is written again with Src the revision (once: a later revision leaves it); a module computed at another commit is left (a merge heard later is stamped // when it is written, applyMerged), and a repository none of the indexer's modules is built from writes nothing. The // modules of a repository are those the modules bucket holds of it (ADR 0501 clause 6). -func (ix *Indexer) ApplySource(ctx context.Context, rev uint64, s meshdelivery.SourceByMainValue) error { +func (ix *Indexer) ApplySource(ctx context.Context, a meshdelivery.SourceAnswer) error { + rev := uint64(a.Revision) + s := struct{ Heads map[string]string }{Heads: map[string]string{}} + for _, r := range a.Repositories { + s.Heads[r.Repository] = r.Head + } ix.kickJudge(ctx) ix.mu.Lock() if rev < ix.source.rev { diff --git a/modules/mesh-delivery-indexer/internal/index/source_test.go b/modules/mesh-delivery-indexer/internal/index/source_test.go index 0bed051a..9fc5b906 100644 --- a/modules/mesh-delivery-indexer/internal/index/source_test.go +++ b/modules/mesh-delivery-indexer/internal/index/source_test.go @@ -10,8 +10,20 @@ import ( const otherMerge = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb" -func sourceOf(repo, head string) meshdelivery.SourceByMainValue { - return meshdelivery.SourceByMainValue{Group: "g", Heads: map[string]string{repo: head}} +// srcT is a source revision's heads as the tests write them; applyS says it as the verb source answers it. +type srcT struct { + Group string + Heads map[string]string +} + +func sourceOf(repo, head string) srcT { return srcT{Group: "g", Heads: map[string]string{repo: head}} } + +func applyS(ix *Indexer, ctx context.Context, rev uint64, s srcT) error { + a := meshdelivery.SourceAnswer{Branch: "main", Group: s.Group, Revision: int64(rev)} + for r, h := range s.Heads { + a.Repositories = append(a.Repositories, meshdelivery.SourceHead{Repository: r, Head: h, Trunk: "main"}) + } + return ix.ApplySource(ctx, a) } func TestASourceRevisionStampsTheModulesOfTheRepositoriesHeadItNames(t *testing.T) { // 1344/AC1 @@ -25,8 +37,8 @@ func TestASourceRevisionStampsTheModulesOfTheRepositoriesHeadItNames(t *testing. t.Fatalf("module before the source = %+v, want Src 0", before) } // The source names the head merged for the delivery repository, and another repository's head no module of ours is built from. - s := meshdelivery.SourceByMainValue{Group: "g", Heads: map[string]string{"novox/mesh-delivery": merge, "novox/mesh-catalog": otherMerge}} - if err := ix.ApplySource(ctx, 7, s); err != nil { + s := srcT{Group: "g", Heads: map[string]string{"novox/mesh-delivery": merge, "novox/mesh-catalog": otherMerge}} + if err := applyS(ix, ctx, 7, s); err != nil { t.Fatal(err) } m, ok := moduleOf(t, mods, "a") @@ -39,10 +51,10 @@ func TestASourceRevisionStampsTheModulesOfTheRepositoriesHeadItNames(t *testing. t.Errorf("module c = %+v: written for a head no merge message of the indexer carried", c) } // The same revision again changes nothing, and so does a later one: Src is the first revision naming the commit. - if err := ix.ApplySource(ctx, 7, s); err != nil { + if err := applyS(ix, ctx, 7, s); err != nil { t.Fatal(err) } - if err := ix.ApplySource(ctx, 9, s); err != nil { + if err := applyS(ix, ctx, 9, s); err != nil { t.Fatal(err) } if m, _ := moduleOf(t, mods, "a"); m.Src != 7 { @@ -56,15 +68,15 @@ func TestAMergeHeardAfterTheSourceNamesItsHeadIsStampedWhenWritten(t *testing.T) ix, mods := modulesIndexer(t, rec(t, "build-1", "a", "novox/mesh-delivery", "modules/a", "main", "", "2026-10-19T09:00:00Z", aInputs)) ctx := context.Background() _ = ix.Apply(ctx, mergedMsg("novox/mesh-delivery", map[string]string{"modules/a": "newtree"})) - _ = ix.ApplySource(ctx, 9, sourceOf("novox/mesh-delivery", merge)) - _ = ix.ApplySource(ctx, 7, sourceOf("novox/mesh-delivery", merge)) + _ = applyS(ix, ctx, 9, sourceOf("novox/mesh-delivery", merge)) + _ = applyS(ix, ctx, 7, sourceOf("novox/mesh-delivery", merge)) if m, _ := moduleOf(t, mods, "a"); m.Src != 9 { t.Errorf("module a Src = %d, want 9", m.Src) } }) ix, mods := modulesIndexer(t, rec(t, "build-1", "a", "novox/mesh-delivery", "modules/a", "main", "", "2026-10-19T09:00:00Z", aInputs)) ctx := context.Background() - if err := ix.ApplySource(ctx, 7, sourceOf("novox/mesh-delivery", merge)); err != nil { + if err := applyS(ix, ctx, 7, sourceOf("novox/mesh-delivery", merge)); err != nil { t.Fatal(err) } if _, ok := moduleOf(t, mods, "a"); ok { @@ -89,7 +101,7 @@ func TestAMergeThatNoSourceRevisionNamesIsWrittenAsToday(t *testing.T) { // 1344 t.Fatalf("module a = %+v (%v), want the merge written with Src 0", m, ok) } // A source revision that names another head of the repository: a merge of the head it does not name keeps Src 0. - if err := ix.ApplySource(ctx, 7, sourceOf("novox/mesh-delivery", otherMerge)); err != nil { + if err := applyS(ix, ctx, 7, sourceOf("novox/mesh-delivery", otherMerge)); err != nil { t.Fatal(err) } if m, _ := moduleOf(t, mods, "a"); m.Src != 0 { diff --git a/modules/mesh-delivery-indexer/internal/index/stale_none2_test.go b/modules/mesh-delivery-indexer/internal/index/stale_none2_test.go index 6ec511d1..e92d94b7 100644 --- a/modules/mesh-delivery-indexer/internal/index/stale_none2_test.go +++ b/modules/mesh-delivery-indexer/internal/index/stale_none2_test.go @@ -10,7 +10,7 @@ import ( // not read. func unreadableAvahi(t *testing.T) string { t.Helper() - raw := strings.Replace(catalogueManifest(t, "avahi"), `"version":"1.0",`, `"version":"1.0", "scope": "everything",`, 1) + raw := strings.Replace(catalogueManifest(t, "avahi"), `"version": "1",`, `"version": "1", "scope": "everything",`, 1) if raw == catalogueManifest(t, "avahi") { t.Fatal("the fixture has no version line") } diff --git a/modules/mesh-delivery-indexer/internal/index/toolchains_test.go b/modules/mesh-delivery-indexer/internal/index/toolchains_test.go index 16eb83ce..17667eef 100644 --- a/modules/mesh-delivery-indexer/internal/index/toolchains_test.go +++ b/modules/mesh-delivery-indexer/internal/index/toolchains_test.go @@ -15,7 +15,6 @@ import ( "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" - "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg" sdkstep "git.novox.be/novox/mesh-delivery/sdk/buildstep" @@ -38,7 +37,7 @@ func renamed(t *testing.T, from, to string) string { if out == raw { t.Fatalf("%s's manifest has no module line", from) } - return out + return strings.Replace(out, `"version": "1",`, `"version": "1.0",`, 1) // the validator reads MAJOR.MINOR only (design 47 G §368) } func standingOn(t *testing.T, raw, base string) string { @@ -150,8 +149,8 @@ func TestAgreedToolchainChangesAGoModulesReasonWithNoMerge(t *testing.T) { // 14 } // The release that last merged avahi: its source revision names avahi's commit, so Src (and FPSrc) are 7. merged, _ := moduleOf(t, mods, "avahi") - src := meshdelivery.SourceByMainValue{Group: "g", Heads: map[string]string{merged.Repository: merged.Commit}} - if err := ix.ApplySource(context.Background(), 7, src); err != nil { + src := srcT{Group: "g", Heads: map[string]string{merged.Repository: merged.Commit}} + if err := applyS(ix, context.Background(), 7, src); err != nil { t.Fatal(err) } if m, _ := moduleOf(t, mods, "avahi"); m.Src != 7 || m.FpSrc != 7 { @@ -183,7 +182,7 @@ func TestAgreedToolchainChangesAGoModulesReasonWithNoMerge(t *testing.T) { // 14 if m.FpSrc != 0 { t.Errorf("a fingerprint moved by a recompute: fp-src %d, want 0", m.FpSrc) } - if err := ix.ApplySource(context.Background(), 8, src); err != nil { + if err := applyS(ix, context.Background(), 8, src); err != nil { t.Fatal(err) } if again, _ := moduleOf(t, mods, "avahi"); again.FpSrc != 0 { @@ -344,7 +343,7 @@ func TestBasesAreComputedFirstAndNoFingerprintTakesTheBuilder(t *testing.T) { // // verb edges of two modules calling each other, from either side; a third calling one of them keeps its edge. invoking := func(name, invokes string) mod { raw := renamed(t, "avahi", name) - return mod{name, strings.Replace(raw, `"version":"1.0",`, `"version":"1.0", "invokes": [`+invokes+`],`, 1)} + return mod{name, strings.Replace(raw, `"version": "1.0",`, `"version": "1.0", "invokes": [`+invokes+`],`, 1)} } vx, _ := newFormIndexer(t, invoking("caller", `"callee.avahi_status"`), invoking("callee", `"caller.avahi_browse", "*"`), invoking("third", `"callee.avahi_status"`)) diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index 09028dc2..cb7db957 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -130,14 +130,14 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build t.Errorf("the indexer names its bucket %s with %d revisions, its manifest's states are %+v", bucket, history, m.State) } } - for bucket, read := range map[string]string{index.SourceBucket: "mesh-delivery-gate.source", index.PreviewsBucket: "mesh-delivery-checker.previews", + for bucket, read := range map[string]string{index.PreviewsBucket: "mesh-delivery-checker.previews", index.ToolchainsBucket(index.BuildGoModule): "mesh-build-go.toolchains", index.ToolchainsBucket(index.BuildNodeModule): "mesh-build-node.toolchains"} { if !slices.Contains(m.Reads, read) || strings.Replace(read, ".", "_", 1) != bucket { t.Errorf("the indexer reads the bucket %s, its manifest's reads are %v (want %s)", bucket, m.Reads, read) } } // DLV-F1r (ADR 0615 §1): the assigner's assignments is read through its first view and views family, never its state. - if want := []string{"mesh-build-go.toolchains", "mesh-build-node.toolchains", "mesh-delivery-builder.build-log", "mesh-delivery-checker.previews", "mesh-delivery-gate.source"}; !reflect.DeepEqual(sorted(m.Reads), want) { + if want := []string{"mesh-build-go.toolchains", "mesh-build-node.toolchains", "mesh-delivery-builder.build-log", "mesh-delivery-checker.previews"}; !reflect.DeepEqual(sorted(m.Reads), want) { t.Errorf("reads = %v, want %v", m.Reads, want) } // "Reads FORGE" is a consumes of the indexer's stream: the broker grants only buckets by reads (mesh-control @@ -145,7 +145,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build // The registry seats' published and the builder's numbered fill each 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-delivery-assigner.views.assignments", "mesh-delivery-builder.version.numbered", "mesh-delivery-checker.preview.written", "mesh-delivery-checker.try.ended", "mesh-delivery-checker.try.started", "mesh-delivery-checker.verdict", "mesh-delivery-forge-bridge.forge.**", - "node-engine.manifest-reads-changed", "npm-package-registry.published"}; !reflect.DeepEqual(sorted(m.Consumes), want) { // ADR 0632 §2: the counted engines' reads + "mesh-delivery.source-moved", "node-engine.manifest-reads-changed", "npm-package-registry.published"}; !reflect.DeepEqual(sorted(m.Consumes), want) { // ADR 0632 §2: the counted engines' reads t.Errorf("consumes = %v, want %v", m.Consumes, want) } if b, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 || b != "${dir:state}/broker" { @@ -157,7 +157,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{"mesh-delivery-assigner.assignments", "gitea.gitea_commit_trees", "gitea.gitea_contains", "gitea.gitea_get_file", "gitea.gitea_open_pulls", "mesh-delivery-checker.checker_tries", "mesh-delivery-forge-bridge.bridge_merged_message", "seat:node-engine.composition", "seat:node-engine.manifest-reads", "seat:mesh-hand-acts.hand-act", "seat:mesh-watchdogs.presence"}; !reflect.DeepEqual(m.Invokes, want) { + if want := []string{"mesh-delivery-assigner.assignments", "gitea.gitea_commit_trees", "gitea.gitea_contains", "gitea.gitea_get_file", "gitea.gitea_open_pulls", "mesh-delivery-checker.checker_tries", "mesh-delivery-forge-bridge.bridge_merged_message", "seat:node-engine.composition", "seat:node-engine.manifest-reads", "seat:mesh-delivery.source", "seat:mesh-hand-acts.hand-act", "seat:mesh-watchdogs.presence"}; !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 diff --git a/modules/mesh-delivery-indexer/module.json b/modules/mesh-delivery-indexer/module.json index c820abfc..0392d211 100644 --- a/modules/mesh-delivery-indexer/module.json +++ b/modules/mesh-delivery-indexer/module.json @@ -69,6 +69,7 @@ "mesh-delivery-checker.try.started", "mesh-delivery-checker.verdict", "mesh-delivery-forge-bridge.forge.**", + "mesh-delivery.source-moved", "node-engine.manifest-reads-changed", "npm-package-registry.published" ], @@ -118,8 +119,7 @@ "mesh-build-go.toolchains", "mesh-build-node.toolchains", "mesh-delivery-builder.build-log", - "mesh-delivery-checker.previews", - "mesh-delivery-gate.source" + "mesh-delivery-checker.previews" ], "invokes": [ "mesh-delivery-assigner.assignments", @@ -131,6 +131,7 @@ "mesh-delivery-forge-bridge.bridge_merged_message", "seat:node-engine.composition", "seat:node-engine.manifest-reads", + "seat:mesh-delivery.source", "seat:mesh-hand-acts.hand-act", "seat:mesh-watchdogs.presence" ], diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main.go new file mode 100644 index 00000000..bebc4d19 --- /dev/null +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main.go @@ -0,0 +1,91 @@ +// Command mesh-delivery-pipeline-tools serves the seat mesh-pipeline-runner's verbs as tools named . +// (design 47 G §83, §306), answered from the runner's own states; the tool runner describes each from the seat's +// definition, so the tools carry no description or input. +package main + +import ( + "encoding/json" + "fmt" + "os" + + seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner" + stdio "git.novox.be/novox/mesh-sdk/go" + + "mesh-delivery-pipeline/internal/pipeline" +) + +// kept reads one of the runner's states through the runtime. +type kept struct{ s stdio.KeptState } + +func (k kept) Keys() ([]string, error) { return k.s.Keys() } + +func (k kept) Value(key string) (json.RawMessage, error) { + e, err := k.s.Get(key) + if err != nil || e == nil { + return nil, err + } + return e.Value, nil +} + +// as decodes a call's arguments into a verb's input, refusing one that does not fit it. +func as[T any](args map[string]any) (T, error) { + var in T + raw, err := json.Marshal(args) + if err == nil { + err = json.Unmarshal(raw, &in) + } + if err != nil { + return in, fmt.Errorf("the arguments do not fit the verb's input: %w", err) + } + return in, nil +} + +// Tools are the runner's read verbs over its states pipelines and kept. +func Tools(r pipeline.Reads) []stdio.Tool { + runs := map[string]func(map[string]any) (any, error){ + seat.VerbShow: func(a map[string]any) (any, error) { + in, err := as[seat.ShowInput](a) + if err != nil { + return nil, err + } + return r.Show(in) + }, + seat.VerbList: func(a map[string]any) (any, error) { + in, err := as[seat.ListInput](a) + if err != nil { + return nil, err + } + return r.List(in) + }, + seat.VerbStands: func(map[string]any) (any, error) { return r.Stands() }, + seat.VerbWhyCarried: func(a map[string]any) (any, error) { + in, err := as[seat.WhyCarriedInput](a) + if err != nil { + return nil, err + } + return r.WhyCarried(in) + }, + seat.VerbKept: func(a map[string]any) (any, error) { + in, err := as[seat.KeptInput](a) + if err != nil { + return nil, err + } + return r.KeptResults(in) + }, + } + var out []stdio.Tool + for _, v := range seat.Verbs() { + if run, ok := runs[v]; ok { + out = append(out, stdio.Tool{Name: seat.Seat + "." + v, Run: run}) + } + } + return out +} + +func main() { + r := pipeline.Reads{Pipelines: kept{stdio.State(seat.StatePipelines)}, Kept: kept{stdio.State(seat.StateKept)}} + if err := stdio.Serve("", Tools(r)); err != nil { + fmt.Fprintf(os.Stderr, "[mesh-delivery-pipeline-tools] %v\n", err) + os.Exit(1) + } +} diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main_test.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main_test.go new file mode 100644 index 00000000..0565790a --- /dev/null +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main_test.go @@ -0,0 +1,26 @@ +package main + +import ( + "encoding/json" + "strings" + "testing" + + "mesh-delivery-pipeline/internal/pipeline" +) + +type none struct{} + +func (none) Keys() ([]string, error) { return nil, nil } +func (none) Value(string) (json.RawMessage, error) { return nil, nil } + +// TestTools: the bundle serves the five read verbs as . (§83, §306; RN15/AC1). +func TestTools(t *testing.T) { + var names []string + for _, tl := range Tools(pipeline.Reads{Pipelines: none{}, Kept: none{}}) { + names = append(names, tl.Name) + } + want := "mesh-pipeline-runner.show mesh-pipeline-runner.list mesh-pipeline-runner.stands mesh-pipeline-runner.why-carried mesh-pipeline-runner.kept" + if got := strings.Join(names, " "); got != want { + t.Errorf("tools %q, want %q", got, want) + } +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/read.go b/modules/mesh-delivery-pipeline/internal/pipeline/read.go new file mode 100644 index 00000000..00163110 --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/read.go @@ -0,0 +1,191 @@ +// Package pipeline is the pipeline runner's own logic, holder of the seat mesh-pipeline-runner (design 47 G §21). It names +// no module, branch or repository (§7): every name comes from what it reads. +package pipeline + +import ( + "encoding/json" + "fmt" + "sort" + "strings" + "time" + + seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner" +) + +// Kept is one of the runner's old state sections as it reads them: every key, and one key's value (nil when absent). +type Kept interface { + Keys() ([]string, error) + Value(key string) (json.RawMessage, error) +} + +// Reads answers the runner's read verbs (§83, §306) from its own states pipelines and kept, never another part's. +type Reads struct { + Pipelines Kept + Kept Kept +} + +// Show is one pipeline whole, with its steps and the names of their outputs in step-outputs (§83). +func (r Reads) Show(in seat.ShowInput) (seat.ShowAnswer, error) { + p, ok, err := r.pipeline(in.Pipeline) + if err != nil { + return seat.ShowAnswer{}, err + } + if !ok { + return seat.ShowAnswer{}, fmt.Errorf("no pipeline %q in %s", in.Pipeline, seat.StatePipelines) + } + a := seat.ShowAnswer{Pipeline: p} + for _, s := range p.Steps { + if s.Output != "" { + a.Outputs = append(a.Outputs, s.Output) + } + } + return a, nil +} + +// List is the pipelines and runs of a module, a branch, a commit or a run, newest started first, at most Limit when it +// names one (§83). +func (r Reads) List(in seat.ListInput) (seat.ListAnswer, error) { + all, err := r.pipelines() + if err != nil { + return seat.ListAnswer{}, err + } + out := []seat.Pipeline{} + for _, p := range all { + if (in.Module == "" || p.Module == in.Module) && (in.Branch == "" || p.Branch == in.Branch) && + (in.Commit == "" || p.Commit == in.Commit) && (in.Run == "" || p.Run == in.Run) { + out = append(out, p) + } + } + sort.SliceStable(out, func(i, j int) bool { return startedOf(out[i]).After(startedOf(out[j])) }) + if in.Limit > 0 && int64(len(out)) > in.Limit { + out = out[:in.Limit] + } + return seat.ListAnswer{Pipelines: out}, nil +} + +// Stands is every step that has started and not ended, with what it waits on and since when (§75, §83): the watchman +// seeds its open waits from it. +func (r Reads) Stands() (seat.StandsAnswer, error) { + all, err := r.pipelines() + if err != nil { + return seat.StandsAnswer{}, err + } + out := []seat.Wait{} + for _, p := range all { + for _, s := range p.Steps { + if s.Ended != nil || s.Started == nil { + continue + } + if s.Outcome != seat.StepResultOutcomeRunning && s.Outcome != seat.StepResultOutcomeWaiting { + continue + } + out = append(out, seat.Wait{Pipeline: p.Pipeline, Module: p.Module, Branch: p.Branch, Commit: p.Commit, + Group: p.Group, Run: p.Run, Source: p.Source, Revision: p.Revision, Fp: p.Fp, Step: s.Step, + Artifact: s.Artifact, On: s.On, Since: *s.Started, V: 1}) + } + } + sort.SliceStable(out, func(i, j int) bool { return out[i].Since.Before(out[j].Since) }) + return seat.StandsAnswer{Stands: out}, nil +} + +// WhyCarried is where a step's result was carried from (§25, §83): the pipeline it names. +func (r Reads) WhyCarried(in seat.WhyCarriedInput) (seat.WhyCarriedAnswer, error) { + p, ok, err := r.pipeline(in.Pipeline) + if err != nil { + return seat.WhyCarriedAnswer{}, err + } + if !ok { + return seat.WhyCarriedAnswer{}, fmt.Errorf("no pipeline %q in %s", in.Pipeline, seat.StatePipelines) + } + for _, s := range p.Steps { + if s.Step != in.Step || (in.Artifact != "" && s.Artifact != in.Artifact) { + continue + } + if s.Outcome != seat.StepResultOutcomeCarried { + return seat.WhyCarriedAnswer{}, fmt.Errorf("step %s of %s %s was not carried: it ended %q", s.Step, + in.Pipeline, s.Artifact, s.Outcome) + } + return seat.WhyCarriedAnswer{CarriedFrom: s.CarriedFrom}, nil + } + return seat.WhyCarriedAnswer{}, fmt.Errorf("pipeline %q has no step %s %s", in.Pipeline, in.Step, in.Artifact) +} + +// KeptResults is the kept passing results, by level, digest or package (§83). +func (r Reads) KeptResults(in seat.KeptInput) (seat.KeptAnswer, error) { + keys, err := r.Kept.Keys() + if err != nil { + return seat.KeptAnswer{}, fmt.Errorf("%s cannot be read: %w", seat.StateKept, err) + } + sort.Strings(keys) + out := []seat.KeptResult{} + for _, k := range keys { + level, digest, ok := seat.ParseKeptByLevelDigestKey(k) + if !ok || (in.Level != "" && level != in.Level) || (in.Digest != "" && digest != in.Digest) { + continue + } + raw, err := r.Kept.Value(k) + if err != nil { + return seat.KeptAnswer{}, fmt.Errorf("%s %s cannot be read: %w", seat.StateKept, k, err) + } + if raw == nil { + continue + } + v, err := seat.DecodeKeptByLevelDigestValue(raw) + if err != nil { + return seat.KeptAnswer{}, fmt.Errorf("%s %s: %w", seat.StateKept, k, err) + } + if in.Package != "" && v.Package != in.Package { + continue + } + out = append(out, v) + } + return seat.KeptAnswer{Kept: out}, nil +} + +func (r Reads) pipeline(key string) (seat.Pipeline, bool, error) { + key = strings.TrimSpace(key) + if key == "" { + return seat.Pipeline{}, false, fmt.Errorf("name the pipeline's key, ..") + } + raw, err := r.Pipelines.Value(key) + if err != nil { + return seat.Pipeline{}, false, fmt.Errorf("%s %s cannot be read: %w", seat.StatePipelines, key, err) + } + if raw == nil { + return seat.Pipeline{}, false, nil + } + p, err := seat.DecodePipelinesByBranchRevisionModuleValue(raw) + if err != nil { + return seat.Pipeline{}, false, fmt.Errorf("%s %s: %w", seat.StatePipelines, key, err) + } + if p.Pipeline == "" { + p.Pipeline = key + } + return p, true, nil +} + +func (r Reads) pipelines() ([]seat.Pipeline, error) { + keys, err := r.Pipelines.Keys() + if err != nil { + return nil, fmt.Errorf("%s cannot be read: %w", seat.StatePipelines, err) + } + sort.Strings(keys) + var out []seat.Pipeline + for _, k := range keys { + p, ok, err := r.pipeline(k) + if err != nil { + return nil, err + } + if ok { + out = append(out, p) + } + } + return out, nil +} + +func startedOf(p seat.Pipeline) time.Time { + if p.Started == nil { + return time.Time{} + } + return *p.Started +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/read_test.go b/modules/mesh-delivery-pipeline/internal/pipeline/read_test.go new file mode 100644 index 00000000..af516211 --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/read_test.go @@ -0,0 +1,129 @@ +package pipeline + +import ( + "encoding/json" + "strings" + "testing" + "time" + + seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner" +) + +type memKept map[string]json.RawMessage + +func (m memKept) Keys() ([]string, error) { + var out []string + for k := range m { + out = append(out, k) + } + return out, nil +} + +func (m memKept) Value(key string) (json.RawMessage, error) { return m[key], nil } + +func put(t *testing.T, m memKept, key string, v any) { + t.Helper() + raw, err := json.Marshal(v) + if err != nil { + t.Fatal(err) + } + m[key] = raw +} + +// TestReads holds the runner's read verbs to §83 and §306, one row per case (RN15/AC1, test 93's runner case). +func TestReads(t *testing.T) { + t0 := time.Date(2026, 11, 11, 10, 0, 0, 0, time.UTC) + t1, t2 := t0.Add(time.Minute), t0.Add(2*time.Minute) + pipes, kept := memKept{}, memKept{} + put(t, pipes, "main.7.alpha", seat.Pipeline{Pipeline: "main.7.alpha", Module: "alpha", Branch: "main", Commit: "c7", + Source: 7, Outcome: seat.PipelineOutcomeRunning, Started: &t0, Steps: []seat.StepResult{ + {Step: seat.StepResultStepLint, Artifact: "alpha:program", Status: "required", Outcome: seat.StepResultOutcomePassed, + Started: &t0, Ended: &t1, Output: "main.7.alpha.lint.alpha:program"}, + {Step: seat.StepResultStepUnit, Artifact: "alpha:program", Status: "required", Outcome: seat.StepResultOutcomeCarried, + CarriedFrom: "main.6.alpha"}, + {Step: seat.StepResultStepApproval, Status: "required", Outcome: seat.StepResultOutcomeWaiting, On: seat.WaitOnPerson, + Started: &t1}, + }}) + put(t, pipes, "beta.3.alpha", seat.Pipeline{Pipeline: "beta.3.alpha", Module: "alpha", Branch: "beta", Commit: "b3", + Outcome: seat.PipelineOutcomePassed, Started: &t2}) + put(t, pipes, "main.7.gamma", seat.Pipeline{Pipeline: "main.7.gamma", Module: "gamma", Branch: "main", Commit: "c7", + Outcome: seat.PipelineOutcomeRunning, Started: &t1}) + put(t, kept, "unit.sha256-aa", seat.KeptResult{Level: "unit", Digest: "sha256-aa", Package: "alpha/x"}) + put(t, kept, "integration.sha256-bb", seat.KeptResult{Level: "integration", Digest: "sha256-bb", Package: "alpha/y"}) + r := Reads{Pipelines: pipes, Kept: kept} + + names := func(ps []seat.Pipeline) string { + var s []string + for _, p := range ps { + s = append(s, p.Pipeline) + } + return strings.Join(s, " ") + } + rows := []struct { + name string + got func() (string, error) + want string + }{ + {"show answers the pipeline whole and its outputs", func() (string, error) { + a, err := r.Show(seat.ShowInput{Pipeline: "main.7.alpha"}) + return a.Pipeline.Module + " " + strings.Join(a.Outputs, ","), err + }, "alpha main.7.alpha.lint.alpha:program"}, + {"show refuses a pipeline it does not hold", func() (string, error) { + _, err := r.Show(seat.ShowInput{Pipeline: "main.9.alpha"}) + return "", err + }, `error: no pipeline "main.9.alpha" in pipelines`}, + {"list by module, newest first", func() (string, error) { + a, err := r.List(seat.ListInput{Module: "alpha"}) + return names(a.Pipelines), err + }, "beta.3.alpha main.7.alpha"}, + {"list by commit", func() (string, error) { + a, err := r.List(seat.ListInput{Commit: "c7"}) + return names(a.Pipelines), err + }, "main.7.gamma main.7.alpha"}, + {"list by branch with a limit", func() (string, error) { + a, err := r.List(seat.ListInput{Branch: "main", Limit: 1}) + return names(a.Pipelines), err + }, "main.7.gamma"}, + {"stands names a waiting step and since when", func() (string, error) { + a, err := r.Stands() + var s []string + for _, w := range a.Stands { + s = append(s, w.Pipeline+" "+w.Step+" "+w.On+" "+w.Since.Format("15:04")) + } + return strings.Join(s, ";"), err + }, "main.7.alpha approval person 10:01"}, + {"why-carried names where a carried step came from", func() (string, error) { + a, err := r.WhyCarried(seat.WhyCarriedInput{Pipeline: "main.7.alpha", Step: "unit", Artifact: "alpha:program"}) + return a.CarriedFrom, err + }, "main.6.alpha"}, + {"why-carried refuses a step that ran", func() (string, error) { + _, err := r.WhyCarried(seat.WhyCarriedInput{Pipeline: "main.7.alpha", Step: "lint"}) + return "", err + }, `error: step lint of main.7.alpha alpha:program was not carried: it ended "passed"`}, + {"kept by level", func() (string, error) { + a, err := r.KeptResults(seat.KeptInput{Level: "unit"}) + var s []string + for _, k := range a.Kept { + s = append(s, k.Package) + } + return strings.Join(s, " "), err + }, "alpha/x"}, + {"kept by package", func() (string, error) { + a, err := r.KeptResults(seat.KeptInput{Package: "alpha/y"}) + var s []string + for _, k := range a.Kept { + s = append(s, k.Digest) + } + return strings.Join(s, " "), err + }, "sha256-bb"}, + } + for _, row := range rows { + got, err := row.got() + if err != nil { + got = "error: " + err.Error() + } + if got != row.want { + t.Errorf("%s: got %q, want %q", row.name, got, row.want) + } + } +} diff --git a/modules/mesh-delivery-pipeline/module.json b/modules/mesh-delivery-pipeline/module.json index 02802158..3a618fe0 100644 --- a/modules/mesh-delivery-pipeline/module.json +++ b/modules/mesh-delivery-pipeline/module.json @@ -35,17 +35,41 @@ "mesh-catalog.catalog-revision" ], "state": [ - { "name": "pipelines", "history": 1 }, - { "name": "kept", "history": 1 }, - { "name": "aged", "history": 1 }, - { "name": "findings", "history": 1 }, - { "name": "first-views", "history": 1 } + { + "name": "pipelines", + "history": 1 + }, + { + "name": "kept", + "history": 1 + }, + { + "name": "aged", + "history": 1 + }, + { + "name": "findings", + "history": 1 + }, + { + "name": "first-views", + "history": 1 + } ], "streams": [ - { "name": "step-log", "events": ["step-log.**"], "max-mib": 128 } + { + "name": "step-log", + "events": [ + "step-log.**" + ], + "max-mib": 128 + } ], "objects": [ - { "name": "step-outputs", "max-mib": 1024 } + { + "name": "step-outputs", + "max-mib": 1024 + } ], "own-secrets": { "broker": "${dir:state}/broker" @@ -93,7 +117,20 @@ "system": "arch", "from": "cmd/mesh-delivery-pipeline", "binary": "mesh-delivery-pipeline", - "needs": ["mesh-delivery-assigner:program"] + "needs": [ + "mesh-delivery-assigner:program" + ] + }, + { + "name": "tools", + "kind": "bundle", + "language": "go", + "system": "arch", + "from": "cmd/mesh-delivery-pipeline-tools", + "binary": "mesh-delivery-pipeline-tools", + "loads": [ + "mesh-delivery-pipeline-tools" + ] } ] }