diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller-tools/main.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller-tools/main.go index a7e035c4..f5b5e57d 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller-tools/main.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller-tools/main.go @@ -13,6 +13,7 @@ import ( "time" stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/views" "mesh-delivery-teller/internal/ends" ) @@ -102,7 +103,11 @@ func askEnd(s stdio.KeptState, bound time.Duration) asker { } func main() { - if err := stdio.Serve("", Tools(askEnd(stdio.State(ends.State), 25*time.Second), time.Now)); err != nil { + // teller_first_views and teller_first_view_again: where the teller's first view of the assigner's asks stands, and + // asking it again by hand (ADR 0615, design 25 § 12.2). + tools := append(Tools(askEnd(stdio.State(ends.State), 25*time.Second), time.Now), + views.StatusTools("mesh-delivery-teller", "teller_first_views", "teller_first_view_again", stdio.State(views.StatusState), time.Now)...) + if err := stdio.Serve("", tools); err != nil { fmt.Fprintf(os.Stderr, "[mesh-delivery-teller-tools] %v\n", err) os.Exit(1) } diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go index f7e015f0..5ea0dc00 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go @@ -6,6 +6,7 @@ import ( "context" "encoding/json" "fmt" + "git.novox.be/novox/mesh-delivery/sdk/firstview/firstviewtest" "os" "path/filepath" "strings" @@ -184,7 +185,10 @@ func fullConf(dir string) string { seatAsk, seatCancel, seatCondition, "$JS.API.DIRECT.GET.KV_"+messengerAsks+".$KV."+messengerAsks+".mesh-delivery-teller.>") // The durable consumer on EVENTS the controller makes for the seat it uses, pulled and acknowledged (issue 1126). teller = append(teller, "$JS.API.CONSUMER.INFO.EVENTS."+tellerConsumer, "$JS.API.CONSUMER.MSG.NEXT.EVENTS."+tellerConsumer, "$JS.ACK.EVENTS."+tellerConsumer+".>") - teller = append(append(teller, readOf(assignerAsks)...), readOf(viewAnswers)...) // issue 1297 B10: the manifest's reads + teller = append(teller, readOf(viewAnswers)...) // issue 1297 B10: the manifest's reads + // The assigner's asks by its read tool and its views family (ADR 0615, DLV-M3), and the teller's own first-views and + // findings; the bucket itself is no longer the teller's to read. + teller = append(append(append(teller, firstviewtest.Grants(delivery.AssignerModule, "asks", tellerConsumer)...), keepsOf(firstViewsBucket)...), keepsOf(findingsBucket)...) view := append(append(append(readOf(tellerAnswers), readOf(gateAsksKV)...), readOf(assignerAsks)...), keepsOf(viewAnswers)...) assigner := keepsOf(assignerAsks) tools := append(keepsOf(gateAsksKV), readOf(tellerAnswers)...) @@ -251,7 +255,7 @@ func startFull(t *testing.T) *fullBed { } ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) t.Cleanup(cancel) - for name, history := range map[string]uint8{gateAsksKV: delivery.AsksHistory, tellerAnswers: delivery.AnswersHistory, tellerAsked: 1, ends.Bucket: 1, messengerAsks: 1, assignerAsks: delivery.AssignerAsksHistory, viewAnswers: delivery.ViewAnswersHistory} { + for name, history := range map[string]uint8{gateAsksKV: delivery.AsksHistory, tellerAnswers: delivery.AnswersHistory, tellerAsked: 1, ends.Bucket: 1, messengerAsks: 1, assignerAsks: delivery.AssignerAsksHistory, viewAnswers: delivery.ViewAnswersHistory, firstViewsBucket: 1, findingsBucket: 1} { if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: history}); err != nil { t.Fatal(err) } @@ -267,6 +271,8 @@ func startFull(t *testing.T) *fullBed { AckPolicy: jetstream.AckExplicitPolicy}); err != nil { t.Fatal(err) } + // The assigner as the owner of its asks: its read tool and its views family (ADR 0615). + firstviewtest.Serve(t, ctx, js, ctl, delivery.AssignerModule, "asks", tellerConsumer) return &fullBed{srv: srv, ctl: ctl, js: js, ctx: ctx} } diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go index 8557c036..183f57ef 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go @@ -18,9 +18,13 @@ import ( "syscall" "git.novox.be/novox/mesh-sdk/go/process" + "git.novox.be/novox/mesh-sdk/go/views" + "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" + "git.novox.be/novox/mesh-delivery/sdk/firstview" + "git.novox.be/novox/mesh-sdk/go/asker" "git.novox.be/novox/mesh-sdk/go/delivery" @@ -83,18 +87,28 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // The view's answers to the assigner's asks (issue 1297 B10): both buckets are the controller's to create, and a bucket not // there is said and leaves the teller without that path rather than without the gate's asks (a wait for it needs a timer). failed := make(chan error, 4) - assignerKV, viewKV := process.OptionalBucket(ctx, js, delivery.AssignerAsksBucket, teller, logger.Printf), process.OptionalBucket(ctx, js, delivery.ViewAnswersBucket, teller, logger.Printf) + // The assigner's asks are its state, read by the first view of its read tool and then its views family, never its + // bucket (ADR 0615 §1, DLV-M3): the picture reads as the bucket did. + calls, err := process.Connect(cred, nats.Name(cred.Node+"."+cred.Module+".calls"), nats.RetryOnFailedConnect(true)) + if err != nil { + return fmt.Errorf("the teller's connection for its tool calls: %w", err) + } + defer calls.Close() + reader := &firstview.Reader{Module: delivery.TellerModule, Call: calls.Call, Logf: logger.Printf} + findings := process.AwaitBucket(ctx, js, conn, findingsBucket, "a first view unanswered past its bound is said in the log alone", logger.Printf) + reader.Findings = findings.KV + assignerKV := jetstream.KeyValue(reader.Follow(delivery.AssignerModule, assignerAsksState)) + reader.Start(ctx, js, cred, process.AwaitBucket(ctx, js, conn, firstViewsBucket, "teller_first_views reads nothing and teller_first_view_again is not heard", logger.Printf)) + viewKV := process.OptionalBucket(ctx, js, delivery.ViewAnswersBucket, teller, logger.Printf) // The assigner's asks are raised on the operator channel whether or not the view's answers can be read (issue 1906). if viewKV != nil { t.View = kvGate{ctx: ctx, kv: viewKV} } - if assignerKV != nil { - t.Assigner = kvGate{ctx: ctx, kv: assignerKV} - go func() { - failed <- watchAsks(ctx, assignerKV, func(key string, _ uint64, _ []byte) { t.OnAssignerAsk(key) }, t.OnAssignerAskGone, nil) - }() - } - if assignerKV != nil && viewKV != nil { + t.Assigner = kvGate{ctx: ctx, kv: assignerKV} + go func() { + failed <- watchAsks(ctx, assignerKV, func(key string, _ uint64, _ []byte) { t.OnAssignerAsk(key) }, t.OnAssignerAskGone, nil) + }() + if viewKV != nil { go func() { failed <- watchAsks(ctx, viewKV, func(key string, _ uint64, _ []byte) { t.OnViewAnswer(key) }, func(string) {}, nil) }() @@ -131,5 +145,15 @@ const teller = "the asks and answers it holds are not carried until the teller i // raised (manifest state entry asked, novox/hq ADR 0437 §4). const askedBucket = delivery.TellerModule + "_asked" +// The teller's own states of its first view of the assigner's asks (views.StatusState) and of its findings, where the +// condition keeper raises a first view unanswered past its bound (ADR 0615 §6). +var ( + firstViewsBucket = delivery.TellerModule + "_" + views.StatusState + findingsBucket = delivery.TellerModule + "_findings" +) + +// assignerAsksState is the assigner's state of its asks (delivery.AssignerAsksBucket). +const assignerAsksState = "asks" + // waiting is said once at start. const waiting = "waiting for the gate's asks, whose bucket does not exist yet; the teller is started again with the gate, and nothing is read, written or published until then" diff --git a/modules/mesh-delivery-teller/go.mod b/modules/mesh-delivery-teller/go.mod index ee19602f..add68f57 100644 --- a/modules/mesh-delivery-teller/go.mod +++ b/modules/mesh-delivery-teller/go.mod @@ -20,3 +20,7 @@ require ( golang.org/x/sys v0.48.0 // indirect golang.org/x/time v0.15.0 // indirect ) + +require git.novox.be/novox/mesh-delivery/sdk v0.0.0-00010101000000-000000000000 + +replace git.novox.be/novox/mesh-delivery/sdk => ../../sdk diff --git a/modules/mesh-delivery-teller/manifest_test.go b/modules/mesh-delivery-teller/manifest_test.go index 74420f55..8de6b9e8 100644 --- a/modules/mesh-delivery-teller/manifest_test.go +++ b/modules/mesh-delivery-teller/manifest_test.go @@ -40,6 +40,8 @@ type manifest struct { Uses []string `json:"uses"` State []state `json:"state"` Reads []string `json:"reads"` + Invokes []string `json:"invokes"` + Consumes []string `json:"consumes"` OwnSecrets map[string]string `json:"own-secrets"` SecretsOwner string `json:"secrets-owner"` Resources []map[string]any `json:"resources"` @@ -80,13 +82,26 @@ func TestManifestDeclarations(t *testing.T) { // 1240/AC1 } // answers is the teller's answer to each ask; asked is what it asked of the router, which condition it raised and which // gate key and revision each ask was made at (issue 1343); ends is the ends asked by hand with teller_end_ask (issue 774). - if want := []state{{Name: "answers", History: 5}, {Name: "asked", History: 1}, {Name: "ends", History: 1}}; !reflect.DeepEqual(m.State, want) { + // first-views is where its first view of the assigner's asks stands and findings its condition past the bound (ADR 0615). + if want := []state{{Name: "answers", History: 5}, {Name: "asked", History: 1}, {Name: "ends", History: 1}, {Name: "first-views", History: 1}, {Name: "findings", History: 1}}; !reflect.DeepEqual(m.State, want) { t.Errorf("state = %+v, want %+v", m.State, want) } - // The assigner's asks and the view's answers to them are read so the view's answer fills an ask that needs no proof (1297/B10). - if want := []string{"mesh-delivery-assigner.asks", "mesh-delivery-gate.asks", "mesh-delivery-view.answers"}; !reflect.DeepEqual(sorted(m.Reads), want) { + // The view's answers to the assigner's asks are read so the view's answer fills an ask that needs no proof (1297/B10); the + // assigner's asks are its state, asked by its read tool and followed by its views family, never read (ADR 0615 §1, DLV-M3). + if want := []string{"mesh-delivery-gate.asks", "mesh-delivery-view.answers"}; !reflect.DeepEqual(sorted(m.Reads), want) { t.Errorf("reads = %v, want %v", m.Reads, want) } + for _, r := range m.Reads { + if strings.HasPrefix(r, "mesh-delivery-indexer.") || strings.HasPrefix(r, "mesh-delivery-assigner.") { + t.Errorf("reads %s: the indexer's and the assigner's state is read by their tools and views families alone", r) + } + } + if want := []string{"mesh-delivery-assigner.assigner_read_asks"}; !reflect.DeepEqual(m.Invokes, want) { + t.Errorf("invokes = %v, want %v", m.Invokes, want) + } + if want := []string{"mesh-delivery-assigner.views.asks"}; !reflect.DeepEqual(m.Consumes, want) { + t.Errorf("consumes = %v, want %v", m.Consumes, want) + } if want := []string{"operator-channel"}; !reflect.DeepEqual(sorted(m.Uses), want) { t.Errorf("uses = %v, want %v", m.Uses, want) } @@ -97,11 +112,11 @@ func TestManifestDeclarations(t *testing.T) { // 1240/AC1 t.Errorf("secrets-owner = %q", m.SecretsOwner) } // 774: teller_end_ask, served by its tools bundle through the state ends; the process serves none. - if want := []string{"teller_end_ask"}; !reflect.DeepEqual(m.Tools, want) { + if want := []string{"teller_end_ask", "teller_first_views", "teller_first_view_again"}; !reflect.DeepEqual(m.Tools, want) { t.Errorf("tools = %v, want %v", m.Tools, want) } - // No consumes, invokes, emits, streams, bus or seat (ADR 0437 §4). - for _, k := range []string{"consumes", "invokes", "streams", "logs", "emits", "claims", "provides", "requires", "defines-seats", "seat", "seats", "bus"} { + // No emits, streams, bus or seat (ADR 0437 §4). + for _, k := range []string{"streams", "logs", "emits", "claims", "provides", "requires", "defines-seats", "seat", "seats", "bus"} { if _, ok := top[k]; ok { t.Errorf("the manifest declares %q: the teller has none", k) } diff --git a/modules/mesh-delivery-teller/module.json b/modules/mesh-delivery-teller/module.json index 6d814c0e..be0af507 100644 --- a/modules/mesh-delivery-teller/module.json +++ b/modules/mesh-delivery-teller/module.json @@ -3,15 +3,24 @@ "version": "1", "slug": "teller", "tools": [ - "teller_end_ask" + "teller_end_ask", + "teller_first_views", + "teller_first_view_again" + ], + "invokes": [ + "mesh-delivery-assigner.assigner_read_asks" + ], + "consumes": [ + "mesh-delivery-assigner.views.asks" ], "state": [ { "name": "answers", "history": 5 }, { "name": "asked", "history": 1 }, - { "name": "ends", "history": 1 } + { "name": "ends", "history": 1 }, + { "name": "first-views", "history": 1 }, + { "name": "findings", "history": 1 } ], "reads": [ - "mesh-delivery-assigner.asks", "mesh-delivery-gate.asks", "mesh-delivery-view.answers" ], diff --git a/sdk/firstview/firstview.go b/sdk/firstview/firstview.go new file mode 100644 index 00000000..57d82e26 --- /dev/null +++ b/sdk/firstview/firstview.go @@ -0,0 +1,532 @@ +// Package firstview is a process's pictures of the indexer's and the assigner's states it used to read (novox/hq ADR +// 0615 §1, design 25 § 12.2), for the mesh-delivery parts that run as processes (the builder, the checker, the teller, +// the view, the watchman): each state is asked once through the owner's ordinary read tool with `values` (the first +// view), then kept from the owner's views family, asked again only on a gap or by hand (mesh-sdk go/views Follower). +// +// A Picture reads as the key-value bucket the process used to bind (jetstream.KeyValue: Get, Keys, ListKeys, Watch, +// WatchAll), so a reader's code is unchanged; every write is refused, for the owner's state is the owner's alone. The +// picture is kept in memory: the owner's state is the truth and a restart asks again (ADR 0615 §4). +// +// It is the process side of mesh-sdk go/views Reader (the stdio side) and moves there whole when mesh-sdk takes it. +package firstview + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "slices" + "strings" + "sync" + "time" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/process" + "git.novox.be/novox/mesh-sdk/go/views" +) + +// ErrReadOnly is every write to a picture: the state is its owner's. +var ErrReadOnly = errors.New("a first view is read-only: the state is its owner's") + +// ReadTool is the owner's ordinary read tool of state, as a reader's `invokes` names it (design 25 § 12.2): +// `._read_`, short the owner's name past its last dash-separated `mesh-delivery-` prefix +// (indexer, assigner) and the state's dashes as underscores. +func ReadTool(owner, state string) string { + short := strings.TrimPrefix(owner, "mesh-delivery-") + return owner + "." + short + "_read_" + strings.ReplaceAll(state, "-", "_") +} + +// Reader is one process's pictures of its owners' states. +type Reader struct { + // Module is the reading module, named in the finding of a first view unanswered past its bound. + Module string + // Call asks a module's tool (process.Conn.Call): the read tools are asked with `values`. + Call func(key string, body any) (json.RawMessage, error) + // Findings is the reader's own state `findings`, where the condition keeper raises and clears the condition of a + // first view unanswered past its bound (ADR 0615 §6); nil says it in the log only. + Findings func() jetstream.KeyValue + Logf func(string, ...any) + // Now, Bound and Wait are for tests: zero are the clock, views.Bound and five seconds. + Now func() time.Time + Bound time.Duration + Wait time.Duration + + mu sync.Mutex + pics map[string]*Picture // by bucket name _ + order []string +} + +func (r *Reader) logf(format string, args ...any) { + if r.Logf != nil { + r.Logf(format, args...) + } +} + +// Follow adds the owner's state to follow through its read tool (ReadTool). Call it before Start. +func (r *Reader) Follow(owner, state string) *Picture { + r.mu.Lock() + defer r.mu.Unlock() + if r.pics == nil { + r.pics = map[string]*Picture{} + } + bucket := owner + "_" + state + if p := r.pics[bucket]; p != nil { + return p + } + tool := ReadTool(owner, state) + p := &Picture{bucket: bucket, ctx: context.Background()} + call := r.Call + ask := views.AskVia(func(key string, body any) (json.RawMessage, error) { + m, _ := body.(map[string]any) + if m == nil { + m = map[string]any{} + } + m["values"] = true + if call == nil { + return nil, errors.New("this reader asks no tool") + } + return call(key, m) + }, tool) + p.f = &views.Follower{Owner: owner, Tool: tool, State: state, Ask: ask, Logf: r.Logf, Now: r.Now, Bound: r.Bound, Wait: r.Wait, + Raise: func(s views.Status) { r.finding(s, true) }, Clear: func(s views.Status) { r.finding(s, false) }, + Changed: p.changed} + r.pics[bucket] = p + r.order = append(r.order, bucket) + return p +} + +// Picture is the picture of the bucket the process used to bind, by its bus name (`_`); nil when not +// followed. +func (r *Reader) Picture(bucket string) *Picture { + r.mu.Lock() + defer r.mu.Unlock() + return r.pics[bucket] +} + +// Followers are the followers, in the order they were added. +func (r *Reader) Followers() []*views.Follower { + r.mu.Lock() + defer r.mu.Unlock() + out := make([]*views.Follower, 0, len(r.order)) + for _, b := range r.order { + out = append(out, r.pics[b].f) + } + return out +} + +func (r *Reader) finding(s views.Status, raised bool) { + if raised { + r.logf("the first view %s of %s's %s is unanswered past its bound: %s", s.Tool, s.Owner, s.State, s.Error) + } + if r.Findings == nil { + return + } + kv := r.Findings() + if kv == nil { + return + } + at := time.Now() + if r.Now != nil { + at = r.Now() + } + key, value, err := views.UnansweredFinding(r.Module, s, at) + if err == nil && raised { + _, err = kv.Put(context.Background(), key, value) + } else if err == nil { + err = kv.Delete(context.Background(), key) + } + if err != nil { + r.logf("views: the finding of the first view %s is not written: %v", s.Tool, err) + } +} + +// Hear is the handler of the owner's views family on the reader's durable consumer (cred.EventConsumer on the owner's +// views stream): each event to the follower of its state. +func (r *Reader) Hear(ctx context.Context, owner string) func(jetstream.Msg) { + fs := map[string]*views.Follower{} + for _, f := range r.Followers() { + if f.Owner == owner { + fs[f.State] = f + } + } + return views.Hear(ctx, fs, r.logf) +} + +// Owners are the owners followed, sorted. +func (r *Reader) Owners() []string { + var out []string + for _, f := range r.Followers() { + if !slices.Contains(out, f.Owner) { + out = append(out, f.Owner) + } + } + slices.Sort(out) + return out +} + +// Ask asks every first view until answered, each on a goroutine of its own, so the process starts at once and waits for +// an owner that is down, saying so (ADR 0615 §6). Said, when set, keeps each status (views.KeepStatus). +func (r *Reader) Ask(ctx context.Context, said func(views.Status)) { + for _, f := range r.Followers() { + if said != nil { + f.Said = said + } + go func() { + if err := f.Start(ctx); err != nil { + r.logf("the first view %s was not answered: %v", f.Tool, err) + } + }() + } +} + +// Ready is closed once every followed state had its first view. +func (r *Reader) Ready() <-chan struct{} { + fs := r.Followers() + done := make(chan struct{}) + go func() { + for _, f := range fs { + <-f.Ready() + } + close(done) + }() + return done +} + +// Picture is one owner's state as the process reads it: a read-only jetstream.KeyValue over a views.Follower. Its +// revisions are the picture's own, one per change it took, so a watcher sees them rise; they are not the owner's. +type Picture struct { + jetstream.KeyValue // never set: every method a reader may call is the picture's own + + bucket string + ctx context.Context + f *views.Follower + + mu sync.Mutex + seq uint64 + revs map[string]uint64 + watchers []*watcher +} + +// Follower is the picture's follower (its status, a gap or a hand ask). +func (p *Picture) Follower() *views.Follower { return p.f } + +func (p *Picture) changed(key string, value json.RawMessage, deleted bool, at time.Time) { + p.mu.Lock() + p.seq++ + if p.revs == nil { + p.revs = map[string]uint64{} + } + p.revs[key] = p.seq + op := jetstream.KeyValuePut + if deleted { + op, value = jetstream.KeyValueDelete, nil + } + e := entry{bucket: p.bucket, key: key, value: value, rev: p.seq, at: at, op: op} + ws := slices.Clone(p.watchers) + p.mu.Unlock() + for _, w := range ws { + w.send(e) + } +} + +// answered is nil once the first view is answered, else the error a read before it gets (never a wait). +func (p *Picture) answered() error { + select { + case <-p.f.Ready(): + return nil + default: + return fmt.Errorf("no first view of %s yet: %s has not answered", p.bucket, p.f.Tool) + } +} + +func (p *Picture) ready(ctx context.Context) error { + select { + case <-p.f.Ready(): + return nil + case <-ctx.Done(): + return fmt.Errorf("no first view of %s yet (%s has not answered): %w", p.bucket, p.f.Tool, ctx.Err()) + } +} + +// Bucket is the bus name of the owner's state the picture stands for. +func (p *Picture) Bucket() string { return p.bucket } + +// Get is one key; before the first view is answered it is an error, never a wait. +func (p *Picture) Get(_ context.Context, key string) (jetstream.KeyValueEntry, error) { + if err := p.answered(); err != nil { + return nil, err + } + v, ok := p.f.Get(key) + if !ok { + return nil, jetstream.ErrKeyNotFound + } + p.mu.Lock() + rev := p.revs[key] + p.mu.Unlock() + return entry{bucket: p.bucket, key: key, value: v, rev: rev, op: jetstream.KeyValuePut}, nil +} + +// Keys is every key; none is jetstream.ErrNoKeysFound, as the bus says it; before the first view is answered it is an +// error, never a wait. +func (p *Picture) Keys(_ context.Context, _ ...jetstream.WatchOpt) ([]string, error) { + if err := p.answered(); err != nil { + return nil, err + } + var keys []string + for k := range p.f.Values() { + keys = append(keys, k) + } + if len(keys) == 0 { + return nil, jetstream.ErrNoKeysFound + } + slices.Sort(keys) + return keys, nil +} + +// ListKeys is Keys on a channel. +func (p *Picture) ListKeys(ctx context.Context, opts ...jetstream.WatchOpt) (jetstream.KeyLister, error) { + keys, err := p.Keys(ctx, opts...) + if err != nil && !errors.Is(err, jetstream.ErrNoKeysFound) { + return nil, err + } + ch := make(chan string, len(keys)) + for _, k := range keys { + ch <- k + } + close(ch) + return lister{ch}, nil +} + +// WatchAll is Watch of every key. +func (p *Picture) WatchAll(ctx context.Context, opts ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) { + return p.Watch(ctx, ">", opts...) +} + +// Watch hands, once the first view is answered, every current value of the keys matching (a subject pattern: `*` one +// token, `>` the rest), then nil, then each change, as the bus's watch does. Options are not read: a picture holds no +// history, and deletes are always handed. +func (p *Picture) Watch(ctx context.Context, keys string, _ ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) { + w := &watcher{ch: make(chan jetstream.KeyValueEntry, 64), match: keys, wake: make(chan struct{}, 1), stop: make(chan struct{})} + go func() { + defer close(w.ch) + if p.ready(ctx) != nil { + return + } + p.mu.Lock() + p.watchers = append(p.watchers, w) + p.mu.Unlock() + defer p.unwatch(w) + vals := p.f.Values() + names := make([]string, 0, len(vals)) + for k := range vals { + names = append(names, k) + } + slices.Sort(names) + p.mu.Lock() + first := make([]jetstream.KeyValueEntry, 0, len(names)+1) + for _, k := range names { + if matches(keys, k) { + first = append(first, entry{bucket: p.bucket, key: k, value: vals[k], rev: p.revs[k], op: jetstream.KeyValuePut}) + } + } + p.mu.Unlock() + first = append(first, nil) + w.run(ctx, first) + }() + return w, nil +} + +func (p *Picture) unwatch(w *watcher) { + p.mu.Lock() + defer p.mu.Unlock() + p.watchers = slices.DeleteFunc(p.watchers, func(x *watcher) bool { return x == w }) +} + +// The writes and the reads a picture cannot answer are refused, never handed to a bucket. +func (p *Picture) Put(context.Context, string, []byte) (uint64, error) { return 0, ErrReadOnly } +func (p *Picture) PutString(context.Context, string, string) (uint64, error) { return 0, ErrReadOnly } +func (p *Picture) Create(context.Context, string, []byte, ...jetstream.KVCreateOpt) (uint64, error) { + return 0, ErrReadOnly +} +func (p *Picture) Update(context.Context, string, []byte, uint64) (uint64, error) { + return 0, ErrReadOnly +} +func (p *Picture) Delete(context.Context, string, ...jetstream.KVDeleteOpt) error { return ErrReadOnly } +func (p *Picture) Purge(context.Context, string, ...jetstream.KVDeleteOpt) error { return ErrReadOnly } +func (p *Picture) PurgeDeletes(context.Context, ...jetstream.KVPurgeOpt) error { return ErrReadOnly } +func (p *Picture) GetRevision(context.Context, string, uint64) (jetstream.KeyValueEntry, error) { + return nil, errors.New("a first view holds no history") +} +func (p *Picture) History(context.Context, string, ...jetstream.WatchOpt) ([]jetstream.KeyValueEntry, error) { + return nil, errors.New("a first view holds no history") +} +func (p *Picture) WatchFiltered(ctx context.Context, keys []string, opts ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) { + if len(keys) != 1 { + return nil, errors.New("a first view watches one pattern") + } + return p.Watch(ctx, keys[0], opts...) +} +func (p *Picture) ListKeysFiltered(ctx context.Context, filters ...string) (jetstream.KeyLister, error) { + keys, err := p.Keys(ctx) + if err != nil && !errors.Is(err, jetstream.ErrNoKeysFound) { + return nil, err + } + ch := make(chan string, len(keys)) + for _, k := range keys { + if len(filters) == 0 || slices.ContainsFunc(filters, func(f string) bool { return matches(f, k) }) { + ch <- k + } + } + close(ch) + return lister{ch}, nil +} +func (p *Picture) Status(context.Context) (jetstream.KeyValueStatus, error) { + return nil, errors.New("a first view has no bucket status: the owner's read tool says its fill") +} + +// matches says whether key matches a subject pattern of the bus's watch. +func matches(pattern, key string) bool { + if pattern == "" || pattern == ">" { + return true + } + pt, kt := strings.Split(pattern, "."), strings.Split(key, ".") + for i, t := range pt { + if t == ">" { + return len(kt) > i + } + if i >= len(kt) || (t != "*" && t != kt[i]) { + return false + } + } + return len(pt) == len(kt) +} + +// watcher hands a watch's entries in order: the current values first, then each change heard, never blocking the +// follower (the changes wait in a queue of its own). +type watcher struct { + ch chan jetstream.KeyValueEntry + match string + wake chan struct{} + stop chan struct{} + once sync.Once + + mu sync.Mutex + queue []jetstream.KeyValueEntry +} + +func (w *watcher) send(e entry) { + if !matches(w.match, e.key) { + return + } + w.mu.Lock() + w.queue = append(w.queue, e) + w.mu.Unlock() + select { + case w.wake <- struct{}{}: + default: + } +} + +func (w *watcher) run(ctx context.Context, first []jetstream.KeyValueEntry) { + out := func(e jetstream.KeyValueEntry) bool { + select { + case w.ch <- e: + return true + case <-ctx.Done(): + return false + case <-w.stop: + return false + } + } + for _, e := range first { + if !out(e) { + return + } + } + for { + w.mu.Lock() + q := w.queue + w.queue = nil + w.mu.Unlock() + for _, e := range q { + if !out(e) { + return + } + } + select { + case <-w.wake: + case <-ctx.Done(): + return + case <-w.stop: + return + } + } +} + +func (w *watcher) Updates() <-chan jetstream.KeyValueEntry { return w.ch } + +func (w *watcher) Stop() error { + w.once.Do(func() { close(w.stop) }) + return nil +} + +type lister struct{ ch chan string } + +func (l lister) Keys() <-chan string { return l.ch } +func (l lister) Stop() error { return nil } + +type entry struct { + bucket, key string + value []byte + rev uint64 + at time.Time + op jetstream.KeyValueOp +} + +func (e entry) Bucket() string { return e.bucket } +func (e entry) Key() string { return e.key } +func (e entry) Value() []byte { return e.value } +func (e entry) Revision() uint64 { return e.rev } +func (e entry) Created() time.Time { return e.at } +func (e entry) Delta() uint64 { return 0 } +func (e entry) Operation() jetstream.KeyValueOp { return e.op } + +// ViewsStream is the bus name of owner's views family (views.IndexerStream, views.AssignerStream): `-views` +// as the owner's old `streams` section names it. +func ViewsStream(owner string) string { + return catalogue.StreamBusName(strings.TrimPrefix(owner, "mesh-delivery-") + "-views") +} + +// Start hears each owner's views family on the process's durable consumer of it (its manifest's `consumes` +// `.views.`), keeps each status in the process's state `first-views` and hears the hand asks the tools +// bundle's `_first_view_again` writes there (views.StatusTools), then asks every first view (Ask). A consumer +// the bus does not have yet is said: the picture is then the first view's alone, asked again on a gap or by hand. +func (r *Reader) Start(ctx context.Context, js jetstream.JetStream, cred process.Account, status *process.AwaitedBucket) { + for _, owner := range r.Owners() { + stream := ViewsStream(owner) + cons, err := cred.EventConsumer(ctx, js, stream) + if err == nil { + _, err = process.HearEvents(cons, r.Hear(ctx, owner), r.logf) + } + if err != nil { + r.logf("the consumer %s on %s is not heard (%v): %s's states are the first view's alone, asked again by hand, until the %s starts again", + cred.EventConsumerName(), stream, err, owner, r.Module) + } + } + var said func(views.Status) + if status != nil { + status.Ready(func(kv jetstream.KeyValue) { + if err := views.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) + } + }) + said = func(s views.Status) { + if kv := status.KV(); kv != nil { + views.KeepStatus(ctx, kv, r.logf)(s) + } + } + } + r.Ask(ctx, said) +} diff --git a/sdk/firstview/firstviewtest/owner.go b/sdk/firstview/firstviewtest/owner.go new file mode 100644 index 00000000..761e13ad --- /dev/null +++ b/sdk/firstview/firstviewtest/owner.go @@ -0,0 +1,109 @@ +// Package firstviewtest is an owner of a state as a reader's whole-module test needs it (novox/hq ADR 0615): the +// owner's read tool answering the first view from its bucket at its mark, and its views family said by the real +// views emitter onto its views stream, which the reader's durable consumer reads. +package firstviewtest + +import ( + "context" + "encoding/json" + "errors" + "strings" + "testing" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/process" + "git.novox.be/novox/mesh-sdk/go/views" + + "git.novox.be/novox/mesh-delivery/sdk/firstview" +) + +// Serve is owner's state on the bus with the owner's side of it: the bucket `_` (made when absent, history +// 1), the owner's marks, its views stream with a durable consumer per reader (`_`), the emitter saying +// each change, and the read tool (firstview.ReadTool) answering on nc. It returns the bucket the test writes. +func Serve(t testing.TB, ctx context.Context, js jetstream.JetStream, nc *nats.Conn, owner, state string, readers ...string) jetstream.KeyValue { + t.Helper() + bucket := owner + "_" + state + kv, err := js.KeyValue(ctx, bucket) + if err != nil { + if kv, err = js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: bucket, History: 1}); err != nil { + t.Fatal(err) + } + } + marks, err := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: owner + "_" + views.MarksState, History: 1}) + if err != nil { + t.Fatal(err) + } + stream := firstview.ViewsStream(owner) + if _, err := js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{Name: stream, Subjects: []string{process.ModuleStreamSubject(owner, views.EventPrefix+">")}}); err != nil { + t.Fatal(err) + } + for _, r := range readers { + if _, err := js.CreateOrUpdateConsumer(ctx, stream, jetstream.ConsumerConfig{Durable: r, AckPolicy: jetstream.AckExplicitPolicy}); err != nil { + t.Fatal(err) + } + } + em := views.Emitter{Owner: owner, Subject: func(e string) string { return process.ModuleStreamSubject(owner, e) }, + Publish: views.JetStreamPublish(js), Marks: marks} + if err := em.Start(ctx, state, kv, nil); err != nil { + t.Fatal(err) + } + tool := strings.TrimPrefix(firstview.ReadTool(owner, state), owner+".") + sub, err := nc.Subscribe("mesh.mod."+owner+".call."+tool+".>", func(m *nats.Msg) { + var in struct { + After string `json:"after"` + } + _ = json.Unmarshal(m.Data, &in) + 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()} + } + raw, _ := json.Marshal(out) + _ = m.Respond(raw) + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = sub.Unsubscribe() }) + return kv +} + +// Grants are the subjects a reader's account publishes to follow owner's state (its `invokes` and `consumes`): the +// read tool's call subjects, its durable consumer's info, next and acknowledgements on the owner's views stream. +func Grants(owner, state, durable string) []string { + tool := strings.TrimPrefix(firstview.ReadTool(owner, state), owner+".") + stream := firstview.ViewsStream(owner) + return []string{"mesh.mod." + owner + ".call." + tool + ".>", "mesh.mod." + owner + ".call." + tool, + "$JS.API.CONSUMER.INFO." + stream + "." + durable, "$JS.API.CONSUMER.MSG.NEXT." + stream + "." + durable, + "$JS.ACK." + stream + "." + durable + ".>"} +} + +// Source is a bucket as a views.Source. +type Source struct { + Ctx context.Context + KV jetstream.KeyValue +} + +// Keys is every key; none is no error. +func (s Source) Keys() ([]string, error) { + keys, err := s.KV.Keys(s.Ctx) + if errors.Is(err, jetstream.ErrNoKeysFound) { + return nil, nil + } + return keys, err +} + +// Get is one key's value and revision. +func (s Source) Get(key string) (*views.Entry, bool, error) { + e, err := s.KV.Get(s.Ctx, key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return nil, false, nil + } + if err != nil { + return nil, false, err + } + return &views.Entry{Value: e.Value(), Revision: e.Revision()}, true, nil +} diff --git a/sdk/go.mod b/sdk/go.mod index 7e692d40..c6bae45b 100644 --- a/sdk/go.mod +++ b/sdk/go.mod @@ -2,4 +2,7 @@ module git.novox.be/novox/mesh-delivery/sdk go 1.26.0 -require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261103221822-85ba0a9a70d8 +require ( + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261103221822-85ba0a9a70d8 + github.com/nats-io/nats.go v1.54.0 +)