Merge branch 'feat/47-G-fix11-a7' into feat/47-G-join11
This commit is contained in:
@@ -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:<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) }
|
||||
@@ -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) }
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
+6
-5
@@ -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 "<owner>.<state>"
|
||||
@@ -589,7 +590,7 @@ func (r *Reader) Rows(name string, rows func() any) {
|
||||
|
||||
// firstViews is what `<prefix>_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()
|
||||
|
||||
+9
-2
@@ -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 "+
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user