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) { keys := delivery.NewPlacementKeys() err := r.each(PlacementRead, func(k string, v []byte) { _, _ = keys.Put(k, v, 0, false) // a key that does not read places nothing, as before }) return keys.All(), 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 }