Merge pull request '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) from fix/1748-transport-retry into main

This commit was merged in pull request #649.
This commit is contained in:
2026-11-09 22:58:43 +00:00
77 changed files with 1022 additions and 142 deletions
+25 -1
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 // 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 // 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 // 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 package failurekit
import ( import (
@@ -13,6 +14,7 @@ import (
"io/fs" "io/fs"
"os" "os"
"path/filepath" "path/filepath"
"slices"
"sort" "sort"
"strings" "strings"
) )
@@ -28,8 +30,14 @@ type manifest struct {
Name string `json:"name"` Name string `json:"name"`
} `json:"claims"` } `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-* // Judge reads every modules/<m>/module.json of the catalogue at root and judges each module that claims a node-build-*
// seat. // seat.
func Judge(root string) ([]Judged, error) { 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 "+ 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)) "(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, "; ") j.Refused = strings.Join(refused, "; ")
out = append(out, j) 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 // 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 // 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) { func TestABuildModuleWithATestStepMustRunTheKit(t *testing.T) {
root := t.TempDir() root := t.TempDir()
const imp = "package main\n\nimport \"git.novox.be/novox/mesh-catalog/modules/mesh-build-go/buildstep/conformance\"\n\n" 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 kit := imp + "func TestKit(t *testing.T) { conformance.Run(t, conformance.GoTest, nil) }\n" + sweep
for path, body := range map[string]string{ 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/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/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/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/c/cmd/c/kit_test.go": imp + sweep,
"modules/f/module.json": `{"claims":[{"name":"node-build-f"}],"tools":["build_f_build"]}`, "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/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/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/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/module.json": `{"claims":[{"name":"node-build-e"}],"tools":["build_e_test"]}`,
"modules/e/cmd/e/kit.go": kit, // not a test file "modules/e/cmd/e/kit.go": kit, // not a test file
} { } {
@@ -40,9 +45,9 @@ func TestABuildModuleWithATestStepMustRunTheKit(t *testing.T) {
if err != nil { if err != nil {
t.Fatal(err) 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) { 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 { for _, j := range got {
is := "passes" 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 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. 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 ( import (
"embed" "embed"
"fmt" "fmt"
"slices"
"strings" "strings"
"testing" "testing"
@@ -199,3 +200,80 @@ func describe(failed []buildstep.Failure) string {
} }
func quote(s string) string { return fmt.Sprintf("%q", s) } 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 ( import (
"context" "context"
"encoding/json" "encoding/json"
"errors"
"strings" "strings"
"testing" "testing"
"time" "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 // 1890/AC7: the SDK's start sweep keeps the left-running case, and the kit refuses a sweep that ends nothing or ends
// everything. // everything.
t.Run("the left-running case", func(t *testing.T) { t.Run("the left-running case", func(t *testing.T) {
+13 -2
View File
@@ -9,6 +9,7 @@ import (
"regexp" "regexp"
"strings" "strings"
sdkstep "git.novox.be/novox/mesh-delivery/sdk/buildstep"
dforgekey "git.novox.be/novox/mesh-delivery/sdk/forgekey" dforgekey "git.novox.be/novox/mesh-delivery/sdk/forgekey"
"git.novox.be/novox/mesh-sdk/go/forgekey" "git.novox.be/novox/mesh-sdk/go/forgekey"
) )
@@ -52,6 +53,16 @@ var (
refOK = regexp.MustCompile(`^refs/[A-Za-z0-9._/-]+$`) 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 // 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 // (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 // 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...) full := append([]string{"-c", "credential.helper=", "-C", dir}, args...)
r := run(ctx, env, "git", full...) r := run(ctx, env, "git", full...)
if r.Status != 0 || r.Err != "" { 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 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 if verb == "-C" && len(args) > 2 { // the mirror's commands are named by what they do, not by -C
verb = args[2] 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 return nil
} }
@@ -2,6 +2,7 @@ package buildstep
import ( import (
"context" "context"
sdkstep "git.novox.be/novox/mesh-delivery/sdk/buildstep"
"strings" "strings"
"testing" "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 { 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") 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. // 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 go 1.26.0
require ( 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/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/gopackageregistryclient v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/contracts/go/clients/npmpackageregistryclient 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...) full := append([]string{"-c", "credential.helper=", "-C", dir}, args...)
r := run(ctx, env, "git", full...) r := run(ctx, env, "git", full...)
if r.Status != 0 || r.Err != "" { 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 return r.Stdout, nil
} }
@@ -15,6 +15,7 @@ import (
"os" "os"
"strings" "strings"
assigner "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner"
imageclient "git.novox.be/novox/mesh-delivery/contracts/go/clients/containerimageregistryclient" imageclient "git.novox.be/novox/mesh-delivery/contracts/go/clients/containerimageregistryclient"
goclient "git.novox.be/novox/mesh-delivery/contracts/go/clients/gopackageregistryclient" goclient "git.novox.be/novox/mesh-delivery/contracts/go/clients/gopackageregistryclient"
npmclient "git.novox.be/novox/mesh-delivery/contracts/go/clients/npmpackageregistryclient" 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" goseat "git.novox.be/novox/mesh-delivery/contracts/go/seats/gopackageregistry"
npmseat "git.novox.be/novox/mesh-delivery/contracts/go/seats/npmpackageregistry" npmseat "git.novox.be/novox/mesh-delivery/contracts/go/seats/npmpackageregistry"
ociseat "git.novox.be/novox/mesh-delivery/contracts/go/seats/ociartifactregistry" 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" "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 // 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. // (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) { 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 { switch seat {
case ImageSeat: case ImageSeat:
s, err := imageclient.Publish(ctx, imageclient.Call(call), in, ps) s, err := imageclient.Publish(ctx, imageclient.Call(call), in, ps)
+51 -2
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 // 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. // 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 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 // AskSource is where a puller takes asks from: the seat's workers on the bus (JetStreamAsks) or a test double
@@ -65,6 +68,9 @@ type Puller struct {
Run func(ctx context.Context, ask sdkstep.StepAsk) (any, error) Run func(ctx context.Context, ask sdkstep.StepAsk) (any, error)
// PullWait and Heartbeat default to DefaultPullWait and DefaultHeartbeat. // 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 Heartbeat time.Duration
Logf func(format string, args ...any) 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 // 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 // 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. // pause never ends by itself while the steps state drops its old keys. Put when nil.
PausePut Put 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 // 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 // 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. // double. Until it hands a value the puller judges by Capacity.
@@ -109,6 +123,8 @@ type Puller struct {
mu sync.Mutex mu sync.Mutex
inHand int inHand int
running map[string]RunningAsk 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 paused bool
wake chan struct{} wake chan struct{}
once sync.Once 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. // 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.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 p.follow() // WatchConfigServed waits for Serve, which a module starts after its pull
go p.watchSettings(ctx) go p.watchSettings(ctx)
wait := p.PullWait wait := p.PullWait
@@ -612,11 +630,15 @@ func (p *Puller) take(ctx context.Context, msg *Pulled) {
p.refuseUnread(msg, err) p.refuseUnread(msg, err)
return 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 { if lacks := sdkstep.Lacks(ask.Needs, p.Has); len(lacks) > 0 {
p.decline(msg, ask, lacks) p.decline(msg, ask, lacks)
return 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 { if err != nil {
p.logf("the ask %s cannot start on %s: %v", ask.ID, p.Node, err) p.logf("the ask %s cannot start on %s: %v", ask.ID, p.Node, err)
_ = msg.Nak() _ = 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)} 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) return p.Run(context.WithValue(ctx, phaseKey{}, ph), ask)
}) })
step = ph.end() // its end keeps the phase it ended in step = ph.end() // its end keeps the phase it ended in
close(done) 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 var dec *Declining
if errors.As(err, &dec) && ctx.Err() == nil { if errors.As(err, &dec) && ctx.Err() == nil {
p.declineRun(msg, ask, key, step, dec) 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. // The process stops: the ask is neither ended nor acknowledged here, so the stream gives it to a node that runs.
return 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 { 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 // 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). // 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 delivered = meta.NumDelivered
} }
return &Pulled{Data: append([]byte(nil), msg.Data()...), Delivered: delivered, 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(): case <-ctx.Done():
return nil, ctx.Err() 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)
}
}
}
+10 -1
View File
@@ -16,6 +16,9 @@ type FakeAsks struct {
added chan struct{} added chan struct{}
// Acked, Termed and Naked are the asks' ids or data by fate, InProgress how many heartbeats were given. // Acked, Termed and Naked are the asks' ids or data by fate, InProgress how many heartbeats were given.
Acked, Termed, Naked []string Acked, Termed, Naked []string
// 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 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, when set, is the delivery count every ask handed over says (1 when 0): a test of a poisoned ask sets it.
Delivered uint64 Delivered uint64
@@ -108,6 +111,12 @@ func (f *FakeAsks) pulled(data []byte) *Pulled {
if delivered == 0 { if delivered == 0 {
delivered = 1 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 }} InProgress: func() error { f.mu.Lock(); f.InProgress++; f.mu.Unlock(); return nil }}
} }
@@ -4,6 +4,7 @@ package buildstep
import ( import (
"context" "context"
"errors"
"strings" "strings"
"sync" "sync"
"testing" "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) } 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 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) Pull func(worker string, wait time.Duration) (*stdio.Pulled, error)
Settle func(id, fate string) 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. // 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 { if err != nil || p == nil {
return nil, err 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) } } 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), 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 ( import (
"context" "context"
"encoding/json"
"fmt"
"strings"
"sync"
"testing" "testing"
"time" "time"
nodebuildgo "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo"
sdkstep "git.novox.be/novox/mesh-delivery/sdk/buildstep" sdkstep "git.novox.be/novox/mesh-delivery/sdk/buildstep"
stdio "git.novox.be/novox/mesh-sdk/go" stdio "git.novox.be/novox/mesh-sdk/go"
) )
@@ -38,3 +43,106 @@ func TestSeatAsksPullsTheKindsWorkerAndSettlesByID(t *testing.T) {
t.Errorf("settled %v", settled) 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 { if acked, _, naked := src.Fates(); len(acked)+len(naked) != 0 {
t.Errorf("acked %v naked %v, want it termed alone", acked, naked) 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 // 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 } 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, return map[string]any{"node": p.Node, "capacity": c.Effective, "capacity_why": c.Why, "capacity_detail": c,
"running": p.Running(), "paused": p.Paused()}, nil "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, "+ 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, "+ "bounded by the cores its node's build slice gives over the cores a step needs) and why, "+
"and whether it is paused.", current), "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). // 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) { 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) key, err := sdkstep.StepKey(node, step.ID)
if err != nil { if err != nil {
return sdkstep.Step{}, "", err 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. // the step's ask is not acknowledged as done.
func finish(put Put, key string, step sdkstep.Step, result any, err error, logf func(format string, args ...any)) error { func finish(put Put, key string, step sdkstep.Step, result any, err error, logf func(format string, args ...any)) error {
step.Ended = now().UTC() step.Ended = now().UTC()
switch ref := (*Refusal)(nil); { switch ref, drop := (*Refusal)(nil), (*Dropped)(nil); {
case err == nil: case err == nil:
step.State = sdkstep.StepPassed step.State = sdkstep.StepPassed
if result != nil { if result != nil {
@@ -79,10 +84,13 @@ func finish(put Put, key string, step sdkstep.Step, result any, err error, logf
step.Result = raw step.Result = raw
} }
} }
case errors.As(err, &drop):
step.State, step.Reason = sdkstep.StepDropped, err.Error()
case errors.As(err, &ref): case errors.As(err, &ref):
step.State, step.Reason = sdkstep.StepRefused, err.Error() step.State, step.Reason = sdkstep.StepRefused, err.Error()
default: default:
step.State, step.Reason = sdkstep.StepFailed, err.Error() 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. // 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 result != nil {
if raw, merr := json.Marshal(result); merr == 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 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. // writeEnd writes the step's end, trying EndAttempts times.
func writeEnd(put Put, key string, step sdkstep.Step) error { func writeEnd(put Put, key string, step sdkstep.Step) error {
for attempt := 1; ; attempt++ { 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" "strings"
"time" "time"
"git.novox.be/novox/mesh-catalog/modules/mesh-build-go/buildstep"
"git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/catalogue"
"github.com/therootcompany/xz" "github.com/therootcompany/xz"
) )
@@ -145,7 +147,7 @@ func (m *Module) pinnedFile(ctx context.Context, workspace string, art catalogue
} }
response, err := client.Do(request) response, err := client.Do(request)
if err != nil { 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() defer response.Body.Close()
if response.StatusCode != http.StatusOK { if response.StatusCode != http.StatusOK {
@@ -162,7 +164,7 @@ func (m *Module) pinnedFile(ctx context.Context, workspace string, art catalogue
err = cerr err = cerr
} }
if err != nil { 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 { if n > maxUpstream {
return "", fmt.Errorf("%s is larger than %d bytes, more than a machine unpacks; nothing was kept", art.URL, 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" "context"
"crypto/sha256" "crypto/sha256"
"encoding/hex" "encoding/hex"
"errors"
"io" "io"
"net" "net"
"net/http" "net/http"
@@ -223,8 +224,17 @@ func TestAPinnedFileOfAnotherDigestIsRefused(t *testing.T) {
func TestAPinnedFileTheServerRefusesFailsTheStep(t *testing.T) { func TestAPinnedFileTheServerRefusesFailsTheStep(t *testing.T) {
m, states := pinnedModule(t, "/gone", releaseSHA256, `,"file":"x"`) m, states := pinnedModule(t, "/gone", releaseSHA256, `,"file":"x"`)
if _, st := runBuild(t, m, states, "files"); st.State != sdkstep.StepFailed || !strings.Contains(st.Reason, "404") { 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", st.State, st.Reason) 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 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, 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) { Run: func(ctx context.Context, a sdkstep.StepAsk) (any, error) {
if err := m.Check(ctx); err != nil { 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. // 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) { func TestManifestClaimsTheGoBuildSeatAndReadsItsCapacity(t *testing.T) {
m := readManifest(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) t.Fatalf("claims = %+v", m.Claims)
} }
if !slices.Contains(m.Reads, KeptState) { 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) return BuildResult{}, fmt.Errorf("packing %s: %w", artifact, err)
} }
if err := m.publishArchive(ctx, name, artifact, kind, digest, archive); err != nil { 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 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 return res, nil
} }
if err := m.publishOn(ctx, seat, path, "package", version, digest, transferName(path, version, digest), body); err != 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 res.Reference = path + "@" + version
return res, nil return res, nil
@@ -2,6 +2,10 @@ package main
import ( import (
"context" "context"
"encoding/json"
"git.novox.be/novox/mesh-catalog/modules/mesh-build-go/buildstep"
"git.novox.be/novox/mesh-sdk/go/forgekey"
"strings"
"testing" "testing"
"time" "time"
@@ -32,3 +36,23 @@ func TestTheStartSweepEndsWhatAPreviousProcessLeft(t *testing.T) {
(&Module{Node: node}).endLeftRunning(ctx, scan, before, logf) (&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)), ReadSteps: readOf(stdio.State(seat.StateSteps)),
Pauses: stdio.State(sdkstep.PausesState).Put, Pauses: stdio.State(sdkstep.PausesState).Put,
ReadPauses: readOf(stdio.State(sdkstep.PausesState)), ReadPauses: readOf(stdio.State(sdkstep.PausesState)),
ForgetPause: stdio.State(sdkstep.PausesState).Delete,
WatchPauses: stdio.State(sdkstep.PausesState).Watch,
ReadToolchainsState: readOf(stdio.State(seat.StateToolchains)), ReadToolchainsState: readOf(stdio.State(seat.StateToolchains)),
Ctx: ctx, Ctx: ctx,
Say: logf, Say: logf,
@@ -4,6 +4,7 @@ import (
"context" "context"
"encoding/json" "encoding/json"
"fmt" "fmt"
stdio "git.novox.be/novox/mesh-sdk/go"
"net/http" "net/http"
"sync" "sync"
"time" "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 is the Put of the pauses state, which keeps the node's pause key without an age; ReadPauses reads it back.
Pauses buildstep.Put Pauses buildstep.Put
ReadPauses StateRead 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 // 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). // the toolchains key, so the indexer, which holds no timer, learns that the bucket exists (issue 1444).
Emit func(eventType string, body any) error 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 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 Ctx context.Context
Say func(format string, a ...any) 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, // 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. // both outside the account's home (ADR 0466 §1); Home is the account's home.
Run forgekey.Runner Run forgekey.Runner
@@ -4,6 +4,7 @@ import (
"context" "context"
"encoding/json" "encoding/json"
"errors" "errors"
"os"
"strings" "strings"
"sync" "sync"
"testing" "testing"
@@ -198,3 +199,15 @@ func TestParseGoVersion(t *testing.T) {
t.Errorf("accepted a text that is not `go version`'s") 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 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") 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 != "" { 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). // The forge did not answer: its own cause, never held as a refusal until the module changes (issue 1748); a network's
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)) // 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 := "", "" commit, ref := "", ""
for _, l := range strings.Split(r.Stdout, "\n") { for _, l := range strings.Split(r.Stdout, "\n") {
@@ -14,6 +14,7 @@ import (
"sync" "sync"
"testing" "testing"
"git.novox.be/novox/mesh-catalog/modules/mesh-build-go/buildstep"
registry "git.novox.be/novox/mesh-delivery/contracts/go/seats/containerimageregistry" registry "git.novox.be/novox/mesh-delivery/contracts/go/seats/containerimageregistry"
builds "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds" builds "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
sdkstep "git.novox.be/novox/mesh-delivery/sdk/buildstep" 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") { !strings.Contains(st.Reason, "oci-artifact-registry") {
t.Errorf("no holder: the step ended %s: %q", st.State, st.Reason) 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 // 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" "strconv"
"strings" "strings"
"sync" "sync"
"time"
sdkstep "git.novox.be/novox/mesh-delivery/sdk/buildstep" sdkstep "git.novox.be/novox/mesh-delivery/sdk/buildstep"
"git.novox.be/novox/mesh-sdk/go/forgekey" "git.novox.be/novox/mesh-sdk/go/forgekey"
@@ -97,37 +96,14 @@ func saidSuffix(said string) string {
return ": " + said return ": " + said
} }
// downloadAttempts bounds the downloads of one asked version, and downloadBackoff is the wait after the first failed one // downloadFailure is a failed download said in go's own words; one whose words are a network's (a lookup, a timeout, a reset
// (twice that after the second): a network failure is taken again, any other fails at once. // connection) is a sdkstep.Passing, so the build module gives the ask again after a delay instead of failing it (ADR 0664 §1).
const ( func downloadFailure(failure, format string, args ...any) error {
downloadAttempts = 3 err := fmt.Errorf("%s: %s", fmt.Sprintf(format, args...), causeLine(failure))
downloadBackoff = 3 * time.Second if sdkstep.PassingCause(failure) {
) return &sdkstep.Passing{Err: err}
// 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()
} }
return err
} }
// takeVersion takes one asked version into dir's modfile (takeVersions). // 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) tagged := sdkstep.PackageSemver.MatchString(tv.Asked)
var r forgekey.Ran var r forgekey.Ran
var info downloadInfo 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) r = m.Run(ctx, m.goEnv(), "go", "-C", pinsDir, "mod", "download", "-json", tv.Package+"@"+tv.Asked)
info = downloadInfo{}
_ = json.Unmarshal([]byte(r.Stdout), &info) _ = json.Unmarshal([]byte(r.Stdout), &info)
failed = r.Status != 0 || r.Err != "" || info.Version == "" || info.Error != "" 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)
}
if failed && tagged { if failed && tagged {
failure := strings.TrimSpace(info.Error + "\n" + r.Err + "\n" + r.Stderr) 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 { 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. // 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() 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 // 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). // 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 { if err == nil {
err = m.replaceInModfile(dir, tv.Package, where) err = m.replaceInModfile(dir, tv.Package, where)
@@ -350,7 +306,7 @@ func comparePre(a, b string) int {
func unknownRevision(failure string) bool { 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 // 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. // no such revision, whatever else the text says.
if networkFailure(failure) { if sdkstep.PassingCause(failure) {
return false return false
} }
return strings.Contains(failure, "unknown revision") || strings.Contains(failure, "invalid version") return strings.Contains(failure, "unknown revision") || strings.Contains(failure, "invalid version")
@@ -7,7 +7,6 @@ import (
"path/filepath" "path/filepath"
"strings" "strings"
"testing" "testing"
"time"
builds "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds" builds "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
sdkstep "git.novox.be/novox/mesh-delivery/sdk/buildstep" sdkstep "git.novox.be/novox/mesh-delivery/sdk/buildstep"
@@ -29,7 +28,8 @@ func TestAnAskedVersionIsWrittenIntoTheModfileWhateverGoModNames(t *testing.T) {
built, said, edit string built, said, edit string
failsWith string // a download failure that is not an unknown revision: the step fails with go's own words 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 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, {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="}`, answer: `{"Path":"` + sdk + `","Version":"v0.1.15-0.20261101000000-aaaaaaaaaaaa","Sum":"h1:zip=","GoModSum":"h1:mod="}`,
@@ -47,15 +47,15 @@ func TestAnAskedVersionIsWrittenIntoTheModfileWhateverGoModNames(t *testing.T) {
{name: "a head go cannot even look up is placed", asked: commitSHA, failing: true, {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"}`, 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"}, 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"}`}, 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 // 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". // 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"}`}, 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). // 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, {name: "a forge that cannot be listed is retaken", asked: commitSHA, failing: true, forgeDown: true,
failsWith: "unable to access the forge", failsWith: "unable to access the forge", passing: true,
answer: `{"Path":"` + sdk + `","Error":"invalid version: unknown revision ` + pinShort + `"}`}, 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 // 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. // whatever go.mod names; one the Go seat lacks fails naming it, never looked for among the pull request heads.
@@ -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"}, 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", {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"}`}, 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. // 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
{name: "a timeout twice is taken again", asked: trunk, timeouts: 2, downloads: 3, // delay, no retry here; another failure is not one.
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"}`},
{name: "an unknown revision is not taken again", asked: offHead, failing: true, failsWith: "on no pull request head", downloads: 1, {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 + `"}`}, 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"}`}, 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) { t.Run(c.name, func(t *testing.T) {
h, states := newPinHarness(t, pinnedGoMod, false, pullList) h, states := newPinHarness(t, pinnedGoMod, false, pullList)
h.f.gomods = "go.mod\nother/go.mod\n" 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 downloads := 0
base := h.m.Run base := h.m.Run
h.m.Run = func(ctx context.Context, env []string, name string, args ...string) forgekey.Ran { 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.cmds = append(h.cmds, cmdRecord{env: env, line: line})
h.mu.Unlock() h.mu.Unlock()
downloads++ 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 { if c.failing {
return forgekey.Ran{Status: 1, Stdout: c.answer} return forgekey.Ran{Status: 1, Stdout: c.answer}
} }
@@ -122,18 +113,18 @@ func TestAnAskedVersionIsWrittenIntoTheModfileWhateverGoModNames(t *testing.T) {
t.Fatal(err) t.Fatal(err)
} }
st := states.waitEnd(t) st := states.waitEnd(t)
if c.downloads != 0 && (downloads != c.downloads || len(paused) != c.downloads-1) { if c.downloads != 0 && downloads != c.downloads {
t.Errorf("go downloaded %d times with %d waits, want %d downloads", downloads, len(paused), 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 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) t.Errorf("the unknown revision ended %s: %q", st.State, st.Reason)
} }
return return
} }
if c.failsWith != "" { if c.failsWith != "" {
if st.State != sdkstep.StepFailed || !strings.Contains(st.Reason, c.failsWith) || strings.Contains(st.Reason, "no pull request head") || 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) t.Errorf("the download failure ended %s: %q", st.State, st.Reason)
} }
return return
+1
View File
@@ -102,6 +102,7 @@ replace git.novox.be/novox/mesh-catalog/modules/ssh-client/sshkeys => ../ssh-cli
require ( 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/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/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/npmpackageregistryclient v0.0.0-00010101000000-000000000000 // indirect
git.novox.be/novox/mesh-delivery/contracts/go/clients/ociartifactregistryclient 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", "runs-as": "mesh-build-go",
"capabilities": [ "capabilities": [
"package-manager", "package-manager",
"seat-settle-delay",
"user-namespaces" "user-namespaces"
], ],
"state": [ "state": [
@@ -20,7 +21,7 @@
{ {
"name": "node-build-go", "name": "node-build-go",
"scope": "node", "scope": "node",
"serves": ["current", "pause", "resume"] "serves": ["current", "drop", "pause", "resume"]
} }
], ],
"consumes": [ "consumes": [
@@ -35,7 +36,8 @@
"seat:oci-artifact-registry.publish-part", "seat:oci-artifact-registry.publish-part",
"seat:go-package-registry.publish", "seat:go-package-registry.publish",
"seat:go-package-registry.publish-part", "seat:go-package-registry.publish-part",
"seat:mesh-delivery-checker.input" "seat:mesh-delivery-checker.input",
"seat:mesh-module-assigner.assignments"
], ],
"config": { "config": {
"capacity": { "capacity": {
@@ -62,7 +62,7 @@ func (m *Module) Puller(steps []StepTool, source buildstep.AskSource) *buildstep
if m.ReadPauses.Get != nil { if m.ReadPauses.Get != nil {
load = func(key string) ([]byte, error) { return m.ReadPauses.Get(key) } 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, 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) { Run: func(ctx context.Context, a sdkstep.StepAsk) (any, error) {
if err := m.Check(ctx); err != nil { 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) argv = append(argv, contextDir)
if r := m.Run(ctx, m.env(), "buildah", argv...); r.Status != 0 || r.Err != "" { 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) 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) defer m.remove(local)
digest, err := m.localDigest(ctx, local) digest, err := m.localDigest(ctx, local)
@@ -178,6 +178,8 @@ func TestBuildFailsWithBuildahsError(t *testing.T) {
[]string{"Error: the end"}, []string{"early noise", long}}, []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}, "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}, "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 { for name, c := range cases {
m, f, states := setup(t, same) m, f, states := setup(t, same)
@@ -200,6 +202,10 @@ func TestBuildFailsWithBuildahsError(t *testing.T) {
if len(st.Reason) > CauseBytes+200 { if len(st.Reason) > CauseBytes+200 {
t.Errorf("%s: the reason is %d bytes, more than its bound", name, len(st.Reason)) 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 { if len(f.called("push")) != 0 {
t.Errorf("%s: a failed build pushed", name) t.Errorf("%s: a failed build pushed", name)
} }
@@ -24,7 +24,7 @@ func TestTheModuleInvokesTheImageSeatAndHoldsNoRegistry(t *testing.T) {
t.Fatal(err) t.Fatal(err)
} }
if want := []string{"seat:container-image-registry.publish", "seat:container-image-registry.publish-part", 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) 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. // 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}) raw, err := m.call()("seat:"+buildstep.ImageSeat+"."+reg.VerbMirror, reg.MirrorInput{Name: module + "/" + name, Source: from, Digest: digest})
if err != nil { if err != nil {
said.Refused(err) 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 var answer reg.MirrorAnswer
_ = json.Unmarshal(raw, &answer) _ = json.Unmarshal(raw, &answer)
@@ -2,6 +2,11 @@ package main
import ( import (
"context" "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" "testing"
"time" "time"
@@ -17,3 +22,25 @@ func TestTheStartSweepEndsWhatAPreviousProcessLeft(t *testing.T) {
(&Module{Node: node}).endLeftRunning(ctx, scan, before, logf) (&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)
}
})
}
@@ -37,6 +37,8 @@ func main() {
ReadSteps: readOf(stdio.State(seat.StateSteps)), ReadSteps: readOf(stdio.State(seat.StateSteps)),
Pauses: stdio.State(sdkstep.PausesState).Put, Pauses: stdio.State(sdkstep.PausesState).Put,
ReadPauses: readOf(stdio.State(sdkstep.PausesState)), ReadPauses: readOf(stdio.State(sdkstep.PausesState)),
ForgetPause: stdio.State(sdkstep.PausesState).Delete,
WatchPauses: stdio.State(sdkstep.PausesState).Watch,
Ctx: ctx, Ctx: ctx,
Say: logf, Say: logf,
Work: WorkDir, Work: WorkDir,
@@ -72,7 +72,7 @@ func TestManifestDeclaresTheAccountTheCapabilityAndThePackages(t *testing.T) {
if home["type"] != "directory" || home["mode"] != "0700" || home["owner"] != "mesh-build-image" || home["place"] != "." { 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) 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) { if !slices.Contains(m.Capabilities, c) {
t.Errorf("capabilities = %v, want %s", 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) { if want := []string{"keys", MirrorWaitsState, "pauses", "steps"}; !slices.Equal(states, want) {
t.Errorf("state = %v, want %v", 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) t.Errorf("claims = %v, want the seat %s alone", m.Claims, Seat)
} }
raw, _ := os.ReadFile(filepath.Join(moduleDir, "module.json")) raw, _ := os.ReadFile(filepath.Join(moduleDir, "module.json"))
@@ -4,6 +4,7 @@ import (
"context" "context"
"errors" "errors"
"fmt" "fmt"
stdio "git.novox.be/novox/mesh-sdk/go"
"sync" "sync"
"time" "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 is the Put of the pauses state, which keeps the node's pause key without an age; ReadPauses reads it back.
Pauses buildstep.Put Pauses buildstep.Put
ReadPauses StateRead 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 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 Ctx context.Context
Say func(format string, a ...any) 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)} Transfer: buildstep.TransferName(name, q, img.Digest)}
sent, err := buildstep.PublishFiles(ctx, m.call(), buildstep.ImageSeat, in, img.Files) sent, err := buildstep.PublishFiles(ctx, m.call(), buildstep.ImageSeat, in, img.Files)
if err != nil { 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)) 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 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 replace git.novox.be/novox/mesh-catalog/modules/ssh-client/sshkeys => ../ssh-client/sshkeys
require ( 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/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/meshcatalogclient 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/npmpackageregistryclient v0.0.0-00010101000000-000000000000 // indirect
+4 -2
View File
@@ -9,6 +9,7 @@
"runs-as": "mesh-build-image", "runs-as": "mesh-build-image",
"capabilities": [ "capabilities": [
"package-manager", "package-manager",
"seat-settle-delay",
"user-namespaces" "user-namespaces"
], ],
"state": [ "state": [
@@ -20,7 +21,7 @@
{ {
"name": "node-build-image", "name": "node-build-image",
"scope": "node", "scope": "node",
"serves": ["current", "pause", "resume"] "serves": ["current", "drop", "pause", "resume"]
} }
], ],
"config": { "config": {
@@ -63,7 +64,8 @@
"invokes": [ "invokes": [
"seat:container-image-registry.publish", "seat:container-image-registry.publish",
"seat:container-image-registry.publish-part", "seat:container-image-registry.publish-part",
"seat:container-image-registry.mirror" "seat:container-image-registry.mirror",
"seat:mesh-module-assigner.assignments"
], ],
"consumes": [ "consumes": [
"mesh-delivery-builder.version.numbered", "mesh-delivery-builder.version.numbered",
@@ -72,7 +72,7 @@ func (m *Module) Puller(steps []StepTool, source buildstep.AskSource) *buildstep
if m.ReadPauses.Get != nil { if m.ReadPauses.Get != nil {
load = func(key string) ([]byte, error) { return m.ReadPauses.Get(key) } 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, Capacity: capacity, SettingsFile: settings, WatchConfig: watch, CoresPerStep: cores, BuildHours: hours, Logf: m.Say,
Run: func(ctx context.Context, a sdkstep.StepAsk) (any, error) { Run: func(ctx context.Context, a sdkstep.StepAsk) (any, error) {
if err := m.Prepare(ctx); err != nil { 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 // 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). // 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 { 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{} files := []string{}
_ = filepath.WalkDir(bundleDir, func(p string, d os.DirEntry, err error) error { _ = 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) "--no-audit", "--no-fund", "--prefix", scratch, "--userconfig", npmrc)
os.Remove(npmrc) os.Remove(npmrc)
if why := ranFailed(r); why != "" { 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") nm := filepath.Join(scratch, "node_modules")
if err := removeSDKCopies(nm); err != nil { if err := removeSDKCopies(nm); err != nil {
@@ -2,6 +2,8 @@ package main
import ( import (
"context" "context"
"git.novox.be/novox/mesh-catalog/modules/mesh-build-go/buildstep"
"strings"
"testing" "testing"
"time" "time"
@@ -30,3 +32,20 @@ func TestTheStartSweepEndsWhatAPreviousProcessLeft(t *testing.T) {
(&Module{Node: node}).endLeftRunning(ctx, scan, before, logf) (&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)), ReadSteps: readOf(stdio.State(seat.StateSteps)),
Pauses: stdio.State(sdkstep.PausesState).Put, Pauses: stdio.State(sdkstep.PausesState).Put,
ReadPauses: readOf(stdio.State(sdkstep.PausesState)), ReadPauses: readOf(stdio.State(sdkstep.PausesState)),
ForgetPause: stdio.State(sdkstep.PausesState).Delete,
WatchPauses: stdio.State(sdkstep.PausesState).Watch,
ReadToolchainsState: readOf(stdio.State(seat.StateToolchains)), ReadToolchainsState: readOf(stdio.State(seat.StateToolchains)),
Work: WorkDir, Work: WorkDir,
Run: dforgekey.ExecRunnerClean(0), 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"] != "." { 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) 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") { if !slices.Contains(m.Capabilities, "user-namespaces") {
t.Errorf("capabilities = %v, want user-namespaces", m.Capabilities) 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) { if want := []string{"keys", "pauses", "steps", "toolchains"}; !slices.Equal(states, want) {
t.Errorf("state = %v, want %v", 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) t.Errorf("claims = %v, want the seat %s alone", m.Claims, Seat)
} }
if !slices.Contains(m.Tools, HealthTool) { if !slices.Contains(m.Tools, HealthTool) {
@@ -3,6 +3,7 @@ package main
import ( import (
"context" "context"
"fmt" "fmt"
stdio "git.novox.be/novox/mesh-sdk/go"
"net/http" "net/http"
"os" "os"
"sync" "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 is the Put of the pauses state, which keeps the node's pause key without an age; ReadPauses reads it back.
Pauses buildstep.Put Pauses buildstep.Put
ReadPauses StateRead 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 // 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). // the toolchains key, so the indexer, which holds no timer, learns that the bucket exists (issue 1444).
Emit func(eventType string, body any) error 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") 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()) 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) 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 != "" { 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)) 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 return res, nil
} }
if err := m.publishOn(ctx, seat, p.Name, "package", p.Version, p.Digest, p.tgz); err != 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 res.Reference = p.Name + "@" + p.Version
return res, nil return res, nil
@@ -17,6 +17,7 @@ import (
"git.novox.be/novox/mesh-catalog/modules/mesh-build-go/buildstep" "git.novox.be/novox/mesh-catalog/modules/mesh-build-go/buildstep"
builds "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds" 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. // 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)) { 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) 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 var archives []string
filepath.WalkDir(work, func(p string, d os.DirEntry, err error) error { filepath.WalkDir(work, func(p string, d os.DirEntry, err error) error {
if err == nil && strings.HasSuffix(p, ".tar.gz") { if err == nil && strings.HasSuffix(p, ".tar.gz") {
@@ -10,6 +10,7 @@ package main
import ( import (
"context" "context"
"fmt" "fmt"
"git.novox.be/novox/mesh-catalog/modules/mesh-build-go/buildstep"
"io/fs" "io/fs"
"os" "os"
"path/filepath" "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) "npm", "ci", "--ignore-scripts", "--no-bin-links", "--no-audit", "--no-fund", "--prefix", dir, "--userconfig", npmrc)
os.Remove(npmrc) os.Remove(npmrc)
if ran.Err != "" || ran.Status != 0 { if ran.Err != "" || ran.Status != 0 {
return fmt.Errorf("npm ci --ignore-scripts for the toolchain %s failed (status %d %s): %s", 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)) version, ran.Status, ran.Err, tail(ran.Stderr)))
} }
if err := os.WriteFile(filepath.Join(dir, installedMark), []byte(version+"\n"), 0o600); err != nil { if err := os.WriteFile(filepath.Join(dir, installedMark), []byte(version+"\n"), 0o600); err != nil {
return err return err
@@ -3,6 +3,7 @@ package main
import ( import (
"context" "context"
"errors" "errors"
sdkstep "git.novox.be/novox/mesh-delivery/sdk/buildstep"
"os" "os"
"path/filepath" "path/filepath"
"strings" "strings"
@@ -86,6 +87,18 @@ func TestAMissingToolchainDirectoryIsInstalledOnceFromTheLock(t *testing.T) {
if err := tc.Ensure(context.Background(), version); err != nil || len(fake.calls) != 1 { 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) 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 // 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 replace git.novox.be/novox/mesh-catalog/modules/ssh-client/sshkeys => ../ssh-client/sshkeys
require ( 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/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/meshcatalogclient 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/npmpackageregistryclient v0.0.0-00010101000000-000000000000 // indirect
+4 -2
View File
@@ -9,6 +9,7 @@
"runs-as": "mesh-build-node", "runs-as": "mesh-build-node",
"capabilities": [ "capabilities": [
"package-manager", "package-manager",
"seat-settle-delay",
"user-namespaces" "user-namespaces"
], ],
"state": [ "state": [
@@ -20,7 +21,7 @@
{ {
"name": "node-build-node", "name": "node-build-node",
"scope": "node", "scope": "node",
"serves": ["current", "pause", "resume"] "serves": ["current", "drop", "pause", "resume"]
} }
], ],
"config": { "config": {
@@ -76,7 +77,8 @@
"seat:oci-artifact-registry.publish", "seat:oci-artifact-registry.publish",
"seat:oci-artifact-registry.publish-part", "seat:oci-artifact-registry.publish-part",
"seat:npm-package-registry.publish", "seat:npm-package-registry.publish",
"seat:npm-package-registry.publish-part" "seat:npm-package-registry.publish-part",
"seat:mesh-module-assigner.assignments"
], ],
"own-secrets": { "own-secrets": {
"broker": "${dir:state}/broker" "broker": "${dir:state}/broker"