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

J3 (lead's a): go/views names no module; its caller gives the owner's read tool and views stream

Browse Source 
ProcessReader.Follow(owner, state, tool, stream) takes the owner's read tool and its views stream as the owner's
declaration names them, and refuses another owner's tool or a second stream for one owner; OwnerReadTool, StreamOf,
IndexerStream and AssignerStream are gone, and viewstest takes the same two values. The second instance of a module
name in shared code (the earlier review finding), so the package test now refuses a module's name in go/views.

Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
This commit is contained in:
jschoubben
2026-11-04 14:35:56 +01:00
parent 4827df4d48
commit dae61959b9
7 changed files with 69 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go/packages_list_test.go
+15 -1
View File
@@ -9,6 +9,7 @@ import (
"os"
"path/filepath"
"reflect"
"regexp"
"slices"
"strconv"
"strings"
@@ -35,6 +36,10 @@ var clientSeats = []string{contractseats.OperatorChannel, contractseats.MeshVaul
contractseats.NpmPackageRegistry, contractseats.GoPackageRegistry, contractseats.ContainerImageRegistry,
contractseats.OciArtifactRegistry}
// moduleWord is a module's name in a literal (mesh-<word>[-<word>...], or one of mesh-delivery's parts by its short name
// before -views or _read_), but a seat's that go/views records through (mesh-hand-acts) and an import path's mesh-sdk.
var moduleWord = regexp.MustCompile(`\bmesh-(?:delivery|grants|vault|host|config|provisioner|broker|knowledge|enrolment)\b[-a-z]*|\b(?:indexer|assigner|teller|watchman|checker|builder)(?:-views|_read_)`)
// notInSDK are the packages that moved to a seat's client or contract (ADR 0627 §1, §4) and may not come back.
var notInSDK = []string{"asks", "asker", "publish"}
@@ -72,7 +77,7 @@ func TestMeshSDKHoldsOnlyWhatEveryModuleShares(t *testing.T) {
}
return false
}
var seatNames, pending []string
var seatNames, pending, modules []string
clientNamed := func(v string) bool {
for _, seat := range clientSeats {
if v == seat || strings.HasPrefix(v, "mesh.seat."+seat+".") {
@@ -114,6 +119,12 @@ func TestMeshSDKHoldsOnlyWhatEveryModuleShares(t *testing.T) {
ast.Inspect(file, func(n ast.Node) bool {
switch n := n.(type) {
case *ast.BasicLit:
// views is shared by every owner and reader: it names no module, the caller gives the owner's tool and
// stream from the owner's declaration (the lead's ruling, 2026-11-04, the second instance of the kind).
if v, err := strconv.Unquote(n.Value); err == nil && n.Kind == token.STRING && (dir == "views" || strings.HasPrefix(dir, "views/")) &&
moduleWord.MatchString(v) {
modules = append(modules, where+moduleWord.FindString(v))
}
if v, err := strconv.Unquote(n.Value); err == nil && n.Kind == token.STRING && namesSeat(v) {
seatNames = append(seatNames, where+v)
if !inCatalogue && clientNamed(v) {
@@ -186,6 +197,9 @@ func TestMeshSDKHoldsOnlyWhatEveryModuleShares(t *testing.T) {
}
}
slices.Sort(found)
for _, m := range modules {
t.Errorf("%s: go/views names no module; its caller gives the owner's read tool and views stream from the owner's declaration", m)
}
for _, dir := range found {
if slices.Contains(notInSDK, dir) {
pending = append(pending, dir+": a package that moved")
go/views/process_reader.go
+23 -25
View File
@@ -28,14 +28,6 @@ import (
// 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")
// OwnerReadTool 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 OwnerReadTool(owner, state string) string {
short := strings.TrimPrefix(owner, "mesh-delivery-")
return owner + "." + short + "_read_" + strings.ReplaceAll(state, "-", "_")
}
// ProcessReader is one process's pictures of its owners' states.
type ProcessReader struct {
// Module is the reading module, named in the finding of a first view unanswered past its bound.
@@ -51,9 +43,10 @@ type ProcessReader struct {
Bound time.Duration
Wait time.Duration
mu sync.Mutex
pics map[string]*Picture // by bucket name <owner>_<state>
order []string
mu sync.Mutex
pics map[string]*Picture // by bucket name <owner>_<state>
order []string
streams map[string]string // each owner's views stream, by owner
}
func (r *ProcessReader) logf(format string, args ...any) {
@@ -62,18 +55,27 @@ func (r *ProcessReader) logf(format string, args ...any) {
}
}
// Follow adds the owner's state to follow through its read tool (OwnerReadTool). Call it before Start.
func (r *ProcessReader) Follow(owner, state string) *Picture {
// Follow adds the owner's state to follow, as the owner declares them (design 25 § 12.2): tool is its ordinary read tool
// of state as the reader's `invokes` names it (`<owner>.<tool>`, from the owner's manifest's tools), stream its views
// family as the owner's manifest's `streams` names it (its bus name is catalogue.StreamBusName's). views names no owner:
// the caller gives both. An owner has one views stream; a second name for it is refused. Call it before Start.
func (r *ProcessReader) Follow(owner, state, tool, stream string) (*Picture, error) {
r.mu.Lock()
defer r.mu.Unlock()
if r.pics == nil {
r.pics = map[string]*Picture{}
if owner == "" || state == "" || !strings.HasPrefix(tool, owner+".") || stream == "" {
return nil, fmt.Errorf("%s follows %s's %s by its read tool <owner>.<tool> and its views stream, not %q and %q", r.Module, owner, state, tool, stream)
}
if have, ok := r.streams[owner]; ok && have != stream {
return nil, fmt.Errorf("%s's views stream is %s, not %s", owner, have, stream)
}
if r.pics == nil {
r.pics, r.streams = map[string]*Picture{}, map[string]string{}
}
r.streams[owner] = stream
bucket := owner + "_" + state
if p := r.pics[bucket]; p != nil {
return p
return p, nil
}
tool := OwnerReadTool(owner, state)
p := &Picture{bucket: bucket, ctx: context.Background()}
call := r.Call
ask := AskVia(func(key string, body any) (json.RawMessage, error) {
@@ -92,7 +94,7 @@ func (r *ProcessReader) Follow(owner, state string) *Picture {
Changed: p.changed}
r.pics[bucket] = p
r.order = append(r.order, bucket)
return p
return p, nil
}
// Picture is the picture of the bucket the process used to bind, by its bus name (`<owner>_<state>`); nil when not
@@ -507,19 +509,15 @@ 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 }
// StreamOf is the bus name of owner's views family (IndexerStream, AssignerStream): `<short>-views`
// as the owner's old `streams` section names it.
func StreamOf(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 (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 *ProcessReader) Start(ctx context.Context, js jetstream.JetStream, cred process.Account, status *process.AwaitedBucket) {
for _, owner := range r.Owners() {
stream := StreamOf(owner)
r.mu.Lock()
stream := catalogue.StreamBusName(r.streams[owner])
r.mu.Unlock()
cons, err := cred.EventConsumer(ctx, js, stream)
if err == nil {
_, err = process.HearEvents(cons, r.Hear(ctx, owner), r.logf)
go/views/reader.go
+1 -1
View File
@@ -13,7 +13,7 @@ import (
"git.novox.be/novox/mesh-sdk/go/handacts"
)
// Reader is a module's pictures of the states of the indexer and the assigner it used to read (novox/hq ADR 0615 §1): it
// Reader is a module's pictures of the owners' states it used to read (novox/hq ADR 0615 §1): it
// asks each owner's read tool for the first view, subscribes the owner's views family, and answers the module's own
// `Keys` and `Get` of a state from the picture, by the name the module's `reads` used to carry (`<owner>.<state>`).
// A module in a stdio tools bundle declares `invokes` for the tools and `consumes` for the events, and names no `reads`
go/views/status.go
+2 -2
View File
@@ -99,7 +99,7 @@ func HearAgain(ctx context.Context, kv jetstream.KeyValue, followers []*Follower
func StatusTools(reader, read, again string, st stdio.KeptState, acts handacts.Recorder, now func() time.Time) []stdio.Tool {
return []stdio.Tool{
{Name: read,
Description: fmt.Sprintf("%s's first views of the indexer's and the assigner's states (novox/hq ADR 0615): per owner and state, "+
Description: fmt.Sprintf("%s's first views of the owners' states it follows (novox/hq ADR 0615): per owner and state, "+
"the tool it asks, the revision its picture held when last answered, when it asked and was answered, how many asks "+
"(by hand among them) and whether its condition (unanswered past two minutes) is raised.", reader),
Input: map[string]any{"type": "object", "properties": map[string]any{}},
@@ -128,7 +128,7 @@ func StatusTools(reader, read, again string, st stdio.KeptState, acts handacts.R
Description: fmt.Sprintf("Ask %s's first view of an owner's state again now, by hand (a hand act, its caller and why kept in "+
"the status %s reads): for a picture a person doubts. The part asks once and follows the owner's events after the answer.", reader, read),
Input: map[string]any{"type": "object", "required": []string{"owner", "state", "why"}, "properties": map[string]any{
"owner": map[string]any{"type": "string", "description": "the owner: mesh-delivery-indexer or mesh-delivery-assigner"},
"owner": map[string]any{"type": "string", "description": "the owner module, as the read tool names it"},
"state": map[string]any{"type": "string", "description": "the owner's state, as the read tool names it"},
"why": map[string]any{"type": "string", "description": "why it is asked again, in plain words"}}},
Run: func(args map[string]any) (any, error) {
go/views/views.go
+4 -6
View File
@@ -1,5 +1,5 @@
// Package views is the owners' side and the readers' side of a first view (novox/hq ADR 0615, ADR 0621, ADR 0622;
// design 25 § 12.2): the indexer and the assigner answer a reader's first view of one of their states as a tool of
// design 25 § 12.2): an owner answers a reader's first view of one of their states as a tool of
// their own module (Answer, FirstView), say each change of that state as one event on their views family (Event,
// Emitter), and a reader keeps its picture from the answer and the events, asking the tool again only on a gap in
// revisions (Follower). The tools' read side (Read) answers the ordinary read tools of § 12.2 too.
@@ -19,12 +19,10 @@ const Version = 1
// The owners' views families (design 25 § 12.2; ADR 0615 §5, ADR 0619 §2): one stream each on its owner's old `streams`
// section, kept by limits 7 days and at most 256 MiB, dropping its oldest when full, so a reader that missed a dropped
// event sees the gap in revisions and asks once more. A stream's name is the mesh's, so each carries its owner's name.
// event sees the gap in revisions and asks once more. A stream's name is its owner's to declare; views names none.
const (
IndexerStream = "indexer-views"
AssignerStream = "assigner-views"
StreamDays = 7
StreamMiB = 256
StreamDays = 7
StreamMiB = 256
// EventPrefix starts each event's local name: `views.<state>`, one event per changed key of that state.
EventPrefix = "views."
// EventPattern is every event of a views family, as a stream's events and a reader's consumes name it.
go/views/views_test.go
+14 -5
View File
@@ -265,28 +265,37 @@ func TestViews(t *testing.T) {
t.Errorf("asked again by hand: %v, %v", again, err)
}
})
t.Run("a process's reader reads its owner's state as a read-only bucket, each key's written time handed on, and opens no other", func(t *testing.T) {
t.Run("a process's reader follows the read tool and views stream its caller names, reads the state as a read-only bucket, each key's written time handed on, and opens no other", func(t *testing.T) {
r := &ProcessReader{Module: "reader", Bound: time.Nanosecond, Wait: time.Millisecond, Now: now,
Call: func(tool string, body any) (json.RawMessage, error) {
if m, _ := body.(map[string]any); tool != "mesh-delivery-indexer.indexer_read_modules" || m["values"] != true {
if m, _ := body.(map[string]any); tool != "o.o_read_modules" || m["values"] != true {
return nil, fmt.Errorf("asked %s with %v", tool, body)
}
a, err := FirstView("o", "modules", values, marks, "", now())
raw, _ := json.Marshal(map[string]any{"result": a})
return raw, err
}}
p := r.Follow("mesh-delivery-indexer", "modules")
p, err := r.Follow("o", "modules", "o.o_read_modules", "o-views")
if err != nil {
t.Fatal(err)
}
for name, c := range map[string][2]string{"another owner's tool": {"x.read_more", "o-views"}, "no stream": {"o.read_more", ""},
"a second stream for the owner": {"o.read_more", "o-other-views"}} {
if _, err := r.Follow("o", "more", c[0], c[1]); err == nil {
t.Fatalf("%s is followed", name) // a picture no owner answers would hold Ready
}
}
if _, err := p.Get(context.Background(), "a"); err == nil {
t.Errorf("a picture with no first view is read")
}
if _, err := r.KeyValue(context.Background(), "mesh-delivery-indexer_other"); err == nil {
if _, err := r.KeyValue(context.Background(), "o_other"); err == nil {
t.Errorf("a bucket not followed is opened")
}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
r.Ask(ctx, nil)
<-r.Ready()
kv, err := r.KeyValue(ctx, "mesh-delivery-indexer_modules")
kv, err := r.KeyValue(ctx, "o_modules")
if err != nil {
t.Fatal(err)
}
go/views/viewstest/owner.go
+10 -7
View File
@@ -14,14 +14,17 @@ import (
"github.com/nats-io/nats.go"
"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"
)
// 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 (views.OwnerReadTool) 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 {
// each change, and the read tool answering on nc: tool is the owner's read tool as a reader's invokes names it
// (`<owner>.<tool>`), stream its views stream as its manifest's `streams` names it, both given as the reader's
// ProcessReader.Follow takes them. It returns the bucket the test writes.
func Serve(t testing.TB, ctx context.Context, js jetstream.JetStream, nc *nats.Conn, owner, state, tool, stream string, readers ...string) jetstream.KeyValue {
t.Helper()
bucket := owner + "_" + state
kv, err := js.KeyValue(ctx, bucket)
@@ -34,7 +37,7 @@ func Serve(t testing.TB, ctx context.Context, js jetstream.JetStream, nc *nats.C
if err != nil {
t.Fatal(err)
}
stream := views.StreamOf(owner)
stream = catalogue.StreamBusName(stream)
if _, err := js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{Name: stream, Subjects: []string{process.ModuleStreamSubject(owner, views.EventPrefix+">")}}); err != nil {
t.Fatal(err)
}
@@ -48,7 +51,7 @@ func Serve(t testing.TB, ctx context.Context, js jetstream.JetStream, nc *nats.C
if err := em.Start(ctx, state, kv, nil); err != nil {
t.Fatal(err)
}
tool := strings.TrimPrefix(views.OwnerReadTool(owner, state), owner+".")
tool = strings.TrimPrefix(tool, owner+".")
sub, err := nc.Subscribe("mesh.mod."+owner+".call."+tool+".>", func(m *nats.Msg) {
var in struct {
After string `json:"after"`
@@ -71,9 +74,9 @@ func Serve(t testing.TB, ctx context.Context, js jetstream.JetStream, nc *nats.C
// 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(views.OwnerReadTool(owner, state), owner+".")
stream := views.StreamOf(owner)
func Grants(owner, tool, stream, durable string) []string {
tool = strings.TrimPrefix(tool, owner+".")
stream = catalogue.StreamBusName(stream)
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 + ".>"}