The build modules end a transport fault's run again, serve drop, and are held to the transport-fault classes (issue 1748; group fix/1748-transport-retry, merge last) #649

Merged
mesh-tools merged 13 commits from fix/1748-transport-retry into main 2026-11-09 22:58:44 +00:00
77 changed files with 1022 additions and 142 deletions
+26 -2
View File
@@ -4,7 +4,8 @@
// passing line in its place). A build module with no test step (mesh-build-image builds images and runs no tests) is
// exempt from that, and said so. Every build module, test step or not, must also hold a test that calls
// conformance.LeftRunning with its own start sweep, so a step a previous process left running is ended at its start
// (issue 1890).
// (issue 1890), and one that calls conformance.Passing, so a step whose network did not answer is said as a transport fault
// (issue 1748). Its manifest asks the capability seat-settle-delay (ADR 0664 §5).
package failurekit
import (
@@ -13,6 +14,7 @@ import (
"io/fs"
"os"
"path/filepath"
"slices"
"sort"
"strings"
)
@@ -27,9 +29,15 @@ type manifest struct {
Claims []struct {
Name string `json:"name"`
} `json:"claims"`
Tools []json.RawMessage `json:"tools"`
Tools []json.RawMessage `json:"tools"`
Capabilities []string `json:"capabilities"`
}
// settleDelay is the capability a tool runner states when it can settle a seat's ask with a delayed negative acknowledgement
// (mesh-sdk CapSeatSettleDelay): a build module gives its ask again after a delay by it, so the delivery plan places a tool
// runner that states it before the module (ADR 0664 §1, §5).
const settleDelay = "seat-settle-delay"
// Judge reads every modules/<m>/module.json of the catalogue at root and judges each module that claims a node-build-*
// seat.
func Judge(root string) ([]Judged, error) {
@@ -83,6 +91,22 @@ func Judge(root string) ([]Judged, error) {
refused = append(refused, fmt.Sprintf("it claims %s, and no test of it calls conformance.LeftRunning "+
"(mesh-build-go buildstep/conformance): run the kit's left-running case with the module's own start sweep (issue 1890)", seat))
}
// Every build module meets the network (a fetch, a download, an install, a publish), so every one runs the passing case: its
// steps say a network that did not answer as a transport fault the bus gives the ask again for (ADR 0664, issue 1748).
says, err := runsTheKit(dir, "conformance.Passing(")
if err != nil {
return nil, err
}
if !says {
refused = append(refused, fmt.Sprintf("it claims %s, and no test of it calls conformance.Passing "+
"(mesh-build-go buildstep/conformance): run the kit's passing case with the module's own path from a failed run to its step's error (issue 1748)", seat))
}
// Every build module gives a step that met a transport fault again by the bus (ADR 0664 §5): it asks the tool runner's
// capability in its manifest, so the delivery plan places a tool runner that states it first.
if !slices.Contains(m.Capabilities, settleDelay) {
refused = append(refused, fmt.Sprintf("it claims %s, and its manifest's capabilities do not ask %s: a step given again by the bus needs "+
"a tool runner that can settle with a delay (ADR 0664 §5, issue 1748)", seat, settleDelay))
}
j.Refused = strings.Join(refused, "; ")
out = append(out, j)
}
+11 -6
View File
@@ -8,23 +8,28 @@ import (
// 1901/AC7 and 1890/AC8: a build module with a test step and no kit is refused, one that runs the kit passes, one with no
// test step is exempt from the failure cases but not from the left-running case, a module that runs only the failure
// cases is refused, and a module that claims no node-build-* seat is not judged.
// cases is refused, a module that does not ask the capability seat-settle-delay is refused (ADR 0664 §5), and a module that
// claims no node-build-* seat is not judged.
func TestABuildModuleWithATestStepMustRunTheKit(t *testing.T) {
root := t.TempDir()
const imp = "package main\n\nimport \"git.novox.be/novox/mesh-catalog/modules/mesh-build-go/buildstep/conformance\"\n\n"
const sweep = "func TestSweep(t *testing.T) { conformance.LeftRunning(t, nil) }\n"
const sweep = "func TestSweep(t *testing.T) { conformance.LeftRunning(t, nil) }\nfunc TestPassing(t *testing.T) { conformance.Passing(t, nil, nil) }\n"
kit := imp + "func TestKit(t *testing.T) { conformance.Run(t, conformance.GoTest, nil) }\n" + sweep
for path, body := range map[string]string{
"modules/a/module.json": `{"claims":[{"name":"node-build-a"}],"tools":["build_a_fetch","build_a_test"]}`,
"modules/a/cmd/a/check_test.go": "package main\n",
"modules/b/module.json": `{"claims":[{"name":"node-build-b"}],"tools":[{"name":"build_b_test"}]}`,
"modules/b/module.json": `{"claims":[{"name":"node-build-b"}],"capabilities":["seat-settle-delay"],"tools":[{"name":"build_b_test"}]}`,
"modules/b/cmd/b/kit_test.go": kit,
"modules/c/module.json": `{"claims":[{"name":"node-build-c"}],"tools":["build_c_build"]}`,
"modules/c/module.json": `{"claims":[{"name":"node-build-c"}],"capabilities":["seat-settle-delay"],"tools":["build_c_build"]}`,
"modules/c/cmd/c/kit_test.go": imp + sweep,
"modules/f/module.json": `{"claims":[{"name":"node-build-f"}],"tools":["build_f_build"]}`,
"modules/g/module.json": `{"claims":[{"name":"node-build-g"}],"tools":["build_g_test"]}`,
"modules/g/cmd/g/kit_test.go": imp + "func TestKit(t *testing.T) { conformance.Run(t, conformance.GoTest, nil) }\n",
"modules/d/module.json": `{"claims":[{"name":"other"}],"tools":["d_test"]}`,
"modules/h/module.json": `{"claims":[{"name":"node-build-h"}],"tools":["build_h_build"]}`, // the sweep, and no passing case
"modules/h/cmd/h/kit_test.go": imp + "func TestSweep(t *testing.T) { conformance.LeftRunning(t, nil) }\n",
"modules/i/module.json": `{"claims":[{"name":"node-build-i"}],"capabilities":["other"],"tools":["build_i_build"]}`, // the whole kit, and no seat-settle-delay
"modules/i/cmd/i/kit_test.go": imp + sweep,
"modules/e/module.json": `{"claims":[{"name":"node-build-e"}],"tools":["build_e_test"]}`,
"modules/e/cmd/e/kit.go": kit, // not a test file
} {
@@ -40,9 +45,9 @@ func TestABuildModuleWithATestStepMustRunTheKit(t *testing.T) {
if err != nil {
t.Fatal(err)
}
want := map[string]string{"a": "refused", "b": "passes", "c": "exempt", "e": "refused", "f": "refused", "g": "refused"}
want := map[string]string{"a": "refused", "b": "passes", "c": "exempt", "e": "refused", "f": "refused", "g": "refused", "h": "refused", "i": "refused"}
if len(got) != len(want) {
t.Fatalf("judged %+v, want a, b, c, e, f and g", got)
t.Fatalf("judged %+v, want a, b, c, e, f, g, h and i", got)
}
for _, j := range got {
is := "passes"
@@ -71,3 +71,11 @@ the command is mesh-build-node's type-check line without a tsconfig.json:
node <typescript>/bin/tsc --noEmit --module NodeNext --moduleResolution NodeNext --target ES2022 --rootDir . ok.ts seat.ts
and the file is its stdout followed by its stderr.
## passing/ — a runner's words when the network did not answer (novox/hq issue 1748, ADR 0664 §1)
Not captured by running: a network cannot be cut on demand in a test. Each file holds the words a runner wrote when a
build step met the fault on the mesh (the journal of ace on 2026-11-09 for ssh and go, the forge's sshd penalty; npm's
error codes and buildah's pull failure as those programs print them), and for each a negative case, a failure of the
source whose words are never a network's (an unknown revision, a checksum, a missing package, a failed RUN). Name:
`<runner>-<case>.txt`; the kit's table (conformance.go, PassingCases) says which are transport faults.
@@ -0,0 +1 @@
Error: building at STEP "RUN false": exit status 1
@@ -0,0 +1 @@
Error: creating build container: initializing source docker://alpine: pinging container registry registry-1.docker.io: Get "https://registry-1.docker.io/v2/": dial tcp: lookup registry-1.docker.io on 10.10.0.3:53: i/o timeout
@@ -0,0 +1,2 @@
Connection closed by 10.10.0.1 port 222
fatal: Could not read from remote repository.
@@ -0,0 +1,6 @@
kex_exchange_identification: read: Connection reset by peer
Connection reset by 10.10.0.1 port 222
fatal: Could not read from remote repository.
Please make sure you have the correct access rights
and the repository exists.
@@ -0,0 +1 @@
fatal: couldn't find remote ref 0123456789abcdef0123456789abcdef01234567
@@ -0,0 +1,2 @@
ssh: Could not resolve hostname mesh-forge: Temporary failure in name resolution
fatal: Could not read from remote repository.
@@ -0,0 +1 @@
{"Path":"git.novox.be/novox/mesh-sdk/go","Error":"verifying module: checksum mismatch"}
@@ -0,0 +1 @@
{"Path":"git.novox.be/novox/mesh-sdk/go","Error":"git.novox.be/novox/mesh-sdk/go@abc: dial tcp 10.10.0.1:5100: i/o timeout"}
@@ -0,0 +1 @@
{"Path":"git.novox.be/novox/mesh-sdk/go","Error":"git.novox.be/novox/mesh-sdk/go@abc: dial tcp: lookup git.novox.be on 10.10.0.3:53: read udp 10.10.0.3:42292->10.10.0.3:53: i/o timeout"}
@@ -0,0 +1 @@
{"Path":"git.novox.be/novox/mesh-sdk/go","Error":"invalid version: unknown revision abc"}
@@ -0,0 +1,2 @@
npm error code E404
npm error 404 Not Found - GET https://git.novox.be/api/packages/novox/npm/@novox%2fx
@@ -0,0 +1,3 @@
npm error code EAI_AGAIN
npm error syscall getaddrinfo
npm error errno EAI_AGAIN
@@ -0,0 +1,4 @@
npm error code ECONNRESET
npm error syscall read
npm error errno -104
npm error network aborted
@@ -12,6 +12,7 @@ package conformance
import (
"embed"
"fmt"
"slices"
"strings"
"testing"
@@ -199,3 +200,80 @@ func describe(failed []buildstep.Failure) string {
}
func quote(s string) string { return fmt.Sprintf("%q", s) }
// The runners whose network failures the kit holds words of (Passing).
const (
GitFetch = "git fetch"
GoDownload = "go mod download"
NpmCI = "npm ci"
BuildahBud = "buildah bud"
)
// PassingCase is the words of one runner and whether they are a transport fault: a network that did not answer, which the
// bus gives the ask again for (novox/hq ADR 0664 §1), or a failure of the source, which fails the step at once.
type PassingCase struct {
Name, Runner string
Transport bool
}
// PassingCases are the kit's cases of the network's failures (captured/passing).
var PassingCases = []PassingCase{
{"git-fetch-kex", GitFetch, true}, {"git-fetch-closed", GitFetch, true}, {"git-fetch-resolve", GitFetch, true}, {"git-fetch-noref", GitFetch, false},
{"go-download-lookup", GoDownload, true}, {"go-download-dial", GoDownload, true}, {"go-download-unknown", GoDownload, false}, {"go-download-checksum", GoDownload, false},
{"npm-ci-econnreset", NpmCI, true}, {"npm-ci-eaiagain", NpmCI, true}, {"npm-ci-404", NpmCI, false},
{"buildah-bud-timeout", BuildahBud, true}, {"buildah-bud-run", BuildahBud, false},
}
// Failed is a build module's own path from a runner's words to the error its step ends with: the same code its step runs on
// a failed command, fed the words of a runner's failed run.
type Failed func(runner, words string) error
// Passing runs every case of the runners through failed, each as a subtest, and fails a build module whose step does not
// say a network's failure as a buildstep.Passing (the bus then gives the ask again) or says a failure of the source as one.
// A build module names the runners its steps use; the case's words are in captured/passing.
func Passing(t *testing.T, runners []string, failed Failed) {
t.Helper()
ran := 0
for _, c := range PassingCases {
if !slices.Contains(runners, c.Runner) {
continue
}
ran++
t.Run(c.Name, func(t *testing.T) {
if problem := judgePassing(c, failed); problem != "" {
t.Error(problem)
}
})
}
if ran == 0 {
t.Fatalf("the kit holds no passing case of the runners %v", runners)
}
}
// JudgePassing is the problems of failed over every case of the runners: what Passing fails a build module for.
func JudgePassing(runners []string, failed Failed) []string {
var out []string
for _, c := range PassingCases {
if slices.Contains(runners, c.Runner) {
if p := judgePassing(c, failed); p != "" {
out = append(out, p)
}
}
}
return out
}
func judgePassing(c PassingCase, failed Failed) string {
words, err := captured.ReadFile("captured/passing/" + c.Name + ".txt")
if err != nil {
return err.Error()
}
got := failed(c.Runner, string(words))
switch {
case got == nil:
return c.Name + ": the step passed on a failed run"
case buildstep.IsPassing(got) != c.Transport:
return fmt.Sprintf("%s: the step ends %q, a transport fault %v, want %v", c.Name, got, buildstep.IsPassing(got), c.Transport)
}
return ""
}
@@ -3,6 +3,7 @@ package conformance
import (
"context"
"encoding/json"
"errors"
"strings"
"testing"
"time"
@@ -23,6 +24,19 @@ func TestTheFailureContract(t *testing.T) {
})
})
// ADR 0664 §1 (issue 1748): the shared wrapper keeps the passing cases; a step that never says a network's failure as a
// transport fault, or says every failure as one, is refused.
t.Run("the passing cases", func(t *testing.T) {
all := []string{GitFetch, GoDownload, NpmCI, BuildahBud}
Passing(t, all, func(_, words string) error { return buildstep.Transport(errors.New(words)) })
if p := JudgePassing(all, func(_, words string) error { return errors.New(words) }); len(p) != 8 {
t.Errorf("a step that wraps nothing: %d problems, want 8 (one per transport case):\n%s", len(p), strings.Join(p, "\n"))
}
if p := JudgePassing(all, func(_, words string) error { return &sdkstep.Passing{Err: errors.New(words)} }); len(p) != 5 {
t.Errorf("a step that wraps everything: %d problems, want 5 (one per source failure):\n%s", len(p), strings.Join(p, "\n"))
}
})
// 1890/AC7: the SDK's start sweep keeps the left-running case, and the kit refuses a sweep that ends nothing or ends
// everything.
t.Run("the left-running case", func(t *testing.T) {
+13 -2
View File
@@ -9,6 +9,7 @@ import (
"regexp"
"strings"
sdkstep "git.novox.be/novox/mesh-delivery/sdk/buildstep"
dforgekey "git.novox.be/novox/mesh-delivery/sdk/forgekey"
"git.novox.be/novox/mesh-sdk/go/forgekey"
)
@@ -52,6 +53,16 @@ var (
refOK = regexp.MustCompile(`^refs/[A-Za-z0-9._/-]+$`)
)
// gitFailure is a failed git command said in git's own words; one whose words are a network's (a lookup, a timeout, ssh's key
// exchange cut by a forge's per-source penalty) is a sdkstep.Passing, so the ask is given again after a delay (ADR 0664 §1).
func gitFailure(verb, said string) error {
err := fmt.Errorf("git %s: %s", verb, said)
if sdkstep.PassingCause(said) {
return &sdkstep.Passing{Err: err}
}
return err
}
// FetchCommit fetches one commit of a forge repository into dir, an empty directory, and checks it out detached
// (novox/hq ADR 0466 §4, §5; issue 1474). remote is ssh://<ssh name>/<owner>/<repo>.git, the ssh name being the
// one the build module's templates set up (ADR 0409 §1); commit is a full 40-hex sha; ref is optional, and names
@@ -72,7 +83,7 @@ func FetchCommit(ctx context.Context, run forgekey.Runner, dir, remote, commit,
full := append([]string{"-c", "credential.helper=", "-C", dir}, args...)
r := run(ctx, env, "git", full...)
if r.Status != 0 || r.Err != "" {
return "", fmt.Errorf("git %s: %s", args[0], dforgekey.Said(r.Err+"\n"+r.Stderr))
return "", gitFailure(args[0], dforgekey.Said(r.Err+"\n"+r.Stderr))
}
return r.Stdout, nil
}
@@ -133,7 +144,7 @@ func UpdateMirror(ctx context.Context, run forgekey.Runner, mirror, remote, comm
if verb == "-C" && len(args) > 2 { // the mirror's commands are named by what they do, not by -C
verb = args[2]
}
return fmt.Errorf("git %s: %s", verb, dforgekey.Said(r.Err+"\n"+r.Stderr))
return gitFailure(verb, dforgekey.Said(r.Err+"\n"+r.Stderr))
}
return nil
}
@@ -2,6 +2,7 @@ package buildstep
import (
"context"
sdkstep "git.novox.be/novox/mesh-delivery/sdk/buildstep"
"strings"
"testing"
@@ -93,6 +94,28 @@ func TestFetchCommitFallsBackToThePullRef(t *testing.T) {
if err := FetchCommit(context.Background(), h.run, "/w", "ssh://mesh-forge/o/r.git", sha, "refs/pull/1/head"); err == nil {
t.Error("a commit that is not in the ref was accepted")
}
// ADR 0664 §1 (issue 1748): the words a forge's sshd penalty leaves (captured on ace, 2026-11-09) are a transport fault the bus
// gives the ask again for; a missing commit is not.
for _, c := range []struct {
said string
passing bool
}{
{"kex_exchange_identification: read: Connection reset by peer\nfatal: Could not read from remote repository.", true},
{"Connection closed by 10.10.0.1 port 222", true},
{"ssh: Could not resolve hostname mesh-forge: Temporary failure in name resolution", true},
{"fatal: couldn't find remote ref " + sha, false},
} {
x := &fakeGit{rule: func(l string) forgekey.Ran {
if strings.Contains(l, " fetch ") {
return forgekey.Ran{Status: 128, Stderr: c.said}
}
return forgekey.Ran{}
}}
if err := FetchCommit(context.Background(), x.run, "/w", "ssh://mesh-forge/o/r.git", sha, ""); err == nil || sdkstep.IsPassing(err) != c.passing {
t.Errorf("%q: %v, passing %v, want %v", c.said, err, err != nil && sdkstep.IsPassing(err), c.passing)
}
}
}
// 1474/AC3: HEAD differs from the commit after the checkout: an error naming both.
+1
View File
@@ -3,6 +3,7 @@ module git.novox.be/novox/mesh-catalog/modules/mesh-build-go/buildstep
go 1.26.0
require (
git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/contracts/go/clients/containerimageregistryclient v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/contracts/go/clients/gopackageregistryclient v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/contracts/go/clients/npmpackageregistryclient v0.0.0-00010101000000-000000000000
+1 -1
View File
@@ -56,7 +56,7 @@ func MergeInto(ctx context.Context, run forgekey.Runner, dir, remote, base, comm
full := append([]string{"-c", "credential.helper=", "-C", dir}, args...)
r := run(ctx, env, "git", full...)
if r.Status != 0 || r.Err != "" {
return "", fmt.Errorf("git %s: %s", args[0], dforgekey.Said(r.Err+"\n"+r.Stderr))
return "", gitFailure(args[0], dforgekey.Said(r.Err+"\n"+r.Stderr))
}
return r.Stdout, nil
}
@@ -15,6 +15,7 @@ import (
"os"
"strings"
assigner "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner"
imageclient "git.novox.be/novox/mesh-delivery/contracts/go/clients/containerimageregistryclient"
goclient "git.novox.be/novox/mesh-delivery/contracts/go/clients/gopackageregistryclient"
npmclient "git.novox.be/novox/mesh-delivery/contracts/go/clients/npmpackageregistryclient"
@@ -23,6 +24,7 @@ import (
goseat "git.novox.be/novox/mesh-delivery/contracts/go/seats/gopackageregistry"
npmseat "git.novox.be/novox/mesh-delivery/contracts/go/seats/npmpackageregistry"
ociseat "git.novox.be/novox/mesh-delivery/contracts/go/seats/ociartifactregistry"
sdkstep "git.novox.be/novox/mesh-delivery/sdk/buildstep"
"git.novox.be/novox/mesh-delivery/sdk/parts"
)
@@ -76,6 +78,41 @@ func SentHow(n int) func(parts.Sent) string {
// PublishParts publishes in with ps as its parts on seat, through that seat's client: in's fields are the seat's own
// (the four registry seats share the publish verb's input, design 51), the parts are written by the client.
func PublishParts(ctx context.Context, call SeatCall, seat string, in reg.PublishInput, ps []parts.Part) (parts.Sent, error) {
sent, err := publishParts(ctx, call, seat, in, ps)
return sent, HolderSilentOr(call, seat, err)
}
// AssignerHolders is the verb that tells who holds a seat: `mesh-module-assigner.assignments` with seats
// (the state seat-holders). The build modules' manifests `invokes` it.
const AssignerHolders = "seat:" + assigner.Seat + "." + assigner.VerbAssignments
// HolderSilentOr is err, or when it is a seat verb's "no responders" and the assigner names a holder of seat, a
// sdkstep.HolderSilent (a Passing: the holder's tool runner is being replaced and answers a moment later, ADR 0664 §1). The
// bus says "no responders" for a seat nothing holds too, so the words alone are no transport fault: with no holder named, or
// when the holders cannot be asked, the error stays as it is, said with why, and fails at once.
func HolderSilentOr(call SeatCall, seat string, err error) error {
if err == nil || !strings.Contains(strings.ToLower(err.Error()), "no responders") {
return err
}
raw, aerr := call(AssignerHolders, map[string]any{"seats": assigner.AssignmentsInputSeatsTrue})
if aerr != nil {
return fmt.Errorf("%w (who holds %s could not be asked: %v)", err, seat, aerr)
}
var answer struct {
Seats []assigner.SeatHolders `json:"seats"`
}
if jerr := json.Unmarshal(raw, &answer); jerr != nil {
return fmt.Errorf("%w (who holds %s was answered in what is not read: %v)", err, seat, jerr)
}
for _, sh := range answer.Seats {
if sh.Seat == seat && len(sh.Holders) > 0 {
return fmt.Errorf("%w: %w", err, &sdkstep.HolderSilent{Seat: seat, Holder: sh.Holders[0].Module, Node: sh.Holders[0].Node})
}
}
return fmt.Errorf("%w (nothing holds %s)", err, seat)
}
func publishParts(ctx context.Context, call SeatCall, seat string, in reg.PublishInput, ps []parts.Part) (parts.Sent, error) {
switch seat {
case ImageSeat:
s, err := imageclient.Publish(ctx, imageclient.Call(call), in, ps)
+52 -3
View File
@@ -32,6 +32,9 @@ type Pulled struct {
// Ack ends it: it is never delivered again. InProgress tells the bus it is still being worked, restarting its
// acknowledgement wait. Nak hands it back to be delivered again at once (to this node or another). Term drops it.
Ack, InProgress, Nak, Term func() error
// NakAfter hands it back to be delivered again after delay, not before (a step that ended on a transport fault, ADR 0664
// §1); nil when the source cannot, and a refusal (a tool runner that does not state the capability) is an error.
NakAfter func(delay time.Duration) error
}
// AskSource is where a puller takes asks from: the seat's workers on the bus (JetStreamAsks) or a test double
@@ -64,7 +67,10 @@ type Puller struct {
// Run runs one ask's job; a *Refusal ends its step refused. Its result is the step's result.
Run func(ctx context.Context, ask sdkstep.StepAsk) (any, error)
// PullWait and Heartbeat default to DefaultPullWait and DefaultHeartbeat.
PullWait time.Duration
PullWait time.Duration
// Delay is how long the bus waits before it gives an ask again after its run-th run ended on a transport fault; nil is
// sdkstep.AgainAfter (a minute, then four). A test's own is short.
Delay func(run int) time.Duration
Heartbeat time.Duration
Logf func(format string, args ...any)
// Load reads one key of the state the node's pause is kept in (nil and no error when it is absent). When set, Pull
@@ -74,6 +80,14 @@ type Puller struct {
// PausePut writes the node's pause key: the module's pauses state (sdkstep.PausesState), which has no age, so a
// pause never ends by itself while the steps state drops its old keys. Put when nil.
PausePut Put
// ForgetPause deletes one key of the pauses state and PauseKeys lists its keys (nil for none): a dropped ask's key is
// deleted when the ask is refused at pull, and the start sweeps the keys of asks the stream no longer holds (Drop).
ForgetPause func(key string) error
PauseKeys func() ([]string, error)
// WatchPauses follows the keys of the pauses state that key names (`**` for the rest): stdio.State(...).Watch in a module. The
// state is the module's, shared by every node that holds the seat, so a drop another node's holder wrote reaches the node
// that runs the ask (Drop).
WatchPauses func(key string, handle func(stdio.StateChange) error) error
// WatchConfig, when set, follows the module's config value `capacity` (CapacitySetting) as mesh-config resolves it
// for this node, and wakes the puller on each change (watchCapacity): WatchConfigServed in a module, a test's own
// double. Until it hands a value the puller judges by Capacity.
@@ -109,6 +123,8 @@ type Puller struct {
mu sync.Mutex
inHand int
running map[string]RunningAsk
// cancels end a running ask's step by its id: the seat's verb drop (Drop).
cancels map[string]context.CancelCauseFunc
paused bool
wake chan struct{}
once sync.Once
@@ -425,6 +441,8 @@ func (p *Puller) Pull(ctx context.Context) error {
// node kept from building is held by its capacity (0), not by a pause.
p.logf("the pause of %s cannot be read: %v; pulling as not paused", p.Node, err)
}
p.sweepDropped()
p.watchDrops()
p.follow() // WatchConfigServed waits for Serve, which a module starts after its pull
go p.watchSettings(ctx)
wait := p.PullWait
@@ -612,11 +630,15 @@ func (p *Puller) take(ctx context.Context, msg *Pulled) {
p.refuseUnread(msg, err)
return
}
if d := p.droppedAsk(ask.ID); d != nil {
p.refuseDropped(msg, ask, d)
return
}
if lacks := sdkstep.Lacks(ask.Needs, p.Has); len(lacks) > 0 {
p.decline(msg, ask, lacks)
return
}
step, key, err := begin(p.Put, p.Node, ask.Kind+" ask", ask.ID, buildrun.AskWhat(askRead(ask)))
step, key, err := beginRun(p.Put, p.Node, ask.Kind+" ask", ask.ID, buildrun.AskWhat(askRead(ask)), int(msg.Delivered))
if err != nil {
p.logf("the ask %s cannot start on %s: %v", ask.ID, p.Node, err)
_ = msg.Nak()
@@ -674,11 +696,20 @@ func (p *Puller) run(ctx context.Context, msg *Pulled, ask sdkstep.StepAsk, key
}
}()
ph := &phases{put: p.Put, key: key, step: step, logf: p.Logf, prefix: fmt.Sprintf("the %s ask %s on %s", ask.Kind, ask.ID, p.Node)}
result, err := protect(ctx, func(ctx context.Context) (any, error) {
runCtx := p.cancellable(ctx, ask.ID)
result, err := protect(runCtx, func(ctx context.Context) (any, error) {
return p.Run(context.WithValue(ctx, phaseKey{}, ph), ask)
})
step = ph.end() // its end keeps the phase it ended in
close(done)
if d := droppedCause(runCtx); d != nil && ctx.Err() == nil {
// The seat's verb drop ended it (ADR 0664 §4, §5): dropped, naming who dropped it and why; acknowledged, never given again.
err = &Dropped{Text: d.Error()}
defer p.forgetDrop(ask.ID)
}
p.mu.Lock()
delete(p.cancels, ask.ID)
p.mu.Unlock()
var dec *Declining
if errors.As(err, &dec) && ctx.Err() == nil {
p.declineRun(msg, ask, key, step, dec)
@@ -688,6 +719,24 @@ func (p *Puller) run(ctx context.Context, msg *Pulled, ask sdkstep.StepAsk, key
// The process stops: the ask is neither ended nor acknowledged here, so the stream gives it to a node that runs.
return
}
if run := int(msg.Delivered); err != nil && sdkstep.IsPassing(err) && run < sdkstep.AskRuns && msg.NakAfter != nil {
// A transport fault (ADR 0664 §1): the step says it is given again and the bus gives it after a delay, to this node or
// another; it is not acknowledged, so its asker sees no end. Run AskRuns times it ends failed, said as a transport fault.
delay := sdkstep.AgainAfter(run)
if p.Delay != nil {
delay = p.Delay(run)
}
if aerr := again(p.Put, key, step, err, delay); aerr == nil {
p.logf("the %s ask %s ended on %s on a transport fault in its run %d (%v), and is given again in %s", ask.Kind, ask.ID, p.Node, run, err, delay)
if nerr := msg.NakAfter(delay); nerr != nil {
p.logf("the ask %s could not be handed back with a delay (%v), so it is handed back at once", ask.ID, nerr)
_ = msg.Nak()
}
return
} else {
p.logf("the ask %s ended on a transport fault and its step could not be written again (%v): it ends failed", ask.ID, aerr)
}
}
if ferr := finish(p.Put, key, step, result, err, p.Logf); ferr != nil {
// No end was written: the ask is not acknowledged as done, so the stream gives it again and a run that can
// write its end (or the poison rule, issue 1907) ends it where its asker sees it (issue 1930).
+2 -1
View File
@@ -70,7 +70,8 @@ func (s *JetStreamAsks) Next(ctx context.Context, kind string, wait time.Duratio
delivered = meta.NumDelivered
}
return &Pulled{Data: append([]byte(nil), msg.Data()...), Delivered: delivered,
Ack: msg.Ack, InProgress: msg.InProgress, Nak: msg.Nak, Term: msg.Term}, nil
Ack: msg.Ack, InProgress: msg.InProgress, Nak: msg.Nak, Term: msg.Term,
NakAfter: func(delay time.Duration) error { return msg.NakWithDelay(delay) }}, nil
case <-ctx.Done():
return nil, ctx.Err()
}
@@ -0,0 +1,183 @@
package buildstep
import (
"context"
"encoding/json"
"errors"
"fmt"
"strings"
"time"
nodebuildgo "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo"
buildrun "git.novox.be/novox/mesh-delivery/sdk/buildrun"
sdkstep "git.novox.be/novox/mesh-delivery/sdk/buildstep"
stdio "git.novox.be/novox/mesh-sdk/go"
)
// **The seat's verb drop ends an ask its caller no longer waits for** (novox/hq ADR 0664 §5, issue 1748; research 086
// choice 28's cancel). It writes the ask in this node's own pauses state under dropped.<ask id> (the part build-step), so a
// holder that restarts still refuses the ask when the bus gives it again, and ends the step if it runs here: the step's
// context is cancelled, its process ended, and its key is written dropped naming who dropped it and why. The state is the
// module's, shared by every node: any holder answers the verb, every node refuses the ask when it pulls it (an ask that
// waits leaves the stream at its next pull, with no work done), and the node that runs it hears the key by its watch
// (watchDrops) and ends the step.
// DroppedKeep is how long a dropped ask's key stays: the start deletes older ones. An ask the seat's stream still holds is
// younger than its stream's maximum age, which no contract this module reads states, so the bound is a generous week.
const DroppedKeep = 7 * 24 * time.Hour
// droppedError is why a step's context was cancelled: the seat's verb drop.
type droppedError struct{ by, why string }
func (d *droppedError) Error() string {
by := d.by
if by == "" {
by = "its asker"
}
return fmt.Sprintf("the ask was dropped by %s: %s", by, d.why)
}
// droppedCause is the drop that cancelled ctx, nil when it was none.
func droppedCause(ctx context.Context) *droppedError {
var d *droppedError
if errors.As(context.Cause(ctx), &d) {
return d
}
return nil
}
// cancellable is ctx with a cancel the verb drop reaches by the ask's id, until the step ends with ctx.
func (p *Puller) cancellable(ctx context.Context, id string) context.Context {
c, cancel := context.WithCancelCause(ctx)
p.mu.Lock()
if p.cancels == nil {
p.cancels = map[string]context.CancelCauseFunc{}
}
p.cancels[id] = cancel
p.mu.Unlock()
return c
}
// Drop is the seat's verb drop: it keeps the ask's drop in the pauses state and ends the step if it runs here.
func (p *Puller) Drop(id, by, why string) error {
if strings.TrimSpace(id) == "" || strings.TrimSpace(why) == "" {
return errors.New("buildstep: give the ask's id and why it is dropped")
}
at := now().UTC()
raw, err := json.Marshal(nodebuildgo.Drop{V: int64(sdkstep.Version), By: by, Why: why, At: &at})
if err != nil {
return err
}
put := p.PausePut
if put == nil {
put = p.Put
}
if _, err := put(sdkstep.DropKey(id), json.RawMessage(raw)); err != nil {
return fmt.Errorf("buildstep: writing the drop of %s: %w", id, err)
}
p.mu.Lock()
cancel := p.cancels[id]
p.mu.Unlock()
if cancel != nil {
cancel(&droppedError{by: by, why: why})
}
p.Wake()
return nil
}
// watchDrops follows the drops written in the pauses state, by any holder of the seat: one for an ask that runs here ends its
// step. An ask that waits is refused when it is pulled (take), on whichever node pulls it. A watch that cannot start is said;
// the pull then refuses a dropped ask but cannot end a running one.
func (p *Puller) watchDrops() {
if p.WatchPauses == nil {
p.logf("no watch of the drops in the pauses of %s is set: a dropped ask that runs here is not ended, only refused when it is pulled", p.Node)
return
}
err := p.WatchPauses(sdkstep.DroppedPrefix+"**", func(ch stdio.StateChange) error {
id, ok := strings.CutPrefix(ch.Key, sdkstep.DroppedPrefix)
if !ok || ch.Op != "put" {
return nil
}
var d nodebuildgo.Drop
if json.Unmarshal(ch.Value, &d) != nil {
return nil
}
p.mu.Lock()
cancel := p.cancels[id]
p.mu.Unlock()
if cancel != nil {
cancel(&droppedError{by: d.By, why: d.Why})
}
return nil
})
if err != nil {
p.logf("the drops in the pauses of %s cannot be followed: %v; a dropped ask that runs here is not ended", p.Node, err)
}
}
// droppedAsk is the drop kept for the ask, nil for none (or no state to read it from).
func (p *Puller) droppedAsk(id string) *nodebuildgo.Drop {
if p.Load == nil {
return nil
}
raw, err := p.Load(sdkstep.DropKey(id))
if err != nil || raw == nil {
return nil
}
var d nodebuildgo.Drop
if json.Unmarshal(raw, &d) != nil {
return nil
}
return &d
}
// refuseDropped ends a dropped ask the bus gave again: its step refused naming who dropped it, the message ended and the
// drop's key deleted.
func (p *Puller) refuseDropped(msg *Pulled, ask sdkstep.StepAsk, d *nodebuildgo.Drop) {
key, err := sdkstep.AskStepKey(p.Node, ask.ID)
if err == nil {
at := now().UTC()
step := sdkstep.Step{ID: ask.ID, Tool: ask.Kind + " ask", Node: p.Node, State: sdkstep.StepDropped, Started: at, Ended: at, Owner: buildrun.Owner,
Run: int(msg.Delivered), Reason: (&droppedError{by: d.By, why: d.Why}).Error()}
err = writeEnd(p.Put, key, step)
}
if err != nil {
p.logf("the dropped ask %s cannot be refused on %s, so it is given back: %v", ask.ID, p.Node, err)
_ = msg.Nak()
return
}
p.logf("the ask %s is refused on %s: it was dropped (%s)", ask.ID, p.Node, d.Why)
_ = msg.Term()
p.forgetDrop(ask.ID)
}
// forgetDrop deletes an ask's drop key, said when it cannot be.
func (p *Puller) forgetDrop(id string) {
if p.ForgetPause == nil {
return
}
if err := p.ForgetPause(sdkstep.DropKey(id)); err != nil {
p.logf("the drop of the ask %s cannot be deleted: %v", id, err)
}
}
// sweepDropped deletes, at the start, the drops older than DroppedKeep: their asks are gone from the stream.
func (p *Puller) sweepDropped() {
if p.PauseKeys == nil || p.ForgetPause == nil || p.Load == nil {
return
}
keys, err := p.PauseKeys()
if err != nil {
p.logf("the pauses of %s cannot be listed to sweep the drops: %v", p.Node, err)
return
}
for _, k := range keys {
id, ok := strings.CutPrefix(k, sdkstep.DroppedPrefix)
if !ok {
continue
}
if d := p.droppedAsk(id); d != nil && (d.At == nil || now().Sub(*d.At) > DroppedKeep) {
p.forgetDrop(id)
}
}
}
+11 -2
View File
@@ -16,7 +16,10 @@ type FakeAsks struct {
added chan struct{}
// Acked, Termed and Naked are the asks' ids or data by fate, InProgress how many heartbeats were given.
Acked, Termed, Naked []string
InProgress int
// Delayed are the asks handed back to be given again after a delay (NakAfter), and Delays that delay of each.
Delayed []string
Delays []time.Duration
InProgress int
// Delivered, when set, is the delivery count every ask handed over says (1 when 0): a test of a poisoned ask sets it.
Delivered uint64
}
@@ -108,6 +111,12 @@ func (f *FakeAsks) pulled(data []byte) *Pulled {
if delivered == 0 {
delivered = 1
}
return &Pulled{Data: data, Delivered: delivered, Ack: fate(&f.Acked), Nak: fate(&f.Naked), Term: fate(&f.Termed),
nakAfter := func(delay time.Duration) error {
f.mu.Lock()
f.Delayed, f.Delays = append(f.Delayed, id), append(f.Delays, delay)
f.mu.Unlock()
return nil
}
return &Pulled{Data: data, Delivered: delivered, NakAfter: nakAfter, Ack: fate(&f.Acked), Nak: fate(&f.Naked), Term: fate(&f.Termed),
InProgress: func() error { f.mu.Lock(); f.InProgress++; f.mu.Unlock(); return nil }}
}
@@ -4,6 +4,7 @@ package buildstep
import (
"context"
"errors"
"strings"
"sync"
"testing"
@@ -141,4 +142,57 @@ func TestPullRedeliversAKilledPullersAsk(t *testing.T) {
}
}
// ADR 0664 §1, §5 (issue 1748) on the real server: a run that ends on a transport fault is written again and the bus gives the
// ask again after the delay, not before, as the second run (its delivery count one higher), and that run's end is the ask's;
// a step that runs when its ask is dropped ends dropped, naming who dropped it, and the ask is not given again.
func TestAnAskThatMetTheNetworkIsGivenAgainAndADroppedOneIsEnded(t *testing.T) {
js := askBus(t, 30*time.Second)
a, doomed := testAsk(sdkstep.AskDelivery), testAsk(sdkstep.AskDelivery)
publishAsk(t, js, a)
st, kv := newSteps(), &pauseKV{}
const delay = 600 * time.Millisecond
var mu sync.Mutex
var runs []time.Time
hold := make(chan struct{})
p := &Puller{Source: &JetStreamAsks{JS: js, Language: sdkstep.AskGo}, Language: sdkstep.AskGo, Node: "ace", Put: st.put, PausePut: kv.put, Load: kv.load,
ForgetPause: kv.forget, PauseKeys: kv.keys, Capacity: func() int { return 1 }, PullWait: 100 * time.Millisecond,
Delay: func(int) time.Duration { return delay },
Run: func(ctx context.Context, got sdkstep.StepAsk) (any, error) {
if got.ID == doomed.ID {
close(hold)
<-ctx.Done()
return nil, ctx.Err()
}
mu.Lock()
runs = append(runs, time.Now())
n := len(runs)
mu.Unlock()
if n == 1 {
return nil, &sdkstep.Passing{Err: errors.New("lookup git.novox.be: i/o timeout")}
}
return nil, nil
}}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go p.Pull(ctx)
key, _ := sdkstep.AskStepKey("ace", a.ID)
waitFor(t, "the ask's second run ended", func() bool { s, ok := st.get(key); return ok && s.State == sdkstep.StepPassed })
mu.Lock()
gap := runs[1].Sub(runs[0])
mu.Unlock()
if s, _ := st.get(key); len(runs) != 2 || gap < delay || s.Run != 2 {
t.Errorf("runs %d, %s apart (delay %s), the step's run %d: want two runs, the second after the delay, as run 2", len(runs), gap, delay, s.Run)
}
publishAsk(t, js, doomed)
<-hold
if err := p.Drop(doomed.ID, "mesh-delivery-builder", "superseded"); err != nil {
t.Fatal(err)
}
dkey, _ := sdkstep.AskStepKey("ace", doomed.ID)
waitFor(t, "the dropped step dropped", func() bool { s, ok := st.get(dkey); return ok && s.State == sdkstep.StepDropped })
if s, _ := st.get(dkey); !contains(s.Reason, "dropped by mesh-delivery-builder: superseded") {
t.Errorf("the dropped step says %q", s.Reason)
}
}
func contains(s, sub string) bool { return strings.Contains(s, sub) }
+8 -1
View File
@@ -16,6 +16,8 @@ type SeatAsks struct {
// Pull and Settle are stdio.SeatPull and stdio.SeatSettle when nil; a test gives its own.
Pull func(worker string, wait time.Duration) (*stdio.Pulled, error)
Settle func(id, fate string) error
// SettleAfter is stdio.SeatSettleAfter when nil (ADR 0664 §1).
SettleAfter func(id string, delay time.Duration) error
}
// Next is AskSource's.
@@ -38,7 +40,12 @@ func (s SeatAsks) Next(ctx context.Context, kind string, wait time.Duration) (*P
if err != nil || p == nil {
return nil, err
}
after := s.SettleAfter
if after == nil {
after = stdio.SeatSettleAfter
}
fate := func(f string) func() error { return func() error { return settle(p.ID, f) } }
return &Pulled{Data: p.Body, Delivered: p.Delivered, Ack: fate(stdio.SettleAck), InProgress: fate(stdio.SettleProgress),
Nak: fate(stdio.SettleNak), Term: fate(stdio.SettleTerm)}, nil
Nak: fate(stdio.SettleNak), Term: fate(stdio.SettleTerm),
NakAfter: func(delay time.Duration) error { return after(p.ID, delay) }}, nil
}
@@ -2,9 +2,14 @@ package buildstep
import (
"context"
"encoding/json"
"fmt"
"strings"
"sync"
"testing"
"time"
nodebuildgo "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo"
sdkstep "git.novox.be/novox/mesh-delivery/sdk/buildstep"
stdio "git.novox.be/novox/mesh-sdk/go"
)
@@ -38,3 +43,106 @@ func TestSeatAsksPullsTheKindsWorkerAndSettlesByID(t *testing.T) {
t.Errorf("settled %v", settled)
}
}
// pauseKV is a pauses state in memory: its put, load, delete and keys.
type pauseKV struct {
mu sync.Mutex
m map[string][]byte
}
func (k *pauseKV) put(key string, v any) (uint64, error) {
raw, _ := v.(json.RawMessage)
k.mu.Lock()
defer k.mu.Unlock()
if k.m == nil {
k.m = map[string][]byte{}
}
k.m[key] = raw
return 1, nil
}
func (k *pauseKV) load(key string) ([]byte, error) {
k.mu.Lock()
defer k.mu.Unlock()
return k.m[key], nil
}
func (k *pauseKV) forget(key string) error {
k.mu.Lock()
defer k.mu.Unlock()
delete(k.m, key)
return nil
}
func (k *pauseKV) keys() ([]string, error) {
k.mu.Lock()
defer k.mu.Unlock()
var out []string
for key := range k.m {
out = append(out, key)
}
return out, nil
}
// ADR 0664 §5 (issue 1748): the seat's verb drop. An ask dropped before it is given is refused at pull, naming who dropped it
// and why, its key deleted, and never run; one that runs when it is dropped has its step's context cancelled and ends
// refused the same way, acknowledged; the start deletes drops older than DroppedKeep.
func TestADroppedAskIsRefusedAtPullAndARunningOneIsEnded(t *testing.T) {
for _, running := range []bool{false, true} {
src, st, kv := NewFakeAsks(), newSteps(), &pauseKV{}
a := testAsk(sdkstep.AskDelivery)
started := make(chan struct{})
ran := false
var heard func(stdio.StateChange) error
p := &Puller{Source: src, Language: sdkstep.AskGo, Node: "shanks", Put: st.put, PausePut: kv.put, Load: kv.load, ForgetPause: kv.forget, PauseKeys: kv.keys,
WatchPauses: func(_ string, h func(stdio.StateChange) error) error { heard = h; return nil },
Capacity: func() int { return 1 }, PullWait: 10 * time.Millisecond,
Run: func(ctx context.Context, _ sdkstep.StepAsk) (any, error) {
ran = true
close(started)
<-ctx.Done()
return nil, ctx.Err()
}}
if !running {
if err := p.Drop(a.ID, "mesh-delivery-builder", "superseded"); err != nil {
t.Fatal(err)
}
}
_ = src.PublishAsk(a)
ctx, cancel := context.WithCancel(context.Background())
go p.Pull(ctx)
if running {
<-started
// Another node's holder served the verb: its write reaches this node by the watch of the shared pauses state.
raw, _ := json.Marshal(nodebuildgo.Drop{V: int64(sdkstep.Version), By: "mesh-delivery-builder", Why: "superseded", At: ptr(time.Now())})
kv.put(sdkstep.DropKey(a.ID), json.RawMessage(raw))
if err := heard(stdio.StateChange{Key: sdkstep.DropKey(a.ID), Op: "put", Value: raw}); err != nil {
t.Fatal(err)
}
}
waitFor(t, "the ask ended", func() bool { ack, tm, _ := src.Fates(); return len(ack)+len(tm) == 1 })
cancel()
key, _ := sdkstep.AskStepKey("shanks", a.ID)
s, ok := st.get(key)
if !ok || s.State != sdkstep.StepDropped || !strings.Contains(s.Reason, "dropped by mesh-delivery-builder: superseded") || ran != running {
t.Errorf("running %v: step %+v (%v), ran %v", running, s, ok, ran)
}
waitFor(t, fmt.Sprint("the drop key deleted, running ", running), func() bool { kv.mu.Lock(); defer kv.mu.Unlock(); _, held := kv.m[sdkstep.DropKey(a.ID)]; return !held })
}
// The start deletes a drop older than DroppedKeep and keeps a newer one.
kv := &pauseKV{}
old, _ := json.Marshal(nodebuildgo.Drop{V: int64(sdkstep.Version), Why: "old", At: ptr(time.Now().Add(-DroppedKeep - time.Hour))})
fresh, _ := json.Marshal(nodebuildgo.Drop{V: int64(sdkstep.Version), Why: "fresh", At: ptr(time.Now())})
kv.put(sdkstep.DropKey("old"), json.RawMessage(old))
kv.put(sdkstep.DropKey("fresh"), json.RawMessage(fresh))
// Review 1: a module that sets no watch of the drops says so (a dropped ask running here is then not ended).
var said []string
(&Puller{Node: "shanks", Logf: func(f string, a ...any) { said = append(said, fmt.Sprintf(f, a...)) }}).watchDrops()
if len(said) != 1 || !strings.Contains(said[0], "no watch of the drops") {
t.Errorf("a puller with no watch said %q", said)
}
p := &Puller{Load: kv.load, ForgetPause: kv.forget, PauseKeys: kv.keys}
p.sweepDropped()
if keys, _ := kv.keys(); len(keys) != 1 || keys[0] != sdkstep.DropKey("fresh") {
t.Errorf("keys after the sweep = %v, want the fresh drop alone", keys)
}
}
func ptr[T any](v T) *T { return &v }
@@ -518,6 +518,49 @@ func TestAPoisonedAskIsSaidAndRunNoMore(t *testing.T) {
if acked, _, naked := src.Fates(); len(acked)+len(naked) != 0 {
t.Errorf("acked %v naked %v, want it termed alone", acked, naked)
}
// ADR 0664 §1 (issue 1748): a run that ends on a transport fault below AskRuns is written again, with its run and when the
// bus gives it, and handed back with that delay, not acknowledged; the AskRuns-th ends failed, passing; a fault of the
// source ends failed at once, in its first run.
for _, c := range []struct {
name string
run uint64
err error
state string
delay time.Duration
passing bool
ackedOnly bool
}{
{"a transport fault in run 1", 1, &sdkstep.Passing{Err: errors.New("lookup proxy: i/o timeout")}, sdkstep.StepAgain, time.Minute, true, false},
{"a transport fault in run 2", 2, &sdkstep.Passing{Err: errors.New("lookup proxy: i/o timeout")}, sdkstep.StepAgain, 4 * time.Minute, true, false},
{"a transport fault in the last run", sdkstep.AskRuns, &sdkstep.Passing{Err: errors.New("lookup proxy: i/o timeout")}, sdkstep.StepFailed, 0, true, true},
{"a compile error in run 1", 1, errors.New("undefined: x"), sdkstep.StepFailed, 0, false, true},
} {
src, st := NewFakeAsks(), newSteps()
src.Delivered = c.run
a := testAsk(sdkstep.AskDelivery)
_ = src.PublishAsk(a)
p := &Puller{Source: src, Language: sdkstep.AskGo, Node: "shanks", Put: st.put, Capacity: func() int { return 1 },
PullWait: 10 * time.Millisecond, Run: func(ctx context.Context, a sdkstep.StepAsk) (any, error) { return nil, c.err }}
ctx, cancel := context.WithCancel(context.Background())
go p.Pull(ctx)
key, _ := sdkstep.AskStepKey("shanks", a.ID)
waitFor(t, c.name, func() bool {
ack, _, _ := src.Fates()
src.mu.Lock()
defer src.mu.Unlock()
return len(ack)+len(src.Delayed) == 1
})
cancel()
s, _ := st.get(key)
src.mu.Lock()
delays := append([]time.Duration(nil), src.Delays...)
src.mu.Unlock()
acked, _, _ := src.Fates()
if s.State != c.state || s.Passing != c.passing || s.Run != int(c.run) || (len(acked) == 1) != c.ackedOnly || (!c.ackedOnly && (len(delays) != 1 || delays[0] != c.delay)) ||
(c.state == sdkstep.StepAgain && s.AgainAt.IsZero()) {
t.Errorf("%s: step %+v, acked %v, delays %v; want %s, passing %v, delay %v", c.name, s, acked, delays, c.state, c.passing, c.delay)
}
}
}
// An ask whose job this build module does not know is answered on its step key as refused, naming why, never only dropped
@@ -27,3 +27,9 @@ type Declining struct {
}
func (d *Declining) Error() string { return d.Text }
// Dropped is a step the seat's verb drop ended: its asker no longer waited for it. A step's run ends with one and the runner
// writes the step dropped with its text, the cause its caller named (ADR 0664 §4).
type Dropped struct{ Text string }
func (d *Dropped) Error() string { return d.Text }
+14 -1
View File
@@ -18,7 +18,20 @@ func SeatTools(seat, what string, p *Puller) []stdio.Tool {
return map[string]any{"node": p.Node, "capacity": c.Effective, "capacity_why": c.Why, "capacity_detail": c,
"running": p.Running(), "paused": p.Paused()}, nil
}
return []stdio.Tool{
drop := stdio.Tool{Name: seat + ".drop", Description: "End an " + what + " ask its caller no longer waits for: it is kept as dropped on this node " +
"(so a bus that gives it again has it refused), and a step running here for it is ended, written refused naming who dropped it and why.",
Input: map[string]any{"type": "object", "required": []string{"id", "why"}, "properties": map[string]any{
"id": map[string]any{"type": "string", "description": "the ask's id"},
"by": map[string]any{"type": "string", "description": "who drops it"},
"why": map[string]any{"type": "string", "description": "why the caller no longer waits"}}},
Run: func(args map[string]any) (any, error) {
str := func(k string) string { s, _ := args[k].(string); return s }
if err := p.Drop(str("id"), str("by"), str("why")); err != nil {
return nil, err
}
return map[string]any{}, nil
}}
return []stdio.Tool{drop,
verb("current", "The "+what+" asks this node runs now (each one's id, kind and when it started), its build capacity (the configured one, "+
"bounded by the cores its node's build slice gives over the cores a step needs) and why, "+
"and whether it is paused.", current),
+18 -2
View File
@@ -44,7 +44,12 @@ func Start(ctx context.Context, put Put, node, tool string, run func(ctx context
// begin writes the step running under id, owned by this process (Owner).
func begin(put Put, node, tool, id, what string) (sdkstep.Step, string, error) {
step := sdkstep.Step{ID: id, Tool: tool, Node: node, State: sdkstep.StepRunning, Started: now().UTC(), What: what, Owner: buildrun.Owner}
return beginRun(put, node, tool, id, what, 0)
}
// beginRun is begin for the run-th run of an ask (the bus's delivery count; 0 for a step that is no ask).
func beginRun(put Put, node, tool, id, what string, run int) (sdkstep.Step, string, error) {
step := sdkstep.Step{ID: id, Tool: tool, Node: node, State: sdkstep.StepRunning, Started: now().UTC(), What: what, Owner: buildrun.Owner, Run: run}
key, err := sdkstep.StepKey(node, step.ID)
if err != nil {
return sdkstep.Step{}, "", err
@@ -68,7 +73,7 @@ func end(ctx context.Context, put Put, key string, step sdkstep.Step, run func(c
// the step's ask is not acknowledged as done.
func finish(put Put, key string, step sdkstep.Step, result any, err error, logf func(format string, args ...any)) error {
step.Ended = now().UTC()
switch ref := (*Refusal)(nil); {
switch ref, drop := (*Refusal)(nil), (*Dropped)(nil); {
case err == nil:
step.State = sdkstep.StepPassed
if result != nil {
@@ -79,10 +84,13 @@ func finish(put Put, key string, step sdkstep.Step, result any, err error, logf
step.Result = raw
}
}
case errors.As(err, &drop):
step.State, step.Reason = sdkstep.StepDropped, err.Error()
case errors.As(err, &ref):
step.State, step.Reason = sdkstep.StepRefused, err.Error()
default:
step.State, step.Reason = sdkstep.StepFailed, err.Error()
step.Passing = sdkstep.IsPassing(err) // it ended on a transport fault: its asker may ask it again (ADR 0664 §3)
// A failed step may answer what it found on the way (a build's out-of-step files): kept on its end.
if result != nil {
if raw, merr := json.Marshal(result); merr == nil {
@@ -112,6 +120,14 @@ func finish(put Put, key string, step sdkstep.Step, result any, err error, logf
return nil
}
// again writes the step as given again after a transport fault: the state again with the run's cause, and when the bus gives
// the ask again (ADR 0664 §1). The key is not ended: its asker keeps waiting on it.
func again(put Put, key string, step sdkstep.Step, err error, delay time.Duration) error {
step.State, step.Reason, step.Passing, step.Result = sdkstep.StepAgain, err.Error(), true, nil
step.Ended, step.AgainAt = time.Time{}, now().UTC().Add(delay)
return writeEnd(put, key, step)
}
// writeEnd writes the step's end, trying EndAttempts times.
func writeEnd(put Put, key string, step sdkstep.Step) error {
for attempt := 1; ; attempt++ {
@@ -0,0 +1,14 @@
package buildstep
import sdkstep "git.novox.be/novox/mesh-delivery/sdk/buildstep"
// Transport is err as a sdkstep.Passing when its words are those of a network that did not answer (a lookup, a timeout, a
// reset connection; sdkstep.PassingCause), so the build module's ask is given again after a delay instead of failing
// (novox/hq ADR 0664 §1, issue 1748); any other error, nil included, is returned as it is. Every build module wraps a step
// that fetches over the network with it: a failure of the source (an unknown revision, a wrong checksum) never is one.
func Transport(err error) error {
if err == nil || sdkstep.IsPassing(err) || !sdkstep.PassingCause(err.Error()) {
return err
}
return &sdkstep.Passing{Err: err}
}
@@ -23,6 +23,8 @@ import (
"strings"
"time"
"git.novox.be/novox/mesh-catalog/modules/mesh-build-go/buildstep"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"github.com/therootcompany/xz"
)
@@ -145,7 +147,7 @@ func (m *Module) pinnedFile(ctx context.Context, workspace string, art catalogue
}
response, err := client.Do(request)
if err != nil {
return "", fmt.Errorf("cannot fetch %s: %w", art.URL, err)
return "", buildstep.Transport(fmt.Errorf("cannot fetch %s: %w", art.URL, err))
}
defer response.Body.Close()
if response.StatusCode != http.StatusOK {
@@ -162,7 +164,7 @@ func (m *Module) pinnedFile(ctx context.Context, workspace string, art catalogue
err = cerr
}
if err != nil {
return "", fmt.Errorf("fetching %s: %w", art.URL, err)
return "", buildstep.Transport(fmt.Errorf("fetching %s: %w", art.URL, err))
}
if n > maxUpstream {
return "", fmt.Errorf("%s is larger than %d bytes, more than a machine unpacks; nothing was kept", art.URL, maxUpstream)
@@ -6,6 +6,7 @@ import (
"context"
"crypto/sha256"
"encoding/hex"
"errors"
"io"
"net"
"net/http"
@@ -223,8 +224,17 @@ func TestAPinnedFileOfAnotherDigestIsRefused(t *testing.T) {
func TestAPinnedFileTheServerRefusesFailsTheStep(t *testing.T) {
m, states := pinnedModule(t, "/gone", releaseSHA256, `,"file":"x"`)
if _, st := runBuild(t, m, states, "files"); st.State != sdkstep.StepFailed || !strings.Contains(st.Reason, "404") {
t.Errorf("the step ended %s: %q", st.State, st.Reason)
if _, st := runBuild(t, m, states, "files"); st.State != sdkstep.StepFailed || !strings.Contains(st.Reason, "404") || st.Passing {
t.Errorf("the step ended %s: %q (passing %v): a missing file is no transport fault", st.State, st.Reason, st.Passing)
}
// ADR 0664 §1 (issue 1748): a server that cannot be reached is a transport fault, ended failed as passing for the bus to
// give the ask again.
m, states = pinnedModule(t, "/release.tar.gz", releaseSHA256, `,"file":"x"`)
m.HTTP.Transport.(*http.Transport).DialContext = func(context.Context, string, string) (net.Conn, error) {
return nil, errors.New("dial tcp 192.0.2.1:443: connect: connection refused")
}
if _, st := runBuild(t, m, states, "files"); st.State != sdkstep.StepFailed || !st.Passing {
t.Errorf("an unreachable server ended %s: %q (passing %v)", st.State, st.Reason, st.Passing)
}
}
@@ -113,7 +113,7 @@ func (m *Module) Puller(steps []StepTool, source buildstep.AskSource) *buildstep
return raw, err
}
}
return &buildstep.Puller{Source: source, Language: sdkstep.AskGo, Node: m.Node, Put: m.Steps, PausePut: m.Pauses, Load: load,
return &buildstep.Puller{Source: source, Language: sdkstep.AskGo, Node: m.Node, Put: m.Steps, PausePut: m.Pauses, Load: load, ForgetPause: m.ForgetPause, WatchPauses: m.WatchPauses, PauseKeys: m.ReadPauses.Keys,
Capacity: capacity, SettingsFile: settings, WatchConfig: watch, CoresPerStep: cores, BuildHours: hours, Logf: m.Say, Has: m.Has,
Run: func(ctx context.Context, a sdkstep.StepAsk) (any, error) {
if err := m.Check(ctx); err != nil {
@@ -118,7 +118,7 @@ func mustJSON(v any) []byte { b, _ := json.Marshal(v); return b }
// the node's build capacity as a setting, with the builder's default; the seat's verbs are served as its tools.
func TestManifestClaimsTheGoBuildSeatAndReadsItsCapacity(t *testing.T) {
m := readManifest(t)
if len(m.Claims) != 1 || m.Claims[0].Name != Seat || !slices.Equal(m.Claims[0].Serves, []string{"current", "pause", "resume"}) {
if len(m.Claims) != 1 || m.Claims[0].Name != Seat || !slices.Equal(m.Claims[0].Serves, []string{"current", "drop", "pause", "resume"}) {
t.Fatalf("claims = %+v", m.Claims)
}
if !slices.Contains(m.Reads, KeptState) {
@@ -178,7 +178,7 @@ func (m *Module) packInto(ctx context.Context, workspace, module, name, artifact
return BuildResult{}, fmt.Errorf("packing %s: %w", artifact, err)
}
if err := m.publishArchive(ctx, name, artifact, kind, digest, archive); err != nil {
return BuildResult{}, fmt.Errorf("%s%s: %v", builds.NotStored, artifact, err)
return BuildResult{}, fmt.Errorf("%s%s: %w", builds.NotStored, artifact, err)
}
return BuildResult{Artifact: artifact, Digest: digest, Size: len(archive), Reference: artifactref.SeatReference(ArtifactSeat, name, artifact, digest)}, nil
}
@@ -82,7 +82,7 @@ func (m *Module) buildGoPackage(ctx context.Context, workspace, dir string, art
return res, nil
}
if err := m.publishOn(ctx, seat, path, "package", version, digest, transferName(path, version, digest), body); err != nil {
return BuildResult{}, fmt.Errorf("%s%s: %v", builds.NotStored, art.Name, err)
return BuildResult{}, fmt.Errorf("%s%s: %w", builds.NotStored, art.Name, err)
}
res.Reference = path + "@" + version
return res, nil
@@ -2,6 +2,10 @@ package main
import (
"context"
"encoding/json"
"git.novox.be/novox/mesh-catalog/modules/mesh-build-go/buildstep"
"git.novox.be/novox/mesh-sdk/go/forgekey"
"strings"
"testing"
"time"
@@ -32,3 +36,23 @@ func TestTheStartSweepEndsWhatAPreviousProcessLeft(t *testing.T) {
(&Module{Node: node}).endLeftRunning(ctx, scan, before, logf)
})
}
// ADR 0664 §1 (issue 1748): the module's own paths from a failed fetch or download to its step's error say a network that did
// not answer as a transport fault, and a failure of the source (an unknown revision, a checksum) as none.
func TestTheStepsSayAFailedNetworkAsATransportFault(t *testing.T) {
conformance.Passing(t, []string{conformance.GitFetch, conformance.GoDownload}, func(runner, words string) error {
switch runner {
case conformance.GitFetch:
run := func(context.Context, []string, string, ...string) forgekey.Ran {
return forgekey.Ran{Status: 128, Stderr: words}
}
return buildstep.FetchCommit(context.Background(), run, t.TempDir(), "ssh://mesh-forge/o/r.git", strings.Repeat("a", 40), "")
default:
var info downloadInfo
if err := json.Unmarshal([]byte(words), &info); err != nil {
t.Fatal(err)
}
return downloadFailure(info.Error, "downloading x@y")
}
})
}
@@ -44,6 +44,8 @@ func main() {
ReadSteps: readOf(stdio.State(seat.StateSteps)),
Pauses: stdio.State(sdkstep.PausesState).Put,
ReadPauses: readOf(stdio.State(sdkstep.PausesState)),
ForgetPause: stdio.State(sdkstep.PausesState).Delete,
WatchPauses: stdio.State(sdkstep.PausesState).Watch,
ReadToolchainsState: readOf(stdio.State(seat.StateToolchains)),
Ctx: ctx,
Say: logf,
@@ -4,6 +4,7 @@ import (
"context"
"encoding/json"
"fmt"
stdio "git.novox.be/novox/mesh-sdk/go"
"net/http"
"sync"
"time"
@@ -35,14 +36,16 @@ type Module struct {
// Pauses is the Put of the pauses state, which keeps the node's pause key without an age; ReadPauses reads it back.
Pauses buildstep.Put
ReadPauses StateRead
// ForgetPause deletes one key of the pauses state: a dropped ask's, once it is refused (the seat's verb drop).
ForgetPause func(key string) error
// WatchPauses follows the drops other holders write in the shared pauses state (stdio.State(...).Watch).
WatchPauses func(key string, handle func(stdio.StateChange) error) error
// Emit tells the bus an event (stdio.Emit); nil emits nothing. The guard emits toolchain.reported after each write of
// the toolchains key, so the indexer, which holds no timer, learns that the bucket exists (issue 1444).
Emit func(eventType string, body any) error
// Ctx is the process's context, which a step runs under (the tool call's is cut at 30 s, the step's is not).
Ctx context.Context
Say func(format string, a ...any)
// Pause waits between a download and its retaking (versions.go); nil waits on a timer. A test sets it to record.
Pause func(d time.Duration)
// Run runs the commands of a step (StepRunner); Work holds the workspaces and Cache the module and build caches,
// both outside the account's home (ADR 0466 §1); Home is the account's home.
Run forgekey.Runner
@@ -4,6 +4,7 @@ import (
"context"
"encoding/json"
"errors"
"os"
"strings"
"sync"
"testing"
@@ -198,3 +199,15 @@ func TestParseGoVersion(t *testing.T) {
t.Errorf("accepted a text that is not `go version`'s")
}
}
// ADR 0664 §4 (issue 1748, review 1): the module follows the drops other holders write in the shared pauses state, or a drop
// served by another node would never end a step running here. main wires the watch.
func TestMainWiresTheWatchOfTheDrops(t *testing.T) {
raw, err := os.ReadFile("main.go")
if err != nil {
t.Fatal(err)
}
if !strings.Contains(string(raw), "WatchPauses:") {
t.Error("main.go does not set Module.WatchPauses: a drop written by another holder would not end a running step")
}
}
@@ -150,8 +150,14 @@ func (m *Module) placePin(ctx context.Context, root string, p pin, failure strin
repository := "novox/" + p.repo
r := m.Run(ctx, []string{"GIT_TERMINAL_PROMPT=0"}, "git", "-c", "credential.helper=", "ls-remote", m.remote(repository), "refs/pull/*/head")
if r.Status != 0 || r.Err != "" {
// The forge did not answer: retaken, with its own cause, never held as a refusal until the module changes (issue 1748).
return "", fmt.Errorf("the pin of %s at commit %s: listing the pull requests of %s failed (a download failure that may be retaken): %s", p.module, p.short, repository, dforgekey.Said(r.Err+"\n"+r.Stderr))
// The forge did not answer: its own cause, never held as a refusal until the module changes (issue 1748); a network's
// words make it a sdkstep.Passing, given again after a delay (ADR 0664 §1).
said := dforgekey.Said(r.Err + "\n" + r.Stderr)
err := fmt.Errorf("the pin of %s at commit %s: listing the pull requests of %s failed: %s", p.module, p.short, repository, said)
if sdkstep.PassingCause(said) {
err = &sdkstep.Passing{Err: err}
}
return "", err
}
commit, ref := "", ""
for _, l := range strings.Split(r.Stdout, "\n") {
@@ -14,6 +14,7 @@ import (
"sync"
"testing"
"git.novox.be/novox/mesh-catalog/modules/mesh-build-go/buildstep"
registry "git.novox.be/novox/mesh-delivery/contracts/go/seats/containerimageregistry"
builds "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
sdkstep "git.novox.be/novox/mesh-delivery/sdk/buildstep"
@@ -139,6 +140,27 @@ func TestARefusedPushFailsTheStepNotStored(t *testing.T) {
!strings.Contains(st.Reason, "oci-artifact-registry") {
t.Errorf("no holder: the step ended %s: %q", st.State, st.Reason)
}
// ADR 0664 §1 (issue 1748): "no responders" is the same words for a seat nothing holds and for a holder whose tool runner is
// being replaced; the assigner says which. A named holder makes it a transport fault (the bus gives the ask again), none
// fails at once naming it.
for _, c := range []struct {
name, holders string
passing bool
words string
}{
{"a holder named", `{"seats":[{"seat":"oci-artifact-registry","holders":[{"module":"mesh-registry","node":"novox"}]}],"holds":1}`, true, "the holder is silent: seat oci-artifact-registry, held by mesh-registry on novox"},
{"nothing holds it", `{"seats":[{"seat":"go-package-registry","holders":[{"module":"mesh-registry","node":"novox"}]}],"holds":1}`, false, "nothing holds oci-artifact-registry"},
} {
m.Call = func(key string, _ any) (json.RawMessage, error) {
if key == buildstep.AssignerHolders {
return json.RawMessage(c.holders), nil
}
return nil, errors.New("nats: no responders available for request")
}
if _, st := runBuild(t, m, states, "tools"); st.State != sdkstep.StepFailed || st.Passing != c.passing || !strings.Contains(st.Reason, c.words) {
t.Errorf("%s: the step ended %s: %q (passing %v)", c.name, st.State, st.Reason, st.Passing)
}
}
}
// 1587-S4 (the lead's decision L9): the packed file is not left on the node once its push answered, a candidate's
@@ -30,7 +30,6 @@ import (
"strconv"
"strings"
"sync"
"time"
sdkstep "git.novox.be/novox/mesh-delivery/sdk/buildstep"
"git.novox.be/novox/mesh-sdk/go/forgekey"
@@ -97,37 +96,14 @@ func saidSuffix(said string) string {
return ": " + said
}
// downloadAttempts bounds the downloads of one asked version, and downloadBackoff is the wait after the first failed one
// (twice that after the second): a network failure is taken again, any other fails at once.
const (
downloadAttempts = 3
downloadBackoff = 3 * time.Second
)
// networkFailure says go's words are those of a network that did not answer (a name lookup, a timeout, a refused or
// reset connection), not of a version or a checksum that is wrong.
func networkFailure(s string) bool {
s = strings.ToLower(s)
for _, w := range []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 "} {
if strings.Contains(s, w) {
return true
}
}
return false
}
// retryWait waits d (m.Pause in a test) or until ctx ends.
func (m *Module) retryWait(ctx context.Context, d time.Duration) error {
if m.Pause != nil {
m.Pause(d)
return ctx.Err()
}
select {
case <-time.After(d):
return nil
case <-ctx.Done():
return ctx.Err()
// downloadFailure is a failed download said in go's own words; one whose words are a network's (a lookup, a timeout, a reset
// connection) is a sdkstep.Passing, so the build module gives the ask again after a delay instead of failing it (ADR 0664 §1).
func downloadFailure(failure, format string, args ...any) error {
err := fmt.Errorf("%s: %s", fmt.Sprintf(format, args...), causeLine(failure))
if sdkstep.PassingCause(failure) {
return &sdkstep.Passing{Err: err}
}
return err
}
// takeVersion takes one asked version into dir's modfile (takeVersions).
@@ -140,32 +116,12 @@ func (m *Module) takeVersion(ctx context.Context, root, dir string, tv *TakenVer
tagged := sdkstep.PackageSemver.MatchString(tv.Asked)
var r forgekey.Ran
var info downloadInfo
var failed bool
attempts := 0
for {
attempts++
r = m.Run(ctx, m.goEnv(), "go", "-C", pinsDir, "mod", "download", "-json", tv.Package+"@"+tv.Asked)
info = downloadInfo{}
_ = json.Unmarshal([]byte(r.Stdout), &info)
failed = r.Status != 0 || r.Err != "" || info.Version == "" || info.Error != ""
if !failed || attempts >= downloadAttempts || !networkFailure(info.Error+"\n"+r.Err+"\n"+r.Stderr) {
break
}
// A name lookup, a timeout or a refused connection says nothing of the version: the download is taken again.
if m.Say != nil {
m.Say("%s: downloading %s@%s failed on the network (attempt %d of %d), taken again: %s", filepath.Base(dir), tv.Package, tv.Asked, attempts, downloadAttempts, causeLine(info.Error+"\n"+r.Err+"\n"+r.Stderr))
}
if err := m.retryWait(ctx, time.Duration(attempts)*downloadBackoff); err != nil {
return err
}
}
tried := "taken once"
if attempts > 1 {
tried = fmt.Sprintf("taken %d times", attempts)
}
r = m.Run(ctx, m.goEnv(), "go", "-C", pinsDir, "mod", "download", "-json", tv.Package+"@"+tv.Asked)
_ = json.Unmarshal([]byte(r.Stdout), &info)
failed := r.Status != 0 || r.Err != "" || info.Version == "" || info.Error != ""
if failed && tagged {
failure := strings.TrimSpace(info.Error + "\n" + r.Err + "\n" + r.Stderr)
return fmt.Errorf("downloading the group's tagged version %s of %s through the Go seat failed (a download failure, %s, that may be retaken by a new ask of this fingerprint): %s", tv.Asked, tv.Package, tried, causeLine(failure))
return downloadFailure(failure, "downloading the group's tagged version %s of %s through the Go seat failed", tv.Asked, tv.Package)
}
if failed {
// Not a commit Go can fetch: the head of a group's pull request (ADR 0573 §4), placed from the forge's heads.
@@ -183,7 +139,7 @@ func (m *Module) takeVersion(ctx context.Context, root, dir string, tv *TakenVer
pins.Unlock()
// On no head, and go failed for another cause (the forge or the resolver did not answer): it fails with go's own
// words, and not as a kind the builder holds until the module changes (issue 1748).
return fmt.Errorf("downloading the asked version %s of %s failed (a download failure, %s, that may be retaken by a new ask of this fingerprint): %s", tv.Asked, tv.Package, tried, causeLine(failure))
return downloadFailure(failure, "downloading the asked version %s of %s failed", tv.Asked, tv.Package)
}
if err == nil {
err = m.replaceInModfile(dir, tv.Package, where)
@@ -350,7 +306,7 @@ func comparePre(a, b string) int {
func unknownRevision(failure string) bool {
// go words any failure of a proxy as "invalid version: ..." (issue 1966): a failure of the network is not go knowing
// no such revision, whatever else the text says.
if networkFailure(failure) {
if sdkstep.PassingCause(failure) {
return false
}
return strings.Contains(failure, "unknown revision") || strings.Contains(failure, "invalid version")
@@ -7,7 +7,6 @@ import (
"path/filepath"
"strings"
"testing"
"time"
builds "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
sdkstep "git.novox.be/novox/mesh-delivery/sdk/buildstep"
@@ -29,7 +28,8 @@ func TestAnAskedVersionIsWrittenIntoTheModfileWhateverGoModNames(t *testing.T) {
built, said, edit string
failsWith string // a download failure that is not an unknown revision: the step fails with go's own words
forgeDown bool // the forge's listing of pull request heads fails too
timeouts, downloads int // downloads that time out before go answers; the downloads asked in all (0: not counted)
passing bool // the step ends failed as a transport fault, which the bus gives again (ADR 0664 §1)
downloads int // the downloads asked in all (0: not counted)
}{
{name: "a trunk commit newer than go.mod's", asked: trunk,
answer: `{"Path":"` + sdk + `","Version":"v0.1.15-0.20261101000000-aaaaaaaaaaaa","Sum":"h1:zip=","GoModSum":"h1:mod="}`,
@@ -47,16 +47,16 @@ func TestAnAskedVersionIsWrittenIntoTheModfileWhateverGoModNames(t *testing.T) {
{name: "a head go cannot even look up is placed", asked: commitSHA, failing: true,
answer: `{"Path":"` + sdk + `","Error":"git.novox.be/novox/mesh-sdk/go@` + commitSHA + `: dial tcp: lookup git.novox.be on [::1]:53: connection refused"}`,
built: commitSHA, said: "placed from the pull request's head"},
{name: "a lookup that times out is no pin", asked: offHead, failing: true, failsWith: "i/o timeout",
{name: "a lookup that times out is no pin", asked: offHead, failing: true, failsWith: "i/o timeout", passing: true, downloads: 1,
answer: `{"Path":"` + sdk + `","Error":"git.novox.be/novox/mesh-sdk/go@` + offHead + `: dial tcp: lookup git.novox.be on 10.10.0.3:53: read udp 10.10.0.3:42292->10.10.0.3:53: i/o timeout"}`},
// Issue 1966: go words a public proxy it cannot reach as "invalid version"; that is the network's failure, not an
// unknown revision, so a trunk commit on no pull request head is not labelled "on no pull request head".
{name: "an unreachable proxy worded invalid version is no pin", asked: offHead, failing: true, failsWith: "network is unreachable",
{name: "an unreachable proxy worded invalid version is no pin", asked: offHead, failing: true, failsWith: "network is unreachable", passing: true,
answer: `{"Path":"` + sdk + `","Error":"invalid version: Get \"https://proxy.golang.org/` + sdk + `/@v/` + offHead + `.info\": dial tcp [2a00:1450::1]:443: connect: network is unreachable"}`},
// The forge's own listing failing is a failure that may be retaken too, with its own cause (review 3).
{name: "a forge that cannot be listed is retaken", asked: commitSHA, failing: true, forgeDown: true,
failsWith: "unable to access the forge",
answer: `{"Path":"` + sdk + `","Error":"invalid version: unknown revision ` + pinShort + `"}`},
failsWith: "unable to access the forge", passing: true,
answer: `{"Path":"` + sdk + `","Error":"invalid version: unknown revision ` + pinShort + `"}`},
// ADR 0607 §3, §5 (design 51 *Tagged versions*): a group's tagged version is fetched by that version and built
// whatever go.mod names; one the Go seat lacks fails naming it, never looked for among the pull request heads.
{name: "a group's tagged version", asked: "v0.1.15-feat-x.2",
@@ -67,22 +67,16 @@ func TestAnAskedVersionIsWrittenIntoTheModfileWhateverGoModNames(t *testing.T) {
built: "v0.1.14-feat-x.2", said: "the group's tagged version", edit: "-require=" + sdk + "@v0.1.14-feat-x.2"},
{name: "a tagged version the Go seat lacks", asked: "v0.1.15-feat-x.3", failing: true, failsWith: "the group's tagged version v0.1.15-feat-x.3",
answer: `{"Path":"` + sdk + `","Error":"` + sdk + `@v0.1.15-feat-x.3: reading https://seat/go/@v/v0.1.15-feat-x.3.info: 404 Not Found"}`},
// Issue 1966: a network failure is taken again a bounded number of times; another failure is not.
{name: "a timeout twice is taken again", asked: trunk, timeouts: 2, downloads: 3,
answer: `{"Path":"` + sdk + `","Version":"v0.1.15-0.20261101000000-aaaaaaaaaaaa","Sum":"h1:zip=","GoModSum":"h1:mod="}`,
built: "v0.1.15-0.20261101000000-aaaaaaaaaaaa", edit: "-require=" + sdk + "@v0.1.15-0.20261101000000-aaaaaaaaaaaa"},
{name: "a timeout that stays fails after three downloads", asked: offHead, failing: true, failsWith: "taken 3 times", downloads: 3,
answer: `{"Path":"` + sdk + `","Error":"lookup git.novox.be on 10.10.0.3:53: i/o timeout"}`},
// ADR 0664 §1 (issue 1748): one download; a failure of the network is a transport fault the bus gives the ask again for after a
// delay, no retry here; another failure is not one.
{name: "an unknown revision is not taken again", asked: offHead, failing: true, failsWith: "on no pull request head", downloads: 1,
answer: `{"Path":"` + sdk + `","Error":"invalid version: unknown revision ` + pinShort + `"}`},
{name: "a refused connection is no pin", asked: offHead, failing: true, failsWith: "connection refused",
{name: "a refused connection is no pin", asked: offHead, failing: true, failsWith: "connection refused", passing: true,
answer: `{"Path":"` + sdk + `","Error":"git ls-remote -q origin in /tmp/gopath: dial tcp 10.10.0.1:443: connect: connection refused"}`},
} {
t.Run(c.name, func(t *testing.T) {
h, states := newPinHarness(t, pinnedGoMod, false, pullList)
h.f.gomods = "go.mod\nother/go.mod\n"
var paused []time.Duration
h.m.Pause = func(d time.Duration) { paused = append(paused, d) }
downloads := 0
base := h.m.Run
h.m.Run = func(ctx context.Context, env []string, name string, args ...string) forgekey.Ran {
@@ -107,9 +101,6 @@ func TestAnAskedVersionIsWrittenIntoTheModfileWhateverGoModNames(t *testing.T) {
h.cmds = append(h.cmds, cmdRecord{env: env, line: line})
h.mu.Unlock()
downloads++
if downloads <= c.timeouts {
return forgekey.Ran{Status: 1, Stdout: `{"Path":"` + sdk + `","Error":"lookup git.novox.be on 10.10.0.3:53: i/o timeout"}`}
}
if c.failing {
return forgekey.Ran{Status: 1, Stdout: c.answer}
}
@@ -122,18 +113,18 @@ func TestAnAskedVersionIsWrittenIntoTheModfileWhateverGoModNames(t *testing.T) {
t.Fatal(err)
}
st := states.waitEnd(t)
if c.downloads != 0 && (downloads != c.downloads || len(paused) != c.downloads-1) {
t.Errorf("go downloaded %d times with %d waits, want %d downloads", downloads, len(paused), c.downloads)
if c.downloads != 0 && downloads != c.downloads {
t.Errorf("go downloaded %d times, want %d: no retry here", downloads, c.downloads)
}
if c.failsWith == "on no pull request head" { // not a network failure: the forge's heads said, no retaking
if st.State != sdkstep.StepFailed || !strings.Contains(st.Reason, c.failsWith) {
if st.State != sdkstep.StepFailed || !strings.Contains(st.Reason, c.failsWith) || st.Passing {
t.Errorf("the unknown revision ended %s: %q", st.State, st.Reason)
}
return
}
if c.failsWith != "" {
if st.State != sdkstep.StepFailed || !strings.Contains(st.Reason, c.failsWith) || strings.Contains(st.Reason, "no pull request head") ||
!strings.Contains(st.Reason, "may be retaken") || builds.FailedUntilChanged(st.Reason) {
strings.Contains(st.Reason, "may be retaken") || st.Passing != c.passing || builds.FailedUntilChanged(st.Reason) {
t.Errorf("the download failure ended %s: %q", st.State, st.Reason)
}
return
+1
View File
@@ -102,6 +102,7 @@ replace git.novox.be/novox/mesh-catalog/modules/ssh-client/sshkeys => ../ssh-cli
require (
git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions v0.0.0-00010101000000-000000000000 // indirect
git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner v0.0.0-00010101000000-000000000000 // indirect
git.novox.be/novox/mesh-delivery/contracts/go/clients/gopackageregistryclient v0.0.0-00010101000000-000000000000 // indirect
git.novox.be/novox/mesh-delivery/contracts/go/clients/npmpackageregistryclient v0.0.0-00010101000000-000000000000 // indirect
git.novox.be/novox/mesh-delivery/contracts/go/clients/ociartifactregistryclient v0.0.0-00010101000000-000000000000 // indirect
+4 -2
View File
@@ -9,6 +9,7 @@
"runs-as": "mesh-build-go",
"capabilities": [
"package-manager",
"seat-settle-delay",
"user-namespaces"
],
"state": [
@@ -20,7 +21,7 @@
{
"name": "node-build-go",
"scope": "node",
"serves": ["current", "pause", "resume"]
"serves": ["current", "drop", "pause", "resume"]
}
],
"consumes": [
@@ -35,7 +36,8 @@
"seat:oci-artifact-registry.publish-part",
"seat:go-package-registry.publish",
"seat:go-package-registry.publish-part",
"seat:mesh-delivery-checker.input"
"seat:mesh-delivery-checker.input",
"seat:mesh-module-assigner.assignments"
],
"config": {
"capacity": {
@@ -62,7 +62,7 @@ func (m *Module) Puller(steps []StepTool, source buildstep.AskSource) *buildstep
if m.ReadPauses.Get != nil {
load = func(key string) ([]byte, error) { return m.ReadPauses.Get(key) }
}
return &buildstep.Puller{Source: source, Language: sdkstep.AskImage, Node: m.Node, Put: m.Steps, PausePut: m.Pauses, Load: load,
return &buildstep.Puller{Source: source, Language: sdkstep.AskImage, Node: m.Node, Put: m.Steps, PausePut: m.Pauses, Load: load, ForgetPause: m.ForgetPause, WatchPauses: m.WatchPauses, PauseKeys: m.ReadPauses.Keys,
Capacity: capacity, SettingsFile: settings, WatchConfig: watch, CoresPerStep: cores, BuildHours: hours, Eligible: m.Eligible, Logf: m.Say,
Run: func(ctx context.Context, a sdkstep.StepAsk) (any, error) {
if err := m.Check(ctx); err != nil {
@@ -173,7 +173,7 @@ func (m *Module) build(ctx context.Context, args map[string]any) (any, error) {
argv = append(argv, contextDir)
if r := m.Run(ctx, m.env(), "buildah", argv...); r.Status != 0 || r.Err != "" {
m.sayFailure("buildah bud of "+art.Name+" of "+manifest.Module, r)
return nil, fmt.Errorf("buildah bud of %s: %s", art.Name, cause(r))
return nil, buildstep.Transport(fmt.Errorf("buildah bud of %s: %s", art.Name, cause(r)))
}
defer m.remove(local)
digest, err := m.localDigest(ctx, local)
@@ -178,6 +178,8 @@ func TestBuildFailsWithBuildahsError(t *testing.T) {
[]string{"Error: the end"}, []string{"early noise", long}},
"nothing on stderr": {forgekey.Ran{Status: 2, Stdout: steps}, []string{"exit status 2", "nothing on stderr", "dl-cdn"}, nil},
"never ran": {forgekey.Ran{Err: "ENOENT"}, []string{"ENOENT", "wrote nothing"}, nil},
"a pull the network cut": {forgekey.Ran{Status: 125, Stdout: steps, Stderr: "Error: creating build container: initializing source docker://alpine: pinging container registry registry-1.docker.io: Get \"https://registry-1.docker.io/v2/\": dial tcp: lookup registry-1.docker.io on 10.10.0.3:53: i/o timeout\n"},
[]string{"i/o timeout"}, nil},
}
for name, c := range cases {
m, f, states := setup(t, same)
@@ -200,6 +202,10 @@ func TestBuildFailsWithBuildahsError(t *testing.T) {
if len(st.Reason) > CauseBytes+200 {
t.Errorf("%s: the reason is %d bytes, more than its bound", name, len(st.Reason))
}
// ADR 0664 §1: a pull the network cut ends passing, for the bus to give the ask again; no other failure does.
if st.Passing != strings.Contains(name, "the network cut") {
t.Errorf("%s: passing = %v", name, st.Passing)
}
if len(f.called("push")) != 0 {
t.Errorf("%s: a failed build pushed", name)
}
@@ -24,7 +24,7 @@ func TestTheModuleInvokesTheImageSeatAndHoldsNoRegistry(t *testing.T) {
t.Fatal(err)
}
if want := []string{"seat:container-image-registry.publish", "seat:container-image-registry.publish-part",
"seat:container-image-registry.mirror"}; !slices.Equal(m.Invokes, want) {
"seat:container-image-registry.mirror", "seat:mesh-module-assigner.assignments"}; !slices.Equal(m.Invokes, want) {
t.Errorf("invokes = %v, want %v", m.Invokes, want)
}
// The builder's version.numbered for the late publish (ADR 0617 §1), and the seat's outcomes a copy waits on.
@@ -112,7 +112,9 @@ func (m *Module) copy(ctx context.Context, args map[string]any) (any, error) {
raw, err := m.call()("seat:"+buildstep.ImageSeat+"."+reg.VerbMirror, reg.MirrorInput{Name: module + "/" + name, Source: from, Digest: digest})
if err != nil {
said.Refused(err)
return nil, fmt.Errorf("%s%s: %s's mirror: %v", builds.NotStored, name, buildstep.ImageSeat, err)
// A seat verb's "no responders" with a holder named is a transport fault (ADR 0664 §1): the seat's tool runner is replaced.
err = buildstep.HolderSilentOr(m.call(), buildstep.ImageSeat, err)
return nil, fmt.Errorf("%s%s: %s's mirror: %w", builds.NotStored, name, buildstep.ImageSeat, err)
}
var answer reg.MirrorAnswer
_ = json.Unmarshal(raw, &answer)
@@ -2,6 +2,11 @@ package main
import (
"context"
"errors"
"git.novox.be/novox/mesh-catalog/modules/mesh-build-go/buildstep"
sdkstep "git.novox.be/novox/mesh-delivery/sdk/buildstep"
"git.novox.be/novox/mesh-sdk/go/forgekey"
"strings"
"testing"
"time"
@@ -17,3 +22,25 @@ func TestTheStartSweepEndsWhatAPreviousProcessLeft(t *testing.T) {
(&Module{Node: node}).endLeftRunning(ctx, scan, before, logf)
})
}
// ADR 0664 §1 (issue 1748): the module's own paths from a failed fetch or image build to its step's error say a network that
// did not answer as a transport fault, and a failure of the source (a RUN line that fails) as none.
func TestTheStepsSayAFailedNetworkAsATransportFault(t *testing.T) {
conformance.Passing(t, []string{conformance.GitFetch, conformance.BuildahBud}, func(runner, words string) error {
switch runner {
case conformance.GitFetch:
run := func(context.Context, []string, string, ...string) forgekey.Ran {
return forgekey.Ran{Status: 128, Stderr: words}
}
return buildstep.FetchCommit(context.Background(), run, t.TempDir(), "ssh://mesh-forge/o/r.git", strings.Repeat("a", 40), "")
default:
m, f, states := setup(t, same)
f.failBud = forgekey.Ran{Status: 125, Stderr: words}
_, st := runBuild(t, m, states, nil)
if st.Passing {
return &sdkstep.Passing{Err: errors.New(st.Reason)}
}
return errors.New(st.Reason)
}
})
}
@@ -32,15 +32,17 @@ func logf(format string, a ...any) { fmt.Fprintf(os.Stderr, "[mesh-build-image]
func main() {
ctx := context.Background()
m := &Module{
Node: strings.TrimSpace(os.Getenv("MESH_NODE")),
Steps: stdio.State(seat.StateSteps).Put,
ReadSteps: readOf(stdio.State(seat.StateSteps)),
Pauses: stdio.State(sdkstep.PausesState).Put,
ReadPauses: readOf(stdio.State(sdkstep.PausesState)),
Ctx: ctx,
Say: logf,
Work: WorkDir,
Store: StoreDir,
Node: strings.TrimSpace(os.Getenv("MESH_NODE")),
Steps: stdio.State(seat.StateSteps).Put,
ReadSteps: readOf(stdio.State(seat.StateSteps)),
Pauses: stdio.State(sdkstep.PausesState).Put,
ReadPauses: readOf(stdio.State(sdkstep.PausesState)),
ForgetPause: stdio.State(sdkstep.PausesState).Delete,
WatchPauses: stdio.State(sdkstep.PausesState).Watch,
Ctx: ctx,
Say: logf,
Work: WorkDir,
Store: StoreDir,
}
m.Floor = floorOf(m.Node)
if u, err := user.Current(); err == nil {
@@ -72,7 +72,7 @@ func TestManifestDeclaresTheAccountTheCapabilityAndThePackages(t *testing.T) {
if home["type"] != "directory" || home["mode"] != "0700" || home["owner"] != "mesh-build-image" || home["place"] != "." {
t.Errorf("home = %v, want its directory at mode 0700 owned by the account", home)
}
for _, c := range []string{"user-namespaces", "package-manager"} {
for _, c := range []string{"user-namespaces", "package-manager", "seat-settle-delay"} {
if !slices.Contains(m.Capabilities, c) {
t.Errorf("capabilities = %v, want %s", m.Capabilities, c)
}
@@ -125,7 +125,7 @@ func TestManifestKeepsItsStatesClaimsItsBuildSeatAndDeclaresItsHealthTool(t *tes
if want := []string{"keys", MirrorWaitsState, "pauses", "steps"}; !slices.Equal(states, want) {
t.Errorf("state = %v, want %v", states, want)
}
if len(m.Claims) != 1 || m.Claims[0].Name != Seat || !slices.Equal(m.Claims[0].Serves, []string{"current", "pause", "resume"}) {
if len(m.Claims) != 1 || m.Claims[0].Name != Seat || !slices.Equal(m.Claims[0].Serves, []string{"current", "drop", "pause", "resume"}) {
t.Errorf("claims = %v, want the seat %s alone", m.Claims, Seat)
}
raw, _ := os.ReadFile(filepath.Join(moduleDir, "module.json"))
@@ -4,6 +4,7 @@ import (
"context"
"errors"
"fmt"
stdio "git.novox.be/novox/mesh-sdk/go"
"sync"
"time"
@@ -30,6 +31,10 @@ type Module struct {
// Pauses is the Put of the pauses state, which keeps the node's pause key without an age; ReadPauses reads it back.
Pauses buildstep.Put
ReadPauses StateRead
// ForgetPause deletes one key of the pauses state: a dropped ask's, once it is refused (the seat's verb drop).
ForgetPause func(key string) error
// WatchPauses follows the drops other holders write in the shared pauses state (stdio.State(...).Watch).
WatchPauses func(key string, handle func(stdio.StateChange) error) error
// Ctx is the process's context, which a step runs under (the tool call's is cut at 30 s, the step's is not).
Ctx context.Context
Say func(format string, a ...any)
@@ -136,7 +136,7 @@ func (m *Module) publish(ctx context.Context, workspace, module, local string, r
Transfer: buildstep.TransferName(name, q, img.Digest)}
sent, err := buildstep.PublishFiles(ctx, m.call(), buildstep.ImageSeat, in, img.Files)
if err != nil {
return nil, fmt.Errorf("%s%s: %v", builds.NotStored, res.Artifact, err)
return nil, fmt.Errorf("%s%s: %w", builds.NotStored, res.Artifact, err)
}
m.say("published %s on %s: %d of its %d parts sent, the rest held", name, buildstep.ImageSeat, len(sent.Parts), len(img.Files))
res.Digest, res.Pushed = img.Digest, true
+1
View File
@@ -34,6 +34,7 @@ require git.novox.be/novox/mesh-catalog/modules/ssh-client/sshkeys v0.0.0-000101
replace git.novox.be/novox/mesh-catalog/modules/ssh-client/sshkeys => ../ssh-client/sshkeys
require (
git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner v0.0.0-00010101000000-000000000000 // indirect
git.novox.be/novox/mesh-delivery/contracts/go/clients/gopackageregistryclient v0.0.0-00010101000000-000000000000 // indirect
git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient v0.0.0-00010101000000-000000000000 // indirect
git.novox.be/novox/mesh-delivery/contracts/go/clients/npmpackageregistryclient v0.0.0-00010101000000-000000000000 // indirect
+4 -2
View File
@@ -9,6 +9,7 @@
"runs-as": "mesh-build-image",
"capabilities": [
"package-manager",
"seat-settle-delay",
"user-namespaces"
],
"state": [
@@ -20,7 +21,7 @@
{
"name": "node-build-image",
"scope": "node",
"serves": ["current", "pause", "resume"]
"serves": ["current", "drop", "pause", "resume"]
}
],
"config": {
@@ -63,7 +64,8 @@
"invokes": [
"seat:container-image-registry.publish",
"seat:container-image-registry.publish-part",
"seat:container-image-registry.mirror"
"seat:container-image-registry.mirror",
"seat:mesh-module-assigner.assignments"
],
"consumes": [
"mesh-delivery-builder.version.numbered",
@@ -72,7 +72,7 @@ func (m *Module) Puller(steps []StepTool, source buildstep.AskSource) *buildstep
if m.ReadPauses.Get != nil {
load = func(key string) ([]byte, error) { return m.ReadPauses.Get(key) }
}
return &buildstep.Puller{Source: source, Language: sdkstep.AskNode, Node: m.Node, Put: m.Steps, PausePut: m.Pauses, Load: load,
return &buildstep.Puller{Source: source, Language: sdkstep.AskNode, Node: m.Node, Put: m.Steps, PausePut: m.Pauses, Load: load, ForgetPause: m.ForgetPause, WatchPauses: m.WatchPauses, PauseKeys: m.ReadPauses.Keys,
Capacity: capacity, SettingsFile: settings, WatchConfig: watch, CoresPerStep: cores, BuildHours: hours, Logf: m.Say,
Run: func(ctx context.Context, a sdkstep.StepAsk) (any, error) {
if err := m.Prepare(ctx); err != nil {
@@ -168,7 +168,7 @@ func (m *Module) build(ctx context.Context, args map[string]any) (any, error) {
// Published from this process once the build ended, a candidate's as a trunk build's (ADR 0498 §1 §2; CAT-BN3): the
// step answers the seat's reference, or fails not stored. The packed bytes are never written to the node (L9).
if err := m.publishOn(ctx, buildstep.ArtifactSeat, a.module+"/"+a.Name, "bundle", "", digest, archive); err != nil {
return nil, fmt.Errorf("%s%s: %v", builds.NotStored, a.Name, err)
return nil, fmt.Errorf("%s%s: %w", builds.NotStored, a.Name, err)
}
files := []string{}
_ = filepath.WalkDir(bundleDir, func(p string, d os.DirEntry, err error) error {
@@ -353,7 +353,7 @@ func (m *Module) installOwn(ctx context.Context, root, dir, rel string) (install
"--no-audit", "--no-fund", "--prefix", scratch, "--userconfig", npmrc)
os.Remove(npmrc)
if why := ranFailed(r); why != "" {
return false, "", fmt.Errorf("installing the packages of %s (%s): %s", rel, strings.Join(names, ", "), why)
return false, "", buildstep.Transport(fmt.Errorf("installing the packages of %s (%s): %s", rel, strings.Join(names, ", "), why))
}
nm := filepath.Join(scratch, "node_modules")
if err := removeSDKCopies(nm); err != nil {
@@ -2,6 +2,8 @@ package main
import (
"context"
"git.novox.be/novox/mesh-catalog/modules/mesh-build-go/buildstep"
"strings"
"testing"
"time"
@@ -30,3 +32,20 @@ func TestTheStartSweepEndsWhatAPreviousProcessLeft(t *testing.T) {
(&Module{Node: node}).endLeftRunning(ctx, scan, before, logf)
})
}
// ADR 0664 §1 (issue 1748): the module's own paths from a failed fetch or install to its step's error say a network that did
// not answer as a transport fault, and a failure of the source (a package the registry lacks) as none.
func TestTheStepsSayAFailedNetworkAsATransportFault(t *testing.T) {
conformance.Passing(t, []string{conformance.GitFetch, conformance.NpmCI}, func(runner, words string) error {
switch runner {
case conformance.GitFetch:
run := func(context.Context, []string, string, ...string) forgekey.Ran {
return forgekey.Ran{Status: 128, Stderr: words}
}
return buildstep.FetchCommit(context.Background(), run, t.TempDir(), "ssh://mesh-forge/o/r.git", strings.Repeat("a", 40), "")
default:
bad := &fakeNpm{ran: forgekey.Ran{Status: 1, Stderr: words}}
return toolchainIn(t, bad).Ensure(context.Background(), Recipe{Steps: "k", Lock: "k"}.Version())
}
})
}
@@ -50,6 +50,8 @@ func main() {
ReadSteps: readOf(stdio.State(seat.StateSteps)),
Pauses: stdio.State(sdkstep.PausesState).Put,
ReadPauses: readOf(stdio.State(sdkstep.PausesState)),
ForgetPause: stdio.State(sdkstep.PausesState).Delete,
WatchPauses: stdio.State(sdkstep.PausesState).Watch,
ReadToolchainsState: readOf(stdio.State(seat.StateToolchains)),
Work: WorkDir,
Run: dforgekey.ExecRunnerClean(0),
@@ -65,6 +65,11 @@ func TestManifestDeclaresTheAccountTheCapabilityAndThePackages(t *testing.T) {
if home["type"] != "directory" || home["mode"] != "0700" || home["owner"] != "mesh-build-node" || home["place"] != "." {
t.Errorf("home = %v, want its directory at mode 0700 owned by the account", home)
}
// ADR 0664 §1 (issue 1748): the delay a step given again waits is the tool runner's capability; the want waits on a node whose
// runner does not state it.
if !slices.Contains(m.Capabilities, "seat-settle-delay") {
t.Errorf("capabilities = %v, want seat-settle-delay", m.Capabilities)
}
if !slices.Contains(m.Capabilities, "user-namespaces") {
t.Errorf("capabilities = %v, want user-namespaces", m.Capabilities)
}
@@ -112,7 +117,7 @@ func TestManifestKeepsItsStatesClaimsItsBuildSeatAndDeclaresItsHealthTool(t *tes
if want := []string{"keys", "pauses", "steps", "toolchains"}; !slices.Equal(states, want) {
t.Errorf("state = %v, want %v", states, want)
}
if len(m.Claims) != 1 || m.Claims[0].Name != Seat || !slices.Equal(m.Claims[0].Serves, []string{"current", "pause", "resume"}) {
if len(m.Claims) != 1 || m.Claims[0].Name != Seat || !slices.Equal(m.Claims[0].Serves, []string{"current", "drop", "pause", "resume"}) {
t.Errorf("claims = %v, want the seat %s alone", m.Claims, Seat)
}
if !slices.Contains(m.Tools, HealthTool) {
@@ -3,6 +3,7 @@ package main
import (
"context"
"fmt"
stdio "git.novox.be/novox/mesh-sdk/go"
"net/http"
"os"
"sync"
@@ -49,6 +50,10 @@ type Module struct {
// Pauses is the Put of the pauses state, which keeps the node's pause key without an age; ReadPauses reads it back.
Pauses buildstep.Put
ReadPauses StateRead
// ForgetPause deletes one key of the pauses state: a dropped ask's, once it is refused (the seat's verb drop).
ForgetPause func(key string) error
// WatchPauses follows the drops other holders write in the shared pauses state (stdio.State(...).Watch).
WatchPauses func(key string, handle func(stdio.StateChange) error) error
// Emit tells the bus an event (stdio.Emit); nil emits nothing. The guard emits toolchain.reported after each write of
// the toolchains key, so the indexer, which holds no timer, learns that the bucket exists (issue 1444).
Emit func(eventType string, body any) error
@@ -228,3 +228,15 @@ func TestParseNodeVersion(t *testing.T) {
t.Errorf("accepted a text that is not `node --version`'s")
}
}
// ADR 0664 §4 (issue 1748, review 1): the module follows the drops other holders write in the shared pauses state, or a drop
// served by another node would never end a step running here. main wires the watch.
func TestMainWiresTheWatchOfTheDrops(t *testing.T) {
raw, err := os.ReadFile("main.go")
if err != nil {
t.Fatal(err)
}
if !strings.Contains(string(raw), "WatchPauses:") {
t.Error("main.go does not set Module.WatchPauses: a drop written by another holder would not end a running step")
}
}
@@ -93,7 +93,7 @@ func (m *Module) installDev(ctx context.Context, root, dir, rel string) error {
env := append(m.baseEnv(), "npm_config_cache="+m.npmCache())
r := m.Run(ctx, env, "npm", "ci", "--ignore-scripts", "--no-bin-links", "--no-audit", "--no-fund", "--prefix", dir, "--userconfig", npmrc)
if why := ranFailed(r); why != "" {
return fmt.Errorf("installing the dependencies of the package in %s: %s", rel, why)
return buildstep.Transport(fmt.Errorf("installing the dependencies of the package in %s: %s", rel, why))
}
return writeShims(filepath.Join(dir, "node_modules"), shimDir(root, rel))
}
@@ -551,7 +551,7 @@ func (m *Module) buildPackage(ctx context.Context, workspace, module, root, dir,
return res, nil
}
if err := m.publishOn(ctx, seat, p.Name, "package", p.Version, p.Digest, p.tgz); err != nil {
return nil, fmt.Errorf("%s%s: %v", builds.NotStored, a.Name, err)
return nil, fmt.Errorf("%s%s: %w", builds.NotStored, a.Name, err)
}
res.Reference = p.Name + "@" + p.Version
return res, nil
@@ -17,6 +17,7 @@ import (
"git.novox.be/novox/mesh-catalog/modules/mesh-build-go/buildstep"
builds "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
sdkstep "git.novox.be/novox/mesh-delivery/sdk/buildstep"
)
// pushed are the blobs the fake holder took, by digest, for the tests that read an archive back.
@@ -130,6 +131,18 @@ func TestARefusedPushFailsNotStoredAndNoPackedFileStays(t *testing.T) {
if c.words != "" && (err == nil || !strings.HasPrefix(err.Error(), builds.NotStored+"tools: ") || !strings.Contains(err.Error(), c.words)) {
t.Errorf("a refused publish answered %v", err)
}
if c.noSeat {
// ADR 0664 §1: with a holder named by the assigner the same words are a transport fault, the bus gives the ask again.
m.Call = func(key string, _ any) (json.RawMessage, error) {
if key == buildstep.AssignerHolders {
return json.RawMessage(`{"seats":[{"seat":"oci-artifact-registry","holders":[{"module":"mesh-registry","node":"novox"}]}],"holds":1}`), nil
}
return nil, errors.New("nats: no responders available for request")
}
if _, err := m.build(context.Background(), buildArgs()); err == nil || !sdkstep.IsPassing(err) || !strings.Contains(err.Error(), "the holder is silent") {
t.Errorf("no responders with a holder named answered %v, want a transport fault", err)
}
}
var archives []string
filepath.WalkDir(work, func(p string, d os.DirEntry, err error) error {
if err == nil && strings.HasSuffix(p, ".tar.gz") {
@@ -10,6 +10,7 @@ package main
import (
"context"
"fmt"
"git.novox.be/novox/mesh-catalog/modules/mesh-build-go/buildstep"
"io/fs"
"os"
"path/filepath"
@@ -99,8 +100,8 @@ func (t Toolchain) Ensure(ctx context.Context, version string) error {
"npm", "ci", "--ignore-scripts", "--no-bin-links", "--no-audit", "--no-fund", "--prefix", dir, "--userconfig", npmrc)
os.Remove(npmrc)
if ran.Err != "" || ran.Status != 0 {
return fmt.Errorf("npm ci --ignore-scripts for the toolchain %s failed (status %d %s): %s",
version, ran.Status, ran.Err, tail(ran.Stderr))
return buildstep.Transport(fmt.Errorf("npm ci --ignore-scripts for the toolchain %s failed (status %d %s): %s",
version, ran.Status, ran.Err, tail(ran.Stderr)))
}
if err := os.WriteFile(filepath.Join(dir, installedMark), []byte(version+"\n"), 0o600); err != nil {
return err
@@ -3,6 +3,7 @@ package main
import (
"context"
"errors"
sdkstep "git.novox.be/novox/mesh-delivery/sdk/buildstep"
"os"
"path/filepath"
"strings"
@@ -86,6 +87,18 @@ func TestAMissingToolchainDirectoryIsInstalledOnceFromTheLock(t *testing.T) {
if err := tc.Ensure(context.Background(), version); err != nil || len(fake.calls) != 1 {
t.Errorf("a present directory ran npm again: %d calls, %v", len(fake.calls), err)
}
// ADR 0664 §1 (issue 1748): an install the network cut is a transport fault, the bus gives the ask again; a package the
// registry lacks is not.
for _, c := range []struct {
stderr string
passing bool
}{{"npm error code ECONNRESET\nnpm error network aborted", true}, {"npm error code EAI_AGAIN", true}, {"npm error code E404\nnpm error 404 Not Found", false}} {
bad := &fakeNpm{ran: forgekey.Ran{Status: 1, Stderr: c.stderr}}
err := toolchainIn(t, bad).Ensure(context.Background(), Recipe{Steps: "2", Lock: "y"}.Version())
if err == nil || sdkstep.IsPassing(err) != c.passing {
t.Errorf("%q: %v, passing %v, want %v", c.stderr, err, err != nil && sdkstep.IsPassing(err), c.passing)
}
}
}
// 1488/AC4: the install is told the mesh's registry for @novox with the account's credential, and only through a
+1
View File
@@ -94,6 +94,7 @@ require git.novox.be/novox/mesh-catalog/modules/ssh-client/sshkeys v0.0.0-000101
replace git.novox.be/novox/mesh-catalog/modules/ssh-client/sshkeys => ../ssh-client/sshkeys
require (
git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner v0.0.0-00010101000000-000000000000 // indirect
git.novox.be/novox/mesh-delivery/contracts/go/clients/gopackageregistryclient v0.0.0-00010101000000-000000000000 // indirect
git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient v0.0.0-00010101000000-000000000000 // indirect
git.novox.be/novox/mesh-delivery/contracts/go/clients/npmpackageregistryclient v0.0.0-00010101000000-000000000000 // indirect
+4 -2
View File
@@ -9,6 +9,7 @@
"runs-as": "mesh-build-node",
"capabilities": [
"package-manager",
"seat-settle-delay",
"user-namespaces"
],
"state": [
@@ -20,7 +21,7 @@
{
"name": "node-build-node",
"scope": "node",
"serves": ["current", "pause", "resume"]
"serves": ["current", "drop", "pause", "resume"]
}
],
"config": {
@@ -76,7 +77,8 @@
"seat:oci-artifact-registry.publish",
"seat:oci-artifact-registry.publish-part",
"seat:npm-package-registry.publish",
"seat:npm-package-registry.publish-part"
"seat:npm-package-registry.publish-part",
"seat:mesh-module-assigner.assignments"
],
"own-secrets": {
"broker": "${dir:state}/broker"