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 index bebc4d19..dcc14e41 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main.go @@ -7,6 +7,10 @@ import ( "encoding/json" "fmt" "os" + "time" + + "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient" + "git.novox.be/novox/mesh-sdk/go/views" seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner" stdio "git.novox.be/novox/mesh-sdk/go" @@ -82,9 +86,22 @@ func Tools(r pipeline.Reads) []stdio.Tool { return out } +// Module and Slug are the runner's module and its tools' prefix, as its manifest names them. +const ( + Module = "mesh-delivery-pipeline" + Slug = "pipeline" +) + +// FirstViewTools are the runner's `_first_views` and `_first_view_again` over its state first-views +// (design 47 G §77, §200): each view it holds or waits for, since when, and asked again by hand, a hand act. +func FirstViewTools(st stdio.KeptState, acts views.HandActs, now func() time.Time) []stdio.Tool { + return views.StatusTools(Module, Slug+"_first_views", Slug+"_first_view_again", st, acts, now) +} + 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 { + tools := append(Tools(r), FirstViewTools(stdio.State(views.StatusState), meshhandactsclient.Recorder{Ask: stdio.Ask}, time.Now)...) + if err := stdio.Serve("", tools); 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 index 0565790a..73d96db0 100644 --- 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 @@ -5,6 +5,9 @@ import ( "strings" "testing" + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/views" + "mesh-delivery-pipeline/internal/pipeline" ) @@ -13,13 +16,17 @@ 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). +// TestTools: the bundle serves the five read verbs as . (§83, §306; RN15/AC1), and the first-view tools +// (§77, §200; RN2/AC4). 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" + for _, tl := range FirstViewTools(stdio.State(views.StatusState), nil, nil) { + 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 pipeline_first_views pipeline_first_view_again" if got := strings.Join(names, " "); got != want { t.Errorf("tools %q, want %q", got, want) } diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go index a931e9eb..5805b1e0 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go @@ -5,6 +5,7 @@ package main import ( "context" + "encoding/json" "fmt" "io" "log" @@ -12,8 +13,12 @@ import ( "os/signal" "syscall" + "github.com/nats-io/nats.go" + seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner" "git.novox.be/novox/mesh-sdk/go/process" + + "mesh-delivery-pipeline/internal/pipeline" ) func main() { @@ -39,9 +44,39 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro defer conn.Close() // The state pipelines, `_pipelines`, holds every pipeline by compare-and-set (§27): without it the // runner keeps nothing, so it stops. - if _, err := process.EssentialBucket(ctx, js, "pipeline runner", ownBucket(cred.Module, seat.StatePipelines), "keep a pipeline"); err != nil { + pipelines, err := process.EssentialBucket(ctx, js, "pipeline runner", ownBucket(cred.Module, seat.StatePipelines), "keep a pipeline") + if err != nil { return err } + // The registrations and each watched branch's catalog half, first views then events (§21, §195, §200): until a + // branch's view is held the runner acts on nothing new of it, and says so. + calls, err := process.Connect(cred, nats.Name(cred.Node+"."+cred.Module+".calls"), nats.RetryOnFailedConnect(true)) + if err != nil { + return err + } + defer calls.Close() + hub := process.NewEventHub(logger.Printf) + rv, err := startViews(ctx, js, conn, cred, hub, calls.Call, logger.Printf) + if err != nil { + return err + } + // A pipeline starts by compare-and-set on its key's absence and says pipeline-started (§54, §103); a publish starts + // one per dependent behind (§53, §334). + starter := &pipeline.Starter{Pipelines: createdKV{ctx: ctx, kv: pipelines}, Logf: logger.Printf, + Say: func(subject string, body any) error { + data, err := json.Marshal(body) + if err == nil { + _, err = js.Publish(ctx, subject, data) + } + return err + }} + if err := hearPublished(ctx, hub, starter, calls, rv.Catalog, logger.Printf); err != nil { + return err + } + if err := hub.StartUntilBound(ctx, js, cred, nil, nil); err != nil { + logger.Printf("the consumer %s is not bound yet (%v): bound once the bus makes it", cred.EventConsumerName(), err) + } + go func() { <-ctx.Done(); hub.Stop() }() logger.Printf("holding %s", seat.Seat) <-ctx.Done() return nil diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/start.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/start.go new file mode 100644 index 00000000..a2341544 --- /dev/null +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/start.go @@ -0,0 +1,71 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + + "github.com/nats-io/nats.go/jetstream" + + builds "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshbuilds" + catalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" + "git.novox.be/novox/mesh-sdk/go/buscalls" + "git.novox.be/novox/mesh-sdk/go/process" + + "mesh-delivery-pipeline/internal/pipeline" +) + +// outdatedVerb is the catalog's verb a publication's dependents behind are asked by, as the manifest's invokes names it. +const outdatedVerb = "seat:" + catalog.Seat + "." + catalog.VerbOutdated + +// createdKV is the state pipelines written by compare-and-set on a key's absence (§54). +type createdKV struct { + ctx context.Context + kv jetstream.KeyValue +} + +func (c createdKV) Create(key string, value []byte) error { + _, err := c.kv.Create(c.ctx, key, value) + if errors.Is(err, jetstream.ErrKeyExists) { + return pipeline.ErrExists + } + return err +} + +// hearPublished starts the pipelines of each package-published heard (§53, §334): the dependents the catalog's outdated +// names behind, each asked as a kept call. The message is acknowledged once its starts are written, on a goroutine of its +// own, so a slow catalog holds no other event. +func hearPublished(ctx context.Context, hub *process.EventHub, starter *pipeline.Starter, calls *process.Conn, v *pipeline.Views, logf func(string, ...any)) error { + outdated := func(in catalog.OutdatedInput) (catalog.OutdatedAnswer, error) { + raw, err := calls.CallKept(outdatedVerb, in) + if err != nil { + return catalog.OutdatedAnswer{}, err + } + answer, said, ok := buscalls.Unwrap(raw) + if !ok { + return catalog.OutdatedAnswer{}, fmt.Errorf("%s answered: %s", outdatedVerb, said) + } + var a catalog.OutdatedAnswer + err = json.Unmarshal(answer, &a) + return a, err + } + return hub.Handle(builds.SubjectPackagePublished, func(m jetstream.Msg) { + var ev builds.PackagePublished + if err := json.Unmarshal(m.Data(), &ev); err != nil { + logf("%s is no package-published: %v", m.Subject(), err) + _ = m.Ack() + return + } + go func() { + n, err := starter.Published(ev, outdated, v.View) + if err != nil { + logf("the publish of %s %s on %s: %d pipelines started, then %v; heard again", ev.Artifact, ev.Version, ev.Branch, n, err) + _ = m.Nak() + return + } + logf("the publish of %s %s on %s started %d pipelines", ev.Artifact, ev.Version, ev.Branch, n) + _ = m.Ack() + }() + }) +} diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/views.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/views.go new file mode 100644 index 00000000..0ff4228f --- /dev/null +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/views.go @@ -0,0 +1,140 @@ +package main + +import ( + "context" + "sync" + "time" + + "github.com/nats-io/nats.go/jetstream" + + mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" + sdkfindings "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient/findings" + "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" + catalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" + "git.novox.be/novox/mesh-sdk/go/process" + "git.novox.be/novox/mesh-sdk/go/views" + + "mesh-delivery-pipeline/internal/pipeline" +) + +// catalogRevisions is every watched branch's catalog-revision, `mesh.seat.mesh-catalog.event.catalog-revision.`. +const catalogRevisions = "mesh.seat." + catalog.Seat + ".event." + catalog.EventCatalogRevision + ".>" + +// runnerViews are the runner's first views (design 47 G §21, §195, §200): the registrations, then the catalog half of +// each watched branch they name, each said in `_first_views` (its state first-views) and, past five minutes +// without a view, as view-waits in its findings. +type runnerViews struct { + Registrations *meshcatalogclient.Registrations + Catalog *pipeline.Views +} + +// startViews wires the runner's views on hub and asks each first view; register every other handler on hub before or +// after, then start hub once. +func startViews(ctx context.Context, js jetstream.JetStream, conn process.Subscriber, cred process.Account, hub *process.EventHub, + call meshcatalogclient.Call, logf func(string, ...any)) (*runnerViews, error) { + findings := process.AwaitBucket(ctx, js, conn, mcc.FindingsBucket(cred.Module), "a view waited for past five minutes is said in the log alone", logf) + status := process.AwaitBucket(ctx, js, conn, cred.Module+"_"+views.StatusState, "the runner's first views are not said in its first-view tool, nor asked again by hand", logf) + writer := func() views.FindingWriter { return sdkfindings.Writer(ctx, findings.KV) } + reader := &views.ProcessReader{Module: cred.Module, Call: call, Logf: logf, Hub: hub, Findings: writer} + + // A watched branch first named by a registration gets its view of the catalog half (CT10), its conditions and its + // status kept as every first view's, asked at once. + wire := func(f *views.Follower) { + f.Logf = logf + f.Raise = func(s views.Status) { + putFinding(writer(), views.UnansweredFinding(cred.Module, s, time.Now()), true, logf) + } + f.Clear = func(s views.Status) { + putFinding(writer(), views.UnansweredFinding(cred.Module, s, time.Now()), false, logf) + } + f.RaiseWaits = func(s views.Status) { + logf("waits for %s's view %s since %s", s.Owner, s.WaitsFor, s.WaitingSince.Format(time.RFC3339)) + putFinding(writer(), views.ViewWaitsFinding(cred.Module, s, time.Now()), true, logf) + } + f.ClearWaits = func(s views.Status) { + putFinding(writer(), views.ViewWaitsFinding(cred.Module, s, time.Now()), false, logf) + } + f.Said = func(s views.Status) { + if kv := status.KV(); kv != nil { + views.KeepStatus(ctx, kv, logf)(s) + } + } + status.Ready(func(kv jetstream.KeyValue) { + if err := views.HearAgain(ctx, kv, []*views.Follower{f}, logf); err != nil { + logf("the hand asks of the first view %s are not heard (%v) until the runner starts again", f.Tool, err) + } + }) + } + var ( + mu sync.Mutex + made []*views.Follower + ) + cv := &pipeline.Views{Logf: logf, Open: func(branch string) (pipeline.BranchView, error) { + c, err := meshcatalogclient.NewCatalog(call, branch) + if err != nil { + return nil, err + } + wire(c.Follower) + mu.Lock() + made = append(made, c.Follower) + mu.Unlock() + go func() { + if err := c.Start(ctx); err != nil { + logf("the first view of the catalog half of %s was not answered: %v", branch, err) + } + }() + return catalogView{c}, nil + }} + regs, _, err := meshcatalogclient.WireRegistrationsProcess(reader, call, func(reg catalog.Registration, gone bool) { + if !gone { + cv.Registered(reg) + } + }) + if err != nil { + return nil, err + } + if err := hub.Handle(catalogRevisions, func(m jetstream.Msg) { + if err := cv.HearRevision(ctx, m.Data()); err != nil { + logf("%s is not taken: %v", m.Subject(), err) + } + _ = m.Ack() + }); err != nil { + return nil, err + } + // A consumer made after a view was asked does not deliver what was said before it: each branch's view asks again. + hub.OnBound(func(created time.Time) { + mu.Lock() + fs := append([]*views.Follower(nil), made...) + mu.Unlock() + for _, f := range fs { + f.ConsumerMade(ctx, created) + } + }) + reader.Start(ctx, js, cred, status) + return &runnerViews{Registrations: regs, Catalog: cv}, nil +} + +func putFinding(w views.FindingWriter, f views.Finding, raised bool, logf func(string, ...any)) { + var err error + if raised { + err = w.Put(f) + } else { + err = w.Delete(f) + } + if err != nil { + logf("the finding %s is not written: %v", f.Kind, err) + } +} + +// catalogView is a branch's meshcatalogclient.Catalog as the runner reads it. +type catalogView struct{ c *meshcatalogclient.Catalog } + +func (v catalogView) Hear(ctx context.Context, data []byte) error { return v.c.Hear(ctx, data) } + +func (v catalogView) View() (pipeline.Half, bool) { + h, ok := v.c.View() + if !ok { + return pipeline.Half{}, false + } + return pipeline.Half{Revision: h.Revision.Revision, Source: h.Revision.Source, Entries: h.Entries}, true +} diff --git a/modules/mesh-delivery-pipeline/go.mod b/modules/mesh-delivery-pipeline/go.mod index 511f96a5..daa61da4 100644 --- a/modules/mesh-delivery-pipeline/go.mod +++ b/modules/mesh-delivery-pipeline/go.mod @@ -9,6 +9,12 @@ require ( ) require ( + git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-delivery/contracts/go/clients/meshpipelinerunnerclient v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-delivery/contracts/go/seats/meshbuilds v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog v0.0.0-00010101000000-000000000000 github.com/klauspost/compress v1.20.1 // indirect github.com/nats-io/nkeys v0.4.16 // indirect github.com/nats-io/nuid v1.0.1 // indirect diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/registrations.go b/modules/mesh-delivery-pipeline/internal/pipeline/registrations.go new file mode 100644 index 00000000..4eaa2ae2 --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/registrations.go @@ -0,0 +1,170 @@ +package pipeline + +import ( + "context" + "fmt" + "sort" + "sync" + + catalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" + seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner" +) + +// Status is a step's status for one artifact of a module, read from the module's registration and never from a +// manifest (design 47 G §21, §23): the artifact's own status where its policy names one, else the module's. A step +// neither names is "" (design question Q-RN8, open). +func Status(reg catalog.Registration, artifact, step string) string { + if a, ok := reg.Policy.Artifacts[artifact]; ok { + if s := statusOf(a, step); s != "" { + return s + } + } + return statusOf(reg.Policy.Steps, step) +} + +func statusOf(s catalog.StepStatuses, step string) string { + switch step { + case seat.StepResultStepLint: + return s.Lint + case seat.StepResultStepUnit: + return s.Unit + case seat.StepResultStepIntegration: + return s.Integration + case seat.StepResultStepComposition: + return s.Composition + case seat.StepResultStepReview: + return s.Review + case seat.StepResultStepApproval: + return s.Approval + } + return "" +} + +// buildStep is the acting rule's step a build seat names for the languages it builds (design 47 G §323). +const buildStep = "build" + +// BuildSeat is the build seat that runs lint and the test steps of an artifact of language, read from the catalog half +// of the pipeline's revision view (design 47 G §323): the seat whose acting rule `build` has `on` naming that language, +// as an entry of the half that claims it carries it in `acts`. Two seats naming one language are an error naming both. +func BuildSeat(entries map[string]catalog.CatalogEntry, language string) (string, error) { + var seats []string + for _, e := range entries { + for _, r := range e.Acts { + if r.Step == buildStep && r.On == language && r.Seat != "" && !contains(seats, r.Seat) { + seats = append(seats, r.Seat) + } + } + } + sort.Strings(seats) + switch len(seats) { + case 1: + return seats[0], nil + case 0: + return "", fmt.Errorf("no build seat's acting rule %s names the language %q in the catalog half", buildStep, language) + } + return "", fmt.Errorf("the build seats %v each name the language %q in their acting rule %s", seats, language, buildStep) +} + +func contains(list []string, s string) bool { + for _, x := range list { + if x == s { + return true + } + } + return false +} + +// BranchView is the runner's view of one watched branch's catalog half (design 47 G §195, §200, §479): a first view, then +// that branch's catalog-revision events. meshcatalogclient.Catalog is one. +type BranchView interface { + Hear(ctx context.Context, data []byte) error + View() (Half, bool) +} + +// Half is the catalog half a branch view holds, as the runner reads it: its revision and its entries by module. +type Half struct { + Revision int64 + Source int64 + Entries map[string]catalog.CatalogEntry +} + +// Views is the runner's views of the catalog half, one per watched branch the registrations name (design 47 G §21, +// §195): a branch first named by a registration gets its view made and started by Open, and every catalog-revision +// heard goes to every branch's view, each taking its own branch's alone. +type Views struct { + // Open makes and starts the view of a branch; its error is said in the journal and the branch asked again at the + // next registration heard. + Open func(branch string) (BranchView, error) + Logf func(string, ...any) + + mu sync.Mutex + branches map[string]BranchView +} + +// Registered takes the watched branches of a module's registration (design 47 G §21): each branch the views do not hold +// yet is opened. +func (v *Views) Registered(reg catalog.Registration) { + for _, b := range reg.Branches { + v.ensure(b) + } +} + +func (v *Views) ensure(branch string) { + v.mu.Lock() + defer v.mu.Unlock() + if v.branches == nil { + v.branches = map[string]BranchView{} + } + if _, ok := v.branches[branch]; ok || branch == "" { + return + } + bv, err := v.Open(branch) + if err != nil { + if v.Logf != nil { + v.Logf("the catalog half of the watched branch %s is not followed (asked again at the next registration heard): %v", branch, err) + } + return + } + v.branches[branch] = bv +} + +// Branches are the watched branches the views hold, sorted. +func (v *Views) Branches() []string { + v.mu.Lock() + defer v.mu.Unlock() + out := make([]string, 0, len(v.branches)) + for b := range v.branches { + out = append(out, b) + } + sort.Strings(out) + return out +} + +// View is the catalog half the view of branch holds; false while it has none, when the runner acts on nothing new of +// that branch (design 47 G §200). +func (v *Views) View(branch string) (Half, bool) { + v.mu.Lock() + bv := v.branches[branch] + v.mu.Unlock() + if bv == nil { + return Half{}, false + } + return bv.View() +} + +// HearRevision hands one catalog-revision event to every branch's view. +func (v *Views) HearRevision(ctx context.Context, data []byte) error { + v.mu.Lock() + all := make([]BranchView, 0, len(v.branches)) + for _, bv := range v.branches { + all = append(all, bv) + } + v.mu.Unlock() + var first error + for _, bv := range all { + if err := bv.Hear(ctx, data); err != nil && first == nil { + first = err + } + } + return first +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/registrations_test.go b/modules/mesh-delivery-pipeline/internal/pipeline/registrations_test.go new file mode 100644 index 00000000..0f150d36 --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/registrations_test.go @@ -0,0 +1,85 @@ +package pipeline + +import ( + "context" + "errors" + "strings" + "testing" + + catalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" +) + +// The policy, the build seat by language and the views per watched branch (design 47 G §21, §23, §323, §195). +func TestPolicyAndBuildSeat(t *testing.T) { + reg := catalog.Registration{Module: "m", Branches: []string{"main"}, Policy: catalog.Policy{ + Steps: catalog.StepStatuses{Lint: "required", Unit: "required", Integration: "optional", Approval: "required"}, + Artifacts: map[string]catalog.StepStatuses{"tools": {Unit: "bypass"}}, + }} + for _, c := range []struct{ artifact, step, want string }{ + {"program", "unit", "required"}, + {"tools", "unit", "bypass"}, + {"tools", "integration", "optional"}, + {"tools", "approval", "required"}, + {"program", "composition", ""}, + } { + if got := Status(reg, c.artifact, c.step); got != c.want { + t.Errorf("Status(%s, %s) = %q, want %q", c.artifact, c.step, got, c.want) + } + } + + entries := map[string]catalog.CatalogEntry{ + "mesh-build-go": {Module: "mesh-build-go", Acts: []catalog.ActingRule{{Seat: "node-build-go", Step: "build", On: "go"}}}, + "mesh-build-node": {Module: "mesh-build-node", Acts: []catalog.ActingRule{{Seat: "node-build-node", Step: "build", On: "typescript"}, {Seat: "node-build-node", Step: "check", On: "go"}}}, + "app": {Module: "app", Artifacts: []catalog.Artifact{{Ref: "app:program", Language: "go"}}}, + } + for _, c := range []struct{ language, want, err string }{ + {"go", "node-build-go", ""}, + {"typescript", "node-build-node", ""}, + {"rust", "", `"rust"`}, + } { + got, err := BuildSeat(entries, c.language) + if got != c.want || (c.err == "") != (err == nil) || (err != nil && !strings.Contains(err.Error(), c.err)) { + t.Errorf("BuildSeat(%s) = %q, %v; want %q, an error naming %s", c.language, got, err, c.want, c.err) + } + } + entries["other"] = catalog.CatalogEntry{Acts: []catalog.ActingRule{{Seat: "node-build-other", Step: "build", On: "go"}}} + if _, err := BuildSeat(entries, "go"); err == nil || !strings.Contains(err.Error(), "node-build-other") { + t.Errorf("two seats naming go: %v, want an error naming both", err) + } + + heard := map[string]int{} + opened := 0 + v := &Views{Open: func(b string) (BranchView, error) { + opened++ + if b == "broken" { + return nil, errors.New("refused") + } + return fakeView{branch: b, heard: heard}, nil + }} + v.Registered(reg) + v.Registered(catalog.Registration{Module: "n", Branches: []string{"main", "feature/x", "broken"}}) + if got := v.Branches(); strings.Join(got, ",") != "feature/x,main" || opened != 3 { + t.Fatalf("branches %v opened %d, want feature/x and main, three opens", got, opened) + } + v.Registered(catalog.Registration{Module: "n", Branches: []string{"broken"}}) + if opened != 4 { + t.Errorf("a branch whose view was refused is opened again at the next registration: %d opens", opened) + } + if err := v.HearRevision(context.Background(), []byte(`{}`)); err != nil || heard["main"] != 1 || heard["feature/x"] != 1 { + t.Errorf("a revision heard reaches every branch's view: %v %v", err, heard) + } + if h, ok := v.View("main"); !ok || h.Revision != 7 { + t.Errorf("View(main) = %v %v", h, ok) + } + if _, ok := v.View("gone"); ok { + t.Error("a branch not watched has no view") + } +} + +type fakeView struct { + branch string + heard map[string]int +} + +func (f fakeView) Hear(context.Context, []byte) error { f.heard[f.branch]++; return nil } +func (f fakeView) View() (Half, bool) { return Half{Revision: 7}, true } diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/start.go b/modules/mesh-delivery-pipeline/internal/pipeline/start.go new file mode 100644 index 00000000..3d693669 --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/start.go @@ -0,0 +1,136 @@ +package pipeline + +import ( + "encoding/json" + "errors" + "fmt" + "strconv" + "time" + + runnerclient "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshpipelinerunnerclient" + builds "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshbuilds" + catalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" + seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner" + "git.novox.be/novox/mesh-sdk/go/keys" +) + +// ErrExists is a Create of a pipeline key that is written already: the second start, which writes nothing (§54). +var ErrExists = errors.New("the pipeline's key is written already") + +// Created is the runner's state pipelines as a start writes it: a key written only when it is absent (compare-and-set on +// its absence), ErrExists when it is not. +type Created interface { + Create(key string, value []byte) error +} + +// Begin is what a start names: the pipeline's key (watched branch, source revision, module; design 47 G §27, §54) and +// what the catalog half of that branch's view says of the module. +type Begin struct { + Branch string + Source int64 + Module string + Commit string + Fp string + Group string + Repository string + Revision int64 +} + +// Key is a pipeline's key in the state pipelines, `..`, the branch escaped as a key's +// name is (§27). +func Key(branch string, source int64, module string) string { + return seat.PipelinesByBranchRevisionModuleKey(keys.EscapeName(branch), strconv.FormatInt(source, 10), module) +} + +// Starter starts pipelines (design 47 G §54, §103): it writes a pipeline's key by compare-and-set on its absence, so one +// source revision and module has one pipeline whichever event starts it first, and says pipeline-started for the one it +// wrote. +type Starter struct { + Pipelines Created + // Say says one of the seat's events on its subject. + Say func(subject string, body any) error + Now func() time.Time + Logf func(string, ...any) +} + +// Start writes b's pipeline, running, and says pipeline-started; false, nil when its key was written already (the second +// start writes nothing, §54). +func (s *Starter) Start(b Begin) (bool, error) { + if b.Branch == "" || b.Module == "" || b.Source <= 0 { + return false, fmt.Errorf("a pipeline names its watched branch, source revision and module: %+v", b) + } + now := time.Now().UTC() + if s.Now != nil { + now = s.Now().UTC() + } + key := Key(b.Branch, b.Source, b.Module) + p := seat.Pipeline{V: 1, Pipeline: key, Branch: b.Branch, Source: b.Source, Module: b.Module, Commit: b.Commit, Fp: b.Fp, + Group: b.Group, Repository: b.Repository, Revision: b.Revision, Outcome: seat.PipelineOutcomeRunning, Started: &now, + Steps: []seat.StepResult{}} + raw, err := json.Marshal(p) + if err != nil { + return false, err + } + if err := s.Pipelines.Create(key, raw); errors.Is(err, ErrExists) { + s.logf("%s is started already: this start writes nothing", key) + return false, nil + } else if err != nil { + return false, fmt.Errorf("the pipeline %s is not written: %w", key, err) + } + ev := seat.PipelineStarted{V: 1, At: &now, Pipeline: key, Branch: b.Branch, Source: b.Source, Module: b.Module, + Commit: b.Commit, Fp: b.Fp, Group: b.Group, Revision: b.Revision} + commitOrGroup := b.Commit + if b.Group != "" { + commitOrGroup = b.Group + } + if err := s.Say(runnerclient.Subject(seat.EventPipelineStarted, b.Module, b.Branch, commitOrGroup), ev); err != nil { + s.logf("pipeline-started of %s is not said: %v", key, err) + } + return true, nil +} + +func (s *Starter) logf(format string, args ...any) { + if s.Logf != nil { + s.Logf(format, args...) + } +} + +// outdatedKind is the edge `outdated` names a dependent behind by for a rebuild on a publish: B published before A builds, +// a package A imports (§53). +const outdatedKind = 1 + +// Published starts, for a package-published, a pipeline for each dependent the catalog's `outdated` names behind for that +// package on that branch by kind 1, keyed by the publication's watched branch and source revision and the dependent +// (design 47 G §53, §334). outdated is the kept call of the verb; view the catalog half of the publication's branch. A +// dependent the half does not hold is said and started with what the event names. +func (s *Starter) Published(ev builds.PackagePublished, outdated func(catalog.OutdatedInput) (catalog.OutdatedAnswer, error), view func(branch string) (Half, bool)) (int, error) { + if ev.Artifact == "" || ev.Branch == "" || ev.Source <= 0 { + return 0, fmt.Errorf("a package-published names its artifact, watched branch and source revision: %+v", ev) + } + a, err := outdated(catalog.OutdatedInput{Branch: ev.Branch, Dependency: ev.Artifact}) + if err != nil { + return 0, fmt.Errorf("outdated of %s on %s: %w", ev.Artifact, ev.Branch, err) + } + half, _ := view(ev.Branch) + started := 0 + var first error + for _, o := range a.Behind { + if o.Kind != outdatedKind || o.Module == "" { + continue + } + b := Begin{Branch: ev.Branch, Source: ev.Source, Module: o.Module, Revision: half.Revision} + if e, ok := half.Entries[o.Module]; ok { + b.Commit, b.Fp, b.Repository = e.Commit, e.Fp, e.Repository + } else { + s.logf("the catalog half of %s holds no entry of %s, behind %s: its pipeline starts without its commit", ev.Branch, o.Module, ev.Artifact) + } + ok, err := s.Start(b) + if err != nil && first == nil { + first = err + } + if ok { + started++ + } + } + return started, first +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/start_test.go b/modules/mesh-delivery-pipeline/internal/pipeline/start_test.go new file mode 100644 index 00000000..da18b6ba --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/start_test.go @@ -0,0 +1,87 @@ +package pipeline + +import ( + "encoding/json" + "errors" + "strings" + "testing" + + builds "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshbuilds" + catalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" + seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner" +) + +type created map[string][]byte + +func (c created) Create(key string, value []byte) error { + if _, ok := c[key]; ok { + return ErrExists + } + c[key] = value + return nil +} + +// The starts (design 47 G §54, §103, §334): a pipeline's key written by compare-and-set on its absence, so a second start +// of the same source revision and module writes nothing; a publish starts one pipeline per dependent outdated names +// behind by kind 1, keyed by the publishing branch and revision. +func TestStarts(t *testing.T) { + store := created{} + var said []string + s := &Starter{Pipelines: store, Say: func(subject string, body any) error { + ev := body.(seat.PipelineStarted) + said = append(said, subject+" "+ev.Pipeline) + return nil + }} + view := func(branch string) (Half, bool) { + return Half{Revision: 9, Entries: map[string]catalog.CatalogEntry{ + "app": {Module: "app", Commit: "c1", Fp: "src2:aa", Repository: "o/app"}}}, true + } + var asked catalog.OutdatedInput + outdated := func(in catalog.OutdatedInput) (catalog.OutdatedAnswer, error) { + asked = in + return catalog.OutdatedAnswer{Behind: []catalog.Outdated{ + {Module: "app", Kind: 1}, {Module: "tool", Kind: 1}, {Module: "ordered", Kind: 4}}}, nil + } + n, err := s.Published(builds.PackagePublished{Artifact: "mesh-sdk:go", Branch: "main", Source: 41, Module: "mesh-sdk", Version: "0.1.16"}, outdated, view) + if err != nil || n != 2 { + t.Fatalf("a publish started %d (%v), want the two dependents behind by kind 1", n, err) + } + if asked.Dependency != "mesh-sdk:go" || asked.Branch != "main" { + t.Errorf("outdated asked %+v, want the package on its branch", asked) + } + var p seat.Pipeline + if err := json.Unmarshal(store["main.41.app"], &p); err != nil || p.Commit != "c1" || p.Fp != "src2:aa" || p.Revision != 9 || + p.Outcome != seat.PipelineOutcomeRunning || p.Source != 41 { + t.Errorf("the pipeline main.41.app = %+v (%v)", p, err) + } + if _, ok := store["main.41.ordered"]; ok { + t.Error("an ordering edge starts no pipeline") + } + // The take-in of the same source revision and module is the same pipeline: the second start writes nothing. + before := string(store["main.41.app"]) + ok, err := s.Start(Begin{Branch: "main", Source: 41, Module: "app", Commit: "c2"}) + if ok || err != nil || string(store["main.41.app"]) != before || len(said) != 2 { + t.Errorf("a second start wrote %v (%v); said %v", ok, err, said) + } + if !strings.HasPrefix(said[0], "mesh.seat.mesh-pipeline-runner.event.pipeline-started.app.main.c1 main.41.app") { + t.Errorf("pipeline-started said on %q", said[0]) + } + // A branch holding a dot is escaped in the key, so the key reads back; a start naming no source revision is refused. + if _, err := s.Start(Begin{Branch: "release.1", Source: 3, Module: "app"}); err != nil { + t.Fatal(err) + } + if b, r, m, ok := seat.ParsePipelinesByBranchRevisionModuleKey(Key("release.1", 3, "app")); !ok || r != "3" || m != "app" || b == "release" { + t.Errorf("the key of release.1 is %q", Key("release.1", 3, "app")) + } + if _, err := s.Start(Begin{Branch: "main", Module: "app"}); err == nil { + t.Error("a start with no source revision is refused") + } + s.Pipelines = failing{} + if _, err := s.Start(Begin{Branch: "main", Source: 5, Module: "app"}); err == nil { + t.Error("a write the bus refuses is an error") + } +} + +type failing struct{} + +func (failing) Create(string, []byte) error { return errors.New("bus away") } diff --git a/modules/mesh-delivery-pipeline/manifest_test.go b/modules/mesh-delivery-pipeline/manifest_test.go index 59aa7378..c60c345f 100644 --- a/modules/mesh-delivery-pipeline/manifest_test.go +++ b/modules/mesh-delivery-pipeline/manifest_test.go @@ -49,8 +49,13 @@ func TestManifest(t *testing.T) { // §27, §85, §330, §355, §77: the old state sections. {"old state sections", equalSets(states, []string{seat.StatePipelines, seat.StateKept, "aged", "findings", "first-views"})}, // §27, §85: the step log on an old stream, its outputs in an old objects section. - {"old streams section", equalSets(streams, []string{seat.StateStepLog})}, - {"step-log 128 MiB", len(m.Streams) == 1 && m.Streams[0].MaxMiB == 128}, + // RN24 (§383, §387): beside it, the finding stream its findings are said on. + {"old streams section", equalSets(streams, []string{seat.StateStepLog, "mesh-delivery-pipeline-findings"})}, + {"step-log 128 MiB", len(m.Streams) == 2 && m.Streams[0].MaxMiB == 128}, + {"finding stream 7 days 16 MiB", len(m.Streams) == 2 && m.Streams[1].MaxAgeDays == 7 && m.Streams[1].MaxMiB == 16 && + equalSets(m.Streams[1].Events, []string{"finding-written.*", "finding-cleared.*", "findings-stated.*"})}, + // RN24 (§368): the version as MAJOR.MINOR, 0.1 with none published. + {"version 0.1", m.Version == "0.1"}, {"old objects section", equalSets(objects, []string{seat.StateStepOutputs})}, {"step-outputs 1 GiB", len(m.Objects) == 1 && m.Objects[0].MaxMiB == 1024}, // §333, §54, §44, §53, §376, §24, §37, §4, §77: what the runner hears. @@ -58,7 +63,13 @@ func TestManifest(t *testing.T) { "mesh-delivery-indexer.take-in.taken", "mesh-builds.package-published", "mesh-builds.toolchain-agreed", "operator-channel.decided", "node-build-go.step-ended", "node-build-node.step-ended", "node-build-image.step-ended", "mesh-wants.step-ended", "mesh-catalog.registration-changed", - "mesh-catalog.catalog-revision"})}, + "mesh-catalog.catalog-revision", "mesh-conditions.findings-asked"})}, + // RN2 (§21, §195, §200, §77): the registrations and the catalog half asked by their verbs, the first views read and + // asked again by hand (a hand act). + {"invokes", equalSets(m.Invokes, []string{"seat:mesh-catalog.registrations", "seat:mesh-catalog.revision", "seat:mesh-hand-acts.hand-act", + // RN3 (§53, §334): a publication's dependents behind. + "seat:mesh-catalog.outdated"})}, + {"first-view tools", equalSets(m.Tools, []string{"pipeline_first_views", "pipeline_first_view_again"})}, // RN1/AC5 (§95, §130): the new assigner is delivered before the runner. {"program needs the assigner's program", slices.Equal(program.Needs, []string{"mesh-delivery-assigner:program"})}, } diff --git a/modules/mesh-delivery-pipeline/module.json b/modules/mesh-delivery-pipeline/module.json index 3a618fe0..d0055720 100644 --- a/modules/mesh-delivery-pipeline/module.json +++ b/modules/mesh-delivery-pipeline/module.json @@ -1,6 +1,6 @@ { "module": "mesh-delivery-pipeline", - "version": "1", + "version": "0.1", "slug": "pipeline", "claims": [ { @@ -22,6 +22,10 @@ ] } ], + "tools": [ + "pipeline_first_views", + "pipeline_first_view_again" + ], "consumes": [ "mesh-delivery-indexer.take-in.taken", "mesh-builds.package-published", @@ -32,7 +36,14 @@ "node-build-image.step-ended", "mesh-wants.step-ended", "mesh-catalog.registration-changed", - "mesh-catalog.catalog-revision" + "mesh-catalog.catalog-revision", + "mesh-conditions.findings-asked" + ], + "invokes": [ + "seat:mesh-catalog.registrations", + "seat:mesh-catalog.revision", + "seat:mesh-hand-acts.hand-act", + "seat:mesh-catalog.outdated" ], "state": [ { @@ -63,6 +74,16 @@ "step-log.**" ], "max-mib": 128 + }, + { + "name": "mesh-delivery-pipeline-findings", + "events": [ + "finding-written.*", + "finding-cleared.*", + "findings-stated.*" + ], + "max-age-days": 7, + "max-mib": 16 } ], "objects": [