From 637c33b7f81e12f43f9ff2482543c906f20ba899 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 22 Oct 2026 14:15:33 +0200 Subject: [PATCH 1/7] issue 1840/S1c: name declared-packages among the controller's readable states, so a manifest may read it --- go/catalogue/state.go | 4 ++-- go/catalogue/state_test.go | 9 +++++++++ 2 files changed, 11 insertions(+), 2 deletions(-) diff --git a/go/catalogue/state.go b/go/catalogue/state.go index 87e402f..a6b183b 100644 --- a/go/catalogue/state.go +++ b/go/catalogue/state.go @@ -134,8 +134,8 @@ func StateProblems(m Manifest) []string { // second way and granted to the module's own principal as well. Named here so a read of one is a read of // kept state. The bus's list is broker.ControllerMirrors; a test keeps the two agreeing. // plans-building is the plans in their build phase, read by the checker to hold checks while a delivery builds (ADR 0483 -// clause 2, issue 1687). -var ControllerMirrors = []string{"assignments", "declared-config", "predefined", "builds", "rollouts", "checks-running", "plans-building"} +// clause 2, issue 1687). declared-packages is the packages a node's declaration names (ADR 0507, issue 1840/S1c). +var ControllerMirrors = []string{"assignments", "declared-config", "predefined", "builds", "rollouts", "checks-running", "plans-building", "declared-packages"} // StateReadsNothingDeclares is every read across a catalogue whose owner is present and declares no // such state. An absent owner says nothing — a module may be installed long before the one whose diff --git a/go/catalogue/state_test.go b/go/catalogue/state_test.go index 475ea2e..bae68b9 100644 --- a/go/catalogue/state_test.go +++ b/go/catalogue/state_test.go @@ -92,6 +92,15 @@ func TestAReadOfThePlansBuildingIsAReadOfKeptState(t *testing.T) { } } +// A module may read `mesh-controller.declared-packages` (issue 1840/S1c, ADR 0507): the packages a node's declaration names. +func TestAReadOfTheDeclaredPackagesIsAReadOfKeptState(t *testing.T) { + controller := Manifest{Module: ControllerSeatName} + reader := Manifest{Module: "some-reader", Reads: []string{"mesh-controller.declared-packages"}} + if p := StateReadsNothingDeclares([]Manifest{controller, reader}); len(p) != 0 { + t.Fatalf("a read of the controller's declared-packages was refused: %v", p) + } +} + // **Across the whole catalogue**: every state name is local, and every read whose owner is present // names state that owner keeps. func TestEveryManifestsStateIsLocalAndEveryReadIsKept(t *testing.T) { From 26517ac00b3b8c59d81b4fe2086daadf70246b5e Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 22 Oct 2026 14:16:13 +0200 Subject: [PATCH 2/7] issue 1840/S1: let a process put a key whole and list a state's keys, as a tools bundle can --- go/process/state.go | 39 ++++++++++++++++++ go/process/state_integration_test.go | 35 +++++++++++++++++ go/process/state_test.go | 59 ++++++++++++++++++++++++++++ 3 files changed, 133 insertions(+) diff --git a/go/process/state.go b/go/process/state.go index 43221fa..e6c0655 100644 --- a/go/process/state.go +++ b/go/process/state.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "sort" "time" "github.com/nats-io/nats.go" @@ -128,3 +129,41 @@ func (s State) Update(key string, value any, revision uint64) (uint64, error) { } return rev, nil } + +// Put writes a key whole, whatever it held, and returns the revision written. +func (s State) Put(key string, value any) (uint64, error) { + ctx, cancel := context.WithTimeout(context.Background(), stateTimeout) + defer cancel() + kv, err := s.store(ctx) + if err != nil { + return 0, err + } + data, err := json.Marshal(value) + if err != nil { + return 0, err + } + rev, err := kv.Put(ctx, key, data) + if err != nil { + return 0, fmt.Errorf("%s's state %s, key %s: %w", s.Module, s.Name, key, err) + } + return rev, nil +} + +// Keys is every key with a current value, sorted; none is an empty list. +func (s State) Keys() ([]string, error) { + ctx, cancel := context.WithTimeout(context.Background(), stateTimeout) + defer cancel() + kv, err := s.store(ctx) + if err != nil { + return nil, err + } + keys, err := kv.Keys(ctx) + if errors.Is(err, jetstream.ErrNoKeysFound) { + return []string{}, nil + } + if err != nil { + return nil, fmt.Errorf("%s's state %s, keys: %w", s.Module, s.Name, err) + } + sort.Strings(keys) + return keys, nil +} diff --git a/go/process/state_integration_test.go b/go/process/state_integration_test.go index cc01c15..01085b3 100644 --- a/go/process/state_integration_test.go +++ b/go/process/state_integration_test.go @@ -94,3 +94,38 @@ func TestStateRefusesAnUngrantedBucketNamingIt(t *testing.T) { t.Errorf("the refusal %q does not name the bucket mesh-mail_unrelated", err.Error()) } } + +// 1840/S1: given a process connection granted its bucket, when it puts keys whole and lists them, then Put +// returns a new revision each time and Keys lists them sorted (none for a bucket with none). +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 { + t.Fatalf("the admin could not make the bucket: %v", err) + } + conn, err := Connect(account) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + s := conn.State("sends") + + if keys, err := s.Keys(); err != nil || len(keys) != 0 { + t.Fatalf("Keys of an empty bucket: %v, %v", keys, err) + } + rev, err := s.Put("b", map[string]any{"v": 1}) + if err != nil || rev == 0 { + t.Fatalf("Put on a granted bucket: %d, %v", rev, err) + } + rev2, err := s.Put("b", map[string]any{"v": 2}) + if err != nil || rev2 <= rev { + t.Fatalf("Put over a key: %d, %v", rev2, err) + } + if _, err := s.Put("a", 1); err != nil { + t.Fatal(err) + } + keys, err := s.Keys() + if err != nil || strings.Join(keys, ",") != "a,b" { + t.Fatalf("Keys: %v, %v, want a,b", keys, err) + } +} diff --git a/go/process/state_test.go b/go/process/state_test.go index 542153b..ba02ad9 100644 --- a/go/process/state_test.go +++ b/go/process/state_test.go @@ -3,6 +3,8 @@ package process import ( "context" "errors" + "sort" + "strings" "sync" "testing" @@ -68,6 +70,28 @@ func (f *fakeKV) Update(_ context.Context, key string, value []byte, revision ui return f.next, nil } +func (f *fakeKV) Put(_ context.Context, key string, value []byte) (uint64, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.next++ + f.values[key], f.revs[key] = value, f.next + return f.next, nil +} + +func (f *fakeKV) Keys(_ context.Context, _ ...jetstream.WatchOpt) ([]string, error) { + f.mu.Lock() + defer f.mu.Unlock() + if len(f.values) == 0 { + return nil, jetstream.ErrNoKeysFound + } + keys := make([]string, 0, len(f.values)) + for k := range f.values { + keys = append(keys, k) + } + sort.Sort(sort.Reverse(sort.StringSlice(keys))) // as a bucket lists them: in no order the caller may rely on + return keys, nil +} + func stateOn(kv jetstream.KeyValue) State { return State{Module: "mesh-mail", Name: "sends", kv: func(context.Context) (jetstream.KeyValue, error) { return kv, nil }} } @@ -103,3 +127,38 @@ func TestStateSatisfiesAskerStore(t *testing.T) { t.Fatal("Update did not advance the revision") } } + +// 1840/S1: given a state, when a key is put twice, then each Put returns a new revision and the key reads +// back the last value, whether or not it had one. +func TestStatePutWritesAKeyWhole(t *testing.T) { + s := stateOn(newFakeKV()) + rev, err := s.Put("k", map[string]any{"a": 1}) + if err != nil || rev == 0 { + t.Fatalf("Put of a new key: %d, %v", rev, err) + } + rev2, err := s.Put("k", map[string]any{"a": 2}) + if err != nil || rev2 <= rev { + t.Fatalf("Put over a key: %d, %v, want a revision after %d", rev2, err, rev) + } + e, err := s.Get("k") + if err != nil || e == nil || e.Revision != rev2 || string(e.Value) != `{"a":2}` { + t.Fatalf("Get after Put: %+v, %v", e, err) + } +} + +// 1840/S1: given a state, when its keys are listed, then they come sorted, and none is an empty list, not an error. +func TestStateKeysAreSortedAndEmptyWhenNone(t *testing.T) { + s := stateOn(newFakeKV()) + if keys, err := s.Keys(); err != nil || len(keys) != 0 { + t.Fatalf("Keys of an empty state: %v, %v", keys, err) + } + for _, k := range []string{"b", "c", "a"} { + if _, err := s.Put(k, 1); err != nil { + t.Fatal(err) + } + } + keys, err := s.Keys() + if err != nil || strings.Join(keys, ",") != "a,b,c" { + t.Fatalf("Keys: %v, %v, want a,b,c", keys, err) + } +} From 076e1f7947db3277e572cee2425011121b8b2ca8 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 22 Oct 2026 14:16:59 +0200 Subject: [PATCH 3/7] issue 1840/S2: let a process watch a state, current values first then each change, as a tools bundle can --- go/process/state.go | 98 ++++++++++++++++++++++- go/process/state_integration_test.go | 105 ++++++++++++++++++++++++ go/process/state_test.go | 115 +++++++++++++++++++++++++++ 3 files changed, 317 insertions(+), 1 deletion(-) diff --git a/go/process/state.go b/go/process/state.go index e6c0655..1e79302 100644 --- a/go/process/state.go +++ b/go/process/state.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "sort" + "strings" "time" "github.com/nats-io/nats.go" @@ -28,6 +29,8 @@ type State struct { // kv binds this state's underlying key-value store: a live connection's own (Conn.State), or, in a // unit test, a fake that never reaches a bus. kv func(ctx context.Context) (jetstream.KeyValue, error) + // closed says the connection is gone, so a watch ends; nil in a unit test. + closed func() bool } // bucket is this state's name on the bus. @@ -39,7 +42,7 @@ func (s State) bucket() string { return s.Module + "_" + s.Name } // not granted is refused there, promptly, as nats.ErrPermissionViolation. func (c *Conn) State(name string) State { bucket := c.account.Module + "_" + name - return State{Module: c.account.Module, Name: name, kv: func(ctx context.Context) (jetstream.KeyValue, error) { + return State{Module: c.account.Module, Name: name, closed: func() bool { return c.nc.IsClosed() }, kv: func(ctx context.Context) (jetstream.KeyValue, error) { js, err := jetstream.New(c.nc) if err != nil { return nil, err @@ -167,3 +170,96 @@ func (s State) Keys() ([]string, error) { sort.Strings(keys) return keys, nil } + +// watchPattern is a key narrowing as the bus names it: `*` stays one name, `**` is the rest, empty is every key. +func watchPattern(key string) string { + if key == "" { + return jetstream.AllKeys + } + tokens := strings.Split(key, ".") + for i, t := range tokens { + if t == "**" { + tokens[i] = ">" + } + } + return strings.Join(tokens, ".") +} + +// Watch hands handle every current value — none that is deleted, each with Current set — 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 changes go on for the life of the +// connection; an error from handle, or the connection closing, ends them. Before the current values are +// handled, either is returned as the error. +func (s State) Watch(key string, handle func(stdio.StateChange) error) error { + // the watch outlives this call, so its context is not the call's timeout: the connection ends it. + ctx, cancel := context.WithCancel(context.Background()) + bind, stopBind := context.WithTimeout(ctx, stateTimeout) + kv, err := s.store(bind) + stopBind() + if err != nil { + cancel() + return err + } + w, err := kv.Watch(ctx, watchPattern(key)) + if err != nil { + cancel() + return fmt.Errorf("%s's state %s, watching %q: %w", s.Module, s.Name, key, err) + } + end := func() { w.Stop(); cancel() } + next := func() (jetstream.KeyValueEntry, bool, error) { + tick := time.NewTicker(250 * time.Millisecond) + defer tick.Stop() + for { + select { + case e, open := <-w.Updates(): + if !open { + return nil, false, fmt.Errorf("%s's state %s: the watch ended", s.Module, s.Name) + } + return e, true, nil + case <-tick.C: + if s.closed != nil && s.closed() { + return nil, false, fmt.Errorf("%s's state %s: the connection closed", s.Module, s.Name) + } + } + } + } + deliver := func(e jetstream.KeyValueEntry, current bool) error { + c := stdio.StateChange{State: s.Name, Key: e.Key(), Op: "put", Revision: e.Revision(), Current: current} + if e.Operation() == jetstream.KeyValuePut { + c.Value = json.RawMessage(e.Value()) + } else { + c.Op = "delete" + } + return handle(c) + } + for { + e, _, err := next() + if err != nil { + end() + return err + } + if e == nil { + break + } + if e.Operation() != jetstream.KeyValuePut { + continue + } + if err := deliver(e, true); err != nil { + end() + return err + } + } + go func() { + defer end() + for { + e, _, err := next() + if err != nil || e == nil { + return + } + if deliver(e, false) != nil { + return + } + } + }() + return nil +} diff --git a/go/process/state_integration_test.go b/go/process/state_integration_test.go index 01085b3..f0b1cfd 100644 --- a/go/process/state_integration_test.go +++ b/go/process/state_integration_test.go @@ -6,6 +6,7 @@ import ( "errors" "strings" "testing" + "time" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" @@ -129,3 +130,107 @@ func TestStatePutAndKeysOnAGrantedBucket(t *testing.T) { t.Fatalf("Keys: %v, %v, want a,b", keys, err) } } + +// 1840/S2: given a process connection granted its bucket, which already holds two keys and a deleted one, +// when it watches, then it gets the two current values marked Current, and then a later put and a delete in +// order; a narrowed watch gets only the keys it names. +func TestStateWatchOnAGrantedBucket(t *testing.T) { + url, account := grantedSendProcess(t) + _, adminJetStream := adminJS(t, url) + kv, err := adminJetStream.CreateKeyValue(ctxFor(t), jetstream.KeyValueConfig{Bucket: "mesh-mail_sends"}) + if err != nil { + t.Fatalf("the admin could not make the bucket: %v", err) + } + ctx := ctxFor(t) + for k, v := range map[string]string{"a.x": `1`, "a.y": `2`, "b.z": `3`, "gone": `4`} { + if _, err := kv.Put(ctx, k, []byte(v)); err != nil { + t.Fatal(err) + } + } + if err := kv.Delete(ctx, "gone"); err != nil { + t.Fatal(err) + } + conn, err := Connect(account) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + s := conn.State("sends") + + all := make(chan stdio.StateChange, 16) + if err := s.Watch("", func(c stdio.StateChange) error { all <- c; return nil }); err != nil { + t.Fatalf("Watch on a granted bucket: %v", err) + } + current := map[string]bool{} + for len(all) > 0 { + c := <-all + if !c.Current || c.Op != "put" { + t.Fatalf("a current value arrived as %+v", c) + } + current[c.Key] = true + } + if len(current) != 3 || current["gone"] { + t.Fatalf("current values %v, want a.x, a.y, b.z and not gone", current) + } + + narrow := make(chan stdio.StateChange, 16) + if err := s.Watch("a.*", func(c stdio.StateChange) error { narrow <- c; return nil }); err != nil { + t.Fatal(err) + } + if len(narrow) != 2 { + t.Fatalf("a narrowed watch handed %d current values, want 2", len(narrow)) + } + + if _, err := s.Put("a.x", 5); err != nil { + t.Fatal(err) + } + if err := kv.Delete(ctx, "a.y"); err != nil { + t.Fatal(err) + } + for _, want := range []struct{ key, op string }{{"a.x", "put"}, {"a.y", "delete"}} { + select { + case c := <-all: + if c.Key != want.key || c.Op != want.op || c.Current { + t.Fatalf("got %+v, want %s %s, not current", c, want.op, want.key) + } + case <-time.After(10 * time.Second): + t.Fatalf("no %s of %s", want.op, want.key) + } + } + if _, err := s.Put("b.q", 6); err != nil { + t.Fatal(err) + } + for _, want := range []string{"a.x", "a.y"} { + select { + case c := <-narrow: + if c.Key != want { + t.Fatalf("narrowed watch got %+v, want %s", c, want) + } + case <-time.After(10 * time.Second): + t.Fatalf("narrowed watch missed %s", want) + } + } + select { + case c := <-narrow: + t.Fatalf("narrowed watch got %+v for a key it did not name", c) + case <-time.After(500 * time.Millisecond): + } +} + +// 1840/AC8: given an account not granted a bucket, when it watches it, then the refusal names the bucket. +func TestStateWatchRefusesAnUngrantedBucketNamingIt(t *testing.T) { + url, account := grantedSendProcess(t) + _, adminJetStream := adminJS(t, url) + if _, err := adminJetStream.CreateKeyValue(ctxFor(t), jetstream.KeyValueConfig{Bucket: "mesh-mail_unrelated"}); err != nil { + t.Fatalf("the admin could not make the bucket: %v", err) + } + conn, err := Connect(account) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + err = conn.State("unrelated").Watch("", func(stdio.StateChange) error { return nil }) + if err == nil || !errors.Is(err, nats.ErrPermissionViolation) || !strings.Contains(err.Error(), "mesh-mail_unrelated") { + t.Fatalf("Watch on an ungranted bucket: %v, want a refusal naming mesh-mail_unrelated", err) + } +} diff --git a/go/process/state_test.go b/go/process/state_test.go index ba02ad9..847abd2 100644 --- a/go/process/state_test.go +++ b/go/process/state_test.go @@ -7,6 +7,7 @@ import ( "strings" "sync" "testing" + "time" "github.com/nats-io/nats.go/jetstream" @@ -34,6 +35,9 @@ type fakeKV struct { values map[string][]byte revs map[string]uint64 next uint64 + // initial is what a Watch hands first, before its nil marker; watcher is the last one made. + initial []jetstream.KeyValueEntry + watcher *fakeWatcher } func newFakeKV() *fakeKV { return &fakeKV{values: map[string][]byte{}, revs: map[string]uint64{}} } @@ -92,6 +96,28 @@ func (f *fakeKV) Keys(_ context.Context, _ ...jetstream.WatchOpt) ([]string, err return keys, nil } +// fakeWatcher is a jetstream.KeyWatcher fed by a test. +type fakeWatcher struct { + jetstream.KeyWatcher + updates chan jetstream.KeyValueEntry + pattern string +} + +func (w *fakeWatcher) Updates() <-chan jetstream.KeyValueEntry { return w.updates } +func (w *fakeWatcher) Stop() error { return nil } + +func (f *fakeKV) Watch(_ context.Context, pattern string, _ ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) { + f.mu.Lock() + defer f.mu.Unlock() + w := &fakeWatcher{updates: make(chan jetstream.KeyValueEntry, 16), pattern: pattern} + f.watcher = w + for _, e := range f.initial { + w.updates <- e + } + w.updates <- nil + return w, nil +} + func stateOn(kv jetstream.KeyValue) State { return State{Module: "mesh-mail", Name: "sends", kv: func(context.Context) (jetstream.KeyValue, error) { return kv, nil }} } @@ -162,3 +188,92 @@ func TestStateKeysAreSortedAndEmptyWhenNone(t *testing.T) { t.Fatalf("Keys: %v, %v, want a,b,c", keys, err) } } + +type watchEntry struct { + fakeEntry + op jetstream.KeyValueOp +} + +func (e watchEntry) Operation() jetstream.KeyValueOp { return e.op } + +// 1840/S2: given a state with current values (one of them deleted), when it is watched, then the handler gets +// each current value marked Current and not the deleted one, Watch returns, and a later put and delete arrive +// in order unmarked. +func TestStateWatchHandsCurrentValuesThenChanges(t *testing.T) { + kv := newFakeKV() + kv.initial = []jetstream.KeyValueEntry{ + watchEntry{fakeEntry{key: "a", value: []byte(`1`), rev: 1}, jetstream.KeyValuePut}, + watchEntry{fakeEntry{key: "gone", rev: 2}, jetstream.KeyValueDelete}, + watchEntry{fakeEntry{key: "b", value: []byte(`2`), rev: 3}, jetstream.KeyValuePut}, + } + got := make(chan stdio.StateChange, 8) + s := stateOn(kv) + if err := s.Watch("", func(c stdio.StateChange) error { got <- c; return nil }); err != nil { + t.Fatal(err) + } + for _, want := range []stdio.StateChange{ + {State: "sends", Key: "a", Op: "put", Value: []byte(`1`), Revision: 1, Current: true}, + {State: "sends", Key: "b", Op: "put", Value: []byte(`2`), Revision: 3, Current: true}, + } { + select { + case c := <-got: + if c.Key != want.Key || c.Op != want.Op || string(c.Value) != string(want.Value) || c.Revision != want.Revision || c.Current != want.Current || c.State != want.State { + t.Fatalf("got %+v, want %+v", c, want) + } + default: + t.Fatalf("Watch returned before handling %+v", want) + } + } + if len(got) != 0 { + t.Fatalf("a deleted key was handed: %+v", <-got) + } + kv.watcher.updates <- watchEntry{fakeEntry{key: "a", value: []byte(`3`), rev: 4}, jetstream.KeyValuePut} + kv.watcher.updates <- watchEntry{fakeEntry{key: "b", rev: 5}, jetstream.KeyValueDelete} + for _, want := range []stdio.StateChange{ + {Key: "a", Op: "put", Revision: 4}, {Key: "b", Op: "delete", Revision: 5}, + } { + select { + case c := <-got: + if c.Key != want.Key || c.Op != want.Op || c.Revision != want.Revision || c.Current { + t.Fatalf("got %+v, want %+v", c, want) + } + case <-time.After(5 * time.Second): + t.Fatalf("no change %+v", want) + } + } +} + +// 1840/S2: a key narrowing reaches the bus as a subject pattern: `*` one name, `**` the rest, empty every key. +func TestStateWatchNarrowing(t *testing.T) { + for key, want := range map[string]string{"": ">", "a.b": "a.b", "a.*": "a.*", "a.**": "a.>", "**": ">"} { + if got := watchPattern(key); got != want { + t.Errorf("watchPattern(%q) = %q, want %q", key, got, want) + } + } +} + +// 1840/S2: given a handler that refuses a current value, then Watch returns its error. +func TestStateWatchReturnsAHandlersErrorOnCurrentValues(t *testing.T) { + kv := newFakeKV() + kv.initial = []jetstream.KeyValueEntry{watchEntry{fakeEntry{key: "a", value: []byte(`1`), rev: 1}, jetstream.KeyValuePut}} + boom := errors.New("boom") + if err := stateOn(kv).Watch("", func(stdio.StateChange) error { return boom }); !errors.Is(err, boom) { + t.Fatalf("Watch: %v, want the handler's error", err) + } +} + +// 1840/S2: given a connection that closes before the current values arrive, then Watch returns an error. +func TestStateWatchEndsWithAnErrorWhenTheConnectionCloses(t *testing.T) { + s := State{Module: "mesh-mail", Name: "sends", closed: func() bool { return true }, + kv: func(context.Context) (jetstream.KeyValue, error) { return silentKV{}, nil }} + if err := s.Watch("", func(stdio.StateChange) error { return nil }); err == nil { + t.Fatal("Watch on a closed connection returned no error") + } +} + +// silentKV watches but never delivers. +type silentKV struct{ jetstream.KeyValue } + +func (silentKV) Watch(context.Context, string, ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) { + return &fakeWatcher{updates: make(chan jetstream.KeyValueEntry)}, nil +} From 5ea6c215a48bbcd09c7c82874b4843b2307ab214 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 22 Oct 2026 14:36:24 +0200 Subject: [PATCH 4/7] issue 1840/S1: let a process delete a key, as a tools bundle can --- go/process/state.go | 14 ++++++++++++++ go/process/state_integration_test.go | 28 ++++++++++++++++++++++++++++ go/process/state_test.go | 25 +++++++++++++++++++++++++ 3 files changed, 67 insertions(+) diff --git a/go/process/state.go b/go/process/state.go index 1e79302..07d62b4 100644 --- a/go/process/state.go +++ b/go/process/state.go @@ -263,3 +263,17 @@ func (s State) Watch(key string, handle func(stdio.StateChange) error) error { }() return nil } + +// Delete removes a key. A key that was not there is not an error. +func (s State) Delete(key string) error { + ctx, cancel := context.WithTimeout(context.Background(), stateTimeout) + defer cancel() + kv, err := s.store(ctx) + if err != nil { + return err + } + if err := kv.Delete(ctx, key); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) { + return fmt.Errorf("%s's state %s, key %s: %w", s.Module, s.Name, key, err) + } + return nil +} diff --git a/go/process/state_integration_test.go b/go/process/state_integration_test.go index f0b1cfd..eca097a 100644 --- a/go/process/state_integration_test.go +++ b/go/process/state_integration_test.go @@ -234,3 +234,31 @@ func TestStateWatchRefusesAnUngrantedBucketNamingIt(t *testing.T) { t.Fatalf("Watch on an ungranted bucket: %v, want a refusal naming mesh-mail_unrelated", err) } } + +// 1840/S1: given a process connection granted its bucket, when it deletes a key, then the key is gone and +// deleting it again is no error. +func TestStateDeleteOnAGrantedBucket(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 { + t.Fatalf("the admin could not make the bucket: %v", err) + } + conn, err := Connect(account) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + s := conn.State("sends") + if _, err := s.Put("k", 1); err != nil { + t.Fatal(err) + } + if err := s.Delete("k"); err != nil { + t.Fatalf("Delete on a granted bucket: %v", err) + } + if e, err := s.Get("k"); err != nil || e != nil { + t.Fatalf("Get after Delete: %+v, %v", e, err) + } + if err := s.Delete("k"); err != nil { + t.Fatalf("Delete of an absent key: %v", err) + } +} diff --git a/go/process/state_test.go b/go/process/state_test.go index 847abd2..02b084a 100644 --- a/go/process/state_test.go +++ b/go/process/state_test.go @@ -82,6 +82,14 @@ func (f *fakeKV) Put(_ context.Context, key string, value []byte) (uint64, error return f.next, nil } +func (f *fakeKV) Delete(_ context.Context, key string, _ ...jetstream.KVDeleteOpt) error { + f.mu.Lock() + defer f.mu.Unlock() + delete(f.values, key) + delete(f.revs, key) + return nil +} + func (f *fakeKV) Keys(_ context.Context, _ ...jetstream.WatchOpt) ([]string, error) { f.mu.Lock() defer f.mu.Unlock() @@ -277,3 +285,20 @@ type silentKV struct{ jetstream.KeyValue } func (silentKV) Watch(context.Context, string, ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) { return &fakeWatcher{updates: make(chan jetstream.KeyValueEntry)}, nil } + +// 1840/S1: given a key, when it is deleted, then it reads as absent; deleting a key that was not there is no error. +func TestStateDeleteRemovesAKeyAndAnAbsentOneIsNoError(t *testing.T) { + s := stateOn(newFakeKV()) + if _, err := s.Put("k", 1); err != nil { + t.Fatal(err) + } + if err := s.Delete("k"); err != nil { + t.Fatalf("Delete of a key: %v", err) + } + if e, err := s.Get("k"); err != nil || e != nil { + t.Fatalf("Get after Delete: %+v, %v, want absent", e, err) + } + if err := s.Delete("never"); err != nil { + t.Fatalf("Delete of an absent key: %v", err) + } +} From c154bdedd315cfd2f53571e3b22e0ed3346d80a2 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 22 Oct 2026 15:03:17 +0200 Subject: [PATCH 5/7] Keep a machine that does not resolve on the private network at its place, so one machine's modules refuse only its own declaration (issue 1833) A machine whose own set could not be resolved was left off the private network; when it was the hub, the network could not be composed and every machine's declaration failed. Its placement (key and address) is its last good place and does not change when its modules fail, so it stays on the network and only its own declaration is refused. A refused machine with no place is left off, and a network that cannot be composed from what remains still fails. --- go/compose/network.go | 13 ++- go/compose/network_test.go | 11 ++- go/compose/one_machine_alone_test.go | 142 +++++++++++++++++++++++++++ 3 files changed, 161 insertions(+), 5 deletions(-) create mode 100644 go/compose/one_machine_alone_test.go diff --git a/go/compose/network.go b/go/compose/network.go index 1e66c51..0ea9d56 100644 --- a/go/compose/network.go +++ b/go/compose/network.go @@ -74,7 +74,10 @@ func (c *composer) overlayRange() string { } // network builds the private network over the machines that resolved the module for it. A machine is on it because it -// was given the module, or because it is joining with a key and an address (novox/hq ADR 0169). +// was given the module, or because it is joining with a key and an address (novox/hq ADR 0169), or because its own set +// could not be resolved and it holds a place (a key and an address): that place is its last good one on the network, and +// it does not change because its modules failed, so every other machine keeps its path to it -- the hub above all -- and +// only its own declaration is refused (issue 1833). A machine refused with no place is left off. func (c *composer) network(on map[string]bool, refused map[string]string) (*overlay.Generator, error) { tunnels := c.in.Tunnels carried := c.in.CarriedPeers @@ -85,9 +88,15 @@ func (c *composer) network(on map[string]bool, refused map[string]string) (*over var nodes []overlay.Node for _, node := range c.in.sorted() { p := node.Place - if !on[node.Name] && !(isJoining[node.Name] && p.Key != "" && p.Address != "") { + placed := p.Key != "" && p.Address != "" + _, wasRefused := refused[node.Name] + if !on[node.Name] && !((isJoining[node.Name] || wasRefused) && placed) { continue } + if !on[node.Name] && wasRefused && !isJoining[node.Name] { + c.say("%s could not be resolved and keeps its place on the private network at %s, so the others keep their path to it", + node.Name, p.Address) + } n := overlay.Node{ Name: node.Name, Key: p.Key, Endpoint: p.Endpoint, Site: p.Site, Hub: p.Hub, Address: p.Address, diff --git a/go/compose/network_test.go b/go/compose/network_test.go index 9f5717f..ff73e08 100644 --- a/go/compose/network_test.go +++ b/go/compose/network_test.go @@ -159,9 +159,11 @@ func TestANodeNotOnThePrivateNetworkGetsNoAddressAndNoNetwork(t *testing.T) { } // 1360/AC3: a node whose resolution is refused is said with today's words: the network missing its hub names the -// machine that could not be resolved, which is almost always why. +// machine that could not be resolved, which is almost always why. The hub here holds no complete place (no key): a refused +// machine with one keeps it on the network (issue 1833), so only one without is missing from it. func TestARefusedNodeIsSaidWhenTheNetworkCannotBeComposed(t *testing.T) { in := aMesh(t) + nodeOf(t, &in, "anchor").Place.Key = "" one := catalogue.Manifest{Module: "rival-one", Version: "1", Claims: []catalogue.Claim{{Name: "the-seat", Scope: catalogue.ScopeNode}}} two := catalogue.Manifest{Module: "rival-two", Version: "1", Claims: []catalogue.Claim{{Name: "the-seat", Scope: catalogue.ScopeNode}}} in.Catalogue["rival-one"], in.Catalogue["rival-two"] = one, two @@ -210,9 +212,10 @@ func TestEachFieldOfTheInputReachesTheGenerator(t *testing.T) { if len(base) != 2 { t.Fatalf("the base mesh has %v", base) } - t.Run("Catalogue: without the module nobody is on the network", func(t *testing.T) { + t.Run("Catalogue: without the module offering it nobody is on the network", func(t *testing.T) { in := aMesh(t) - in.Catalogue = map[string]catalogue.Manifest{} + // The module stays known, so every machine still resolves: a machine that does not resolve keeps its place (issue 1833). + in.Catalogue = map[string]catalogue.Manifest{overlay.Name: {Module: overlay.Name, Version: "1"}} if got := addressesOf(in); len(got) != 0 { t.Errorf("got %v", got) } @@ -290,6 +293,8 @@ func TestEachFieldOfTheInputReachesTheGenerator(t *testing.T) { } laptop := nodeOf(t, &in, "laptop") laptop.Assigned = append(laptop.Assigned, "holder-one", "holder-two") + // Without a key: a machine that does not resolve and holds a complete place keeps it on the network (issue 1833). + laptop.Place.Key = "" if got := addressesOf(in); got["laptop"] != "" { t.Fatalf("a machine whose set claims one seat twice is on the network: %v", got) } diff --git a/go/compose/one_machine_alone_test.go b/go/compose/one_machine_alone_test.go new file mode 100644 index 0000000..3fc89e5 --- /dev/null +++ b/go/compose/one_machine_alone_test.go @@ -0,0 +1,142 @@ +package compose + +import ( + "strings" + "testing" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/overlay" +) + +// fourOnTheNetwork is aMesh and two more machines on the private network at the site `there`: `ace`, dialled at +// ace.example:51820, at 10.77.0.3, and `box`, behind no endpoint, at 10.77.0.4. A module `office` requires +// `document-server`, which nothing in the catalogue provides: the shape of issue 1833 (nextcloud on the hub wanting a +// document server nobody offered). +func fourOnTheNetwork(t *testing.T) Input { + t.Helper() + in := aMesh(t) + in.Catalogue["office"] = catalogue.Manifest{Module: "office", Version: "1", Requires: []string{"document-server"}} + caps := map[string]bool{"container-runtime": true, "wireguard": true, "systemd": true} + in.Nodes = append(in.Nodes, + Node{Name: "ace", Place: Place{Key: keyOf(t, "ace"), Endpoint: "ace.example:51820", Site: "there", Address: "10.77.0.3"}, + Assigned: []string{overlay.Name}, Capabilities: caps}, + Node{Name: "box", Place: Place{Key: keyOf(t, "box"), Site: "there", Address: "10.77.0.4"}, + Assigned: []string{overlay.Name}, Capabilities: caps}) + return in +} + +// unresolvableBy makes one machine's own set fail to resolve: it is also given `office`, whose requirement nothing provides. +func unresolvableBy(t *testing.T, in *Input, name string) { + t.Helper() + n := nodeOf(t, in, name) + n.Assigned = append(n.Assigned, "office") +} + +// networkOf is the generator of an Input's private network, failing the test when it is not composed. +func networkOf(t *testing.T, in Input) *overlay.Generator { + t.Helper() + gens, err := Generators(in) + if err != nil { + t.Fatalf("the private network was not composed: %v", err) + } + return gens[overlay.Name].(*overlay.Generator) +} + +func placesOf(g *overlay.Generator) map[string]overlay.Node { + out := map[string]overlay.Node{} + for _, n := range g.Nodes() { + out[n.Name] = n + } + return out +} + +// 1833/AC1: four machines, the hub's own set does not resolve; the other three compose their declarations, and the hub alone is +// refused, with an error that is its own set failing and names what could not be resolved. +func TestOneMachineThatDoesNotResolveIsRefusedAlone(t *testing.T) { + in := fourOnTheNetwork(t) + unresolvableBy(t, &in, "anchor") + for _, node := range []string{"laptop", "ace", "box"} { + d, problems, err := Compose(in, node) + if err != nil || len(problems) != 0 { + t.Errorf("%s does not compose because another machine does not resolve: %v %v", node, err, problems) + continue + } + if !has(resourceIDs(d), "file:mesh-wireguard.overlay-config") { + t.Errorf("%s is given no private network: %v", node, resourceIDs(d)) + } + } + _, _, err := Compose(in, "anchor") + if err == nil || !IsUnresolvable(err) { + t.Fatalf("the machine that does not resolve is not refused as its own set failing: %v", err) + } + if !strings.Contains(err.Error(), `nothing provides "document-server", wanted by office`) { + t.Errorf("the refusal does not name what could not be resolved: %v", err) + } +} + +// 1833/AC2: the private network is composed with every machine on it: the hub that does not resolve keeps its place (key, +// address, endpoint, site, hub role), so every other machine keeps its path through it. +func TestAMachineThatDoesNotResolveKeepsItsPlaceOnTheNetwork(t *testing.T) { + in := fourOnTheNetwork(t) + unresolvableBy(t, &in, "anchor") + got := placesOf(networkOf(t, in)) + if len(got) != 4 { + t.Fatalf("the network is over %v, want all four machines", got) + } + hub := got["anchor"] + if !hub.Hub || hub.Key != keyOf(t, "anchor") || hub.Address != "10.77.0.1" || hub.Endpoint != "anchor.example:51820" || hub.Site != "here" { + t.Errorf("the hub did not keep its place: %+v", hub) + } + nodes, graph, err := Graph(in) + if err != nil || len(nodes) != 4 { + t.Fatalf("the graph is not shown with every machine: %v %v", nodes, err) + } + for _, node := range []string{"ace", "box"} { + through := false + for _, p := range graph[node] { + through = through || (p.Name == "anchor" && p.Endpoint == "anchor.example:51820") + } + if !through { + t.Errorf("%s has no path through the hub: %+v", node, graph[node]) + } + } +} + +// 1833/AC3: a machine that does not resolve and was never placed (a new machine: no key, no address) is left off the network, +// which is composed from the rest -- here with the hub that does not resolve on its last place -- and that machine alone is refused. +func TestANewMachineThatDoesNotResolveIsLeftOffAndRefusedAlone(t *testing.T) { + in := fourOnTheNetwork(t) + unresolvableBy(t, &in, "anchor") + in.Nodes = append(in.Nodes, Node{Name: "fresh", Place: Place{Site: "there"}, + Assigned: []string{overlay.Name, "office"}, Capabilities: map[string]bool{"wireguard": true, "systemd": true}}) + got := placesOf(networkOf(t, in)) + if _, on := got["fresh"]; on || len(got) != 4 { + t.Fatalf("the network is over %v, want the four placed machines without fresh", got) + } + if _, _, err := Compose(in, "fresh"); err == nil || !IsUnresolvable(err) { + t.Errorf("the new machine is not refused as its own set failing: %v", err) + } + if _, problems, err := Compose(in, "box"); err != nil || len(problems) != 0 { + t.Errorf("another machine does not compose: %v %v", err, problems) + } +} + +// 1833/AC4: a network that truly cannot be composed from what remains still fails: the hub does not resolve and has no +// complete place (no key), so nothing is the hub; the refusal is today's, naming the machine that could not be resolved. +func TestANetworkThatCannotBeComposedFromWhatRemainsStillFails(t *testing.T) { + in := fourOnTheNetwork(t) + unresolvableBy(t, &in, "anchor") + nodeOf(t, &in, "anchor").Place.Key = "" + _, err := Generators(in) + if err == nil { + t.Fatal("a network without a hub was composed") + } + for _, want := range []string{overlay.ErrNoHub.Error(), "and 1 node(s) could not be resolved at all, which is probably why:", " anchor: "} { + if !strings.Contains(err.Error(), want) { + t.Errorf("the refusal does not say %q:\n%v", want, err) + } + } + if _, _, err := Compose(in, "laptop"); err == nil { + t.Error("a declaration was composed over a network that cannot be") + } +} From bd7731f3588cf336620526b032bbfe9867f7837e Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 22 Oct 2026 15:26:48 +0200 Subject: [PATCH 6/7] Count a machine that does not resolve and holds a place as on the network for its name and the mesh's sources too (issue 1833) The tunnel kept such a machine while the names and the packet filter's sources from the mesh, read from their own unchecked resolve, dropped it when that pass refused it too (a seat claimed twice): every other machine was sent the hub as a peer without its name or its address. onTheNetwork keeps it by the same rule as network(). --- go/compose/network.go | 10 +++++++++- go/compose/node_test.go | 4 +++- go/compose/one_machine_alone_test.go | 25 +++++++++++++++++++++++++ 3 files changed, 37 insertions(+), 2 deletions(-) diff --git a/go/compose/network.go b/go/compose/network.go index 0ea9d56..afc8363 100644 --- a/go/compose/network.go +++ b/go/compose/network.go @@ -242,7 +242,9 @@ func (c *composer) whereEveryoneIs() (map[string]string, error) { } // onTheNetwork is every placed machine that resolves the private network: has an address AND runs what puts it there -// (novox/hq issue 079). A machine that does not resolve is not counted, and said. +// (novox/hq issue 079). A machine that does not resolve keeps its place when it holds one (a key and an address), by the +// same rule as network(), so its name and its address stay where the tunnel to it does (issue 1833); one without is not +// counted, and said. func (c *composer) onTheNetwork() []Node { var out []Node for _, n := range c.in.sorted() { @@ -256,6 +258,12 @@ func (c *composer) onTheNetwork() []Node { catalogue.Node{Name: n.Name, Site: n.Place.Site, Capabilities: n.Capabilities}, catalogue.World{Unchecked: true, Holdings: c.in.Holdings}) if err != nil { + if n.Place.Key != "" { + c.say("%s could not be resolved and keeps its place on the private network at %s, so the others keep their path to it", + n.Name, n.Place.Address) + out = append(out, n) + continue + } c.say("%s is not counted as on the network: it does not resolve: %v", n.Name, err) continue } diff --git a/go/compose/node_test.go b/go/compose/node_test.go index b7be07a..c2485dc 100644 --- a/go/compose/node_test.go +++ b/go/compose/node_test.go @@ -387,9 +387,11 @@ func TestAnotherNodesKeptBuildDoesNotChangeTheRestOfTheMeshOrWhoIsOnTheNetwork(t func overlayName() string { return overlay.Name } -// Round 1, finding 6: the nodes the composition did not count as on the network are returned, said, not lost. +// Round 1, finding 6: the nodes the composition did not count as on the network are returned, said, not lost. laptop holds no +// complete place (no key): one that does keeps it on the network (issue 1833) and is counted. func TestANodeNotCountedAsOnTheNetworkIsSaidInTheResult(t *testing.T) { in := aMesh(t) + nodeOf(t, &in, "laptop").Place.Key = "" in.Catalogue["rival-one"] = catalogue.Manifest{Module: "rival-one", Version: "1", Claims: []catalogue.Claim{{Name: "the-seat", Scope: catalogue.ScopeNode}}} in.Catalogue["rival-two"] = catalogue.Manifest{Module: "rival-two", Version: "1", Claims: []catalogue.Claim{{Name: "the-seat", Scope: catalogue.ScopeNode}}} laptop := nodeOf(t, &in, "laptop") diff --git a/go/compose/one_machine_alone_test.go b/go/compose/one_machine_alone_test.go index 3fc89e5..aa6b4c9 100644 --- a/go/compose/one_machine_alone_test.go +++ b/go/compose/one_machine_alone_test.go @@ -1,6 +1,7 @@ package compose import ( + "slices" "strings" "testing" @@ -140,3 +141,27 @@ func TestANetworkThatCannotBeComposedFromWhatRemainsStillFails(t *testing.T) { t.Error("a declaration was composed over a network that cannot be") } } + +// 1833/AC2, a failure the composition's unchecked pass refuses too (two modules claiming one seat on the hub, the 1360/AC3 +// shape): the hub keeps its place in what every other machine is sent besides the tunnel -- its internal name and its address +// among the sources "from the mesh" -- and another machine still composes. +func TestAMachineThatDoesNotResolveKeepsItsNameAndItsAddressInTheMesh(t *testing.T) { + in := fourOnTheNetwork(t) + in.Catalogue["rival-one"] = catalogue.Manifest{Module: "rival-one", Version: "1", Claims: []catalogue.Claim{{Name: "the-seat", Scope: catalogue.ScopeNode}}} + in.Catalogue["rival-two"] = catalogue.Manifest{Module: "rival-two", Version: "1", Claims: []catalogue.Claim{{Name: "the-seat", Scope: catalogue.ScopeNode}}} + hub := nodeOf(t, &in, "anchor") + hub.Assigned = append(hub.Assigned, "rival-one", "rival-two") + if _, err := Resolve(in, "anchor"); !IsUnresolvable(err) { + t.Fatalf("the hub is meant not to resolve: %v", err) + } + c := newComposer(in) + if got := c.namesInTheMesh()["anchor."+c.suffix()]; got != "10.77.0.1" { + t.Errorf("the hub's name is not in the mesh's names: %v", c.namesInTheMesh()) + } + if got := c.onThePrivateNetwork(); !slices.Contains(got, "10.77.0.1") || len(got) != 4 { + t.Errorf("the sources from the mesh are %v, want all four machines with the hub", got) + } + if _, problems, err := Compose(in, "ace"); err != nil || len(problems) != 0 { + t.Errorf("another machine does not compose: %v %v", err, problems) + } +} From f8b8f3a666d816071e7e30e2e1eeec8a29ab3d86 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 22 Oct 2026 17:45:58 +0200 Subject: [PATCH 7/7] issue 1840/S2: take a narrowed watch's current values as a set, since the bus hands them in the order of their last revision --- go/process/state_integration_test.go | 16 ++++++++++++++-- 1 file changed, 14 insertions(+), 2 deletions(-) diff --git a/go/process/state_integration_test.go b/go/process/state_integration_test.go index eca097a..32c1e72 100644 --- a/go/process/state_integration_test.go +++ b/go/process/state_integration_test.go @@ -180,6 +180,18 @@ func TestStateWatchOnAGrantedBucket(t *testing.T) { if len(narrow) != 2 { t.Fatalf("a narrowed watch handed %d current values, want 2", len(narrow)) } + // the current values come in the order of their last revision, not of their keys: taken as a set. + narrowCurrent := map[string]bool{} + for len(narrow) > 0 { + c := <-narrow + if !c.Current || c.Op != "put" { + t.Fatalf("a narrowed watch's current value arrived as %+v", c) + } + narrowCurrent[c.Key] = true + } + if !narrowCurrent["a.x"] || !narrowCurrent["a.y"] { + t.Fatalf("a narrowed watch's current values %v, want a.x and a.y", narrowCurrent) + } if _, err := s.Put("a.x", 5); err != nil { t.Fatal(err) @@ -203,8 +215,8 @@ func TestStateWatchOnAGrantedBucket(t *testing.T) { for _, want := range []string{"a.x", "a.y"} { select { case c := <-narrow: - if c.Key != want { - t.Fatalf("narrowed watch got %+v, want %s", c, want) + if c.Key != want || c.Current { + t.Fatalf("narrowed watch got %+v, want the change of %s", c, want) } case <-time.After(10 * time.Second): t.Fatalf("narrowed watch missed %s", want)