diff --git a/go/process/binding.go b/go/process/binding.go new file mode 100644 index 00000000..56fb669e --- /dev/null +++ b/go/process/binding.go @@ -0,0 +1,184 @@ +package process + +import ( + "encoding/json" + "fmt" + "slices" + "strings" + "sync" + "time" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/membership" + "git.novox.be/novox/mesh-sdk/go/seatdef" +) + +// Binding is how a process with an account of its own reaches the kept data its `requires` declares: Names takes a local +// name and gives the entry's bus names (bucket, stream and subjects, object stores). It asks `show` of the seat +// the bus's seat for the module and each local name at its start, as a kept call, and asks again when it hears the seat's +// `recreated` or `dropped` for that entry (Register). Until `show` answers, Names answers an error naming the entry and +// the process acts on nothing that needs it; Waiting says so for the part's health (design 47 G §577; ADR 0283). +type Binding struct { + seat string + module string + call func(key string, body any) (json.RawMessage, error) + logf func(string, ...any) + + mu sync.Mutex + entries map[string]*boundEntry + locals []string + // spawn runs an ask of its own; a test makes it synchronous. + spawn func(func()) +} + +type boundEntry struct { + names membership.BusBinding + have bool + why string // why there is no answer yet + ask int // the latest ask; an older answer is dropped +} + +// NewBinding is a Binding of module's entries, asked of the seat that holds the bus's state (the caller names it, as it +// names the seat in its `invokes` as `seat:.show`: this package names no seat), by their local names, asking through call (a Conn's CallKept); logf says +// what it waits for (nil says nothing). +func NewBinding(seat, module string, call func(key string, body any) (json.RawMessage, error), logf func(string, ...any), locals ...string) *Binding { + if logf == nil { + logf = func(string, ...any) {} + } + b := &Binding{seat: seat, module: module, call: call, logf: logf, entries: map[string]*boundEntry{}, spawn: func(f func()) { go f() }} + for _, l := range locals { + if _, ok := b.entries[l]; !ok { + b.entries[l] = &boundEntry{why: "not asked yet"} + b.locals = append(b.locals, l) + } + } + return b +} + +// Names are the bus names of the entry local, or an error naming the entry when `show` has not answered for it (or it is +// no entry this Binding was made for). +func (b *Binding) Names(local string) (membership.BusBinding, error) { + b.mu.Lock() + defer b.mu.Unlock() + e := b.entries[local] + switch { + case e == nil: + return membership.BusBinding{}, fmt.Errorf("%s asks no bus entry %q: its Binding holds %s", b.module, local, strings.Join(b.locals, ", ")) + case !e.have: + return membership.BusBinding{}, fmt.Errorf("the bus entry %q of %s is not bound yet: %s", local, b.module, e.why) + } + return e.names, nil +} + +// Waiting is why the process waits, naming each entry `show` has not answered for, or empty when every one is bound. +func (b *Binding) Waiting() string { + b.mu.Lock() + defer b.mu.Unlock() + var out []string + for _, l := range b.locals { + if e := b.entries[l]; !e.have { + out = append(out, fmt.Sprintf("the bus entry %q (%s)", l, e.why)) + } + } + if len(out) == 0 { + return "" + } + return "waits for " + strings.Join(out, ", ") +} + +// Start asks `show` for every entry once, each as a kept call of its own that returns when answered, so the process's +// start waits for none. +func (b *Binding) Start() { + for _, l := range b.locals { + l := l + b.spawn(func() { b.Ask(l) }) + } +} + +// Ask asks `show` for the entry local once more, as a kept call, and keeps its answer; the answer of an ask replaced by a +// later one is dropped. +func (b *Binding) Ask(local string) { + b.mu.Lock() + e := b.entries[local] + if e == nil { + b.mu.Unlock() + return + } + e.ask++ + mine := e.ask + b.mu.Unlock() + names, err := b.show(local) + b.mu.Lock() + defer b.mu.Unlock() + if e.ask != mine { + return + } + if err != nil { + e.have, e.why = false, err.Error() + b.logf("binding: the bus entry %q of %s is not bound: %v", local, b.module, err) + return + } + e.names, e.have, e.why = names, true, "" +} + +func (b *Binding) show(local string) (membership.BusBinding, error) { + raw, err := b.call("seat:"+b.seat+".show", map[string]string{"module": b.module, "local-name": local}) + if err != nil { + return membership.BusBinding{}, err + } + var held struct { + Binding *membership.BusBinding `json:"binding"` + State string `json:"state"` + Why string `json:"why"` + } + if err := json.Unmarshal(raw, &held); err != nil { + return membership.BusBinding{}, fmt.Errorf("show answered something that is not an entry: %w", err) + } + if held.Binding == nil { + return membership.BusBinding{}, fmt.Errorf("the bus holds no objects for it yet (%s %s)", held.State, held.Why) + } + return *held.Binding, nil +} + +// Register makes the Binding hear `recreated` and `dropped` of the seat on hub, and asks every entry again at each bind +// of the hub's consumer, an event said before it was made being undelivered (ADR 0668). Register before the hub starts. +func (b *Binding) Register(hub *EventHub) error { + for _, event := range []string{"recreated", "dropped"} { + if err := hub.Handle(seatdef.EventSubject(b.seat, event), b.Hear); err != nil { + return err + } + } + hub.OnBound(func(time.Time) { + for _, l := range b.locals { + l := l + b.spawn(func() { b.Ask(l) }) + } + }) + return nil +} + +// Hear takes a `recreated` or `dropped` of the seat: one for an entry of this module's Binding is asked again, and a +// dropped one is unbound until `show` answers. Any other is acknowledged and left. +func (b *Binding) Hear(m jetstream.Msg) { + defer func() { _ = m.Ack() }() + var said struct { + Module string `json:"module"` + Local string `json:"local-name"` + } + if json.Unmarshal(m.Data(), &said) != nil || said.Module != b.module { + return + } + b.mu.Lock() + e := b.entries[said.Local] + if e != nil && strings.HasSuffix(m.Subject(), ".dropped") { + e.have, e.why = false, "the bus dropped it" + } + b.mu.Unlock() + if e != nil { + b.spawn(func() { b.Ask(said.Local) }) + } +} + +// Locals are the local names this Binding holds, in the order made. +func (b *Binding) Locals() []string { return slices.Clone(b.locals) } diff --git a/go/process/binding_test.go b/go/process/binding_test.go new file mode 100644 index 00000000..efb72d81 --- /dev/null +++ b/go/process/binding_test.go @@ -0,0 +1,93 @@ +package process + +import ( + "encoding/json" + "errors" + "strings" + "sync" + "testing" + + "github.com/nats-io/nats.go/jetstream" +) + +type bindMsg struct { + jetstream.Msg + subject string + data string + acked bool +} + +func (m *bindMsg) Subject() string { return m.subject } +func (m *bindMsg) Data() []byte { return []byte(m.data) } +func (m *bindMsg) Ack() error { m.acked = true; return nil } + +// design 47 G §577: a process reaches the kept data its `requires` declares through its Binding: `show` of the seat +// mesh-broker at the start, asked again on `recreated` and `dropped` of the entry, and a wait naming the entry until it answers. +func TestABindingAsksShowAndAgainOnRecreatedAndDropped(t *testing.T) { + var mu sync.Mutex + bucket, answers := "m_performed", true + var asked []string + call := func(key string, body any) (json.RawMessage, error) { + mu.Lock() + defer mu.Unlock() + asked = append(asked, key+" "+toJSON(body)) + if !answers { + return nil, errors.New("no responders") + } + return json.RawMessage(`{"binding":{"local":"performed","storage":"key-value","bucket":"` + bucket + `"}}`), nil + } + b := NewBinding("mesh-broker", "m", call, nil, "performed", "events") + b.spawn = func(f func()) { f() } + if _, err := b.Names("performed"); err == nil || !strings.Contains(err.Error(), `"performed"`) { + t.Fatalf("before show answered: %v", err) + } + if w := b.Waiting(); !strings.Contains(w, `"performed"`) || !strings.Contains(w, `"events"`) { + t.Fatalf("waiting words %q name neither entry", w) + } + if _, err := b.Names("nope"); err == nil { + t.Fatal("an entry this Binding was not made for was answered") + } + b.Ask("performed") + b.Ask("events") + got, err := b.Names("performed") + if err != nil || got.Bucket != "m_performed" || b.Waiting() != "" { + t.Fatalf("after show: %+v, %v, waiting %q", got, err, b.Waiting()) + } + if !strings.Contains(asked[0], `seat:mesh-broker.show {"local-name":"performed","module":"m"}`) { + t.Fatalf("asked %v", asked) + } + // recreated for this module's entry: asked again, the new names kept + mu.Lock() + bucket = "m_performed2" + mu.Unlock() + for _, m := range []*bindMsg{ + {subject: "mesh.seat.mesh-broker.event.recreated", data: `{"module":"other","local-name":"performed"}`}, + {subject: "mesh.seat.mesh-broker.event.recreated", data: `{"module":"m","local-name":"performed"}`}, + } { + b.Hear(m) + if !m.acked { + t.Errorf("%s not acknowledged", m.data) + } + } + if n, _ := b.Names("performed"); n.Bucket != "m_performed2" { + t.Fatalf("recreated was not asked again: %+v", n) + } + // dropped: unbound until show answers, which it does not + mu.Lock() + answers = false + mu.Unlock() + b.Hear(&bindMsg{subject: "mesh.seat.mesh-broker.event.dropped", data: `{"module":"m","local-name":"performed"}`}) + if _, err := b.Names("performed"); err == nil || !strings.Contains(err.Error(), "no responders") { + t.Fatalf("dropped left it bound: %v", err) + } + // an entry the bus holds no objects for is waited for, not bound + b2 := NewBinding("mesh-broker", "m", func(string, any) (json.RawMessage, error) { + return json.RawMessage(`{"state":"waiting","why":"not made"}`), nil + }, nil, "x") + b2.Ask("x") + if _, err := b2.Names("x"); err == nil || !strings.Contains(err.Error(), "not made") { + t.Fatalf("an entry with no binding: %v", err) + } +} + +func toJSON(v any) string { d, _ := json.Marshal(v); return string(d) } diff --git a/go/views/ownbuild.go b/go/views/ownbuild.go new file mode 100644 index 00000000..edb670fe --- /dev/null +++ b/go/views/ownbuild.go @@ -0,0 +1,44 @@ +package views + +import ( + "context" + "encoding/json" + "os" + + "git.novox.be/novox/mesh-sdk/go/process" +) + +// OwnBuildKey is where a process keeps its own build in its state `first-views`, and the field both answers of a part's +// first views carry beside their list (design 47 G §674, §424, §519): the fingerprint the node-engine composed into +// the part's environment, or OwnBuildUnknown when it carries none. +const OwnBuildKey = "own-build" + +// OwnBuildUnknown is what a part's first views say of its own build when its environment carries no fingerprint. +const OwnBuildUnknown = "unknown" + +func ownBuildWord(b process.Build) string { + if b.Fingerprint == "" { + return OwnBuildUnknown + } + return b.Fingerprint +} + +// ownBuild is the build this reader's bundle was composed with, read from its environment at each answer. +func (r *Reader) ownBuild() process.Build { + getenv := r.Getenv + if getenv == nil { + getenv = os.Getenv + } + return process.ReadOwnBuild(getenv) +} + +// KeepOwnBuild writes a process's own build at OwnBuildKey of kv (its state `first-views`) at its start, for its tools +// bundle's status tool to read (design 47 G §674). +func KeepOwnBuild(ctx context.Context, kv interface { + Put(ctx context.Context, key string, value []byte) (uint64, error) +}, b process.Build, logf func(string, ...any)) { + data, _ := json.Marshal(ownBuildWord(b)) + if _, err := kv.Put(ctx, OwnBuildKey, data); err != nil && logf != nil { + logf("views: the own build is not kept in %s: %v", StatusState, err) + } +} diff --git a/go/views/process_reader.go b/go/views/process_reader.go index 5115032f..6a999fef 100644 --- a/go/views/process_reader.go +++ b/go/views/process_reader.go @@ -833,6 +833,7 @@ func (r *ProcessReader) Start(ctx context.Context, js jetstream.JetStream, cred var said func(Status) if status != nil { status.Ready(func(kv jetstream.KeyValue) { + KeepOwnBuild(ctx, kv, process.OwnBuild(), r.logf) if err := hearAgain(ctx, kv, r.Followers, r.logf); err != nil { r.logf("views: the hand asks of %s's first views are not heard (%v) until it starts again", r.Module, err) } diff --git a/go/views/reader.go b/go/views/reader.go index 9dcd3380..9b37efcd 100644 --- a/go/views/reader.go +++ b/go/views/reader.go @@ -35,10 +35,11 @@ type Reader struct { // Acts records a re-read by hand before it is asked (nil records nothing // and the re-read is refused). Acts HandActs - // Now, Bound and Wait are for tests: zero are the clock, Bound and five seconds. - Now func() time.Time - Bound time.Duration - Wait time.Duration + // Now, Bound, Wait and Getenv are for tests: zero are the clock, Bound, five seconds and os.Getenv. + Getenv func(string) string + Now func() time.Time + Bound time.Duration + Wait time.Duration mu sync.Mutex followers map[string]*Follower // by "." @@ -589,7 +590,7 @@ func (r *Reader) Rows(name string, rows func() any) { // firstViews is what `_first_views` answers: the views' statuses and each list added by Rows. func (r *Reader) firstViews() map[string]any { - out := map[string]any{"views": r.Statuses()} + out := map[string]any{"views": r.Statuses(), OwnBuildKey: ownBuildWord(r.ownBuild())} r.mu.Lock() rows := maps.Clone(r.rows) r.mu.Unlock() diff --git a/go/views/status.go b/go/views/status.go index bc7f3af4..8a6a9021 100644 --- a/go/views/status.go +++ b/go/views/status.go @@ -125,7 +125,7 @@ func StatusTools(reader, read, again string, st keptStatus, acts HandActs, now f } out := []Status{} for _, k := range keys { - if strings.HasPrefix(k, "again.") { + if strings.HasPrefix(k, "again.") || k == OwnBuildKey { continue } e, err := st.Get(k) @@ -137,7 +137,14 @@ func StatusTools(reader, read, again string, st keptStatus, acts HandActs, now f out = append(out, s) } } - return map[string]any{"first-views": out}, nil + own := OwnBuildUnknown + if e, err := st.Get(OwnBuildKey); err == nil && e != nil { + var w string + if json.Unmarshal(e.Value, &w) == nil && strings.TrimSpace(w) != "" { + own = w + } + } + return map[string]any{"first-views": out, OwnBuildKey: own}, 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 "+ diff --git a/go/views/views_test.go b/go/views/views_test.go index 07bdd0ac..1e3c5429 100644 --- a/go/views/views_test.go +++ b/go/views/views_test.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "maps" "reflect" "runtime" "slices" @@ -1234,3 +1235,49 @@ func TestViews(t *testing.T) { t.Run("Views", subViews) t.Run("StdioWriterPublishesAndListensThroughTheModulesOwnPlumbing", subStdioWriterPublishesAndListensThroughTheModulesOwnPlumbing) } + +type putRecorder struct{ puts map[string][]byte } + +func (p *putRecorder) Put(_ context.Context, key string, value []byte) (uint64, error) { + p.puts[key] = value + return 1, nil +} + +// statusWith is a state `first-views` holding the given values. +type statusWith map[string][]byte + +func (s statusWith) Keys() ([]string, error) { return slices.Collect(maps.Keys(s)), nil } +func (s statusWith) Get(k string) (*stdio.StateEntry, error) { + if v, ok := s[k]; ok { + return &stdio.StateEntry{Value: v}, nil + } + return nil, nil +} +func (statusWith) Put(string, any) (uint64, error) { return 0, errors.New("read only") } + +// design 47 G §674: both answers of a part's first views say its own build beside their list. +func TestFirstViewsSayTheOwnBuild(t *testing.T) { + for name, c := range map[string]struct{ env, want string }{"a fingerprint": {"abc123", "abc123"}, "none": {"", "unknown"}} { + getenv := func(string) string { return c.env } + // the bundle reader's answer + r := &Reader{Module: "m", Getenv: getenv} + got, _ := r.Tools("m")[0].Run(nil) + if v := got.(map[string]any)[OwnBuildKey]; v != c.want { + t.Errorf("%s: the bundle reader's first views say %v, want %s", name, v, c.want) + } + // the process writes it at its start, and its status tool reads it + rec := &putRecorder{puts: map[string][]byte{}} + KeepOwnBuild(context.Background(), rec, process.ReadOwnBuild(getenv), nil) + st := statusWith{OwnBuildKey: rec.puts[OwnBuildKey]} + read, err := StatusTools("m", "m_first_views", "m_first_view_again", st, nil, time.Now)[0].Run(nil) + m := read.(map[string]any) + if err != nil || m[OwnBuildKey] != c.want || len(m["first-views"].([]Status)) != 0 { + t.Errorf("%s: the status tool says %v, %v, want %s", name, m, err, c.want) + } + } + // a process that kept nothing yet says unknown + read, _ := StatusTools("m", "r", "a", statusWith{}, nil, time.Now)[0].Run(nil) + if read.(map[string]any)[OwnBuildKey] != "unknown" { + t.Errorf("nothing kept: %v", read) + } +}