diff --git a/go/buildstep/askrun.go b/go/buildstep/askrun.go index f1ef95b..193f7e9 100644 --- a/go/buildstep/askrun.go +++ b/go/buildstep/askrun.go @@ -37,6 +37,11 @@ type AskSteps struct { // Has says whether this node has a need a package's tests declare (delivery.AskNeeds, issue 890); a package needing // what it lacks is not tested here but named in the level's Unmet, for the asker to place. Nil has nothing. Has func(need string) bool + // Narrate says an event on the module's own stream (stdio.Emit) and Module is the module's name: a failed test's whole + // output is said on the module's log stream while the ask runs, and the step's result names where (narrate). Nil + // says nothing, and the whole output is dropped from the result all the same. + Narrate func(event string, body any) error + Module string } // RunAsk runs ask's job with steps and answers a delivery.BuildAnswer or a delivery.CheckAnswer. @@ -221,6 +226,7 @@ func (s AskSteps) level(ctx context.Context, job delivery.CheckJob, level string if err != nil { return false, err } + r = s.stepLog(ctx).narrate(r, s.Logf) la.Tests = append(la.Tests, r) var t struct { Passed bool `json:"passed"` diff --git a/go/buildstep/askrun_test.go b/go/buildstep/askrun_test.go index 49ea614..7f1fc43 100644 --- a/go/buildstep/askrun_test.go +++ b/go/buildstep/askrun_test.go @@ -5,10 +5,12 @@ import ( "encoding/json" "errors" "fmt" + "slices" "strings" "testing" "time" + stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/delivery" ) @@ -291,4 +293,65 @@ func TestRunAskRunsEveryPackageAfterAFailure(t *testing.T) { if a := out.(delivery.CheckAnswer); len(r.calls) != 3 || len(a.Levels) != 1 || len(a.Levels[0].Tests) != 2 { t.Fatalf("calls %v, levels %+v: want both directories of the unit level and no integration level", r.calls, a.Levels) } + + // The operator, 2026-10-30: a failure's whole output is said on the module's log stream under the run and the step, + // the result names where and holds no whole output, and the log reads back whole, in pages, its lines in order. + var lines []string + for i := range 3000 { + lines = append(lines, fmt.Sprintf("line %04d %s", i, strings.Repeat("x", 40))) + } + var said []stdio.StreamEntry + s = r.steps() + s.Digest, s.Module = nil, "mesh-build-node" + s.Narrate = func(event string, body any) error { + b, _ := json.Marshal(body) + said = append(said, stdio.StreamEntry{Sequence: uint64(len(said) + 1), Event: event, Body: b}) + return nil + } + s.Test = func(ctx context.Context, args map[string]any) (any, error) { + return map[string]any{"passed": false, "failed": []Failure{{Package: "b", Test: "t", Cause: "boom", Output: "boom", Full: strings.Join(lines, "\n")}}}, nil + } + r.fetch["modules"] = []string{"b"} + ctx := context.WithValue(context.Background(), phaseKey{}, &phases{step: delivery.Step{ID: "s1", Node: "ace"}, run: "r1"}) + out, err = RunAsk(ctx, s, ask(delivery.CheckJob{Levels: []string{delivery.LevelUnit}, Fetch: map[string]any{"repository": "novox/x", "commit": runCommit}})) + if err != nil { + t.Fatal(err) + } + var res struct{ Failed []Failure } + _ = json.Unmarshal(out.(delivery.CheckAnswer).Levels[0].Tests[0], &res) + if f := res.Failed[0]; f.Full != "" || f.Step != "s1" || f.Module != "mesh-build-node" || f.Machine != "ace" || f.Run != "r1" || len(said) < 2 || said[0].Event != "log.r1.s1" { + t.Fatalf("failure %+v, %d events (first %+v): want its address, no whole output, and its log in several events", f, len(said), said[0].Event) + } + read := func(events string, from uint64, limit int) (*stdio.StreamPage, error) { + p := &stdio.StreamPage{Last: uint64(len(said))} + for _, e := range said { + if e.Sequence >= max(from, 1) && e.Event == events && len(p.Entries) < limit { + p.Entries = append(p.Entries, e) + } + } + if n := len(p.Entries); n > 0 && p.Entries[n-1].Sequence < p.Last { + p.Next = p.Entries[n-1].Sequence + 1 + } + return p, nil + } + var got []string + pages := 0 + for from := uint64(0); ; pages++ { + page, err := ReadStepLog(read, "r1", "s1", from) + if err != nil || page.Gap || page.Bytes > LogPageBytes { + t.Fatalf("page from %d: %+v, %v", from, page.Next, err) + } + got = append(got, page.Lines...) + if from = page.Next; from == 0 { + break + } + } + // Each page opens with the failure it reads: "=== ". + own := slices.DeleteFunc(slices.Clone(got), func(l string) bool { return l == "=== b t" }) + if pages < 2 || got[0] != "=== b t" || strings.Join(own, "\n") != strings.Join(lines, "\n") { + t.Errorf("%d pages, %d lines read back, want %d in several pages, headed by the failure", pages+1, len(own), len(lines)) + } + if _, err := ReadStepLog(read, "r1", "s1.>", 0); err == nil { + t.Error("a step id that is no token was read") + } } diff --git a/go/buildstep/failure.go b/go/buildstep/failure.go index 78c27af..55e2313 100644 --- a/go/buildstep/failure.go +++ b/go/buildstep/failure.go @@ -29,6 +29,16 @@ type Failure struct { // Output is the failing test's own output, its last lines up to OutputBytes, Cause first: an older checker reads // only its first line. Output string `json:"output,omitempty"` + // Full is the failing test's whole output, its last FullBytes, passing lines kept: it crosses from a shut-off level + // process to the step, which says it on the build module's own log stream and empties it, so it never reaches the + // step's result (narrate). + Full string `json:"full_output,omitempty"` + // Step, Module, Machine and Run say which build step said the whole output on its log, of which build module on which + // machine, under which run (delivery.StepLogEvent); set when it was said, absent when it was not. + Step string `json:"step,omitempty"` + Module string `json:"module,omitempty"` + Machine string `json:"machine,omitempty"` + Run string `json:"run,omitempty"` } // OutputBytes is the most a failure keeps of its output, and LineBytes the most of one line: four kilobytes holds a long @@ -109,6 +119,19 @@ func Clip(lines []string, cause string) string { return strings.Join(kept, "\n") } +// FullOf joins a failure's whole output, its last FullBytes when longer, starting on a whole line. +func FullOf(lines []string) string { + s := strings.Join(lines, "\n") + if len(s) <= FullBytes { + return s + } + tail := s[len(s)-FullBytes:] + if i := strings.IndexByte(tail, '\n'); i >= 0 { + tail = tail[i+1:] + } + return fmt.Sprintf("... (%d earlier bytes)\n", len(s)-len(tail)) + strings.ToValidUTF8(tail, "") +} + // CutLine cuts a line to LineBytes, on a character boundary. func CutLine(l string) string { if len(l) <= LineBytes { @@ -143,5 +166,5 @@ func ConsoleFailure(pkg, test, out string) Failure { cause = test + " failed and wrote nothing that says why" } cause = CutLine(cause) - return Failure{Package: pkg, Test: test, Cause: cause, Output: Clip(lines, cause)} + return Failure{Package: pkg, Test: test, Cause: cause, Output: Clip(lines, cause), Full: FullOf(lines)} } diff --git a/go/buildstep/gotest.go b/go/buildstep/gotest.go index 7db4c8c..ec86e95 100644 --- a/go/buildstep/gotest.go +++ b/go/buildstep/gotest.go @@ -211,7 +211,7 @@ func (r *goReader) subtestTakesTheCause(pkg, test string) bool { took := false for i, f := range r.res.Failed { if f.Package == pkg && strings.HasPrefix(f.Test, test+"/") && f.Cause == noMessage { - r.res.Failed[i].Cause, r.res.Failed[i].Output = cause, Clip(r.testOut[key], cause) + r.res.Failed[i].Cause, r.res.Failed[i].Output, r.res.Failed[i].Full = cause, Clip(r.testOut[key], cause), FullOf(r.testOut[key]) took = true } } @@ -231,7 +231,7 @@ func (r *goReader) testFailure(pkg, test string) Failure { if cause == "" { cause = noMessage } - return Failure{Package: pkg, Test: test, Output: Clip(r.testOut[key], cause), Cause: cause} + return Failure{Package: pkg, Test: test, Output: Clip(r.testOut[key], cause), Cause: cause, Full: FullOf(r.testOut[key])} } // goSummaryLine says a line is go test's summary of a package, in go's exact forms only: "PASS" or "FAIL" alone, @@ -380,7 +380,7 @@ func (r *goReader) packageFailure(name string) Failure { if cause == "" { cause = "the package failed with no test to name and no panic, signal or build error in its output" } - return Failure{Package: name, Test: goTestNames(running), Output: Clip(lines, cause), Cause: cause} + return Failure{Package: name, Test: goTestNames(running), Output: Clip(lines, cause), Cause: cause, Full: FullOf(lines)} } var goFileAndLine = regexp.MustCompile(`^\S+\.go:\d+: `) diff --git a/go/buildstep/phase.go b/go/buildstep/phase.go index 130178a..a8155f6 100644 --- a/go/buildstep/phase.go +++ b/go/buildstep/phase.go @@ -34,6 +34,9 @@ type phases struct { ended bool logf func(format string, args ...any) prefix string + // run is the ask's run (delivery.StepAsk.Run), and log the step's log (steplog.go), made at the first line said. + run string + log *stepLog } func (r *phases) enter(phase string) { diff --git a/go/buildstep/pull.go b/go/buildstep/pull.go index da1e540..85efa42 100644 --- a/go/buildstep/pull.go +++ b/go/buildstep/pull.go @@ -619,7 +619,7 @@ func (p *Puller) run(ctx context.Context, msg *Pulled, ask delivery.StepAsk, key } } }() - ph := &phases{put: p.Put, key: key, step: step, logf: p.Logf, prefix: fmt.Sprintf("the %s ask %s on %s", ask.Kind, ask.ID, p.Node)} + ph := &phases{put: p.Put, key: key, step: step, run: ask.Run, 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) { return p.Run(context.WithValue(ctx, phaseKey{}, ph), ask) }) diff --git a/go/buildstep/step.go b/go/buildstep/step.go index f5ba8f0..daf7327 100644 --- a/go/buildstep/step.go +++ b/go/buildstep/step.go @@ -86,7 +86,7 @@ func finish(put Put, key string, step delivery.Step, result any, err error, logf if merr != nil { step.State, step.Reason = delivery.StepFailed, "the result is not JSON: "+merr.Error() } else { - step.Result = raw + step.Result = (*stepLog)(nil).narrate(raw, nil) // a step run by hand keeps no whole output in its result } } case errors.As(err, &ref): diff --git a/go/buildstep/step_test.go b/go/buildstep/step_test.go index 8bee52c..c0b4588 100644 --- a/go/buildstep/step_test.go +++ b/go/buildstep/step_test.go @@ -86,6 +86,10 @@ func TestStepEndsPassedFailedOrRefusedOnItsKey(t *testing.T) { result string }{ {"result", func(context.Context) (any, error) { return map[string]string{"sha": "abc"}, nil }, delivery.StepPassed, "", `{"sha":"abc"}`}, + // A step run by hand says no log: a failure's whole output never reaches its result. + {"no whole output", func(context.Context) (any, error) { + return map[string]any{"failed": []map[string]string{{"package": "p", "full_output": "all of it"}}}, nil + }, delivery.StepPassed, "", `{"failed":[{"package":"p"}]}`}, {"error", func(context.Context) (any, error) { return nil, errors.New("exit 2") }, delivery.StepFailed, "exit 2", ""}, {"refusal", func(context.Context) (any, error) { return nil, &Refusal{Text: "go 1.27.1 installed, the recipe builds with go 1.27.0", Want: "1.27.0", Have: "1.27.1"} diff --git a/go/buildstep/steplog.go b/go/buildstep/steplog.go new file mode 100644 index 0000000..8daa0d3 --- /dev/null +++ b/go/buildstep/steplog.go @@ -0,0 +1,221 @@ +package buildstep + +import ( + "context" + "encoding/json" + "fmt" + "strings" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// A build step's log on its build module's own stream (the operator, 2026-10-30: a verdict cut a failure to its last four +// kilobytes and named no place that held the rest, so a check's cause was guessed; novox/hq ADR 0157 for the shape: a +// build says what it does on the bus, under its id, in sequenced lines, read back from the stream). The step's result +// keeps each failure's Output clipped to OutputBytes, the summary; the failure's whole output, up to FullBytes, is said +// as delivery.StepLog events on delivery.StepLogEvent(run, step), at most LogEventBytes each, on the stream the module +// declares for `log.**` (max age and size in its manifest, dropping its oldest). ReadStepLog reads it back in pages. + +const ( + // FullBytes is the most of one failure's output that is said whole (its last bytes). + FullBytes = 1 << 20 + // LogEventBytes is the most lines one log event holds, and LogPageBytes the most one page of a read answers. + LogEventBytes = 64 << 10 + LogPageBytes = 64 << 10 +) + +// stepLog says one step's log: its node, id and run, the build module, how to say an event, and the next line's number. +type stepLog struct { + say func(event string, body any) error + module, node string + id, run string + seq int +} + +// stepLog is the log of the step whose ask runs under ctx, or nil when no ask runs (a tool called by hand) or the module +// says no log. +func (s AskSteps) stepLog(ctx context.Context) *stepLog { + r, ok := ctx.Value(phaseKey{}).(*phases) + if !ok || r == nil || s.Narrate == nil { + return nil + } + r.mu.Lock() + defer r.mu.Unlock() + if r.log == nil { + r.log = &stepLog{say: s.Narrate, module: s.Module, node: r.step.Node, id: r.step.ID, run: r.run} + } + return r.log +} + +// narrate says every failure's Full of a step answer's "failed" on the step's log and empties it, setting the failure's +// Step, Module, Machine and Run; a failure whose lines could not all be said keeps no address, and that is said through +// logf. With l nil, Full is only emptied: the step's result never holds it. +func (l *stepLog) narrate(raw json.RawMessage, logf func(string, ...any)) json.RawMessage { + if !strings.Contains(string(raw), `"full_output"`) { + return raw + } + var answer map[string]json.RawMessage + if err := json.Unmarshal(raw, &answer); err != nil || answer["failed"] == nil { + return raw + } + var failed []map[string]json.RawMessage + if err := json.Unmarshal(answer["failed"], &failed); err != nil { + return raw + } + for i, f := range failed { + var full string + _ = json.Unmarshal(f["full_output"], &full) + delete(failed[i], "full_output") + if full == "" || l == nil { + continue + } + var name struct { + Package string `json:"package"` + Test string `json:"test"` + } + b, _ := json.Marshal(f) + _ = json.Unmarshal(b, &name) + if err := l.say1(name.Package, name.Test, strings.Split(full, "\n")); err != nil { + if logf != nil { + logf("the whole output of %s %s in the step %s is not on its log: %v", name.Package, name.Test, l.id, err) + } + continue + } + for field, v := range map[string]string{"step": l.id, "module": l.module, "machine": l.node, "run": l.run} { + failed[i][field], _ = json.Marshal(v) + } + } + answer["failed"], _ = json.Marshal(failed) + out, err := json.Marshal(answer) + if err != nil { + return raw + } + return out +} + +// say1 says one failure's lines, in events of at most LogEventBytes, numbered on from the step's last line. +func (l *stepLog) say1(pkg, test string, lines []string) error { + event := delivery.StepLogEvent(l.run, l.id) + for len(lines) > 0 { + n, size := 0, 0 + for n < len(lines) && (n == 0 || size+len(lines[n])+1 <= LogEventBytes) { + if len(lines[n]) > LogEventBytes { + lines[n] = strings.ToValidUTF8(lines[n][:LogEventBytes], "") // one line longer than an event is cut + } + size += len(lines[n]) + 1 + n++ + } + if err := l.say(event, delivery.StepLog{Seq: l.seq + 1, Package: pkg, Test: test, Lines: lines[:n]}); err != nil { + return err + } + l.seq += n + lines = lines[n:] + } + return nil +} + +// StreamRead reads a module's own stream: stdio.Stream(name).Read's shape. +type StreamRead func(events string, from uint64, limit int) (*stdio.StreamPage, error) + +// StepLogPage is one page of a step's log: the lines from the stream sequence From on, headed "=== " +// where a failure's lines start on the page, Next the sequence to read on from (0 when nothing follows yet; a step still running may +// say more), Last the stream's last sequence, and Gap when a line number was skipped (the stream dropped it with age). +type StepLogPage struct { + Run string `json:"run"` + Step string `json:"step"` + Event string `json:"event"` + From uint64 `json:"from"` + Lines []string `json:"lines"` + Bytes int `json:"bytes"` + Next uint64 `json:"next"` + Last uint64 `json:"last"` + Gap bool `json:"gap,omitempty"` +} + +// ReadStepLog reads one page of about LogPageBytes of the log of the step id under run, from the stream sequence from on. +func ReadStepLog(read StreamRead, run, id string, from uint64) (StepLogPage, error) { + if !stepToken(run) || !stepToken(id) { + return StepLogPage{}, fmt.Errorf("a step's log is named by its run and its step id, each one token of letters, digits and -: %q, %q", run, id) + } + event := delivery.StepLogEvent(run, id) + page := StepLogPage{Run: run, Step: id, Event: event, From: from, Lines: []string{}} + seq, test := 0, "\x00" + for { + got, err := read(event, from, 16) + if err != nil { + return page, fmt.Errorf("the log %s cannot be read: %w", event, err) + } + page.Last = got.Last + for _, e := range got.Entries { + var l delivery.StepLog + if err := json.Unmarshal(e.Body, &l); err != nil { + return page, fmt.Errorf("the log %s holds at %d what is no log event: %v", event, e.Sequence, err) + } + size := 0 + for _, line := range l.Lines { + size += len(line) + 1 + } + if page.Bytes > 0 && page.Bytes+size > LogPageBytes { + page.Next = e.Sequence + return page, nil + } + if seq > 0 && l.Seq != seq { + page.Gap = true + } + if name := l.Package + " " + l.Test; name != test { + page.Lines = append(page.Lines, "=== "+strings.TrimSpace(name)) + test = name + } + page.Lines = append(page.Lines, l.Lines...) + page.Bytes += size + seq = l.Seq + len(l.Lines) + } + if got.Next == 0 { + return page, nil + } + from = got.Next + } +} + +// stepToken is one subject token of a run or a step id. +func stepToken(s string) bool { + if s == "" { + return false + } + for _, r := range s { + if !(r >= 'a' && r <= 'z' || r >= 'A' && r <= 'Z' || r >= '0' && r <= '9' || r == '-' || r == '_') { + return false + } + } + return true +} + +// LogStreamOf is the local name of a build module's own log stream, -build-logs (a stream's bus name is +// mesh-wide): LogStreamOf names it. +func LogStreamOf(language string) string { return language + "-build-logs" } + +// LogTool is a build module's tool that reads one step's log from its own stream, {run, step, from}: a failed test's +// whole output, which the step's result and the checker's verdict keep cut. The stream is the mesh's, so any machine's +// instance answers for a step of any machine. +func LogTool(name string, read StreamRead) stdio.Tool { + return stdio.Tool{Name: name, + Description: "The full build log of one build step: the whole output of every test that failed in it, which a check's " + + "verdict and the step's result keep cut to their last lines (\"... (N earlier lines)\"). One page of about 64 KB; " + + "read on with from = the answer's next until it is 0 (a step still running may say more: read again from next). " + + "run is the check (its heads hash) or the delivery plan the step ran for, \"none\" when the ask named none; step " + + "is the step id; both are in a failed verdict (checker_failure, checker_pull). Kept 14 days. Read-only.", + Input: map[string]any{"type": "object", "required": []string{"run", "step"}, "properties": map[string]any{ + "run": map[string]any{"type": "string", "description": "the check or delivery plan the step ran for, or none"}, + "step": map[string]any{"type": "string", "description": "the step id (the ask's id)"}, + "from": map[string]any{"type": "integer", "description": "the stream sequence to read on from: the last answer's next; 0 is the start"}}}, + Run: func(args map[string]any) (any, error) { + run, _ := args["run"].(string) + step, _ := args["step"].(string) + from, _ := args["from"].(float64) + if from < 0 { + return nil, fmt.Errorf("from is a stream sequence, 0 or more") + } + return ReadStepLog(read, run, step, uint64(from)) + }} +} diff --git a/go/buildstep/tap.go b/go/buildstep/tap.go index 09331a0..2df7ced 100644 --- a/go/buildstep/tap.go +++ b/go/buildstep/tap.go @@ -107,7 +107,7 @@ func ReadTAP(pkg, script, dir, report, console string) []Failure { if dir != "" { file = strings.TrimPrefix(file, strings.TrimSuffix(dir, "/")+"/") } - f := Failure{Package: pkg, Test: name, File: file, Cause: cause, Output: Clip(block, cause)} + f := Failure{Package: pkg, Test: name, File: file, Cause: cause, Output: Clip(block, cause), Full: FullOf(block)} if fields["failureType"] == "cancelledByParent" { cancelled = append(cancelled, f) continue diff --git a/go/buscompose/create_test.go b/go/buscompose/create_test.go index 68cd04b..5625e30 100644 --- a/go/buscompose/create_test.go +++ b/go/buscompose/create_test.go @@ -1,6 +1,7 @@ package buscompose import ( + "strings" "testing" "time" @@ -13,6 +14,9 @@ func TestDeclaredStreamsAreCreatedAsTheControllerAssertsThem(t *testing.T) { keyed := BucketStreamConfig(Bucket{Module: "m", Name: "presence", History: 5, TTLSeconds: 60, KeyTTL: true}) queue := ModuleStreamConfig(ModuleStream{Module: "b", Name: "ASKS", Events: []string{"ask"}, MaxAgeDays: 7, MaxMiB: 32, WorkQueue: true, Duplicates: 1}) + logs := ModuleStream{Module: "mesh-build-go", Name: "GO_BUILD_LOGS", Events: []string{"log.>"}, MaxAgeDays: 14, MaxMiB: 2048} + lcfg := ModuleStreamConfig(logs) + grants := strings.Join(ownStreamGrants("mesh-build-go", []ModuleStream{logs, {Module: "b", Name: "ASKS", Events: []string{"ask"}, WorkQueue: true}}), " ") for _, row := range []struct { what string got, want any @@ -37,6 +41,13 @@ func TestDeclaredStreamsAreCreatedAsTheControllerAssertsThem(t *testing.T) { {"queue cap", queue.MaxBytes, int64(32 << 20)}, {"queue discards new", queue.Discard, jetstream.DiscardNew}, {"queue duplicates", queue.Duplicates, time.Hour}, + {"queue is read by its consumer, never directly", queue.AllowDirect, false}, + {"a module's stream is read back directly", lcfg.AllowDirect, true}, + {"a module's stream drops its oldest", lcfg.Discard, jetstream.DiscardOld}, + {"a module's stream cap", lcfg.MaxBytes, int64(2048 << 20)}, + {"a module's stream age", lcfg.MaxAge, 14 * 24 * time.Hour}, + {"its owner publishes and reads it, and no other's", grants, + "mesh.stream.mesh-build-go.event.log.> $JS.API.STREAM.INFO.GO_BUILD_LOGS $JS.API.DIRECT.GET.GO_BUILD_LOGS"}, {"queue metadata", queue.Metadata[WorkQueueMetadata] + queue.Metadata[RestrictedMetadata], "truetrue"}, } { if row.got != row.want { diff --git a/go/buscompose/modstreams.go b/go/buscompose/modstreams.go index d270738..d71a02b 100644 --- a/go/buscompose/modstreams.go +++ b/go/buscompose/modstreams.go @@ -89,6 +89,8 @@ func (s ModuleStream) Stream() Stream { Why: why, // Said on the bus, so a dead letter kept from it ages out with it (novox/hq ADR 0376 ยง3). Restricted: s.Restricted, + // Its owner reads it back by direct gets, with no consumer (its runtime's stream.read, as a log's read). + Direct: !s.WorkQueue, } if s.WorkQueue { out.Retention = RetentionWorkQueue @@ -248,7 +250,9 @@ func heardGrants(p Principal, hears []Heard) []string { return out } -// ownStreamGrants are the subjects a module publishes its own streams' events on, and no other's. +// ownStreamGrants are the subjects a module publishes its own streams' events on, and no other's, and the reading of +// each back by direct gets (its runtime's stream.read: the stream's info and the batch direct get, with no consumer), as +// a log is read. A work-queue stream is its one consumer's to read, never by direct get. func ownStreamGrants(module string, streams []ModuleStream) []string { var out []string for _, s := range streams { @@ -256,6 +260,9 @@ func ownStreamGrants(module string, streams []ModuleStream) []string { continue } out = append(out, s.Subjects()...) + if !s.WorkQueue && safeSubject.MatchString(s.Name) { + out = append(out, "$JS.API.STREAM.INFO."+s.Name, "$JS.API.DIRECT.GET."+s.Name) + } } return out } diff --git a/go/delivery/checker.go b/go/delivery/checker.go index 6b07ede..ffd0b66 100644 --- a/go/delivery/checker.go +++ b/go/delivery/checker.go @@ -141,6 +141,12 @@ type FailedFirst struct { Test string `json:"test,omitempty"` Machine string `json:"machine,omitempty"` Words string `json:"words,omitempty"` + // Step, Module and Run say where a failed test's whole output is: the build step (its ask's id) of which build + // module, under which run (the check), its log on that module's own stream as StepLogEvent(Run, Step), read with + // the module's log tool. Absent for a failure that is no test's, and in a verdict written before them. + Step string `json:"step,omitempty"` + Module string `json:"module,omitempty"` + Run string `json:"run,omitempty"` } // MaxFailures is the most entries a verdict's Failures holds (issue 1921): fifty tests named one by one, and an entry diff --git a/go/delivery/stepasks.go b/go/delivery/stepasks.go index fba438c..8621def 100644 --- a/go/delivery/stepasks.go +++ b/go/delivery/stepasks.go @@ -253,6 +253,30 @@ type StepAsk struct { Compose *ComposeJob `json:"compose,omitempty"` // Needs are what the node that runs the ask must have (AskNeeds); a check ask's alone. Empty: any node. Needs []string `json:"needs,omitempty"` + // Run is what the ask is part of, one token: the check (its heads hash) or the delivery plan. A build step's log is + // said under it and the ask's id (StepLogEvent). Empty in an ask put before it: the log is said under NoRun. + Run string `json:"run,omitempty"` +} + +// NoRun is the run of an ask that names none. +const NoRun = "none" + +// StepLogEvent is the event a build step's log is said as on its build module's own stream (BuildLogStream): +// log.., run NoRun when empty. +func StepLogEvent(run, step string) string { + if run == "" { + run = NoRun + } + return "log." + run + "." + step +} + +// StepLog is one event of a build step's log: its lines in order, Seq the first line's number from 1, so a reader sees +// order and gaps (novox/hq ADR 0157 clause 6), and the failure they are of, when they are a failed test's output. +type StepLog struct { + Seq int `json:"seq"` + Package string `json:"package,omitempty"` + Test string `json:"test,omitempty"` + Lines []string `json:"lines"` } // ComposeJob is a compose ask's job (novox/hq ADR 0482 clause 4, issue 1622): fetch the commit, build the composer diff --git a/go/stream.go b/go/stream.go new file mode 100644 index 0000000..de95724 --- /dev/null +++ b/go/stream.go @@ -0,0 +1,59 @@ +package stdio + +import ( + "encoding/json" + "fmt" +) + +// A module's own stream read back (novox/hq ADR 0363): every module may read the streams it declared under `streams`, +// and no other's, by the runtime's stream.read, a batch of direct gets with no consumer, as a log is read. A module's +// tools read with it what its stream holds (a build step's log, issue: the full failure output), within the stream's age. + +// StreamEntry is one event of a module's own stream: its sequence, its event's name (the subject after +// `mesh.stream..event.`), its body and when the bus took it. +type StreamEntry struct { + Sequence uint64 `json:"sequence"` + Event string `json:"event"` + Body json.RawMessage `json:"body"` + At string `json:"at"` +} + +// StreamPage is one page of a read: its entries in order, the sequence to read on from (0 when nothing follows) and the +// stream's last sequence when it was read. +type StreamPage struct { + Entries []StreamEntry `json:"entries"` + Next uint64 `json:"next"` + Last uint64 `json:"last"` +} + +// KeptStream is one stream this module declared. +type KeptStream struct{ name string } + +// Stream is a stream this module declared, by its local name. +func Stream(name string) KeptStream { return KeptStream{name: name} } + +// Read is the events from the sequence from on (0 is the first), whose event names match events (`*` for one name, +// `>` for the rest, last; empty is every event the stream carries), oldest first, at most limit (0 is the runtime's +// page, 100; at most 1000). Read on from the page's Next until it is 0. +func (s KeptStream) Read(events string, from uint64, limit int) (*StreamPage, error) { + if limit < 0 { + return nil, fmt.Errorf("a read of %s asks for at least no entries: limit %d", s.name, limit) + } + c, err := live() + if err != nil { + return nil, err + } + params := map[string]any{"stream": s.name, "from": from} + if events != "" { + params["subject"] = events + } + if limit > 0 { + params["limit"] = limit + } + raw, err := c.ask("mesh/stream.read", params, logTimeout) + if err != nil { + return nil, err + } + var p StreamPage + return &p, json.Unmarshal(raw, &p) +}