Files
mesh-sdk/go/views/status.go
T
jschoubben 38659c3937
mesh/delivery merged: source.main moved to novox/mesh-sdk@55f71b78114f
mesh/delivery-group group fix/1966-views-durable merged: every member is merged
mesh/check checking 38659c3937e7 on 55f71b78114f | novox/mesh-sdk#174 in group fix/1966-views-durable
A reader asks again when its consumer was made after its first view, says every silent drop, and carries what it heard in its status (issue 1966)
A consumer is made from now (ADR 0443): an event said before it was made is never delivered to it and no later event
shows the loss. The event hub tells a part when the bus made its consumer, and a follower asked before that asks
again. Events passed over, at or below the held revision, or of a state not followed are now said in the log. The
status carries the highest revision heard and why the picture is behind it.

Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
2026-11-09 02:26:10 +01:00

153 lines
6.3 KiB
Go

package views
import (
"context"
"encoding/json"
"errors"
"fmt"
"slices"
"strings"
"time"
"github.com/nats-io/nats.go/jetstream"
stdio "git.novox.be/novox/mesh-sdk/go"
)
// StatusState is a reader's own state of its first views (design 25 § 12.2): at StatusKey each first view's Status, as
// its process last said it, and at AgainKey a hand ask of its re-read tool, which the process takes and deletes. A
// process serves no tool (ADR 0436), so its read and re-read tools, in its tools bundle, meet it here.
const StatusState = "first-views"
// StatusKey is where a reader keeps its first view of owner's state.
func StatusKey(owner, state string) string { return owner + "." + state }
// AgainKey is where a re-read tool asks a reader's process to ask owner's first view of state again.
func AgainKey(owner, state string) string { return "again." + owner + "." + state }
// HandAsk is one ask of a re-read tool: who asked (the runtime's proof), why and when.
type HandAsk struct {
V int `json:"v"`
Owner string `json:"owner"`
State string `json:"state"`
Why string `json:"why"`
By string `json:"by"`
At time.Time `json:"at"`
}
// KeepStatus is a Follower's Said that writes its status at StatusKey of kv (the reader's StatusState).
func KeepStatus(ctx context.Context, kv jetstream.KeyValue, logf func(string, ...any)) func(Status) {
return func(s Status) {
data, err := json.Marshal(s)
if err == nil {
_, err = kv.Put(ctx, StatusKey(s.Owner, s.State), data)
}
if err != nil && logf != nil {
logf("views: the status of the first view %s is not kept: %v", s.Tool, err)
}
}
}
// HearAgain takes each hand ask written at an AgainKey of kv, those standing at the start included, asks that
// follower's first view again with the ask's words and who asked, and deletes the ask. It returns once the watch is set.
func HearAgain(ctx context.Context, kv jetstream.KeyValue, followers []*Follower, logf func(string, ...any)) error {
w, err := kv.Watch(ctx, "again.>")
if err != nil {
return err
}
go func() {
defer func() { _ = w.Stop() }()
for {
select {
case <-ctx.Done():
return
case e, ok := <-w.Updates():
if !ok {
return
}
if e == nil || e.Operation() != jetstream.KeyValuePut {
continue
}
var a HandAsk
if err := json.Unmarshal(e.Value(), &a); err != nil {
logf("views: the hand ask %s is not read: %v", e.Key(), err)
_ = kv.Delete(ctx, e.Key())
continue
}
i := slices.IndexFunc(followers, func(f *Follower) bool { return f.Owner == a.Owner && f.State == a.State })
if i < 0 {
logf("views: the hand ask %s names no first view this part holds", e.Key())
_ = kv.Delete(ctx, e.Key())
continue
}
logf("views: %s asks the first view %s again by hand: %s", a.By, followers[i].Tool, a.Why)
_ = kv.Delete(ctx, e.Key(), jetstream.LastRevision(e.Revision()))
f := followers[i]
go func() { _ = f.Again(ctx, a.Why+" ("+a.By+")") }()
}
}
}()
return nil
}
// StatusTools are a reader's read tool of its first views (read) and its re-read tool (again), in its tools bundle,
// over its StatusState: read lists what the process last said of each; again writes a hand ask the process takes,
// naming the caller the runtime proved, and is recorded in the status's by-hand count and why. again is a hand act:
// acts records it before the ask is written, and one that is not taken is said and nothing is asked (nil acts
// record nothing, so again refuses).
func StatusTools(reader, read, again string, st stdio.KeptState, acts HandActs, now func() time.Time) []stdio.Tool {
return []stdio.Tool{
{Name: read,
Description: fmt.Sprintf("%s's first views of the owners' states it follows (novox/hq ADR 0615): per owner and state, "+
"the tool it asks, the revision its picture holds now (events taken in included), the highest revision it heard and why "+
"it is behind it, if it is, when it asked and was answered, how many asks (by hand among them) and whether its "+
"condition (unanswered past two minutes) is raised.", reader),
Input: map[string]any{"type": "object", "properties": map[string]any{}},
Run: func(map[string]any) (any, error) {
keys, err := st.Keys()
if err != nil {
return nil, fmt.Errorf("the state %s cannot be listed: %w", StatusState, err)
}
out := []Status{}
for _, k := range keys {
if strings.HasPrefix(k, "again.") {
continue
}
e, err := st.Get(k)
if err != nil || e == nil {
continue
}
var s Status
if json.Unmarshal(e.Value, &s) == nil {
out = append(out, s)
}
}
return map[string]any{"first-views": out}, nil
}},
{Name: again,
Description: fmt.Sprintf("Ask %s's first view of an owner's state again now, by hand (a hand act, its caller and why kept in "+
"the status %s reads): for a picture a person doubts. The part asks once and follows the owner's events after the answer.", reader, read),
Input: map[string]any{"type": "object", "required": []string{"owner", "state", "why"}, "properties": map[string]any{
"owner": map[string]any{"type": "string", "description": "the owner module, as the read tool names it"},
"state": map[string]any{"type": "string", "description": "the owner's state, as the read tool names it"},
"why": map[string]any{"type": "string", "description": "why it is asked again, in plain words"}}},
Run: func(args map[string]any) (any, error) {
owner, _ := args["owner"].(string)
state, _ := args["state"].(string)
why, _ := args["why"].(string)
if owner == "" || state == "" || strings.TrimSpace(why) == "" {
return nil, errors.New("again names the owner, the state and why")
}
a := HandAsk{V: Version, Owner: owner, State: state, Why: strings.TrimSpace(why), By: stdio.CallerOf(args), At: now().UTC()}
if err := recordHandAct(acts, HandAct{What: fmt.Sprintf("asked %s's first view of %s's %s again by hand", reader, owner, state),
Why: a.Why, Cause: "first-view-again", Verb: again}); err != nil {
return nil, fmt.Errorf("the hand act was not recorded, so nothing was asked again: %w", err)
}
if _, err := st.Put(AgainKey(owner, state), a); err != nil {
return nil, fmt.Errorf("the hand ask is not written: %w", err)
}
return map[string]any{"asked": a, "read": read}, nil
}},
}
}