A-99: the build is every pipeline's first step, asked of mesh-builds for the module's own entry
The runner's step enum gains build; Place puts the build first, required whatever the policy names, and asks it of the seat mesh-builds naming the module's own entry by fingerprint; the hearer takes mesh-builds' step-ended as the other seats' ends, held to its one entry. Builds: design 47 G §230, A-91, A-97, A-99 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
@@ -391,6 +391,7 @@ const (
|
||||
|
||||
// The values of LogEntry's step.
|
||||
const (
|
||||
LogEntryStepBuild = "build"
|
||||
LogEntryStepLint = "lint"
|
||||
LogEntryStepUnit = "unit"
|
||||
LogEntryStepIntegration = "integration"
|
||||
@@ -651,6 +652,7 @@ const (
|
||||
|
||||
// The values of StepResult's step.
|
||||
const (
|
||||
StepResultStepBuild = "build"
|
||||
StepResultStepLint = "lint"
|
||||
StepResultStepUnit = "unit"
|
||||
StepResultStepIntegration = "integration"
|
||||
@@ -715,6 +717,7 @@ const (
|
||||
|
||||
// The values of Wait's step.
|
||||
const (
|
||||
WaitStepBuild = "build"
|
||||
WaitStepLint = "lint"
|
||||
WaitStepUnit = "unit"
|
||||
WaitStepIntegration = "integration"
|
||||
|
||||
@@ -771,6 +771,7 @@
|
||||
"step": {
|
||||
"type": "string",
|
||||
"enum": [
|
||||
"build",
|
||||
"lint",
|
||||
"unit",
|
||||
"integration",
|
||||
@@ -1406,6 +1407,7 @@
|
||||
"step": {
|
||||
"type": "string",
|
||||
"enum": [
|
||||
"build",
|
||||
"lint",
|
||||
"unit",
|
||||
"integration",
|
||||
@@ -1466,6 +1468,7 @@
|
||||
"step": {
|
||||
"type": "string",
|
||||
"enum": [
|
||||
"build",
|
||||
"lint",
|
||||
"unit",
|
||||
"integration",
|
||||
|
||||
+3
-3
@@ -164,7 +164,7 @@ export interface LogEntry {
|
||||
/** the output's name in step-outputs */
|
||||
"output"?: string;
|
||||
"pipeline": string;
|
||||
"step": "lint" | "unit" | "integration" | "composition" | "review" | "approval";
|
||||
"step": "build" | "lint" | "unit" | "integration" | "composition" | "review" | "approval";
|
||||
"what": "started" | "ended";
|
||||
}
|
||||
|
||||
@@ -342,7 +342,7 @@ export interface StepResult {
|
||||
"started"?: string;
|
||||
/** the step's status in the module's policy */
|
||||
"status": "required" | "optional" | "bypass";
|
||||
"step": "lint" | "unit" | "integration" | "composition" | "review" | "approval";
|
||||
"step": "build" | "lint" | "unit" | "integration" | "composition" | "review" | "approval";
|
||||
"tests"?: TestCounts;
|
||||
/** what the outcome rests on: the part not heard, the seat with no holder, what a waiting step waits on */
|
||||
"why"?: string;
|
||||
@@ -389,7 +389,7 @@ export interface Wait {
|
||||
/** the source revision of the watched branch; the pipeline is keyed by it (ADR 0666 section 9) */
|
||||
"source"?: number;
|
||||
/** a pipeline step a wait inside a pipeline is in; pipeline is the acting step of design 47 G 203, for a wait of the entries the runner holds for its new version (design 47 G 534) */
|
||||
"step": "lint" | "unit" | "integration" | "composition" | "review" | "approval" | "pipeline";
|
||||
"step": "build" | "lint" | "unit" | "integration" | "composition" | "review" | "approval" | "pipeline";
|
||||
"v": number;
|
||||
/** for a wait on a person, who: the operator, a reviewer */
|
||||
"who"?: string;
|
||||
|
||||
@@ -8,7 +8,6 @@ import (
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
nodebuildgo "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
|
||||
"mesh-delivery-pipeline/internal/pipeline"
|
||||
@@ -22,18 +21,18 @@ type busStepAsker struct {
|
||||
cred process.Account
|
||||
}
|
||||
|
||||
func (b busStepAsker) AskStep(accept string, a nodebuildgo.StepAsk) error {
|
||||
data, err := json.Marshal(a)
|
||||
func (b busStepAsker) AskStep(accept, id string, body any) error {
|
||||
data, err := json.Marshal(body)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
msg := nats.NewMsg(accept)
|
||||
msg.Header.Set("x-event-id", a.ID)
|
||||
msg.Header.Set("x-event-id", id)
|
||||
msg.Header.Set("x-source", b.cred.Module)
|
||||
msg.Header.Set("x-node", b.cred.Node)
|
||||
msg.Header.Set("x-time", time.Now().UTC().Format(time.RFC3339Nano))
|
||||
msg.Header.Set("content-type", "application/json")
|
||||
msg.Header.Set(nats.MsgIdHdr, a.ID)
|
||||
msg.Header.Set(nats.MsgIdHdr, id)
|
||||
msg.Data = data
|
||||
_, err = b.js.PublishMsg(b.ctx, msg)
|
||||
return err
|
||||
|
||||
@@ -77,9 +77,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
hearer.Asker = busAsker{conn: conn, cred: cred}
|
||||
hearer.Where = routerRecordOf(cred.Module, calls.Call, calls.Ask)
|
||||
// Its asking half (§22, §24): a pipeline's steps are placed from the module's registration and each build seat is asked
|
||||
// for its step, an approval due at the start asked as the hearer asks it.
|
||||
// for its step.
|
||||
starter := &pipeline.Starter{Pipelines: createdKV{ctx: ctx, kv: pipelines}, Logf: logger.Printf, Say: say,
|
||||
Registration: rv.Registrations.Registration, Steps: busStepAsker{ctx: ctx, js: js, cred: cred}, Name: cred.Module, After: hearer}
|
||||
Registration: rv.Registrations.Registration, Steps: busStepAsker{ctx: ctx, js: js, cred: cred}, Name: cred.Module}
|
||||
if err := hearStepEnded(hub, hearer, logger.Printf); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -101,13 +101,20 @@ func hearStepEnded(hub *process.EventHub, h *pipeline.Hearer, logf func(string,
|
||||
for _, bs := range pipeline.StepSeats {
|
||||
bs := bs
|
||||
err := hub.Handle(bs.Prefix+"*", func(m jetstream.Msg) {
|
||||
var ev pipeline.StepEnded
|
||||
if err := json.Unmarshal(m.Data(), &ev); err != nil {
|
||||
decode := bs.Decode
|
||||
if decode == nil {
|
||||
decode = func(raw []byte) (pipeline.StepEnded, error) {
|
||||
var ev pipeline.StepEnded
|
||||
return ev, json.Unmarshal(raw, &ev)
|
||||
}
|
||||
}
|
||||
ev, err := decode(m.Data())
|
||||
if err != nil {
|
||||
logf("%s is no step-ended: %v", m.Subject(), err)
|
||||
_ = m.Term()
|
||||
return
|
||||
}
|
||||
_, err := h.OnStepEnded(bs, m.Subject(), ev)
|
||||
_, err = h.OnStepEnded(bs, m.Subject(), ev)
|
||||
switch {
|
||||
case errors.Is(err, pipeline.ErrRefused):
|
||||
logf("%v", err)
|
||||
|
||||
@@ -8,6 +8,7 @@ import (
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
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"
|
||||
nodebuildgo "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo"
|
||||
@@ -18,11 +19,12 @@ import (
|
||||
// StepAsker puts one step's ask on a build seat's work queue, at the subject accept names (the seat's accepted kind `step`
|
||||
// on its stream `check`, design 47 G §22, §24). An ask asked twice under one id is kept once.
|
||||
type StepAsker interface {
|
||||
AskStep(accept string, a nodebuildgo.StepAsk) error
|
||||
AskStep(accept, id string, body any) error
|
||||
}
|
||||
|
||||
// acceptStep is where each build seat is sent its step, from its contract.
|
||||
var acceptStep = map[string]string{
|
||||
builds.Seat: builds.AcceptSubjectStep,
|
||||
nodebuildgo.Seat: nodebuildgo.AcceptSubjectCheckStep,
|
||||
nodebuildnode.Seat: nodebuildnode.AcceptSubjectCheckStep,
|
||||
nodebuildimage.Seat: nodebuildimage.AcceptSubjectCheckStep,
|
||||
@@ -56,6 +58,13 @@ func Place(p seat.Pipeline, reg catalog.Registration, entries map[string]catalog
|
||||
sort.Slice(arts, func(i, j int) bool { return arts[i].Ref < arts[j].Ref })
|
||||
var steps []seat.StepResult
|
||||
var asks []Asked
|
||||
// The build step is every pipeline's first, required whatever the policy names, asked of the seat mesh-builds for the
|
||||
// module's own entry only (A-99, A-97): a group's other members are built by their own pipelines.
|
||||
build := seat.StepResult{Step: seat.StepResultStepBuild, Status: seat.StepResultStatusRequired, Fingerprint: p.Fingerprint,
|
||||
Seat: builds.Seat, Outcome: seat.StepResultOutcomeRunning, Started: &now, Ask: AskID(p.Pipeline, "", seat.StepResultStepBuild)}
|
||||
steps = append(steps, build)
|
||||
asks = append(asks, Asked{Seat: builds.Seat, ID: build.Ask, Ask: builds.StepRun{Pipeline: p.Pipeline, Step: builds.StepRunStepBuild, Branch: p.Branch,
|
||||
Revision: p.Revision, Source: p.Source, Entries: []builds.BuildEntry{{Module: p.Module, Commit: p.Commit, Fingerprint: p.Fingerprint}}}})
|
||||
for _, a := range arts {
|
||||
name := a.Ref
|
||||
if _, after, ok := strings.Cut(a.Ref, ":"); ok {
|
||||
@@ -85,7 +94,7 @@ func Place(p seat.Pipeline, reg catalog.Registration, entries map[string]catalog
|
||||
s.Seat, s.Outcome, s.Started = bs, seat.StepResultOutcomeRunning, &now
|
||||
s.Ask = AskID(p.Pipeline, a.Ref, step)
|
||||
steps = append(steps, s)
|
||||
asks = append(asks, Asked{Seat: bs, Ask: stepAsk(p, entry, a, step, s.Ask, asker, now)})
|
||||
asks = append(asks, Asked{Seat: bs, ID: s.Ask, Ask: stepAsk(p, entry, a, step, s.Ask, asker, now)})
|
||||
}
|
||||
}
|
||||
for _, step := range moduleSteps {
|
||||
@@ -105,7 +114,8 @@ func Place(p seat.Pipeline, reg catalog.Registration, entries map[string]catalog
|
||||
// Asked is one ask to put on a build seat's work queue.
|
||||
type Asked struct {
|
||||
Seat string
|
||||
Ask nodebuildgo.StepAsk
|
||||
ID string
|
||||
Ask any
|
||||
}
|
||||
|
||||
func stepAsk(p seat.Pipeline, e catalog.CatalogEntry, a catalog.Artifact, step, id, asker string, now time.Time) nodebuildgo.StepAsk {
|
||||
@@ -122,10 +132,10 @@ func askAll(asker StepAsker, key string, asks []Asked) error {
|
||||
accept, ok := acceptStep[a.Seat]
|
||||
err := fmt.Errorf("the seat %s is no build seat the runner asks", a.Seat)
|
||||
if ok {
|
||||
err = asker.AskStep(accept, a.Ask)
|
||||
err = asker.AskStep(accept, a.ID, a.Ask)
|
||||
}
|
||||
if err != nil && first == nil {
|
||||
first = fmt.Errorf("the step %s of %s is not asked of %s: %w", a.Ask.Step.Step, key, a.Seat, err)
|
||||
first = fmt.Errorf("the step %s of %s is not asked of %s: %w", a.ID, key, a.Seat, err)
|
||||
}
|
||||
}
|
||||
return first
|
||||
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"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"
|
||||
nodebuildgo "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo"
|
||||
@@ -12,11 +13,12 @@ import (
|
||||
|
||||
type stepAsks struct {
|
||||
at []string
|
||||
asks []nodebuildgo.StepAsk
|
||||
ids []string
|
||||
asks []any
|
||||
}
|
||||
|
||||
func (s *stepAsks) AskStep(accept string, a nodebuildgo.StepAsk) error {
|
||||
s.at, s.asks = append(s.at, accept), append(s.asks, a)
|
||||
func (s *stepAsks) AskStep(accept, id string, body any) error {
|
||||
s.at, s.ids, s.asks = append(s.at, accept), append(s.ids, id), append(s.asks, body)
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -52,66 +54,76 @@ func TestStartPlacesTheStepsThePolicyNamesAndAsksEachBuildSeat(t *testing.T) {
|
||||
for _, st := range p.Steps {
|
||||
got = append(got, st.Artifact+" "+st.Step+" "+st.Outcome)
|
||||
}
|
||||
want := []string{"app:program lint running", "app:program unit running", "app:tools lint bypassed", "app:tools unit running",
|
||||
want := []string{" build running", "app:program lint running", "app:program unit running", "app:tools lint bypassed", "app:tools unit running",
|
||||
"app:web lint could not run", "app:web unit could not run", " approval waiting"}
|
||||
if strings.Join(got, "|") != strings.Join(want, "|") {
|
||||
t.Errorf("steps = %q, want %q (integration and composition are named by no policy: no step)", got, want)
|
||||
}
|
||||
if !strings.Contains(p.Steps[4].Why, `"rust"`) {
|
||||
t.Errorf("the step of an artifact no seat builds says %q, want it to name the language", p.Steps[4].Why)
|
||||
if !strings.Contains(p.Steps[5].Why, `"rust"`) {
|
||||
t.Errorf("the step of an artifact no seat builds says %q, want it to name the language", p.Steps[5].Why)
|
||||
}
|
||||
if len(asks.asks) != 3 {
|
||||
t.Fatalf("asked %d steps, want the three running ones", len(asks.asks))
|
||||
if len(asks.asks) != 4 {
|
||||
t.Fatalf("asked %d steps, want the build and the three running ones", len(asks.asks))
|
||||
}
|
||||
a := asks.asks[0]
|
||||
if asks.at[0] != nodebuildgo.AcceptSubjectCheckStep || a.Kind != "step" || a.Asker != "the-runner" || a.ID != p.Steps[0].Ask ||
|
||||
// A-99: the build is every pipeline's first step, required whatever the policy names, asked of mesh-builds for the
|
||||
// module's own entry only.
|
||||
b, ok := asks.asks[0].(builds.StepRun)
|
||||
if !ok || asks.at[0] != builds.AcceptSubjectStep || p.Steps[0].Status != seat.StepResultStatusRequired || asks.ids[0] != p.Steps[0].Ask ||
|
||||
len(b.Entries) != 1 || b.Entries[0] != (builds.BuildEntry{Module: "app", Commit: "c1", Fingerprint: "src2:aa"}) || b.Pipeline != "main.4.app" {
|
||||
t.Errorf("the build ask = %+v at %s", asks.asks[0], asks.at[0])
|
||||
}
|
||||
a := asks.asks[1].(nodebuildgo.StepAsk)
|
||||
if asks.at[1] != nodebuildgo.AcceptSubjectCheckStep || a.Kind != "step" || a.Asker != "the-runner" || a.ID != p.Steps[1].Ask || asks.ids[1] != a.ID ||
|
||||
a.Step.Commit != "c1" || a.Step.Fingerprint != "src2:aa" || a.Step.Pipeline != "main.4.app" || a.Step.Artifact != "app:program" ||
|
||||
a.Step.Step != "lint" || a.Step.Fetch.Repository != "o/app" || a.Step.Fetch.Dir != "modules/app" || a.Step.Revision != 4 {
|
||||
t.Errorf("the ask = %+v at %s", a, asks.at[0])
|
||||
t.Errorf("the ask = %+v at %s", a, asks.at[1])
|
||||
}
|
||||
// A second start of the revision asks nothing.
|
||||
if ok, err := s.Start(Begin{Branch: "main", Source: 4, Module: "app", Commit: "c1", Entries: entries}); ok || err != nil || len(asks.asks) != 3 {
|
||||
if ok, err := s.Start(Begin{Branch: "main", Source: 4, Module: "app", Commit: "c1", Entries: entries}); ok || err != nil || len(asks.asks) != 4 {
|
||||
t.Errorf("second start = %v, %v, %d asks", ok, err, len(asks.asks))
|
||||
}
|
||||
}
|
||||
|
||||
// A pipeline whose every other step is bypassed asks its approval at its start, and one whose steps all ended ends at once.
|
||||
func TestStartAsksTheApprovalDueAndEndsAPipelineWithNothingToWaitFor(t *testing.T) {
|
||||
// A pipeline whose other steps are all bypassed still has its build, required whatever the policy names, so its approval is
|
||||
// not due at its start (A-99, §26).
|
||||
func TestTheBuildHoldsTheApprovalWhateverThePolicyNames(t *testing.T) {
|
||||
reg, entries := placeFixture()
|
||||
reg.Policy = catalog.Policy{Steps: catalog.StepStatuses{Lint: "bypass", Unit: "bypass", Approval: "required"}}
|
||||
store := created{}
|
||||
var said []string
|
||||
approval := &askRecorder{}
|
||||
s := &Starter{Pipelines: store, Say: func(subject string, _ any) error {
|
||||
said = append(said, subject[strings.Index(subject, ".event.")+7:])
|
||||
return nil
|
||||
},
|
||||
Steps: &stepAsks{}, Registration: func(string) (catalog.Registration, bool) { return reg, true },
|
||||
After: &Hearer{Name: "the-runner", Asker: approval, Say: func(subject string, _ any) error {
|
||||
said = append(said, subject[strings.Index(subject, ".event.")+7:])
|
||||
return nil
|
||||
}}}
|
||||
s := &Starter{Pipelines: store, Say: func(string, any) error { return nil }, Steps: &stepAsks{},
|
||||
Registration: func(string) (catalog.Registration, bool) { return reg, true }}
|
||||
if _, err := s.Start(Begin{Branch: "main", Source: 4, Module: "app", Commit: "c1", Fp: "src2:aa", Entries: entries}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(approval.asked) != 1 {
|
||||
t.Fatalf("approval asks = %d, want 1 at the start", len(approval.asked))
|
||||
}
|
||||
var p seat.Pipeline
|
||||
_ = json.Unmarshal(store["main.4.app"], &p)
|
||||
if last := p.Steps[len(p.Steps)-1]; last.Step != "approval" || last.Outcome != seat.StepResultOutcomeRunning || last.Ask != approval.asked[0].ID {
|
||||
t.Errorf("approval step = %+v", last)
|
||||
}
|
||||
// Nothing required and nothing named: the pipeline ends at once, passed.
|
||||
reg.Policy = catalog.Policy{Steps: catalog.StepStatuses{Lint: "bypass"}}
|
||||
store2 := created{}
|
||||
s.Pipelines = store2
|
||||
said = nil
|
||||
if _, err := s.Start(Begin{Branch: "main", Source: 5, Module: "app", Entries: entries}); err != nil {
|
||||
if p.Steps[0].Step != seat.StepResultStepBuild || p.Steps[0].Status != seat.StepResultStatusRequired || approvalDue(p) != -1 {
|
||||
t.Errorf("steps = %+v; want the build first, required, and no approval due", p.Steps)
|
||||
}
|
||||
}
|
||||
|
||||
// The seat mesh-builds' end of the build step is taken as the others are, held to its one entry's module, commit and
|
||||
// fingerprint; an end naming two entries is refused (A-97, A-99).
|
||||
func TestTheBuildStepsEndIsTakenOnItsOneEntry(t *testing.T) {
|
||||
bs := StepSeats[len(StepSeats)-1]
|
||||
if bs.Seat != builds.Seat {
|
||||
t.Fatalf("the last step seat is %s, want mesh-builds", bs.Seat)
|
||||
}
|
||||
h, store, _, _, _, _ := fixture()
|
||||
e := store[pkey]
|
||||
e.p.Steps = append([]seat.StepResult{{Step: seat.StepResultStepBuild, Status: seat.StepResultStatusRequired, Outcome: seat.StepResultOutcomeRunning,
|
||||
Ask: "b1", Seat: builds.Seat}}, e.p.Steps...)
|
||||
store[pkey] = e
|
||||
raw := `{"id":"b1","pipeline":"main.5.app","step":"build","node":"n1","outcome":"passed","entries":[{"module":"app","commit":"c1","fingerprint":"src2:aa"}]}`
|
||||
ev, err := bs.Decode([]byte(raw))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
p, err := h.OnStepEnded(bs, bs.Prefix+"n1", ev)
|
||||
if err != nil || p.Steps[0].Outcome != seat.StepResultOutcomePassed {
|
||||
t.Fatalf("the build end: %v, step %+v", err, p.Steps[0])
|
||||
}
|
||||
_ = json.Unmarshal(store2["main.5.app"], &p)
|
||||
if p.Outcome != seat.PipelineOutcomePassed || !strings.HasPrefix(said[len(said)-1], seat.EventPipelineEnded) {
|
||||
t.Errorf("pipeline = %s, said %v; want it ended passed and said", p.Outcome, said)
|
||||
if _, err := bs.Decode([]byte(`{"id":"b1","entries":[{"module":"a","fingerprint":"x"},{"module":"b","fingerprint":"y"}]}`)); err == nil {
|
||||
t.Error("an end of two entries is refused")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -57,10 +57,9 @@ type Starter struct {
|
||||
// pipeline starts with no step and says so.
|
||||
Registration func(module string) (catalog.Registration, bool)
|
||||
// Steps asks each build seat for its step, as the pipeline's asking half (§22, §24); Name is the runner's own name as an
|
||||
// asker. After says what a start wrote (an approval asked, a pipeline ended already) as the hearer does.
|
||||
// asker.
|
||||
Steps StepAsker
|
||||
Name string
|
||||
After *Hearer
|
||||
}
|
||||
|
||||
// Start writes b's pipeline, running, and says pipeline-started; false, nil when its key was written already (the second
|
||||
@@ -85,15 +84,6 @@ func (s *Starter) Start(b Begin) (bool, error) {
|
||||
s.logf("%s has no registration yet: its pipeline starts with no step", key)
|
||||
}
|
||||
}
|
||||
// A pipeline whose steps all ended at once (bypassed, or no seat to run them) ends now; an approval due is asked now.
|
||||
var asked, ended bool
|
||||
if s.After != nil {
|
||||
var err error
|
||||
if asked, err = askApproval(key, &p, now, s.After.Asker); err != nil {
|
||||
return false, err
|
||||
}
|
||||
ended = endPipeline(&p, now)
|
||||
}
|
||||
raw, err := json.Marshal(p)
|
||||
if err != nil {
|
||||
return false, err
|
||||
@@ -113,11 +103,6 @@ func (s *Starter) Start(b Begin) (bool, error) {
|
||||
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)
|
||||
}
|
||||
if s.After != nil {
|
||||
if err := s.After.said(p, asked, ended); err != nil {
|
||||
return true, err
|
||||
}
|
||||
}
|
||||
if s.Steps != nil && len(asks) > 0 {
|
||||
if err := askAll(s.Steps, key, asks); err != nil {
|
||||
return true, err
|
||||
|
||||
@@ -1,12 +1,14 @@
|
||||
package pipeline
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
|
||||
builds "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshbuilds"
|
||||
seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner"
|
||||
nodebuildgo "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo"
|
||||
nodebuildimage "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildimage"
|
||||
@@ -20,6 +22,8 @@ type StepSeat struct {
|
||||
Seat string
|
||||
// Prefix is the subject before <node>.
|
||||
Prefix string
|
||||
// Decode reads the seat's step-ended body; nil is the build seats' one form.
|
||||
Decode func([]byte) (StepEnded, error)
|
||||
}
|
||||
|
||||
// StepSeats are the three build seats, from their contracts.
|
||||
@@ -27,6 +31,24 @@ var StepSeats = []StepSeat{
|
||||
{Seat: nodebuildgo.Seat, Prefix: strings.TrimSuffix(nodebuildgo.SubjectStepEndedAll, "*")},
|
||||
{Seat: nodebuildnode.Seat, Prefix: strings.TrimSuffix(nodebuildnode.SubjectStepEndedAll, "*")},
|
||||
{Seat: nodebuildimage.Seat, Prefix: strings.TrimSuffix(nodebuildimage.SubjectStepEndedAll, "*")},
|
||||
// The seat mesh-builds answers the build step of a candidate, one entry named by its fingerprint (A-97, A-99).
|
||||
{Seat: builds.Seat, Prefix: strings.TrimSuffix(builds.SubjectStepEndedAll, "*"), Decode: decodeBuildEnded},
|
||||
}
|
||||
|
||||
// decodeBuildEnded reads the build step's end of the seat mesh-builds as the form the hearer holds each end to: its one
|
||||
// entry names the module, the commit and the fingerprint the step was asked at; an end naming another number of entries is
|
||||
// refused (A-99).
|
||||
func decodeBuildEnded(raw []byte) (StepEnded, error) {
|
||||
var b builds.StepEnded
|
||||
if err := json.Unmarshal(raw, &b); err != nil {
|
||||
return StepEnded{}, err
|
||||
}
|
||||
if len(b.Entries) != 1 {
|
||||
return StepEnded{}, fmt.Errorf("%w: ask %s ended a build of %d entries, asked of one", ErrRefused, b.ID, len(b.Entries))
|
||||
}
|
||||
e := b.Entries[0]
|
||||
return StepEnded{V: b.V, ID: b.ID, Pipeline: b.Pipeline, Step: b.Step, Module: e.Module, Commit: e.Commit, Fingerprint: e.Fingerprint,
|
||||
Node: b.Node, Outcome: b.Outcome, Said: b.Said, Rerun: b.Rerun, By: b.By, Started: b.Started, Ended: b.Ended}, nil
|
||||
}
|
||||
|
||||
// StepEnded is a build seat's step-ended. The three seats' bodies are one form, the part build-step's, held equal by
|
||||
|
||||
@@ -61,7 +61,7 @@ func TestManifest(t *testing.T) {
|
||||
{"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.
|
||||
{"consumes", equalSets(m.Consumes, []string{
|
||||
"mesh-delivery-indexer.take-in.taken", "mesh-builds.package-published", "mesh-builds.toolchain-agreed",
|
||||
"mesh-delivery-indexer.take-in.taken", "mesh-builds.package-published", "mesh-builds.toolchain-agreed", "mesh-builds.step-ended",
|
||||
"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-conditions.findings-asked"})},
|
||||
|
||||
@@ -30,6 +30,7 @@
|
||||
"mesh-delivery-indexer.take-in.taken",
|
||||
"mesh-builds.package-published",
|
||||
"mesh-builds.toolchain-agreed",
|
||||
"mesh-builds.step-ended",
|
||||
"operator-channel.decided",
|
||||
"node-build-go.step-ended",
|
||||
"node-build-node.step-ended",
|
||||
|
||||
Reference in New Issue
Block a user