From 50f3dde9f5822d0a02f973eedf4557009a567caf Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 06:07:59 +0100 Subject: [PATCH] RN17 (show/kept fill): show and kept carry each store's fill, read through the bundle's calls MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G §366, §623, A-214 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../cmd/mesh-delivery-pipeline-tools/main.go | 43 ++++++++++++++++++- .../mesh-delivery-pipeline-tools/main_test.go | 31 +++++++++++++ .../internal/pipeline/read.go | 34 ++++++++++++++- .../internal/pipeline/read_test.go | 23 ++++++++++ 4 files changed, 128 insertions(+), 3 deletions(-) diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main.go index 444dc4b8..be77f16b 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main.go @@ -10,6 +10,8 @@ import ( "time" "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient" + "git.novox.be/novox/mesh-sdk/go/buscalls" + "git.novox.be/novox/mesh-sdk/go/busstate" "git.novox.be/novox/mesh-sdk/go/views" seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" @@ -108,8 +110,47 @@ func HealthTool(pipelines, consumers pipeline.Kept, now func() time.Time) stdio. Run: func(map[string]any) (any, error) { return pipeline.Health(pipelines, consumers, now()), nil }} } +// StoresFill is each of the runner's stores' fill through the bundle's calls (design 47 G §366, §623, A-214): the states +// pipelines and kept by `mesh/state.info`, step-outputs by the object store's fill and step-log by the log's; the runner's +// process writes nothing for it. A store whose fill cannot be read is an error naming it. +func StoresFill(state func(string) (*stdio.StateInfo, error), objects func(string) (buscalls.Fill, error), + log func(string) (*stdio.LogInfo, error), now func() time.Time) func() ([]seat.Fill, error) { + return func() ([]seat.Fill, error) { + at := now().UTC() + var out []seat.Fill + for _, name := range []string{seat.StatePipelines, seat.StateStepOutputs, seat.StateKept, seat.StateStepLog} { + f := seat.Fill{Store: name, At: &at} + switch name { + case seat.StateStepOutputs: + o, err := objects(name) + if err != nil { + return nil, fmt.Errorf("%s: %w", name, err) + } + f.Bytes, f.Cap = int64(o.Bytes), int64(o.MaxBytes) + case seat.StateStepLog: + l, err := log(name) + if err != nil { + return nil, fmt.Errorf("%s: %w", name, err) + } + f.Bytes, f.Cap = int64(l.Bytes), int64(l.MaxBytes) + default: + s, err := state(name) + if err != nil { + return nil, fmt.Errorf("%s: %w", name, err) + } + f.Bytes, f.Cap = int64(s.Bytes), s.MaxBytes + } + out = append(out, f) + } + return out, nil + } +} + func main() { - r := pipeline.Reads{Pipelines: kept{stdio.State(seat.StatePipelines)}, Kept: kept{stdio.State(seat.StateKept)}} + fill := StoresFill(func(n string) (*stdio.StateInfo, error) { return stdio.State(n).Info() }, + func(n string) (buscalls.Fill, error) { return busstate.ObjectsOf(n).Fill() }, + func(n string) (*stdio.LogInfo, error) { return stdio.Log(n).Info() }, time.Now) + r := pipeline.Reads{Pipelines: kept{stdio.State(seat.StatePipelines)}, Kept: kept{stdio.State(seat.StateKept)}, Fill: fill} tools := append(Tools(r), FirstViewTools(stdio.State(views.StatusState), meshhandactsclient.Recorder{Ask: stdio.Ask}, time.Now)...) tools = append(tools, UnloggedTools(stdio.State(seat.StatePipelines), stdio.Log(seat.StateStepLog), meshhandactsclient.Recorder{Ask: stdio.Ask}, func(f string, a ...any) { fmt.Fprintf(os.Stderr, "[mesh-delivery-pipeline-tools] "+f+"\n", a...) })...) diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main_test.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main_test.go index fddd68ee..38e78f32 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main_test.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main_test.go @@ -3,8 +3,10 @@ package main import ( "encoding/json" "errors" + "git.novox.be/novox/mesh-sdk/go/buscalls" "strings" "testing" + "time" stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/views" @@ -58,3 +60,32 @@ func TestUnloggedAppendIsAHandAct(t *testing.T) { t.Errorf("an append whose hand act is not recorded: %v", err) } } + +// StoresFill reads the two states through the state call, the object store and the log through their own, in the seat's order, +// and names a store it could not read (design 47 G §366, §623, A-214). +func TestStoresFillIsReadThroughTheBundlesCalls(t *testing.T) { + now := func() time.Time { return time.Date(2026, 11, 12, 8, 0, 0, 0, time.UTC) } + var asked []string + fill := StoresFill( + func(n string) (*stdio.StateInfo, error) { + asked = append(asked, "state:"+n) + return &stdio.StateInfo{Bytes: 5, MaxBytes: 64}, nil + }, + func(n string) (buscalls.Fill, error) { + asked = append(asked, "objects:"+n) + return buscalls.Fill{Bytes: 7, MaxBytes: 70}, nil + }, + func(n string) (*stdio.LogInfo, error) { + asked = append(asked, "log:"+n) + return &stdio.LogInfo{Bytes: 9, MaxBytes: 90}, nil + }, now) + got, err := fill() + want := "state:pipelines objects:step-outputs state:kept log:step-log" + if err != nil || strings.Join(asked, " ") != want || len(got) != 4 || got[1].Bytes != 7 || got[3].Cap != 90 || got[2].Store != "kept" { + t.Fatalf("fill %+v, %v, asked %v; want %q", got, err, asked, want) + } + failing := StoresFill(func(string) (*stdio.StateInfo, error) { return nil, errors.New("no bus") }, nil, nil, now) + if _, err := failing(); err == nil || !strings.Contains(err.Error(), "pipelines") { + t.Errorf("a store unread: %v", err) + } +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/read.go b/modules/mesh-delivery-pipeline/internal/pipeline/read.go index ced5a604..b244afbb 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/read.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/read.go @@ -22,6 +22,21 @@ type Kept interface { type Reads struct { Pipelines Kept Kept Kept + // Fill is each of the runner's stores' fill now, pipelines, step-outputs, kept and step-log, read through the bundle's calls + // (the state's `mesh/state.info`, the object store's and the log's fill; design 47 G §366, §623, A-214); nil says none. + Fill func() ([]seat.Fill, error) +} + +// fills is the stores' fill, or an error saying it could not be read: an answer never leaves it out unsaid. +func (r Reads) fills() ([]seat.Fill, error) { + if r.Fill == nil { + return nil, nil + } + f, err := r.Fill() + if err != nil { + return nil, fmt.Errorf("the stores' fill cannot be read: %w", err) + } + return f, nil } // Show is one pipeline whole, with its steps and the names of their outputs in step-outputs (§83). @@ -33,7 +48,11 @@ func (r Reads) Show(in seat.ShowInput) (seat.ShowAnswer, error) { if !ok { return seat.ShowAnswer{}, fmt.Errorf("no pipeline %q in %s", in.Pipeline, seat.StatePipelines) } - a := seat.ShowAnswer{Pipeline: p} + fills, err := r.fills() + if err != nil { + return seat.ShowAnswer{}, err + } + a := seat.ShowAnswer{Pipeline: p, Fill: fills} for _, s := range p.Steps { if s.Output != "" { a.Outputs = append(a.Outputs, s.Output) @@ -139,7 +158,18 @@ func (r Reads) KeptResults(in seat.KeptInput) (seat.KeptAnswer, error) { } out = append(out, v) } - return seat.KeptAnswer{Kept: out}, nil + fills, err := r.fills() + if err != nil { + return seat.KeptAnswer{}, err + } + a := seat.KeptAnswer{Kept: out} + for _, f := range fills { + if f.Store == seat.StateKept { + f := f + a.Fill = &f + } + } + return a, nil } func (r Reads) pipeline(key string) (seat.Pipeline, bool, error) { diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/read_test.go b/modules/mesh-delivery-pipeline/internal/pipeline/read_test.go index 601e639c..20c544ab 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/read_test.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/read_test.go @@ -2,6 +2,7 @@ package pipeline import ( "encoding/json" + "errors" "strings" "testing" "time" @@ -127,3 +128,25 @@ func TestReads(t *testing.T) { } } } + +// show carries each store's fill (pipelines, step-outputs, kept, step-log) and kept the kept store's, read through the bundle's state +// call, never counted from the keys; a fill that cannot be read is an error, not an answer without it (design 47 G §366, §623, A-214). +func TestShowAndKeptCarryTheStoresFill(t *testing.T) { + pipes, kept := memKept{}, memKept{} + put(t, pipes, "main.7.alpha", seat.Pipeline{Pipeline: "main.7.alpha", Module: "alpha"}) + fills := []seat.Fill{{Store: seat.StatePipelines, Bytes: 10, Cap: 100}, {Store: seat.StateStepOutputs, Bytes: 20, Cap: 200}, + {Store: seat.StateKept, Bytes: 30, Cap: 300}, {Store: seat.StateStepLog, Bytes: 40, Cap: 400}} + r := Reads{Pipelines: pipes, Kept: kept, Fill: func() ([]seat.Fill, error) { return fills, nil }} + shown, err := r.Show(seat.ShowInput{Pipeline: "main.7.alpha"}) + if err != nil || len(shown.Fill) != 4 || shown.Fill[1].Store != seat.StateStepOutputs || shown.Fill[3].Cap != 400 { + t.Errorf("show: %+v, %v; want the four stores' fill", shown.Fill, err) + } + k, err := r.KeptResults(seat.KeptInput{}) + if err != nil || k.Fill == nil || k.Fill.Store != seat.StateKept || k.Fill.Bytes != 30 { + t.Errorf("kept: %+v, %v; want the kept store's fill", k.Fill, err) + } + r.Fill = func() ([]seat.Fill, error) { return nil, errors.New("the bus does not answer") } + if _, err := r.Show(seat.ShowInput{Pipeline: "main.7.alpha"}); err == nil || !strings.Contains(err.Error(), "fill") { + t.Errorf("show with the fill unread: %v", err) + } +}