From fe83d66b62e6c0f73efb0ba554ab79a257e6887e Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 4 Nov 2026 14:48:21 +0100 Subject: [PATCH] J4 (A2j finding 1): an owner's read tool answers a key's kept history, so a reader sees what changed before its first view The view's module history, heard messages and a want's since asked the owners' read tools for a key's history and were refused: views.ReadTool takes key with history and answers {history: [{revision, at, op, value}]} from the state's kept values (KeptSource through mesh/state.history; viewstest.Source and Serve from the bucket). Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig --- go/views/tools.go | 75 +++++++++++++++++++++++++++++++++---- go/views/views_test.go | 30 ++++++++++++++- go/views/viewstest/owner.go | 36 ++++++++++++++++-- 3 files changed, 130 insertions(+), 11 deletions(-) diff --git a/go/views/tools.go b/go/views/tools.go index 20b164e..6951e13 100644 --- a/go/views/tools.go +++ b/go/views/tools.go @@ -25,6 +25,20 @@ func (k KeptSource) Get(key string) (*Entry, bool, error) { return &Entry{Value: e.Value, Revision: e.Revision, Written: e.At}, true, nil } +// History is every value the state still keeps for key, oldest first (stdio.KeptState.History): as many as the +// state declares it keeps, deletes included. +func (k KeptSource) History(key string) ([]Revision, error) { + kept, err := k.State.History(key) + if err != nil { + return nil, err + } + out := make([]Revision, 0, len(kept)) + for _, r := range kept { + out = append(out, Revision{Revision: r.Revision, At: r.At, Op: r.Op, Value: r.Value}) + } + return out, nil +} + // ParseCallers is a callers list as a manifest writes it into an env value: comma-separated, each a module (on any // machine), . or person. (stdio.Callers). func ParseCallers(list string) stdio.Callers { @@ -76,22 +90,28 @@ func FirstViewTool(owner, name, state, words string, callers stdio.Callers, valu } // ReadTool is an owner's ordinary read tool of one of its states (design 25 ยง 12.2): the keys, one key's value and -// revision, or a page of values in the first view's form, and the state's fill against its cap as its mark last said -// it. Granted, as any tool is, to every reader that invokes it, and callable by a session. +// revision, one key's kept history (values a Historian), or a page of values in the first view's form, and the state's +// fill against its cap as its mark last said it. Granted, as any tool is, to every reader that invokes it, and callable +// by a session. func ReadTool(owner, name, state, words string, values, marks Source, now func() time.Time) stdio.Tool { return stdio.Tool{ Name: name, Description: fmt.Sprintf("Read %s's state %s: %s With no argument, its keys and how full it is against its cap (as the "+ - "owner last marked it); with key, that key's value and revision; with values, a page of values whole by key, at the "+ - "revision the owner said its views family up to, continued with after (the page's next).", owner, state, words), + "owner last marked it); with key, that key's value and revision; with key and history, every value the state still "+ + "keeps for that key, oldest first, each with its revision, time and op; with values, a page of values whole by key, "+ + "at the revision the owner said its views family up to, continued with after (the page's next).", owner, state, words), Input: map[string]any{"type": "object", "properties": map[string]any{ - "key": map[string]any{"type": "string", "description": "one key: its value and revision"}, - "values": map[string]any{"type": "boolean", "description": "a page of values whole by key"}, - "after": map[string]any{"type": "string", "description": "with values: the key the page before ended at"}}}, + "key": map[string]any{"type": "string", "description": "one key: its value and revision"}, + "history": map[string]any{"type": "boolean", "description": "with key: every value the state keeps for it, oldest first"}, + "values": map[string]any{"type": "boolean", "description": "a page of values whole by key"}, + "after": map[string]any{"type": "string", "description": "with values: the key the page before ended at"}}}, Run: func(args map[string]any) (any, error) { key, _ := args["key"].(string) after, _ := args["after"].(string) all, _ := args["values"].(bool) + if history, _ := args["history"].(bool); history { + return ReadHistory(owner, state, values, strings.TrimSpace(key)) + } return Read(owner, state, values, marks, strings.TrimSpace(key), all, strings.TrimSpace(after), now()) }, } @@ -119,6 +139,47 @@ type KeyAnswer struct { Revision uint64 `json:"revision,omitempty"` } +// Revision is one value a state keeps for a key: its revision, when the bus wrote it, its op (put, or delete) and +// its value, null for a delete. +type Revision struct { + Revision uint64 `json:"revision"` + At time.Time `json:"at"` + Op string `json:"op"` + Value json.RawMessage `json:"value"` +} + +// Historian is a Source that also keeps each key's history (KeptSource: the state's own kept values). +type Historian interface { + History(key string) ([]Revision, error) +} + +// HistoryAnswer is a read tool's answer of one key's history: every value kept, oldest first; none kept is empty. +type HistoryAnswer struct { + Owner string `json:"owner"` + State string `json:"state"` + Key string `json:"key"` + History []Revision `json:"history"` +} + +// ReadHistory is a read tool's answer of key's history from values, refused when no key is named or values keeps none. +func ReadHistory(owner, state string, values Source, key string) (HistoryAnswer, error) { + if key == "" { + return HistoryAnswer{}, fmt.Errorf("a history of %s is of one key: name it", state) + } + h, ok := values.(Historian) + if !ok { + return HistoryAnswer{}, fmt.Errorf("%s's state %s keeps no history of its keys here", owner, state) + } + kept, err := h.History(key) + if err != nil { + return HistoryAnswer{}, fmt.Errorf("the history of %s.%s cannot be read: %w", state, key, err) + } + if kept == nil { + kept = []Revision{} + } + return HistoryAnswer{Owner: owner, State: state, Key: key, History: kept}, nil +} + // Read is a read tool's answer: one key, a page of values, or the keys and fill. func Read(owner, state string, values, marks Source, key string, all bool, after string, now time.Time) (any, error) { switch { diff --git a/go/views/views_test.go b/go/views/views_test.go index 8a3227a..e1a990f 100644 --- a/go/views/views_test.go +++ b/go/views/views_test.go @@ -32,6 +32,14 @@ func (m mapSource) Get(key string) (*Entry, bool, error) { return &e, true, nil } +// keptSource is a mapSource that keeps each key's history, as a Historian. +type keptSource struct { + mapSource + kept map[string][]Revision +} + +func (k keptSource) History(key string) ([]Revision, error) { return k.kept[key], nil } + func markOf(state string, rev uint64, made time.Time) Entry { raw, _ := json.Marshal(Mark{V: Version, State: state, Revision: rev, Created: made, Fill: Fill{Keys: 2, Bytes: 100, MaxBytes: 1000}}) return Entry{Value: raw, Revision: 1} @@ -97,7 +105,7 @@ func TestViews(t *testing.T) { t.Errorf("an unproven call naming a caller is answered") } }) - t.Run("a read tool says the keys and the fill, one key, or a page of values", func(t *testing.T) { + t.Run("a read tool says the keys and the fill, one key, one key's kept history, or a page of values", func(t *testing.T) { r, err := Read("o", "modules", values, marks, "", false, "", now()) ra, ok := r.(ReadAnswer) if err != nil || !ok || len(ra.Keys) != 2 || ra.Fill == nil || ra.Fill.MaxBytes != 1000 || ra.Revision != 9 { @@ -111,6 +119,26 @@ func TestViews(t *testing.T) { if ra := r.(ReadAnswer); ra.Fill != nil || ra.Words == "" { t.Errorf("an unmarked state says its fill as %+v", ra) } + kept := []Revision{{Revision: 3, At: written, Op: "put", Value: json.RawMessage(`{"n":0}`)}, + {Revision: 5, At: written.Add(time.Second), Op: "delete", Value: json.RawMessage(`null`)}, {Revision: 7, At: written, Op: "put", Value: json.RawMessage(`{"n":1}`)}} + tool := ReadTool("o", "o_read_modules", "modules", "w", keptSource{values, map[string][]Revision{"a": kept}}, marks, now) + r, err = tool.Run(map[string]any{"key": "a", "history": true}) + raw, _ := json.Marshal(r) + var h struct { + History []Revision `json:"history"` + } + if err != nil || json.Unmarshal(raw, &h) != nil || len(h.History) != 3 || h.History[1].Op != "delete" || h.History[2].Revision != 7 || !h.History[0].At.Equal(written) { + t.Errorf("a key's history = %s, %v", raw, err) + } + if r, err = tool.Run(map[string]any{"key": "none", "history": true}); err != nil || r.(HistoryAnswer).History == nil { + t.Errorf("a key with nothing kept answers %+v, %v, not an empty history", r, err) + } + if _, err := tool.Run(map[string]any{"history": true}); err == nil { + t.Error("a history naming no key is answered") + } + if _, err := ReadTool("o", "o_read_modules", "modules", "w", values, marks, now).Run(map[string]any{"key": "a", "history": true}); err == nil { + t.Error("a source keeping no history answers one") + } }) // A follower whose owner answers from values and marks. diff --git a/go/views/viewstest/owner.go b/go/views/viewstest/owner.go index 9f11002..4b62535 100644 --- a/go/views/viewstest/owner.go +++ b/go/views/viewstest/owner.go @@ -23,7 +23,8 @@ import ( // 1), the owner's marks, its views stream with a durable consumer per reader (`_`), the emitter saying // each change, and the read tool answering on nc: tool is the owner's read tool as a reader's invokes names it // (`.`), stream its views stream as its manifest's `streams` names it, both given as the reader's -// ProcessReader.Follow takes them. It returns the bucket the test writes. +// ProcessReader.Follow takes them. The tool answers a key's history too (key with history), from what the bucket keeps: +// a test that reads history makes the bucket first with the history it needs. It returns the bucket the test writes. func Serve(t testing.TB, ctx context.Context, js jetstream.JetStream, nc *nats.Conn, owner, state, tool, stream string, readers ...string) jetstream.KeyValue { t.Helper() bucket := owner + "_" + state @@ -54,10 +55,18 @@ func Serve(t testing.TB, ctx context.Context, js jetstream.JetStream, nc *nats.C tool = strings.TrimPrefix(tool, owner+".") sub, err := nc.Subscribe("mesh.mod."+owner+".call."+tool+".>", func(m *nats.Msg) { var in struct { - After string `json:"after"` + After string `json:"after"` + Key string `json:"key"` + History bool `json:"history"` } _ = json.Unmarshal(m.Data, &in) - a, err := views.FirstView(owner, state, Source{ctx, kv}, Source{ctx, marks}, in.After, time.Now()) + var a any + var err error + if in.History { + a, err = views.ReadHistory(owner, state, Source{ctx, kv}, in.Key) + } else { + a, err = views.FirstView(owner, state, Source{ctx, kv}, Source{ctx, marks}, in.After, time.Now()) + } out := map[string]any{"result": a, "node": "n0"} if err != nil { out = map[string]any{"error": err.Error()} @@ -108,3 +117,24 @@ func (s Source) Get(key string) (*views.Entry, bool, error) { } return &views.Entry{Value: e.Value(), Revision: e.Revision(), Written: e.Created()}, true, nil } + +// History is every value the bucket keeps for key, oldest first, as the tool runner's mesh/state.history says it +// (a put, or a delete with a null value); none kept is an empty history. +func (s Source) History(key string) ([]views.Revision, error) { + kept, err := s.KV.History(s.Ctx, key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return []views.Revision{}, nil + } + if err != nil { + return nil, err + } + out := make([]views.Revision, 0, len(kept)) + for _, e := range kept { + r := views.Revision{Revision: e.Revision(), At: e.Created().UTC(), Op: "delete", Value: json.RawMessage("null")} + if e.Operation() == jetstream.KeyValuePut { + r.Op, r.Value = "put", json.RawMessage(e.Value()) + } + out = append(out, r) + } + return out, nil +}