diff --git a/go/packages_list_test.go b/go/packages_list_test.go index 6aa3664..e69d9b2 100644 --- a/go/packages_list_test.go +++ b/go/packages_list_test.go @@ -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-[-...], 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") diff --git a/go/views/process_reader.go b/go/views/process_reader.go index c12ce8a..a45fcb4 100644 --- a/go/views/process_reader.go +++ b/go/views/process_reader.go @@ -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): -// `._read_`, 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 _ - order []string + mu sync.Mutex + pics map[string]*Picture // by bucket name _ + 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 (`.`, 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 . 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 (`_`); 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): `-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` // `.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 (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) diff --git a/go/views/reader.go b/go/views/reader.go index 5e5d30b..e98b67b 100644 --- a/go/views/reader.go +++ b/go/views/reader.go @@ -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 (`.`). // A module in a stdio tools bundle declares `invokes` for the tools and `consumes` for the events, and names no `reads` diff --git a/go/views/status.go b/go/views/status.go index d60b89b..e040a80 100644 --- a/go/views/status.go +++ b/go/views/status.go @@ -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) { diff --git a/go/views/views.go b/go/views/views.go index 287f058..4e1a515 100644 --- a/go/views/views.go +++ b/go/views/views.go @@ -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.`, 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. diff --git a/go/views/views_test.go b/go/views/views_test.go index 7c4b703..8a3227a 100644 --- a/go/views/views_test.go +++ b/go/views/views_test.go @@ -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) } diff --git a/go/views/viewstest/owner.go b/go/views/viewstest/owner.go index eb5a1f8..9f11002 100644 --- a/go/views/viewstest/owner.go +++ b/go/views/viewstest/owner.go @@ -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 `_` (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 (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 +// (`.`), 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 + ".>"}