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