novox/mesh-delivery
10
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

A-290 B1: a step answer over the steps state's bound is kept as the object its key names

Browse Source 
A step's key keeps its answer only within what the steps state holds per value (kvlimit.ModuleStateValueBytes less
the headers' room). Over it, finish keeps the answer with the holder's KeepOutput as the object output.<ask> in its
outputs store and the key names it in output-object: a check's whole answer, its result emptied; a run's output, the
key keeping the run answer without it. An answer that cannot be kept ends the key failed, saying why. The Puller says
step-ended from the end finish wrote, so the key and the event say one outcome: a run whose program exited other than
0 ends its key failed as well, its answer kept; step-ended names the object finish kept instead of putting it twice.
Step.Whole reads a key's object back through the seat's verb output on the step's node. The part build-step's Step
gains output-object, the verb output says it answers a step's answer; fixtures and the generated contracts follow.

Builds: design 47 G A-290, A-254, §467

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
jschoubben
2026-11-12 16:44:07 +01:00
parent 9fdd8a76c2
commit 386095bd53
22 changed files with 409 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-contracts/go/seats/nodebuildgo/contract.go
+4 -2
View File
@@ -701,7 +701,9 @@ type Step struct {
Ended *time.Time `json:"ended,omitempty"`
ID string `json:"id"`
Node string `json:"node"`
Owner string `json:"owner,omitempty"`
// the name of the object in the build module's outputs store that holds the step's answer when the step's value would be over what the steps state holds (design 47 G A-290): a check's whole answer, its result then empty; a run's output, its result the run's answer without it. Read through the verb output
OutputObject string `json:"output-object,omitempty"`
Owner string `json:"owner,omitempty"`
// true when the step ended on a fault that passes (a transport fault), so a failed end after the last run says so
Passing bool `json:"passing,omitempty"`
Phase string `json:"phase,omitempty"`
@@ -1070,7 +1072,7 @@ type StepsAnswer struct {
// OutputInput is what output takes.
type OutputInput struct {
// the ask's id, the `output-object` of its step-ended naming it
// the ask's id, the `output-object` of its step key or its step-ended naming it
Ask string `json:"ask"`
// 128 random bits the caller chooses: the subject token of this send's chunks and acknowledgements
Call string `json:"call"`
modules/mesh-delivery-contracts/go/seats/nodebuildgo/contract_test.go
+2
View File
@@ -97,6 +97,8 @@ func TestFixtures(t *testing.T) {
err := json.Unmarshal(b, &v)
return v, err
}, subjectIs: nil},
{name: "a check's step whose answer is kept as an object (G A-290)", subject: "", given: "{\"ended\":\"2026-11-05T10:05:00Z\",\"id\":\"c3d4e5f6a7b8c9\",\"node\":\"ace\",\"output-object\":\"output.c3d4e5f6a7b8c9\",\"started\":\"2026-11-05T10:00:00Z\",\"state\":\"passed\",\"tool\":\"check\",\"v\":1}", wire: "{\"ended\":\"2026-11-05T10:05:00Z\",\"id\":\"c3d4e5f6a7b8c9\",\"node\":\"ace\",\"output-object\":\"output.c3d4e5f6a7b8c9\",\"started\":\"2026-11-05T10:00:00Z\",\"state\":\"passed\",\"tool\":\"check\",\"v\":1}",
decode: func(b []byte) (any, error) { var v StepsByNodeIDValue; err := json.Unmarshal(b, &v); return v, err }, subjectIs: nil},
{name: "drop names the ask and why", subject: "", given: "{\"by\":\"mesh-delivery-builder\",\"id\":\"a1b2\",\"why\":\"a newer fingerprint supersedes it\"}", wire: "{\"by\":\"mesh-delivery-builder\",\"id\":\"a1b2\",\"why\":\"a newer fingerprint supersedes it\"}",
decode: func(b []byte) (any, error) { var v DropInput; err := json.Unmarshal(b, &v); return v, err }, subjectIs: nil},
{name: "kept-artifact asks", subject: "", given: "{\"digest\":\"dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd\",\"fp\":\"src2:1480985e\",\"module\":\"artifact-store\"}", wire: "{\"digest\":\"dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd\",\"fp\":\"src2:1480985e\",\"module\":\"artifact-store\"}",
modules/mesh-delivery-contracts/go/seats/nodebuildimage/contract.go
+4 -2
View File
@@ -699,7 +699,9 @@ type Step struct {
Ended *time.Time `json:"ended,omitempty"`
ID string `json:"id"`
Node string `json:"node"`
Owner string `json:"owner,omitempty"`
// the name of the object in the build module's outputs store that holds the step's answer when the step's value would be over what the steps state holds (design 47 G A-290): a check's whole answer, its result then empty; a run's output, its result the run's answer without it. Read through the verb output
OutputObject string `json:"output-object,omitempty"`
Owner string `json:"owner,omitempty"`
// true when the step ended on a fault that passes (a transport fault), so a failed end after the last run says so
Passing bool `json:"passing,omitempty"`
Phase string `json:"phase,omitempty"`
@@ -1068,7 +1070,7 @@ type StepsAnswer struct {
// OutputInput is what output takes.
type OutputInput struct {
// the ask's id, the `output-object` of its step-ended naming it
// the ask's id, the `output-object` of its step key or its step-ended naming it
Ask string `json:"ask"`
// 128 random bits the caller chooses: the subject token of this send's chunks and acknowledgements
Call string `json:"call"`
modules/mesh-delivery-contracts/go/seats/nodebuildimage/contract_test.go
+2
View File
@@ -79,6 +79,8 @@ func TestFixtures(t *testing.T) {
err := json.Unmarshal(b, &v)
return v, err
}, subjectIs: nil},
{name: "a check's step whose answer is kept as an object (G A-290)", subject: "", given: "{\"ended\":\"2026-11-05T10:05:00Z\",\"id\":\"c3d4e5f6a7b8c9\",\"node\":\"ace\",\"output-object\":\"output.c3d4e5f6a7b8c9\",\"started\":\"2026-11-05T10:00:00Z\",\"state\":\"passed\",\"tool\":\"check\",\"v\":1}", wire: "{\"ended\":\"2026-11-05T10:05:00Z\",\"id\":\"c3d4e5f6a7b8c9\",\"node\":\"ace\",\"output-object\":\"output.c3d4e5f6a7b8c9\",\"started\":\"2026-11-05T10:00:00Z\",\"state\":\"passed\",\"tool\":\"check\",\"v\":1}",
decode: func(b []byte) (any, error) { var v StepsByNodeIDValue; err := json.Unmarshal(b, &v); return v, err }, subjectIs: nil},
{name: "drop names the ask and why", subject: "", given: "{\"by\":\"mesh-delivery-builder\",\"id\":\"a1b2\",\"why\":\"a newer fingerprint supersedes it\"}", wire: "{\"by\":\"mesh-delivery-builder\",\"id\":\"a1b2\",\"why\":\"a newer fingerprint supersedes it\"}",
decode: func(b []byte) (any, error) { var v DropInput; err := json.Unmarshal(b, &v); return v, err }, subjectIs: nil},
{name: "kept-artifact asks", subject: "", given: "{\"digest\":\"dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd\",\"fp\":\"src2:1480985e\",\"module\":\"artifact-store\"}", wire: "{\"digest\":\"dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd\",\"fp\":\"src2:1480985e\",\"module\":\"artifact-store\"}",
modules/mesh-delivery-contracts/go/seats/nodebuildnode/contract.go
+4 -2
View File
@@ -699,7 +699,9 @@ type Step struct {
Ended *time.Time `json:"ended,omitempty"`
ID string `json:"id"`
Node string `json:"node"`
Owner string `json:"owner,omitempty"`
// the name of the object in the build module's outputs store that holds the step's answer when the step's value would be over what the steps state holds (design 47 G A-290): a check's whole answer, its result then empty; a run's output, its result the run's answer without it. Read through the verb output
OutputObject string `json:"output-object,omitempty"`
Owner string `json:"owner,omitempty"`
// true when the step ended on a fault that passes (a transport fault), so a failed end after the last run says so
Passing bool `json:"passing,omitempty"`
Phase string `json:"phase,omitempty"`
@@ -1068,7 +1070,7 @@ type StepsAnswer struct {
// OutputInput is what output takes.
type OutputInput struct {
// the ask's id, the `output-object` of its step-ended naming it
// the ask's id, the `output-object` of its step key or its step-ended naming it
Ask string `json:"ask"`
// 128 random bits the caller chooses: the subject token of this send's chunks and acknowledgements
Call string `json:"call"`
modules/mesh-delivery-contracts/go/seats/nodebuildnode/contract_test.go
+2
View File
@@ -83,6 +83,8 @@ func TestFixtures(t *testing.T) {
err := json.Unmarshal(b, &v)
return v, err
}, subjectIs: nil},
{name: "a check's step whose answer is kept as an object (G A-290)", subject: "", given: "{\"ended\":\"2026-11-05T10:05:00Z\",\"id\":\"c3d4e5f6a7b8c9\",\"node\":\"ace\",\"output-object\":\"output.c3d4e5f6a7b8c9\",\"started\":\"2026-11-05T10:00:00Z\",\"state\":\"passed\",\"tool\":\"check\",\"v\":1}", wire: "{\"ended\":\"2026-11-05T10:05:00Z\",\"id\":\"c3d4e5f6a7b8c9\",\"node\":\"ace\",\"output-object\":\"output.c3d4e5f6a7b8c9\",\"started\":\"2026-11-05T10:00:00Z\",\"state\":\"passed\",\"tool\":\"check\",\"v\":1}",
decode: func(b []byte) (any, error) { var v StepsByNodeIDValue; err := json.Unmarshal(b, &v); return v, err }, subjectIs: nil},
{name: "drop names the ask and why", subject: "", given: "{\"by\":\"mesh-delivery-builder\",\"id\":\"a1b2\",\"why\":\"a newer fingerprint supersedes it\"}", wire: "{\"by\":\"mesh-delivery-builder\",\"id\":\"a1b2\",\"why\":\"a newer fingerprint supersedes it\"}",
decode: func(b []byte) (any, error) { var v DropInput; err := json.Unmarshal(b, &v); return v, err }, subjectIs: nil},
{name: "kept-artifact asks", subject: "", given: "{\"digest\":\"dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd\",\"fp\":\"src2:1480985e\",\"module\":\"artifact-store\"}", wire: "{\"digest\":\"dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd\",\"fp\":\"src2:1480985e\",\"module\":\"artifact-store\"}",
modules/mesh-delivery-contracts/parts/build-step.json
+6 -2
View File
@@ -243,13 +243,13 @@
"seat:mesh-delivery-checker",
"seat:mesh-pipeline-runner"
],
"description": "A run's answer by its ask, when the step-ended of the run names it as `output-object` because the answer is over what the event carries (design 47 G A-254): the answer's bytes in chunks of at most 512 KiB on the chunk subject of the call, each acknowledged by the caller on the ack subject within 30 seconds; the answer is of any size. An answer inline in the event is never asked here.",
"description": "A step's answer by its ask, when the step's key or a run's step-ended names it as `output-object` because the answer is over what the steps state's value or the event carries (design 47 G A-254, A-290): the answer's bytes in chunks of at most 512 KiB on the chunk subject of the call, each acknowledged by the caller on the ack subject within 30 seconds; the answer is of any size. An answer inline in the key or the event is never asked here.",
"input": {
"type": "object",
"properties": {
"ask": {
"type": "string",
"description": "the ask's id, the `output-object` of its step-ended naming it",
"description": "the ask's id, the `output-object` of its step key or its step-ended naming it",
"minLength": 1
},
"call": {
@@ -678,6 +678,10 @@
"result": {
"$ref": "#/defs/StepResult"
},
"output-object": {
"type": "string",
"description": "the name of the object in the build module's outputs store that holds the step's answer when the step's value would be over what the steps state holds (design 47 G A-290): a check's whole answer, its result then empty; a run's output, its result the run's answer without it. Read through the verb output"
},
"what": {
"type": "string"
},
modules/mesh-delivery-contracts/seats/node-build-go/fixtures/states.json
+14
View File
@@ -543,5 +543,19 @@
"state": "running",
"started": "2026-11-05T10:00:00Z"
}
},
{
"name": "a check's step whose answer is kept as an object (G A-290)",
"of": "state:steps/<node>.<id>",
"given": {
"v": 1,
"id": "c3d4e5f6a7b8c9",
"tool": "check",
"node": "ace",
"state": "passed",
"started": "2026-11-05T10:00:00Z",
"ended": "2026-11-05T10:05:00Z",
"output-object": "output.c3d4e5f6a7b8c9"
}
}
]
modules/mesh-delivery-contracts/seats/node-build-image/fixtures/states.json
+14
View File
@@ -203,5 +203,19 @@
"state": "running",
"started": "2026-11-05T10:00:00Z"
}
},
{
"name": "a check's step whose answer is kept as an object (G A-290)",
"of": "state:steps/<node>.<id>",
"given": {
"v": 1,
"id": "c3d4e5f6a7b8c9",
"tool": "check",
"node": "ace",
"state": "passed",
"started": "2026-11-05T10:00:00Z",
"ended": "2026-11-05T10:05:00Z",
"output-object": "output.c3d4e5f6a7b8c9"
}
}
]
modules/mesh-delivery-contracts/seats/node-build-node/fixtures/states.json
+14
View File
@@ -300,5 +300,19 @@
"state": "running",
"started": "2026-11-05T10:00:00Z"
}
},
{
"name": "a check's step whose answer is kept as an object (G A-290)",
"of": "state:steps/<node>.<id>",
"given": {
"v": 1,
"id": "c3d4e5f6a7b8c9",
"tool": "check",
"node": "ace",
"state": "passed",
"started": "2026-11-05T10:00:00Z",
"ended": "2026-11-05T10:05:00Z",
"output-object": "output.c3d4e5f6a7b8c9"
}
}
]
modules/mesh-delivery-contracts/ts/seats/nodebuildgo/parts/build-step.d.ts
Vendored
+3 -1
View File
@@ -332,6 +332,8 @@ export interface Step {
"ended"?: string;
"id": string;
"node": string;
/** the name of the object in the build module's outputs store that holds the step's answer when the step's value would be over what the steps state holds (design 47 G A-290): a check's whole answer, its result then empty; a run's output, its result the run's answer without it. Read through the verb output */
"output-object"?: string;
"owner"?: string;
/** true when the step ended on a fault that passes (a transport fault), so a failed end after the last run says so */
"passing"?: boolean;
@@ -661,7 +663,7 @@ export interface StepsAnswer {
/** is what output takes. */
export interface OutputInput {
/** the ask's id, the `output-object` of its step-ended naming it */
/** the ask's id, the `output-object` of its step key or its step-ended naming it */
"ask": string;
/** 128 random bits the caller chooses: the subject token of this send's chunks and acknowledgements */
"call": string;
modules/mesh-delivery-contracts/ts/seats/nodebuildimage/parts/build-step.d.ts
Vendored
+3 -1
View File
@@ -332,6 +332,8 @@ export interface Step {
"ended"?: string;
"id": string;
"node": string;
/** the name of the object in the build module's outputs store that holds the step's answer when the step's value would be over what the steps state holds (design 47 G A-290): a check's whole answer, its result then empty; a run's output, its result the run's answer without it. Read through the verb output */
"output-object"?: string;
"owner"?: string;
/** true when the step ended on a fault that passes (a transport fault), so a failed end after the last run says so */
"passing"?: boolean;
@@ -661,7 +663,7 @@ export interface StepsAnswer {
/** is what output takes. */
export interface OutputInput {
/** the ask's id, the `output-object` of its step-ended naming it */
/** the ask's id, the `output-object` of its step key or its step-ended naming it */
"ask": string;
/** 128 random bits the caller chooses: the subject token of this send's chunks and acknowledgements */
"call": string;
modules/mesh-delivery-contracts/ts/seats/nodebuildnode/parts/build-step.d.ts
Vendored
+3 -1
View File
@@ -332,6 +332,8 @@ export interface Step {
"ended"?: string;
"id": string;
"node": string;
/** the name of the object in the build module's outputs store that holds the step's answer when the step's value would be over what the steps state holds (design 47 G A-290): a check's whole answer, its result then empty; a run's output, its result the run's answer without it. Read through the verb output */
"output-object"?: string;
"owner"?: string;
/** true when the step ended on a fault that passes (a transport fault), so a failed end after the last run says so */
"passing"?: boolean;
@@ -661,7 +663,7 @@ export interface StepsAnswer {
/** is what output takes. */
export interface OutputInput {
/** the ask's id, the `output-object` of its step-ended naming it */
/** the ask's id, the `output-object` of its step key or its step-ended naming it */
"ask": string;
/** 128 random bits the caller chooses: the subject token of this send's chunks and acknowledgements */
"call": string;
modules/mesh-delivery-sdk/buildstep/stepoutput.go
+49 -9
View File
@@ -9,7 +9,7 @@ import (
"fmt"
)
// OutputVerb is the verb of a build seat that answers a run's answer kept as an object (design 47 G A-254).
// OutputVerb is the verb of a build seat that answers a step's answer kept as an object (design 47 G A-254, A-290).
const OutputVerb = "output"
// AskVerb asks a seat verb as the runtime calls it (stdio.Ask): the key `seat:<seat>.<verb>[@<node>]` and the body in, the
@@ -23,20 +23,60 @@ func (e StepEnded) ReadOutput(language string, ask AskVerb) ([]byte, error) {
if e.OutputObject == "" {
return []byte(e.Output), nil
}
return readObject(language, e.Node, e.Ask, e.OutputObject, ask)
}
// Whole is the step with its answer whole (design 47 G A-290): a step whose key names its answer as the object
// `output-object`, its value being over what the steps state holds, has it read through the verb `output` of the seat of
// language on the step's node, and put back where the key holds the answer itself: a run's (run true) as its run answer's
// output, any other's as its result. A step whose key holds its answer is answered as it is. Every reader of a step's answer
// reads it so, the checker and the builder alike (no step reads passed in one place and failed in another).
func (s Step) Whole(language string, run bool, ask AskVerb) (Step, error) {
if s.OutputObject == "" {
return s, nil
}
body, err := readObject(language, s.Node, s.ID, s.OutputObject, ask)
if err != nil {
return s, err
}
if !run {
if !json.Valid(body) {
return s, fmt.Errorf("buildstep: the answer %s of the ask %s is not JSON", s.OutputObject, s.ID)
}
s.Result, s.OutputObject = json.RawMessage(body), ""
return s, nil
}
var answer RunAnswer
if len(s.Result) > 0 {
if err := json.Unmarshal(s.Result, &answer); err != nil {
return s, fmt.Errorf("buildstep: the run answer of the ask %s is not one: %w", s.ID, err)
}
}
answer.Output = json.RawMessage(body)
raw, err := json.Marshal(answer)
if err != nil {
return s, fmt.Errorf("buildstep: the run answer of the ask %s with its output %s: %w", s.ID, s.OutputObject, err)
}
s.Result, s.OutputObject = raw, ""
return s, nil
}
// readObject reads the object a build seat's holder on node keeps of the ask askID through the seat's verb `output`.
func readObject(language, node, askID, object string, ask AskVerb) ([]byte, error) {
seat, err := BuildSeat(language)
if err != nil {
return nil, err
}
if ask == nil {
return nil, errors.New("buildstep: the answer is in the output object " + e.OutputObject + ", which this reader has no verb to read")
return nil, errors.New("buildstep: the answer is in the output object " + object + ", which this reader has no verb to read")
}
var call [16]byte
if _, err := rand.Read(call[:]); err != nil {
return nil, err
}
raw, err := ask("seat:"+seat+"."+OutputVerb+"@"+e.Node, map[string]any{"ask": e.Ask, "call": hex.EncodeToString(call[:])})
raw, err := ask("seat:"+seat+"."+OutputVerb+"@"+node, map[string]any{"ask": askID, "call": hex.EncodeToString(call[:])})
if err != nil {
return nil, fmt.Errorf("buildstep: asking %s for the answer %s of the ask %s: %w", seat, e.OutputObject, e.Ask, err)
return nil, fmt.Errorf("buildstep: asking %s for the answer %s of the ask %s: %w", seat, object, askID, err)
}
var got struct {
Ask string `json:"ask"`
@@ -44,17 +84,17 @@ func (e StepEnded) ReadOutput(language string, ask AskVerb) ([]byte, error) {
Chunks int `json:"chunks"`
}
if err := json.Unmarshal(raw, &got); err != nil {
return nil, fmt.Errorf("buildstep: the verb output of %s answered no answer for %s: %w", seat, e.Ask, err)
return nil, fmt.Errorf("buildstep: the verb output of %s answered no answer for %s: %w", seat, askID, err)
}
if got.Ask != e.Ask {
return nil, fmt.Errorf("buildstep: the verb output of %s answered the ask %q, asked %s", seat, got.Ask, e.Ask)
if got.Ask != askID {
return nil, fmt.Errorf("buildstep: the verb output of %s answered the ask %q, asked %s", seat, got.Ask, askID)
}
if got.File == "" {
return nil, fmt.Errorf("buildstep: %s on %s keeps no answer of the ask %s", seat, e.Node, e.Ask)
return nil, fmt.Errorf("buildstep: %s on %s keeps no answer of the ask %s", seat, node, askID)
}
body, err := base64.StdEncoding.DecodeString(got.File)
if err != nil {
return nil, fmt.Errorf("buildstep: the answer of %s in no base64: %w", e.Ask, err)
return nil, fmt.Errorf("buildstep: the answer of %s in no base64: %w", askID, err)
}
return body, nil
}
modules/mesh-delivery-sdk/buildstep/stepoutput_test.go
+48
View File
@@ -1,6 +1,7 @@
package buildstep
import (
"encoding/base64"
"encoding/json"
"errors"
"strings"
@@ -61,3 +62,50 @@ func TestStepEndedRefusesAnswerTwice(t *testing.T) {
t.Fatalf("err %v", err)
}
}
// Design 47 G A-290: a step whose key names its answer as an object is read whole through the seat's verb output on its
// node: a check's answer as its result, a run's output put back in its run answer; a key holding its answer is answered as it is.
func TestStepWholeReadsTheObjectItsKeyNames(t *testing.T) {
const ask = "m2kx9a1b3c4d5f"
var key string
verb := func(file string, err error) AskVerb {
return func(k string, _ any) (json.RawMessage, error) {
key = k
return json.RawMessage(`{"ask":"` + ask + `","file":"` + file + `"}`), err
}
}
b64 := func(s string) string { return base64.StdEncoding.EncodeToString([]byte(s)) }
named := Step{ID: ask, Node: "ace", State: StepPassed, OutputObject: OutputObjectName(ask)}
for _, c := range []struct {
name string
step Step
run bool
ask AskVerb
want, fail string
}{
{name: "the key holds it", step: Step{ID: ask, Node: "ace", Result: json.RawMessage(`{"a":1}`)}, want: `{"a":1}`},
{name: "a check's answer", step: named, ask: verb(b64(`{"fetch":{"workspace":"w"}}`), nil), want: `{"fetch":{"workspace":"w"}}`},
{name: "a run's output", step: func() Step { s := named; s.Result = json.RawMessage(`{"built":true,"key":"k","exit":0}`); return s }(), run: true,
ask: verb(b64(`{"verdict":"ok"}`), nil), want: `{"built":true,"key":"k","exit":0,"output":{"verdict":"ok"}}`},
{name: "a check's answer that is no JSON", step: named, ask: verb(b64(`not json`), nil), fail: "not JSON"},
{name: "the verb fails", step: named, ask: verb("", errors.New("no responders")), fail: "no responders"},
{name: "no verb", step: named, fail: "no verb"},
} {
t.Run(c.name, func(t *testing.T) {
key = ""
got, err := c.step.Whole("go", c.run, c.ask)
if c.fail != "" {
if err == nil || !strings.Contains(err.Error(), c.fail) {
t.Fatalf("err %v, want %q", err, c.fail)
}
return
}
if err != nil || string(got.Result) != c.want || got.OutputObject != "" {
t.Fatalf("got %s (object %q), %v; want %s", got.Result, got.OutputObject, err, c.want)
}
if c.ask != nil && key != "seat:node-build-go.output@ace" {
t.Fatalf("asked %q", key)
}
})
}
}
modules/mesh-delivery-sdk/buildstep/steps.go
+9 -4
View File
@@ -49,10 +49,15 @@ type Step struct {
Ended time.Time `json:"ended,omitzero"`
Reason string `json:"reason,omitempty"`
Result json.RawMessage `json:"result,omitempty"`
What string `json:"what,omitempty"`
Phase string `json:"phase,omitempty"`
PhaseAt time.Time `json:"phase-at,omitzero"`
Owner string `json:"owner,omitempty"`
// OutputObject names the object in the build module's outputs store (OutputObjectName) that holds the step's answer
// when the whole value would be over what the steps state holds (design 47 G A-290): a check's whole answer, Result
// then empty; a run's output, Result the run's answer without it. A reader reads the answer through the seat's verb
// output (Step.Answer).
OutputObject string `json:"output-object,omitempty"`
What string `json:"what,omitempty"`
Phase string `json:"phase,omitempty"`
PhaseAt time.Time `json:"phase-at,omitzero"`
Owner string `json:"owner,omitempty"`
// Run is which run of the ask this is, counted by the bus's delivery count (1 for the first); AgainAt when the bus
// gives it again (the state again); Passing says the step ended on a transport fault, so a failed end after the
// last run says so (ADR 0664 §1, §2).
modules/mesh-delivery-sdk/steprun/askers.go
+1 -1
View File
@@ -106,7 +106,7 @@ func (p *Puller) refuseAsker(msg *Pulled, ask sdkstep.StepAsk, why string) {
}
p.logf("the ask %s is refused on %s: %s", ask.ID, p.Node, why)
if ask.Step == nil {
p.sayEnded(msg, ask, sdkstep.OutcomeFailed, why, nil)
p.sayEnded(msg, ask, sdkstep.OutcomeFailed, why, nil, "")
} else if end, err := sdkstep.EndOf(ask, p.Node, sdkstep.OutcomeFailed); err == nil {
// A pipeline step's end names its pipeline, step, commit and fingerprint (its own form, §602).
end.Said, end.At, end.Asker = why, now().UTC(), askerOf(msg, ask)
modules/mesh-delivery-sdk/steprun/digestshape_test.go
+3 -3
View File
@@ -119,7 +119,7 @@ func TestADigestOfAPackageNamedTokenIsStoredAsAList(t *testing.T) {
t.Fatalf("the rule does not refuse the form before issue 1930 (%s): the test proves nothing", old)
}
st := newRuntimeState()
if err := finish(st.put, "shanks.a1", sdkstep.Step{ID: "a1", Tool: "check ask", Node: "shanks", State: sdkstep.StepRunning, Started: time.Now()}, out, nil, t.Logf); err != nil {
if _, err := finish(st.put, nil, "shanks.a1", sdkstep.Step{ID: "a1", Tool: "check ask", Node: "shanks", State: sdkstep.StepRunning, Started: time.Now()}, out, nil, t.Logf); err != nil {
t.Fatal(err)
}
s, ok := st.last("shanks.a1")
@@ -160,7 +160,7 @@ func TestARefusedEndIsSaidAndEndsTheStepFailed(t *testing.T) {
}
var logged []string
logf := func(format string, args ...any) { logged = append(logged, fmt.Sprintf(format, args...)) }
err := finish(put, "shanks.a2", sdkstep.Step{ID: "a2", Tool: "check ask", Node: "shanks", State: sdkstep.StepRunning, Started: time.Now()}, map[string]any{"levels": 1}, nil, logf)
_, err := finish(put, nil, "shanks.a2", sdkstep.Step{ID: "a2", Tool: "check ask", Node: "shanks", State: sdkstep.StepRunning, Started: time.Now()}, map[string]any{"levels": 1}, nil, logf)
if err != nil {
t.Fatalf("finish answered %v, want the small end written", err)
}
@@ -176,7 +176,7 @@ func TestARefusedEndIsSaidAndEndsTheStepFailed(t *testing.T) {
t.Errorf("logged %q, want the refusal said", logged)
}
none := func(string, any) (uint64, error) { return 0, errors.New("the bus is gone") }
if err := finish(none, "shanks.a3", sdkstep.Step{ID: "a3", Tool: "check ask", Node: "shanks", Started: time.Now()}, nil, nil, logf); err == nil || !strings.Contains(err.Error(), "the bus is gone") {
if _, err := finish(none, nil, "shanks.a3", sdkstep.Step{ID: "a3", Tool: "check ask", Node: "shanks", Started: time.Now()}, nil, nil, logf); err == nil || !strings.Contains(err.Error(), "the bus is gone") {
t.Errorf("finish with no write possible answered %v, want the error", err)
}
}
modules/mesh-delivery-sdk/steprun/pull.go
+25 -28
View File
@@ -81,9 +81,9 @@ type Puller struct {
// every ProgressEvery while it runs, step-ended when an ask outside a pipeline ends (design 47 G §496, §602). Nil says
// none, a test's. ProgressEvery defaults to DefaultProgressEvery.
Say func(event string, body any, id string) error
// KeepOutput keeps a run's answer over what step-ended carries inline (sdkstep.StepEndedInlineMost) as the object named
// object, which the seat's verb `output` answers (design 47 G A-254); a holder keeps it in the object store its manifest
// requires. Nil keeps none: such an answer ends the ask failed, saying so.
// KeepOutput keeps a step's answer over what its key holds in the steps state (a check's or a run's, design 47 G A-290)
// as the object named object, which the key and a run's step-ended name and the seat's verb `output` answers (A-254); a
// holder keeps it in the object store its manifest requires. Nil keeps none: such an answer ends the ask failed, saying so.
KeepOutput func(object string, body []byte) error
// Callers says which seats the build seat's definition lists as callers of an accepted kind (CallersFromDefinitions);
// found is false while the seat list holds no such kind. Nil checks no named asker (design 47 G §543, A-163).
@@ -673,7 +673,7 @@ func (p *Puller) take(ctx context.Context, msg *Pulled) {
// (issue 1907): the asker's key fails with why.
why := fmt.Errorf("the ask was taken %d times and no run of it ended: each took its build node down or died with it, so it is run no more (poison)", msg.Delivered-1)
p.logf("the ask %s is ended on %s: %v", ask.ID, p.Node, why)
_ = finish(p.Put, key, step, nil, why, p.Logf) // run no more either way: the asker's bound says it then
_, _ = finish(p.Put, p.KeepOutput, key, step, nil, why, p.Logf) // run no more either way: the asker's bound says it then
_ = msg.Term()
return
}
@@ -757,7 +757,7 @@ func (p *Puller) run(ctx context.Context, msg *Pulled, ask sdkstep.StepAsk, key
var dec *Declining
if errors.As(err, &dec) && ctx.Err() == nil {
p.declineRun(msg, ask, key, step, dec)
p.sayEnded(msg, ask, sdkstep.OutcomeCouldNotRun, dec.Text, nil)
p.sayEnded(msg, ask, sdkstep.OutcomeCouldNotRun, dec.Text, nil, "")
return
}
if ctx.Err() != nil {
@@ -782,22 +782,28 @@ func (p *Puller) run(ctx context.Context, msg *Pulled, ask sdkstep.StepAsk, key
p.logf("the ask %s ended on a transport fault and its step could not be written again (%v): it ends failed", ask.ID, aerr)
}
}
if ferr := finish(p.Put, key, step, result, err, p.Logf); ferr != nil {
written, ferr := finish(p.Put, p.KeepOutput, key, step, result, err, p.Logf)
if ferr != nil {
// No end was written: the ask is not acknowledged as done, so the stream gives it again and a run that can
// write its end (or the poison rule, issue 1907) ends it where its asker sees it (issue 1930).
p.logf("the ask %s ran on %s and no end of it could be written, so it is given back: %v", ask.ID, p.Node, ferr)
_ = msg.Nak()
return
}
// step-ended says the end the key holds (design 47 G A-290): its outcome, and the answer only when the key kept it, so no
// step reads passed in one place and failed in another.
if len(written.Result) == 0 && written.OutputObject == "" {
result = nil
}
outcome := "passed"
if err != nil {
outcome = "did not pass: " + err.Error()
if written.State != sdkstep.StepPassed {
outcome = "did not pass: " + written.Reason
if len(outcome) > 300 {
outcome = outcome[:300] + "..."
}
p.sayEnded(msg, ask, sdkstep.OutcomeFailed, err.Error(), nil)
p.sayEnded(msg, ask, sdkstep.OutcomeFailed, written.Reason, result, written.OutputObject)
} else {
p.sayEnded(msg, ask, sdkstep.OutcomePassed, "", result)
p.sayEnded(msg, ask, sdkstep.OutcomePassed, "", result, written.OutputObject)
}
p.logf("the %s ask %s ended on %s after %s, in its phase %q; %s", ask.Kind, ask.ID, p.Node,
now().Sub(step.Started).Round(time.Second), step.Phase, outcome)
@@ -981,25 +987,24 @@ func (p *Puller) say(event string, body any, id string) {
// sayEnded says step-ended for an ask outside a pipeline, whatever its asker (design 47 G §602): a pipeline step's end is
// its run's own (AskSteps.SayStep). The end of a run carries the run's answer (A-254): the last line of its program's
// standard output, inline as `output` up to sdkstep.StepEndedInlineMost, else as `output-object`, kept by KeepOutput; a
// program that exits other than 0 ends failed with its answer carried.
func (p *Puller) sayEnded(msg *Pulled, ask sdkstep.StepAsk, outcome, said string, result any) {
// standard output, inline as `output` up to sdkstep.StepEndedInlineMost, else as `output-object`, the object kept, the step's
// key naming it (design 47 G A-290: finish kept it with KeepOutput, so it is not put twice); a program that exits other than 0
// ends failed with its answer carried.
func (p *Puller) sayEnded(msg *Pulled, ask sdkstep.StepAsk, outcome, said string, result any, kept string) {
if p.Say == nil || ask.Step != nil {
return
}
end := sdkstep.StepEnded{ID: ask.ID, Asker: askerOf(msg, ask), Ask: ask.ID, Node: p.Node, Outcome: outcome, At: now().UTC()}
if ans, ok := runAnswerOf(ask, result); ok {
if ans.Exit != 0 && outcome == sdkstep.OutcomePassed {
outcome, said = sdkstep.OutcomeFailed, fmt.Sprintf("the program exited %d: %s", ans.Exit, ans.Said)
outcome, said = sdkstep.OutcomeFailed, exitedWhy(ans)
}
if len(ans.Output) <= sdkstep.StepEndedInlineMost {
end.Output = string(ans.Output)
} else if object := sdkstep.OutputObjectName(ask.ID); p.KeepOutput == nil {
outcome, said = sdkstep.OutcomeFailed, fmt.Sprintf("the run's answer is %d bytes, over what the end carries, and this holder keeps no object for it", len(ans.Output))
} else if err := p.KeepOutput(object, ans.Output); err != nil {
outcome, said = sdkstep.OutcomeFailed, fmt.Sprintf("the run's answer of %d bytes could not be kept: %v", len(ans.Output), err)
} else if kept != "" {
end.OutputObject = kept
} else {
end.OutputObject = object
outcome, said = sdkstep.OutcomeFailed, fmt.Sprintf("the run's answer is %d bytes, over what the end carries, and no object of it was kept", len(ans.Output))
}
}
if len(said) > 300 {
@@ -1022,13 +1027,5 @@ func runAnswerOf(ask sdkstep.StepAsk, result any) (sdkstep.RunAnswer, bool) {
if ask.Run == nil {
return sdkstep.RunAnswer{}, false
}
switch a := result.(type) {
case sdkstep.RunAnswer:
return a, true
case *sdkstep.RunAnswer:
if a != nil {
return *a, true
}
}
return sdkstep.RunAnswer{}, false
return runAnswerIn(result)
}
modules/mesh-delivery-sdk/steprun/pull_integration_test.go
+67
View File
@@ -4,7 +4,10 @@ package steprun
import (
"context"
"encoding/base64"
"encoding/json"
"errors"
"fmt"
"strings"
"sync"
"testing"
@@ -14,6 +17,7 @@ import (
"github.com/nats-io/nats.go/jetstream"
sdkstep "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildstep"
"git.novox.be/novox/mesh-sdk/go/kvlimit"
"git.novox.be/novox/mesh-sdk/go/testbus"
)
@@ -233,3 +237,66 @@ func TestTheTwoWorkersTakeTheTwoFormsOfTheAcceptSubject(t *testing.T) {
}
}
}
// Design 47 G A-290 on the real server: a check's answer over what the steps state holds per value (a bucket made with
// kvlimit.ModuleStateValueBytes, as every module's is) is kept as the object output.<ask> in the outputs store, the step's key
// names it and ends passed, and the answer reads back whole through the seat's verb output (Step.Whole).
func TestACheckAnswerOverTheStepsStateIsKeptAndReadBackWhole(t *testing.T) {
js := askBus(t, 30*time.Second)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "steps-a290", MaxValueSize: kvlimit.ModuleStateValueBytes})
if err != nil {
t.Fatal(err)
}
objects, err := js.CreateObjectStore(ctx, jetstream.ObjectStoreConfig{Bucket: "outputs-a290"})
if err != nil {
t.Fatal(err)
}
put := func(key string, value any) (uint64, error) {
raw, err := json.Marshal(value)
if err != nil {
return 0, err
}
return kv.Put(ctx, key, raw)
}
answer := map[string]any{"fetch": map[string]any{"workspace": "/w"}, "levels": []any{map[string]any{"level": "unit", "said": strings.Repeat("z", 2*kvlimit.ModuleStateValueBytes)}}}
a := testAsk(sdkstep.AskCheck)
publishAsk(t, js, a)
p := &Puller{Source: &JetStreamAsks{JS: js, Language: sdkstep.AskGo}, Language: sdkstep.AskGo, Node: "ace", Put: put,
Capacity: func() int { return 1 }, PullWait: 100 * time.Millisecond,
KeepOutput: func(object string, body []byte) error { _, err := objects.PutBytes(ctx, object, body); return err },
Run: func(context.Context, sdkstep.StepAsk) (any, error) { return answer, nil }}
go p.Pull(ctx)
key, _ := sdkstep.AskStepKey("ace", a.ID)
var step sdkstep.Step
waitFor(t, "the check's step ended", func() bool {
e, err := kv.Get(ctx, key)
if err != nil {
return false
}
step, err = sdkstep.DecodeStep(e.Value())
return err == nil && sdkstep.Ended(step)
})
if step.State != sdkstep.StepPassed || step.OutputObject != sdkstep.OutputObjectName(a.ID) || len(step.Result) != 0 {
t.Fatalf("the key ended %s (%s) naming %q with %d bytes of result, want passed, naming the object and no result", step.State, step.Reason, step.OutputObject, len(step.Result))
}
verb := func(k string, body any) (json.RawMessage, error) {
if k != "seat:node-build-go.output@ace" {
return nil, fmt.Errorf("asked %s", k)
}
data, err := objects.GetBytes(ctx, sdkstep.OutputObjectName(body.(map[string]any)["ask"].(string)))
if err != nil {
return nil, err
}
return json.Marshal(map[string]any{"ask": a.ID, "file": base64.StdEncoding.EncodeToString(data)})
}
whole, err := step.Whole(sdkstep.AskGo, false, verb)
if err != nil {
t.Fatal(err)
}
want, _ := json.Marshal(answer)
if string(whole.Result) != string(want) {
t.Fatalf("read back %d bytes, want the whole answer of %d", len(whole.Result), len(want))
}
}
modules/mesh-delivery-sdk/steprun/run_output_test.go
+60 -14
View File
@@ -10,50 +10,69 @@ import (
"time"
sdkstep "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildstep"
"git.novox.be/novox/mesh-sdk/go/kvlimit"
)
// Design 47 G A-254: a build seat's step-ended of a run carries the run's answer, the last line of its program's standard output:
// inline as `output` up to the event bound, else as `output-object`, the object the holder keeps, which the verb `output` answers.
// Design 47 G A-254, A-290: a build seat's step-ended of a run carries the run's answer, the last line of its program's standard
// output: inline as `output` up to the event bound, else as `output-object`, the object the holder keeps, which the verb `output`
// answers. The step's key holds the answer only within what the steps state holds per value: over it, a check's whole answer or a
// run's output is the object output.<ask> the key names, kept once; the key and step-ended always say the same outcome, a run
// whose program exited other than 0 failed in both, its answer kept (A-290).
func TestStepEndedOfARunCarriesItsAnswer(t *testing.T) {
big := `"` + strings.Repeat("x", sdkstep.StepEndedInlineMost) + `"`
overKey := `"` + strings.Repeat("y", kvlimit.ModuleStateValueBytes) + `"` // over the steps state, within the event
checkAnswer := map[string]any{"fetch": map[string]any{"workspace": "/w"}, "levels": []any{map[string]any{"level": "unit", "said": strings.Repeat("z", kvlimit.ModuleStateValueBytes)}}}
for _, c := range []struct {
name string
check bool // a check ask answering checkAnswer, not a run
answer sdkstep.RunAnswer
keepErr error
outcome string
inline string
objectKept bool
objectKept bool // step-ended names the object
keyObject string // what the object the key names holds: "" none, else the answer's bytes
said string
}{
{name: "a small answer is inline", answer: sdkstep.RunAnswer{Output: json.RawMessage(`{"verdict":true}`)}, outcome: sdkstep.OutcomePassed, inline: `{"verdict":true}`},
{name: "an answer over the event bound is an object", answer: sdkstep.RunAnswer{Output: json.RawMessage(big)}, outcome: sdkstep.OutcomePassed, objectKept: true},
{name: "an answer over the event bound is an object", answer: sdkstep.RunAnswer{Output: json.RawMessage(big)}, outcome: sdkstep.OutcomePassed, objectKept: true, keyObject: big},
{name: "an answer over the steps state within the event is inline and the object the key names", answer: sdkstep.RunAnswer{Built: true, Key: "k", Output: json.RawMessage(overKey)},
outcome: sdkstep.OutcomePassed, inline: overKey, keyObject: overKey},
{name: "a program that exits other than 0 failed, its answer kept", answer: sdkstep.RunAnswer{Exit: 3, Output: json.RawMessage(`"no"`), Said: "bad input"},
outcome: sdkstep.OutcomeFailed, inline: `"no"`, said: "the program exited 3: bad input"},
{name: "an answer the holder cannot keep is said", answer: sdkstep.RunAnswer{Output: json.RawMessage(big)}, keepErr: errors.New("store full"),
outcome: sdkstep.OutcomeFailed, said: "store full"},
{name: "a check's answer over the steps state is the object the key names", check: true, outcome: sdkstep.OutcomePassed, keyObject: "check"},
{name: "a check's answer the holder cannot keep fails in its key and its end alike", check: true, keepErr: errors.New("store full"),
outcome: sdkstep.OutcomeFailed, said: "store full"},
} {
t.Run(c.name, func(t *testing.T) {
src, st, sy := NewFakeAsks(), newSteps(), &sayer{}
a := testAsk(sdkstep.AskCheck)
a.Check = nil
in, _ := sdkstep.RunInputOf("seat:mesh-catalog.input", []byte(`{"v":1}`))
commit := "0123456789abcdef0123456789abcdef01234567"
a.Run = &sdkstep.RunJob{Program: sdkstep.RunProgram{Repository: "novox/mesh-delivery", Commit: commit, Dir: "modules/x", Package: "./cmd/x"}, Input: in}
var answer any = c.answer
if c.check {
answer = checkAnswer
} else {
a.Check = nil
in, _ := sdkstep.RunInputOf("seat:mesh-catalog.input", []byte(`{"v":1}`))
commit := "0123456789abcdef0123456789abcdef01234567"
a.Run = &sdkstep.RunJob{Program: sdkstep.RunProgram{Repository: "novox/mesh-delivery", Commit: commit, Dir: "modules/x", Package: "./cmd/x"}, Input: in}
}
_ = src.PublishAsk(a)
var mu sync.Mutex
kept := map[string][]byte{}
kept, keeps := map[string][]byte{}, 0
p := &Puller{Source: src, Language: sdkstep.AskGo, Node: "shanks", Put: st.put, Capacity: func() int { return 1 },
PullWait: 10 * time.Millisecond, Heartbeat: time.Hour, Say: sy.say,
KeepOutput: func(object string, body []byte) error {
mu.Lock()
defer mu.Unlock()
keeps++
if c.keepErr != nil {
return c.keepErr
}
kept[object] = body
return nil
},
Run: func(ctx context.Context, a sdkstep.StepAsk) (any, error) { return c.answer, nil }}
Run: func(ctx context.Context, a sdkstep.StepAsk) (any, error) { return answer, nil }}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go p.Pull(ctx)
@@ -63,20 +82,47 @@ func TestStepEndedOfARunCarriesItsAnswer(t *testing.T) {
t.Fatalf("outcome %v, want %s (%v)", end["outcome"], c.outcome, end)
}
if c.inline != "" && end["output"] != c.inline {
t.Fatalf("output %v, want %s inline", end["output"], c.inline)
t.Fatalf("output %.80v, want %.80s inline", end["output"], c.inline)
}
if c.said != "" && !strings.Contains(end["said"].(string), c.said) {
t.Fatalf("said %v, want it to hold %q", end["said"], c.said)
}
key, _ := sdkstep.StepKey("shanks", a.ID)
step, _ := st.get(key)
if raw, err := sdkstep.EncodeStep(step); err != nil || kvlimit.Check("steps", key, len(raw), kvlimit.ModuleStateValueBytes) != nil {
t.Fatalf("the step's key is %d bytes (%v), over what the steps state holds", len(raw), err)
}
if want := map[string]string{sdkstep.OutcomePassed: sdkstep.StepPassed, sdkstep.OutcomeFailed: sdkstep.StepFailed}[c.outcome]; step.State != want {
t.Fatalf("the key ended %s (%s), step-ended %s: no step ends passed in one place and failed in another", step.State, step.Reason, c.outcome)
}
if c.answer.Exit != 0 && (!strings.Contains(step.Reason, c.said) || !strings.Contains(string(step.Result), c.inline)) {
t.Fatalf("the key's reason %q and answer %s, want the exit said and the answer kept", step.Reason, step.Result)
}
mu.Lock()
defer mu.Unlock()
name := sdkstep.OutputObjectName(a.ID)
if c.objectKept {
name := sdkstep.OutputObjectName(a.ID)
if end["output-object"] != name || end["output"] != nil || string(kept[name]) != big {
t.Fatalf("end %v and kept %d bytes, want the object %s holding the whole answer and no inline output", end["output-object"], len(kept[name]), name)
}
} else if end["output-object"] != nil || len(kept) != 0 {
t.Fatalf("an object %v was named or kept (%d), want none", end["output-object"], len(kept))
} else if end["output-object"] != nil {
t.Fatalf("an object %v was named, want none", end["output-object"])
}
switch {
case c.keyObject == "":
if step.OutputObject != "" || (c.keepErr == nil && keeps != 0) {
t.Fatalf("the key names %q and %d objects were kept, want none", step.OutputObject, keeps)
}
case c.check:
var whole map[string]any
if step.OutputObject != name || len(step.Result) != 0 || keeps != 1 || json.Unmarshal(kept[name], &whole) != nil || whole["fetch"] == nil {
t.Fatalf("the key names %q with result %.80s, %d keeps; want the object %s holding the whole check answer, kept once", step.OutputObject, step.Result, keeps, name)
}
default:
var rest sdkstep.RunAnswer
if step.OutputObject != name || keeps != 1 || string(kept[name]) != c.keyObject || json.Unmarshal(step.Result, &rest) != nil || rest.Output != nil || rest.Key != c.answer.Key {
t.Fatalf("the key names %q with result %.80s, %d keeps; want the object %s holding the output, kept once, and the run answer without it", step.OutputObject, step.Result, keeps, name)
}
}
})
}
modules/mesh-delivery-sdk/steprun/step.go
+72 -9
View File
@@ -14,11 +14,16 @@ import (
buildrun "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildrun"
sdkstep "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildstep"
"git.novox.be/novox/mesh-sdk/go/kvlimit"
)
// Put writes one key of a state: the steps state.Put's shape (go/state.go).
type Put func(key string, value any) (uint64, error)
// Keep keeps a step's answer that is over what the steps state holds per value as the object named object in the build
// module's outputs store (design 47 G A-290), the object the seat's verb output answers: Puller.KeepOutput, KeepOutputIn.
type Keep func(object string, body []byte) error
// EndAttempts and EndRetry bound how a step's end is written: a write that fails is tried again, since the end
// is the only way the step's reader learns it is over.
var (
@@ -63,15 +68,16 @@ func beginRun(put Put, node, tool, id, what string, run int) (sdkstep.Step, stri
// end runs run and writes the step's end.
func end(ctx context.Context, put Put, key string, step sdkstep.Step, run func(ctx context.Context) (any, error)) {
result, err := protect(ctx, run)
_ = finish(put, key, step, result, err, log.Printf) // finish says a refused end itself
_, _ = finish(put, nil, key, step, result, err, log.Printf) // finish says a refused end itself
}
// finish writes the step's end from run's result or error: passed, failed or refused (a *Refusal). A write that fails is
// tried EndAttempts times. When the whole end is refused every time (a result the state will not take: issue 1930), the
// finish writes the step's end from run's result or error: passed, failed or refused (a *Refusal), and answers the end it
// wrote. An end whose value would be over what the steps state holds per value keeps its answer as an object with keep
// (keepOver, design 47 G A-290). A write that fails is tried EndAttempts times. When the whole end is refused every time (a result the state will not take: issue 1930), the
// refusal is logged and a small failed end without the result is written instead, naming the write's error, so the
// step's reader learns it is over and why. finish answers an error only when no end could be written at all, and then
// the step's ask is not acknowledged as done.
func finish(put Put, key string, step sdkstep.Step, result any, err error, logf func(format string, args ...any)) error {
func finish(put Put, keep Keep, key string, step sdkstep.Step, result any, err error, logf func(format string, args ...any)) (sdkstep.Step, error) {
step.Ended = now().UTC()
switch ref, drop := (*Refusal)(nil), (*Dropped)(nil); {
case err == nil:
@@ -84,6 +90,10 @@ func finish(put Put, key string, step sdkstep.Step, result any, err error, logf
step.Result = raw
}
}
if ans, ok := runAnswerIn(result); ok && ans.Exit != 0 && step.State == sdkstep.StepPassed {
// A run whose program exited other than 0 ends failed, its answer kept, as its step-ended says (design 47 G A-290).
step.State, step.Reason = sdkstep.StepFailed, exitedWhy(ans)
}
case errors.As(err, &drop):
step.State, step.Reason = sdkstep.StepDropped, err.Error()
case errors.As(err, &ref):
@@ -98,9 +108,10 @@ func finish(put Put, key string, step sdkstep.Step, result any, err error, logf
}
}
}
step = keepOver(keep, step, result)
werr := writeEnd(put, key, step)
if werr == nil {
return nil
return step, nil
}
if logf != nil {
logf("the end of the step %s (%s) could not be written in %d tries: %v", key, step.State, EndAttempts, werr)
@@ -110,20 +121,72 @@ func finish(put Put, key string, step sdkstep.Step, result any, err error, logf
if step.Reason != "" {
why = fmt.Sprintf("the step ended %s (%s), and its end could not be written: %v", ended, step.Reason, werr)
}
step.State, step.Reason, step.Result = sdkstep.StepFailed, why, nil
step.State, step.Reason, step.Result, step.OutputObject = sdkstep.StepFailed, why, nil, ""
if ferr := writeEnd(put, key, step); ferr != nil {
if logf != nil {
logf("the failed end of the step %s could not be written either: %v", key, ferr)
}
return fmt.Errorf("steprun: no end of the step %s could be written: %w", key, ferr)
return step, fmt.Errorf("steprun: no end of the step %s could be written: %w", key, ferr)
}
return nil
return step, nil
}
// keepOver is step as its key keeps it (design 47 G A-290): a value within what the steps state holds per value
// (kvlimit.ModuleStateValueBytes less the headers' room) keeps its answer itself; over it, the answer is kept with keep as the
// object sdkstep.OutputObjectName(step.ID), which the key names as its output-object: a run's output, the key keeping the
// run answer without it, else the whole answer, the key's result then empty. A step whose answer cannot be kept ends failed,
// saying why, with no answer, so it never reads passed where its answer is lost.
func keepOver(keep Keep, step sdkstep.Step, result any) sdkstep.Step {
raw, err := sdkstep.EncodeStep(step)
if err != nil || len(step.Result) == 0 || kvlimit.Check("steps", "", len(raw), kvlimit.ModuleStateValueBytes) == nil {
return step
}
object, body, rest := sdkstep.OutputObjectName(step.ID), []byte(step.Result), json.RawMessage(nil)
if ans, ok := runAnswerIn(result); ok {
body, ans.Output = ans.Output, nil
if rest, err = json.Marshal(ans); err != nil {
rest = nil
}
}
lost := func(why string) sdkstep.Step {
if step.Reason != "" {
why = step.Reason + "; " + why
}
step.State, step.Reason, step.Result, step.Passing = sdkstep.StepFailed, why, nil, false
return step
}
if keep == nil {
return lost(fmt.Sprintf("the step's answer makes its value %d bytes, over what the steps state holds, and this holder keeps no object for it", len(raw)))
}
if err := keep(object, body); err != nil {
return lost(fmt.Sprintf("the step's answer makes its value %d bytes, over what the steps state holds, and could not be kept as the object %s: %v", len(raw), object, err))
}
step.Result, step.OutputObject = rest, object
return step
}
// exitedWhy is why a run whose program exited other than 0 failed: its exit status and the first line of its standard error.
func exitedWhy(ans sdkstep.RunAnswer) string {
return fmt.Sprintf("the program exited %d: %s", ans.Exit, ans.Said)
}
// runAnswerIn is the run answer a step handed back, whatever its form (the value or a pointer).
func runAnswerIn(result any) (sdkstep.RunAnswer, bool) {
switch a := result.(type) {
case sdkstep.RunAnswer:
return a, true
case *sdkstep.RunAnswer:
if a != nil {
return *a, true
}
}
return sdkstep.RunAnswer{}, false
}
// again writes the step as given again after a transport fault: the state again with the run's cause, and when the bus gives
// the ask again (ADR 0664 §1). The key is not ended: its asker keeps waiting on it.
func again(put Put, key string, step sdkstep.Step, err error, delay time.Duration) error {
step.State, step.Reason, step.Passing, step.Result = sdkstep.StepAgain, err.Error(), true, nil
step.State, step.Reason, step.Passing, step.Result, step.OutputObject = sdkstep.StepAgain, err.Error(), true, nil, ""
step.Ended, step.AgainAt = time.Time{}, now().UTC().Add(delay)
return writeEnd(put, key, step)
}