diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go index d9faf0d0..85d21677 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -664,7 +664,7 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, // The composition step (design 47 G §22, §354): the seat mesh-wants' accepted kind step, run by one run ask per machine. if bus != nil && js != nil && inputs != nil && p.publishWants != nil { run := &compositionRun{inputs: &compositionInputs{machines: w.machinesOf, call: bus.CallKeptOn}, keeper: inputs, call: bus.CallKept, - trunk: trunkOf(iv, w.trunkName), running: w.runningOf, away: away, put: runAskPublisher(js), ends: ends, ask: bus.Call, now: time.Now, logf: logger.Printf} + trunk: trunkOf(iv, w.trunkName), running: w.runningOf, away: away, put: runAskPublisher(js), ends: ends, ask: bus.CallFile, now: time.Now, logf: logger.Printf} steps := &compositionSteps{node: thisNode, say: p.publishWants, run: run.run, now: time.Now, logf: logger.Printf, asker: stepAsker} if err := serveCompositionSteps(ctx, nc, steps); err != nil { logger.Printf("the composition step is not heard, so no composition step runs until the assigner starts again: %v", err) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go index fd61db8c..426c570e 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go @@ -480,6 +480,7 @@ func (b *fakeBus) CallKept(key string, body any) (json.RawMessage, error) { retu func (b *fakeBus) CallKeptOn(node, key string, body any) (json.RawMessage, error) { return b.CallOn(node, key, body) } +func (b *fakeBus) CallFile(key string, body any) (json.RawMessage, error) { return b.Call(key, body) } func (b *fakeBus) Emit(eventType string, body any) error { b.events = append(b.events, fmt.Sprintf("%s %+v", eventType, body)) return nil diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go index 30f5d9e3..ed4c9c78 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go @@ -44,6 +44,8 @@ type caller interface { CallKept(key string, body any) (json.RawMessage, error) // CallKeptOn is CallOn for a kept call, safe to call again (process.Conn.CallKeptOn; ADR 0668). CallKeptOn(node, key string, body any) (json.RawMessage, error) + // CallFile is Call of a verb marked chunks, its answer read whole (process.Conn.CallFile; design 47 G A-291). + CallFile(key string, body any) (json.RawMessage, error) Emit(eventType string, body any) error } diff --git a/modules/mesh-delivery-builder/buildrun/asks.go b/modules/mesh-delivery-builder/buildrun/asks.go index d65d845e..77abbc1e 100644 --- a/modules/mesh-delivery-builder/buildrun/asks.go +++ b/modules/mesh-delivery-builder/buildrun/asks.go @@ -58,6 +58,10 @@ type Asker struct { Expiry *Expiry // Logf says a drop that failed (the log's Printf when unset): the ask then stays on its stream, and it is said. Logf func(format string, args ...any) + // Outputs reads the answer a step's key names as its output-object, being over what the steps state holds (design 47 G + // A-290), through the build seat's verb output on the step's node (buildstep.Step.Whole); Wait answers the step whole. + // Nil reads none: such a step's wait is an error saying so. + Outputs buildstep.AskVerb } // StepBounds keeps one bound as a key with its own time to live: Arm writes key afresh to end after bound, End deletes it. @@ -332,7 +336,11 @@ func (k Asker) Wait(ctx context.Context, a buildstep.StepAsk) (buildstep.Step, e seat, _ := buildstep.BuildSeat(a.Language) step.Reason = TransportFault(FaultClass(step.Reason), seat, step.Node, step.Reason) } - return step, nil + whole, err := step.Whole(a.Language, a.Run != nil, k.Outputs) + if err != nil { + return buildstep.Step{}, fmt.Errorf("the ask %s's step on %s ended %s, and its answer %s cannot be read: %w", a.ID, step.Node, step.State, step.OutputObject, err) + } + return whole, nil } if k.Bound <= 0 { continue diff --git a/modules/mesh-delivery-builder/buildrun/asks_test.go b/modules/mesh-delivery-builder/buildrun/asks_test.go index 5f847802..9f00b53f 100644 --- a/modules/mesh-delivery-builder/buildrun/asks_test.go +++ b/modules/mesh-delivery-builder/buildrun/asks_test.go @@ -55,6 +55,23 @@ func TestAnAskerPutsOnceAndReadsTheStep(t *testing.T) { if _, err := k.Ask(context.Background(), goAsk("mvl0frv18d1958")); err == nil || !strings.Contains(err.Error(), "shanks refused: go 1.27 installed") { t.Fatalf("a refused step: %v", err) } + // Design 47 G A-290: a step whose key names its answer as the object output. is answered whole, read through the + // build seat's verb output on the step's node; with no verb to read it the wait says so, never the step without its answer. + f.Answer = func(a buildstep.StepAsk) buildstep.Step { + return buildstep.Step{ID: a.ID, Node: "ace", State: buildstep.StepPassed, OutputObject: buildstep.OutputObjectName(a.ID)} + } + var asked string + k.Outputs = func(key string, body any) (json.RawMessage, error) { + asked = key + return json.RawMessage(`{"ask":"` + body.(map[string]any)["ask"].(string) + `","file":"eyJhcnRpZmFjdHMiOltdfQ=="}`), nil + } + if step, err := k.Ask(context.Background(), goAsk("mvl0frv18d1961")); err != nil || string(step.Result) != `{"artifacts":[]}` || step.OutputObject != "" || asked != "seat:node-build-go.output@ace" { + t.Fatalf("a step whose answer is an object: %+v, %v, asked %q", step, err, asked) + } + k.Outputs = nil + if _, err := k.Ask(context.Background(), goAsk("mvl0frv18d1962")); err == nil || !strings.Contains(err.Error(), "no verb to read") { + t.Fatalf("a step whose answer is an object no verb reads: %v", err) + } // Issue 1966 (ADR 0668 §1, §2; 1748 note 27): the put is a kept call under the ask's id, made again through a deadline and // no responders until the stream takes it; a refusal of the stream is not made again; a put that meets a transport fault // past the bound says it in buildstep.TransportFaultWords, so the key it fails is the hand tool's to build again. diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go index d819103b..a9f662c1 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go @@ -293,7 +293,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro Asker: engineAsks{call: caller.CallOn}, Placed: placedNodes{kv: assignedNodes(&assignments)}.Placed, Bound: answerBound, - Sandbox: seatSandbox{asker: buildrun.Asker{Seats: buildSeats, Publisher: asks, Waiter: waiter, Bound: stepBound, Dropper: buildrun.SeatDrops{Caller: caller}}}, + Sandbox: seatSandbox{asker: buildrun.Asker{Seats: buildSeats, Publisher: asks, Waiter: waiter, Bound: stepBound, Dropper: buildrun.SeatDrops{Caller: caller}, Outputs: caller.CallFile}}, Verdicts: verdicts.all, // The node-engine's trunk head: a group with no node-engine member composes with the composer built there (ADR 0656 §5). Composer: func() string { return trunks.Heads()[check.NodeEngineRepository] }, @@ -321,7 +321,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro ch.Runs = repos // the node-engine a check reads beside at its running commit (ADR 0548 §6) ch.Says = verdictSays{js: js, node: cred.Node, now: time.Now} // a pull request's verdict event, on the stream asks (design 26 *Out of step*) ch.Judge = newJudge(judgeDeps{ - asks: buildrun.Asker{Seats: buildSeats, Publisher: asks, Waiter: waiter, Bound: stepBound, Dropper: buildrun.SeatDrops{Caller: caller}}, tools: tools, trunks: trunks, + asks: buildrun.Asker{Seats: buildSeats, Publisher: asks, Waiter: waiter, Bound: stepBound, Dropper: buildrun.SeatDrops{Caller: caller}, Outputs: caller.CallFile}, tools: tools, trunks: trunks, kept: kvStore{buckets[check.CheckerKeptBucket]}, previews: kvStore{buckets[check.PreviewsBucket]}, candidates: tryCandidates{candidates}, repos: repos, published: published, builds: buildStates, ends: ends, buildWait: set.buildWait, forge: forgeReads{call: caller.Call}, logf: logger.Printf, announce: announce, compositions: compositions, inputs: runInputs{awaitedKV{inputsKV}}, beginBound: beginBound, diff --git a/modules/mesh-delivery-checker/internal/check/examiner_test.go b/modules/mesh-delivery-checker/internal/check/examiner_test.go index 4307767c..7f6410d1 100644 --- a/modules/mesh-delivery-checker/internal/check/examiner_test.go +++ b/modules/mesh-delivery-checker/internal/check/examiner_test.go @@ -396,6 +396,12 @@ func TestACleanMergeIsFetchedAtItsTrunkAndBothLevelsPass(t *testing.T) { // 1679 s.State, s.Reason = buildstep.StepFailed, "the program does not build: undefined: rules.Nothing" return s, true }, fails: "by the checking program: its run ask", runs: 1}, + // Design 47 G A-290: a run whose program exited other than 0 ends its key failed, its answer kept; the checker reads it. + {name: "a program that exits other than 0", run: func(_ buildstep.StepAsk, s buildstep.Step) (buildstep.Step, bool) { + s.State, s.Reason = buildstep.StepFailed, "the program exited 2: bad flag" + s.Result, _ = json.Marshal(buildstep.RunAnswer{Key: "program-0123456789abcdef", Built: true, Exit: 2, Said: "bad flag"}) + return s, true + }, fails: "it exited 2 on shanks: bad flag", runs: 1}, {name: "a rule the change alters", at: head1, run: func(a buildstep.StepAsk, s buildstep.Step) (buildstep.Step, bool) { if a.Run.Args[0] != "-verdict" { return s, false diff --git a/modules/mesh-delivery-checker/internal/check/programs.go b/modules/mesh-delivery-checker/internal/check/programs.go index 5ef36dc7..2fd0a1e8 100644 --- a/modules/mesh-delivery-checker/internal/check/programs.go +++ b/modules/mesh-delivery-checker/internal/check/programs.go @@ -122,10 +122,14 @@ func (st *state) runProgram(ctx context.Context, args []string, body []byte) (js if err != nil { return nil, nil, fmt.Errorf("its run ask %s: %w", a.ID, err) } + var answer buildstep.RunAnswer if step.State != buildstep.StepPassed { + // A run whose program exited other than 0 ends its key failed with its answer kept (design 47 G A-290). + if json.Unmarshal(step.Result, &answer) == nil && answer.Exit != 0 { + return nil, nil, fmt.Errorf("it exited %d on %s: %s", answer.Exit, step.Node, answer.Said) + } return nil, nil, fmt.Errorf("its run ask %s ended %s on %s: %s", a.ID, step.State, step.Node, step.Reason) } - var answer buildstep.RunAnswer if err := json.Unmarshal(step.Result, &answer); err != nil { return nil, nil, fmt.Errorf("its run ask %s answered no run answer: %w", a.ID, err) } diff --git a/modules/mesh-delivery-checker/manifest_test.go b/modules/mesh-delivery-checker/manifest_test.go index d57e1e72..e883554b 100644 --- a/modules/mesh-delivery-checker/manifest_test.go +++ b/modules/mesh-delivery-checker/manifest_test.go @@ -135,6 +135,8 @@ func TestManifestDeclarations(t *testing.T) { // 1678/AC1 "seat:git.file", "seat:mesh-catalog.revision", "seat:mesh-catalog.seat-definitions", "seat:mesh-config.renames", "seat:mesh-hand-acts.hand-act", "seat:mesh-module-assigner.assignments", "seat:node-build-go.drop", "seat:node-build-node.drop", "seat:node-engine.compose-candidate", // The image build seat's drop too: a check of a module built by it is asked of it, and dropped of it (buildrun.SeatDrops). "seat:node-build-image.drop", + // A-290, A-291: a step's answer its key keeps as an object, read through each build seat's verb output. + "seat:node-build-go.output", "seat:node-build-node.output", "seat:node-build-image.output", // A-231: the check asks on each build seat's queues check and recheck. "seat:node-build-go.accept.check", "seat:node-build-go.accept.recheck", "seat:node-build-node.accept.check", "seat:node-build-node.accept.recheck", "seat:node-build-image.accept.check", "seat:node-build-image.accept.recheck"}; !reflect.DeepEqual(sorted(m.Invokes), sorted(want)) { diff --git a/modules/mesh-delivery-checker/module.json b/modules/mesh-delivery-checker/module.json index 8efc3476..adfd7767 100644 --- a/modules/mesh-delivery-checker/module.json +++ b/modules/mesh-delivery-checker/module.json @@ -97,6 +97,9 @@ "seat:node-build-node.drop", "seat:node-engine.compose-candidate", "seat:node-build-image.drop", + "seat:node-build-go.output", + "seat:node-build-node.output", + "seat:node-build-image.output", "seat:node-build-go.accept.check", "seat:node-build-go.accept.recheck", "seat:node-build-node.accept.check", 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 634ed0a9..94d6d352 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -482,7 +482,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // The validation's run ask (design 47 G §713, A-247, A-248), put under the indexer's own asker. ValidationAsks: buildrun.Asker{Name: index.ValidationAsker, Publisher: buildrun.JetStreamAsks{JS: js}}, // The build seat's verb output, for a validation's answer kept as an object (A-254). - AskVerb: calls.Call} + AskVerb: calls.CallFile} // The registrations and the changes that wait for their warrant (design 47 G §2, §667): their states are bound when they // appear; a warrant is heard on the router's decided event for this module, below. registry := newRegistry(js, conn, cred, calls.Call,