Merge pull request 'A level's digest is stored as a list, and an end the state refuses is said, so a check of a repository with a package named token ends (issue 1930, with mesh-catalog and mesh-delivery)' (#74) from fix/1930-digest-shape into main

Reviewed-on: #74

go/buildstep/askrun.go

@@ -293,11 +293,7 @@ func (s AskSteps) plan(ctx context.Context, job delivery.CheckJob, level string,
 	if err != nil {
 		return nil, true, true
 	}
-	var d struct {
-		Packages map[string]string `json:"packages"`
-		Beside   []string          `json:"beside"`
-		JudgedBy string            `json:"judged_by"`
-	}
+	var d delivery.LevelDigest // packages kept as a sorted list, never keyed by a package path (issue 1930)
 	if err := json.Unmarshal(r, &d); err != nil {
 		return nil, true, true
 	}
@@ -311,21 +307,17 @@ func (s AskSteps) plan(ctx context.Context, job delivery.CheckJob, level string,
 	if len(d.Packages) == 0 || !slices.Equal(d.Beside, besideOf(job.Fetch)) {
 		return nil, true, runner != "" // a digest that names nothing or not the siblings keys no verdict: the level runs whole
 	}
-	la.Digest = r
-	pkgs := make([]string, 0, len(d.Packages))
-	for p := range d.Packages {
-		pkgs = append(pkgs, p)
-	}
-	sort.Strings(pkgs)
+	la.Digest = &d
 	var todo []string
-	for _, p := range pkgs {
+	for _, pd := range d.Packages {
+		p := pd.Package
 		if job.Kept && s.Kept != nil {
-			tests, found, err := s.Kept(ctx, level, d.Packages[p])
+			tests, found, err := s.Kept(ctx, level, pd.Digest)
 			if err != nil && s.Logf != nil {
 				s.Logf("the kept verdict of %s at %s cannot be read, so it runs: %v", p, level, err)
 			}
 			if err == nil && found {
-				la.Kept = append(la.Kept, delivery.KeptTaken{Package: p, Digest: d.Packages[p], Tests: tests})
+				la.Kept = append(la.Kept, delivery.KeptTaken{Package: p, Digest: pd.Digest, Tests: tests})
 				continue
 			}
 		}

go/buildstep/askrun_test.go

@@ -135,7 +135,7 @@ test {"level":"unit","packages":["x/pkg/b"],"workspace":"/w"}`
 		t.Fatalf("levels %+v, want the unit level alone", a.Levels)
 	}
 	u := a.Levels[0]
-	if u.Level != delivery.LevelUnit || len(u.Kept) != 1 || u.Kept[0].Package != "x/pkg/k" || u.Kept[0].Tests != 7 || len(u.Tests) != 1 || len(u.Digest) == 0 {
+	if u.Level != delivery.LevelUnit || len(u.Kept) != 1 || u.Kept[0].Package != "x/pkg/k" || u.Kept[0].Tests != 7 || len(u.Tests) != 1 || u.Digest == nil {
 		t.Fatalf("the unit level %+v", u)
 	}
 }
@@ -169,7 +169,7 @@ test {"level":"unit","runner":"workspace","workspace":"/w"}`},
 		if tc.judged == "installed" && (len(l.Kept) != 1 || l.Judged != "") {
 			t.Errorf("an installed module's level %+v, want the kept verdict taken", l)
 		}
-		if tc.judged != "installed" && (len(l.Kept) != 0 || len(l.Digest) != 0 || l.Judged != tc.judged) {
+		if tc.judged != "installed" && (len(l.Kept) != 0 || l.Digest != nil || l.Judged != tc.judged) {
 			t.Errorf("the candidate's level %+v, want nothing kept and the program named", l)
 		}
 	}
@@ -179,7 +179,7 @@ test {"level":"unit","runner":"workspace","workspace":"/w"}`},
 	if err != nil {
 		t.Fatal(err)
 	}
-	if l := out.(delivery.CheckAnswer).Levels[0]; len(l.Kept) != 0 || len(l.Digest) != 0 {
+	if l := out.(delivery.CheckAnswer).Levels[0]; len(l.Kept) != 0 || l.Digest != nil {
 		t.Fatalf("a digest without the siblings kept %+v", l)
 	}
 }

go/buildstep/digestshape_test.go

@@ -0,0 +1,234 @@
+package buildstep
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"sort"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+// credentialField is the node runtime's rule for a state value (mesh-tools node-tools/internal/bus/state.go, novox/hq
+// ADR 0201), copied whole: the first field anywhere in the value whose name, lower-cased and without - _ and spaces, ends
+// in a credential word, or "". A step's end is refused when it answers a field.
+func credentialField(value json.RawMessage) string {
+	endings := []string{"password", "passwd", "secret", "token", "credential", "credentials",
+		"authorization", "apikey", "privatekey", "accesskey", "cookie"}
+	var v any
+	if json.Unmarshal(value, &v) != nil {
+		return ""
+	}
+	var walk func(any) string
+	walk = func(v any) string {
+		switch t := v.(type) {
+		case map[string]any:
+			keys := make([]string, 0, len(t))
+			for k := range t {
+				keys = append(keys, k)
+			}
+			sort.Strings(keys)
+			for _, k := range keys {
+				norm := strings.NewReplacer("-", "", "_", "", " ", "").Replace(strings.ToLower(k))
+				for _, end := range endings {
+					if strings.HasSuffix(norm, end) {
+						return k
+					}
+				}
+				if found := walk(t[k]); found != "" {
+					return found
+				}
+			}
+		case []any:
+			for _, x := range t {
+				if found := walk(x); found != "" {
+					return found
+				}
+			}
+		}
+		return ""
+	}
+	return walk(v)
+}
+
+// runtimeState is a steps state that refuses a value as the node runtime does, and keeps every value it took.
+type runtimeState struct {
+	mu      sync.Mutex
+	took    map[string][]delivery.Step
+	refused int
+}
+
+func newRuntimeState() *runtimeState { return &runtimeState{took: map[string][]delivery.Step{}} }
+
+func (s *runtimeState) put(key string, value any) (uint64, error) {
+	raw, _ := value.(json.RawMessage)
+	if f := credentialField(raw); f != "" {
+		s.mu.Lock()
+		s.refused++
+		s.mu.Unlock()
+		return 0, fmt.Errorf("steps.%s carries a field %q, which names a credential", key, f)
+	}
+	st, err := delivery.DecodeStep(raw)
+	if err != nil {
+		return 0, err
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.took[key] = append(s.took[key], st)
+	return uint64(len(s.took[key])), nil
+}
+
+func (s *runtimeState) last(key string) (delivery.Step, bool) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	t := s.took[key]
+	if len(t) == 0 {
+		return delivery.Step{}, false
+	}
+	return t[len(t)-1], true
+}
+
+func fastEnds(t *testing.T) {
+	attempts, retry := EndAttempts, EndRetry
+	EndAttempts, EndRetry = 2, time.Millisecond
+	t.Cleanup(func() { EndAttempts, EndRetry = attempts, retry })
+}
+
+// 1930/AC1: given a check of a repository with the package .../internal/token, when its ask's answer is stored as the
+// step's end, then no field is keyed by a package path: the runtime's credential rule takes the value, and the package
+// and its digest read back from the stored end.
+func TestADigestOfAPackageNamedTokenIsStoredAsAList(t *testing.T) {
+	fastEnds(t)
+	r := &recorder{
+		fetch:   map[string]any{"workspace": "/w"},
+		digests: map[string]string{"x/internal/token": "d-token", "x/pkg/a": "d-a"},
+		tests:   map[string]bool{"x/internal/token": true, "x/pkg/a": true},
+	}
+	out, err := RunAsk(context.Background(), r.steps(), ask(delivery.CheckJob{Levels: []string{delivery.LevelUnit},
+		Fetch: map[string]any{"repository": "novox/x", "commit": runCommit}}))
+	if err != nil {
+		t.Fatal(err)
+	}
+	old, _ := json.Marshal(map[string]any{"levels": []any{map[string]any{"level": "unit", "digest": map[string]any{"packages": r.digests}}}})
+	if credentialField(old) != "x/internal/token" {
+		t.Fatalf("the rule does not refuse the form before issue 1930 (%s): the test proves nothing", old)
+	}
+	st := newRuntimeState()
+	if err := finish(st.put, "shanks.a1", delivery.Step{ID: "a1", Tool: "check ask", Node: "shanks", State: delivery.StepRunning, Started: time.Now()}, out, nil, t.Logf); err != nil {
+		t.Fatal(err)
+	}
+	s, ok := st.last("shanks.a1")
+	if !ok || s.State != delivery.StepPassed || st.refused != 0 {
+		t.Fatalf("the end %+v (written %v, refused %d), want it passed and taken at once", s, ok, st.refused)
+	}
+	var a delivery.CheckAnswer
+	if err := json.Unmarshal(s.Result, &a); err != nil {
+		t.Fatal(err)
+	}
+	if len(a.Levels) != 1 || a.Levels[0].Digest == nil {
+		t.Fatalf("levels %+v, want the unit level with its digest", a.Levels)
+	}
+	want := []delivery.PackageDigest{{Package: "x/internal/token", Digest: "d-token"}, {Package: "x/pkg/a", Digest: "d-a"}}
+	if got := a.Levels[0].Digest.Packages; fmt.Sprint(got) != fmt.Sprint(want) {
+		t.Errorf("packages %v, want %v", got, want)
+	}
+	if !strings.Contains(string(s.Result), `"package":"x/internal/token"`) {
+		t.Errorf("the stored end does not name the package as a value: %s", s.Result)
+	}
+}
+
+// 1930/AC2: given an end the state refuses every time, when finish writes it, then it logs the refusal and writes a small
+// failed end without the result that names the write's error; and when no end can be written at all it answers an error.
+func TestARefusedEndIsSaidAndEndsTheStepFailed(t *testing.T) {
+	fastEnds(t)
+	big := errors.New("the value carries a field \"x/internal/token\", which names a credential")
+	var refused int
+	var stored []delivery.Step
+	put := func(key string, value any) (uint64, error) {
+		st, _ := delivery.DecodeStep(value.(json.RawMessage))
+		if len(st.Result) > 0 {
+			refused++
+			return 0, big
+		}
+		stored = append(stored, st)
+		return 1, nil
+	}
+	var logged []string
+	logf := func(format string, args ...any) { logged = append(logged, fmt.Sprintf(format, args...)) }
+	err := finish(put, "shanks.a2", delivery.Step{ID: "a2", Tool: "check ask", Node: "shanks", State: delivery.StepRunning, Started: time.Now()}, map[string]any{"levels": 1}, nil, logf)
+	if err != nil {
+		t.Fatalf("finish answered %v, want the small end written", err)
+	}
+	if refused != EndAttempts || len(stored) != 1 {
+		t.Fatalf("refused %d, stored %d, want %d refusals and one small end", refused, len(stored), EndAttempts)
+	}
+	s := stored[0]
+	if s.State != delivery.StepFailed || len(s.Result) != 0 || !strings.Contains(s.Reason, "could not be written") ||
+		!strings.Contains(s.Reason, big.Error()) || !strings.Contains(s.Reason, "passed") {
+		t.Errorf("the small end %+v, want failed, no result, naming the write's error and that it passed", s)
+	}
+	if len(logged) == 0 || !strings.Contains(logged[0], big.Error()) {
+		t.Errorf("logged %q, want the refusal said", logged)
+	}
+	none := func(string, any) (uint64, error) { return 0, errors.New("the bus is gone") }
+	if err := finish(none, "shanks.a3", delivery.Step{ID: "a3", Tool: "check ask", Node: "shanks", Started: time.Now()}, nil, nil, logf); err == nil || !strings.Contains(err.Error(), "the bus is gone") {
+		t.Errorf("finish with no write possible answered %v, want the error", err)
+	}
+}
+
+// 1930/AC2: given an ask whose end the state refuses, when it runs, then the ask is acknowledged only once its small
+// failed end is written; when no end can be written at all, it is given back, never acknowledged as done.
+func TestAnAskWhoseEndCannotBeWrittenIsNotAcked(t *testing.T) {
+	fastEnds(t)
+	for _, tc := range []struct {
+		name      string
+		smallOK   bool
+		wantAcked bool
+	}{{"the small end written", true, true}, {"no end written", false, false}} {
+		t.Run(tc.name, func(t *testing.T) {
+			src := NewFakeAsks()
+			a := testAsk(delivery.AskCheck)
+			_ = src.PublishAsk(a)
+			var mu sync.Mutex
+			var ends []delivery.Step
+			put := func(key string, value any) (uint64, error) {
+				st, err := delivery.DecodeStep(value.(json.RawMessage))
+				if err != nil {
+					return 0, err
+				}
+				if st.State != delivery.StepRunning && (len(st.Result) > 0 || !tc.smallOK) {
+					return 0, errors.New("refused")
+				}
+				mu.Lock()
+				ends = append(ends, st)
+				mu.Unlock()
+				return 1, nil
+			}
+			p := &Puller{Source: src, Language: delivery.AskGo, Node: "shanks", Put: put, Capacity: func() int { return 1 },
+				PullWait: 10 * time.Millisecond, Logf: t.Logf,
+				Run: func(ctx context.Context, a delivery.StepAsk) (any, error) { return map[string]any{"ran": true}, nil }}
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+			go p.Pull(ctx)
+			waitFor(t, "the ask's fate", func() bool { ac, tm, nk := src.Fates(); return len(ac)+len(tm)+len(nk) > 0 })
+			cancel()
+			acked, termed, naked := src.Fates()
+			if tc.wantAcked && (len(acked) != 1 || len(naked) != 0) || !tc.wantAcked && (len(acked) != 0 || len(naked) == 0) || len(termed) != 0 {
+				t.Fatalf("acked %v naked %v termed %v", acked, naked, termed)
+			}
+			mu.Lock()
+			defer mu.Unlock()
+			if tc.smallOK {
+				last := ends[len(ends)-1]
+				if last.State != delivery.StepFailed || !strings.Contains(last.Reason, "refused") {
+					t.Errorf("the end %+v, want the small failed end", last)
+				}
+			}
+		})
+	}
+}

go/buildstep/pull.go

@@ -286,7 +286,7 @@ func (p *Puller) take(ctx context.Context, msg *Pulled) {
 		// (issue 1907): the asker's key fails with why.
 		why := fmt.Errorf("the ask was taken %d times and no run of it ended: each took its build node down or died with it, so it is run no more (poison)", msg.Delivered-1)
 		p.logf("the ask %s is ended on %s: %v", ask.ID, p.Node, why)
-		finish(p.Put, key, step, nil, why)
+		_ = finish(p.Put, key, step, nil, why, p.Logf) // run no more either way: the asker's bound says it then
 		_ = msg.Term()
 		return
 	}
@@ -337,7 +337,13 @@ func (p *Puller) run(ctx context.Context, msg *Pulled, ask delivery.StepAsk, key
 		// The process stops: the ask is neither ended nor acknowledged here, so the stream gives it to a node that runs.
 		return
 	}
-	finish(p.Put, key, step, result, err)
+	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).
+		p.logf("the ask %s ran on %s and no end of it could be written, so it is given back: %v", ask.ID, p.Node, ferr)
+		_ = msg.Nak()
+		return
+	}
 	if aerr := msg.Ack(); aerr != nil {
 		p.logf("the ask %s ended on %s and its acknowledgement failed: %v", ask.ID, p.Node, aerr)
 	}

go/buildstep/step.go

@@ -10,6 +10,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"log"
 	"strconv"
 	"time"
 
@@ -66,11 +67,15 @@ func begin(put Put, node, tool, id string) (delivery.Step, string, error) {
 // end runs run and writes the step's end.
 func end(ctx context.Context, put Put, key string, step delivery.Step, run func(ctx context.Context) (any, error)) {
 	result, err := protect(ctx, run)
-	finish(put, key, step, result, err)
+	_ = finish(put, key, step, result, err, log.Printf) // finish says a refused end itself
 }
 
-// finish writes the step's end from run's result or error: passed, failed or refused (a *Refusal).
-func finish(put Put, key string, step delivery.Step, result any, err error) {
+// finish writes the step's end from run's result or error: passed, failed or refused (a *Refusal). A write that fails is
+// tried EndAttempts times. When the whole end is refused every time (a result the state will not take: issue 1930), the
+// refusal is logged and a small failed end without the result is written instead, naming the write's error, so the
+// step's reader learns it is over and why. finish answers an error only when no end could be written at all, and then
+// the step's ask is not acknowledged as done.
+func finish(put Put, key string, step delivery.Step, result any, err error, logf func(format string, args ...any)) error {
 	step.Ended = now().UTC()
 	switch ref := (*Refusal)(nil); {
 	case err == nil:
@@ -88,9 +93,34 @@ func finish(put Put, key string, step delivery.Step, result any, err error) {
 	default:
 		step.State, step.Reason = delivery.StepFailed, err.Error()
 	}
+	werr := writeEnd(put, key, step)
+	if werr == nil {
+		return nil
+	}
+	if logf != nil {
+		logf("the end of the step %s (%s) could not be written in %d tries: %v", key, step.State, EndAttempts, werr)
+	}
+	ended := step.State
+	why := fmt.Sprintf("the step ended %s, and its end could not be written: %v", ended, werr)
+	if step.Reason != "" {
+		why = fmt.Sprintf("the step ended %s (%s), and its end could not be written: %v", ended, step.Reason, werr)
+	}
+	step.State, step.Reason, step.Result = delivery.StepFailed, why, nil
+	if ferr := writeEnd(put, key, step); ferr != nil {
+		if logf != nil {
+			logf("the failed end of the step %s could not be written either: %v", key, ferr)
+		}
+		return fmt.Errorf("buildstep: no end of the step %s could be written: %w", key, ferr)
+	}
+	return nil
+}
+
+// writeEnd writes the step's end, trying EndAttempts times.
+func writeEnd(put Put, key string, step delivery.Step) error {
 	for attempt := 1; ; attempt++ {
-		if werr := write(put, key, step); werr == nil || attempt >= EndAttempts {
-			return
+		werr := write(put, key, step)
+		if werr == nil || attempt >= EndAttempts {
+			return werr
 		}
 		time.Sleep(EndRetry)
 	}

go/delivery/stepasks.go

@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"regexp"
+	"sort"
 	"strings"
 	"time"
 )
@@ -361,11 +362,74 @@ type LevelAnswer struct {
 	// Judged is what the digest under the candidate's runner said of the program that ran it ("candidate <merge>"),
 	// when the candidate judged the level (ADR 0535 §4, §5); empty otherwise.
 	Judged string            `json:"judged,omitempty"`
-	Digest json.RawMessage   `json:"digest,omitempty"`
+	Digest *LevelDigest      `json:"digest,omitempty"`
 	Kept   []KeptTaken       `json:"kept,omitempty"`
 	Tests  []json.RawMessage `json:"tests,omitempty"`
 }
 
+// LevelDigest is the digest's answer as a level keeps it in the ask's step result: its packages a list of
+// {package, digest} sorted by package, never a map keyed by the package's path (issue 1930). The step result is a
+// state value, and the node runtime refuses a value with a field name ending in a credential word (novox/hq ADR 0201):
+// a Go package such as .../internal/token is such a name as a map key, and is a plain value in the list. It decodes
+// the digest tool's own answer, and a value stored before issue 1930, with packages as a map too.
+type LevelDigest struct {
+	Level        string          `json:"level,omitempty"`
+	Packages     []PackageDigest `json:"packages"`
+	Beside       []string        `json:"beside,omitempty"`
+	WithoutTests []string        `json:"without_tests,omitempty"`
+	JudgedBy     string          `json:"judged_by,omitempty"`
+}
+
+// PackageDigest is one package's digest at a level.
+type PackageDigest struct {
+	Package string `json:"package"`
+	Digest  string `json:"digest"`
+}
+
+// UnmarshalJSON reads packages as the list or as the map keyed by package path (the digest tool's answer, and a level
+// answer stored before issue 1930), and sorts them by package.
+func (d *LevelDigest) UnmarshalJSON(raw []byte) error {
+	var v struct {
+		Level        string          `json:"level"`
+		Packages     json.RawMessage `json:"packages"`
+		Beside       []string        `json:"beside"`
+		WithoutTests []string        `json:"without_tests"`
+		JudgedBy     string          `json:"judged_by"`
+	}
+	if err := json.Unmarshal(raw, &v); err != nil {
+		return err
+	}
+	var pkgs []PackageDigest
+	if p := strings.TrimSpace(string(v.Packages)); p != "" && p != "null" {
+		if strings.HasPrefix(p, "{") {
+			var m map[string]string
+			if err := json.Unmarshal(v.Packages, &m); err != nil {
+				return fmt.Errorf("a digest's packages: %w", err)
+			}
+			for pkg, digest := range m {
+				pkgs = append(pkgs, PackageDigest{Package: pkg, Digest: digest})
+			}
+		} else if err := json.Unmarshal(v.Packages, &pkgs); err != nil {
+			return fmt.Errorf("a digest's packages: %w", err)
+		}
+	}
+	sort.Slice(pkgs, func(i, j int) bool { return pkgs[i].Package < pkgs[j].Package })
+	*d = LevelDigest{Level: v.Level, Packages: pkgs, Beside: v.Beside, WithoutTests: v.WithoutTests, JudgedBy: v.JudgedBy}
+	return nil
+}
+
+// Map is the digests by package, for a lookup; nil when it names none.
+func (d *LevelDigest) Map() map[string]string {
+	if d == nil || len(d.Packages) == 0 {
+		return nil
+	}
+	m := make(map[string]string, len(d.Packages))
+	for _, p := range d.Packages {
+		m[p.Package] = p.Digest
+	}
+	return m
+}
+
 // KeptTaken is a package whose kept verdict the check took instead of testing it.
 type KeptTaken struct {
 	Package string `json:"package"`

go/delivery/stepasks_test.go

@@ -1,6 +1,7 @@
 package delivery
 
 import (
+	"encoding/json"
 	"errors"
 	"reflect"
 	"strings"
@@ -146,3 +147,34 @@ func TestABuildingKeyNamesItsAsks(t *testing.T) {
 		}
 	}
 }
+
+// 1930/AC1: a level answer stored before issue 1930, its digest's packages a map keyed by package path, and the digest
+// tool's own answer, read as the list sorted by package; the list form reads back the same and keys no field by a path.
+func TestALevelDigestReadsTheMapAndTheList(t *testing.T) {
+	var old LevelAnswer
+	if err := json.Unmarshal([]byte(`{"level":"unit","digest":{"level":"unit","packages":{"x/pkg/a":"d-a","x/internal/token":"d-t"},"beside":["novox/y abc"],"judged_by":"installed"}}`), &old); err != nil {
+		t.Fatal(err)
+	}
+	want := []PackageDigest{{Package: "x/internal/token", Digest: "d-t"}, {Package: "x/pkg/a", Digest: "d-a"}}
+	if old.Digest == nil || !reflect.DeepEqual(old.Digest.Packages, want) || old.Digest.JudgedBy != "installed" || len(old.Digest.Beside) != 1 {
+		t.Fatalf("the old form read %+v", old.Digest)
+	}
+	raw, err := json.Marshal(old)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if strings.Contains(string(raw), `"x/internal/token":`) {
+		t.Errorf("the list form keys a field by a package path: %s", raw)
+	}
+	var back LevelAnswer
+	if err := json.Unmarshal(raw, &back); err != nil {
+		t.Fatal(err)
+	}
+	if !reflect.DeepEqual(back.Digest, old.Digest) || back.Digest.Map()["x/internal/token"] != "d-t" {
+		t.Errorf("the list form read back %+v, want %+v", back.Digest, old.Digest)
+	}
+	var none LevelAnswer
+	if err := json.Unmarshal([]byte(`{"level":"unit"}`), &none); err != nil || none.Digest != nil || none.Digest.Map() != nil {
+		t.Errorf("a level without a digest read %+v, %v", none.Digest, err)
+	}
+}