From 403a5d72ab934af19257884e5b00d9df2b1d5518 Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 23 Oct 2026 23:32:00 +0200 Subject: [PATCH] Keep a level's digest as a list and say an end the state refuses, so a check of a repository with a package named token ends (issue 1930) A check ask's answer kept the digest's packages as a map keyed by Go import path; the node runtime refuses a state value with a field ending in a credential word, so the end of any check of mesh-controller (internal/token) was never written and the ask was acked anyway. The level's digest is now delivery.LevelDigest, packages a sorted list of {package, digest}, reading the old map form too. finish logs an end refused every time, writes a small failed end naming the write's error, and the puller gives the ask back instead of acking it when no end could be written. --- go/buildstep/askrun.go | 20 +-- go/buildstep/askrun_test.go | 6 +- go/buildstep/digestshape_test.go | 234 +++++++++++++++++++++++++++++++ go/buildstep/pull.go | 10 +- go/buildstep/step.go | 40 +++++- go/delivery/stepasks.go | 66 ++++++++- go/delivery/stepasks_test.go | 32 +++++ 7 files changed, 383 insertions(+), 25 deletions(-) create mode 100644 go/buildstep/digestshape_test.go diff --git a/go/buildstep/askrun.go b/go/buildstep/askrun.go index c59d839..44d2fc1 100644 --- a/go/buildstep/askrun.go +++ b/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 } } diff --git a/go/buildstep/askrun_test.go b/go/buildstep/askrun_test.go index ebc8f40..fa1bf24 100644 --- a/go/buildstep/askrun_test.go +++ b/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) } } diff --git a/go/buildstep/digestshape_test.go b/go/buildstep/digestshape_test.go new file mode 100644 index 0000000..aa2b757 --- /dev/null +++ b/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) + } + } + }) + } +} diff --git a/go/buildstep/pull.go b/go/buildstep/pull.go index 9fb676b..c4cd09b 100644 --- a/go/buildstep/pull.go +++ b/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) } diff --git a/go/buildstep/step.go b/go/buildstep/step.go index 8b9270b..8a01aaa 100644 --- a/go/buildstep/step.go +++ b/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) } diff --git a/go/delivery/stepasks.go b/go/delivery/stepasks.go index 4e1f001..1e8cdd3 100644 --- a/go/delivery/stepasks.go +++ b/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 "), // 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"` diff --git a/go/delivery/stepasks_test.go b/go/delivery/stepasks_test.go index 9ac39a5..7029fe9 100644 --- a/go/delivery/stepasks_test.go +++ b/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) + } +}