From 59311407e2401f3282eb9c38ec5c2407c3abeb32 Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 30 Oct 2026 17:50:56 +0100 Subject: [PATCH 01/11] One bad module must not drop its node: plans are composed module by module, a foreseen own secret fills a file, a built bus manifest keeps its bundle (issue 774) PlansHeld holds a module whose build cannot be taken in or whose plan refuses, by composeEach's rule, and composes every other module's plan; sealedFor gives a file the stand-in its own secret's file has (baserow on ace); shipsABundle reads a built manifest's Bundles (nats on novox). --- go/catalogue/bus_snapshot_test.go | 19 ++++- go/catalogue/manifest.go | 5 ++ go/catalogue/secrets_into_files.go | 4 + go/catalogue/secrets_into_files_test.go | 62 ++++++++------- go/compose/frombus/frombus.go | 62 +++++++++++++++ .../frombus/frombustest/frombustest_test.go | 76 +++++++++++++++---- 6 files changed, 186 insertions(+), 42 deletions(-) diff --git a/go/catalogue/bus_snapshot_test.go b/go/catalogue/bus_snapshot_test.go index e767d48..5f8edb3 100644 --- a/go/catalogue/bus_snapshot_test.go +++ b/go/catalogue/bus_snapshot_test.go @@ -1,6 +1,7 @@ package catalogue import ( + "encoding/json" "os" "path/filepath" "strings" @@ -27,14 +28,28 @@ func TestTheBussAccountSaysNothingOnTheBus(t *testing.T) { "claims":[{"name":"mesh-broker","scope":"mesh"}],"own-secrets":{"broker":"/run/broker"}, "emits":["bus.slow-consumer"],"tools":["bus_streams"], "build":{"artifacts":[{"name":"tools","kind":"bundle","language":"go","system":"arch","from":"cmd/t","binary":"t"}]}}`) - if _, err := ParseManifest(bundled); err != nil { + m, err := ParseManifest(bundled) + if err != nil { t.Fatalf("a bus module whose tools bundle declares what it emits was refused: %v", err) } + // And so does its built manifest, its bundle in Bundles and its build read away (issue 774: nats's built manifest + // was refused, and with it every plan of the control-node). + built, err := m.Resolve([]Built{{Name: "tools", Kind: ArtifactBundle, Reference: "store/tools:1", Digest: "sha256:1"}}) + if err != nil { + t.Fatal(err) + } + raw, err = json.Marshal(built) + if err != nil { + t.Fatal(err) + } + if _, err := ParseManifestForInputs(raw); err != nil { + t.Fatalf("the built manifest of a bus module whose tools bundle declares what it emits was refused: %v", err) + } quiet := []byte(`{"module":"bus","version":"1","provides":[{"name":"mesh-bus","scope":"mesh"}], "claims":[{"name":"mesh-broker","scope":"mesh"}],"own-secrets":{"broker":"/run/broker"}, "tools":["bus_streams"]}`) - m, err := ParseManifest(quiet) + m, err = ParseManifest(quiet) if err != nil { t.Fatalf("a bus module with an account and tools was refused: %v", err) } diff --git a/go/catalogue/manifest.go b/go/catalogue/manifest.go index 04e86b8..5db0e07 100644 --- a/go/catalogue/manifest.go +++ b/go/catalogue/manifest.go @@ -3378,6 +3378,11 @@ func whileStoppedProblems(m Manifest, r map[string]any, scheduled bool) []string // shipsABundle says the module builds a tools bundle, which the node's runtime carries and speaks for. func (m Manifest) shipsABundle() bool { + // A built manifest says its bundles in Bundles, its build read away (Resolve; issue 774: nats's built manifest was + // refused for the emits its tools bundle says). + if len(m.Bundles) > 0 { + return true + } if m.Build == nil { return false } diff --git a/go/catalogue/secrets_into_files.go b/go/catalogue/secrets_into_files.go index 0ed07c3..7baca99 100644 --- a/go/catalogue/secrets_into_files.go +++ b/go/catalogue/secrets_into_files.go @@ -59,6 +59,10 @@ func sealedFor(m Manifest, needs []Needed, with Rendering) (map[string]string, e for name := range m.OwnSecrets.Plain() { if value := with.Needed[m.Module][name]; value != "" { sealed[name] = value + } else if with.Foreseen[m.Module][name] { + // A stand-in where its own file has one (the Foreseeing and Planning modes, issue 774): a file naming it + // composes as its own file does, never refused for a value no such composition reads. + sealed[name] = ForeseenSealed } } for _, to := range m.SecretRequirements() { diff --git a/go/catalogue/secrets_into_files_test.go b/go/catalogue/secrets_into_files_test.go index 1a92a7e..f8d9bb3 100644 --- a/go/catalogue/secrets_into_files_test.go +++ b/go/catalogue/secrets_into_files_test.go @@ -21,34 +21,42 @@ func fileNamed(out []map[string]any, id string) map[string]any { return nil } -// A module's own credential, put into a file it wrote. +// A module's own credential, put into a file it wrote: the value sealed to the node, or, in a composition asked ahead of +// the send or for a plan alone, the stand-in its own file has (issue 774: baserow's config.json refused ace's plans). func TestAFileGetsTheSecretItsContentAsksFor(t *testing.T) { - r := Resolution{Node: "anchor", Modules: []Manifest{{ - Module: "gitea", - OwnSecrets: OwnSecrets{"admin": {Path: "/var/lib/gitea/admin.env"}}, - Resources: []map[string]any{{ - "id": "conf", "type": "file", "path": "/etc/gitea/app.ini", - "content": "[security]\nSECRET_KEY = ${secret:admin}\n", - }}, - }}} - out, err := r.Declaration(Rendering{ - Needed: map[string]map[string]string{"gitea": {"admin": "sealed-admin"}}, - }) - if err != nil { - t.Fatal(err) - } - file := fileNamed(out, "gitea.conf") - if file == nil { - t.Fatalf("no file in %v", out) - } - got, _ := file["secrets"].(map[string]any) - if got["admin"] != "sealed-admin" { - t.Errorf("the file was sent without its secret: %v", file) - } - // The hole is still a hole on the wire. The mesh does not fill it in; the host does, on the - // machine, which is the only place both halves exist. - if !strings.Contains(file["content"].(string), "${secret:admin}") { - t.Errorf("the mesh substituted it itself: %v", file["content"]) + for _, row := range []struct { + name string + with Rendering + want string + }{ + {"sealed", Rendering{Needed: map[string]map[string]string{"gitea": {"admin": "sealed-admin"}}}, "sealed-admin"}, + {"foreseen", Rendering{Foreseen: map[string]map[string]bool{"gitea": {"admin": true}}}, ForeseenSealed}, + } { + r := Resolution{Node: "anchor", Modules: []Manifest{{ + Module: "gitea", + OwnSecrets: OwnSecrets{"admin": {Path: "/var/lib/gitea/admin.env"}}, + Resources: []map[string]any{{ + "id": "conf", "type": "file", "path": "/etc/gitea/app.ini", + "content": "[security]\nSECRET_KEY = ${secret:admin}\n", + }}, + }}} + out, err := r.Declaration(row.with) + if err != nil { + t.Fatalf("%s: %v", row.name, err) + } + file := fileNamed(out, "gitea.conf") + if file == nil { + t.Fatalf("%s: no file in %v", row.name, out) + } + got, _ := file["secrets"].(map[string]any) + if got["admin"] != row.want { + t.Errorf("%s: the file was sent without its secret: %v", row.name, file) + } + // The hole is still a hole on the wire. The mesh does not fill it in; the host does, on the + // machine, which is the only place both halves exist. + if !strings.Contains(file["content"].(string), "${secret:admin}") { + t.Errorf("%s: the mesh substituted it itself: %v", row.name, file["content"]) + } } } diff --git a/go/compose/frombus/frombus.go b/go/compose/frombus/frombus.go index 2fedc4c..ac67876 100644 --- a/go/compose/frombus/frombus.go +++ b/go/compose/frombus/frombus.go @@ -439,6 +439,68 @@ func Plans(ctx context.Context, r Reader, mesh Mesh, s Scope) (map[string]catalo return PlansOf(in, s.Node) } +// PlansHeld is Plans composed module by module (issue 774: one bad module must not drop its node): a module placed on +// the scope's node whose build cannot be read or taken in, or whose plan refuses the node's, is held, by the one rule +// Compose follows (composeEach, culprit): composed as its stub so what it provides and requires still shapes the rest, +// named in held with why, and none of its plan in plans. Only a refusal that is no single module's refuses the node, +// err. The seat mesh-provisioner's holder composes so: a node with one bad manifest still has every other module's plan. +func PlansHeld(ctx context.Context, r Reader, mesh Mesh, s Scope) (plans map[string]catalogue.Plan, held map[string]string, err error) { + held = map[string]string{} + in, err := input(ctx, r, mesh, s, func(module string, err error) { held[module] = err.Error() }) + if err != nil { + return nil, held, err + } + plans, err = plansHeldOf(in, s.Node, held) + return plans, held, err +} + +// plansHeldOf is the plans of node over in, every module in held held, and each module a refusal is of held too until +// the plans compose (culprit's rule over the Planning mode: its stub first, then its bare stub). +func plansHeldOf(in compose.Input, node string, held map[string]string) (map[string]catalogue.Plan, error) { + for _, m := range sortedKeys(held) { + in = stubbed(in, node, m, "") + } + compose1 := func(in compose.Input) (map[string]catalogue.Plan, error) { return plansIn(in, node, compose.Planning) } + // Each try holds one module more or leaves out all of one held module's offer: bounded by twice the modules placed. + for tries := 0; tries <= 2*len(assignedOn(in, node))+1; tries++ { + plans, err := compose1(in) + if err == nil { + for m := range held { + delete(plans, m) + } + return plans, nil + } + var order []string + for _, m := range assignedOn(in, node) { + if strings.Contains(err.Error(), m) && len(order) < maxCulprits { + order = append(order, m) + } + } + module, leave := "", "" + find: + for _, l := range []string{"", leaveAll} { + for _, m := range order { + if _, done := held[m]; done && l == "" { + continue + } + if _, again := compose1(stubbed(in, node, m, l)); again == nil || again.Error() != err.Error() { + module, leave = m, l + break find + } + } + } + if module == "" { + return nil, err + } + if _, done := held[module]; !done { + held[module] = err.Error() + } + in = stubbed(in, node, module, leave) + } + return nil, fmt.Errorf("frombus: the plans of %s do not compose with %d module(s) held: %s", node, len(held), + strings.Join(sortedKeys(held), ", ")) +} + // PlansOf is Plans over an input already filled, rendered in the Planning mode (compose.Planning): every own secret the // input holds none for stands in, so no value and no record of one is needed. func PlansOf(in compose.Input, node string) (map[string]catalogue.Plan, error) { diff --git a/go/compose/frombus/frombustest/frombustest_test.go b/go/compose/frombus/frombustest/frombustest_test.go index edb8388..476bb1d 100644 --- a/go/compose/frombus/frombustest/frombustest_test.go +++ b/go/compose/frombus/frombustest/frombustest_test.go @@ -2,25 +2,75 @@ package frombustest import ( "context" + "strings" "testing" "git.novox.be/novox/mesh-sdk/go/compose/frombus" + "git.novox.be/novox/mesh-sdk/go/delivery" ) // ADR 0570 clause 2 (issue 1954; a3's H5): the in-memory mesh composes a plan from the bus that hands restic to the -// provisioner and names its postgres-database, provided by postgres on n1. +// provisioner and names its postgres-database, provided by postgres on n1. And one bad module must not drop its node +// (issue 774): with a module placed beside it whose built manifest cannot be used, or whose plan refuses, PlansHeld +// still hands restic's plan and names the bad one with why, where Plans refuses the whole node. func TestTheProvisionedMeshHandsAPlanWithItsProvision(t *testing.T) { - b, err := Provisioned() - if err != nil { - t.Fatal(err) - } - plans, err := frombus.Plans(context.Background(), b, frombus.Mesh{}, frombus.Scope{Node: Node}) - if err != nil { - t.Fatal(err) - } - restic, handed := plans["restic"] - if !handed || len(restic.Provisions) != 1 || restic.Provisions[0].Provision != "postgres-database" || - restic.Provisions[0].Provider != Node || restic.Provisions[0].ProviderModule != "postgres" { - t.Fatalf("the plans are %+v", plans) + fp := "src2:" + strings.Repeat("9", 64) + for _, row := range []struct { + name string + bad string // the indexer's manifest of a module "backup" placed on n1; empty places none + // built is the manifest backup's done build names; empty gives it no build + built string + why string // words backup's hold says + }{ + {name: "every module composes"}, + {name: "a built manifest that cannot be used", + bad: `{"module":"backup","version":"1","build":{"artifacts":[{"name":"files","kind":"archive","from":"files"}]}}`, + built: `{"module":"backup","version":"1","requires":["amqp"]}`, why: "backup"}, + {name: "a plan that refuses", + bad: `{"module":"backup","version":"1","resources":[{"id":"conf","type":"file","path":"/etc/backup.conf",` + + `"mode":"0600","content":"key=${secret:nope}"}]}`, why: "${secret:nope}"}, + } { + b, err := Provisioned() + if err != nil { + t.Fatal(err) + } + if row.bad != "" { + p := b.Placement[Node] + p.Modules = append(p.Modules, delivery.PlacedModule{Module: "backup"}) + b.Placement[Node] = p + digest := delivery.ManifestDigest([]byte(row.bad)) + b.Manifests[digest] = []byte(row.bad) + b.Module["backup"] = delivery.Module{FP2: fp, Manifest: digest} + if row.built != "" { + builtDigest := delivery.ManifestDigest([]byte(row.built)) + b.Manifests[builtDigest] = []byte(row.built) + b.Builds["backup "+fp] = &delivery.BuilderBuild{State: delivery.BuildDone, Manifest: builtDigest, + Artifacts: []delivery.BuiltArtifact{{Name: "files", Kind: "archive", Reference: "artifact-store://backup/files/blobs/x", Digest: "sha256:x"}}} + } + if _, err := frombus.Plans(context.Background(), b, frombus.Mesh{}, frombus.Scope{Node: Node}); err == nil { + t.Errorf("%s: Plans composed the node; the row does not refuse it", row.name) + } + } + plans, held, err := frombus.PlansHeld(context.Background(), b, frombus.Mesh{}, frombus.Scope{Node: Node}) + if err != nil { + t.Fatalf("%s: %v", row.name, err) + } + restic, handed := plans["restic"] + if !handed || len(restic.Provisions) != 1 || restic.Provisions[0].Provision != "postgres-database" || + restic.Provisions[0].Provider != Node || restic.Provisions[0].ProviderModule != "postgres" { + t.Fatalf("%s: the plans are %+v (held %v)", row.name, plans, held) + } + if row.bad == "" { + if len(held) > 0 { + t.Errorf("%s: held %v", row.name, held) + } + continue + } + if why, ok := held["backup"]; !ok || !strings.Contains(why, row.why) || len(held) != 1 { + t.Errorf("%s: held %v, want backup alone, saying %q", row.name, held, row.why) + } + if _, planned := plans["backup"]; planned { + t.Errorf("%s: the held module has a plan", row.name) + } } } From 7b02f63daf0e6accaca38028c7ec9e987b0ce98a Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 30 Oct 2026 18:02:18 +0100 Subject: [PATCH 02/11] Refuse a key-value write too large for its bucket before it is put (issue 774) mesh-grants' sealed grants and the controller's merge record each met "message size exceeds maximum" (JetStream 10054) only after the put, and a writer that retries looped on a value that can never land; an update over the limit was even answered as a revision mismatch, which every compare-and-set writer retries. Quick fix: cap the two values in their modules. Thorough fix, chosen because the kind was met twice and every module writes through the SDK: the package kvlimit checks a value against its bucket's largest message (read from its stream and kept) before the put, naming the bucket, the key, the size and the limit, and turns the bus's own refusal into the same *TooLarge (errors.Is kvlimit.ErrTooLarge), so no writer retries it. The SDK's own writers use it: process.State (Put, Create, CreateTTL, Update), process.AwaitedBucket, the runtime state of a bundle (stdio KeptState, against the module state's 256 KiB) and mesh-hand-compose. A value that can outgrow its bucket is split over keys or kept in the object store with its name in the key. --- go/cmd/mesh-hand-compose/main.go | 5 +- go/kvlimit/kvlimit.go | 175 +++++++++++++++++++++++++++ go/process/buckets.go | 6 +- go/process/state.go | 9 +- go/process/state_integration_test.go | 25 +++- go/state.go | 22 +++- go/state_test.go | 15 +++ 7 files changed, 247 insertions(+), 10 deletions(-) create mode 100644 go/kvlimit/kvlimit.go diff --git a/go/cmd/mesh-hand-compose/main.go b/go/cmd/mesh-hand-compose/main.go index cf59e98..624b0b2 100644 --- a/go/cmd/mesh-hand-compose/main.go +++ b/go/cmd/mesh-hand-compose/main.go @@ -32,6 +32,7 @@ import ( "git.novox.be/novox/mesh-sdk/go/compose" "git.novox.be/novox/mesh-sdk/go/declaration" "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/kvlimit" ) func main() { @@ -190,7 +191,7 @@ func writeIt(url, creds, node, bucket, key string, encoded []byte, digest string if err != nil { return fmt.Errorf("the bucket %s: %w", bucket, err) } - if _, err := composed.Put(ctx, key, encoded); err != nil { + if _, err := kvlimit.Put(ctx, composed, key, encoded); err != nil { return fmt.Errorf("%s in %s: %w", key, bucket, err) } value, err := delivery.EncodeComposedDigest(delivery.ComposedDigest{Digest: digest, Judgement: delivery.JudgedCannot, @@ -202,7 +203,7 @@ func writeIt(url, creds, node, bucket, key string, encoded []byte, digest string if err != nil { return fmt.Errorf("the bucket %s: %w", delivery.ControllerComposedBucket, err) } - if _, err := digests.Put(ctx, key, value); err != nil { + if _, err := kvlimit.Put(ctx, digests, key, value); err != nil { return fmt.Errorf("%s in %s: %w", key, delivery.ControllerComposedBucket, err) } return nil diff --git a/go/kvlimit/kvlimit.go b/go/kvlimit/kvlimit.go new file mode 100644 index 0000000..95d51ee --- /dev/null +++ b/go/kvlimit/kvlimit.go @@ -0,0 +1,175 @@ +// Package kvlimit refuses a key-value write whose value is larger than its bucket takes, before it is put, naming the +// bucket, the key, the size and the limit (issue 774, sweep item 5). The bus refuses such a write too, but only after +// the put, as "message size exceeds maximum" (JetStream error 10054); a writer that retries on errors then loops on a +// value that can never land. A refusal from here, or the bus's own one passed through Refused, is a *TooLarge that +// errors.Is ErrTooLarge: a writer stops on it and says it, it never writes the same value again. +// +// A value that may outgrow its bucket is shaped one of two ways, never retried: +// - split over keys: one key per part that changes on its own (per node, per module), each well under the limit; +// - or kept in the object store, with only its name and digest in the key. +package kvlimit + +import ( + "context" + "errors" + "fmt" + "strings" + "sync" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" +) + +// ModuleStateValueBytes is the most a value in a module's declared state may hold: every module bucket is made with +// it (buscompose.StateMaxValueBytes). +const ModuleStateValueBytes = 256 * 1024 + +// HeaderRoom is what a write's headers may take of the limit: the bus counts headers and value together, and a create, +// an update or a key's time to live adds one header each. A value closer to the limit than this is refused. +const HeaderRoom = 128 + +// codeMessageExceedsMaximum is JetStream's error code for a message over the stream's largest (JSStreamMessageExceedsMaximumErr). +const codeMessageExceedsMaximum jetstream.ErrorCode = 10054 + +// ErrTooLarge is what every *TooLarge is: a value its bucket does not take, which no retry will change. +var ErrTooLarge = errors.New("a value larger than its bucket takes") + +// TooLarge is one refused write. Limit is zero when the bus refused it and its limit was not known here. +type TooLarge struct { + Bucket, Key string + Size, Limit int64 +} + +func (e *TooLarge) Error() string { + limit := "the most its bucket takes" + if e.Limit > 0 { + limit = fmt.Sprintf("the %d bytes the bucket takes (headers included)", e.Limit) + } + return fmt.Sprintf("the value for %s in %s is %d bytes, over %s: it is refused and written again it is refused again; "+ + "split it over keys or keep it in the object store with its name in the key", e.Key, e.Bucket, e.Size, limit) +} + +func (e *TooLarge) Unwrap() error { return ErrTooLarge } + +// Check refuses a value of size bytes for key in bucket when it does not fit limit with HeaderRoom to spare; a limit of +// zero or less is no limit. +func Check(bucket, key string, size int, limit int64) error { + if limit > 0 && int64(size)+HeaderRoom > limit { + return &TooLarge{Bucket: bucket, Key: key, Size: int64(size), Limit: limit} + } + return nil +} + +// Refused is err as a *TooLarge when it is the bus's refusal of a value too large (the stream's 10054, the client's +// maximum payload, or either said in words by a runtime that passed it on), and err unchanged otherwise. +func Refused(bucket, key string, size int, limit int64, err error) error { + if err == nil || errors.Is(err, ErrTooLarge) { + return err + } + var api *jetstream.APIError + tooLarge := errors.As(err, &api) && api.ErrorCode == codeMessageExceedsMaximum + tooLarge = tooLarge || errors.Is(err, nats.ErrMaxPayload) + words := err.Error() + tooLarge = tooLarge || strings.Contains(words, "message size exceeds maximum") || strings.Contains(words, "maximum payload exceeded") + if !tooLarge { + return err + } + forget(bucket) + return fmt.Errorf("%w (the bus: %v)", &TooLarge{Bucket: bucket, Key: key, Size: int64(size), Limit: limit}, err) +} + +// limits are the buckets' largest messages as read, by bucket name: a limit is kept once read, and only a limit (an +// unlimited bucket is read again each time). A kept limit that refuses a value is read again before the refusal is +// given, so a bucket made again with a larger limit is never refused by an old one; one made with a smaller limit is +// refused by the bus, and Refused forgets the kept one. +var limits sync.Map // bucket name → int64 + +func forget(bucket string) { limits.Delete(bucket) } + +// LimitOf is the largest message kv's bucket takes, read from its stream and kept; zero when it has no limit or it +// cannot be read (the write then goes, and Refused names the bus's refusal). +func LimitOf(ctx context.Context, kv jetstream.KeyValue) int64 { + if l, ok := limits.Load(nameOf(kv)); ok { + return l.(int64) + } + return readLimit(ctx, kv) +} + +func readLimit(ctx context.Context, kv jetstream.KeyValue) (limit int64) { + name := nameOf(kv) + if name == "" { + return 0 + } + // A test's fake that implements only the writes it needs panics on Status; its limit is not known. + defer func() { + if recover() != nil { + limit = 0 + } + }() + st, err := kv.Status(ctx) + if err != nil { + return 0 + } + s, ok := st.(*jetstream.KeyValueBucketStatus) + if !ok || s.StreamInfo() == nil || s.StreamInfo().Config.MaxMsgSize <= 0 { + forget(name) + return 0 + } + limit = int64(s.StreamInfo().Config.MaxMsgSize) + limits.Store(name, limit) + return limit +} + +// fits is the limit of kv's bucket and nil when a value of size bytes fits it, or the *TooLarge refusing it, checked +// against a freshly read limit. +func fits(ctx context.Context, kv jetstream.KeyValue, key string, size int) (int64, error) { + limit := LimitOf(ctx, kv) + if Check(nameOf(kv), key, size, limit) == nil { + return limit, nil + } + limit = readLimit(ctx, kv) + return limit, Check(nameOf(kv), key, size, limit) +} + +// nameOf is kv's bucket name, or "" for none or a test's fake that does not say it. +func nameOf(kv jetstream.KeyValue) (name string) { + if kv == nil { + return "" + } + defer func() { + if recover() != nil { + name = "" + } + }() + return kv.Bucket() +} + +// Put is kv.Put, refused before it is put when value does not fit the bucket. +func Put(ctx context.Context, kv jetstream.KeyValue, key string, value []byte) (uint64, error) { + limit, err := fits(ctx, kv, key, len(value)) + if err != nil { + return 0, err + } + rev, err := kv.Put(ctx, key, value) + return rev, Refused(nameOf(kv), key, len(value), limit, err) +} + +// Create is kv.Create, refused before it is put when value does not fit the bucket. +func Create(ctx context.Context, kv jetstream.KeyValue, key string, value []byte, opts ...jetstream.KVCreateOpt) (uint64, error) { + limit, err := fits(ctx, kv, key, len(value)) + if err != nil { + return 0, err + } + rev, err := kv.Create(ctx, key, value, opts...) + return rev, Refused(nameOf(kv), key, len(value), limit, err) +} + +// Update is kv.Update, refused before it is put when value does not fit the bucket. +func Update(ctx context.Context, kv jetstream.KeyValue, key string, value []byte, revision uint64) (uint64, error) { + limit, err := fits(ctx, kv, key, len(value)) + if err != nil { + return 0, err + } + rev, err := kv.Update(ctx, key, value, revision) + return rev, Refused(nameOf(kv), key, len(value), limit, err) +} diff --git a/go/process/buckets.go b/go/process/buckets.go index bc2421c..a84d406 100644 --- a/go/process/buckets.go +++ b/go/process/buckets.go @@ -10,6 +10,8 @@ import ( "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/kvlimit" ) // **A part does not stop for a bucket it can work without** (novox/hq issue 1945, met twice on 2026-10-25: the delivery @@ -167,7 +169,7 @@ func (b *AwaitedBucket) Put(ctx context.Context, key string, value []byte) (uint kv = b.kv b.mu.Unlock() } - return kv.Put(ctx, key, value) + return kvlimit.Put(ctx, kv, key, value) } // bind tries one STREAM.INFO of the bucket and, when it is there, binds it; it answers the bucket or nil. @@ -196,7 +198,7 @@ func (b *AwaitedBucket) bound(kv jetstream.KeyValue, appeared bool) jetstream.Ke written, failed := 0, 0 for _, key := range b.order { ctx, cancel := context.WithTimeout(b.ctx, bindWithin) - _, err := kv.Put(ctx, key, b.pending[key]) + _, err := kvlimit.Put(ctx, kv, key, b.pending[key]) cancel() if err != nil { failed++ diff --git a/go/process/state.go b/go/process/state.go index 90523b4..d2a462f 100644 --- a/go/process/state.go +++ b/go/process/state.go @@ -13,6 +13,7 @@ import ( "github.com/nats-io/nats.go/jetstream" stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/kvlimit" ) // stateTimeout bounds a state call (Get, Create, Update) and an Emit (emit.go): a denied publish is @@ -103,7 +104,7 @@ func (s State) Create(key string, value any) (uint64, error) { if err != nil { return 0, err } - rev, err := kv.Create(ctx, key, data) + rev, err := kvlimit.Create(ctx, kv, key, data) if errors.Is(err, jetstream.ErrKeyExists) { return 0, fmt.Errorf("%s: %w", key, stdio.ErrChanged) } @@ -134,7 +135,7 @@ func (s State) CreateTTL(key string, value any, ttl time.Duration) (uint64, erro if err != nil { return 0, err } - rev, err := kv.Create(ctx, key, data, jetstream.KeyTTL(ttl)) + rev, err := kvlimit.Create(ctx, kv, key, data, jetstream.KeyTTL(ttl)) if errors.Is(err, jetstream.ErrKeyExists) { return 0, fmt.Errorf("%s: %w", key, stdio.ErrChanged) } @@ -171,7 +172,7 @@ func (s State) Update(key string, value any, revision uint64) (uint64, error) { if err != nil { return 0, err } - rev, err := kv.Update(ctx, key, data, revision) + rev, err := kvlimit.Update(ctx, kv, key, data, revision) if errors.Is(err, jetstream.ErrKeyRevisionMismatch) { return 0, fmt.Errorf("%s: %w", key, stdio.ErrChanged) } @@ -193,7 +194,7 @@ func (s State) Put(key string, value any) (uint64, error) { if err != nil { return 0, err } - rev, err := kv.Put(ctx, key, data) + rev, err := kvlimit.Put(ctx, kv, key, data) if err != nil { return 0, fmt.Errorf("%s's state %s, key %s: %w", s.Module, s.Name, key, err) } diff --git a/go/process/state_integration_test.go b/go/process/state_integration_test.go index 7d86e70..588959d 100644 --- a/go/process/state_integration_test.go +++ b/go/process/state_integration_test.go @@ -12,6 +12,7 @@ import ( "github.com/nats-io/nats.go/jetstream" stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/kvlimit" ) // 1188/AC1: given a process connection to a test bus granted the bucket mesh-mail_sends, when it creates a @@ -148,7 +149,8 @@ func TestStateRefusesAnUngrantedBucketNamingIt(t *testing.T) { func TestStatePutAndKeysOnAGrantedBucket(t *testing.T) { url, account := grantedSendProcess(t) _, adminJetStream := adminJS(t, url) - if _, err := adminJetStream.CreateKeyValue(ctxFor(t), jetstream.KeyValueConfig{Bucket: "mesh-mail_sends"}); err != nil { + admins, err := adminJetStream.CreateKeyValue(ctxFor(t), jetstream.KeyValueConfig{Bucket: "mesh-mail_sends", MaxValueSize: 1024}) + if err != nil { t.Fatalf("the admin could not make the bucket: %v", err) } conn, err := Connect(account) @@ -158,6 +160,27 @@ func TestStatePutAndKeysOnAGrantedBucket(t *testing.T) { defer conn.Close() s := conn.State("sends") + // Issue 774 (sweep item 5): a value over the bucket's largest is refused before it is put, naming the bucket, the key, + // its size and the limit, as kvlimit.ErrTooLarge, so a writer stops on it; nothing reaches the bucket. + big := strings.Repeat("x", 2000) + for name, write := range map[string]func() (uint64, error){ + "put": func() (uint64, error) { return s.Put("big", big) }, + "create": func() (uint64, error) { return s.Create("big", big) }, + "update": func() (uint64, error) { return s.Update("big", big, 1) }, + } { + _, err := write() + var tooLarge *kvlimit.TooLarge + if !errors.Is(err, kvlimit.ErrTooLarge) || !errors.As(err, &tooLarge) || tooLarge.Limit != 1024 || + tooLarge.Bucket != "mesh-mail_sends" || tooLarge.Key != "big" || tooLarge.Size != 2002 || strings.Contains(err.Error(), "the bus:") { + t.Errorf("a %s of 2002 bytes into a bucket of 1024: %v", name, err) + } + } + // The bus's own refusal (JetStream 10054), as a writer that put past the check meets it, is the same refusal. + _, err = admins.Put(ctxFor(t), "big", []byte(big)) + if err == nil || !errors.Is(kvlimit.Refused("mesh-mail_sends", "big", 2000, 1024, err), kvlimit.ErrTooLarge) { + t.Errorf("the bus's refusal is not ErrTooLarge: %v", err) + } + if keys, err := s.Keys(); err != nil || len(keys) != 0 { t.Fatalf("Keys of an empty bucket: %v, %v", keys, err) } diff --git a/go/state.go b/go/state.go index ac3d5fd..ad49de2 100644 --- a/go/state.go +++ b/go/state.go @@ -5,6 +5,8 @@ import ( "errors" "fmt" "time" + + "git.novox.be/novox/mesh-sdk/go/kvlimit" ) // State — the current value of something a module keeps on the bus, which every machine sees, @@ -64,12 +66,30 @@ func (s KeptState) ask(verb string, params map[string]any) (json.RawMessage, err if err := refusedUnannounced(s.name); err != nil { return nil, err } + // A write's value is measured before it is sent: one over the bucket's limit is refused here, by name, and a + // refusal the bus gives anyway comes back as the same *kvlimit.TooLarge, so no writer retries it (issue 774). + size := -1 + key, _ := params["key"].(string) + if value, ok := params["value"]; ok { + data, err := json.Marshal(value) + if err != nil { + return nil, fmt.Errorf("the value for %s in the state %s: %w", key, s.name, err) + } + if err := kvlimit.Check("the state "+s.name, key, len(data), kvlimit.ModuleStateValueBytes); err != nil { + return nil, err + } + params["value"], size = json.RawMessage(data), len(data) + } c, err := live() if err != nil { return nil, err } params["state"] = s.name - return c.ask("mesh/state."+verb, params, stateTimeout) + raw, err := c.ask("mesh/state."+verb, params, stateTimeout) + if size >= 0 { + err = kvlimit.Refused("the state "+s.name, key, size, kvlimit.ModuleStateValueBytes, err) + } + return raw, err } // Get is a key's current value, or nil when it has none. diff --git a/go/state_test.go b/go/state_test.go index 7b5f109..dc80765 100644 --- a/go/state_test.go +++ b/go/state_test.go @@ -122,6 +122,21 @@ func TestABundleKeepsAndWatchesStateThroughTheRuntime(t *testing.T) { if r := next(); r["id"] != float64(2) || r["result"].(map[string]any)["content"].([]any)[0].(map[string]any)["text"] != "9" { t.Fatalf("the put answered %v", r) } + // Issue 774 (sweep item 5): a value over a module state's largest is refused before the runtime is asked, naming the + // key, its size and the limit; one the runtime's bus refuses anyway comes back as the same refusal. + send(map[string]any{"jsonrpc": "2.0", "id": 30, "method": "tools/call", "params": map[string]any{"name": "remember", + "arguments": map[string]any{"key": "all.big", "value": strings.Repeat("x", 300*1024)}}}) + if r := next(); r["id"] != float64(30) || !strings.Contains(fmt.Sprint(r), "all.big in the state servers is 307202 bytes, over the 262144") { + t.Fatalf("a value over the limit was not refused before the runtime was asked: %v", r) + } + send(map[string]any{"jsonrpc": "2.0", "id": 31, "method": "tools/call", "params": map[string]any{"name": "remember", + "arguments": map[string]any{"key": "all.near", "value": "x"}}}) + near := next() + send(map[string]any{"jsonrpc": "2.0", "id": near["id"], "error": map[string]any{"code": -32000, + "message": "nats: API error: code=400 err_code=10054 description=message size exceeds maximum allowed"}}) + if r := next(); r["id"] != float64(31) || !strings.Contains(fmt.Sprint(r), "written again it is refused again") { + t.Fatalf("the bus's refusal was not named as too large: %v", r) + } // PutTTL and CreateTTL ask the runtime's put and create with the key's time to live in whole seconds; a create the // runtime refuses as changed (-32010) is ErrChanged. for i, create := range []bool{false, true} { From e703e9b0e2e57464a3cab84542a4f7881ca04a75 Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 30 Oct 2026 18:08:02 +0100 Subject: [PATCH 03/11] Judge every read for its reader where the bus's users are composed (issue 774) Since 13:25 the vault, the assigner, the watchman, the grants part and the artifact store's tools were refused mesh-controller.lease, and the vault mesh-grants.accounts, though the composer names them those states' readers (ADR 0547 clause 9): the controller splits reads with its own copy of the reader-less ReadsAnnounced, which knows no named readers, and Users only moved reads from granted to refused, never back. Quick fix: the controller calls ReadsAnnouncedBy with the module. Thorough fix, chosen because the kind was met on four owners and the judgement had two copies: Users judges every read again for its reader against the announcements in Records (withAnnouncedReads), whatever split the caller made, and the reader-less ReadsAnnounced is removed so no caller can judge without the reader. A rule over manifests was not added: ADR 0495 clauses 1 and 5 keep readability out of the manifest and replaced the merge gate's reads check by each module's announcement test. gitea.bound and disk-load.readings stay refused: their owners announce them not readable, which their own code fixes. --- go/buscompose/announced.go | 16 ++++++---------- go/buscompose/buscompose_test.go | 21 +++++++++++++++++++++ go/buscompose/userlist_switch.go | 18 ++++++++++++++++++ go/buscompose/users.go | 8 ++++---- 4 files changed, 49 insertions(+), 14 deletions(-) diff --git a/go/buscompose/announced.go b/go/buscompose/announced.go index 09637ba..90c57d8 100644 --- a/go/buscompose/announced.go +++ b/go/buscompose/announced.go @@ -42,16 +42,12 @@ func ControllerAnnouncement(node string) announce.States { return s } -// ReadsAnnounced splits a reader's `reads` (`.`) by what each owner announced: granted when the owner -// announced the state readable, or announced nothing yet (the transition); refused when the owner announced and did -// not announce that state readable. -func ReadsAnnounced(reads []string, announced map[string]announce.States) (granted, refused []string) { - return ReadsAnnouncedBy("", reads, announced) -} - -// ReadsAnnouncedBy is ReadsAnnounced for one reader module, which also honours the readers an owner named for a state -// (announce.State.Readers) and the composer's named readers (NamedReaders, ADR 0547 clause 9): a read named for other -// modules is refused. An empty reader is judged by the readable mark alone. +// ReadsAnnouncedBy splits one reader module's `reads` (`.`) by what each owner announced: granted when the +// owner announced the state readable (by this reader, where it named its readers: announce.State.Readers), when the +// composer names the reader among the state's readers (NamedReaders, ADR 0547 clause 9), or when the owner announced +// nothing yet (the transition); refused otherwise, and a read named for other modules is refused. There is no judgement +// without the reader (issue 774): one made without it refused every named reader, and Users judges again for the +// reader whatever split it is given (withAnnouncedReads). An empty reader is judged by the readable mark alone. func ReadsAnnouncedBy(reader string, reads []string, announced map[string]announce.States) (granted, refused []string) { for _, r := range reads { owner, name, ok := strings.Cut(r, ".") diff --git a/go/buscompose/buscompose_test.go b/go/buscompose/buscompose_test.go index 5eb1bef..3d02084 100644 --- a/go/buscompose/buscompose_test.go +++ b/go/buscompose/buscompose_test.go @@ -552,6 +552,27 @@ func TestTheCopiedComposerComposesAsTheControllers(t *testing.T) { t.Errorf("%s reads the lease: %v, want %v", reader, granted, want) } } + // Issue 774 (sweep item 2): Users judges every read for its reader again, whatever split the caller made: the + // controller's own reader-less copy refused the vault the lease and the grants part's accounts it is named for. + announced["mesh-grants"] = announce.States{Module: "mesh-grants", States: []announce.State{{Name: "accounts", Readers: []string{VaultSeat}}}} + announced["disk-load"] = announce.States{Module: "disk-load", States: []announce.State{{Name: "readings"}}} + judged, err := Users(Records{Nodes: []string{"a"}, Announced: announced, Assigned: map[string][]Declared{"a": { + {Module: VaultSeat, ReadsRefused: []string{ControllerLease, GrantsAccounts}}, + {Module: "mesh-config", ReadsRefused: []string{ControllerLease}}, + {Module: "restic", ReadsRefused: []string{"disk-load.readings"}}, + {Module: "messenger", Reads: []string{ControllerLease, GrantsAccounts}}}}}) + if err != nil { + t.Fatal(err) + } + reads := map[string]string{} + for _, u := range judged { + if u.Kind == KindModule { + reads[u.Module] = strings.Join(u.Reads, ",") + } + } + if reads[VaultSeat] != ControllerLease+","+GrantsAccounts || reads["mesh-config"] != "" || reads["restic"] != "" || reads["messenger"] != "" { + t.Errorf("the reads granted for their readers: %v", reads) + } c, _ := ConsumerFor(Principal{Kind: KindModule, Node: "a", Module: "m", Consumes: []string{"x.happened"}}) cfg := ConsumerConfig(c) if cfg.Durable != "a_m" || cfg.AckPolicy != jetstream.AckExplicitPolicy || cfg.AckWait != 30*time.Second || diff --git a/go/buscompose/userlist_switch.go b/go/buscompose/userlist_switch.go index 9f57881..cbd7f98 100644 --- a/go/buscompose/userlist_switch.go +++ b/go/buscompose/userlist_switch.go @@ -5,6 +5,7 @@ import ( "strings" "time" + "git.novox.be/novox/mesh-sdk/go/announce" "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/handacts" "git.novox.be/novox/mesh-sdk/go/health" @@ -116,6 +117,23 @@ func readsNamed(module string, reads []string) (granted, refused []string) { return granted, refused } +// withAnnouncedReads is d with its reads judged for d's module (ReadsAnnouncedBy) against what the owners announced: +// every read its manifest names, granted or refused by the caller, judged again here, so a caller that split them +// without the reader (the controller's own copy of a reader-less judgement, issue 774: the vault, the assigner, the +// grants part and the artifact store's tools refused the lease they are named for) cannot refuse a read the reader is +// named for, nor grant one named for another. With no announcements read it is withNamedReaders. +func withAnnouncedReads(d Declared, announced map[string]announce.States) Declared { + if announced == nil { + return withNamedReaders(d) + } + all := append(append([]string(nil), d.Reads...), d.ReadsRefused...) + if len(all) == 0 { + return d + } + d.Reads, d.ReadsRefused = ReadsAnnouncedBy(d.Module, all, announced) + return d +} + // withNamedReaders is d with the reads named for other modules moved to its refused reads. func withNamedReaders(d Declared) Declared { granted, refused := readsNamed(d.Module, d.Reads) diff --git a/go/buscompose/users.go b/go/buscompose/users.go index 662d38b..aaf966e 100644 --- a/go/buscompose/users.go +++ b/go/buscompose/users.go @@ -205,9 +205,9 @@ func Users(r Records) ([]Principal, error) { if !Composed(d) { continue } - // A state with named readers is read by them alone (ADR 0547 clause 9): a read named for another module is - // granted nothing, here and in the runtime carrying the module. - d = withReadLogs(withNamedReaders(d), logs) + // A state with named readers is read by them alone (ADR 0547 clause 9), and a read is granted what its owner + // announced (ADR 0495 clause 3): judged here, for this reader, whatever split the caller made (issue 774). + d = withReadLogs(withAnnouncedReads(d, r.Announced), logs) // **A process account is its own, delivered in its own secret** (broker-, novox/hq // ADR 0436 §1, §2), never the module's `broker` — so it is composed whether or not the module // itself has an account, and is never carried by the machine's runtime, which runs as the @@ -259,7 +259,7 @@ func Users(r Records) ([]Principal, error) { var carried []Declared for _, d := range r.Assigned[node] { if CarriedByRunner(d) { - d = withReadLogs(withNamedReaders(d), logs) + d = withReadLogs(withAnnouncedReads(d, r.Announced), logs) // Never a restricted stream for what the tool runner carries (novox/hq ADR 0376 §1). d.Hears = ComposedHears(d) carried = append(carried, d) From 7115b7bf64089da6f53aaa7a317d97baf1cfe54c Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 30 Oct 2026 18:11:19 +0100 Subject: [PATCH 04/11] Refuse a catalogue where a module invokes a seat verb nothing answers (issue 774) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit messenger invoked seat:node-engine.root-free while nothing answered it, and Telegram held every ask that offers an approval, silently. Nothing judged a seat:. invoke beyond its spelling: not that the seat exists, that it promises the verb, nor that anything holds it. Quick fix: messenger asks the controller's root-free (mesh-catalog bc271896) and the engine announces its seat as discovery reads it (mesh-host 7ecf7ba). Thorough fix, chosen because an invoke nothing answers fails only at the call, silently for whoever waits on it: CatalogueProblems refuses every seat verb a module or one of its processes invokes whose seat no module declares and the mesh does not define, whose seat does not promise the verb, or whose seat no module claims (a seat every node's engine holds, ADR 0563 §1, is held). Over some manifests alone a seat whose holder was not given is a note (IsUndeclaredSeat). Over every manifest of mesh-catalog, mesh-control and mesh-delivery today it refuses nothing. --- go/catalogue/seat_invokes.go | 89 +++++++++++++++++++++++++++++ go/catalogue/seats_declared.go | 7 ++- go/catalogue/seats_declared_test.go | 37 ++++++++++-- 3 files changed, 127 insertions(+), 6 deletions(-) create mode 100644 go/catalogue/seat_invokes.go diff --git a/go/catalogue/seat_invokes.go b/go/catalogue/seat_invokes.go new file mode 100644 index 0000000..583ad4e --- /dev/null +++ b/go/catalogue/seat_invokes.go @@ -0,0 +1,89 @@ +package catalogue + +import ( + "fmt" + "slices" + "strings" +) + +// A seat verb a module invokes is one some holder serves (issue 774, sweep item 1). messenger invoked +// `seat:node-engine.root-free`, which nothing in the mesh answered, so every ask that offers an approval was held on +// Telegram, silently. A manifest names what it calls (novox/hq ADR 0152); that a seat by that name exists, promises +// that verb, and has a holder is only known over the whole catalogue, so it is judged here, beside the seats' other +// rules (CatalogueProblems). + +// UnheldSeat ends the problem of a `seat:.` whose seat no module given claims: over the whole catalogue a +// refusal, over some manifests alone a seat whose holder was not given (IsUndeclaredSeat says both). +const UnheldSeat = "which no module claims, so nothing answers it" + +// SeatInvoke splits an invoke `seat:.` into its seat and verb; ok is false for anything else (a module's +// tool, `*`). +func SeatInvoke(invoke string) (seat, verb string, ok bool) { + rest, isSeat := strings.CutPrefix(invoke, "seat:") + if !isSeat { + return "", "", false + } + seat, verb, ok = strings.Cut(rest, ".") + return seat, verb, ok && seat != "" && verb != "" +} + +// seatInvokeProblems are the `seat:.` a module or one of its processes invokes that nothing answers: a seat +// no module declares and the mesh does not define, a verb the seat does not promise, or a seat with no holder (a +// mesh seat no module claims, unless its node's engine holds it, ADR 0563 §1). +func seatInvokeProblems(shelf Shelf, declared map[string]SeatDeclaration) []string { + claimed := map[string]bool{} + for _, m := range shelf { + for _, c := range m.Claims { + claimed[c.Name] = true + if s, ok := SeatNamed(c.Name); ok { + claimed[s.Name] = true + } + } + } + var problems []string + for _, module := range shelfOrder(shelf) { + m := shelf[module] + for _, invoke := range invokesOf(m) { + seat, verb, ok := SeatInvoke(invoke) + if !ok { + continue + } + var verbs []Verb + held := claimed[seat] + if s, isMesh := SeatNamed(seat); isMesh { + verbs, held = s.Serves, held || claimed[s.Name] || s.HeldByTheEngine + } else if d, isModule := declared[seat]; isModule { + verbs = d.Serves + } else { + problems = append(problems, fmt.Sprintf("%s invokes %s, a verb of the seat %q, "+UndeclaredSeat, module, invoke, seat)) + continue + } + if !slices.ContainsFunc(verbs, func(v Verb) bool { return v.Name == verb }) { + problems = append(problems, fmt.Sprintf("%s invokes %s, and the seat %s serves no verb %q", module, invoke, seat, verb)) + continue + } + if !held { + problems = append(problems, fmt.Sprintf("%s invokes %s, a verb of the seat %q, "+UnheldSeat, module, invoke, seat)) + } + } + } + return problems +} + +// invokesOf is everything a module says it invokes: its own and each process's bus, once each. +func invokesOf(m Manifest) []string { + out := append([]string(nil), m.Invokes...) + for _, r := range m.Resources { + if fmt.Sprint(r["type"]) != "process" { + continue + } + if _, present := r["bus"]; !present { + continue + } + if b, _, err := processBusOf(r); err == nil { + out = append(out, b.Invokes...) + } + } + slices.Sort(out) + return slices.Compact(out) +} diff --git a/go/catalogue/seats_declared.go b/go/catalogue/seats_declared.go index b8342e7..94f0ffa 100644 --- a/go/catalogue/seats_declared.go +++ b/go/catalogue/seats_declared.go @@ -272,7 +272,9 @@ type Shelf map[string]Manifest const UndeclaredSeat = "which no module declares and the mesh does not define" // IsUndeclaredSeat says whether a problem CatalogueProblems gave is a seat no manifest given declares. -func IsUndeclaredSeat(problem string) bool { return strings.HasSuffix(problem, UndeclaredSeat) } +func IsUndeclaredSeat(problem string) bool { + return strings.HasSuffix(problem, UndeclaredSeat) || strings.HasSuffix(problem, UnheldSeat) +} func CatalogueProblems(shelf Shelf) []string { var problems []string @@ -349,6 +351,9 @@ func CatalogueProblems(shelf Shelf) []string { } } + // A seat verb a module invokes is one some holder serves (issue 774). + problems = append(problems, seatInvokeProblems(shelf, declared)...) + // Which seats name each path (novox/hq ADR 0287): a module's seat naming one another seat names loses. named, pathProblems := pathSeats(shelfOrder(shelf), shelf, declaredBy) problems = append(problems, pathProblems...) diff --git a/go/catalogue/seats_declared_test.go b/go/catalogue/seats_declared_test.go index 25a8d0b..fdef894 100644 --- a/go/catalogue/seats_declared_test.go +++ b/go/catalogue/seats_declared_test.go @@ -51,12 +51,39 @@ func TestTwoModulesCannotDeclareTheSameSeat(t *testing.T) { } // Where ADR 0110's guarantee lands under a derived set: a typo is refused, not resolved to -// nothing at runtime. +// nothing at runtime. And a seat verb a module invokes is one a holder serves (issue 774, sweep item 1: messenger +// invoked a verb nothing answered, and Telegram held every ask silently). func TestUsingASeatNobodyDeclaresIsRefused(t *testing.T) { - shop := Manifest{Module: "shop", Uses: []string{"telegram-sendr"}} - got := problemsFor(t, Shelf{"telegram": telegram(), "shop": shop}) - if !strings.Contains(got, "telegram-sendr") || !strings.Contains(got, "no module declares") { - t.Fatalf("a misspelled seat was accepted: %s", got) + unclaimed := telegram() + unclaimed.Claims = nil + for _, row := range []struct { + name, want string + shop Manifest + seat Manifest + note bool // over some manifests alone it is a note (IsUndeclaredSeat) + }{ + {"a misspelled seat used", `uses the seat "telegram-sendr", ` + UndeclaredSeat, + Manifest{Module: "shop", Uses: []string{"telegram-sendr"}}, telegram(), true}, + {"a declared verb invoked of a held seat", "", Manifest{Module: "shop", Invokes: []string{"seat:telegram-sender.status"}}, telegram(), false}, + {"a verb the seat does not promise", `the seat telegram-sender serves no verb "resend"`, + Manifest{Module: "shop", Invokes: []string{"seat:telegram-sender.resend"}}, telegram(), false}, + {"a verb of a misspelled seat", `a verb of the seat "telegram-sendr", ` + UndeclaredSeat, + Manifest{Module: "shop", Invokes: []string{"seat:telegram-sendr.status"}}, telegram(), true}, + {"a verb of a declared seat nobody claims", `a verb of the seat "telegram-sender", ` + UnheldSeat, + Manifest{Module: "shop", Invokes: []string{"seat:telegram-sender.status"}}, unclaimed, true}, + {"a verb of a mesh seat nobody claims", `a verb of the seat "mesh-controller", ` + UnheldSeat, + Manifest{Module: "shop", Invokes: []string{"seat:mesh-controller.root-free"}}, telegram(), true}, + {"the engine's verb: every node's engine holds its seat", "", + Manifest{Module: "shop", Invokes: []string{"seat:node-engine.root-free"}}, telegram(), false}, + {"a verb the engine's seat does not promise", `the seat node-engine serves no verb "root-freed"`, + Manifest{Module: "shop", Invokes: []string{"seat:node-engine.root-freed"}}, telegram(), false}, + } { + got := CatalogueProblems(Shelf{"telegram": row.seat, "shop": row.shop}) + joined := strings.Join(got, "; ") + if row.want == "" && joined != "" || row.want != "" && (len(got) != 1 || !strings.Contains(joined, row.want) || + IsUndeclaredSeat(got[0]) != row.note) { + t.Errorf("%s: %s", row.name, joined) + } } } From 1dc5d8cb3d652821244af54264c3bf4404f1efdc Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 30 Oct 2026 18:15:46 +0100 Subject: [PATCH 05/11] Run a watch's handler beside the runtime, and give a refused watch up loudly (issue 774) The provisioner's watches of gitea.bound, mesh-vault.bound and postgres.bound failed every minute for an hour on the control-node ("the current values ... did not arrive in 10s") and 1257 changes were refused as "names no watch of this bundle". Both sides were at fault. The SDK answered a change only once its handler returned, so a slow handler (Rejudge waits on the lock its composition holds) counted against the runtime's 10 s bound on the current values; and when the runtime gave up, the SDK forgot the id at once. The runtime (mesh-tools node-tools internal/bus/state.go StateWatch) stops its watcher on that bound but still drains and delivers the values already buffered, to an id nobody holds. Quick fix: a longer runtime bound. Thorough fix here: each state and provision watch has a follower that takes a change in order and answers at once, its handler on its own goroutine; Watch still returns only once the current values were handled, and a handler's error is said on standard error. A watch the runtime refuses is said given up and kept by id, so a change still handed to it is refused saying so. KeptState.Follow asks again with backoff until taken, the loop four modules wrote for themselves. A log watch keeps its answer after the handler: that is its back-pressure. --- go/follower.go | 135 +++++++++++++++++++++++++++++++++++++++++++++++ go/provision.go | 7 ++- go/state.go | 12 ++++- go/state_test.go | 58 ++++++++++++++++++++ go/stdio.go | 49 ++++++++--------- 5 files changed, 230 insertions(+), 31 deletions(-) create mode 100644 go/follower.go diff --git a/go/follower.go b/go/follower.go new file mode 100644 index 0000000..17a9407 --- /dev/null +++ b/go/follower.go @@ -0,0 +1,135 @@ +package stdio + +import ( + "fmt" + "os" + "sync" + "time" +) + +// A watch's handler runs beside the runtime, never in its way (issue 774, the sweep of 2026-10-30). +// +// The runtime hands a watch's current values over one at a time and gives up on the watch when none was handed over +// within its bound (mesh-tools node-tools internal/bus/state.go StateWatch, 10 s). The bundle answered each change only +// once its handler returned, so a slow handler (mesh-provisioner's Rejudge waits on the lock its composition holds) +// counted against the bus's bound: the watch was given up, the SDK forgot its id, and the values the runtime still +// delivered to it were refused as "names no watch of this bundle", 1257 times in an hour on the control-node. +// +// So each watch has a follower: a change is taken into it in the order the runtime hands it over and answered at once, +// and the handler runs on the follower's own goroutine, in that order. Watch still returns only once every current +// value was handled. A handler's error is said on the bundle's standard error, naming the watch and the change, since +// the runtime no longer hears it. A log watch keeps its answer after the handler: its runtime sends the next entry only +// after the answer, which is the back-pressure a replay of a long log needs. + +// follower runs one watch's handler on its own goroutine, in order. +type follower[T any] struct { + what string + handle func(T) error + describe func(T) string + mu sync.Mutex + items []followed[T] + wake chan struct{} +} + +// followed is one change, or with no change a mark that closes done once everything before it was handled. +type followed[T any] struct { + change T + mark bool + done chan struct{} +} + +func newFollower[T any](what string, handle func(T) error, describe func(T) string) *follower[T] { + f := &follower[T]{what: what, handle: handle, describe: describe, wake: make(chan struct{}, 1)} + go f.run() + return f +} + +// take queues a change; it never waits on the handler. +func (f *follower[T]) take(change T) { + f.mu.Lock() + f.items = append(f.items, followed[T]{change: change}) + f.mu.Unlock() + f.poke() +} + +// handled is closed once every change taken before it was handled. +func (f *follower[T]) handled() <-chan struct{} { + done := make(chan struct{}) + f.mu.Lock() + f.items = append(f.items, followed[T]{mark: true, done: done}) + f.mu.Unlock() + f.poke() + return done +} + +func (f *follower[T]) poke() { + select { + case f.wake <- struct{}{}: + default: + } +} + +func (f *follower[T]) run() { + for range f.wake { + for { + f.mu.Lock() + if len(f.items) == 0 { + f.mu.Unlock() + break + } + next := f.items[0] + f.items = f.items[1:] + f.mu.Unlock() + if next.mark { + close(next.done) + continue + } + if err := f.handle(next.change); err != nil { + sayf("%s: the handler did not take %s: %v; the next change hands it again", f.what, f.describe(next.change), err) + } + } + } +} + +// sayf says something the runtime cannot say for the bundle, on its standard error, which its journal keeps. +func sayf(format string, args ...any) { + fmt.Fprintf(os.Stderr, "[mesh-sdk] "+format+"\n", args...) +} + +// gaveUpMost bounds what a bundle remembers of the watches it gave up. +const gaveUpMost = 256 + +// gaveUp keeps why a watch was given up, so a change the runtime still hands to it is refused saying so, never as a +// watch nobody asked for. Called with c.mu held. +func (c *channel) gaveUp(id, why string) { + if c.givenUp == nil || len(c.givenUp) >= gaveUpMost { + c.givenUp = map[string]string{} + } + c.givenUp[id] = why +} + +// noWatch is the refusal of a change for watch id, which this bundle never asked or gave up. +func (c *channel) noWatch(kind, id string) string { + c.mu.Lock() + why, given := c.givenUp[id] + c.mu.Unlock() + if given { + sayf("%s: the runtime still hands over changes for the watch %s, which this bundle gave up: %s", kind, id, why) + return fmt.Sprintf("%s: this bundle gave the watch %s up (%s); stop handing it over", kind, id, why) + } + return kind + " names no watch of this bundle" +} + +// Follow watches key of the state as Watch does and asks again until the watch is taken, saying each give-up and +// when it asks again (2 s, doubled to a minute): the loop every module wrote for itself, in one place. It returns once +// the watch is taken; call it beside the code that needs it. +func (s KeptState) Follow(key string, handle func(StateChange) error) { + for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { + err := s.Watch(key, handle) + if err == nil { + return + } + sayf("the state %s is not watched yet (%v); asking again in %s", s.name, err, wait) + time.Sleep(wait) + } +} diff --git a/go/provision.go b/go/provision.go index e4d65c7..be289e2 100644 --- a/go/provision.go +++ b/go/provision.go @@ -3,6 +3,7 @@ package stdio import ( "encoding/json" "errors" + "fmt" "time" ) @@ -42,15 +43,19 @@ func provisionWatch(kind, key string, handle func(ProvisionChange) error) error return err } id := c.newID() + f := newFollower("the watch "+id+" of the "+kind+" "+key, handle, func(ch ProvisionChange) string { return ch.Kind + " " + ch.Key }) c.mu.Lock() - c.provisioned[id] = handle + c.provisioned[id] = f c.mu.Unlock() if _, err := c.askAs(id, "mesh/provision.watch", map[string]any{"kind": kind, "key": key}, provisionTimeout); err != nil { c.mu.Lock() delete(c.provisioned, id) + c.gaveUp(id, fmt.Sprintf("the %s %s at %s: %v", kind, key, time.Now().Format(time.RFC3339), err)) c.mu.Unlock() + sayf("the watch %s of the %s %s is given up: %v", id, kind, key, err) return err } + <-f.handled() return nil } diff --git a/go/state.go b/go/state.go index ad49de2..a085570 100644 --- a/go/state.go +++ b/go/state.go @@ -213,21 +213,29 @@ func (s KeptState) Keys() ([]string, error) { // Watch hands handle every current value — none that is deleted — and then every change, in order, // and returns once the current values have all been handled. key narrows it: a key, with `*` for one -// name and `**` for the rest; empty is every key. The watch lasts as long as the process. +// name and `**` for the rest; empty is every key. The watch lasts as long as the process. handle runs beside the +// runtime (follower.go): its time never counts against the runtime's bound on the current values, and an error it +// returns is said on standard error. A watch the runtime refuses is given up, said, and a change the runtime still +// hands to it is refused saying so; Follow asks again until it is taken. func (s KeptState) Watch(key string, handle func(StateChange) error) error { c, err := live() if err != nil { return err } id := c.newID() + f := newFollower("the watch "+id+" of the state "+s.name, handle, func(ch StateChange) string { return ch.Op + " " + ch.Key }) c.mu.Lock() - c.watches[id] = handle + c.watches[id] = f c.mu.Unlock() if _, err := c.askAs(id, "mesh/state.watch", map[string]any{"state": s.name, "key": key}, stateTimeout); err != nil { c.mu.Lock() delete(c.watches, id) + c.gaveUp(id, fmt.Sprintf("the state %s at %s: %v", s.name, time.Now().Format(time.RFC3339), err)) c.mu.Unlock() + sayf("the watch %s of the state %s is given up: %v", id, s.name, err) return err } + // The runtime hands every current value over before it answers the watch: each is in the follower by now. + <-f.handled() return nil } diff --git a/go/state_test.go b/go/state_test.go index dc80765..cac843b 100644 --- a/go/state_test.go +++ b/go/state_test.go @@ -20,6 +20,7 @@ func TestABundleKeepsAndWatchesStateThroughTheRuntime(t *testing.T) { outR, outW := io.Pipe() seen := make(chan string, 16) watched := make(chan error, 1) + slowWatched, release := make(chan error, 1), make(chan struct{}) go func() { _ = ServeOn("", []Tool{ {Name: "start", Run: func(map[string]any) (any, error) { @@ -31,6 +32,19 @@ func TestABundleKeepsAndWatchesStateThroughTheRuntime(t *testing.T) { }() return "started", nil }}, + {Name: "slow", Run: func(map[string]any) (any, error) { + go func() { + slowWatched <- State("slow").Watch("", func(c StateChange) error { + <-release + seen <- "slow " + c.Key + return errors.New("the handler's own error") + }) + }() + return "started", nil + }}, + {Name: "refused", Run: func(map[string]any) (any, error) { + return nil, State("gone").Watch("", func(StateChange) error { return nil }) + }}, {Name: "remember", Run: func(a map[string]any) (any, error) { return State("servers").Put(a["key"].(string), a["value"]) }}, @@ -111,6 +125,50 @@ func TestABundleKeepsAndWatchesStateThroughTheRuntime(t *testing.T) { t.Fatalf("a change for no watch was answered %v", r) } + // Issue 774 (the provisioner's "names no watch of this bundle"): a change is answered as soon as it is taken, so a + // slow handler never counts against the runtime's bound on the current values; Watch still returns only once the + // current values were handled, and a handler's error stays the bundle's to say. + send(map[string]any{"jsonrpc": "2.0", "id": 60, "method": "tools/call", "params": map[string]any{"name": "slow"}}) + var slowAsk map[string]any + for i := 0; i < 2; i++ { + if m := next(); m["method"] == "mesh/state.watch" { + slowAsk = m + } + } + for i, key := range []string{"a", "b"} { + send(map[string]any{"jsonrpc": "2.0", "id": 61 + i, "method": "mesh/state", "params": map[string]any{"watch": slowAsk["id"], + "change": map[string]any{"state": "slow", "key": key, "op": "put", "value": 1, "current": true}}}) + if r := next(); r["id"] != float64(61+i) || r["error"] != nil { + t.Fatalf("a change to a slow handler was answered %v", r) + } + } + send(map[string]any{"jsonrpc": "2.0", "id": slowAsk["id"], "result": map[string]any{}}) + select { + case err := <-slowWatched: + t.Fatalf("the watch returned before its current values were handled: %v", err) + case <-time.After(50 * time.Millisecond): + } + close(release) + if a, b := <-seen, <-seen; a != "slow a" || b != "slow b" { + t.Fatalf("handled out of order: %q, %q", a, b) + } + if err := <-slowWatched; err != nil { + t.Fatal(err) + } + // A watch the runtime refuses is given up, and a change the runtime still hands to it is refused saying so. + send(map[string]any{"jsonrpc": "2.0", "id": 70, "method": "tools/call", "params": map[string]any{"name": "refused"}}) + goneAsk := next() + send(map[string]any{"jsonrpc": "2.0", "id": goneAsk["id"], "error": map[string]any{"code": -32000, + "message": "the current values of gone did not arrive in 10s"}}) + if r := next(); r["id"] != float64(70) { + t.Fatalf("the refused watch answered %v", r) + } + send(map[string]any{"jsonrpc": "2.0", "id": 71, "method": "mesh/state", "params": map[string]any{"watch": goneAsk["id"], + "change": map[string]any{"state": "gone", "key": "k", "op": "put"}}}) + if r := next(); r["id"] != float64(71) || !strings.Contains(fmt.Sprint(r), "gave the watch") || !strings.Contains(fmt.Sprint(r), "did not arrive in 10s") { + t.Fatalf("a change for a given-up watch was answered %v", r) + } + // Put: asked of the runtime as the module, the revision is the tool's answer. send(map[string]any{"jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": map[string]any{"name": "remember", "arguments": map[string]any{"key": "all.two", "value": map[string]any{"url": "http://two"}}}}) diff --git a/go/stdio.go b/go/stdio.go index 82568f6..71ab95b 100644 --- a/go/stdio.go +++ b/go/stdio.go @@ -69,9 +69,11 @@ type channel struct { // subscribed numbers the listeners, so one whose subscribe failed is the one removed. subscribed int // watches are this bundle's state watches by the id each was asked with (ADR 0201). - watches map[string]func(StateChange) error + watches map[string]*follower[StateChange] // provisioned are this bundle's watches of its values, by the id each was asked with (provision.go). - provisioned map[string]func(ProvisionChange) error + provisioned map[string]*follower[ProvisionChange] + // givenUp are the watches this bundle gave up, by id, each with why (follower.go). + givenUp map[string]string // logWatches are this bundle's watches of a log, by the id each was asked with (log.go). logWatches map[string]func(LogEntry) error // takers and provers are this bundle's seat work and proof handlers, by worker and subject (ADR 0259). @@ -183,8 +185,8 @@ func ServeOn(name string, tools []Tool, in io.Reader, out io.Writer) error { for _, t := range tools { byName[t.Name] = t } - c := &channel{enc: json.NewEncoder(out), waiting: map[string]chan message{}, watches: map[string]func(StateChange) error{}, - provisioned: map[string]func(ProvisionChange) error{}, logWatches: map[string]func(LogEntry) error{}, + c := &channel{enc: json.NewEncoder(out), waiting: map[string]chan message{}, watches: map[string]*follower[StateChange]{}, + provisioned: map[string]*follower[ProvisionChange]{}, logWatches: map[string]func(LogEntry) error{}, takers: map[string]func(Work) error{}, provers: map[string]func(string, json.RawMessage) (any, error){}} currentMu.Lock() current = c @@ -296,27 +298,23 @@ func ServeOn(name string, tools []Tool, in io.Reader, out io.Writer) error { c.say(map[string]any{"id": id, "result": map[string]any{}}) }(id, *p.Envelope, matching) case "mesh/state": - // One change to a state this bundle watches (ADR 0201), named by the watch's id, answered - // once its handler has taken it. Beside the loop, like a tool call. + // One change to a state this bundle watches (ADR 0201), named by the watch's id: taken into the watch's + // follower in the order it came and answered at once, so a slow handler never counts against the runtime's + // bound on the current values (follower.go, issue 774). var p struct { Watch string `json:"watch"` Change *StateChange `json:"change"` } _ = json.Unmarshal(m.Params, &p) c.mu.Lock() - handle := c.watches[p.Watch] + f := c.watches[p.Watch] c.mu.Unlock() - if handle == nil || p.Change == nil { - c.say(map[string]any{"id": id, "error": map[string]any{"code": -32602, "message": "mesh/state names no watch of this bundle"}}) + if f == nil || p.Change == nil { + c.say(map[string]any{"id": id, "error": map[string]any{"code": -32602, "message": c.noWatch("mesh/state", p.Watch)}}) continue } - go func(id any, change StateChange) { - if err := handle(change); err != nil { - c.say(map[string]any{"id": id, "error": map[string]any{"code": -32000, "message": err.Error()}}) - return - } - c.say(map[string]any{"id": id, "result": map[string]any{}}) - }(id, *p.Change) + f.take(*p.Change) + go c.say(map[string]any{"id": id, "result": map[string]any{}}) case "mesh/log": // One entry of a log this bundle watches (log.go), named by the watch's id, answered once its handler has // taken it, in order: the runtime sends the next only after this answer. @@ -340,27 +338,22 @@ func ServeOn(name string, tools []Tool, in io.Reader, out io.Writer) error { c.say(map[string]any{"id": id, "result": map[string]any{}}) }(id, *p.Entry) case "mesh/provision": - // One change to a value of this module's that this bundle watches (provision.go), named by the - // watch's id, answered once its handler has taken it. Beside the loop, like a tool call. + // One change to a value of this module's that this bundle watches (provision.go), named by the watch's id: + // taken into the watch's follower and answered at once, as a state's change is (follower.go, issue 774). var p struct { Watch string `json:"watch"` Change *ProvisionChange `json:"change"` } _ = json.Unmarshal(m.Params, &p) c.mu.Lock() - handle := c.provisioned[p.Watch] + f := c.provisioned[p.Watch] c.mu.Unlock() - if handle == nil || p.Change == nil { - c.say(map[string]any{"id": id, "error": map[string]any{"code": -32602, "message": "mesh/provision names no watch of this bundle"}}) + if f == nil || p.Change == nil { + c.say(map[string]any{"id": id, "error": map[string]any{"code": -32602, "message": c.noWatch("mesh/provision", p.Watch)}}) continue } - go func(id any, change ProvisionChange) { - if err := handle(change); err != nil { - c.say(map[string]any{"id": id, "error": map[string]any{"code": -32000, "message": err.Error()}}) - return - } - c.say(map[string]any{"id": id, "result": map[string]any{}}) - }(id, *p.Change) + f.take(*p.Change) + go c.say(map[string]any{"id": id, "result": map[string]any{}}) case "mesh/work": // One piece of a seat's work this bundle takes (novox/hq ADR 0259): taken by an answer that is // not an error, offered again after one. Beside the loop, like a tool call. From 44dbb22ea76f231befaafbf5e38397bf43f7f683 Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 30 Oct 2026 18:18:32 +0100 Subject: [PATCH 06/11] Wait out a bus that answers nothing for a few seconds instead of losing the write (issue 774) At each mesh-grants restart and user-list reload (16:37:24, 16:55:15) nats-server 2.11's reloadAuthorization left the mesh's account without its imports, $JS.API.> among them, for a few seconds: requests had no responders, JetStream calls said it was not enabled, and a build-test result on ace was lost. Quick fix, chosen as asked; the thorough fix is a reload that keeps the account's imports, in the bus's holder. The package busretry asks a request again, backing off within about 30 s, only while its error says it reached no one (no responders, JetStream not enabled, no response from the stream, or either in words): nothing was done, so even a create or an update is safe to ask again. A timeout or any answer is never asked again. After the bound it is said on standard error and returned. Used by process.State's requests, process.Conn's calls (asked again only while $JS.API.INFO says the bus itself answers nothing, so a callee that is not running still fails at once) and asks, and the bundle's state requests through the runtime. --- go/busretry/busretry.go | 69 ++++++++++++++++++++++++++++++++++++ go/busretry/busretry_test.go | 48 +++++++++++++++++++++++++ go/process/asks.go | 12 +++++-- go/process/call.go | 28 ++++++++++++--- go/process/state.go | 42 ++++++++++++++++++++++ go/process/state_test.go | 14 ++++++++ go/state.go | 6 +++- 7 files changed, 211 insertions(+), 8 deletions(-) create mode 100644 go/busretry/busretry.go create mode 100644 go/busretry/busretry_test.go diff --git a/go/busretry/busretry.go b/go/busretry/busretry.go new file mode 100644 index 0000000..a8adbff --- /dev/null +++ b/go/busretry/busretry.go @@ -0,0 +1,69 @@ +// Package busretry waits out a bus that answers nothing for a few seconds (issue 774, the sweep of 2026-10-30: bus +// blackouts at every user-list reload). nats-server 2.11's reloadAuthorization rebuilds the accounts, and for a few +// seconds the mesh's account has none of its imports, $JS.API.> among them: every request is answered "no responders" +// and every JetStream call "jetstream not enabled". A write or a call made in that window failed, and was lost. +// +// Only an error that says the request reached no one is waited out (Undelivered): nothing was done, so asking again +// is safe even for a write that is not idempotent. A timeout, or any answer, is never asked again: the request may +// have been delivered. The wait is bounded (Bound, about 30 s, backing off), and said on standard error when it ends +// without an answer. +// +// This is the quick fix. The thorough one is a reload that keeps the account's imports, in the bus's holder. +package busretry + +import ( + "errors" + "fmt" + "os" + "strings" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" +) + +// Bound is how long a request that reached no one is asked again; FirstWait the first pause, doubled to MostWait. +var ( + Bound = 30 * time.Second + FirstWait = 250 * time.Millisecond + MostWait = 4 * time.Second +) + +// Undelivered says err means the request reached no one: no responders, JetStream not enabled (for the account), or +// no response from the stream after the client's own retries on no responders; also said in words by a runtime that +// passed the error on. +func Undelivered(err error) bool { + if err == nil { + return false + } + if errors.Is(err, nats.ErrNoResponders) || errors.Is(err, jetstream.ErrJetStreamNotEnabled) || + errors.Is(err, jetstream.ErrJetStreamNotEnabledForAccount) || errors.Is(err, jetstream.ErrNoStreamResponse) || + errors.Is(err, nats.ErrJetStreamNotEnabled) || errors.Is(err, nats.ErrJetStreamNotEnabledForAccount) { + return true + } + words := strings.ToLower(err.Error()) + return strings.Contains(words, "no responders") || strings.Contains(words, "jetstream not enabled") +} + +// Do runs try, and again while its error says the request reached no one, until Bound has passed. blackout, when +// given, is asked after each such error and Do stops waiting when it says the bus is not blacked out: a call nobody +// answers because its callee is not running fails at once, as before. what names the request in the error. +func Do[T any](what string, try func() (T, error), blackout func() bool) (T, error) { + v, err := try() + if !Undelivered(err) || (blackout != nil && !blackout()) { + return v, err + } + start, wait := time.Now(), FirstWait + for time.Since(start) < Bound { + time.Sleep(wait) + wait = min(wait*2, MostWait) + v, err = try() + if !Undelivered(err) || (blackout != nil && !blackout()) { + return v, err + } + } + err = fmt.Errorf("%s: the bus answered nothing for %s, asked again until then (a user-list reload drops the "+ + "account's imports for a few seconds; this was longer): %w", what, time.Since(start).Round(time.Second), err) + fmt.Fprintf(os.Stderr, "[mesh-sdk] %v\n", err) + return v, err +} diff --git a/go/busretry/busretry_test.go b/go/busretry/busretry_test.go new file mode 100644 index 0000000..d6af096 --- /dev/null +++ b/go/busretry/busretry_test.go @@ -0,0 +1,48 @@ +package busretry + +import ( + "context" + "errors" + "fmt" + "strings" + "testing" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" +) + +// Issue 774 (bus blackouts at every user-list reload): only a request that reached no one is asked again, within the +// bound; a blackout probe that says the bus answers ends the wait at once. +func TestABusThatAnswersNothingIsWaitedOut(t *testing.T) { + Bound, FirstWait, MostWait = 60*time.Millisecond, time.Millisecond, 4*time.Millisecond + for _, row := range []struct { + name string + errs []error // what each try answers; the last repeats + blackout func() bool + tries int // at least + atMost int + delivered bool + }{ + {"no responders, then an answer", []error{nats.ErrNoResponders, nats.ErrNoResponders, nil}, nil, 3, 3, true}, + {"jetstream not enabled, then an answer", []error{jetstream.ErrJetStreamNotEnabled, nil}, nil, 2, 2, true}, + {"said in words by a runtime", []error{errors.New("nats: no responders available for request"), nil}, nil, 2, 2, true}, + {"a timeout may have been delivered: never asked again", []error{context.DeadlineExceeded}, nil, 1, 1, false}, + {"an answer that refuses is an answer", []error{errors.New("refused")}, nil, 1, 1, false}, + {"nobody answers on a bus that answers: the callee is not running", []error{nats.ErrNoResponders}, func() bool { return false }, 1, 1, false}, + {"blacked out past the bound: said", []error{jetstream.ErrJetStreamNotEnabledForAccount}, func() bool { return true }, 3, 1000, false}, + } { + tries := 0 + _, err := Do("the test", func() (int, error) { + e := row.errs[min(tries, len(row.errs)-1)] + tries++ + return tries, e + }, row.blackout) + if tries < row.tries || tries > row.atMost || (err == nil) != row.delivered { + t.Errorf("%s: %d tries, %v", row.name, tries, err) + } + if strings.HasPrefix(row.name, "blacked out") && (!Undelivered(err) || !strings.Contains(fmt.Sprint(err), "the bus answered nothing")) { + t.Errorf("%s: %v", row.name, err) + } + } +} diff --git a/go/process/asks.go b/go/process/asks.go index baa431a..460ddbf 100644 --- a/go/process/asks.go +++ b/go/process/asks.go @@ -6,7 +6,10 @@ import ( "fmt" "strings" + "github.com/nats-io/nats.go" + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/busretry" ) // Ask reads the router's record of one of this process's module's own asks, under its own module and no @@ -24,9 +27,12 @@ func (c *Conn) Ask(bucket, id string) (*stdio.StateEntry, error) { return nil, err } subject := "$JS.API.DIRECT.GET.KV_" + bucket + ".$KV." + bucket + "." + c.account.Module + "." + id - ctx, cancel := context.WithTimeout(context.Background(), stateTimeout) - defer cancel() - reply, err := c.nc.RequestWithContext(ctx, subject, nil) + // A read: asked again while the bus answers nothing (busretry). + reply, err := busretry.Do("reading "+subject, func() (*nats.Msg, error) { + ctx, cancel := context.WithTimeout(context.Background(), stateTimeout) + defer cancel() + return c.nc.RequestWithContext(ctx, subject, nil) + }, nil) if err != nil { return nil, fmt.Errorf("reading %s's ask %s from %s: %w", c.account.Module, id, bucket, err) } diff --git a/go/process/call.go b/go/process/call.go index a9f609c..7bf1252 100644 --- a/go/process/call.go +++ b/go/process/call.go @@ -9,6 +9,8 @@ import ( "time" "github.com/nats-io/nats.go" + + "git.novox.be/novox/mesh-sdk/go/busretry" ) // CallTimeout bounds a call to a tool or a seat verb (mesh-tools node-tools/internal/bus/bus.go:35, @@ -47,16 +49,34 @@ func (c *Conn) CallOn(node, key string, body any) (json.RawMessage, error) { return c.callOn(called, fallback, body) } +// callOn asks called, then fallback when nothing answers it; while the bus itself answers nothing (a user-list reload +// drops the account's imports for a few seconds, busretry) both are asked again, within busretry.Bound. A call nobody +// answers on a bus that answers is not asked again: its callee is not running. A call that may have been delivered (a +// timeout, any answer) is never asked again. func (c *Conn) callOn(called, fallback string, body any) (json.RawMessage, error) { data, err := json.Marshal(body) if err != nil { return nil, err } - result, err := c.request(called, data) - if errors.Is(err, nats.ErrNoResponders) { - result, err = c.request(fallback, data) + return busretry.Do("the call on "+fallback, func() (json.RawMessage, error) { + result, err := c.request(called, data) + if errors.Is(err, nats.ErrNoResponders) { + result, err = c.request(fallback, data) + } + return result, err + }, c.blackedOut) +} + +// blackedOut says the bus answers this account nothing at all: its JetStream API, which every account imports, has no +// responder or says it is not enabled. +func (c *Conn) blackedOut() bool { + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + msg, err := c.nc.RequestWithContext(ctx, "$JS.API.INFO", nil) + if err != nil { + return busretry.Undelivered(err) } - return result, err + return strings.Contains(string(msg.Data), `"err_code":10039`) || strings.Contains(string(msg.Data), "jetstream not enabled") } // callerToken is this account's user name as the last token of a call it publishes, each dot written ~ diff --git a/go/process/state.go b/go/process/state.go index d2a462f..508ff8a 100644 --- a/go/process/state.go +++ b/go/process/state.go @@ -13,6 +13,7 @@ import ( "github.com/nats-io/nats.go/jetstream" stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/busretry" "git.novox.be/novox/mesh-sdk/go/kvlimit" ) @@ -74,7 +75,12 @@ func (s State) store(ctx context.Context) (jetstream.KeyValue, error) { } // Get is a key's current value, or nil when it has none. +// Get waits out a bus that answers nothing for a few seconds (busretry): only a request that reached no one is asked again. func (s State) Get(key string) (*stdio.StateEntry, error) { + return busretry.Do(s.what("get "+key), func() (*stdio.StateEntry, error) { return s.get(key) }, nil) +} + +func (s State) get(key string) (*stdio.StateEntry, error) { ctx, cancel := context.WithTimeout(context.Background(), stateTimeout) defer cancel() kv, err := s.store(ctx) @@ -93,7 +99,12 @@ func (s State) Get(key string) (*stdio.StateEntry, error) { // Create writes a key only when it has none, and returns the revision; a key that has one is refused as // stdio.ErrChanged. +// Create waits out a bus that answers nothing for a few seconds (busretry): only a request that reached no one is asked again. func (s State) Create(key string, value any) (uint64, error) { + return busretry.Do(s.what("create "+key), func() (uint64, error) { return s.create(key, value) }, nil) +} + +func (s State) create(key string, value any) (uint64, error) { ctx, cancel := context.WithTimeout(context.Background(), stateTimeout) defer cancel() kv, err := s.store(ctx) @@ -121,7 +132,12 @@ const MinTTL = time.Second // 1952 S6): the bus removes it when ttl passes and its watchers hear the removal, so an expiry is an event and no timer // runs. A key that has a value is refused as stdio.ErrChanged: an act begun once cannot be begun again. The bucket must // take keys with their own time to live (a state declared `key-ttl`). +// CreateTTL waits out a bus that answers nothing for a few seconds (busretry): only a request that reached no one is asked again. func (s State) CreateTTL(key string, value any, ttl time.Duration) (uint64, error) { + return busretry.Do(s.what("create "+key), func() (uint64, error) { return s.createTTL(key, value, ttl) }, nil) +} + +func (s State) createTTL(key string, value any, ttl time.Duration) (uint64, error) { if ttl < MinTTL { return 0, fmt.Errorf("%s's state %s, key %s: a time to live of %s is shorter than the bus keeps (%s)", s.Module, s.Name, key, ttl, MinTTL) } @@ -161,7 +177,12 @@ func (s State) RenewTTL(key string, value any, ttl time.Duration) (uint64, error // Update writes a key only when it is still at revision, and returns the new one; a key changed meanwhile // is refused as stdio.ErrChanged. +// Update waits out a bus that answers nothing for a few seconds (busretry): only a request that reached no one is asked again. func (s State) Update(key string, value any, revision uint64) (uint64, error) { + return busretry.Do(s.what("update "+key), func() (uint64, error) { return s.update(key, value, revision) }, nil) +} + +func (s State) update(key string, value any, revision uint64) (uint64, error) { ctx, cancel := context.WithTimeout(context.Background(), stateTimeout) defer cancel() kv, err := s.store(ctx) @@ -183,7 +204,12 @@ func (s State) Update(key string, value any, revision uint64) (uint64, error) { } // Put writes a key whole, whatever it held, and returns the revision written. +// Put waits out a bus that answers nothing for a few seconds (busretry): only a request that reached no one is asked again. func (s State) Put(key string, value any) (uint64, error) { + return busretry.Do(s.what("put "+key), func() (uint64, error) { return s.put(key, value) }, nil) +} + +func (s State) put(key string, value any) (uint64, error) { ctx, cancel := context.WithTimeout(context.Background(), stateTimeout) defer cancel() kv, err := s.store(ctx) @@ -202,7 +228,12 @@ func (s State) Put(key string, value any) (uint64, error) { } // Keys is every key with a current value, sorted; none is an empty list. +// Keys waits out a bus that answers nothing for a few seconds (busretry): only a request that reached no one is asked again. func (s State) Keys() ([]string, error) { + return busretry.Do(s.what("keys"), func() ([]string, error) { return s.keys() }, nil) +} + +func (s State) keys() ([]string, error) { ctx, cancel := context.WithTimeout(context.Background(), stateTimeout) defer cancel() kv, err := s.store(ctx) @@ -314,7 +345,18 @@ func (s State) Watch(key string, handle func(stdio.StateChange) error) error { } // Delete removes a key. A key that was not there is not an error. +// Delete waits out a bus that answers nothing for a few seconds, as Put does. func (s State) Delete(key string) error { + _, err := busretry.Do(s.what("delete "+key), func() (struct{}, error) { return struct{}{}, s.delete(key) }, nil) + return err +} + +// what names a request on this state for busretry. +func (s State) what(request string) string { + return fmt.Sprintf("%s's state %s, %s", s.Module, s.Name, request) +} + +func (s State) delete(key string) error { ctx, cancel := context.WithTimeout(context.Background(), stateTimeout) defer cancel() kv, err := s.store(ctx) diff --git a/go/process/state_test.go b/go/process/state_test.go index 7cc0a93..0809a29 100644 --- a/go/process/state_test.go +++ b/go/process/state_test.go @@ -13,6 +13,7 @@ import ( stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/asker" + "git.novox.be/novox/mesh-sdk/go/busretry" ) // fakeEntry is a fakeKV's KeyValueEntry: only Key, Value and Revision are read by State. @@ -206,6 +207,19 @@ func TestStatePutWritesAKeyWhole(t *testing.T) { if err != nil || e == nil || e.Revision != rev2 || string(e.Value) != `{"a":2}` { t.Fatalf("Get after Put: %+v, %v", e, err) } + // Issue 774: a bus that answers nothing for a moment (a user-list reload) is waited out, not a lost write. + busretry.FirstWait = time.Millisecond + kv, blackedOut := newFakeKV(), 2 + dark := State{Module: "mesh-mail", Name: "sends", kv: func(context.Context) (jetstream.KeyValue, error) { + if blackedOut > 0 { + blackedOut-- + return nil, jetstream.ErrJetStreamNotEnabled + } + return kv, nil + }} + if rev, err := dark.Put("k", 1); err != nil || rev == 0 || blackedOut != 0 { + t.Fatalf("a Put through a short blackout: %d, %v", rev, err) + } } // 1840/S1: given a state, when its keys are listed, then they come sorted, and none is an empty list, not an error. diff --git a/go/state.go b/go/state.go index a085570..90824d8 100644 --- a/go/state.go +++ b/go/state.go @@ -6,6 +6,7 @@ import ( "fmt" "time" + "git.novox.be/novox/mesh-sdk/go/busretry" "git.novox.be/novox/mesh-sdk/go/kvlimit" ) @@ -85,7 +86,10 @@ func (s KeptState) ask(verb string, params map[string]any) (json.RawMessage, err return nil, err } params["state"] = s.name - raw, err := c.ask("mesh/state."+verb, params, stateTimeout) + // A request the runtime's bus delivered to no one is asked again for a few seconds (busretry): nothing was done. + raw, err := busretry.Do("the state "+s.name+", "+verb+" "+key, func() (json.RawMessage, error) { + return c.ask("mesh/state."+verb, params, stateTimeout) + }, nil) if size >= 0 { err = kvlimit.Refused("the state "+s.name, key, size, kvlimit.ModuleStateValueBytes, err) } From 61b67d61e2aaed1457e6e58410a8e0597805b4c0 Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 30 Oct 2026 18:19:20 +0100 Subject: [PATCH 07/11] Export the accounts block's user entry, so a kept user is written in ComposeAccounts' own lines (issue 774) mesh-grants' refused.go keptEntry wrote a user of the accounts block from grants it keeps by copying ComposeAccounts' lines; a change to one would leave the other writing a list the bus reads differently. buscompose.AccountsEntry(user, passwordHash, perms) is now the one writer, and ComposeAccounts uses it. No change to what is written. --- go/buscompose/buscompose_test.go | 7 +++++++ go/buscompose/nats.go | 31 ++++++++++++++++++++----------- 2 files changed, 27 insertions(+), 11 deletions(-) diff --git a/go/buscompose/buscompose_test.go b/go/buscompose/buscompose_test.go index 3d02084..d483eba 100644 --- a/go/buscompose/buscompose_test.go +++ b/go/buscompose/buscompose_test.go @@ -573,6 +573,13 @@ func TestTheCopiedComposerComposesAsTheControllers(t *testing.T) { if reads[VaultSeat] != ControllerLease+","+GrantsAccounts || reads["mesh-config"] != "" || reads["restic"] != "" || reads["messenger"] != "" { t.Errorf("the reads granted for their readers: %v", reads) } + // AccountsEntry is the very lines ComposeAccounts writes for a user, so a writer of kept users repeats none. + filled, _ := WithPasswords(judged, map[string]string{judged[0].Username(): "$2a$11$h"}) + list, err := ComposeAccounts(filled[:1]) + perms, perr := PermissionsFor(filled[0]) + if err != nil || perr != nil || !strings.Contains(list, " users = [\n"+AccountsEntry(filled[0].Username(), "$2a$11$h", perms)+" ]\n") { + t.Errorf("the accounts entry is not ComposeAccounts' own: %v %v\n%s", err, perr, list) + } c, _ := ConsumerFor(Principal{Kind: KindModule, Node: "a", Module: "m", Consumes: []string{"x.happened"}}) cfg := ConsumerConfig(c) if cfg.Durable != "a_m" || cfg.AckPolicy != jetstream.AckExplicitPolicy || cfg.AckWait != 30*time.Second || diff --git a/go/buscompose/nats.go b/go/buscompose/nats.go index 7e536e2..85dbab2 100644 --- a/go/buscompose/nats.go +++ b/go/buscompose/nats.go @@ -1680,17 +1680,7 @@ func ComposeAccounts(principals []Principal) (string, error) { if p.PasswordHash == "" { return "", fmt.Errorf("%s has no password hash: a user without one is a user anybody is", p.Username()) } - fmt.Fprintf(&b, " { user: %q, password: %q, permissions: {\n", p.Username(), p.PasswordHash) - if len(perms.PublishDeny) > 0 { - fmt.Fprintf(&b, " publish: { allow: [%s], deny: [%s] }\n", quoted(perms.Publish), quoted(perms.PublishDeny)) - } else { - fmt.Fprintf(&b, " publish: { allow: [%s] }\n", quoted(perms.Publish)) - } - fmt.Fprintf(&b, " subscribe: { allow: [%s] }\n", quoted(perms.Subscribe)) - if perms.AllowResponses { - fmt.Fprintf(&b, " allow_responses: { max: 1, ttl: \"%dm\" }\n", int(ResponseTTL/time.Minute)) - } - b.WriteString(" } }\n") + b.WriteString(AccountsEntry(p.Username(), p.PasswordHash, perms)) } b.WriteString(" ]\n }\n") b.WriteString(systemAccount) @@ -1698,6 +1688,25 @@ func ComposeAccounts(principals []Principal) (string, error) { return b.String(), nil } +// AccountsEntry is one user of the accounts block, as ComposeAccounts writes each: its name, its password hash and its +// permissions, in the lines the bus reads. Exported so a writer that keeps a user from grants it holds rather than +// composes (mesh-grants' refused users) writes the same lines, never a copy of them (issue 774). +func AccountsEntry(user, passwordHash string, perms Permissions) string { + var b strings.Builder + fmt.Fprintf(&b, " { user: %q, password: %q, permissions: {\n", user, passwordHash) + if len(perms.PublishDeny) > 0 { + fmt.Fprintf(&b, " publish: { allow: [%s], deny: [%s] }\n", quoted(perms.Publish), quoted(perms.PublishDeny)) + } else { + fmt.Fprintf(&b, " publish: { allow: [%s] }\n", quoted(perms.Publish)) + } + fmt.Fprintf(&b, " subscribe: { allow: [%s] }\n", quoted(perms.Subscribe)) + if perms.AllowResponses { + fmt.Fprintf(&b, " allow_responses: { max: 1, ttl: \"%dm\" }\n", int(ResponseTTL/time.Minute)) + } + b.WriteString(" } }\n") + return b.String() +} + // WatchmanModule is the one module whose principal hears the bus's connect and disconnect events and asks its list of // connections (novox/hq ADR 0431 §2, issue 1147): the watchman of the repository mesh-delivery, which writes presence // from them. No other principal is granted the subjects below, and nothing is a wildcard over them. From 942f906222849d57a15b0ace742490ae1312092e Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 30 Oct 2026 18:25:59 +0100 Subject: [PATCH 08/11] Judge only the verbs of a mesh seat a module invokes, not its holder (issue 774) The seat-invoke rule refused mesh-config, mesh-healers and messenger calling the controller's verbs, and mesh-self-check calling the watchman's seat, over a catalogue that holds no manifest of their holders: the controller answers its seat itself, and the watchman lives in mesh-delivery. mesh-control's TestModuleCheckPassesTheCatalogue (the captured mesh-catalog) failed on it. A mesh seat is held by the mesh or by a module of any repository, so one repository's catalogue cannot say it has no holder: for a mesh seat only the verb is judged; a seat a module declares is still judged for a claimant. Over the captured catalogue, the live mesh-catalog alone, and mesh-catalog, mesh-control, mesh-delivery and mesh-host together: no refusal. --- go/catalogue/seat_invokes.go | 8 ++++++-- go/catalogue/seats_declared_test.go | 8 ++++++-- 2 files changed, 12 insertions(+), 4 deletions(-) diff --git a/go/catalogue/seat_invokes.go b/go/catalogue/seat_invokes.go index 583ad4e..d130b3d 100644 --- a/go/catalogue/seat_invokes.go +++ b/go/catalogue/seat_invokes.go @@ -29,7 +29,8 @@ func SeatInvoke(invoke string) (seat, verb string, ok bool) { // seatInvokeProblems are the `seat:.` a module or one of its processes invokes that nothing answers: a seat // no module declares and the mesh does not define, a verb the seat does not promise, or a seat with no holder (a -// mesh seat no module claims, unless its node's engine holds it, ADR 0563 §1). +// seat a module declares and no module claims; a mesh seat's holder, the mesh itself or a module of any repository, +// is not judged). func seatInvokeProblems(shelf Shelf, declared map[string]SeatDeclaration) []string { claimed := map[string]bool{} for _, m := range shelf { @@ -51,7 +52,10 @@ func seatInvokeProblems(shelf Shelf, declared map[string]SeatDeclaration) []stri var verbs []Verb held := claimed[seat] if s, isMesh := SeatNamed(seat); isMesh { - verbs, held = s.Serves, held || claimed[s.Name] || s.HeldByTheEngine + // A mesh seat is held by the mesh itself or by a module of any repository: one repository's catalogue + // cannot say it has no holder, so only its verbs are judged (the lead, 2026-10-30: mesh-control's check + // runs over the captured mesh-catalog alone, whose mesh-self-check calls the watchman's seat). + verbs, held = s.Serves, true } else if d, isModule := declared[seat]; isModule { verbs = d.Serves } else { diff --git a/go/catalogue/seats_declared_test.go b/go/catalogue/seats_declared_test.go index fdef894..41dfabc 100644 --- a/go/catalogue/seats_declared_test.go +++ b/go/catalogue/seats_declared_test.go @@ -71,8 +71,12 @@ func TestUsingASeatNobodyDeclaresIsRefused(t *testing.T) { Manifest{Module: "shop", Invokes: []string{"seat:telegram-sendr.status"}}, telegram(), true}, {"a verb of a declared seat nobody claims", `a verb of the seat "telegram-sender", ` + UnheldSeat, Manifest{Module: "shop", Invokes: []string{"seat:telegram-sender.status"}}, unclaimed, true}, - {"a verb of a mesh seat nobody claims", `a verb of the seat "mesh-controller", ` + UnheldSeat, - Manifest{Module: "shop", Invokes: []string{"seat:mesh-controller.root-free"}}, telegram(), true}, + {"a verb of a mesh seat whose holder lives in another repository", "", + Manifest{Module: "shop", Invokes: []string{"seat:mesh-watchdogs.signals"}}, telegram(), false}, + {"a verb a mesh seat does not promise", `the seat mesh-controller serves no verb "root-freed"`, + Manifest{Module: "shop", Invokes: []string{"seat:mesh-controller.root-freed"}}, telegram(), false}, + {"the controller's verb: the controller answers its seat", "", + Manifest{Module: "shop", Invokes: []string{"seat:mesh-controller.root-free"}}, telegram(), false}, {"the engine's verb: every node's engine holds its seat", "", Manifest{Module: "shop", Invokes: []string{"seat:node-engine.root-free"}}, telegram(), false}, {"a verb the engine's seat does not promise", `the seat node-engine serves no verb "root-freed"`, From ef1ec1c93544f0fc12b2dca608167bf2af11a846 Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 30 Oct 2026 18:26:23 +0100 Subject: [PATCH 09/11] Name mesh-config among the controller's lease readers (issue 774) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit mesh-config reads mesh-controller.lease to take the config over only once the lease is handed away (ADR 0561 §3, its takeover.go leaseAway), and was refused it as a reader the composer does not name. --- go/buscompose/buscompose_test.go | 2 +- go/buscompose/userlist_switch.go | 7 ++++++- 2 files changed, 7 insertions(+), 2 deletions(-) diff --git a/go/buscompose/buscompose_test.go b/go/buscompose/buscompose_test.go index d483eba..c936737 100644 --- a/go/buscompose/buscompose_test.go +++ b/go/buscompose/buscompose_test.go @@ -570,7 +570,7 @@ func TestTheCopiedComposerComposesAsTheControllers(t *testing.T) { reads[u.Module] = strings.Join(u.Reads, ",") } } - if reads[VaultSeat] != ControllerLease+","+GrantsAccounts || reads["mesh-config"] != "" || reads["restic"] != "" || reads["messenger"] != "" { + if reads[VaultSeat] != ControllerLease+","+GrantsAccounts || reads["mesh-config"] != ControllerLease || reads["restic"] != "" || reads["messenger"] != "" { t.Errorf("the reads granted for their readers: %v", reads) } // AccountsEntry is the very lines ComposeAccounts writes for a user, so a writer of kept users repeats none. diff --git a/go/buscompose/userlist_switch.go b/go/buscompose/userlist_switch.go index cbd7f98..fece755 100644 --- a/go/buscompose/userlist_switch.go +++ b/go/buscompose/userlist_switch.go @@ -84,6 +84,9 @@ const VaultBusAccounts = "mesh-vault.bus-accounts" // ArtifactStoreTools is the module holding the seat mesh-artifact-sweep (novox/hq ADR 0571 clause 2). const ArtifactStoreTools = "artifact-store-tools" +// ConfigModule is the module that holds the seat mesh-config, a reader of the controller's lease (NamedReaders). +const ConfigModule = "mesh-config" + // NamedReaders are the states only the modules named may read, by `.`: the grants part's wanted accounts // are read by the vault alone (novox/hq ADR 0561 §2). // The controller's lease is read by the parts that turn on at its write (ADR 0561 §3, §6), and by the assigner, which @@ -95,7 +98,9 @@ const ArtifactStoreTools = "artifact-store-tools" var NamedReaders = map[string][]string{VaultBusAccounts: {GrantsModule}, GrantsAccounts: {VaultSeat}, ControllerLease: {VaultSeat, GrantsModule, delivery.AssignerModule, health.KeeperModule, health.HealersModule, WatchmanModule, // ADR 0571 clause 2: the artifact store's seat sweeps only once the lease names no controller. - ArtifactStoreTools}, + ArtifactStoreTools, + // ADR 0561 §3: mesh-config takes over the config only once the lease is handed away (its takeover.go leaseAway). + ConfigModule}, ControllerConditions: {health.KeeperModule}, ControllerHandActs: {health.HealersModule, handacts.Module}, HealersWould: {}} // ReaderMay says whether module may read the state read (`.`) by the named readers: always, for a state From 206f794eb319136a93cfaac57e5f18bee0a1ce8b Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 30 Oct 2026 18:35:54 +0100 Subject: [PATCH 10/11] Grant a bar's holder the state a contributor's item shows, at its own machine's key (issue 774) i3status-rust on g14 and shanks was issued nothing to read ("was issued: none", 148 times): ReadsGranted derived a holder's reads from contributions' data alone, and design 57 L3 moved power's draw block to an item (mesh-catalog ca8d6f33: shows: state, state: power.draw). ReadsGranted now reads a module's items of a kind the holder's seat receives too, their attributes in a contribution's data shape, always at the machine's own key (ADR 0260), never another's. Both callers, the controller's busrecords and mesh-grants' records.go, get it through ReadsGranted with no change of their own. --- go/catalogue/seat_data.go | 30 +++++++++++++++++++++++++++++- go/catalogue/seat_data_test.go | 10 ++++++++++ 2 files changed, 39 insertions(+), 1 deletion(-) diff --git a/go/catalogue/seat_data.go b/go/catalogue/seat_data.go index 90a1fb7..ab3bbcf 100644 --- a/go/catalogue/seat_data.go +++ b/go/catalogue/seat_data.go @@ -155,7 +155,9 @@ var stateKeyName = regexp.MustCompile(`^[a-z0-9][a-z0-9-]*$`) // ReadsGranted is the state a holder reads because modules on its machine offer it in pieces of a // kind its seat receives as data (novox/hq ADR 0260), each with the one key the piece shows — the -// machine's own when it names none — sorted and once each. The bus grants the holder these keys and +// machine's own when it names none — sorted and once each. A module's items of such a kind (design 57 L3) grant +// what they show too, always at the machine's own key; an item whose capability the machine lacks grants a key +// that is never written, which is no wider than its own machine's. The bus grants the holder these keys and // no others: the holder names no contributor, and reads only what is offered to it where it runs. func ReadsGranted(holder Manifest, onMachine []Manifest, machine string) []StateRead { seen := map[StateRead]bool{} @@ -183,6 +185,21 @@ func ReadsGranted(holder Manifest, onMachine []Manifest, machine string) []State } } } + // A moved contributor's items of the kind (design 57 L3): what each shows, read from its attributes as a + // contribution's data, and always this machine's own key (ADR 0260), never another's (issue 774: + // power's draw block became an item and i3status-rust was issued nothing). + for _, it := range m.Items { + if it.Collection != "contribution:"+s.Name+":"+r.Kind { + continue + } + for _, sr := range r.Shape.Reads(itemData(it)) { + sr.Key = machine + if module, _, err := ReadState(sr.Read); err == nil && module == m.Module && module != holder.Module && + stateKeyName.MatchString(sr.Key) { + seen[sr] = true + } + } + } } } } @@ -199,6 +216,17 @@ func ReadsGranted(holder Manifest, onMachine []Manifest, machine string) []State return out } +// itemData is an item's attributes in the shape of a contribution's data: each at the top and among the options, where +// a shape's Reads looks for what a piece shows (barBlockReads: shows, options.state). +func itemData(it Item) map[string]any { + data, options := map[string]any{}, map[string]any{} + for k, v := range it.Attributes { + data[k], options[k] = v, v + } + data["options"] = options + return data +} + // wholeIn is v as a whole number within [lo, hi]; JSON gives every number as a float. func wholeIn(v any, lo, hi int) (int, bool) { f, ok := v.(float64) diff --git a/go/catalogue/seat_data_test.go b/go/catalogue/seat_data_test.go index 3d2313e..d39c59e 100644 --- a/go/catalogue/seat_data_test.go +++ b/go/catalogue/seat_data_test.go @@ -394,6 +394,16 @@ func TestAStateBlockIsRenderedForThisMachineAndTheHolderIsGrantedToReadIt(t *tes if got := ReadsGranted(barHolder(), []Manifest{barHolder()}, "laptop"); len(got) != 0 { t.Fatalf("a holder alone is granted %v", got) } + // Issue 774 (design 57 L3): the block as an item grants the same read, at this machine's own key alone, even where + // the item names another machine's. + for _, attrs := range []map[string]string{{"shows": "state", "state": "power.draw"}, {"shows": "state", "state": "power.draw", "key": "other"}} { + moved := Manifest{Module: "power", State: []StateDeclaration{{Name: "draw"}}, Items: []Item{{ + Collection: "contribution:node-bar:block", Name: "power-draw", IfCapability: "power-meter", Attributes: attrs}}} + if got := ReadsGranted(barHolder(), []Manifest{barHolder(), moved}, "laptop"); len(got) != 1 || + got[0] != (StateRead{Read: "power.draw", Key: "laptop"}) { + t.Errorf("the holder is granted %v by the item %v", got, attrs) + } + } } func TestAModuleOffersItsOwnStateOnly(t *testing.T) { From a37181261c24a15b8092c06e11cce63e0e5f8ade Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 30 Oct 2026 18:41:37 +0100 Subject: [PATCH 11/11] Never ask a call again that its callee answered, whatever its words (issue 774) The review found that busretry could ask a delivered call again: Conn.request turned a callee's error answer into errors.New(words), and Undelivered matched the words "no responders" and "jetstream not enabled", so a callee that answered so during a blackout ran its work twice. A callee's answer is now *busretry.Answered, which Undelivered never counts: only the transport's own no responders makes a call undelivered. The other busretry sites read or write through JetStream's API (State, Ask) or the runtime's state verbs, where the words are the transport's. --- go/busretry/busretry.go | 9 ++++++++- go/process/call.go | 18 ++++++++++++++---- go/process/call_test.go | 31 +++++++++++++++++++++++++++++++ 3 files changed, 53 insertions(+), 5 deletions(-) diff --git a/go/busretry/busretry.go b/go/busretry/busretry.go index a8adbff..acc523c 100644 --- a/go/busretry/busretry.go +++ b/go/busretry/busretry.go @@ -33,7 +33,8 @@ var ( // no response from the stream after the client's own retries on no responders; also said in words by a runtime that // passed the error on. func Undelivered(err error) bool { - if err == nil { + var answered *Answered + if err == nil || errors.As(err, &answered) { return false } if errors.Is(err, nats.ErrNoResponders) || errors.Is(err, jetstream.ErrJetStreamNotEnabled) || @@ -45,6 +46,12 @@ func Undelivered(err error) bool { return strings.Contains(words, "no responders") || strings.Contains(words, "jetstream not enabled") } +// Answered is an error a callee answered with: the request was delivered, whatever its words say ("jetstream not +// enabled" in a callee's own answer), so Undelivered never counts it and Do never asks it again. +type Answered struct{ Words string } + +func (a *Answered) Error() string { return a.Words } + // Do runs try, and again while its error says the request reached no one, until Bound has passed. blackout, when // given, is asked after each such error and Do stops waiting when it says the bus is not blacked out: a call nobody // answers because its callee is not running fails at once, as before. what names the request in the error. diff --git a/go/process/call.go b/go/process/call.go index 7bf1252..95cce50 100644 --- a/go/process/call.go +++ b/go/process/call.go @@ -58,15 +58,25 @@ func (c *Conn) callOn(called, fallback string, body any) (json.RawMessage, error if err != nil { return nil, err } + return callWith(c.request, c.blackedOut, called, fallback, data) +} + +// callWith is callOn over a request and a blackout probe. Only the transport's own errors can count as undelivered: a +// callee's answer is *busretry.Answered, whatever its words, so a call that reached its callee is never asked again. +func callWith(request func(string, []byte) (json.RawMessage, error), blackout func() bool, called, fallback string, + data []byte) (json.RawMessage, error) { return busretry.Do("the call on "+fallback, func() (json.RawMessage, error) { - result, err := c.request(called, data) + result, err := request(called, data) if errors.Is(err, nats.ErrNoResponders) { - result, err = c.request(fallback, data) + result, err = request(fallback, data) } return result, err - }, c.blackedOut) + }, blackout) } +// answered is the error a callee answered with: delivered, never asked again (busretry.Answered). +func answered(words string) error { return &busretry.Answered{Words: words} } + // blackedOut says the bus answers this account nothing at all: its JetStream API, which every account imports, has no // responder or says it is not enabled. func (c *Conn) blackedOut() bool { @@ -136,7 +146,7 @@ func (c *Conn) request(subject string, data []byte) (json.RawMessage, error) { return nil, fmt.Errorf("%s answered something that is not a seat's answer: %w", subject, err) } if r.Error != "" { - return nil, errors.New(r.Error) + return nil, answered(r.Error) } return r.Result, nil } diff --git a/go/process/call_test.go b/go/process/call_test.go index 6f33eb2..a892874 100644 --- a/go/process/call_test.go +++ b/go/process/call_test.go @@ -1,8 +1,14 @@ package process import ( + "encoding/json" "strings" "testing" + "time" + + "github.com/nats-io/nats.go" + + "git.novox.be/novox/mesh-sdk/go/busretry" ) // 1191/AC3: given a module tool key ., when called, then the subject is @@ -79,3 +85,28 @@ func TestCallOnRefusesANodeThatIsNotOneToken(t *testing.T) { } } } + +// The review of 2026-10-30 (issue 774): a callee's own answer is delivered, whatever its words, and is never asked +// again during a blackout; only the transport's no responders is. +func TestACallIsAskedAgainOnlyWhenItReachedNoOne(t *testing.T) { + busretry.Bound, busretry.FirstWait, busretry.MostWait = 50*time.Millisecond, time.Millisecond, 2*time.Millisecond + for _, row := range []struct { + name string + answer []error // per ask; the last repeats + asks int + }{ + {"a callee answers jetstream not enabled: asked once", []error{answered("jetstream not enabled")}, 1}, + {"a callee answers no responders in its words: asked once", []error{answered("nats: no responders available for request")}, 1}, + {"no responders on either subject, then an answer", []error{nats.ErrNoResponders, nats.ErrNoResponders, nil}, 3}, + } { + asks := 0 + _, err := callWith(func(string, []byte) (json.RawMessage, error) { + e := row.answer[min(asks, len(row.answer)-1)] + asks++ + return json.RawMessage(`{}`), e + }, func() bool { return true }, "called", "fallback", nil) + if asks != row.asks { + t.Errorf("%s: asked %d times (%v), want %d", row.name, asks, err, row.asks) + } + } +}