novox/mesh-delivery
19
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

DLV-M3: the teller reads the assigner's asks by its read tool's first view and its views family, never its bucket

Browse Source 
ADR 0615 §1: sdk/firstview is a process's picture of an owner's state, read as the bucket it
replaces (Get, Keys, Watch; every write refused), asked through <owner>.<short>_read_<state> with
values and kept from the owner's views family; firstviewtest serves an owner in a whole-module test.
The teller declares invokes assigner_read_asks, consumes views.asks, states first-views and findings,
and the tools teller_first_views and teller_first_view_again.

Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
This commit is contained in:
jschoubben
2026-11-04 13:54:50 +01:00
parent 1592abbf4f
commit 28c3b27fec
9 changed files with 728 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-teller/cmd/mesh-delivery-teller-tools/main.go
+6 -1
View File
@@ -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)
}
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go
+8 -2
View File
@@ -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}
}
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go
+32 -8
View File
@@ -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"
modules/mesh-delivery-teller/go.mod
+4
View File
@@ -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
modules/mesh-delivery-teller/manifest_test.go
+21 -6
View File
@@ -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)
}
modules/mesh-delivery-teller/module.json
+12 -3
View File
@@ -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"
],
sdk/firstview/firstview.go
+532
View File
@@ -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):
// `<owner>.<short>_read_<state>`, 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 <owner>_<state>
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 (`<owner>_<state>`); 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): `<short>-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`
// `<owner>.views.<state>`), keeps each status in the process's state `first-views` and hears the hand asks the tools
// bundle's `<reader>_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)
}
sdk/firstview/firstviewtest/owner.go
+109
View File
@@ -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 `<owner>_<state>` (made when absent, history
// 1), the owner's marks, its views stream with a durable consumer per reader (`<node>_<module>`), 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
}
sdk/go.mod
+4 -1
View File
@@ -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
)