diff --git a/go/keeps.go b/go/keeps.go new file mode 100644 index 0000000..cffdf9c --- /dev/null +++ b/go/keeps.go @@ -0,0 +1,80 @@ +package stdio + +import ( + "encoding/json" + "fmt" + "os" + "strings" + "sync" + + "git.novox.be/novox/mesh-sdk/go/announce" +) + +// What this bundle said it keeps (novox/hq ADR 0495 clause 2; issue 1790). +var ( + keepsMu sync.Mutex + kept map[string]bool // nil until Keeps: no state is refused then + announced []byte // the announcement, published once the bundle serves +) + +// Keeps says on the bus, as this bundle's module, every state it keeps and which of them others may read (novox/hq +// ADR 0495 clause 2): what a reader is granted is what the owner announced (clause 3). Call it before Serve; the +// announcement is published through the runtime once the bundle serves, or at once when it already does. From then +// on a state of the module's own that it did not announce is refused at the call, naming it, so code and announcement +// cannot drift (clause 5). It refuses a state no bucket could be named by and a state given twice. +func Keeps(states ...announce.State) error { + module := os.Getenv("MESH_MODULE") + if module == "" { + module = os.Getenv(ServedAs) + } + data, err := announce.Encode(announce.States{Module: module, Node: os.Getenv("MESH_NODE"), States: states}) + if err != nil { + return err + } + names := map[string]bool{} + for _, s := range states { + names[s.Name] = true + } + keepsMu.Lock() + kept, announced = names, data + keepsMu.Unlock() + currentMu.Lock() + serving := current != nil + currentMu.Unlock() + if serving { + return announce1() + } + return nil +} + +// announce1 publishes the announcement Keeps made, if any. +func announce1() error { + keepsMu.Lock() + data := announced + keepsMu.Unlock() + if data == nil { + return nil + } + return Emit(announce.Event, json.RawMessage(data)) +} + +// refusedUnannounced is the error of a call on a state of the module's own that Keeps did not name, or nil. A read of +// another module's state (`.`) is the grant's business. +func refusedUnannounced(name string) error { + if strings.Contains(name, ".") { + return nil + } + keepsMu.Lock() + defer keepsMu.Unlock() + if kept != nil && !kept[name] { + return fmt.Errorf("the state %q is not among the states this module announced it keeps (stdio.Keeps)", name) + } + return nil +} + +// forgetKeeps is for a test: nothing announced. +func forgetKeeps() { + keepsMu.Lock() + kept, announced = nil, nil + keepsMu.Unlock() +} diff --git a/go/keeps_test.go b/go/keeps_test.go new file mode 100644 index 0000000..9233a62 --- /dev/null +++ b/go/keeps_test.go @@ -0,0 +1,63 @@ +package stdio + +import ( + "bufio" + "encoding/json" + "io" + "strings" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/announce" +) + +// A Go bundle says what it keeps from its code (novox/hq ADR 0495 clauses 1 and 2): Keeps, called before Serve, +// announces through the runtime once the bundle serves, as the module the runtime names; and a state of its own that it +// did not announce is refused at the call, naming it, so code and announcement cannot drift. +func TestABundleAnnouncesWhatItKeepsAndBindsNothingElse(t *testing.T) { + t.Setenv(ServedAs, "chime") + t.Setenv("MESH_MODULE", "chime") + t.Setenv("MESH_NODE", "anchor") + defer forgetKeeps() + // Not yet serving: an earlier test's channel is gone with its test. + currentMu.Lock() + current = nil + currentMu.Unlock() + if err := Keeps(announce.State{Name: "rings", Readable: true}, announce.State{Name: "own"}); err != nil { + t.Fatal(err) + } + if _, err := State("other").Get("k"); err == nil || !strings.Contains(err.Error(), `"other"`) { + t.Fatalf("a state not announced was bound: %v", err) + } + inR, inW := io.Pipe() + outR, outW := io.Pipe() + go func() { _ = ServeOn("", nil, inR, outW) }() + defer inW.Close() + lines := bufio.NewScanner(outR) + got := make(chan map[string]any, 1) + go func() { + if lines.Scan() { + var m map[string]any + _ = json.Unmarshal(lines.Bytes(), &m) + got <- m + } + }() + var asked map[string]any + select { + case asked = <-got: + case <-time.After(3 * time.Second): + t.Fatal("the bundle announced nothing") + } + p, _ := asked["params"].(map[string]any) + if asked["method"] != "mesh/publish" || p["key"] != announce.Event { + t.Fatalf("the announcement was not published through the runtime: %v", asked) + } + raw, _ := json.Marshal(p["body"]) + s, err := announce.Decode(raw) + if err != nil || s.Module != "chime" || s.Node != "anchor" || !s.Readable("rings") || !s.Keeps("own") || s.Readable("own") { + t.Fatalf("announced %s (%v)", raw, err) + } + if err := Keeps(announce.State{Name: "bad_name"}); err == nil { + t.Fatal("an announcement no bucket could be named by was accepted") + } +} diff --git a/go/process/conn.go b/go/process/conn.go index 58efb40..ca46922 100644 --- a/go/process/conn.go +++ b/go/process/conn.go @@ -24,6 +24,12 @@ type Conn struct { mu sync.Mutex next uint64 pending map[string]map[uint64]context.CancelCauseFunc + + // What Keeps announced (keeps.go): the states' names (nil before Keeps), the announcement, and whether a + // reconnect announces it again. + kept map[string]bool + announced []byte + reannounces bool } // errPermissionRefused marks a context cancelled because the bus refused this account's publish of the diff --git a/go/process/fixture_integration_test.go b/go/process/fixture_integration_test.go index 27f8468..73991c5 100644 --- a/go/process/fixture_integration_test.go +++ b/go/process/fixture_integration_test.go @@ -38,6 +38,7 @@ var sendGrants = []string{ "$KV.mesh-mail_sends.>", "mesh.mod.mesh-mail.event.mail.send.failed", "mesh.mod.mesh-mail.event.mail.send.sent", + "mesh.mod.mesh-mail.event.states", "mesh.seat.operator-channel.call.tell.*.novox~mesh-mail~send", "mesh.seat.operator-channel.call.tell.novox~mesh-mail~send", "mesh.seat.operator-channel.tool.tell", diff --git a/go/process/keeps.go b/go/process/keeps.go new file mode 100644 index 0000000..22a1df7 --- /dev/null +++ b/go/process/keeps.go @@ -0,0 +1,71 @@ +package process + +import ( + "encoding/json" + "fmt" + + "github.com/nats-io/nats.go" + + "git.novox.be/novox/mesh-sdk/go/announce" +) + +// Keeps says on the bus, as this process's module, every state it keeps and which of them others may read (novox/hq +// ADR 0495 clause 2; issue 1790), on announce.Subject; what a reader is granted is what the owner announced (clause 3). +// It publishes at once and again after every reconnect, so the controller hears it whenever it missed it. From then on +// a state of the module that it did not announce is refused at the call, naming it, so code and announcement cannot +// drift (clause 5). It refuses a state no bucket could be named by and a state given twice. +func (c *Conn) Keeps(states ...announce.State) error { + if err := c.remember(states); err != nil { + return err + } + c.mu.Lock() + first := !c.reannounces + c.reannounces = true + c.mu.Unlock() + if first { + before := c.nc.Opts.ReconnectedCB + c.nc.SetReconnectHandler(func(nc *nats.Conn) { + if before != nil { + before(nc) + } + _ = c.publishKeeps() + }) + } + return c.publishKeeps() +} + +// remember makes the announcement of states and binds only them from now on. +func (c *Conn) remember(states []announce.State) error { + data, err := announce.Encode(announce.States{Module: c.account.Module, Node: c.account.Node, States: states}) + if err != nil { + return err + } + names := map[string]bool{} + for _, s := range states { + names[s.Name] = true + } + c.mu.Lock() + c.kept, c.announced = names, data + c.mu.Unlock() + return nil +} + +func (c *Conn) publishKeeps() error { + c.mu.Lock() + data := c.announced + c.mu.Unlock() + if err := c.Emit(announce.Event, json.RawMessage(data)); err != nil { + return fmt.Errorf("announcing what %s keeps: %w", c.account.Module, err) + } + return nil +} + +// refusedUnannounced is the error of a bind of a state Keeps did not name, or nil (also before Keeps). +func (c *Conn) refusedUnannounced(name string) error { + c.mu.Lock() + defer c.mu.Unlock() + if c.kept != nil && !c.kept[name] { + return fmt.Errorf("the state %q is not among the states %s announced it keeps (Conn.Keeps)", name, c.account.Module) + } + return nil +} diff --git a/go/process/keeps_integration_test.go b/go/process/keeps_integration_test.go new file mode 100644 index 0000000..8d3b0f3 --- /dev/null +++ b/go/process/keeps_integration_test.go @@ -0,0 +1,45 @@ +//go:build integration + +package process + +import ( + "strings" + "testing" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/announce" +) + +// 1790/S10: on a real bus, a process's Keeps lands on its module's own announcement subject, in the events stream, as +// the module's announcement; a state it did not announce is then refused at the call. +func TestAProcessAnnouncesWhatItKeepsOnTheBus(t *testing.T) { + url, account := grantedSendProcess(t) + _, admin := adminJS(t, url) + stream, err := admin.CreateStream(ctxFor(t), jetstream.StreamConfig{Name: "MAIL_STATES", Subjects: []string{announce.Subject("mesh-mail")}}) + if err != nil { + t.Fatal(err) + } + conn, err := Connect(account) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + if err := conn.Keeps(announce.State{Name: "accounts", Readable: true}, announce.State{Name: "drafts"}); err != nil { + t.Fatal(err) + } + msg, err := stream.GetMsg(ctxFor(t), 1) + if err != nil { + t.Fatalf("no announcement in the stream: %v", err) + } + if msg.Subject != "mesh.mod.mesh-mail.event.states" || msg.Header.Get("x-source") != "mesh-mail" { + t.Fatalf("announced on %s with %v", msg.Subject, msg.Header) + } + got, err := announce.Decode(msg.Data) + if err != nil || !got.Readable("accounts") || !got.Keeps("drafts") || got.Readable("drafts") || got.Node != "novox" { + t.Fatalf("announced %s (%v)", msg.Data, err) + } + if _, err := conn.State("outbox").Get("k"); err == nil || !strings.Contains(err.Error(), `"outbox"`) { + t.Fatalf("an unannounced state was bound: %v", err) + } +} diff --git a/go/process/keeps_test.go b/go/process/keeps_test.go new file mode 100644 index 0000000..387178a --- /dev/null +++ b/go/process/keeps_test.go @@ -0,0 +1,28 @@ +package process + +import ( + "strings" + "testing" + + "git.novox.be/novox/mesh-sdk/go/announce" +) + +// After Keeps a process binds only the states it announced (novox/hq ADR 0495 clause 5): any other of its module's +// states is refused at the call, naming it, before the bus is asked; before Keeps nothing is refused. +func TestAProcessBindsOnlyTheStatesItAnnounced(t *testing.T) { + c := &Conn{account: Account{Module: "mesh-mail", Node: "novox"}} + if err := c.remember([]announce.State{{Name: "accounts", Readable: true}}); err != nil { + t.Fatal(err) + } + _, err := c.State("drafts").Get("k") + if err == nil || !strings.Contains(err.Error(), `"drafts"`) || !strings.Contains(err.Error(), "Keeps") { + t.Fatalf("an unannounced state was bound: %v", err) + } + if err := c.remember([]announce.State{{Name: "a_b"}}); err == nil { + t.Fatal("an announcement no bucket could be named by was accepted") + } + got, err := announce.Decode(c.announced) + if err != nil || got.Module != "mesh-mail" || got.Node != "novox" || !got.Readable("accounts") { + t.Fatalf("the announcement is %s (%v)", c.announced, err) + } +} diff --git a/go/process/state.go b/go/process/state.go index 43221fa..8e294e3 100644 --- a/go/process/state.go +++ b/go/process/state.go @@ -39,6 +39,9 @@ func (s State) bucket() string { return s.Module + "_" + s.Name } func (c *Conn) State(name string) State { bucket := c.account.Module + "_" + name return State{Module: c.account.Module, Name: name, kv: func(ctx context.Context) (jetstream.KeyValue, error) { + if err := c.refusedUnannounced(name); err != nil { + return nil, err + } js, err := jetstream.New(c.nc) if err != nil { return nil, err diff --git a/go/state.go b/go/state.go index 70e1b35..56490b6 100644 --- a/go/state.go +++ b/go/state.go @@ -50,6 +50,9 @@ func State(name string) KeptState { return KeptState{name: name} } const stateTimeout = 30 * time.Second func (s KeptState) ask(verb string, params map[string]any) (json.RawMessage, error) { + if err := refusedUnannounced(s.name); err != nil { + return nil, err + } c, err := live() if err != nil { return nil, err diff --git a/go/stdio.go b/go/stdio.go index a61f4de..a072f40 100644 --- a/go/stdio.go +++ b/go/stdio.go @@ -180,6 +180,9 @@ func ServeOn(name string, tools []Tool, in io.Reader, out io.Writer) error { currentMu.Lock() current = c currentMu.Unlock() + // What the bundle said it keeps is announced once it serves: the runtime answers the publish on this channel, so + // it is asked from beside the loop that reads the answer. + go func() { _ = announce1() }() scanner := bufio.NewScanner(in) scanner.Buffer(make([]byte, 1<<20), 1<<24) for scanner.Scan() { diff --git a/package-lock.json b/package-lock.json index 6f9a353..04ae640 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "@novox/mesh-sdk", - "version": "0.1.14", + "version": "0.1.15", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "@novox/mesh-sdk", - "version": "0.1.14", + "version": "0.1.15", "devDependencies": { "@types/node": "^22.0.0", "typescript": "^5.6.0" diff --git a/package.json b/package.json index 5c158a5..3190b63 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@novox/mesh-sdk", - "version": "0.1.14", + "version": "0.1.15", "description": "The stable spine a Novox Mesh module's own code builds against.", "type": "module", "exports": { diff --git a/src/state/index.ts b/src/state/index.ts index 2f7da14..3b68b3e 100644 --- a/src/state/index.ts +++ b/src/state/index.ts @@ -6,6 +6,12 @@ // (`reads`, as `.`), and names each here the way it named it there. Every instance of // the owner may write; a reader reads and watches. The runtime that launched the bundle does the work // and refuses, with the reason, what the module did not declare. +// +// A module also says on the bus, from its code, which states it keeps and which others may read +// (`keeps`, novox/hq ADR 0495 clause 2): a reader is granted what the owner announced, never what a list +// or the owner's manifest says (clause 3). + +import { emit } from "../events/index.js"; /** One key's current value. */ export interface StateEntry { @@ -61,7 +67,7 @@ export function useStateChannel(c: StateChannel): void { channel = c; } -function bound(): StateChannel { +function boundChannel(): StateChannel { if (!channel) { throw new Error( "no runtime to keep state through — a module's state is reached through the runtime that launched " + @@ -71,8 +77,57 @@ function bound(): StateChannel { return channel; } +/** One state a module keeps, by its local name, and whether other modules may read it. */ +export interface Kept { + readonly name: string; + readonly readable?: boolean; +} + +const stateName = /^[a-z0-9][a-z0-9-]*$/; +let keptNames: Set | undefined; + +/** + * Say on the bus, as this bundle's module, every state it keeps and which of them others may read + * (novox/hq ADR 0495 clause 2; issue 1790): the announcement `states` on the module's own event subject, + * `{v: 1, module, node, states}` with the states sorted by name, which the controller keeps and grants + * readers from (clause 3). Call it as the bundle is imported. From then on a state of the module's own + * that it did not announce is refused at the call, naming it, so code and announcement cannot drift + * (clause 5). A state no bucket could be named by, or one given twice, is refused. + */ +export async function keeps(states: readonly Kept[]): Promise { + const module = process.env.MESH_MODULE || process.env.MESH_SERVED_MODULE || ""; + if (!stateName.test(module)) { + throw new Error(`"${module}" is not a module that keeps state: lower-case letters, digits and hyphens`); + } + const names = new Set(); + for (const k of states) { + if (!stateName.test(k.name)) { + throw new Error(`${module} announces the state "${k.name}", and a state is named locally: lower-case letters, digits and hyphens`); + } + if (names.has(k.name)) throw new Error(`${module} announces the state "${k.name}" twice`); + names.add(k.name); + } + keptNames = names; + const sorted = [...states] + .sort((a, b) => (a.name < b.name ? -1 : a.name > b.name ? 1 : 0)) + .map((k) => (k.readable ? { name: k.name, readable: true } : { name: k.name })); + const node = process.env.MESH_NODE; + await emit("states", { v: 1, module, ...(node ? { node } : {}), states: sorted }); +} + +/** The error of a call on a state of the module's own that `keeps` did not name, before the runtime is asked. */ +function refusedUnannounced(name: string): void { + if (!name.includes(".") && keptNames && !keptNames.has(name)) { + throw new Error(`the state "${name}" is not among the states this module announced it keeps (keeps)`); + } +} + /** A state this module keeps (by its local name) or reads (`.`). */ export function state(name: string): State { + const bound = (): StateChannel => { + refusedUnannounced(name); + return boundChannel(); + }; return { get: async (key) => (await bound().ask("mesh/state.get", { state: name, key })) as StateEntry | null, put: async (key, value) => diff --git a/test/fixtures/stdio-keeps.mjs b/test/fixtures/stdio-keeps.mjs new file mode 100644 index 0000000..4e1cf99 --- /dev/null +++ b/test/fixtures/stdio-keeps.mjs @@ -0,0 +1,11 @@ +// A launched bundle that says what it keeps (novox/hq ADR 0495 clause 2; issue 1790): it announces its states as it is +// imported, and a state of its own it did not announce is refused at the call. +import { keeps, state } from "../../dist/state/index.js"; +import { registerModuleTools } from "../../dist/tools/index.js"; +import { serveRegisteredOverStdio } from "../../dist/stdio/index.js"; + +await keeps([{ name: "servers", readable: true }, { name: "own" }]); +registerModuleTools("keeper", () => [ + { name: "unannounced", description: "reads a state it did not announce", input: {}, run: async () => state("drafts").get("k") }, +]); +await serveRegisteredOverStdio(); diff --git a/test/stdio.test.ts b/test/stdio.test.ts index df6f0e4..8710f99 100644 --- a/test/stdio.test.ts +++ b/test/stdio.test.ts @@ -273,3 +273,40 @@ test("topics match as the runtime's do: * one segment, # and ** any number", asy assert.ok(topicMatches("alpha.**", "alpha.a.b")); assert.ok(!topicMatches("beta.*", "alpha.happened")); }); + +test("a launched bundle announces what it keeps as it is imported, and binds no state it did not announce (ADR 0495, issue 1790)", async () => { + const child = spawn(process.execPath, [fileURLToPath(new URL("./fixtures/stdio-keeps.mjs", import.meta.url))], + { stdio: ["pipe", "pipe", "inherit"], env: { ...process.env, MESH_SERVED_MODULE: "keeper", MESH_MODULE: "keeper", MESH_NODE: "anchor" } }); + const seen: any[] = []; + let buffered = ""; + child.stdout.on("data", (d) => { + buffered += d.toString(); + let at: number; + while ((at = buffered.indexOf("\n")) >= 0) { + seen.push(JSON.parse(buffered.slice(0, at))); + buffered = buffered.slice(at + 1); + } + }); + const until = async (match: (m: any) => boolean) => { + for (let i = 0; i < 150 && !seen.some(match); i++) await new Promise((r) => setTimeout(r, 20)); + return seen.find(match); + }; + const send = (m: unknown) => child.stdin.write(JSON.stringify(m) + "\n"); + try { + const asked = await until((m) => m.method === "mesh/publish"); + assert.ok(asked, "the bundle announced nothing as it was imported"); + assert.equal(asked.params.key, "states"); + assert.deepEqual(asked.params.body, { + v: 1, module: "keeper", node: "anchor", states: [{ name: "own" }, { name: "servers", readable: true }], + }); + send({ jsonrpc: "2.0", id: asked.id, result: {} }); + send({ jsonrpc: "2.0", id: 1, method: "initialize", params: { protocolVersion: "2025-03-26", capabilities: {} } }); + await until((m) => m.id === 1); + send({ jsonrpc: "2.0", id: 2, method: "tools/call", params: { name: "unannounced", arguments: {} } }); + const answered = await until((m) => m.id === 2); + assert.match(JSON.stringify(answered), /drafts/, "a state the bundle did not announce was bound"); + assert.equal(seen.find((m) => m.method === "mesh/state.get"), undefined, "the runtime was asked for an unannounced state"); + } finally { + child.kill(); + } +});