diff --git a/go/compose/frombus/frombustest/frombustest_test.go b/go/compose/frombus/frombustest/frombustest_test.go index c6e3493..84ec290 100644 --- a/go/compose/frombus/frombustest/frombustest_test.go +++ b/go/compose/frombus/frombustest/frombustest_test.go @@ -157,6 +157,16 @@ func TestTheProvisionedMeshHandsAPlanWithItsProvision(t *testing.T) { byKey[frombus.PlacementRead+" watched"] != 1 { t.Errorf("the watched reader read %v, want a build by key each time, a manifest once, the placement watched once", byKey) } + fetched := 0 + keep := fromruntime.KeptFetch(func(context.Context, string) ([]byte, error) { fetched++; return []byte("tar"), nil }) + for range 2 { + if body, err := keep(context.Background(), "http://store/restic/files/blobs/sha256:a"); err != nil || string(body) != "tar" { + t.Fatalf("the kept fetch answered %q, %v", body, err) + } + } + if fetched != 1 { + t.Errorf("a templates archive was fetched %d times, want once", fetched) + } } func planned[P any](plans map[string]P, module string) bool { _, ok := plans[module]; return ok } diff --git a/go/compose/fromruntime/fromruntime.go b/go/compose/fromruntime/fromruntime.go index ae0f323..997fa5a 100644 --- a/go/compose/fromruntime/fromruntime.go +++ b/go/compose/fromruntime/fromruntime.go @@ -8,6 +8,7 @@ import ( "fmt" "io" "net/http" + "sync" stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/compose/frombus" @@ -47,9 +48,39 @@ func (w Watched) Get(state, key string) ([]byte, bool, error) { return w.WatchedStates.Get(state, key) } -// WatchedReader is frombus's Reader over w, fetching archives over HTTP. +// 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}, HTTPFetch) + 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. diff --git a/go/state_test.go b/go/state_test.go index bef4cbd..e8bdc2a 100644 --- a/go/state_test.go +++ b/go/state_test.go @@ -262,6 +262,18 @@ func TestABundleKeepsAndWatchesStateThroughTheRuntime(t *testing.T) { if watches["placement"] != 1 { t.Fatalf("the state was watched %d times", watches["placement"]) } + // A read that starts a state's watch before its follower joins loses that follower no change: the follower is run over + // the copy's current values when it joins, then hears every later change (review of ADR 0580 §2). + if _, _, err := w.Get("haves", "n1.a"); err != nil { + t.Fatal(err) + } + _ = handles["haves"](StateChange{Key: "n1.b", Op: OpPut, Value: []byte("2")}) // in the gap: no follower yet + var joined []string + w.Follow("haves", func(c StateChange) error { joined = append(joined, c.Key); return nil }) + _ = handles["haves"](StateChange{Key: "n2.c", Op: OpPut, Value: []byte("3")}) + if fmt.Sprint(joined) != "[a n1.b n2.c]" || watches["haves"] != 1 { + t.Fatalf("a follower joining a watched state heard %v (watched %d times)", joined, watches["haves"]) + } for range 2 { if v, found, _ := w.Once("manifests", "d1"); !found || string(v) != "d1" { t.Fatalf("once read %q %v", v, found) diff --git a/go/watched.go b/go/watched.go index 1cb7578..1f7953e 100644 --- a/go/watched.go +++ b/go/watched.go @@ -19,9 +19,14 @@ type WatchedStates struct { mu sync.Mutex states map[string]*watchedState - once map[string][]byte + // once holds each value read with Once, by state and key, for the process's life: a few kilobytes per manifest + // digest. Past onceAtMost keys it is emptied, which costs only a read again. + once map[string][]byte } +// onceAtMost bounds the values Once keeps. +const onceAtMost = 4096 + type watchedState struct { ready chan struct{} // closed once the watch's current values are all in the copy, or it was refused err error @@ -68,9 +73,7 @@ func (w *WatchedStates) watched(state string, then func(StateChange) error) (*wa return nil, s.err } if then != nil { - s.mu.Lock() - s.then = append(s.then, then) - s.mu.Unlock() + s.join(state, then) } return s, nil } @@ -117,6 +120,28 @@ func (w *WatchedStates) watched(state string, then func(StateChange) error) (*wa return s, nil } +// join adds then to a state already watched (a read started its watch first) and runs it over the copy's current values +// before any later change, so no change that landed before it joined is lost to it (review of ADR 0580 §2: a read +// beat the follower to the watch at start-up). Under thenMu, so a later change waits for these; a change the copy took +// just before the snapshot may also be handed again, which a then that wakes work takes as a second wake. +func (s *watchedState) join(state string, then func(StateChange) error) { + s.thenMu.Lock() + defer s.thenMu.Unlock() + s.mu.Lock() + s.then = append(s.then, then) + current := make([]StateChange, 0, len(s.values)) + for k, v := range s.values { + current = append(current, StateChange{State: state, Key: k, Op: OpPut, Value: v, Current: true}) + } + s.mu.Unlock() + sort.Slice(current, func(i, j int) bool { return current[i].Key < current[j].Key }) + for _, c := range current { + if err := then(c); err != nil { + sayf("%s %s: %v", state, c.Key, err) + } + } +} + // run hands one change, already in the copy, to every then. func (s *watchedState) run(state string, c StateChange) error { s.thenMu.Lock() @@ -207,6 +232,9 @@ func (w *WatchedStates) Once(state, key string) ([]byte, bool, error) { return nil, false, err } w.mu.Lock() + if len(w.once) >= onceAtMost { + clear(w.once) + } w.once[state+" "+key] = v w.mu.Unlock() return v, true, nil