diff --git a/go/compose/fromruntime/fromruntime.go b/go/compose/fromruntime/fromruntime.go index a2a98e0..ae0f323 100644 --- a/go/compose/fromruntime/fromruntime.go +++ b/go/compose/fromruntime/fromruntime.go @@ -28,9 +28,30 @@ func (States) Get(state, key string) ([]byte, bool, error) { // 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. +// 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 archives over HTTP. +func WatchedReader(w *stdio.WatchedStates) frombus.Reader { + return frombus.StatesReader(Watched{w}, 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) diff --git a/go/state_test.go b/go/state_test.go index cac843b..bef4cbd 100644 --- a/go/state_test.go +++ b/go/state_test.go @@ -230,4 +230,45 @@ func TestABundleKeepsAndWatchesStateThroughTheRuntime(t *testing.T) { t.Fatalf("the get answered %v", r) } _ = inW.Close() + + // ADR 0580 clause 1: the watched copy answers keys and values after the watch's current values, a later put or delete + // changes its answer, a follower's then sees the copy that took its change, each state is watched once, and a + // value kept once is read by key once. + handles, watches, gets := map[string]func(StateChange) error{}, map[string]int{}, 0 + w := WatchedStatesOver(func(state string, handle func(StateChange) error) error { + watches[state]++ + handles[state] = handle + return handle(StateChange{State: state, Key: "a", Op: OpPut, Value: []byte("1"), Current: true}) + }, func(state, key string) ([]byte, bool, error) { gets++; return []byte(key), key != "missing", nil }) + var thenSaw []string + w.Follow("placement", func(c StateChange) error { + v, found, _ := w.Get("placement", c.Key) + thenSaw = append(thenSaw, fmt.Sprintf("%s %s %s %v", c.Op, c.Key, v, found)) + return nil + }) + if keys, err := w.Keys("placement"); err != nil || fmt.Sprint(keys) != "[a]" { + t.Fatalf("the copy's keys are %v (%v)", keys, err) + } + _ = handles["placement"](StateChange{Key: "b", Op: OpPut, Value: []byte("2")}) + _ = handles["placement"](StateChange{Key: "a", Op: OpDelete}) + vals, _ := w.Values("placement") + if v, found, _ := w.Get("placement", "a"); found || string(vals["b"]) != "2" || len(vals) != 1 || v != nil { + t.Fatalf("after a put and a delete the copy is %v", vals) + } + if fmt.Sprint(thenSaw) != "[put a 1 true put b 2 true delete a false]" { + t.Fatalf("then saw %v", thenSaw) + } + _, _ = w.Keys("placement") + if watches["placement"] != 1 { + t.Fatalf("the state was watched %d times", watches["placement"]) + } + for range 2 { + if v, found, _ := w.Once("manifests", "d1"); !found || string(v) != "d1" { + t.Fatalf("once read %q %v", v, found) + } + _, _, _ = w.Once("manifests", "missing") + } + if gets != 3 { + t.Fatalf("a kept value was read by key %d times, want once, and the missing one twice", gets) + } } diff --git a/go/watched.go b/go/watched.go new file mode 100644 index 0000000..1cb7578 --- /dev/null +++ b/go/watched.go @@ -0,0 +1,216 @@ +package stdio + +import ( + "fmt" + "sort" + "sync" + "time" +) + +// WatchedStates is a copy of the states a module reads, each filled by its own watch and kept by its changes (novox/hq +// ADR 0580 clause 1): a read of a whole state is one watch's current values, never a get per key, which took a +// composition past 30 seconds and a compare past its call bound. A state is watched on its first read, or on Follow, +// and the watch hands every current value over before the read is answered, so the copy is complete from the first +// answer. A key whose value never changes once written (a manifest, by its digest) is read once with Once and kept. +// A state helper, no rule: what a module reads is its own. +type WatchedStates struct { + watch func(state string, handle func(StateChange) error) error + get func(state, key string) ([]byte, bool, error) + + mu sync.Mutex + states map[string]*watchedState + once map[string][]byte +} + +type watchedState struct { + ready chan struct{} // closed once the watch's current values are all in the copy, or it was refused + err error + + mu sync.Mutex + values map[string][]byte + then []func(StateChange) error + started bool // the current values are in, so a change runs then at once + pending []StateChange // the current values, run through then once all are in + thenMu sync.Mutex // then runs one change at a time +} + +// NewWatchedStates is the copy over this bundle's runtime: each state by KeptState.Watch, Once by KeptState.Get. +func NewWatchedStates() *WatchedStates { + return WatchedStatesOver( + func(state string, handle func(StateChange) error) error { return State(state).Watch("", handle) }, + func(state, key string) ([]byte, bool, error) { + e, err := State(state).Get(key) + if err != nil || e == nil { + return nil, false, err + } + return e.Value, true, nil + }) +} + +// WatchedStatesOver is the copy over a watch and a get of one's own (a process's bus, a test's): watch hands handle +// every current value and returns once they are handled, then hands every change; get reads one key. +func WatchedStatesOver(watch func(state string, handle func(StateChange) error) error, + get func(state, key string) ([]byte, bool, error)) *WatchedStates { + return &WatchedStates{watch: watch, get: get, states: map[string]*watchedState{}, once: map[string][]byte{}} +} + +// watched is the state's copy, its watch started now when it has none, the caller waiting until its current values +// are in. then, when given, runs after the copy took each change: the current values once they are all in (a then +// that reads the copy reads it whole), each later change as it comes. No lock is held while the watch starts, so a +// then may read any state of the copy. +func (w *WatchedStates) watched(state string, then func(StateChange) error) (*watchedState, error) { + w.mu.Lock() + s, ok := w.states[state] + if ok { + w.mu.Unlock() + <-s.ready + if s.err != nil { + return nil, s.err + } + if then != nil { + s.mu.Lock() + s.then = append(s.then, then) + s.mu.Unlock() + } + return s, nil + } + s = &watchedState{ready: make(chan struct{}), values: map[string][]byte{}} + if then != nil { + s.then = []func(StateChange) error{then} + } + w.states[state] = s + w.mu.Unlock() + err := w.watch(state, func(c StateChange) error { + s.mu.Lock() + if c.Op == OpPut { + s.values[c.Key] = append([]byte(nil), c.Value...) + } else { + delete(s.values, c.Key) + } + if !s.started { + s.pending = append(s.pending, c) + s.mu.Unlock() + return nil + } + s.mu.Unlock() + return s.run(state, c) + }) + if err != nil { + s.err = fmt.Errorf("the state %s is not watched: %w", state, err) + w.mu.Lock() + delete(w.states, state) // asked again on the next read + w.mu.Unlock() + close(s.ready) + return nil, s.err + } + s.mu.Lock() + s.started = true + pending := s.pending + s.pending = nil + s.mu.Unlock() + close(s.ready) + for _, c := range pending { + if err := s.run(state, c); err != nil { + sayf("%v", err) + } + } + return s, nil +} + +// run hands one change, already in the copy, to every then. +func (s *watchedState) run(state string, c StateChange) error { + s.thenMu.Lock() + defer s.thenMu.Unlock() + s.mu.Lock() + then := append([]func(StateChange) error(nil), s.then...) + s.mu.Unlock() + var errs []error + for _, f := range then { + if err := f(c); err != nil { + errs = append(errs, err) + } + } + if len(errs) > 0 { + return fmt.Errorf("%s %s: %v", state, c.Key, errs) + } + return nil +} + +// Follow watches the state into the copy and runs then on each change after the copy took it, so a reader then wakes +// never reads a copy older than its change; it asks again until the watch is taken (2 s, doubled to a minute), as +// KeptState.Follow does. On a state already watched then hears the changes from now on. +func (w *WatchedStates) Follow(state string, then func(StateChange) error) { + for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { + if _, err := w.watched(state, then); err == nil { + return + } else { + sayf("%v; asking again in %s", err, wait) + } + time.Sleep(wait) + } +} + +// Get is a key's value in the copy, false when it has none. +func (w *WatchedStates) Get(state, key string) ([]byte, bool, error) { + s, err := w.watched(state, nil) + if err != nil { + return nil, false, err + } + s.mu.Lock() + defer s.mu.Unlock() + v, ok := s.values[key] + return v, ok, nil +} + +// Keys are the state's keys in the copy, sorted. +func (w *WatchedStates) Keys(state string) ([]string, error) { + s, err := w.watched(state, nil) + if err != nil { + return nil, err + } + s.mu.Lock() + defer s.mu.Unlock() + keys := make([]string, 0, len(s.values)) + for k := range s.values { + keys = append(keys, k) + } + sort.Strings(keys) + return keys, nil +} + +// Values is every key's value in the copy. +func (w *WatchedStates) Values(state string) (map[string][]byte, error) { + s, err := w.watched(state, nil) + if err != nil { + return nil, err + } + s.mu.Lock() + defer s.mu.Unlock() + out := make(map[string][]byte, len(s.values)) + for k, v := range s.values { + out[k] = v + } + return out, nil +} + +// Once is a key whose value never changes once written, read by key the first time it is found and kept after: a +// manifest by its digest. A key not found is not kept, so it is read again on the next ask. +func (w *WatchedStates) Once(state, key string) ([]byte, bool, error) { + w.mu.Lock() + v, ok := w.once[state+" "+key] + w.mu.Unlock() + if ok { + return v, true, nil + } + v, found, err := w.get(state, key) + if err != nil || !found { + return nil, false, err + } + w.mu.Lock() + w.once[state+" "+key] = v + w.mu.Unlock() + return v, true, nil +} + +// Direct is a key read by key now, not from the copy: a value that changes and is read one at a time (a build). +func (w *WatchedStates) Direct(state, key string) ([]byte, bool, error) { return w.get(state, key) }