From db4eb3bbf43fd573a87704105a80fef3de5f2419 Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 1 Nov 2026 03:29:05 +0100 Subject: [PATCH 1/3] One watched copy of the states a module reads, and the composer's reader over it A whole state read key by key took the provisioner's composition of one node past 30 s and the assigner's compare past its call bound (ADR 0580 clause 1): each state is now one watch's current values, kept by its changes; a build is read by key and a manifest once per digest. Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL --- go/compose/fromruntime/fromruntime.go | 23 ++- go/state_test.go | 41 +++++ go/watched.go | 216 ++++++++++++++++++++++++++ 3 files changed, 279 insertions(+), 1 deletion(-) create mode 100644 go/watched.go 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) } From 8696aa9ba47985979f5976547b297e8c480cdf3a Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 1 Nov 2026 03:33:17 +0100 Subject: [PATCH 2/3] The hand-act recorder records through the seat alone, and the watched reader's reads are proved The controller's hand-act and warranted are gone (issue 1953 L5b) and the bus refuses them, so the recorder no longer falls back to them: an act the seat does not take is an error saying nothing was recorded. And ADR 0580 clause 1's reader is shown reading a build by key each time and a manifest once per digest. Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL --- .../frombus/frombustest/frombustest_test.go | 22 +++++++++++ go/handacts/handacts.go | 39 ++++++------------- go/handacts/handacts_test.go | 18 ++++----- 3 files changed, 41 insertions(+), 38 deletions(-) diff --git a/go/compose/frombus/frombustest/frombustest_test.go b/go/compose/frombus/frombustest/frombustest_test.go index 0db70b2..c6e3493 100644 --- a/go/compose/frombus/frombustest/frombustest_test.go +++ b/go/compose/frombus/frombustest/frombustest_test.go @@ -5,7 +5,9 @@ import ( "strings" "testing" + stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/compose/frombus" + "git.novox.be/novox/mesh-sdk/go/compose/fromruntime" "git.novox.be/novox/mesh-sdk/go/delivery" ) @@ -135,6 +137,26 @@ func TestTheProvisionedMeshHandsAPlanWithItsProvision(t *testing.T) { t.Errorf("%s: data items %v held %v: %v", row.name, items, iheld, err) } } + + // ADR 0580 clause 1: the watched reader reads a state it composes from out of the copy, a build by key each time it is + // asked, and a manifest once per digest. + byKey := map[string]int{} + ws := stdio.WatchedStatesOver(func(state string, handle func(stdio.StateChange) error) error { + byKey[state+" watched"]++ + return handle(stdio.StateChange{Key: "placement." + Node, Op: stdio.OpPut, Value: []byte(`{}`), Current: true}) + }, func(state, key string) ([]byte, bool, error) { byKey[state]++; return []byte(key), true, nil }) + w := fromruntime.Watched{WatchedStates: ws} + for range 2 { + _, _, _ = w.Get(frombus.BuildsRead, "restic."+fp) + _, _, _ = w.Get(frombus.ManifestsRead, "sha256-d") + if _, found, _ := w.Get(frombus.PlacementRead, "placement."+Node); !found { + t.Error("the watched reader does not answer the placement from its copy") + } + } + if byKey[frombus.BuildsRead] != 2 || byKey[frombus.ManifestsRead] != 1 || byKey[frombus.PlacementRead] != 0 || + 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) + } } func planned[P any](plans map[string]P, module string) bool { _, ok := plans[module]; return ok } diff --git a/go/handacts/handacts.go b/go/handacts/handacts.go index 4ce5937..26c1514 100644 --- a/go/handacts/handacts.go +++ b/go/handacts/handacts.go @@ -1,8 +1,7 @@ // Package handacts is the hand-act log's contract (novox/hq ADR 0565 clauses 10 and 11, issue 1953): the layout of an act, // moved unchanged from the controller's (mesh-controller internal/link HandAct) so one reader reads both logs alike; the // names of the seat mesh-hand-acts and of what its holder keeps; and the recorder every module records through, which -// asks the seat and, while no holder answers, the controller's verb until the controller hands its lease away, saying -// which log took the act. +// asks the seat alone: the controller's hand-act and warranted are gone (issue 1953 L5b) and the bus refuses them. package handacts import ( @@ -95,13 +94,13 @@ type Taken struct { Answer json.RawMessage `json:"answer,omitempty"` } -// Recorder records acts for a module (ADR 0565 clause 11): through the seat mesh-hand-acts, and while the seat has no -// holder that answers, through the controller's verb until the controller hands its lease away (the release's order -// gives no seat's holder before its callers; 00-plan.md, "Missing" 1). +// Recorder records acts for a module (ADR 0565 clause 11) through the seat mesh-hand-acts alone. An act the seat does +// not take is an error that says nothing was recorded, never a quiet success. type Recorder struct { // Ask calls a seat verb by its address (`seat:.`): stdio.Ask through the runtime. Ask func(key string, body any) (json.RawMessage, error) - // LeaseAway says the controller's lease is handed away: then nothing falls back to it. Nil is never handed away. + // LeaseAway is no longer read: nothing falls back to the controller. Kept so a caller that sets it builds; it goes + // once no module sets it. LeaseAway func() bool } @@ -121,38 +120,22 @@ func (r Recorder) Record(q Request) (Taken, error) { body[k] = v } } - controller := map[string]any{"what": q.What, "why": q.Why, "cause": q.Cause} - if q.Condition != "" { - controller["condition"] = q.Condition - } - if q.Cause == "" { - controller["cause"] = q.Verb - } - return r.either("hand-act", body, "hand-act", controller) + return r.record("hand-act", body) } // RecordWarrant records what the asking module did on the operator's warrant for its ask. func (r Recorder) RecordWarrant(asker, ask string) (Taken, error) { body := map[string]any{"asker": asker, "ask": ask} - return r.either("warranted", body, "warranted", body) + return r.record("warranted", body) } -func (r Recorder) either(verb string, body any, controllerVerb string, controllerBody any) (Taken, error) { +func (r Recorder) record(verb string, body any) (Taken, error) { answer, err := r.Ask("seat:"+Seat+"."+verb, body) if err == nil { return Taken{Log: LogSeat, Answer: answer}, nil } - if !NoResponder(err) { - return Taken{}, fmt.Errorf("%s.%s refused: %w", Seat, verb, err) + if NoResponder(err) { + return Taken{}, fmt.Errorf("%s has no holder that answers: nothing was recorded (%w)", Seat, err) } - if r.LeaseAway != nil && r.LeaseAway() { - return Taken{}, fmt.Errorf("%s has no holder that answers, and the controller handed its lease away: nothing "+ - "was recorded (%w)", Seat, err) - } - answer, cerr := r.Ask("seat:mesh-controller."+controllerVerb, controllerBody) - if cerr != nil { - return Taken{}, fmt.Errorf("%s has no holder that answers (%v), and the controller's %s refused too: %w", Seat, err, - controllerVerb, cerr) - } - return Taken{Log: LogController, Answer: answer}, nil + return Taken{}, fmt.Errorf("%s.%s refused, nothing was recorded: %w", Seat, verb, err) } diff --git a/go/handacts/handacts_test.go b/go/handacts/handacts_test.go index fe56282..6ffb90d 100644 --- a/go/handacts/handacts_test.go +++ b/go/handacts/handacts_test.go @@ -8,9 +8,8 @@ import ( "testing" ) -// ADR 0565 clauses 10 and 11 (issue 1953 L5-S2, S4): an act reads the controller's entries unchanged, and a recorder -// records through the seat, through the controller's verb only while the seat has no holder and the lease is the -// controller's, saying which log took it, and refuses to record nowhere. +// ADR 0565 clauses 10 and 11 (issue 1953 L5-S2, S4, L5b): an act reads the controller's entries unchanged, and a +// recorder records through the seat alone, the controller's verbs being gone, and says loudly when nothing was recorded. func TestAnActIsRecordedInTheLogThatAnswers(t *testing.T) { // The shape of mesh-controller internal/link HandAct, as its bucket keeps an entry (handacts.go at 0fbb5ef3). raw := `{"id":"act-1","at":"2026-10-28T09:00:00Z","by":"person.jochen","verb":"hand-act record","why":"a stuck consumer",` + @@ -39,14 +38,13 @@ func TestAnActIsRecordedInTheLogThatAnswers(t *testing.T) { warrants bool }{ {name: "the seat answers", log: LogSeat, asked: "seat:mesh-hand-acts.hand-act"}, - {name: "no holder yet", seat: noResponders, log: LogController, asked: "seat:mesh-hand-acts.hand-act,seat:mesh-controller.hand-act"}, - {name: "no holder, the lease handed away", seat: noResponders, away: true, refused: "handed its lease away", + {name: "no holder: nothing recorded, the controller never asked", seat: noResponders, refused: "nothing was recorded", asked: "seat:mesh-hand-acts.hand-act"}, - {name: "the seat refuses", seat: errors.New("why is required"), refused: "refused", asked: "seat:mesh-hand-acts.hand-act"}, - {name: "neither answers", seat: noResponders, ctrl: noResponders, refused: "refused too", - asked: "seat:mesh-hand-acts.hand-act,seat:mesh-controller.hand-act"}, - {name: "a warrant without a holder", seat: noResponders, warrants: true, log: LogController, - asked: "seat:mesh-hand-acts.warranted,seat:mesh-controller.warranted"}, + {name: "the seat refuses", seat: errors.New("why is required"), refused: "refused, nothing was recorded", + asked: "seat:mesh-hand-acts.hand-act"}, + {name: "a warrant without a holder", seat: noResponders, warrants: true, refused: "nothing was recorded", + asked: "seat:mesh-hand-acts.warranted"}, + {name: "a warrant recorded", warrants: true, log: LogSeat, asked: "seat:mesh-hand-acts.warranted"}, } { var asked []string r := Recorder{LeaseAway: func() bool { return c.away }, Ask: func(key string, body any) (json.RawMessage, error) { From 790d1c82d027e867cf2fde914e8bc94c80d44544 Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 1 Nov 2026 03:51:53 +0100 Subject: [PATCH 3/3] A follower joining a watched state hears its current values, and a templates archive is fetched once MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- .../frombus/frombustest/frombustest_test.go | 10 ++++++ go/compose/fromruntime/fromruntime.go | 35 ++++++++++++++++-- go/state_test.go | 12 +++++++ go/watched.go | 36 ++++++++++++++++--- 4 files changed, 87 insertions(+), 6 deletions(-) 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