RN17 (show/kept fill): show and kept carry each store's fill, read through the bundle's calls
Builds: design 47 G §366, §623, A-214 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
@@ -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...) })...)
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user