Let a module read its own streams back by a batch direct get, as it reads its log, so its tools can answer what the stream holds (a build step's whole failure output)

This commit is contained in:
2026-10-30 10:54:05 +01:00
parent 95a12105cf
commit b90376ba59
7 changed files with 255 additions and 39 deletions
+42 -1
View File
@@ -21,6 +21,8 @@ func logBuses(t *testing.T, maxBytes int64) (issues, other launch.Bus, mesh *mt.
mesh = mt.New(t)
m := mt.MembershipOf("issues", "anchor", false, nil)
m.Logs = []bus.LogIssued{mesh.Log(t, "issues", "changes", maxBytes)}
mesh.ModuleStream(t, "issues", "ISSUE_MOVES")
m.Streams = []bus.StreamIssued{{Stream: "ISSUE_MOVES", Events: []string{"moved.>"}, Subject: "mesh.stream.issues.event.{event}"}}
mesh.Issue(t, m)
mesh.Issue(t, mt.MembershipOf("other", "anchor", false, nil))
conn := connect(t, "node-tools", "anchor")
@@ -111,7 +113,7 @@ func infoOf(t *testing.T, b launch.Bus) map[string]uint64 {
// RFC 3339; last is a key's last entry or null; info says entries, bytes, cap, first and last — and no
// consumer is made for any of it.
func TestALogAnswersEachVerbOnTheBus(t *testing.T) {
b, _, mesh := logBuses(t, 1<<20)
b, other, mesh := logBuses(t, 1<<20)
before := time.Now().Add(-time.Second)
var seqs []uint64
@@ -189,6 +191,45 @@ func TestALogAnswersEachVerbOnTheBus(t *testing.T) {
if n := mesh.Consumers(t, "LOG_issues_changes"); n != 0 {
t.Fatalf("the runtime made %d consumers of the log; it uses none", n)
}
// A module reads its own stream back as it reads its log (mesh/stream.read): by its local name, an event pattern,
// in pages; another module's stream is refused with the reason.
for i, ev := range []string{"moved.a.1", "moved.b.1", "moved.a.1"} {
if err := b.Publish(json.RawMessage(fmt.Sprintf(`{"key":%q,"body":{"n":%d}}`, ev, i))); err != nil {
t.Fatal(err)
}
}
streamOf := func(b launch.Bus, params string) (wireStream, error) {
var p wireStream
raw, err := b.StreamRead(json.RawMessage(params))
if err == nil {
err = json.Unmarshal(raw, &p)
}
return p, err
}
page, err := streamOf(b, `{"stream":"issue-moves","subject":"moved.a.>","limit":1}`)
if err != nil || len(page.Entries) != 1 || page.Entries[0].Event != "moved.a.1" || string(page.Entries[0].Body) != `{"n":0}` || page.Next == 0 || page.Last != 3 {
t.Fatalf("the first page of moved.a: %+v, %v", page, err)
}
page, err = streamOf(b, fmt.Sprintf(`{"stream":"ISSUE_MOVES","subject":"moved.a.>","from":%d}`, page.Next))
if err != nil || len(page.Entries) != 1 || string(page.Entries[0].Body) != `{"n":2}` || page.Next != 0 {
t.Fatalf("the second page of moved.a: %+v, %v", page, err)
}
if _, err := streamOf(other, `{"stream":"issue-moves"}`); err == nil || !strings.Contains(err.Error(), "keeps no stream") {
t.Fatalf("another module's stream: %v", err)
}
}
// wireStream is a stream read's answer with exactly the contract's field names (mesh-sdk stdio.StreamPage).
type wireStream struct {
Entries []struct {
Sequence uint64 `json:"sequence"`
Event string `json:"event"`
Body json.RawMessage `json:"body"`
At string `json:"at"`
} `json:"entries"`
Next uint64 `json:"next"`
Last uint64 `json:"last"`
}
// A page holds about 4 MiB at most, whatever its limit, and says where to continue.