diff --git a/go/compose/frombus/frombus_test.go b/go/compose/frombus/frombus_test.go index 1a4b562..bb53330 100644 --- a/go/compose/frombus/frombus_test.go +++ b/go/compose/frombus/frombus_test.go @@ -14,6 +14,7 @@ import ( "os" "reflect" "slices" + "sort" "strings" "testing" @@ -724,6 +725,17 @@ func TestInput(t *testing.T) { } } } + // And the same through StatesReader, over the fake's states as a module's runtime gives them (one reader for + // mesh-provisioner and disk-load, the lead 2026-10-29). + viaStates, _, err := DataItems(ctx, StatesReader(statesOf(t, ctx, noSecrets), func(_ context.Context, url string) ([]byte, error) { + for _, body := range noSecrets.archives { + return body, nil + } + return nil, errors.New("no archive") + }), Mesh{}, Scope{Node: "n1"}) + if err != nil || !reflect.DeepEqual(viaStates, items) { + t.Errorf("through the states the data items are %v, %v; through the reader %v", viaStates, err, items) + } // And what a provider on another node serves from its settings (mailu's smtp domain, issue 1126: shanks refused // '${setting:domain} ... nothing sets "domain"'): read for that provider's served keys alone, through Served. @@ -769,3 +781,72 @@ func TestInput(t *testing.T) { } } } + +// mapStates are read states by name, key to value, as a module's runtime gives them. +type mapStates map[string]map[string][]byte + +func (m mapStates) Get(state, key string) ([]byte, bool, error) { + v, ok := m[state][key] + return v, ok, nil +} + +func (m mapStates) Keys(state string) ([]string, error) { + var out []string + for k := range m[state] { + out = append(out, k) + } + sort.Strings(out) + return out, nil +} + +// statesOf is b laid out as the states StatesReader reads. +func statesOf(t *testing.T, ctx context.Context, b *bus) mapStates { + t.Helper() + out := mapStates{} + put := func(state, key string, raw []byte, err error) { + if err != nil { + t.Fatal(err) + } + if out[state] == nil { + out[state] = map[string][]byte{} + } + out[state][key] = raw + } + placements, _ := b.Placements(ctx) + for node, p := range placements { + k, _ := delivery.PlacementKey(node) + raw, err := delivery.EncodePlacement(p) + put(PlacementRead, k, raw, err) + } + records, _ := b.Records(ctx) + for node, r := range records { + k, _ := delivery.HaveNodeKey(node) + rec := r + raw, err := delivery.EncodeHaveNode(delivery.HaveNode{Record: &rec}) + put(HaveRead, k, raw, err) + } + modules, _ := b.Modules(ctx) + for name, m := range modules { + raw, err := delivery.EncodeModule(m) + put(ModulesRead, name, raw, err) + var mf struct{ Module string } + _ = json.Unmarshal(b.manifests[m.Manifest], &mf) + k, _ := delivery.ManifestsKey(m.Manifest) + raw, err = delivery.EncodeManifest(delivery.Manifest{Module: mf.Module, JSON: string(b.manifests[m.Manifest])}) + put(ManifestsRead, k, raw, err) + } + for key, build := range b.builds { + module, fp, _ := strings.Cut(key, " ") + k, _ := delivery.BuilderBuildKey(module, fp) + raw, err := delivery.EncodeBuilderBuild(*build) + put(BuildsRead, k, raw, err) + } + for nm, values := range b.values { + for k, v := range values { + value, _ := json.Marshal(v) + raw, err := json.Marshal(map[string]any{"value": json.RawMessage(value), "source": catalogue.SourceModule}) + put(ResolvedRead, nm+"."+k, raw, err) + } + } + return out +} diff --git a/go/compose/frombus/states.go b/go/compose/frombus/states.go new file mode 100644 index 0000000..9962cb9 --- /dev/null +++ b/go/compose/frombus/states.go @@ -0,0 +1,199 @@ +package frombus + +// **The bus read through a module's declared `reads`** (the lead, 2026-10-29: one reader in place of the copies in +// mesh-provisioner and disk-load): a Reader over the states a module reads, by a runtime's Get and Keys (a bundle's +// stdio.State, a process's own), for a holder that composes plans or data items and reads no value: placement, each +// node's record in its own have key, the indexer's modules and manifests, the builder's builds and mesh-config's resolved +// values. No own secret and no credential is read (compose.Planning). Pure as the rest of this package: a bundle's +// runtime states and an HTTP fetch are compose/fromruntime's. + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "fmt" + "strings" + "time" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/compose" + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// The states a module names in its manifest's `reads` to compose through StatesReader. +const ( + PlacementRead = delivery.PlacementRead + ModulesRead = delivery.IndexerModule + ".modules" + ManifestsRead = delivery.IndexerModule + ".manifests" + BuildsRead = delivery.BuilderModule + ".builds" + HaveRead = "node-engine.have" + ResolvedRead = "mesh-config.resolved" +) + +// StatesReads are those reads, in a manifest's order. +var StatesReads = []string{PlacementRead, ModulesRead, ManifestsRead, BuildsRead, HaveRead, ResolvedRead} + +// States is a module's read states as its runtime gives them: a key's value, false when it has none. +type States interface { + Get(state, key string) (value []byte, found bool, err error) + Keys(state string) ([]string, error) +} + +// Fetch reads an archive at a URL (fromruntime.HTTPFetch over HTTP). +type Fetch func(ctx context.Context, url string) ([]byte, error) + +// StatesReader is the Reader over s; fetch reads a templates archive, nil for a holder that reads none. +func StatesReader(s States, fetch Fetch) Reader { return statesReader{s: s, fetch: fetch} } + +type statesReader struct { + s States + fetch Fetch +} + +func (r statesReader) each(read string, visit func(key string, value []byte)) error { + keys, err := r.s.Keys(read) + if err != nil { + return fmt.Errorf("%s: %w", read, err) + } + for _, k := range keys { + v, found, err := r.s.Get(read, k) + if err != nil { + return fmt.Errorf("%s %s: %w", read, k, err) + } + if found { + visit(k, v) + } + } + return nil +} + +func (r statesReader) Placements(context.Context) (map[string]delivery.Placement, error) { + out := map[string]delivery.Placement{} + err := r.each(PlacementRead, func(k string, v []byte) { + if node, err := delivery.ParsePlacementKey(k); err == nil { + if p, err := delivery.DecodePlacement(v); err == nil && p.Node == node { + out[node] = p + } + } + }) + return out, err +} + +func (r statesReader) Records(context.Context) (map[string]delivery.NodeRecord, error) { + out := map[string]delivery.NodeRecord{} + err := r.each(HaveRead, func(k string, v []byte) { + if node, _, isNode, err := delivery.ParseHaveKey(k); err == nil && isNode { + if h, err := delivery.DecodeHaveNode(v); err == nil && h.Record != nil { + out[node] = *h.Record + } + } + }) + return out, err +} + +func (r statesReader) Modules(context.Context) (map[string]delivery.Module, error) { + out := map[string]delivery.Module{} + err := r.each(ModulesRead, func(k string, v []byte) { + if module, err := delivery.ParseModulesKey(k); err == nil { + if m, err := delivery.DecodeModule(v); err == nil { + out[module] = m + } + } + }) + return out, err +} + +func (r statesReader) Manifest(_ context.Context, digest string) ([]byte, error) { + key, err := delivery.ManifestsKey(digest) + if err != nil { + return nil, err + } + v, found, err := r.s.Get(ManifestsRead, key) + if err != nil { + return nil, err + } + if !found { + return nil, fmt.Errorf("the indexer holds no manifest %s", digest) + } + m, err := delivery.DecodeManifest(v) + if err != nil { + return nil, err + } + return []byte(m.JSON), nil +} + +func (r statesReader) Values(_ context.Context, node, module string) (map[string]any, error) { + var out map[string]any + prefix := node + "." + module + "." + err := r.each(ResolvedRead, func(k string, v []byte) { + name, ok := strings.CutPrefix(k, prefix) + if !ok || strings.Contains(name, ".") { + return + } + if value, laid := catalogue.LaidValue(v); laid { + if out == nil { + out = map[string]any{} + } + out[name] = value + } + }) + return out, err +} + +func (statesReader) OwnSecrets(context.Context, string) ([]compose.OwnSecretRow, error) { + return nil, nil +} + +func (statesReader) Credentials(context.Context, string, []compose.CredentialKey, []compose.CredentialKey) ([]compose.Credential, []compose.Issued, error) { + return nil, nil, nil +} + +func (r statesReader) Build(_ context.Context, module, fp string) (*delivery.BuilderBuild, error) { + key, err := delivery.BuilderBuildKey(module, fp) + if err != nil { + return nil, err + } + v, found, err := r.s.Get(BuildsRead, key) + if err != nil || !found { + return nil, err + } + b, err := delivery.DecodeBuilderBuild(v) + if err != nil { + return nil, err + } + return &b, nil +} + +// archiveWithin bounds one read of a templates archive. +const archiveWithin = 30 * time.Second + +func (r statesReader) Archive(ctx context.Context, url, digest string) ([]byte, error) { + if r.fetch == nil { + return nil, fmt.Errorf("this holder reads no archive (%s)", url) + } + fetching, cancel := context.WithTimeout(ctx, archiveWithin) + defer cancel() + body, err := r.fetch(fetching, url) + if err != nil { + return nil, err + } + sum := sha256.Sum256(body) + if got := hex.EncodeToString(sum[:]); got != strings.TrimPrefix(digest, "sha256:") { + return nil, fmt.Errorf("%s was recorded as %s and what arrived is %s", url, digest, got) + } + return body, nil +} + +func (r statesReader) Served(ctx context.Context, node, module string, keys []string) (map[string]any, error) { + all, err := r.Values(ctx, node, module) + var out map[string]any + for _, k := range keys { + if v, ok := all[k]; ok { + if out == nil { + out = map[string]any{} + } + out[k] = v + } + } + return out, err +} diff --git a/go/compose/fromruntime/fromruntime.go b/go/compose/fromruntime/fromruntime.go new file mode 100644 index 0000000..a2a98e0 --- /dev/null +++ b/go/compose/fromruntime/fromruntime.go @@ -0,0 +1,49 @@ +// Package fromruntime is compose/frombus's StatesReader over a bundle's runtime (the lead, 2026-10-29): the states a +// module's manifest names in `reads` (frombus.StatesReads), read through stdio.State, and a templates archive fetched +// over HTTP. Apart from frombus so that frombus stays pure for the mesh-compose command. +package fromruntime + +import ( + "context" + "fmt" + "io" + "net/http" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/compose/frombus" +) + +// States are a bundle's read states through its runtime. +type States struct{} + +// Get is a key's value, false when it has none. +func (States) Get(state, key string) ([]byte, bool, error) { + e, err := stdio.State(state).Get(key) + if err != nil || e == nil { + return nil, false, err + } + return e.Value, true, nil +} + +// Keys are a state's keys. +func (States) Keys(state string) ([]string, error) { return stdio.State(state).Keys() } + +// Reader is frombus's Reader over this bundle's read states, fetching archives over HTTP. +func Reader() frombus.Reader { return frombus.StatesReader(States{}, HTTPFetch) } + +// HTTPFetch reads an archive over HTTP, at most 256 MiB. +func HTTPFetch(ctx context.Context, url string) ([]byte, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + if err != nil { + return nil, err + } + resp, err := http.DefaultClient.Do(req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("%s answered %s", url, resp.Status) + } + return io.ReadAll(io.LimitReader(resp.Body, 256<<20)) +}