jschoubben
790d1c82d0
mesh/delivery merged: source.main moved to novox/mesh-sdk@cad8250aaa8c
mesh/delivery-group group feat/l5-l7-main merged: every member is merged
mesh/check pass 790d1c82d027 on da41823855a9 | unit 1964 run, 5697 kept, 30 skipped | integration 1999 run, 5829 kept, 38 skipped | over bound: unit, …
A read could start a state's watch before its follower joined, and a change landing in that gap woke nothing (review of ADR 0580 §2): a joining follower now runs over the copy's current values before any later change. The watched reader keeps each archive once per URL, and both caches are bounded. Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL
102 lines
3.3 KiB
Go
102 lines
3.3 KiB
Go
// 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"
|
|
"sync"
|
|
|
|
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 key by key, fetching archives over HTTP. A holder that
|
|
// composes again and again reads through Watched instead (novox/hq ADR 0580).
|
|
func Reader() frombus.Reader { return frombus.StatesReader(States{}, HTTPFetch) }
|
|
|
|
// Watched is frombus's States over a watched copy (novox/hq ADR 0580 clause 1): the states a composition reads whole
|
|
// (placement, haves, modules, resolved values) from the copy, a build by key when asked, a manifest once per digest.
|
|
type Watched struct{ *stdio.WatchedStates }
|
|
|
|
// Get is a key's value: a build read now, a manifest kept once read, any other state's from the copy.
|
|
func (w Watched) Get(state, key string) ([]byte, bool, error) {
|
|
switch state {
|
|
case frombus.BuildsRead:
|
|
return w.Direct(state, key)
|
|
case frombus.ManifestsRead:
|
|
return w.Once(state, key)
|
|
}
|
|
return w.WatchedStates.Get(state, key)
|
|
}
|
|
|
|
// WatchedReader is frombus's Reader over w, fetching each templates archive over HTTP once (KeptFetch): a composition
|
|
// made again on every change reads it again otherwise.
|
|
func WatchedReader(w *stdio.WatchedStates) frombus.Reader {
|
|
return frombus.StatesReader(Watched{w}, KeptFetch(HTTPFetch))
|
|
}
|
|
|
|
// archivesAtMost bounds the archives KeptFetch keeps; past it they are emptied, which costs only a fetch again.
|
|
const archivesAtMost = 64
|
|
|
|
// KeptFetch is fetch kept once per URL: an archive's URL names its blob by digest, so its body never changes under it
|
|
// (frombus checks the digest on every read). A failed fetch is not kept.
|
|
func KeptFetch(fetch frombus.Fetch) frombus.Fetch {
|
|
var mu sync.Mutex
|
|
kept := map[string][]byte{}
|
|
return func(ctx context.Context, url string) ([]byte, error) {
|
|
mu.Lock()
|
|
body, ok := kept[url]
|
|
mu.Unlock()
|
|
if ok {
|
|
return body, nil
|
|
}
|
|
body, err := fetch(ctx, url)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
mu.Lock()
|
|
if len(kept) >= archivesAtMost {
|
|
clear(kept)
|
|
}
|
|
kept[url] = body
|
|
mu.Unlock()
|
|
return body, nil
|
|
}
|
|
}
|
|
|
|
// 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))
|
|
}
|