mesh-delivery-sdk: the runner's stale-step part moves to package buildrun, so buildstep (layouts, keys, codecs) and the builder's published builds package reach no bus; a test each refuses it (B6e finding 5; ADR 0634 §1; issue 1966)

Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
This commit is contained in:
2026-11-05 15:50:31 +01:00
parent eaa84387f7
commit 840a7c5b07
5 changed files with 169 additions and 58 deletions
@@ -0,0 +1,21 @@
package builds
import (
"os/exec"
"strings"
"testing"
)
// The builder's published value is read by modules that never touch the bus (mesh-host's composer tool): nothing it
// imports reaches the bus library.
func TestThePackageReachesNoBus(t *testing.T) {
out, err := exec.Command("go", "list", "-deps", ".").Output()
if err != nil {
t.Fatalf("go list: %v", err)
}
for _, p := range strings.Fields(string(out)) {
if strings.HasPrefix(p, "github.com/nats-io/") {
t.Errorf("builds depends on %s", p)
}
}
}
+52
View File
@@ -0,0 +1,52 @@
package buildrun
import (
"encoding/json"
"sync"
"testing"
"time"
"git.novox.be/novox/mesh-delivery/sdk/buildstep"
)
// keys is a steps state in memory: every value put, in order, per key, and a signal for each end.
type keys struct {
mu sync.Mutex
puts map[string][]buildstep.Step
ended chan string
}
func newKeys() *keys { return &keys{puts: map[string][]buildstep.Step{}, ended: make(chan string, 16)} }
func (k *keys) put(key string, value any) (uint64, error) {
step, err := buildstep.DecodeStep(value.(json.RawMessage))
if err != nil {
return 0, err
}
k.mu.Lock()
k.puts[key] = append(k.puts[key], step)
n := uint64(len(k.puts[key]))
k.mu.Unlock()
if step.State != buildstep.StepRunning {
k.ended <- key
}
return n, nil
}
func (k *keys) last(key string) buildstep.Step {
k.mu.Lock()
defer k.mu.Unlock()
s := k.puts[key]
return s[len(s)-1]
}
func (k *keys) waitEnd(t *testing.T) string {
t.Helper()
select {
case key := <-k.ended:
return key
case <-time.After(3 * time.Second):
t.Fatal("the step never ended on its key")
return ""
}
}
@@ -1,10 +1,11 @@
package buildstep
package buildrun
import (
"context"
"encoding/json"
"errors"
"fmt"
"git.novox.be/novox/mesh-delivery/sdk/buildstep"
"os"
"strconv"
"strings"
@@ -20,7 +21,7 @@ const RestartReason = "the build module restarted while it ran"
// for this process's, which may still run it, so EndOne refuses it.
var Boot = time.Now()
// Owner is this process as a step's owner (Step.Owner): `<boot id>.<pid>.<start>`, the node's boot id
// Owner is this process as a step's owner (buildstep.Step.Owner): `<boot id>.<pid>.<start>`, the node's boot id
// (/proc/sys/kernel/random/boot_id), the pid and Boot in milliseconds, base 36. The start keeps two processes apart
// that a reboot or a pid namespace gave the same pid. A running step whose owner is another is a previous process's.
var Owner = ownerOf(bootID, os.Getpid(), Boot)
@@ -84,15 +85,15 @@ func ScanOf(s stdio.KeptState) Scan {
// has the tool's name). An ask's key is the stream's to settle, never a restart's: the puller leaves an ask it was
// running unacknowledged so the stream redelivers it, and the asker's wait takes the first end it reads on the key.
// A key a gone process owns is deleted, never ended (EndStale); one with no owner is ended only once the asker's bound
// has passed (StepBound), when it is nobody's.
func isAsk(key string, s Step) bool {
_, _, err := ParseAskStepKey(key)
// has passed (buildstep.StepBound), when it is nobody's.
func isAsk(key string, s buildstep.Step) bool {
_, _, err := buildstep.ParseAskStepKey(key)
return err == nil && strings.HasSuffix(s.Tool, " ask")
}
// AskWhat says what an ask is, in one line: its kind, and its repository, head and levels, or its module and
// fingerprint.
func AskWhat(a Ask) string {
func AskWhat(a buildstep.Ask) string {
short := func(s string) string {
if len(s) > 12 {
return s[:12]
@@ -122,9 +123,9 @@ func AskWhat(a Ask) string {
}
// endRunning writes step ended failed with why, only while its key is still at revision.
func endRunning(scan Scan, key string, step Step, revision uint64, why string) error {
step.State, step.Reason, step.Ended, step.Result = StepFailed, why, now().UTC(), nil
raw, err := EncodeStep(step)
func endRunning(scan Scan, key string, step buildstep.Step, revision uint64, why string) error {
step.State, step.Reason, step.Ended, step.Result = buildstep.StepFailed, why, now().UTC(), nil
raw, err := buildstep.EncodeStep(step)
if err != nil {
return err
}
@@ -134,7 +135,7 @@ func endRunning(scan Scan, key string, step Step, revision uint64, why string) e
// EndStale ends this node's running steps that no process of this module runs any more, and answers their ids. Only keys
// of node are read; a value that is no step is left alone.
// - A hand-started step with an owner (Step.Owner) that is not this process (Owner) is ended failed with
// - A hand-started step with an owner (buildstep.Step.Owner) that is not this process (Owner) is ended failed with
// RestartReason at once, whatever its age: its process is gone and nothing redelivers it, so nothing else would
// write its end (issue 1890, note 21).
// - A pulled ask's step with an owner that is not this process is deleted, not ended (novox/hq ADR 0525 §6): the
@@ -146,7 +147,7 @@ func endRunning(scan Scan, key string, step Step, revision uint64, why string) e
// - A step owned by this process is left: it may run now. A redelivered ask begun on this node is owned by it.
// - A step with no owner (written by an older build) keeps the earlier rule: a hand-started step started before
// `before` (the start of this process) is ended; a pulled ask is left to the stream's redelivery (isAsk) until its
// start is older than StepBound, and then ended "past the step bound: its asker stopped waiting at <time>".
// start is older than buildstep.StepBound, and then ended "past the step bound: its asker stopped waiting at <time>".
//
// Each end and delete is logged, naming the step, its owner and why. A key written by another writer since it was read
// (a redelivered ask's start, a real end) is skipped.
@@ -157,7 +158,7 @@ func EndStale(scan Scan, node string, before time.Time, logf func(string, ...any
}
var ended []string
for _, k := range keys {
n, _, perr := ParseStepKey(k)
n, _, perr := buildstep.ParseStepKey(k)
if perr != nil || n != node {
continue
}
@@ -168,8 +169,8 @@ func EndStale(scan Scan, node string, before time.Time, logf func(string, ...any
if len(raw) == 0 {
continue
}
step, err := DecodeStep(raw)
if err != nil || step.State != StepRunning {
step, err := buildstep.DecodeStep(raw)
if err != nil || step.State != buildstep.StepRunning {
continue
}
why, said := staleWhy(k, step, before)
@@ -204,17 +205,17 @@ func EndStale(scan Scan, node string, before time.Time, logf func(string, ...any
// staleWhy is why a running step is ended (or, for an ask with an owner, deleted) by EndStale and what the log says of
// its owner, or "" to leave it.
func staleWhy(key string, step Step, before time.Time) (why, said string) {
func staleWhy(key string, step buildstep.Step, before time.Time) (why, said string) {
switch {
case step.Owner == Owner:
return "", ""
case step.Owner != "":
return RestartReason, "its owner " + step.Owner + " is not this process " + Owner
case isAsk(key, step):
if !step.Started.Add(StepBound).Before(now()) {
if !step.Started.Add(buildstep.StepBound).Before(now()) {
return "", ""
}
return "past the step bound: its asker stopped waiting at " + step.Started.Add(StepBound).UTC().Format("2006-01-02T15:04:05Z"),
return "past the step bound: its asker stopped waiting at " + step.Started.Add(buildstep.StepBound).UTC().Format("2006-01-02T15:04:05Z"),
"it names no owner (an older build wrote it)"
case step.Started.Before(before):
return RestartReason, "it names no owner (an older build wrote it) and started before this process"
@@ -225,45 +226,45 @@ func staleWhy(key string, step Step, before time.Time) (why, said string) {
// EndOne ends this node's running step id (or the key <node>.<id>) as failed with why. A step that is not running is
// refused, and so is one this process owns (Owner; for a step with no owner, one started since Boot: it may still run,
// whether a pulled ask or a step a tool started), one live reports as held, and one whose key changed since it was read.
func EndOne(scan Scan, node, id, why string, live func(id string) bool) (Step, error) {
func EndOne(scan Scan, node, id, why string, live func(id string) bool) (buildstep.Step, error) {
id = strings.TrimSpace(id)
if n, rest, ok := strings.Cut(id, "."); ok {
if n != node {
return Step{}, fmt.Errorf("the step %q is of node %q, this module runs on %q", id, n, node)
return buildstep.Step{}, fmt.Errorf("the step %q is of node %q, this module runs on %q", id, n, node)
}
id = rest
}
if strings.TrimSpace(why) == "" {
return Step{}, fmt.Errorf("say why the step is ended (why)")
return buildstep.Step{}, fmt.Errorf("say why the step is ended (why)")
}
key, err := StepKey(node, id)
key, err := buildstep.StepKey(node, id)
if err != nil {
return Step{}, err
return buildstep.Step{}, err
}
raw, rev, err := scan.Get(key)
if err != nil {
return Step{}, err
return buildstep.Step{}, err
}
if len(raw) == 0 {
return Step{}, fmt.Errorf("no step %s in the steps state", key)
return buildstep.Step{}, fmt.Errorf("no step %s in the steps state", key)
}
step, err := DecodeStep(raw)
step, err := buildstep.DecodeStep(raw)
if err != nil {
return Step{}, err
return buildstep.Step{}, err
}
if step.State != StepRunning {
return Step{}, fmt.Errorf("the step %s is %s, not running", key, step.State)
if step.State != buildstep.StepRunning {
return buildstep.Step{}, fmt.Errorf("the step %s is %s, not running", key, step.State)
}
if step.Owner == Owner || (step.Owner == "" && !step.Started.Before(Boot)) || (live != nil && live(id)) {
return Step{}, fmt.Errorf("the step %s was started by this process and may run now; only a step a previous process left is ended", key)
return buildstep.Step{}, fmt.Errorf("the step %s was started by this process and may run now; only a step a previous process left is ended", key)
}
if err := endRunning(scan, key, step, rev, why); err != nil {
if errors.Is(err, stdio.ErrChanged) {
return Step{}, fmt.Errorf("the step %s changed while it was being ended; read it again", key)
return buildstep.Step{}, fmt.Errorf("the step %s changed while it was being ended; read it again", key)
}
return Step{}, err
return buildstep.Step{}, err
}
step.State, step.Reason = StepFailed, why
step.State, step.Reason = buildstep.StepFailed, why
return step, nil
}
@@ -316,7 +317,7 @@ func EndStaleAtStart(ctx context.Context, scan Scan, node string, before time.Ti
// EndStaleRound is EndStale as the module's periodic round runs it: at once, and only when the runtime serves (a round
// before that is skipped, the start's EndStaleAtStart and the next round cover it). It ends the steps a previous
// process left and the asks past StepBound.
// process left and the asks past buildstep.StepBound.
func EndStaleRound(scan Scan, node string, logf func(string, ...any)) {
select {
case <-stdio.Served():
@@ -327,3 +328,6 @@ func EndStaleRound(scan Scan, node string, logf func(string, ...any)) {
logf("ending the steps left running: %v", err)
}
}
// now is the clock; a test sets it.
var now = time.Now
@@ -1,8 +1,9 @@
package buildstep
package buildrun
import (
"encoding/json"
"fmt"
"git.novox.be/novox/mesh-delivery/sdk/buildstep"
"strings"
"testing"
"time"
@@ -30,7 +31,7 @@ func scanOf(k *keys, race func(key string)) Scan {
if !ok {
return nil, 0, nil
}
raw, err := EncodeStep(s[len(s)-1])
raw, err := buildstep.EncodeStep(s[len(s)-1])
return raw, uint64(len(s)), err
},
Delete: func(key string, revision uint64) error {
@@ -70,25 +71,25 @@ func TestEndStaleAndEndOne(t *testing.T) {
defer func() { Boot = oldBoot }()
gone := ownerOf("b00t", 4242, boot.Add(-time.Hour))
put := func(node, id, tool, state string, started time.Time, owner string) {
key, _ := StepKey(node, id)
_, _ = k.put(key, json.RawMessage(mustStep(t, Step{ID: id, Tool: tool, Node: node, State: state, Started: started, Owner: owner})))
key, _ := buildstep.StepKey(node, id)
_, _ = k.put(key, json.RawMessage(mustStep(t, buildstep.Step{ID: id, Tool: tool, Node: node, State: state, Started: started, Owner: owner})))
}
put("ace", "old1", "build_go_fetch", StepRunning, boot.Add(-time.Hour), "")
put("ace", "mjk3l9x0abcdef", "check ask", StepRunning, boot.Add(-time.Hour), "")
put("ace", "mjk3l9x0abcdee", "check ask", StepRunning, time.Now().Add(-5*time.Minute), "")
put("ace", "mjk3l9x0abcded", "check ask", StepRunning, time.Now().Add(-5*time.Minute), gone) // note 21's ghost
put("ace", "gone2", "build_go_test", StepRunning, boot.Add(time.Second), gone)
put("ace", "mine1", "check ask", StepRunning, boot.Add(-time.Hour), Owner)
put("ace", "raced1", "build_go_test", StepRunning, boot.Add(-time.Hour), "")
put("ace", "new1", "build_go_fetch", StepRunning, boot.Add(time.Second), "")
put("ace", "done1", "build_go_fetch", StepPassed, boot.Add(-time.Hour), gone)
put("g14", "other1", "build_go_fetch", StepRunning, boot.Add(-time.Hour), gone)
put("ace", "old1", "build_go_fetch", buildstep.StepRunning, boot.Add(-time.Hour), "")
put("ace", "mjk3l9x0abcdef", "check ask", buildstep.StepRunning, boot.Add(-time.Hour), "")
put("ace", "mjk3l9x0abcdee", "check ask", buildstep.StepRunning, time.Now().Add(-5*time.Minute), "")
put("ace", "mjk3l9x0abcded", "check ask", buildstep.StepRunning, time.Now().Add(-5*time.Minute), gone) // note 21's ghost
put("ace", "gone2", "build_go_test", buildstep.StepRunning, boot.Add(time.Second), gone)
put("ace", "mine1", "check ask", buildstep.StepRunning, boot.Add(-time.Hour), Owner)
put("ace", "raced1", "build_go_test", buildstep.StepRunning, boot.Add(-time.Hour), "")
put("ace", "new1", "build_go_fetch", buildstep.StepRunning, boot.Add(time.Second), "")
put("ace", "done1", "build_go_fetch", buildstep.StepPassed, boot.Add(-time.Hour), gone)
put("g14", "other1", "build_go_fetch", buildstep.StepRunning, boot.Add(-time.Hour), gone)
// raced1 is ended by its own run between the read and the write.
raced := false
scan := scanOf(k, func(key string) {
if key == "ace.raced1" && !raced {
raced = true
put("ace", "raced1", "build_go_test", StepPassed, boot.Add(-time.Hour), "")
put("ace", "raced1", "build_go_test", buildstep.StepPassed, boot.Add(-time.Hour), "")
}
})
var said []string
@@ -100,18 +101,18 @@ func TestEndStaleAndEndOne(t *testing.T) {
t.Errorf("the ghost ask's key is still held: %+v", k.last("ace.mjk3l9x0abcded"))
}
for _, key := range []string{"ace.old1", "ace.gone2"} {
if s := k.last(key); s.State != StepFailed || s.Reason != RestartReason || s.Ended.IsZero() {
if s := k.last(key); s.State != buildstep.StepFailed || s.Reason != RestartReason || s.Ended.IsZero() {
t.Errorf("%s = %+v", key, s)
}
}
if s := k.last("ace.mjk3l9x0abcdef"); s.State != StepFailed || !strings.HasPrefix(s.Reason, "past the step bound: its asker stopped waiting at ") {
if s := k.last("ace.mjk3l9x0abcdef"); s.State != buildstep.StepFailed || !strings.HasPrefix(s.Reason, "past the step bound: its asker stopped waiting at ") {
t.Errorf("the old ask = %+v", s)
}
if log := strings.Join(said, "\n"); !strings.Contains(log, "ace.mjk3l9x0abcded") || !strings.Contains(log, "deleted, left to the stream's redelivery") || !strings.Contains(log, "its owner "+gone+" is not this process "+Owner) {
t.Errorf("the log does not name the ghost and its owner:\n%s", log)
}
for key, state := range map[string]string{"ace.mjk3l9x0abcdee": StepRunning, "ace.mine1": StepRunning, "ace.raced1": StepPassed,
"ace.new1": StepRunning, "ace.done1": StepPassed, "g14.other1": StepRunning} {
for key, state := range map[string]string{"ace.mjk3l9x0abcdee": buildstep.StepRunning, "ace.mine1": buildstep.StepRunning, "ace.raced1": buildstep.StepPassed,
"ace.new1": buildstep.StepRunning, "ace.done1": buildstep.StepPassed, "g14.other1": buildstep.StepRunning} {
if s := k.last(key); s.State != state {
t.Errorf("%s = %s, want %s", key, s.State, state)
}
@@ -123,14 +124,14 @@ func TestEndStaleAndEndOne(t *testing.T) {
t.Errorf("ended %s with why %q", c.id, c.why)
}
}
put("ace", "old2", "build_go_fetch", StepRunning, boot.Add(time.Second), gone)
put("ace", "old2", "build_go_fetch", buildstep.StepRunning, boot.Add(time.Second), gone)
if _, err := EndOne(scan, "ace", "old2", "x", func(string) bool { return true }); err == nil {
t.Error("ended a held step")
}
if _, err := EndOne(scan, "ace", "ace.old2", "stuck for good", nil); err != nil {
t.Fatal(err)
}
if s := k.last("ace.old2"); s.State != StepFailed || s.Reason != "stuck for good" || s.Owner != gone {
if s := k.last("ace.old2"); s.State != buildstep.StepFailed || s.Reason != "stuck for good" || s.Owner != gone {
t.Errorf("old2 = %+v", s)
}
if a, b := ownerOf("b", 1, boot), ownerOf("b", 1, boot.Add(time.Second)); a == b {
@@ -138,9 +139,9 @@ func TestEndStaleAndEndOne(t *testing.T) {
}
}
func mustStep(t *testing.T, s Step) []byte {
func mustStep(t *testing.T, s buildstep.Step) []byte {
t.Helper()
raw, err := EncodeStep(s)
raw, err := buildstep.EncodeStep(s)
if err != nil {
t.Fatal(err)
}
@@ -149,8 +150,8 @@ func mustStep(t *testing.T, s Step) []byte {
// 1890: AskWhat names the repository, head and levels of a check ask and the module of a build ask.
func TestAskWhat(t *testing.T) {
var check, build Ask
for raw, into := range map[string]*Ask{
var check, build buildstep.Ask
for raw, into := range map[string]*buildstep.Ask{
`{"kind":"check","check":{"fetch":{"repository":"novox/mesh-sdk","commit":"0123456789abcdef0123"},"levels":["unit","integration"]}}`: &check,
`{"kind":"delivery","build":{"module":"mesh-x","fp":"abcdef0123456789"}}`: &build,
} {
+33
View File
@@ -0,0 +1,33 @@
package buildstep
import (
"go/parser"
"go/token"
"path/filepath"
"strconv"
"strings"
"testing"
)
// The layouts, keys and codecs of this package are read by modules that never touch the bus (a composer's tool, the
// builder's published package): it imports neither the bus library nor mesh-sdk's runtime (the root package), whose
// part that runs is package buildrun's.
func TestThePackageIsFreeOfTheBus(t *testing.T) {
files, _ := filepath.Glob("*.go")
fset := token.NewFileSet()
for _, f := range files {
if strings.HasSuffix(f, "_test.go") {
continue
}
parsed, err := parser.ParseFile(fset, f, nil, parser.ImportsOnly)
if err != nil {
t.Fatal(err)
}
for _, imp := range parsed.Imports {
path, _ := strconv.Unquote(imp.Path.Value)
if strings.HasPrefix(path, "github.com/nats-io/") || path == "git.novox.be/novox/mesh-sdk/go" {
t.Errorf("%s imports %s, which reaches the bus", f, path)
}
}
}
}