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

Merge pull request 'A build step's run that meets a transport fault is again; the asker waits through it and drops what it withdraws (issue 1748; group fix/1748-transport-retry)' (#196) from fix/1748-transport-retry into main

Browse Source
This commit was merged in pull request #196.
This commit is contained in:
mesh-tools
2026-11-09 22:58:32 +00:00
parent 10066c5ec0 7e0634205a
commit 8b03cc036e
53 changed files with 1206 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
contracts/go/seats/nodebuildgo/contract.go
+71 -3
View File
@@ -18,6 +18,7 @@ const (
VerbCurrent = "current"
VerbPause = "pause"
VerbResume = "resume"
VerbDrop = "drop"
)
const (
@@ -28,7 +29,7 @@ const (
// Verbs is every verb the seat serves, in its definition's order: a new copy each call.
func Verbs() []string {
return []string{VerbCurrent, VerbPause, VerbResume}
return []string{VerbCurrent, VerbPause, VerbResume, VerbDrop}
}
// AcceptSubjectDelivery is where delivery is sent.
@@ -117,6 +118,41 @@ func DecodePausesByPausedNodeValue(raw []byte) (PausesByPausedNodeValue, error)
return v, nil
}
// PausesByDroppedIDKey is the key of one value of the state pauses under the form "dropped.<id>".
func PausesByDroppedIDKey(id string) string { return "dropped." + id }
// ParsePausesByDroppedIDKey reads the parts of a key of the form "dropped.<id>", in the order of the form; the last result is false for a key of another form.
func ParsePausesByDroppedIDKey(key string) (string, bool) {
rest := key
if !strings.HasPrefix(rest, "dropped.") {
return "", false
}
rest = rest[8:]
if rest == "" {
return "", false
}
v1 := rest
rest = ""
if rest != "" {
return "", false
}
return v1, true
}
// DecodePausesByDroppedIDValue reads one value of the state pauses under the form "dropped.<id>", refusing a field PausesByDroppedIDValue does not know and any text after the value.
func DecodePausesByDroppedIDValue(raw []byte) (PausesByDroppedIDValue, error) {
var v PausesByDroppedIDValue
dec := json.NewDecoder(bytes.NewReader(raw))
dec.DisallowUnknownFields()
if err := dec.Decode(&v); err != nil {
return v, err
}
if dec.More() {
return v, errors.New("text after the value")
}
return v, nil
}
// PausesByCapacityNodeKey is the key of one value of the state pauses under the form "capacity.<node>".
func PausesByCapacityNodeKey(node string) string { return "capacity." + node }
@@ -245,6 +281,16 @@ type ComposeAnswer struct {
Verdict any `json:"verdict,omitempty"`
}
// Drop is an ask a caller no longer waits for.
type Drop struct {
// when (RFC 3339 time)
At *time.Time `json:"at,omitempty"`
By string `json:"by,omitempty"`
// the layout version
V int64 `json:"v,omitempty"`
Why string `json:"why"`
}
// FetchBesideItem is a type of the contract.
type FetchBesideItem struct {
Commit string `json:"commit"`
@@ -433,19 +479,25 @@ type RunProgram struct {
// Step is one step a build module runs on a node.
type Step struct {
// when the bus gives the ask again, for the state again (RFC 3339 time)
AgainAt *time.Time `json:"again-at,omitempty"`
// when (RFC 3339 time)
Ended *time.Time `json:"ended,omitempty"`
ID string `json:"id"`
Node string `json:"node"`
Owner string `json:"owner,omitempty"`
Phase string `json:"phase,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"`
// when (RFC 3339 time)
PhaseAt *time.Time `json:"phase-at,omitempty"`
Reason string `json:"reason,omitempty"`
Result *StepResult `json:"result,omitempty"`
// which run of the ask this is, counted by the bus's delivery count (1 for the first)
Run int64 `json:"run,omitempty"`
// when (RFC 3339 time)
Started time.Time `json:"started"`
// running, passed, failed or refused
// running, again (ended on a transport fault, to be given again by the bus after again-at), passed, failed, refused or dropped (ended by the verb drop, its cause the reason)
State string `json:"state"`
Tool string `json:"tool"`
// the layout version
@@ -588,6 +640,19 @@ type ResumeInput struct {
// ResumeAnswer is what resume answers.
type ResumeAnswer = map[string]any
// DropInput is what drop takes.
type DropInput struct {
// who drops it
By string `json:"by,omitempty"`
// the ask's id
ID string `json:"id"`
// why the caller no longer waits
Why string `json:"why"`
}
// DropAnswer is what drop answers.
type DropAnswer = map[string]any
// DeliveryAccept is the body of delivery.
type DeliveryAccept = StepAsk
@@ -603,6 +668,9 @@ type StepsByNodeIDValue = Step
// PausesByPausedNodeValue is one value of the state pauses under the key paused.<node>.
type PausesByPausedNodeValue = Pause
// PausesByDroppedIDValue is one value of the state pauses under the key dropped.<id>.
type PausesByDroppedIDValue = Drop
// PausesByCapacityNodeValue is one value of the state pauses under the key capacity.<node>.
type PausesByCapacityNodeValue = Capacity
contracts/go/seats/nodebuildgo/contract_test.go
+6
View File
@@ -53,6 +53,12 @@ func TestFixtures(t *testing.T) {
decode: func(b []byte) (any, error) { var v CheckAccept; err := json.Unmarshal(b, &v); return v, err }, subjectIs: regexp.MustCompile("^mesh\\.seat\\.node-build-go\\.accept\\.check$")},
{name: "a finished run step whose program exited non-zero", subject: "", given: "{\"ended\":\"2026-11-05T10:01:00Z\",\"id\":\"r5e6\",\"node\":\"ace\",\"result\":{\"built\":true,\"exit\":2,\"key\":\"program-0123456789abcdef\",\"said\":\"the facts name no head\"},\"started\":\"2026-11-05T10:00:00Z\",\"state\":\"failed\",\"tool\":\"check\",\"v\":1}", wire: "{\"ended\":\"2026-11-05T10:01:00Z\",\"id\":\"r5e6\",\"node\":\"ace\",\"result\":{\"built\":true,\"exit\":2,\"key\":\"program-0123456789abcdef\",\"said\":\"the facts name no head\"},\"started\":\"2026-11-05T10:00:00Z\",\"state\":\"failed\",\"tool\":\"check\",\"v\":1}",
decode: func(b []byte) (any, error) { var v StepsByNodeIDValue; err := json.Unmarshal(b, &v); return v, err }, subjectIs: nil},
{name: "a step ended on a transport fault, given again by the bus", subject: "", given: "{\"again-at\":\"2026-11-05T10:01:30Z\",\"ended\":\"2026-11-05T10:00:30Z\",\"id\":\"a1b2\",\"node\":\"ace\",\"passing\":true,\"reason\":\"fetching the commit: kex_exchange_identification: read: Connection reset by peer\",\"run\":1,\"started\":\"2026-11-05T10:00:00Z\",\"state\":\"again\",\"tool\":\"build\",\"v\":1}", wire: "{\"again-at\":\"2026-11-05T10:01:30Z\",\"ended\":\"2026-11-05T10:00:30Z\",\"id\":\"a1b2\",\"node\":\"ace\",\"passing\":true,\"reason\":\"fetching the commit: kex_exchange_identification: read: Connection reset by peer\",\"run\":1,\"started\":\"2026-11-05T10:00:00Z\",\"state\":\"again\",\"tool\":\"build\",\"v\":1}",
decode: func(b []byte) (any, error) { var v StepsByNodeIDValue; err := json.Unmarshal(b, &v); return v, err }, subjectIs: nil},
{name: "an ask dropped by the caller that no longer waits", subject: "", given: "{\"at\":\"2026-11-05T10:00:00Z\",\"by\":\"mesh-delivery-builder\",\"v\":1,\"why\":\"a newer fingerprint supersedes it\"}", wire: "{\"at\":\"2026-11-05T10:00:00Z\",\"by\":\"mesh-delivery-builder\",\"v\":1,\"why\":\"a newer fingerprint supersedes it\"}",
decode: func(b []byte) (any, error) { var v PausesByDroppedIDValue; 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},
}
for _, c := range cases {
v, err := c.decode([]byte(c.given))
contracts/go/seats/nodebuildimage/contract.go
+71 -3
View File
@@ -18,6 +18,7 @@ const (
VerbCurrent = "current"
VerbPause = "pause"
VerbResume = "resume"
VerbDrop = "drop"
)
const (
@@ -28,7 +29,7 @@ const (
// Verbs is every verb the seat serves, in its definition's order: a new copy each call.
func Verbs() []string {
return []string{VerbCurrent, VerbPause, VerbResume}
return []string{VerbCurrent, VerbPause, VerbResume, VerbDrop}
}
// AcceptSubjectDelivery is where delivery is sent.
@@ -117,6 +118,41 @@ func DecodePausesByPausedNodeValue(raw []byte) (PausesByPausedNodeValue, error)
return v, nil
}
// PausesByDroppedIDKey is the key of one value of the state pauses under the form "dropped.<id>".
func PausesByDroppedIDKey(id string) string { return "dropped." + id }
// ParsePausesByDroppedIDKey reads the parts of a key of the form "dropped.<id>", in the order of the form; the last result is false for a key of another form.
func ParsePausesByDroppedIDKey(key string) (string, bool) {
rest := key
if !strings.HasPrefix(rest, "dropped.") {
return "", false
}
rest = rest[8:]
if rest == "" {
return "", false
}
v1 := rest
rest = ""
if rest != "" {
return "", false
}
return v1, true
}
// DecodePausesByDroppedIDValue reads one value of the state pauses under the form "dropped.<id>", refusing a field PausesByDroppedIDValue does not know and any text after the value.
func DecodePausesByDroppedIDValue(raw []byte) (PausesByDroppedIDValue, error) {
var v PausesByDroppedIDValue
dec := json.NewDecoder(bytes.NewReader(raw))
dec.DisallowUnknownFields()
if err := dec.Decode(&v); err != nil {
return v, err
}
if dec.More() {
return v, errors.New("text after the value")
}
return v, nil
}
// PausesByCapacityNodeKey is the key of one value of the state pauses under the form "capacity.<node>".
func PausesByCapacityNodeKey(node string) string { return "capacity." + node }
@@ -245,6 +281,16 @@ type ComposeAnswer struct {
Verdict any `json:"verdict,omitempty"`
}
// Drop is an ask a caller no longer waits for.
type Drop struct {
// when (RFC 3339 time)
At *time.Time `json:"at,omitempty"`
By string `json:"by,omitempty"`
// the layout version
V int64 `json:"v,omitempty"`
Why string `json:"why"`
}
// FetchBesideItem is a type of the contract.
type FetchBesideItem struct {
Commit string `json:"commit"`
@@ -433,19 +479,25 @@ type RunProgram struct {
// Step is one step a build module runs on a node.
type Step struct {
// when the bus gives the ask again, for the state again (RFC 3339 time)
AgainAt *time.Time `json:"again-at,omitempty"`
// when (RFC 3339 time)
Ended *time.Time `json:"ended,omitempty"`
ID string `json:"id"`
Node string `json:"node"`
Owner string `json:"owner,omitempty"`
Phase string `json:"phase,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"`
// when (RFC 3339 time)
PhaseAt *time.Time `json:"phase-at,omitempty"`
Reason string `json:"reason,omitempty"`
Result *StepResult `json:"result,omitempty"`
// which run of the ask this is, counted by the bus's delivery count (1 for the first)
Run int64 `json:"run,omitempty"`
// when (RFC 3339 time)
Started time.Time `json:"started"`
// running, passed, failed or refused
// running, again (ended on a transport fault, to be given again by the bus after again-at), passed, failed, refused or dropped (ended by the verb drop, its cause the reason)
State string `json:"state"`
Tool string `json:"tool"`
// the layout version
@@ -588,6 +640,19 @@ type ResumeInput struct {
// ResumeAnswer is what resume answers.
type ResumeAnswer = map[string]any
// DropInput is what drop takes.
type DropInput struct {
// who drops it
By string `json:"by,omitempty"`
// the ask's id
ID string `json:"id"`
// why the caller no longer waits
Why string `json:"why"`
}
// DropAnswer is what drop answers.
type DropAnswer = map[string]any
// DeliveryAccept is the body of delivery.
type DeliveryAccept = StepAsk
@@ -603,6 +668,9 @@ type StepsByNodeIDValue = Step
// PausesByPausedNodeValue is one value of the state pauses under the key paused.<node>.
type PausesByPausedNodeValue = Pause
// PausesByDroppedIDValue is one value of the state pauses under the key dropped.<id>.
type PausesByDroppedIDValue = Drop
// PausesByCapacityNodeValue is one value of the state pauses under the key capacity.<node>.
type PausesByCapacityNodeValue = Capacity
contracts/go/seats/nodebuildimage/contract_test.go
+6
View File
@@ -39,6 +39,12 @@ func TestFixtures(t *testing.T) {
}, subjectIs: nil},
{name: "a build ask", subject: "mesh.seat.node-build-image.accept.delivery", given: "{\"asker\":\"mesh-delivery-builder\",\"build\":{\"artifacts\":[{\"artifact\":\"go\"}],\"module\":\"nats\"},\"id\":\"a1b2\",\"kind\":\"build\",\"language\":\"go\"}", wire: "{\"asker\":\"mesh-delivery-builder\",\"build\":{\"artifacts\":[{\"artifact\":\"go\"}],\"module\":\"nats\"},\"id\":\"a1b2\",\"kind\":\"build\",\"language\":\"go\"}",
decode: func(b []byte) (any, error) { var v DeliveryAccept; err := json.Unmarshal(b, &v); return v, err }, subjectIs: regexp.MustCompile("^mesh\\.seat\\.node-build-image\\.accept\\.delivery$")},
{name: "a step ended on a transport fault, given again by the bus", subject: "", given: "{\"again-at\":\"2026-11-05T10:01:30Z\",\"ended\":\"2026-11-05T10:00:30Z\",\"id\":\"a1b2\",\"node\":\"ace\",\"passing\":true,\"reason\":\"fetching the commit: kex_exchange_identification: read: Connection reset by peer\",\"run\":1,\"started\":\"2026-11-05T10:00:00Z\",\"state\":\"again\",\"tool\":\"build\",\"v\":1}", wire: "{\"again-at\":\"2026-11-05T10:01:30Z\",\"ended\":\"2026-11-05T10:00:30Z\",\"id\":\"a1b2\",\"node\":\"ace\",\"passing\":true,\"reason\":\"fetching the commit: kex_exchange_identification: read: Connection reset by peer\",\"run\":1,\"started\":\"2026-11-05T10:00:00Z\",\"state\":\"again\",\"tool\":\"build\",\"v\":1}",
decode: func(b []byte) (any, error) { var v StepsByNodeIDValue; err := json.Unmarshal(b, &v); return v, err }, subjectIs: nil},
{name: "an ask dropped by the caller that no longer waits", subject: "", given: "{\"at\":\"2026-11-05T10:00:00Z\",\"by\":\"mesh-delivery-builder\",\"v\":1,\"why\":\"a newer fingerprint supersedes it\"}", wire: "{\"at\":\"2026-11-05T10:00:00Z\",\"by\":\"mesh-delivery-builder\",\"v\":1,\"why\":\"a newer fingerprint supersedes it\"}",
decode: func(b []byte) (any, error) { var v PausesByDroppedIDValue; 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},
}
for _, c := range cases {
v, err := c.decode([]byte(c.given))
contracts/go/seats/nodebuildnode/contract.go
+71 -3
View File
@@ -18,6 +18,7 @@ const (
VerbCurrent = "current"
VerbPause = "pause"
VerbResume = "resume"
VerbDrop = "drop"
)
const (
@@ -28,7 +29,7 @@ const (
// Verbs is every verb the seat serves, in its definition's order: a new copy each call.
func Verbs() []string {
return []string{VerbCurrent, VerbPause, VerbResume}
return []string{VerbCurrent, VerbPause, VerbResume, VerbDrop}
}
// AcceptSubjectDelivery is where delivery is sent.
@@ -117,6 +118,41 @@ func DecodePausesByPausedNodeValue(raw []byte) (PausesByPausedNodeValue, error)
return v, nil
}
// PausesByDroppedIDKey is the key of one value of the state pauses under the form "dropped.<id>".
func PausesByDroppedIDKey(id string) string { return "dropped." + id }
// ParsePausesByDroppedIDKey reads the parts of a key of the form "dropped.<id>", in the order of the form; the last result is false for a key of another form.
func ParsePausesByDroppedIDKey(key string) (string, bool) {
rest := key
if !strings.HasPrefix(rest, "dropped.") {
return "", false
}
rest = rest[8:]
if rest == "" {
return "", false
}
v1 := rest
rest = ""
if rest != "" {
return "", false
}
return v1, true
}
// DecodePausesByDroppedIDValue reads one value of the state pauses under the form "dropped.<id>", refusing a field PausesByDroppedIDValue does not know and any text after the value.
func DecodePausesByDroppedIDValue(raw []byte) (PausesByDroppedIDValue, error) {
var v PausesByDroppedIDValue
dec := json.NewDecoder(bytes.NewReader(raw))
dec.DisallowUnknownFields()
if err := dec.Decode(&v); err != nil {
return v, err
}
if dec.More() {
return v, errors.New("text after the value")
}
return v, nil
}
// PausesByCapacityNodeKey is the key of one value of the state pauses under the form "capacity.<node>".
func PausesByCapacityNodeKey(node string) string { return "capacity." + node }
@@ -245,6 +281,16 @@ type ComposeAnswer struct {
Verdict any `json:"verdict,omitempty"`
}
// Drop is an ask a caller no longer waits for.
type Drop struct {
// when (RFC 3339 time)
At *time.Time `json:"at,omitempty"`
By string `json:"by,omitempty"`
// the layout version
V int64 `json:"v,omitempty"`
Why string `json:"why"`
}
// FetchBesideItem is a type of the contract.
type FetchBesideItem struct {
Commit string `json:"commit"`
@@ -433,19 +479,25 @@ type RunProgram struct {
// Step is one step a build module runs on a node.
type Step struct {
// when the bus gives the ask again, for the state again (RFC 3339 time)
AgainAt *time.Time `json:"again-at,omitempty"`
// when (RFC 3339 time)
Ended *time.Time `json:"ended,omitempty"`
ID string `json:"id"`
Node string `json:"node"`
Owner string `json:"owner,omitempty"`
Phase string `json:"phase,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"`
// when (RFC 3339 time)
PhaseAt *time.Time `json:"phase-at,omitempty"`
Reason string `json:"reason,omitempty"`
Result *StepResult `json:"result,omitempty"`
// which run of the ask this is, counted by the bus's delivery count (1 for the first)
Run int64 `json:"run,omitempty"`
// when (RFC 3339 time)
Started time.Time `json:"started"`
// running, passed, failed or refused
// running, again (ended on a transport fault, to be given again by the bus after again-at), passed, failed, refused or dropped (ended by the verb drop, its cause the reason)
State string `json:"state"`
Tool string `json:"tool"`
// the layout version
@@ -588,6 +640,19 @@ type ResumeInput struct {
// ResumeAnswer is what resume answers.
type ResumeAnswer = map[string]any
// DropInput is what drop takes.
type DropInput struct {
// who drops it
By string `json:"by,omitempty"`
// the ask's id
ID string `json:"id"`
// why the caller no longer waits
Why string `json:"why"`
}
// DropAnswer is what drop answers.
type DropAnswer = map[string]any
// DeliveryAccept is the body of delivery.
type DeliveryAccept = StepAsk
@@ -603,6 +668,9 @@ type StepsByNodeIDValue = Step
// PausesByPausedNodeValue is one value of the state pauses under the key paused.<node>.
type PausesByPausedNodeValue = Pause
// PausesByDroppedIDValue is one value of the state pauses under the key dropped.<id>.
type PausesByDroppedIDValue = Drop
// PausesByCapacityNodeValue is one value of the state pauses under the key capacity.<node>.
type PausesByCapacityNodeValue = Capacity
contracts/go/seats/nodebuildnode/contract_test.go
+6
View File
@@ -43,6 +43,12 @@ func TestFixtures(t *testing.T) {
decode: func(b []byte) (any, error) { var v CheckAccept; err := json.Unmarshal(b, &v); return v, err }, subjectIs: regexp.MustCompile("^mesh\\.seat\\.node-build-node\\.accept\\.check$")},
{name: "a finished check step of a TypeScript merge preview", subject: "", given: "{\"ended\":\"2026-11-05T10:01:00Z\",\"id\":\"c3d6\",\"node\":\"ace\",\"result\":{\"fetch\":{\"base\":\"abcdef0123456789abcdef0123456789abcdef01\",\"changed\":[\"modules/m/package.json\"],\"commit\":\"0123456789abcdef0123456789abcdef01234567\",\"merge\":\"0123456789abcdef0123456789abcdef01234567\",\"modules\":[\"modules/m\"],\"trees\":{\"modules/m\":{\"manifest\":\"0581b699d9d05517e69b21f5cedbeb2eee8aba2b205280b561be731594d7feef\",\"tree\":\"0123456789abcdef0123456789abcdef01234567\"}},\"workspace\":\"mvii1sxt979f67\"}},\"started\":\"2026-11-05T10:00:00Z\",\"state\":\"passed\",\"tool\":\"check\",\"v\":1}", wire: "{\"ended\":\"2026-11-05T10:01:00Z\",\"id\":\"c3d6\",\"node\":\"ace\",\"result\":{\"fetch\":{\"base\":\"abcdef0123456789abcdef0123456789abcdef01\",\"changed\":[\"modules/m/package.json\"],\"commit\":\"0123456789abcdef0123456789abcdef01234567\",\"merge\":\"0123456789abcdef0123456789abcdef01234567\",\"modules\":[\"modules/m\"],\"trees\":{\"modules/m\":{\"manifest\":\"0581b699d9d05517e69b21f5cedbeb2eee8aba2b205280b561be731594d7feef\",\"tree\":\"0123456789abcdef0123456789abcdef01234567\"}},\"workspace\":\"mvii1sxt979f67\"}},\"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: "a step ended on a transport fault, given again by the bus", subject: "", given: "{\"again-at\":\"2026-11-05T10:01:30Z\",\"ended\":\"2026-11-05T10:00:30Z\",\"id\":\"a1b2\",\"node\":\"ace\",\"passing\":true,\"reason\":\"fetching the commit: kex_exchange_identification: read: Connection reset by peer\",\"run\":1,\"started\":\"2026-11-05T10:00:00Z\",\"state\":\"again\",\"tool\":\"build\",\"v\":1}", wire: "{\"again-at\":\"2026-11-05T10:01:30Z\",\"ended\":\"2026-11-05T10:00:30Z\",\"id\":\"a1b2\",\"node\":\"ace\",\"passing\":true,\"reason\":\"fetching the commit: kex_exchange_identification: read: Connection reset by peer\",\"run\":1,\"started\":\"2026-11-05T10:00:00Z\",\"state\":\"again\",\"tool\":\"build\",\"v\":1}",
decode: func(b []byte) (any, error) { var v StepsByNodeIDValue; err := json.Unmarshal(b, &v); return v, err }, subjectIs: nil},
{name: "an ask dropped by the caller that no longer waits", subject: "", given: "{\"at\":\"2026-11-05T10:00:00Z\",\"by\":\"mesh-delivery-builder\",\"v\":1,\"why\":\"a newer fingerprint supersedes it\"}", wire: "{\"at\":\"2026-11-05T10:00:00Z\",\"by\":\"mesh-delivery-builder\",\"v\":1,\"why\":\"a newer fingerprint supersedes it\"}",
decode: func(b []byte) (any, error) { var v PausesByDroppedIDValue; 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},
}
for _, c := range cases {
v, err := c.decode([]byte(c.given))
contracts/parts/build-step.json
+68 -1
View File
@@ -38,6 +38,36 @@
"type": "object",
"description": "whether the node is paused"
}
},
{
"name": "drop",
"description": "End an ask this node holds for a caller that no longer waits for it: a step not yet running is ended dropped when the bus gives it, one running is cancelled; both end the step dropped, naming who dropped it and why. Writes dropped.<id> in the pauses state; the build module removes the key when it has acted.",
"callers": "the build seats' asker (the builder and the checker)",
"input": {
"type": "object",
"properties": {
"id": {
"type": "string",
"description": "the ask's id"
},
"by": {
"type": "string",
"description": "who drops it"
},
"why": {
"type": "string",
"description": "why the caller no longer waits"
}
},
"required": [
"id",
"why"
]
},
"answer": {
"type": "object",
"description": "whether the ask was held here"
}
}
],
"defs": {
@@ -404,7 +434,7 @@
},
"state": {
"type": "string",
"description": "running, passed, failed or refused"
"description": "running, again (ended on a transport fault, to be given again by the bus after again-at), passed, failed, refused or dropped (ended by the verb drop, its cause the reason)"
},
"started": {
"type": "string",
@@ -435,6 +465,19 @@
},
"owner": {
"type": "string"
},
"run": {
"type": "integer",
"description": "which run of the ask this is, counted by the bus's delivery count (1 for the first)"
},
"again-at": {
"type": "string",
"description": "when the bus gives the ask again, for the state again (RFC 3339 time)",
"format": "date-time"
},
"passing": {
"type": "boolean",
"description": "true when the step ended on a fault that passes (a transport fault), so a failed end after the last run says so"
}
},
"required": [
@@ -476,6 +519,30 @@
],
"description": "the toolchain a node reports"
},
"Drop": {
"type": "object",
"properties": {
"v": {
"type": "integer",
"description": "the layout version"
},
"by": {
"type": "string"
},
"why": {
"type": "string"
},
"at": {
"type": "string",
"description": "when (RFC 3339 time)",
"format": "date-time"
}
},
"required": [
"why"
],
"description": "an ask a caller no longer waits for"
},
"Pause": {
"type": "object",
"properties": {
contracts/seats/node-build-go/fixtures/states.json
+27
View File
@@ -312,5 +312,32 @@
"said": "the facts name no head"
}
}
},
{
"name": "a step ended on a transport fault, given again by the bus",
"of": "state:steps/<node>.<id>",
"given": {
"v": 1,
"id": "a1b2",
"tool": "build",
"node": "ace",
"state": "again",
"started": "2026-11-05T10:00:00Z",
"ended": "2026-11-05T10:00:30Z",
"reason": "fetching the commit: kex_exchange_identification: read: Connection reset by peer",
"run": 1,
"again-at": "2026-11-05T10:01:30Z",
"passing": true
}
},
{
"name": "an ask dropped by the caller that no longer waits",
"of": "state:pauses/dropped.<id>",
"given": {
"v": 1,
"by": "mesh-delivery-builder",
"why": "a newer fingerprint supersedes it",
"at": "2026-11-05T10:00:00Z"
}
}
]
contracts/seats/node-build-go/fixtures/verbs.json
+11
View File
@@ -0,0 +1,11 @@
[
{
"name": "drop names the ask and why",
"of": "verb:drop.input",
"given": {
"id": "a1b2",
"by": "mesh-delivery-builder",
"why": "a newer fingerprint supersedes it"
}
}
]
contracts/seats/node-build-go/seat.json
+7
View File
@@ -58,6 +58,13 @@
},
"description": "whether a node is paused"
},
{
"key": "dropped.<id>",
"value": {
"$ref": "#/defs/Drop"
},
"description": "an ask its caller no longer waits for"
},
{
"key": "capacity.<node>",
"value": {
contracts/seats/node-build-image/fixtures/states.json
+27
View File
@@ -59,5 +59,32 @@
]
}
}
},
{
"name": "a step ended on a transport fault, given again by the bus",
"of": "state:steps/<node>.<id>",
"given": {
"v": 1,
"id": "a1b2",
"tool": "build",
"node": "ace",
"state": "again",
"started": "2026-11-05T10:00:00Z",
"ended": "2026-11-05T10:00:30Z",
"reason": "fetching the commit: kex_exchange_identification: read: Connection reset by peer",
"run": 1,
"again-at": "2026-11-05T10:01:30Z",
"passing": true
}
},
{
"name": "an ask dropped by the caller that no longer waits",
"of": "state:pauses/dropped.<id>",
"given": {
"v": 1,
"by": "mesh-delivery-builder",
"why": "a newer fingerprint supersedes it",
"at": "2026-11-05T10:00:00Z"
}
}
]
contracts/seats/node-build-image/fixtures/verbs.json
+11
View File
@@ -0,0 +1,11 @@
[
{
"name": "drop names the ask and why",
"of": "verb:drop.input",
"given": {
"id": "a1b2",
"by": "mesh-delivery-builder",
"why": "a newer fingerprint supersedes it"
}
}
]
contracts/seats/node-build-image/seat.json
+7
View File
@@ -58,6 +58,13 @@
},
"description": "whether a node is paused"
},
{
"key": "dropped.<id>",
"value": {
"$ref": "#/defs/Drop"
},
"description": "an ask its caller no longer waits for"
},
{
"key": "capacity.<node>",
"value": {
contracts/seats/node-build-node/fixtures/states.json
+27
View File
@@ -131,5 +131,32 @@
}
}
}
},
{
"name": "a step ended on a transport fault, given again by the bus",
"of": "state:steps/<node>.<id>",
"given": {
"v": 1,
"id": "a1b2",
"tool": "build",
"node": "ace",
"state": "again",
"started": "2026-11-05T10:00:00Z",
"ended": "2026-11-05T10:00:30Z",
"reason": "fetching the commit: kex_exchange_identification: read: Connection reset by peer",
"run": 1,
"again-at": "2026-11-05T10:01:30Z",
"passing": true
}
},
{
"name": "an ask dropped by the caller that no longer waits",
"of": "state:pauses/dropped.<id>",
"given": {
"v": 1,
"by": "mesh-delivery-builder",
"why": "a newer fingerprint supersedes it",
"at": "2026-11-05T10:00:00Z"
}
}
]
contracts/seats/node-build-node/fixtures/verbs.json
+11
View File
@@ -0,0 +1,11 @@
[
{
"name": "drop names the ask and why",
"of": "verb:drop.input",
"given": {
"id": "a1b2",
"by": "mesh-delivery-builder",
"why": "a newer fingerprint supersedes it"
}
}
]
contracts/seats/node-build-node/seat.json
+7
View File
@@ -58,6 +58,13 @@
},
"description": "whether a node is paused"
},
{
"key": "dropped.<id>",
"value": {
"$ref": "#/defs/Drop"
},
"description": "an ask its caller no longer waits for"
},
{
"key": "capacity.<node>",
"value": {
contracts/ts/seats/nodebuildgo/src/index.ts
+24
View File
@@ -10,6 +10,7 @@ export const verbs = {
current: "current",
pause: "pause",
resume: "resume",
drop: "drop",
} as const;
export const accepts = {
@@ -66,6 +67,23 @@ export function parsePausesByPausedNodeKey(_key: string): { node: string } | und
return { node };
}
/** The key of one value of the state pauses under the form "dropped.<id>". */
export function pausesByDroppedIDKey(id: string): string {
return "dropped." + id;
}
/** The parts of a key of the form "dropped.<id>", or undefined for a key of another form. */
export function parsePausesByDroppedIDKey(_key: string): { id: string } | undefined {
let _rest = _key;
if (!_rest.startsWith("dropped.")) return undefined;
_rest = _rest.slice(8);
if (_rest === "") return undefined;
const id = _rest;
_rest = "";
if (_rest !== "") return undefined;
return { id };
}
/** The key of one value of the state pauses under the form "capacity.<node>". */
export function pausesByCapacityNodeKey(node: string): string {
return "capacity." + node;
@@ -110,6 +128,9 @@ export type CheckAnswer = buildStep.CheckAnswer;
export type ComposeAnswer = buildStep.ComposeAnswer;
export type CurrentAnswer = buildStep.CurrentAnswer;
export type CurrentInput = buildStep.CurrentInput;
export type Drop = buildStep.Drop;
export type DropAnswer = buildStep.DropAnswer;
export type DropInput = buildStep.DropInput;
export type Fetch = buildStep.Fetch;
export type FetchAnswer = buildStep.FetchAnswer;
export type FetchBesideItem = buildStep.FetchBesideItem;
@@ -159,6 +180,9 @@ export type StepsByNodeIDValue = Step;
/** is one value of the state pauses under the key paused.<node>. */
export type PausesByPausedNodeValue = Pause;
/** is one value of the state pauses under the key dropped.<id>. */
export type PausesByDroppedIDValue = Drop;
/** is one value of the state pauses under the key capacity.<node>. */
export type PausesByCapacityNodeValue = Capacity;
contracts/ts/seats/nodebuildgo/src/parts/build-step.ts
+31 -1
View File
@@ -6,6 +6,7 @@ export const verbs = {
current: "current",
pause: "pause",
resume: "resume",
drop: "drop",
} as const;
/** one artifact of a build ask: built from the fetched tree, or copied from an upstream pin */
@@ -62,6 +63,16 @@ export interface ComposeAnswer {
"verdict"?: unknown;
}
/** an ask a caller no longer waits for */
export interface Drop {
/** when (RFC 3339 time) */
"at"?: string;
"by"?: string;
/** the layout version */
"v"?: number;
"why": string;
}
export interface FetchBesideItem {
"commit": string;
"repository": string;
@@ -244,19 +255,25 @@ export interface RunProgram {
/** one step a build module runs on a node */
export interface Step {
/** when the bus gives the ask again, for the state again (RFC 3339 time) */
"again-at"?: string;
/** when (RFC 3339 time) */
"ended"?: string;
"id": string;
"node": 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;
"phase"?: string;
/** when (RFC 3339 time) */
"phase-at"?: string;
"reason"?: string;
"result"?: StepResult;
/** which run of the ask this is, counted by the bus's delivery count (1 for the first) */
"run"?: number;
/** when (RFC 3339 time) */
"started": string;
/** running, passed, failed or refused */
/** running, again (ended on a transport fault, to be given again by the bus after again-at), passed, failed, refused or dropped (ended by the verb drop, its cause the reason) */
"state": string;
"tool": string;
/** the layout version */
@@ -396,3 +413,16 @@ export interface ResumeInput {
/** is what resume answers. */
export type ResumeAnswer = Record<string, unknown>;
/** is what drop takes. */
export interface DropInput {
/** who drops it */
"by"?: string;
/** the ask's id */
"id": string;
/** why the caller no longer waits */
"why": string;
}
/** is what drop answers. */
export type DropAnswer = Record<string, unknown>;
contracts/ts/seats/nodebuildimage/src/index.ts
+24
View File
@@ -10,6 +10,7 @@ export const verbs = {
current: "current",
pause: "pause",
resume: "resume",
drop: "drop",
} as const;
export const accepts = {
@@ -66,6 +67,23 @@ export function parsePausesByPausedNodeKey(_key: string): { node: string } | und
return { node };
}
/** The key of one value of the state pauses under the form "dropped.<id>". */
export function pausesByDroppedIDKey(id: string): string {
return "dropped." + id;
}
/** The parts of a key of the form "dropped.<id>", or undefined for a key of another form. */
export function parsePausesByDroppedIDKey(_key: string): { id: string } | undefined {
let _rest = _key;
if (!_rest.startsWith("dropped.")) return undefined;
_rest = _rest.slice(8);
if (_rest === "") return undefined;
const id = _rest;
_rest = "";
if (_rest !== "") return undefined;
return { id };
}
/** The key of one value of the state pauses under the form "capacity.<node>". */
export function pausesByCapacityNodeKey(node: string): string {
return "capacity." + node;
@@ -110,6 +128,9 @@ export type CheckAnswer = buildStep.CheckAnswer;
export type ComposeAnswer = buildStep.ComposeAnswer;
export type CurrentAnswer = buildStep.CurrentAnswer;
export type CurrentInput = buildStep.CurrentInput;
export type Drop = buildStep.Drop;
export type DropAnswer = buildStep.DropAnswer;
export type DropInput = buildStep.DropInput;
export type Fetch = buildStep.Fetch;
export type FetchAnswer = buildStep.FetchAnswer;
export type FetchBesideItem = buildStep.FetchBesideItem;
@@ -159,6 +180,9 @@ export type StepsByNodeIDValue = Step;
/** is one value of the state pauses under the key paused.<node>. */
export type PausesByPausedNodeValue = Pause;
/** is one value of the state pauses under the key dropped.<id>. */
export type PausesByDroppedIDValue = Drop;
/** is one value of the state pauses under the key capacity.<node>. */
export type PausesByCapacityNodeValue = Capacity;
contracts/ts/seats/nodebuildimage/src/parts/build-step.ts
+31 -1
View File
@@ -6,6 +6,7 @@ export const verbs = {
current: "current",
pause: "pause",
resume: "resume",
drop: "drop",
} as const;
/** one artifact of a build ask: built from the fetched tree, or copied from an upstream pin */
@@ -62,6 +63,16 @@ export interface ComposeAnswer {
"verdict"?: unknown;
}
/** an ask a caller no longer waits for */
export interface Drop {
/** when (RFC 3339 time) */
"at"?: string;
"by"?: string;
/** the layout version */
"v"?: number;
"why": string;
}
export interface FetchBesideItem {
"commit": string;
"repository": string;
@@ -244,19 +255,25 @@ export interface RunProgram {
/** one step a build module runs on a node */
export interface Step {
/** when the bus gives the ask again, for the state again (RFC 3339 time) */
"again-at"?: string;
/** when (RFC 3339 time) */
"ended"?: string;
"id": string;
"node": 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;
"phase"?: string;
/** when (RFC 3339 time) */
"phase-at"?: string;
"reason"?: string;
"result"?: StepResult;
/** which run of the ask this is, counted by the bus's delivery count (1 for the first) */
"run"?: number;
/** when (RFC 3339 time) */
"started": string;
/** running, passed, failed or refused */
/** running, again (ended on a transport fault, to be given again by the bus after again-at), passed, failed, refused or dropped (ended by the verb drop, its cause the reason) */
"state": string;
"tool": string;
/** the layout version */
@@ -396,3 +413,16 @@ export interface ResumeInput {
/** is what resume answers. */
export type ResumeAnswer = Record<string, unknown>;
/** is what drop takes. */
export interface DropInput {
/** who drops it */
"by"?: string;
/** the ask's id */
"id": string;
/** why the caller no longer waits */
"why": string;
}
/** is what drop answers. */
export type DropAnswer = Record<string, unknown>;
contracts/ts/seats/nodebuildnode/src/index.ts
+24
View File
@@ -10,6 +10,7 @@ export const verbs = {
current: "current",
pause: "pause",
resume: "resume",
drop: "drop",
} as const;
export const accepts = {
@@ -66,6 +67,23 @@ export function parsePausesByPausedNodeKey(_key: string): { node: string } | und
return { node };
}
/** The key of one value of the state pauses under the form "dropped.<id>". */
export function pausesByDroppedIDKey(id: string): string {
return "dropped." + id;
}
/** The parts of a key of the form "dropped.<id>", or undefined for a key of another form. */
export function parsePausesByDroppedIDKey(_key: string): { id: string } | undefined {
let _rest = _key;
if (!_rest.startsWith("dropped.")) return undefined;
_rest = _rest.slice(8);
if (_rest === "") return undefined;
const id = _rest;
_rest = "";
if (_rest !== "") return undefined;
return { id };
}
/** The key of one value of the state pauses under the form "capacity.<node>". */
export function pausesByCapacityNodeKey(node: string): string {
return "capacity." + node;
@@ -110,6 +128,9 @@ export type CheckAnswer = buildStep.CheckAnswer;
export type ComposeAnswer = buildStep.ComposeAnswer;
export type CurrentAnswer = buildStep.CurrentAnswer;
export type CurrentInput = buildStep.CurrentInput;
export type Drop = buildStep.Drop;
export type DropAnswer = buildStep.DropAnswer;
export type DropInput = buildStep.DropInput;
export type Fetch = buildStep.Fetch;
export type FetchAnswer = buildStep.FetchAnswer;
export type FetchBesideItem = buildStep.FetchBesideItem;
@@ -159,6 +180,9 @@ export type StepsByNodeIDValue = Step;
/** is one value of the state pauses under the key paused.<node>. */
export type PausesByPausedNodeValue = Pause;
/** is one value of the state pauses under the key dropped.<id>. */
export type PausesByDroppedIDValue = Drop;
/** is one value of the state pauses under the key capacity.<node>. */
export type PausesByCapacityNodeValue = Capacity;
contracts/ts/seats/nodebuildnode/src/parts/build-step.ts
+31 -1
View File
@@ -6,6 +6,7 @@ export const verbs = {
current: "current",
pause: "pause",
resume: "resume",
drop: "drop",
} as const;
/** one artifact of a build ask: built from the fetched tree, or copied from an upstream pin */
@@ -62,6 +63,16 @@ export interface ComposeAnswer {
"verdict"?: unknown;
}
/** an ask a caller no longer waits for */
export interface Drop {
/** when (RFC 3339 time) */
"at"?: string;
"by"?: string;
/** the layout version */
"v"?: number;
"why": string;
}
export interface FetchBesideItem {
"commit": string;
"repository": string;
@@ -244,19 +255,25 @@ export interface RunProgram {
/** one step a build module runs on a node */
export interface Step {
/** when the bus gives the ask again, for the state again (RFC 3339 time) */
"again-at"?: string;
/** when (RFC 3339 time) */
"ended"?: string;
"id": string;
"node": 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;
"phase"?: string;
/** when (RFC 3339 time) */
"phase-at"?: string;
"reason"?: string;
"result"?: StepResult;
/** which run of the ask this is, counted by the bus's delivery count (1 for the first) */
"run"?: number;
/** when (RFC 3339 time) */
"started": string;
/** running, passed, failed or refused */
/** running, again (ended on a transport fault, to be given again by the bus after again-at), passed, failed, refused or dropped (ended by the verb drop, its cause the reason) */
"state": string;
"tool": string;
/** the layout version */
@@ -396,3 +413,16 @@ export interface ResumeInput {
/** is what resume answers. */
export type ResumeAnswer = Record<string, unknown>;
/** is what drop takes. */
export interface DropInput {
/** who drops it */
"by"?: string;
/** the ask's id */
"id": string;
/** why the caller no longer waits */
"why": string;
}
/** is what drop answers. */
export type DropAnswer = Record<string, unknown>;
modules/mesh-delivery-builder/buildrun/asks.go
+88 -1
View File
@@ -2,8 +2,11 @@ package buildrun
import (
"context"
"encoding/json"
"errors"
"fmt"
"log"
"strings"
"sync"
"time"
@@ -36,6 +39,63 @@ type Asker struct {
// before the asker stops waiting; a step that keeps moving is waited for up to 3 Bounds from its start (issue 1930); zero waits as long as ctx. An ask waiting in its stream is not bounded: the stream's backlog is no fault
// (ADR 0525 §6, §9), and the bound starts again when another node takes the ask.
Bound time.Duration
// Dropper, when set, is told to end an ask the asker stops waiting for without its end (ADR 0664 §5): a step given
// again that the bus did not give within its bound, or a wait ended with a cause its caller names (a superseded
// fingerprint). Never a wait ended by a shutdown (a plain cancel): the ask is then taken over by the next run.
Dropper AskDropper
// 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)
}
// Withdrawn is the cause a caller gives the context of an ask's wait to withdraw the ask (ADR 0664 §4): a build superseded by a
// newer fingerprint, a candidate taken out, an ask asked again by hand. Asker.Wait drops the ask on this cause alone: a
// plain cancel (a shutdown) and any other cause (a bound passing, a check of whether the ask is gone) leave it on its stream
// for the next run. A cause that wraps a Withdrawn (errors.As) withdraws too.
type Withdrawn struct{ Why string }
func (w Withdrawn) Error() string { return w.Why }
// AskDropper ends an ask: the seat's verb drop, served by any holder of the build seat (no node is named). It writes the
// ask in the build module's shared state of pauses, so every node refuses it when the stream gives it, and the node that
// runs it ends the step (ADR 0664 §5).
type AskDropper interface {
DropAsk(ctx context.Context, a buildstep.StepAsk, why string) error
}
// SeatCaller calls a seat's verb on whichever holder answers (mesh-sdk process.Conn.Call).
type SeatCaller interface {
Call(key string, body any) (json.RawMessage, error)
}
// SeatDrops is an AskDropper that calls the verb drop of the ask's build seat over the bus.
type SeatDrops struct{ Caller SeatCaller }
// DropAsk calls <seat>.drop for a, naming its asker and why.
func (d SeatDrops) DropAsk(_ context.Context, a buildstep.StepAsk, why string) error {
seat, err := buildstep.BuildSeat(a.Language)
if err != nil {
return err
}
if _, err := d.Caller.Call("seat:"+seat+".drop", map[string]any{"id": a.ID, "by": a.Asker, "why": why}); err != nil {
return fmt.Errorf("dropping the ask %s: %w", a.ID, err)
}
return nil
}
// dropped ends a without its end. A failed drop is not the wait's error, but it is said (Logf).
func (k Asker) dropped(a buildstep.StepAsk, why string) {
if k.Dropper == nil {
return
}
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
if err := k.Dropper.DropAsk(ctx, a, why); err != nil {
say := k.Logf
if say == nil {
say = log.Printf
}
say("the ask %s was not dropped (%s), it stays on its stream: %v", a.ID, why, err)
}
}
// Put puts a on its stream: the seat's accept subject of its language and kind.
@@ -79,6 +139,7 @@ func (k Asker) Wait(ctx context.Context, a buildstep.StepAsk) (buildstep.Step, e
}
}()
running, since, phase, phaseAt := "", time.Time{}, "", time.Time{}
var agains []string // the node of each run that ended on a transport fault, in order (ADR 0664 §3)
standStill := func() error {
if phase == "" {
return fmt.Errorf("the ask %s's step on %s did not end within %s of its start: it stood still, saying no phase", a.ID, running, k.Bound)
@@ -89,8 +150,15 @@ func (k Asker) Wait(ctx context.Context, a buildstep.StepAsk) (buildstep.Step, e
for {
select {
case <-ctx.Done():
var w Withdrawn
if errors.As(context.Cause(ctx), &w) {
k.dropped(a, context.Cause(ctx).Error())
}
return buildstep.Step{}, ctx.Err()
case <-bound:
if len(agains) > 0 && phase == "given again after a transport fault" {
k.dropped(a, "the bus did not give it again within its bound")
}
return buildstep.Step{}, standStill()
case <-outerCap:
return buildstep.Step{}, fmt.Errorf("the ask %s's step on %s did not end within %s of its start: it kept moving (last phase %q at %s) but never ended",
@@ -102,7 +170,26 @@ func (k Asker) Wait(ctx context.Context, a buildstep.StepAsk) (buildstep.Step, e
if seen := stepSeenOf(ctx); seen != nil {
seen(step)
}
if step.State != buildstep.StepRunning {
if step.State == buildstep.StepAgain {
// A run ended on a transport fault and the bus gives the ask again: not an end. The bound runs on from the
// time it is given again, so a bus that never gives it says so (ADR 0664 §3).
agains = append(agains, step.Node)
running, since, phase, phaseAt = step.Node, step.Started, "given again after a transport fault", step.AgainAt
if k.Bound > 0 {
if timer != nil {
timer.Stop()
}
timer = time.NewTimer(max(time.Until(step.AgainAt.Add(k.Bound)), 0))
bound = timer.C
}
continue
}
if buildstep.Ended(step) {
if step.Passing && step.State == buildstep.StepFailed && (len(agains) > 0 || step.Run > 1) {
// The step's own count of runs (ADR 0664 §2), not only the ones this wait saw: a builder that started again
// reads the key's last value and none of the runs before it.
step.Reason = fmt.Sprintf(buildstep.TransportFaultWords+"%d runs (%s): %s", max(step.Run, len(agains)+1), strings.Join(append(agains, step.Node), ", "), step.Reason)
}
return step, nil
}
if k.Bound <= 0 {
modules/mesh-delivery-builder/buildrun/asks_test.go
+85
View File
@@ -4,6 +4,7 @@ import (
"context"
"encoding/json"
"errors"
"fmt"
"strings"
"sync"
"testing"
@@ -40,6 +41,72 @@ 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)
}
// Issue 1748 (ADR 0664 §3): a run that ended on a transport fault is no end; the next run's end is the answer, in either
// arrival order; three runs then failed say a transport fault on 3 runs with each node; a bus that never gives the ask
// again ends the wait saying so.
for name, order := range map[string][]buildstep.Step{
"again then passed": {{ID: "mvl0frv18d1971", Node: "ace", State: buildstep.StepAgain, Run: 1, AgainAt: time.Now()}, {ID: "mvl0frv18d1971", Node: "g14", State: buildstep.StepPassed, Run: 2}},
"passed then again": {{ID: "mvl0frv18d1971", Node: "g14", State: buildstep.StepPassed, Run: 2}, {ID: "mvl0frv18d1971", Node: "ace", State: buildstep.StepAgain, Run: 1, AgainAt: time.Now()}},
} {
h := &FakeAsks{}
for _, st := range order {
h.Write(st)
}
if got, err := (Asker{Publisher: h, Waiter: h, Bound: time.Minute}).Wait(context.Background(), goAsk("mvl0frv18d1971")); err != nil || got.State != buildstep.StepPassed {
t.Errorf("%s: %+v, %v", name, got, err)
}
}
h := &FakeAsks{}
for i, node := range []string{"ace", "g14", "shanks"} {
st := buildstep.Step{ID: "mvl0frv18d1972", Node: node, State: buildstep.StepAgain, Run: i + 1, AgainAt: time.Now()}
if i == 2 {
st.State, st.Passing, st.Reason = buildstep.StepFailed, true, "lookup proxy: i/o timeout"
}
h.Write(st)
}
if _, err := (Asker{Publisher: h, Waiter: h, Bound: time.Minute}).Ask(context.Background(), goAsk("mvl0frv18d1972")); err == nil || !strings.Contains(err.Error(), "a transport fault on 3 runs (ace, g14, shanks): lookup proxy: i/o timeout") {
t.Errorf("three runs then failed: %v", err)
}
h = &FakeAsks{}
h.Write(buildstep.Step{ID: "mvl0frv18d1973", Node: "ace", State: buildstep.StepAgain, Run: 1, AgainAt: time.Now(), Started: time.Now()})
if _, err := (Asker{Publisher: h, Waiter: h, Bound: 20 * time.Millisecond}).Wait(context.Background(), goAsk("mvl0frv18d1973")); err == nil || !strings.Contains(err.Error(), "given again after a transport fault") {
t.Errorf("an ask never given again: %v", err)
}
// ADR 0664 §5: that ask is dropped; a wait ended by a named cause is dropped too, an ask no node took included (it waits in
// its stream); a shutdown's plain cancel is not.
drops := &seatDrops{}
h = &FakeAsks{}
h.Write(buildstep.Step{ID: "mvl0frv18d1974", Node: "ace", State: buildstep.StepAgain, Run: 1, AgainAt: time.Now(), Started: time.Now()})
(Asker{Publisher: h, Waiter: h, Bound: 20 * time.Millisecond, Dropper: drops}).Wait(context.Background(), goAsk("mvl0frv18d1974"))
named, stop := context.WithCancelCause(context.Background())
h.Write(buildstep.Step{ID: "mvl0frv18d1975", Node: "g14", State: buildstep.StepRunning, Started: time.Now()})
seenStep := WithStepSeen(named, func(buildstep.Step) { stop(Withdrawn{Why: "superseded by a newer fingerprint"}) })
(Asker{Publisher: h, Waiter: h, Bound: time.Hour, Dropper: drops}).Wait(seenStep, goAsk("mvl0frv18d1975"))
shut, off := context.WithCancel(context.Background())
off()
(Asker{Publisher: h, Waiter: h, Bound: time.Hour, Dropper: drops}).Wait(shut, goAsk("mvl0frv18d1977"))
(Asker{Publisher: &FakeAsks{}, Waiter: &FakeAsks{}, Bound: time.Hour, Dropper: drops}).Wait(named, goAsk("mvl0frv18d1976"))
// Review 1 (finding 1): a cause that is no withdrawal (the builder's check bound passing, which waits on) drops nothing.
bounded, passed := context.WithCancelCause(context.Background())
passed(errors.New("the check bound passed"))
(Asker{Publisher: &FakeAsks{}, Waiter: &FakeAsks{}, Bound: time.Hour, Dropper: drops}).Wait(bounded, goAsk("mvl0frv18d1978"))
if got := strings.Join(drops.got, "|"); !strings.Contains(got, "mvl0frv18d1974") || !strings.Contains(got, "mvl0frv18d1975 superseded by a newer fingerprint") || !strings.Contains(got, "mvl0frv18d1976 superseded by a newer fingerprint") || len(drops.got) != 3 {
t.Errorf("drops %q, want the ask never given again, the superseded one and the one no node took, and no other", got)
}
// Review 1: a wait that starts after the first runs (a builder started again) reads the last value only, and its end still says
// how many runs; a drop that failed is said, not discarded.
late := &FakeAsks{}
late.Write(buildstep.Step{ID: "mvl0frv18d1979", Node: "shanks", State: buildstep.StepFailed, Passing: true, Run: 3, Reason: "lookup proxy: i/o timeout"})
if st, err := (Asker{Publisher: late, Waiter: late, Bound: time.Minute}).Wait(context.Background(), goAsk("mvl0frv18d1979")); err != nil || !strings.Contains(st.Reason, "a transport fault on 3 runs") {
t.Errorf("a failed step of run 3 read alone says %q (%v), want it names 3 runs", st.Reason, err)
}
var said []string
bad, off2 := context.WithCancelCause(context.Background())
off2(Withdrawn{Why: "withdrawn"})
(Asker{Publisher: &FakeAsks{}, Waiter: &FakeAsks{}, Bound: time.Hour, Dropper: failingDrops{}, Logf: func(f string, a ...any) { said = append(said, fmt.Sprintf(f, a...)) }}).Wait(bad, goAsk("mvl0frv18d1980"))
if len(said) != 1 || !strings.Contains(said[0], "mvl0frv18d1980") || !strings.Contains(said[0], "no holder") {
t.Errorf("a drop that failed said %q, want it named with its cause", said)
}
// review round 2 item 1: an ask that waits in its stream for a free slot is not bounded; once a node writes its step
// running, the step is bounded from its start, and the error names the node.
g := &FakeAsks{}
@@ -235,3 +302,21 @@ func (c *closingWatch) WatchAsk(ctx context.Context, module, askID string) (<-ch
}
return c.inner.WatchAsk(ctx, module, askID)
}
// seatDrops keeps the asks an Asker dropped: "<id> <why>".
type seatDrops struct{ got []string }
func (d *seatDrops) DropAsk(_ context.Context, a buildstep.StepAsk, why string) error {
if strings.Contains(why, "bound") {
why = ""
}
d.got = append(d.got, strings.TrimSpace(a.ID+" "+why))
return nil
}
// failingDrops is an AskDropper whose verb is not answered.
type failingDrops struct{}
func (failingDrops) DropAsk(context.Context, buildstep.StepAsk, string) error {
return errors.New("no holder answered")
}
modules/mesh-delivery-builder/buildrun/heardends.go
+2 -2
View File
@@ -112,7 +112,7 @@ func (h *HeardEnds) WatchAsk(ctx context.Context, module, askID string) (<-chan
continue
}
for _, s := range steps {
if s.State != buildstep.StepRunning {
if buildstep.Ended(s) {
h.logf("the ask %s's step on %s ended %s, read at a nudge: its watch had not handed it over", askID, s.Node, s.State)
}
if !send(s) {
@@ -206,7 +206,7 @@ func (h *HeardEnds) follow(ctx context.Context, kv jetstream.KeyValue, module st
continue
}
heard = true
if step, err := buildstep.DecodeStep(e.Value()); err == nil && step.State != buildstep.StepRunning {
if step, err := buildstep.DecodeStep(e.Value()); err == nil && buildstep.Ended(step) {
h.Nudge(module)
}
}
modules/mesh-delivery-builder/builds/builds.go
+7
View File
@@ -194,6 +194,13 @@ func FailedUntilChanged(err string) bool {
return strings.HasPrefix(err, NotPinned) || strings.HasPrefix(err, VersionTaken) || strings.HasPrefix(err, RefusedKind)
}
// FailedOnTransportFault says whether a failed key's error is a step's end after its third run met a transport fault
// (buildstep.TransportFaultWords): a fault of the network, not of the source, so the builder does not retake it by itself
// (ADR 0664 §3, which changes ADR 0429 clause 18 for this case); a person's hand act or a builder that starts again does.
func FailedOnTransportFault(err string) bool {
return strings.Contains(err, buildstep.TransportFaultWords)
}
// ToolchainUsed is what a build used for one language (ADR 0405 clauses 2 and 7): the installed toolchain's version and
// the build module's recipe version.
type ToolchainUsed struct {
modules/mesh-delivery-builder/builds/reference_test.go
+4
View File
@@ -103,11 +103,15 @@ func TestOnlyAPushFailureIsBuiltAgainBeforeTheFingerprintChanges(t *testing.T) {
{VersionTaken + "@novox/mesh-sdk@0.1.15 holds another file", false},
{RefusedKind + "a package in go", false},
{"go test: 2 failed", true},
{"a transport fault on 3 runs (ace, g14, novox)", true}, // ADR 0664: not "until changed" (a person's hand act builds it); the builder's own retake is FailedOnTransportFault's
} {
if got := FailedUntilChanged(c.err); got == c.again {
t.Errorf("FailedUntilChanged(%q) = %v", c.err, got)
}
}
if !FailedOnTransportFault("mesh-build-go on ace: a transport fault on 3 runs (ace): i/o timeout") || FailedOnTransportFault("go test: 2 failed") || FailedOnTransportFault(NotStored+"code: 500") {
t.Error("FailedOnTransportFault must name the third run's end alone: the builder's own retake holds it (ADR 0664 §3)")
}
if NotStored != "not stored: " || NotPinned != "not pinned: " || VersionTaken != "version taken: " || RefusedKind != "refused kind: " {
t.Fatal("a prefix changed its words")
}
modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go
+1 -1
View File
@@ -175,7 +175,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
numberer.Publishing = pubs.arm
b := &build.Builder{
Claimer: build.Claimer{Store: kvStore{buildsKV}, By: cred.Node + "." + cred.Module},
Asks: buildrun.Asker{Publisher: asks, Waiter: buildrun.LostAskWaiter{Inner: heard, Dead: dead, Hearing: hearing}, Bound: bound},
Asks: buildrun.Asker{Publisher: asks, Waiter: buildrun.LostAskWaiter{Inner: heard, Dead: dead, Hearing: hearing}, Bound: bound, Dropper: buildrun.SeatDrops{Caller: calls}},
Held: heldOnTheBus{js},
// An ask that can no longer be answered is asked again (issue 1945), read at the step bound: a context's bound.
Gone: heldOnTheBus{js},
modules/mesh-delivery-builder/internal/build/askagain.go
+19 -4
View File
@@ -98,10 +98,11 @@ func Proves(taken, waiting string) bool {
}
// Waits is whether a message ahead still waits to be delivered: no build node wrote a step for it, and no ask of its
// kind asked after it was taken. A message its worker gave up on (delivered its most times) stays on a work queue until
// its age, and is not counted.
// kind asked after it was taken; or a node wrote its step again, a run that ended on a transport fault and that the bus
// gives once more after its delay (ADR 0664). A message its worker gave up on (delivered its most times) stays on a work
// queue until its age, and is not counted; a dropped ask has its step refused and is not counted either.
func (s AskSeen) Waits(q Queued) bool {
return q.Step == "" && !s.passed(q.At)
return q.Step == sdkstep.StepAgain || q.Step == "" && !s.passed(q.At)
}
func (s AskSeen) passed(at time.Time) bool {
@@ -300,6 +301,9 @@ type byHand struct{ why string }
func (h byHand) Error() string { return "asked again by hand: " + h.why }
// Unwrap makes the ask asked again by hand a withdrawal of the old ask: its asker drops it (ADR 0664 §4).
func (h byHand) Unwrap() error { return buildrun.Withdrawn{Why: h.Error()} }
// answer waits for the step of the lane's ask, and asks it again while it can no longer be answered (Gone) or a person
// asked it again (AskAgain); it answers the step and the ask that has it.
func (b *Builder) answer(ctx context.Context, l *lane, ask sdkstep.StepAsk, asked *askNotes, module, fp string) (sdkstep.Step, sdkstep.StepAsk, error) {
@@ -310,7 +314,18 @@ func (b *Builder) answer(ctx context.Context, l *lane, ask sdkstep.StepAsk, aske
var firstDecline time.Time
declines := map[string]string{} // node -> why it declined, the last
for {
step, why, err := b.waitOnce(ctx, ask, again)
// Each run that ended on a transport fault is said in the log and on the key's step; the bus gives the ask again (ADR 0664 §3).
seen := buildrun.WithStepSeen(ctx, func(st sdkstep.Step) {
if st.State != sdkstep.StepAgain {
return
}
said := fmt.Sprintf("run %d of %s on %s ended on a transport fault, given again by the bus at %s: %s", st.Run, l.module, st.Node, st.AgainAt.UTC().Format(time.RFC3339), st.Reason)
b.logf("%s: %s", key, said)
if err := asked.note(ctx, ask.Language, ask.ID, said); err != nil {
b.logf("%s: the run said on the key's step could not be written: %v", key, err)
}
})
step, why, err := b.waitOnce(seen, ask, again)
switch {
case errors.Is(err, errCheck):
continue
modules/mesh-delivery-builder/internal/build/claim.go
+5
View File
@@ -51,6 +51,9 @@ type ClaimOptions struct {
// TakeOver takes a key found building whoever holds it: the builder has just started and looks at the key for the first
// time, and a build left building by a builder that is gone, on this node or another, is nobody's (ADR 0478 clause 6).
TakeOver bool
// ByHand is a person's hand act (BuildAgain): a key failed on a transport fault is taken again only by it, or by a builder
// that has just started (TakeOver), never by the builder's own retake (ADR 0664 §3).
ByHand bool
// ImageBuilder says whether a node runs a healthy image build module now: a key failed for NoImageBuilderPrefix is taken
// again only then. Nil means none does.
ImageBuilder func() bool
@@ -210,6 +213,8 @@ func (c Claimer) Claim(ctx context.Context, module, fingerprint, manifest string
return nil, nil
case cur.State == builds.BuildFailed && (builds.FailedUntilChanged(cur.Error) || strings.HasPrefix(cur.Error, buildrun.DigestsDifferPrefix)):
return nil, nil
case cur.State == builds.BuildFailed && builds.FailedOnTransportFault(cur.Error) && !opt.ByHand && !opt.TakeOver:
return nil, nil
case cur.State == builds.BuildFailed && strings.HasPrefix(cur.Error, NoImageBuilderPrefix) && (opt.ImageBuilder == nil || !opt.ImageBuilder()):
return nil, nil
case cur.State == builds.BuildFailed:
modules/mesh-delivery-builder/internal/build/claim_test.go
+12
View File
@@ -152,6 +152,18 @@ func TestAFailedKeyIsTakenAgainAsANewRevisionExceptARefusedKind(t *testing.T) {
if got, err := (Claimer{Store: st, By: "novox.a"}).Claim(context.Background(), "m", fp1, "", ClaimOptions{}); err != nil || got != nil || st.writes != writes {
t.Errorf("a refused kind: claim = %v, %v, writes %d (was %d), want none and no write", got, err, st.writes, writes)
}
// ADR 0664 §3 (review 1, finding 4): a key that failed on a transport fault is not taken again by the builder's own retake; a
// person's hand act and a builder that has just started take it.
for _, o := range []struct {
opt ClaimOptions
take bool
}{{ClaimOptions{}, false}, {ClaimOptions{ByHand: true}, true}, {ClaimOptions{TakeOver: true}, true}} {
st = newMem()
seed(t, st, builds.BuilderBuild{State: builds.BuildFailed, By: "novox.a", Error: "mesh-build-go on ace: a transport fault on 3 runs (ace, g14, novox): lookup proxy: i/o timeout"})
if got, err := (Claimer{Store: st, By: "novox.a"}).Claim(context.Background(), "m", fp1, "", o.opt); err != nil || (got != nil) != o.take {
t.Errorf("a transport fault on 3 runs with %+v: claim = %v, %v, want taken %v", o.opt, got, err, o.take)
}
}
for _, e := range []string{builds.NotPinned + "up: nats:2 names no digest", builds.VersionTaken + "@novox/mesh-sdk@0.1.15"} { // B3, B5: ADR 0498 clauses 4, 5
st = newMem()
seed(t, st, builds.BuilderBuild{State: builds.BuildFailed, By: "novox.a", Error: e})
modules/mesh-delivery-builder/internal/build/run.go
+32 -2
View File
@@ -121,6 +121,9 @@ type Builder struct {
waiting map[string]map[string]chan string
// hands are the requests to build a key again by hand that wait for the claim and its asks (BuildAgain), by build key.
hands map[string]*handWatch
// builds are the ends that withdraw the asks of a key's build in flight, by build key (ADR 0664 §4): a newer fingerprint of
// the module takes its older one's build out, and the asker drops that build's asks.
builds map[string]context.CancelCauseFunc
}
type planned struct {
@@ -139,6 +142,7 @@ func (b *Builder) init() {
b.known, b.inflight, b.again, b.plans, b.tried = map[string]meshcatalog.ModulesByModuleValue{}, map[string]bool{}, map[string]meshcatalog.ModulesByModuleValue{}, map[string]planned{}, map[string]bool{}
b.published, b.held, b.waiting = map[string]bool{}, map[string]bool{}, map[string]map[string]chan string{}
b.hands = map[string]*handWatch{}
b.builds = map[string]context.CancelCauseFunc{}
})
}
@@ -161,13 +165,25 @@ func (b *Builder) Module(ctx context.Context, key string, value []byte) {
return
}
b.mu.Lock()
old, had := b.known[key]
b.known[key] = m
var superseded context.CancelCauseFunc
if had && old.Fp2 != "" && old.Fp2 != m.Fp2 {
if ok, err := builds.BuilderBuildKey(key, old.Fp2); err == nil {
superseded = b.builds[ok]
}
}
if bk, err := builds.BuilderBuildKey(key, m.Fp2); err == nil && m.Fp2 != "" {
// A fingerprint the trunk carries is built from the trunk's commit, whoever asked for it first.
b.trunkFP[bk] = true
delete(b.via, bk)
}
b.mu.Unlock()
if superseded != nil {
// A newer fingerprint of the module: the build of the older one in flight is taken out, and its asks are withdrawn
// from the build seats (ADR 0664 §4).
superseded(buildrun.Withdrawn{Why: "the build of " + key + " at " + old.Fp2 + " was superseded by the fingerprint " + m.Fp2})
}
b.consider(ctx, key, m)
}
@@ -448,6 +464,7 @@ func (b *Builder) try(ctx context.Context, module string, m meshcatalog.ModulesB
trunk := b.trunkFP[key]
first := !b.tried[key]
b.tried[key] = true
byHand := b.hands[key] != nil
b.mu.Unlock()
var numbering versions.Numbering
tag := ""
@@ -460,7 +477,7 @@ func (b *Builder) try(ctx context.Context, module string, m meshcatalog.ModulesB
return
}
}
cl, err := b.Claimer.Claim(ctx, module, m.Fp2, m.Manifest, ClaimOptions{TakeOver: first, ImageBuilder: imageSeat, Trunk: trunk, Published: b.isPublished,
cl, err := b.Claimer.Claim(ctx, module, m.Fp2, m.Manifest, ClaimOptions{TakeOver: first, ByHand: byHand, ImageBuilder: imageSeat, Trunk: trunk, Published: b.isPublished,
Stored: b.storedOnce(key)})
if err != nil {
b.logf("%s: %v", module, err)
@@ -498,7 +515,20 @@ func (b *Builder) try(ctx context.Context, module string, m meshcatalog.ModulesB
// any other, so the controller has a done key to register it from (issue 1613, ADR 0478 clause 1, ADR 0501 clause 1).
result = builds.BuilderBuild{State: builds.BuildDone, Step: "nothing to build: the manifest names no artifact"}
} else {
result = b.build(ctx, module, m, plan, planErr, prior, only, b.notes(cl))
run, withdraw := context.WithCancelCause(ctx)
b.mu.Lock()
b.builds[key] = withdraw
b.mu.Unlock()
result = b.build(run, module, m, plan, planErr, prior, only, b.notes(cl))
b.mu.Lock()
delete(b.builds, key)
b.mu.Unlock()
withdraw(nil)
var w buildrun.Withdrawn
if ctx.Err() == nil && errors.As(context.Cause(run), &w) {
// Taken out, not stopped: the key says so, and the asks are withdrawn (the asker dropped them).
result = builds.BuilderBuild{State: builds.BuildFailed, Error: keyError([]string{"withdrawn: " + w.Why})}
}
}
if ctx.Err() != nil {
return
modules/mesh-delivery-builder/internal/build/run_test.go
+68
View File
@@ -624,6 +624,7 @@ func TestTwoInstancesStartingTogetherEndWithOneDoneKey(t *testing.T) { // 1607/A
}
func TestAModulesValueThatArrivesWhileItsKeyIsBeingBuiltIsLookedAtAgainWhenItEnds(t *testing.T) { // 1607/AC6 (no event is dropped, no timer)
t.Run("a newer fingerprint withdraws the build of the older one", aNewerFingerprintWithdrawsTheBuildOfTheOlderOne)
r := newRig(t, goBundle)
r.bm.fails["ace/build_go_build"] = "go build of tool: boom"
r.bm.hold, r.bm.entered = make(chan struct{}), make(chan struct{}, 8)
@@ -640,6 +641,60 @@ func TestAModulesValueThatArrivesWhileItsKeyIsBeingBuiltIsLookedAtAgainWhenItEnd
}
}
// ADR 0664 §4 (issue 1748, review 1 finding 2): a newer fingerprint of a module takes the build of its older one out: the key
// says so, and the asker drops the ask the build waits on, carrying the cause. The first ask's step runs and never ends.
func aNewerFingerprintWithdrawsTheBuildOfTheOlderOne(t *testing.T) {
r := newRig(t, goBundle)
drops := &seatDrops{}
var asked int32
running := make(chan struct{})
r.asks.Answer = func(a buildstep.StepAsk) buildstep.Step {
if atomic.AddInt32(&asked, 1) == 1 {
defer close(running)
return buildstep.Step{ID: a.ID, Node: "ace", State: buildstep.StepRunning, Started: time.Now()}
}
return r.answer(a)
}
r.b.Asks = buildrun.Asker{Publisher: r.asks, Waiter: r.asks, Bound: 2 * time.Second, Dropper: drops}
r.b.Module(context.Background(), "m", moduleValue(t, "modules/m"))
<-running
fp2 := "src2:" + hex64('2')
data, err := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Repository: "novox/mesh-catalog", Commit: "7a74cee83f88e6585a5511c3cc53bdf0f5303d20", Fp: "src1:" + hex64('f'), Fp2: fp2,
Manifest: mdigest('a'), Path: "modules/m", Toolchains: CatalogToolchains(agreedToolchains)})
if err != nil {
t.Fatal(err)
}
r.b.Module(context.Background(), "m", data)
r.b.Wait()
old := valueOf(t, r.st, r.key)
if old.State != builds.BuildFailed || !strings.Contains(old.Error, "superseded by the fingerprint "+fp2) {
t.Errorf("the older key = %+v, want failed, naming the newer fingerprint that superseded it", old)
}
first := r.asks.Asks()[0].ID
if got := drops.get(); len(got) != 1 || !strings.HasPrefix(got[0], first+" ") || !strings.Contains(got[0], "superseded by the fingerprint "+fp2) {
t.Errorf("dropped %q, want the first ask withdrawn with the cause", got)
}
}
// seatDrops keeps the asks an Asker withdrew: "<id> <why>".
type seatDrops struct {
mu sync.Mutex
got []string
}
func (d *seatDrops) DropAsk(_ context.Context, a buildstep.StepAsk, why string) error {
d.mu.Lock()
defer d.mu.Unlock()
d.got = append(d.got, a.ID+" "+why)
return nil
}
func (d *seatDrops) get() []string {
d.mu.Lock()
defer d.mu.Unlock()
return append([]string(nil), d.got...)
}
func TestAKeyIsBuiltOnlyWithTheToolchainItsFingerprintNamesAndNothingIsWrittenOtherwise(t *testing.T) { // 1615/AC4
r := newRig(t, goBundle)
old := map[string]builds.ToolchainUsed{"go": {Version: "go1.26.0", Recipe: "1"}}
@@ -779,8 +834,19 @@ func TestAFailedPublishOfACandidatesImageIsWrittenToTheKey(t *testing.T) { // re
t.Fatalf("candidate build = %+v", v)
}
r.bm.fails["ace/build_image_build"] = "push to the registry: connection refused"
// Issue 1748: a run of the ask that ended on a transport fault is said in the log and on the key, and is no end.
inner, log := r.asks.Answer, &logged{}
r.asks.Answer = func(a buildstep.StepAsk) buildstep.Step {
r.asks.Write(buildstep.Step{ID: a.ID, Node: "g14", State: buildstep.StepAgain, Run: 1, AgainAt: time.Now(), Reason: "lookup registry: i/o timeout"})
return inner(a)
}
r.b.Logf = log.logf
r.b.Module(context.Background(), "m", moduleValue(t, "modules/m")) // the merge: the trunk names the fingerprint
r.b.Wait()
r.asks.Answer, r.b.Logf = inner, t.Logf
if !log.has("run 1 of", "on g14 ended on a transport fault", "lookup registry: i/o timeout") {
t.Errorf("a run given again was not said: %v", log.lines)
}
v := valueOf(t, r.st, r.key)
if v.State != builds.BuildFailed || !strings.Contains(v.Error, "connection refused") || !strings.Contains(v.Error, "publish") {
t.Fatalf("build = %+v, want failed naming the publish and its reason", v)
@@ -1144,6 +1210,8 @@ func TestABuildAskWaitingInItsStreamIsNotBounded(t *testing.T) {
// Note 62: seven messages SEAT_NODE_BUILD_GO gave up on sat ahead of a Go ask for five hours.
{"seven given up on ahead, a later ask taken", seen(ahead(7, ""), later), "behind 7 given up on or taken", 0},
{"seven taken ahead (failed as poison), a later ask taken", seen(ahead(7, buildstep.StepFailed), later), "no longer delivers it", 0},
{"two to be given again after a transport fault, a later ask taken", seen(ahead(2, buildstep.StepAgain), later), "", 2},
{"two dropped (step refused) ahead, a later ask taken", seen(ahead(2, buildstep.StepRefused), later), "no longer delivers it", 0},
{"a check ask taken later proves a delivery ask passed", seen(ahead(7, ""), taken(buildstep.AskCheck, put.Add(time.Minute))), "no longer delivers it", 0},
{"a delivery ask taken later does not prove a check ask passed", func() AskSeen { s := seen(ahead(3, ""), later); s.Kind = buildstep.AskCheck; return s }(), "", 3},
} {
modules/mesh-delivery-builder/manifest_test.go
+1 -1
View File
@@ -125,7 +125,7 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1
// Design 51 *The floor*: the floor's three sources, the registry seats' versions, bridge_tags and indexer_requirements.
if want := []string{"mesh-delivery-assigner.assigner_read_assignments", "mesh-delivery-forge-bridge.bridge_tags", "mesh-delivery-indexer.indexer_read_candidates", "mesh-delivery-indexer.indexer_read_manifests",
"mesh-delivery-indexer.indexer_read_modules", "mesh-delivery-indexer.indexer_read_packages", "mesh-delivery-indexer.indexer_requirements", "seat:container-image-registry.remove", "seat:container-image-registry.versions",
"seat:go-package-registry.remove", "seat:go-package-registry.versions", "seat:mesh-artifact-sweep.held", "seat:mesh-hand-acts.hand-act", "seat:node-engine.manifest-reads", "seat:npm-package-registry.remove",
"seat:go-package-registry.remove", "seat:go-package-registry.versions", "seat:mesh-artifact-sweep.held", "seat:mesh-hand-acts.hand-act", "seat:node-build-go.drop", "seat:node-build-image.drop", "seat:node-build-node.drop", "seat:node-engine.manifest-reads", "seat:npm-package-registry.remove",
"seat:npm-package-registry.versions", "seat:oci-artifact-registry.remove", "seat:oci-artifact-registry.versions"}; !reflect.DeepEqual(sorted(m.Invokes), want) || len(m.Config) != 0 {
t.Errorf("invokes = %v, config = %v, want %v and no config", m.Invokes, m.Config, want)
}
modules/mesh-delivery-builder/module.json
+3
View File
@@ -67,6 +67,9 @@
"seat:go-package-registry.versions",
"seat:mesh-artifact-sweep.held",
"seat:mesh-hand-acts.hand-act",
"seat:node-build-go.drop",
"seat:node-build-image.drop",
"seat:node-build-node.drop",
"seat:node-engine.manifest-reads",
"seat:npm-package-registry.remove",
"seat:npm-package-registry.versions",
modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main.go
+1 -1
View File
@@ -397,7 +397,7 @@ func RecoveryTools(r check.Reader, w check.VerdictEntry) []stdio.Tool {
return []stdio.Tool{{Name: check.AskAgainTool,
Description: "Ask a candidate's check again by hand, whose verdict stays running while nothing runs for it (" + check.StatusReadTool + " names it " +
"under lost): its verdict is ended as asked again, saying why, and the checker, which hears its verdicts, stops the wait that judged nothing " +
"and asks the check anew, whole. Refused for a group with no candidate, a verdict that is not running (a pass or failure stands until a push or " +
"and asks the check anew, whole. Refused for a group with no candidate, a verdict that is not running or a failure whose failing steps did not all end on a transport fault (a pass, or any other failure, stands until a push or " +
"a trunk move), and while a step of its pull requests can still run on a build module (its end judges it). The checker reads a step that " +
"ended while its watch was deaf by itself at the next step end of its build module; this is for what that does not reach.",
Input: map[string]any{"type": "object", "required": []string{"group", "why"}, "properties": map[string]any{
modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go
+2 -2
View File
@@ -262,7 +262,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
Asker: engineAsks{call: caller.CallOn},
Placed: placedNodes{kv: func(context.Context) (jetstream.KeyValue, error) { return assignments, nil }}.Placed,
Bound: answerBound,
Sandbox: seatSandbox{asker: buildrun.Asker{Publisher: asks, Waiter: waiter, Bound: stepBound}},
Sandbox: seatSandbox{asker: buildrun.Asker{Publisher: asks, Waiter: waiter, Bound: stepBound, Dropper: buildrun.SeatDrops{Caller: caller}}},
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] },
@@ -304,7 +304,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
}
}
ch.Judge = newJudge(judgeDeps{
asks: buildrun.Asker{Publisher: asks, Waiter: waiter, Bound: stepBound}, tools: tools, trunks: trunks,
asks: buildrun.Asker{Publisher: asks, Waiter: waiter, Bound: stepBound, Dropper: buildrun.SeatDrops{Caller: caller}}, 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, tags: tags, inputs: runInputs{awaitedKV{inputsKV}}, beginBound: beginBound,
modules/mesh-delivery-checker/internal/check/askagain.go
+41 -3
View File
@@ -10,6 +10,8 @@ import (
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliverycheckerclient"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdeliverychecker"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun"
"git.novox.be/novox/mesh-delivery/sdk/buildstep"
)
// **A check is asked again by hand** (the lost check of 2026-10-28; tools-are-how-a-module-is-spoken-to: every recovery
@@ -69,7 +71,7 @@ const AskAgainTool = "checker_ask_again"
const AskedAgainWords = EndedWords + "asked again by hand (" + AskAgainTool + "): "
// errAskedAgain is the cause a running check is ended with when its verdict was asked again by hand.
var errAskedAgain = errors.New("its verdict was asked again by hand (" + AskAgainTool + ")")
var errAskedAgain = buildrun.Withdrawn{Why: "its verdict was asked again by hand (" + AskAgainTool + ")"}
// IsAskedAgain says whether a verdict was written by AskAgainTool.
func IsAskedAgain(v meshdeliverychecker.VerdictsByGroupHeadsValue) bool {
@@ -92,7 +94,7 @@ type AskedAgain struct {
}
// AskAgain ends the running verdict of the group's candidate as asked again by hand, saying why. It refuses a group with
// no candidate, a verdict that is not running (a pass or a failure stands until a push or a trunk move), one a step of
// no candidate, a verdict that is not running (a pass stands, and a failure unless every failing step ended on a transport fault, ADR 0664; both stand until a push or a trunk move), one a step of
// whose pull requests can still run, and one whose asks are not known or one of which no node took yet.
func (r Reader) AskAgain(w VerdictEntry, group, why string) (AskedAgain, error) {
why = strings.TrimSpace(why)
@@ -121,15 +123,26 @@ func (r Reader) AskAgain(w VerdictEntry, group, why string) (AskedAgain, error)
if err != nil {
return AskedAgain{}, fmt.Errorf("%s holds what is not a verdict: %w", key, err)
}
if v.State != meshdeliverycheckerclient.VerdictRunning {
failed := v.State == meshdeliverycheckerclient.VerdictFail
if v.State != meshdeliverycheckerclient.VerdictRunning && !failed {
return AskedAgain{}, fmt.Errorf("%s is %s, not running: a verdict stands until a push or a trunk move asks it again", key, v.State)
}
steps := r.readSteps()
if failed {
// ADR 0664 (operator answer 5, issue 1748): a failure stands, except one whose failing steps all ended on a transport
// fault (a step written passing): the network failed, not the change, so it is asked again like a lost check.
if err := steps.onlyTransportFaults(c); err != nil {
return AskedAgain{}, fmt.Errorf("%s is fail, and a failure stands until a push or a trunk move asks it again, unless every failing step of it ended on a transport fault: %w", key, err)
}
}
for _, rs := range steps.running {
if asksOf(c, rs.What) {
return AskedAgain{}, fmt.Errorf("a step of it runs: %s %s on %s since %s (%s); its end judges it", rs.Module, rs.ID, rs.Node, rs.Since, rs.What)
}
}
if failed {
return r.askedAgain(w, key, rev, v, why)
}
if r.AskState.Get == nil {
return AskedAgain{}, fmt.Errorf("the checker's asks are not readable here, so whether its ask waits in a build seat's stream cannot be told")
}
@@ -151,6 +164,31 @@ func (r Reader) AskAgain(w VerdictEntry, group, why string) (AskedAgain, error)
"the seats' backlog is in mesh-delivery-view.build_usage", id)
}
}
return r.askedAgain(w, key, rev, v, why)
}
// onlyTransportFaults is nil when the candidate's pull requests have a failed or refused step and every one of them ended
// passing (buildstep.Step.Passing); else why not.
func (s stepsRead) onlyTransportFaults(c meshcatalog.CandidatesByGroupValue) error {
n := 0
for _, ms := range s.all {
st := ms.step
if !asksOf(c, st.What) || (st.State != buildstep.StepFailed && st.State != buildstep.StepRefused) {
continue
}
if !st.Passing {
return fmt.Errorf("its step %s on %s ended %s and not on a transport fault: %s", st.ID, st.Node, st.State, st.Reason)
}
n++
}
if n == 0 {
return errors.New("no failed step of it is read (the newest keys of each build module's steps are)")
}
return nil
}
// askedAgain writes the verdict v ended as asked again by hand, why saying so.
func (r Reader) askedAgain(w VerdictEntry, key string, rev uint64, v meshdeliverychecker.VerdictsByGroupHeadsValue, why string) (AskedAgain, error) {
words := AskedAgainWords + why
ended := meshdeliverychecker.VerdictsByGroupHeadsValue{Group: v.Group, Heads: v.Heads, Trunk: v.Trunk, State: meshdeliverycheckerclient.VerdictFail,
Failed: &FailedFirst{Machine: "checker", Words: words}}
modules/mesh-delivery-checker/internal/check/checker.go
+4
View File
@@ -18,6 +18,7 @@ import (
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdeliverychecker"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun"
"git.novox.be/novox/mesh-delivery/sdk/buildstep"
"git.novox.be/novox/mesh-sdk/go/keys"
)
@@ -177,6 +178,9 @@ func (s superseded) Error() string {
return EndedWords + "superseded by " + s.by
}
// Unwrap makes a check ended so a withdrawal of its asks: the asker drops what it put on the build nodes (ADR 0664 §4, §5).
func (s superseded) Unwrap() error { return buildrun.Withdrawn{Why: s.Error()} }
func (ch *Checker) init() {
ch.once.Do(func() {
ch.known, ch.inflight, ch.again, ch.tried = map[string]meshcatalog.CandidatesByGroupValue{}, map[string]bool{}, map[string]bool{}, map[string]bool{}
modules/mesh-delivery-checker/internal/check/checker_test.go
+36
View File
@@ -3,6 +3,7 @@ package check
import (
"context"
"encoding/json"
"errors"
"fmt"
"strings"
"sync"
@@ -15,6 +16,7 @@ import (
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdeliverychecker"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun"
"git.novox.be/novox/mesh-delivery/sdk/buildstep"
"git.novox.be/novox/mesh-sdk/go/keys"
)
@@ -362,6 +364,40 @@ func TestARepositoryWithNoTrunkCannotBeJudgedLoudlyUntilItsTrunkIsKnown(t *testi
}
})
}
// ADR 0664 §5 (issue 1748): a candidate taken out (its pull request merged or closed) ends the check running for it with a
// cause its asks' wait names, never a plain cancel, so its asker drops the asks it put on the build nodes.
t.Run("a candidate taken out ends its check with a named cause", func(t *testing.T) {
ch := &Checker{}
ch.init()
run, end := context.WithCancelCause(context.Background())
ch.live["g.heads"] = liveCheck{group: "g", end: end}
// The check's asker is waiting on an ask of its run when the candidate goes (review 1, finding 7).
drops := &checkDrops{}
fake := &buildrun.FakeAsks{}
done := make(chan struct{})
go func() {
defer close(done)
(buildrun.Asker{Publisher: fake, Waiter: fake, Bound: time.Hour, Dropper: drops}).Wait(run, buildstep.StepAsk{ID: "mvl0frv18d2000", Language: buildstep.AskGo, Kind: buildstep.AskCheck})
}()
ch.Forget("g")
<-done
var s superseded
if cause := context.Cause(run); !errors.As(cause, &s) || s.by != "" || errors.Is(cause, context.Canceled) {
t.Errorf("the check was ended with %v, want superseded with no heads, not a plain cancel", cause)
}
if len(drops.got) != 1 || !strings.Contains(drops.got[0], "mvl0frv18d2000") || !strings.Contains(drops.got[0], "no candidate holds these heads") {
t.Errorf("the asker dropped %q, want its ask withdrawn with the check's cause", drops.got)
}
})
}
// checkDrops keeps the asks an asker dropped: "<id> <why>".
type checkDrops struct{ got []string }
func (d *checkDrops) DropAsk(_ context.Context, a buildstep.StepAsk, why string) error {
d.got = append(d.got, a.ID+" "+why)
return nil
}
func countWith(lines []string, words string) int {
modules/mesh-delivery-checker/internal/check/lostask_test.go
+36
View File
@@ -3,6 +3,9 @@ package check
import (
"context"
"encoding/json"
"fmt"
"maps"
"slices"
"strings"
"sync"
"testing"
@@ -347,7 +350,40 @@ func TestACheckWhoseAskJudgedNothingIsAskedAgainOrRead(t *testing.T) {
if _, err := rd.AskAgain(asVerdicts{r.verdicts}, "feat/x", "again"); err == nil {
t.Error("a judged check was asked again")
}
// ADR 0664 (issue 1748, operator answer 5): a failed verdict whose failing steps all ended on a transport fault
// is asked again; one with a failing step that did not, or with none read, stands.
for _, c := range []struct {
name string
passing []bool
ok bool
}{{"every failing step on a transport fault", []bool{true, true}, true}, {"one step failed on the source", []bool{true, false}, false}, {"no failed step read", nil, false}} {
failedKV := newMemKV()
judgedRaw, _, _, _ := r.verdicts.Get(ctx, key)
failed, _ := meshdeliverycheckerclient.DecodeVerdict(judgedRaw)
failed.State, failed.Failed = meshdeliverycheckerclient.VerdictFail, &FailedFirst{Machine: "ace", Words: "a transport fault on 3 runs"}
raw, _ := meshdeliverycheckerclient.EncodeVerdict(failed)
failedKV.Create(ctx, key, raw)
vals := map[string]json.RawMessage{}
for i, p := range c.passing {
id := fmt.Sprint("mvl0frv18d19", 80+i)
vals["ace."+id], _ = buildstep.EncodeStep(buildstep.Step{ID: id, Node: "ace", Tool: "check ask", State: buildstep.StepFailed, Passing: p, Ended: time.Now(), What: "check " + repoA + "@" + short12(head1)})
}
rd := Reader{CandidateState: kvRead(cands2(t, cand)), VerdictState: kvRead(failedKV), AskState: kvRead(newMemKV()), StepStates: map[string]StateRead{buildrun.GoBuildModule: {
Keys: func() ([]string, error) { return slices.Collect(maps.Keys(vals)), nil }, Get: func(k string) (json.RawMessage, error) { return vals[k], nil }}}}
if got, err := rd.AskAgain(asVerdicts{failedKV}, "feat/x", "the network"); (err == nil) != c.ok || (c.ok && !strings.HasPrefix(got.Said, AskedAgainWords)) {
t.Errorf("%s: asked again = %+v, %v, want asked again %v", c.name, got, err, c.ok)
}
}
}
})
}
}
// cands2 is a candidates state holding cand under feat/x.
func cands2(t *testing.T, cand []byte) *memKV {
t.Helper()
kv := newMemKV()
ckey, _ := meshcatalogclient.CandidatesKey("feat/x")
_, _ = kv.Create(context.Background(), ckey, cand)
return kv
}
modules/mesh-delivery-checker/internal/check/read.go
+1 -1
View File
@@ -710,7 +710,7 @@ func (s stepsRead) lostWhy(c meshcatalog.CandidatesByGroupValue) string {
var last *buildstep.Step
for i := range s.all {
st := s.all[i].step
if st.State != buildstep.StepRunning && asksOf(c, st.What) && (last == nil || st.Ended.After(last.Ended)) {
if buildstep.Ended(st) && asksOf(c, st.What) && (last == nil || st.Ended.After(last.Ended)) {
last = &s.all[i].step
}
}
modules/mesh-delivery-checker/internal/check/try.go
+1 -1
View File
@@ -231,7 +231,7 @@ func (t Try) HoldsANode() bool {
if len(t.Steps) == 0 {
return true
}
return slices.ContainsFunc(t.Steps, func(s TryStep) bool { return s.State == buildstep.StepRunning })
return slices.ContainsFunc(t.Steps, func(s TryStep) bool { return s.State == buildstep.StepRunning || s.State == buildstep.StepAgain })
}
// Over says whether the try came to its end.
modules/mesh-delivery-checker/internal/check/try_test.go
+4
View File
@@ -128,6 +128,10 @@ func TestATryIsAskedQueuedOncePerHeadAndAtMostFourAtOnce(t *testing.T) { // ADR
t.Errorf("%s: answered %v, want refused %q", c.what, err, c.refused)
}
}
// Issue 1748: a try whose step ended on a transport fault, the bus giving its ask again, still runs.
if !(Try{State: TryRunning, Steps: []TryStep{{Ask: "a", State: buildstep.StepAgain}}}).HoldsANode() {
t.Error("a try with a step given again no longer runs")
}
}
func TestATryOfARepositoryWithNoCheckedLevelIsRefused(t *testing.T) { // the agreed error of the try
modules/mesh-delivery-checker/manifest_test.go
+2 -2
View File
@@ -131,8 +131,8 @@ func TestManifestDeclarations(t *testing.T) { // 1678/AC1
"mesh-delivery-assigner.assigner_read_assignments", "mesh-delivery-indexer.indexer_read_candidates", "mesh-delivery-indexer.indexer_read_manifests",
"mesh-delivery-indexer.indexer_read_modules", "mesh-delivery-indexer.indexer_read_packages", "mesh-delivery-indexer.indexer_read_predefined",
"mesh-delivery-indexer.indexer_read_seat_definitions",
"seat:mesh-catalog.seat-definitions", "seat:mesh-config.renames", "seat:mesh-hand-acts.hand-act", "seat:node-engine.compose-candidate"}; !reflect.DeepEqual(sorted(m.Invokes), want) {
t.Errorf("invokes = %v, want the forge's reads and compose-candidate", m.Invokes)
"seat:mesh-catalog.seat-definitions", "seat:mesh-config.renames", "seat:mesh-hand-acts.hand-act", "seat:node-build-go.drop", "seat:node-build-node.drop", "seat:node-engine.compose-candidate"}; !reflect.DeepEqual(sorted(m.Invokes), want) {
t.Errorf("invokes = %v, want the forge's reads, the seats' drop and compose-candidate", m.Invokes)
}
if want := []string{"node-build-go", "node-build-node"}; !reflect.DeepEqual(sorted(m.Uses), want) {
t.Errorf("uses = %v, want the Go and Node build seats", m.Uses)
modules/mesh-delivery-checker/module.json
+2
View File
@@ -93,6 +93,8 @@
"seat:mesh-catalog.seat-definitions",
"seat:mesh-config.renames",
"seat:mesh-hand-acts.hand-act",
"seat:node-build-go.drop",
"seat:node-build-node.drop",
"seat:node-engine.compose-candidate"
],
modules/mesh-grants/internal/buscompose/buscompose_test.go
+15
View File
@@ -18,6 +18,7 @@ import (
"git.novox.be/novox/mesh-contracts/identity-and-access/go/clients/meshvaultclient"
"git.novox.be/novox/mesh-contracts/identity-and-access/go/seats/meshvault"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
"git.novox.be/novox/mesh-delivery/sdk/buildstep"
"git.novox.be/novox/mesh-sdk/go/announce"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"github.com/nats-io/nats.go/jetstream"
@@ -146,6 +147,20 @@ func TestTheCopiedComposerComposesAsTheControllers(t *testing.T) {
t.Errorf("may publish %s: %v, want %v", subject, got, want)
}
}
// ADR 0664 §2 with ADR 0436 §6: the work queue offers a build seat's ask more often than the module runs it, so the stream
// never drops an ask before the module has ended it, and at most ten times.
for _, seat := range []string{"node-build-go", "node-build-node", "node-build-image"} {
held := Seat{Name: seat, Scope: "node", Accepts: []string{"delivery", "check", "recheck"}, PulledByKind: true}
_, workers := SeatTrafficObjects([]Principal{{Kind: KindModule, Node: "one", Module: "m-" + seat, Holds: []Seat{held}}})
if len(workers) == 0 {
t.Fatalf("%s: no worker composed", seat)
}
for _, w := range workers {
if w.MaxDeliver <= buildstep.AskRuns || w.MaxDeliver > 10 {
t.Errorf("%s worker %s offers an ask %d times, want more than %d runs and at most 10", seat, w.Name, w.MaxDeliver, buildstep.AskRuns)
}
}
}
plain, err := PermissionsFor(Principal{Kind: KindModule, Node: "one", Module: "shop",
Uses: []Seat{{Name: "telegram-sender", Accepts: []string{"send"}}}})
if err != nil {
sdk/buildrun/stale_test.go
+4
View File
@@ -83,6 +83,7 @@ func TestEndStaleAndEndOne(t *testing.T) {
put("ace", "raced1", "build_go_test", buildstep.StepRunning, boot.Add(-time.Hour), "")
put("ace", "new1", "build_go_fetch", buildstep.StepRunning, boot.Add(time.Second), "")
put("ace", "done1", "build_go_fetch", buildstep.StepPassed, boot.Add(-time.Hour), gone)
put("ace", "again1", "build_go_fetch", buildstep.StepAgain, boot.Add(-time.Hour), gone) // issue 1748: the bus gives it again; the sweep leaves it
put("g14", "other1", "build_go_fetch", buildstep.StepRunning, boot.Add(-time.Hour), gone)
// raced1 is ended by its own run between the read and the write.
raced := false
@@ -97,6 +98,9 @@ func TestEndStaleAndEndOne(t *testing.T) {
if err != nil || len(ended) != 4 {
t.Fatalf("ended %v, %v", ended, err)
}
if s := k.last("ace.again1"); s.State != buildstep.StepAgain {
t.Errorf("the sweep ended a step the bus gives again: %+v", s)
}
if _, held := k.puts["ace.mjk3l9x0abcded"]; held {
t.Errorf("the ghost ask's key is still held: %+v", k.last("ace.mjk3l9x0abcded"))
}
sdk/buildstep/ask.go
+9
View File
@@ -77,6 +77,15 @@ const (
AskMaxDeliver = AskRuns + 2
)
// AgainAfter is how long the bus waits before it gives an ask again after its run-th run ended on a transport fault: 1
// minute after the first, 4 after the second (ADR 0664 §2). It outlasts a per-source penalty of a forge's sshd.
func AgainAfter(run int) time.Duration {
if run < 1 {
run = 1
}
return time.Minute << (2 * (run - 1))
}
// StepAskVersion is the layout version of an ask.
const StepAskVersion = 1
sdk/buildstep/ask_test.go
+40
View File
@@ -4,6 +4,7 @@ import (
"bytes"
"encoding/json"
"errors"
"fmt"
"os"
"path/filepath"
"strings"
@@ -313,6 +314,43 @@ func TestPauseCapacityAndComposerKey(t *testing.T) {
if ComposerKey("novox/r", testCommit, "d", "p", tc, nil) != ComposerKey("novox/r", testCommit, "d", "p", tc, PackageVersions{}) {
t.Error("an empty ask changed the key")
}
// Issue 1748 (ADR 0664): a transport fault is told from a wrong source by its words; a step given again is no end; the
// bus waits 1 minute after the first run and 4 after the second.
for words, want := range map[string]bool{
`Get "https://proxy": dial tcp: lookup proxy.golang.org: i/o timeout`: true,
`Head http://store.internal:5100/v2/x: dial tcp 10.10.0.1:5100: i/o timeout`: true,
"kex_exchange_identification: read: Connection reset by peer": true,
"Connection closed by 10.10.0.1 port 222": true,
"npm error code EAI_AGAIN": true,
"npm error code ETIMEDOUT": true,
"npm error code ECONNRESET": true,
"fatal: unable to access the forge: Could not resolve host: mesh-forge": true,
"Get \"https://example.com/x.tar.gz\": context deadline exceeded (Client.Timeout exceeded while awaiting headers)": true,
(&HolderSilent{Seat: "go-package-registry", Holder: "gopkgs", Node: "ace"}).Error(): true,
`invalid version: Get "https://proxy.golang.org/x/@v/y.info": dial tcp [2a00::1]:443: connect: network is unreachable`: true,
"dial tcp 10.10.0.1:22: connect: No route to host": true,
"nats: no responders available for request": false,
"fatal: unknown revision or path not in the working tree": false,
"verifying module: checksum mismatch": false,
"./main.go:3:1: syntax error: unexpected }": false,
} {
if PassingCause(words) != want {
t.Errorf("PassingCause(%q) = %v, want %v", words, !want, want)
}
}
if !IsPassing(fmt.Errorf("fetch: %w", &HolderSilent{Seat: "s"})) || IsPassing(errors.New("i/o timeout")) {
t.Error("a HolderSilent is a Passing error and a bare error is none")
}
if Ended(Step{State: StepRunning}) || Ended(Step{State: StepAgain}) || !Ended(Step{State: StepPassed}) || !Ended(Step{State: StepFailed}) {
t.Error("Ended: running and again are no end")
}
// Review 1: a step ended by the seat's verb drop is the state dropped, an end that decodes (ADR 0664 §4).
if raw, err := EncodeStep(Step{ID: "mvl0frv18d1990", Tool: "go ask", Node: "ace", State: StepDropped, Reason: "withdrawn"}); err != nil || !Ended(Step{State: StepDropped}) {
t.Errorf("a dropped step: %v (%s)", err, raw)
}
if AgainAfter(1) != time.Minute || AgainAfter(2) != 4*time.Minute {
t.Errorf("AgainAfter = %v, %v", AgainAfter(1), AgainAfter(2))
}
}
// The part `build-step` is one layout for the three build seats: each seat's fixtures read strictly in the neutral
@@ -350,6 +388,8 @@ func TestTheThreeSeatsFixturesFitTheNeutralForms(t *testing.T) {
case strings.HasPrefix(f.Of, "state:pauses/paused."):
seen++
strict(t, f.Given, &Pause{})
case strings.HasPrefix(f.Of, "state:pauses/dropped."):
seen++ // its layout is the generated contracts' Drop, which seatgen checks the fixture against
case strings.HasPrefix(f.Of, "state:pauses/capacity."):
seen++
strict(t, f.Given, &Capacity{})
sdk/buildstep/passing.go
+58
View File
@@ -0,0 +1,58 @@
package buildstep
import (
"errors"
"fmt"
"strings"
)
// Passing is a fault of the network between a build node and what it reaches (a name lookup, a timeout, a reset
// connection, a forge's sshd refusing a burst, a store that does not answer), not of the source or the toolchain. A
// build module wraps it, and the shared runner then writes the step again and settles its ask with a delayed nak, up to
// AskRuns runs (ADR 0664 §1, §2). Any other error ends the step failed at once.
type Passing struct{ Err error }
func (p *Passing) Error() string { return p.Err.Error() }
func (p *Passing) Unwrap() error { return p.Err }
// IsPassing says err is, or wraps, a Passing.
func IsPassing(err error) bool {
var p *Passing
return errors.As(err, &p)
}
// HolderSilentWords opens the words of a HolderSilent, so PassingCause reads them from a step's reason too.
const HolderSilentWords = "the holder is silent"
// HolderSilent is a seat verb that answered "no responders" while the assigner names a holder of the seat: its tool
// runner is being replaced, a moment later it answers. It is a Passing fault (ADR 0664 §1); a seat nothing holds is not.
type HolderSilent struct{ Seat, Holder, Node string }
func (h *HolderSilent) Error() string {
return fmt.Sprintf("%s: seat %s, held by %s on %s, answered no responders", HolderSilentWords, h.Seat, h.Holder, h.Node)
}
// Unwrap makes a HolderSilent a Passing.
func (h *HolderSilent) Unwrap() error { return &Passing{Err: errors.New(h.Error())} }
// passingWords are the words of a network that did not answer, lower-cased: go's and git's (a lookup, a timeout, a refused or
// reset connection, ssh's key exchange cut by a forge's per-source penalty) and npm's error codes. "no responders" alone is not
// one: it is the same for a seat nothing holds (HolderSilent is the case a holder is named).
var passingWords = []string{"i/o timeout", "no such host", "temporary failure in name resolution", "connection refused",
"connection reset", "tls handshake timeout", "network is unreachable", "no route to host", "lookup ", "kex_exchange_identification", "connection closed by",
"could not resolve host", "client.timeout", "operation timed out", "connection timed out", "eai_again", "etimedout", "econnreset", "econnrefused", "enotfound", strings.ToLower(HolderSilentWords)}
// PassingCause says words (a tool's output or a step's reason) are those of a transport fault.
func PassingCause(words string) bool {
words = strings.ToLower(words)
for _, w := range passingWords {
if strings.Contains(words, w) {
return true
}
}
return false
}
// TransportFaultWords open the reason of a step that ended failed after its runs met a transport fault (the asker says them,
// "a transport fault on 3 runs (nodes): cause"): a reader tells that end from a fault of the source by them (ADR 0664 §3).
const TransportFaultWords = "a transport fault on "
sdk/buildstep/pause.go
+6
View File
@@ -32,6 +32,12 @@ func CapacityKey(node string) (string, error) {
return "capacity." + node, nil
}
// DropKey is the key of an ask's drop in a build seat's pauses state: "dropped.<id>" (the part build-step).
func DropKey(askID string) string { return "dropped." + askID }
// DroppedPrefix opens every key DropKey makes.
const DroppedPrefix = "dropped."
// Pause is whether a node takes no new ask, and who said so.
type Pause struct {
V int `json:"v"`
sdk/buildstep/steps.go
+18 -2
View File
@@ -23,9 +23,15 @@ var ErrUnknownVersion = errors.New("buildstep: unknown value version")
// The states of a step.
const (
StepRunning = "running"
// StepAgain is a step that ended on a transport fault with runs left: the bus gives its ask again after AgainAt
// (ADR 0664 §1). It is not an end (Ended), and nothing acknowledges the ask.
StepAgain = "again"
StepPassed = "passed"
StepFailed = "failed"
StepRefused = "refused"
// StepDropped is a step ended by the seat's verb drop: its asker no longer waited for it (ADR 0664 §4). The cause its
// caller named is its reason. It is an end, and no failure of the source.
StepDropped = "dropped"
)
// StepBound is how long an asker waits for one ask's step once a build node took it (30 minutes): past it the asker
@@ -47,8 +53,18 @@ type Step struct {
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).
Run int `json:"run,omitempty"`
AgainAt time.Time `json:"again-at,omitzero"`
Passing bool `json:"passing,omitempty"`
}
// Ended says a step's state is an end: passed, failed, refused or dropped. A step running, or ended on a transport fault with runs
// left (StepAgain), is not.
func Ended(s Step) bool { return s.State != StepRunning && s.State != StepAgain && s.State != "" }
var (
nameOK = regexp.MustCompile(`^[-/_=a-zA-Z0-9]+$`)
stepIDOK = regexp.MustCompile(`^[a-z0-9]+$`)
@@ -106,9 +122,9 @@ func (s Step) check() error {
return fmt.Errorf("buildstep: a step needs its tool")
}
switch s.State {
case StepRunning, StepPassed, StepFailed, StepRefused:
case StepRunning, StepAgain, StepPassed, StepFailed, StepRefused, StepDropped:
default:
return fmt.Errorf("buildstep: the step state %q is not running, passed, failed or refused", s.State)
return fmt.Errorf("buildstep: the step state %q is not running, again, passed, failed, refused or dropped", s.State)
}
if len(s.Result) > 0 && !json.Valid(s.Result) {
return fmt.Errorf("buildstep: the step's result is not JSON")