Compare commits
2
Commits
main
..
feat/l5-l7
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
676ce657f1 | ||
|
|
67635ef89f |
+1
-1
@@ -8,7 +8,7 @@ require (
|
|||||||
)
|
)
|
||||||
|
|
||||||
require (
|
require (
|
||||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261101235037-c21d8fd9692c
|
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261101023317-8311ecb84911
|
||||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op // indirect
|
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op // indirect
|
||||||
github.com/google/go-tpm v0.9.8 // indirect
|
github.com/google/go-tpm v0.9.8 // indirect
|
||||||
github.com/klauspost/compress v1.20.0 // indirect
|
github.com/klauspost/compress v1.20.0 // indirect
|
||||||
|
|||||||
+2
-4
@@ -1,7 +1,5 @@
|
|||||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261101085317-790d1c82d027 h1:a33dsavoaqllh40eVICbhxi6Kero1EGSjt1V+2fa2Hk=
|
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261101023317-8311ecb84911 h1:+82ygf2hiY5Mhwfmc1xfCe9e/KY/5ObBuE2LecM37qI=
|
||||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261101085317-790d1c82d027/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261101023317-8311ecb84911/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261101235037-c21d8fd9692c h1:/YCxh+NYfqry3PxZYY1nQXaccupobxpsFt8c9BYuGBg=
|
|
||||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261101235037-c21d8fd9692c/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
|
||||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
|
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
|
||||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
|
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
|
||||||
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
|
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
|
||||||
|
|||||||
@@ -37,6 +37,8 @@ import (
|
|||||||
// verb still runs, so a caller is never left guessing past this (novox/hq issue 265).
|
// verb still runs, so a caller is never left guessing past this (novox/hq issue 265).
|
||||||
const RequestTimeout = 30 * time.Second
|
const RequestTimeout = 30 * time.Second
|
||||||
|
|
||||||
|
const assignmentsStream = "ASSIGNMENTS"
|
||||||
|
|
||||||
// Credential is a broker credential as the mesh delivers it (novox/hq ADR 0120).
|
// Credential is a broker credential as the mesh delivers it (novox/hq ADR 0120).
|
||||||
type Credential struct {
|
type Credential struct {
|
||||||
URL string `json:"url"`
|
URL string `json:"url"`
|
||||||
@@ -181,9 +183,12 @@ type Conn struct {
|
|||||||
cred Credential
|
cred Credential
|
||||||
mu sync.Mutex
|
mu sync.Mutex
|
||||||
issued map[string]*Membership // module → membership; present with nil = followed, none issued
|
issued map[string]*Membership // module → membership; present with nil = followed, none issued
|
||||||
onNew []*func(Membership)
|
// fromGrants is every module whose membership the grants part's state memberships has given (novox/hq ADR 0571
|
||||||
subs []*nats.Subscription
|
// clause 7): its assignments stream copy is then a fallback no longer applied.
|
||||||
Logf func(format string, args ...any)
|
fromGrants map[string]bool
|
||||||
|
onNew []*func(Membership)
|
||||||
|
subs []*nats.Subscription
|
||||||
|
Logf func(format string, args ...any)
|
||||||
// answering is every subject this runtime answers on, so a subscription the bus refused can be
|
// answering is every subject this runtime answers on, so a subscription the bus refused can be
|
||||||
// asked for again (novox/hq issue 222).
|
// asked for again (novox/hq issue 222).
|
||||||
answering map[string]*answered
|
answering map[string]*answered
|
||||||
@@ -327,7 +332,7 @@ func Connect(cred Credential) (*Conn, error) {
|
|||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
c = &Conn{nc: nc, js: js, self: cred.Module, node: cred.Node, cred: cred,
|
c = &Conn{nc: nc, js: js, self: cred.Module, node: cred.Node, cred: cred,
|
||||||
issued: map[string]*Membership{}, Logf: log.Printf, answering: map[string]*answered{}}
|
issued: map[string]*Membership{}, fromGrants: map[string]bool{}, Logf: log.Printf, answering: map[string]*answered{}}
|
||||||
c.Follow(cred.Module)
|
c.Follow(cred.Module)
|
||||||
return c, nil
|
return c, nil
|
||||||
}
|
}
|
||||||
@@ -341,10 +346,16 @@ func (c *Conn) Node() string { return c.node }
|
|||||||
// Credential is what this connection was opened with.
|
// Credential is what this connection was opened with.
|
||||||
func (c *Conn) Credential() Credential { return c.cred }
|
func (c *Conn) Credential() Credential { return c.cred }
|
||||||
|
|
||||||
|
// MembershipSubject is the one address a runtime derives for an assignment (ADR 0160).
|
||||||
|
func MembershipSubject(node, module string) string {
|
||||||
|
return "mesh.assignment." + node + "." + module
|
||||||
|
}
|
||||||
|
|
||||||
// Follow reads a module's membership on this machine once and follows it live, so its tools are
|
// Follow reads a module's membership on this machine once and follows it live, so its tools are
|
||||||
// served where the mesh issued them (ADR 0175). It is read from the grants part's state memberships, its key
|
// served where the mesh issued them (ADR 0175). It is read from the grants part's state memberships, its key
|
||||||
// `<node>.<module>` (novox/hq ADR 0571 clause 7 and 10, mesh-sdk buscompose.GrantsMembershipsBucket), and nowhere else:
|
// `<node>.<module>` (novox/hq ADR 0571 clause 7, mesh-sdk buscompose.GrantsMembershipsBucket); the controller's
|
||||||
// the controller's assignments stream is not read and its subject is not heard.
|
// assignments stream is read beside it as the fallback until the lease write, and applied only while the state has given
|
||||||
|
// the module no membership.
|
||||||
func (c *Conn) Follow(module string) {
|
func (c *Conn) Follow(module string) {
|
||||||
c.mu.Lock()
|
c.mu.Lock()
|
||||||
if c.node == "" {
|
if c.node == "" {
|
||||||
@@ -358,24 +369,53 @@ func (c *Conn) Follow(module string) {
|
|||||||
c.issued[module] = nil
|
c.issued[module] = nil
|
||||||
c.mu.Unlock()
|
c.mu.Unlock()
|
||||||
c.followGrants(module)
|
c.followGrants(module)
|
||||||
|
subject := MembershipSubject(c.node, module)
|
||||||
|
// The subject-addressed direct get: the one address the mesh grants this account on the
|
||||||
|
// stream's API.
|
||||||
|
if c.Membership(module) == nil {
|
||||||
|
if got, err := c.nc.Request("$JS.API.DIRECT.GET."+assignmentsStream+"."+subject, nil, 5*time.Second); err == nil {
|
||||||
|
if got.Header.Get("Status") == "" && len(got.Data) > 0 {
|
||||||
|
var m Membership
|
||||||
|
if json.Unmarshal(got.Data, &m) == nil {
|
||||||
|
c.mu.Lock()
|
||||||
|
if !c.fromGrants[module] {
|
||||||
|
c.issued[module] = &m
|
||||||
|
}
|
||||||
|
c.mu.Unlock()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
if c.Membership(module) == nil {
|
if c.Membership(module) == nil {
|
||||||
c.Logf("[mesh-tools] no membership issued for %s on %s yet; serving the derived shape until one arrives", module, c.node)
|
c.Logf("[mesh-tools] no membership issued for %s on %s yet; serving the derived shape until one arrives", module, c.node)
|
||||||
}
|
}
|
||||||
|
sub, err := c.nc.Subscribe(subject, func(msg *nats.Msg) {
|
||||||
|
var m Membership
|
||||||
|
if err := json.Unmarshal(msg.Data, &m); err != nil {
|
||||||
|
c.Logf("[mesh-tools] a membership arrived that is not one: %v", err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
c.issue(module, m, false)
|
||||||
|
})
|
||||||
|
if err == nil {
|
||||||
|
c.track(sub)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// GrantsPatience bounds the first read of a membership from the grants part's state.
|
// GrantsPatience bounds the first read of a membership from the grants part's state.
|
||||||
var GrantsPatience = 2 * time.Second
|
var GrantsPatience = 2 * time.Second
|
||||||
|
|
||||||
// GrantsRetry is the first wait before the memberships state is read again after it could not be; it doubles up to
|
// issue applies a membership that arrived for module, live, and re-serves on it. One from the assignments stream is
|
||||||
// GrantsRetryMost.
|
// passed over once the grants part's state has given the module one.
|
||||||
var (
|
func (c *Conn) issue(module string, m Membership, fromGrants bool) {
|
||||||
GrantsRetry = time.Second
|
|
||||||
GrantsRetryMost = 30 * time.Second
|
|
||||||
)
|
|
||||||
|
|
||||||
// issue applies a membership that arrived for module, live, and re-serves on it.
|
|
||||||
func (c *Conn) issue(module string, m Membership) {
|
|
||||||
c.mu.Lock()
|
c.mu.Lock()
|
||||||
|
if !fromGrants && c.fromGrants[module] {
|
||||||
|
c.mu.Unlock()
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if fromGrants {
|
||||||
|
c.fromGrants[module] = true
|
||||||
|
}
|
||||||
c.issued[module] = &m
|
c.issued[module] = &m
|
||||||
handlers := make([]func(Membership), 0, len(c.onNew))
|
handlers := make([]func(Membership), 0, len(c.onNew))
|
||||||
for _, h := range c.onNew {
|
for _, h := range c.onNew {
|
||||||
@@ -390,38 +430,9 @@ func (c *Conn) issue(module string, m Membership) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// followGrants reads module's key of the grants part's state memberships and watches it: the value there now is applied
|
// followGrants reads module's key of the grants part's state memberships and watches it: the value there now is applied
|
||||||
// before Follow returns, each change after it live. A state that is not there yet, or not readable (a bus answering 503
|
// before Follow returns, each change after it live. A state that is not there yet, or not readable, is said once, and
|
||||||
// right after its boot, a timeout), is said once and read again, with a growing wait, until it can be or the
|
// the assignments stream answers alone.
|
||||||
// connection closes: Follow runs once per module, so nothing else would ever read it again.
|
|
||||||
func (c *Conn) followGrants(module string) {
|
func (c *Conn) followGrants(module string) {
|
||||||
if c.watchGrants(module, true) {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
go func() {
|
|
||||||
wait := GrantsRetry
|
|
||||||
for !c.nc.IsClosed() {
|
|
||||||
time.Sleep(wait)
|
|
||||||
if c.nc.IsClosed() {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
if c.watchGrants(module, false) {
|
|
||||||
c.Logf("[mesh-tools] the grants part's memberships are read again; %s on %s follows its key", module, c.node)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
if wait *= 2; wait > GrantsRetryMost {
|
|
||||||
wait = GrantsRetryMost
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
}
|
|
||||||
|
|
||||||
// watchGrants is one attempt of followGrants: true once the key is watched. A failure is said when say is true.
|
|
||||||
func (c *Conn) watchGrants(module string, say bool) bool {
|
|
||||||
logf := func(format string, a ...any) {
|
|
||||||
if say {
|
|
||||||
c.Logf(format, a...)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
key := buscompose.MembershipKey(c.node, module)
|
key := buscompose.MembershipKey(c.node, module)
|
||||||
// First the key's direct get, which the account is granted alone: a bus that refuses it, or a state not there yet,
|
// First the key's direct get, which the account is granted alone: a bus that refuses it, or a state not there yet,
|
||||||
// answers at once, where binding the bucket would wait out its bound on a refusal.
|
// answers at once, where binding the bucket would wait out its bound on a refusal.
|
||||||
@@ -432,27 +443,27 @@ func (c *Conn) watchGrants(module string, say bool) bool {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
why = err.Error()
|
why = err.Error()
|
||||||
}
|
}
|
||||||
logf("[mesh-tools] the grants part's memberships (%s) cannot be read (%s); %s has no membership "+
|
c.Logf("[mesh-tools] the grants part's memberships (%s) cannot be read (%s); %s's membership is read from the "+
|
||||||
"until it can be", bucket, why, module)
|
"assignments stream alone", bucket, why, module)
|
||||||
return false
|
return
|
||||||
}
|
}
|
||||||
js, err := jetstream.New(c.nc)
|
js, err := jetstream.New(c.nc)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return false
|
return
|
||||||
}
|
}
|
||||||
ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
|
ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
|
||||||
defer cancel()
|
defer cancel()
|
||||||
kv, err := js.KeyValue(ctx, bucket)
|
kv, err := js.KeyValue(ctx, bucket)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
logf("[mesh-tools] the grants part's memberships (%s) cannot be read (%v); %s has no membership "+
|
c.Logf("[mesh-tools] the grants part's memberships (%s) cannot be read (%v); %s's membership is read from the "+
|
||||||
"until it can be", buscompose.GrantsMembershipsBucket, err, module)
|
"assignments stream alone", buscompose.GrantsMembershipsBucket, err, module)
|
||||||
return false
|
return
|
||||||
}
|
}
|
||||||
w, err := kv.Watch(context.Background(), key)
|
w, err := kv.Watch(context.Background(), key)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
logf("[mesh-tools] %s's key %s of the grants part's memberships cannot be watched (%v); %s has no membership "+
|
c.Logf("[mesh-tools] %s's key %s of the grants part's memberships cannot be watched (%v); read from the "+
|
||||||
"until it can be", module, key, err, module)
|
"assignments stream alone", module, key, err)
|
||||||
return false
|
return
|
||||||
}
|
}
|
||||||
apply := func(e jetstream.KeyValueEntry) {
|
apply := func(e jetstream.KeyValueEntry) {
|
||||||
if e.Operation() != jetstream.KeyValuePut {
|
if e.Operation() != jetstream.KeyValuePut {
|
||||||
@@ -460,7 +471,7 @@ func (c *Conn) watchGrants(module string, say bool) bool {
|
|||||||
}
|
}
|
||||||
// Read through mesh-sdk's one codec of this state, the grants part's writer's (issue 774): it takes the bare
|
// Read through mesh-sdk's one codec of this state, the grants part's writer's (issue 774): it takes the bare
|
||||||
// membership and a wrapper the part wrote before of one published or unchanged, and refuses a held-back one's
|
// membership and a wrapper the part wrote before of one published or unchanged, and refuses a held-back one's
|
||||||
// wrapper or a value naming no node or module, which then leaves the module with no membership.
|
// wrapper or a value naming no node or module, which then leaves the assignments stream's copy in force instead.
|
||||||
// A refused value arriving after a good one keeps the last good membership: nothing here clears it.
|
// A refused value arriving after a good one keeps the last good membership: nothing here clears it.
|
||||||
var m Membership
|
var m Membership
|
||||||
body, err := membership.BareGrantsMembership(e.Value())
|
body, err := membership.BareGrantsMembership(e.Value())
|
||||||
@@ -471,7 +482,7 @@ func (c *Conn) watchGrants(module string, say bool) bool {
|
|||||||
c.Logf("[mesh-tools] the membership at %s is not one: %v", key, err)
|
c.Logf("[mesh-tools] the membership at %s is not one: %v", key, err)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
c.issue(module, m)
|
c.issue(module, m, true)
|
||||||
}
|
}
|
||||||
timer := time.NewTimer(StateTimeout)
|
timer := time.NewTimer(StateTimeout)
|
||||||
defer timer.Stop()
|
defer timer.Stop()
|
||||||
@@ -479,7 +490,7 @@ func (c *Conn) watchGrants(module string, say bool) bool {
|
|||||||
select {
|
select {
|
||||||
case e, ok := <-w.Updates():
|
case e, ok := <-w.Updates():
|
||||||
if !ok {
|
if !ok {
|
||||||
return false
|
return
|
||||||
}
|
}
|
||||||
if e == nil {
|
if e == nil {
|
||||||
current = false
|
current = false
|
||||||
@@ -488,9 +499,9 @@ func (c *Conn) watchGrants(module string, say bool) bool {
|
|||||||
apply(e)
|
apply(e)
|
||||||
case <-timer.C:
|
case <-timer.C:
|
||||||
_ = w.Stop()
|
_ = w.Stop()
|
||||||
logf("[mesh-tools] %s's key %s of the grants part's memberships did not answer in %s; %s has no membership "+
|
c.Logf("[mesh-tools] %s's key %s of the grants part's memberships did not answer in %s; read from the "+
|
||||||
"until it can be", module, key, StateTimeout, module)
|
"assignments stream alone", module, key, StateTimeout)
|
||||||
return false
|
return
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
go func() {
|
go func() {
|
||||||
@@ -500,7 +511,6 @@ func (c *Conn) watchGrants(module string, say bool) bool {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
return true
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// Membership is what the mesh issued a module here, or nil when nothing has been issued.
|
// Membership is what the mesh issued a module here, or nil when nothing has been issued.
|
||||||
|
|||||||
@@ -119,24 +119,25 @@ func TestAWatchIsRefusedWhenNoCurrentValueIsTakenWithinTheBound(t *testing.T) {
|
|||||||
|
|
||||||
// **A membership the grants part wrote in its old wrapper is served whole** (issue 774). Until 2026-10-30 the part wrote
|
// **A membership the grants part wrote in its old wrapper is served whole** (issue 774). Until 2026-10-30 the part wrote
|
||||||
// its pass record, the membership inside, into its state memberships; the tool runner read the value as a bare
|
// its pass record, the membership inside, into its state memberships; the tool runner read the value as a bare
|
||||||
// membership, and served no tool and no state. This is what the live mesh held: read through mesh-sdk's codec, its tools
|
// membership, took it over the assignments stream's copy, and served no tool and no state. This is what the live mesh
|
||||||
// are answered where the membership says and its state is reached. A held-back one's wrapper is refused, and the module has
|
// held: read through mesh-sdk's codec, its tools are answered where the membership says and its state is reached.
|
||||||
// no membership then (774/S32/AC2: no stream is read in its place).
|
|
||||||
func TestAMembershipTheGrantsPartWroteWrappedIsServed(t *testing.T) {
|
func TestAMembershipTheGrantsPartWroteWrappedIsServed(t *testing.T) {
|
||||||
mesh := mt.New(t)
|
mesh := mt.New(t)
|
||||||
mesh.Bucket(t, "gitea_repos")
|
mesh.Bucket(t, "gitea_repos")
|
||||||
gitea := mt.MembershipOf("gitea", "anchor", false, nil)
|
gitea := mt.MembershipOf("gitea", "anchor", false, nil)
|
||||||
gitea.State = []bus.StateIssued{{Name: "repos", Bucket: "gitea_repos", Writes: true}}
|
gitea.State = []bus.StateIssued{{Name: "repos", Bucket: "gitea_repos", Writes: true}}
|
||||||
mesh.IssueInGrantsWrapped(t, gitea, "published")
|
mesh.IssueInGrantsWrapped(t, gitea, "published")
|
||||||
// The old wrapper of a membership the part held back carries its unpublished composition: refused, never served.
|
// The old wrapper of a membership the part held back carries its unpublished composition: refused, so the
|
||||||
|
// assignments stream's copy is served, never what the hold kept back.
|
||||||
|
pages := mt.MembershipOf("pages", "anchor", false, nil)
|
||||||
|
mesh.Issue(t, pages)
|
||||||
heldBack := mt.MembershipOf("pages", "anchor", true, nil)
|
heldBack := mt.MembershipOf("pages", "anchor", true, nil)
|
||||||
mesh.IssueInGrantsWrapped(t, heldBack, "not published, a shadow difference: publishing would lose the last holder of x")
|
mesh.IssueInGrantsWrapped(t, heldBack, "not published, a shadow difference: publishing would lose the last holder of x")
|
||||||
conn, err := bus.Connect(bus.Credential{URL: mt.URL(t), Module: "node-tools", Node: "anchor"})
|
conn, err := bus.Connect(bus.Credential{URL: mt.URL(t), Module: "node-tools", Node: "anchor"})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
var logged []string
|
conn.Logf = func(string, ...any) {}
|
||||||
conn.Logf = func(f string, a ...any) { logged = append(logged, fmt.Sprintf(f, a...)) }
|
|
||||||
t.Cleanup(conn.Close)
|
t.Cleanup(conn.Close)
|
||||||
conn.Follow("gitea")
|
conn.Follow("gitea")
|
||||||
mt.Until(t, func() error {
|
mt.Until(t, func() error {
|
||||||
@@ -155,9 +156,14 @@ func TestAMembershipTheGrantsPartWroteWrappedIsServed(t *testing.T) {
|
|||||||
t.Errorf("the state reads back %+v, %v", e, err)
|
t.Errorf("the state reads back %+v, %v", e, err)
|
||||||
}
|
}
|
||||||
conn.Follow("pages")
|
conn.Follow("pages")
|
||||||
time.Sleep(300 * time.Millisecond)
|
mt.Until(t, func() error {
|
||||||
if m := conn.Membership("pages"); m != nil || !strings.Contains(strings.Join(logged, "\n"), "no membership issued for pages") {
|
if m := conn.Membership("pages"); m == nil || len(m.Serves) == 0 {
|
||||||
t.Errorf("a held-back membership's wrapper was served or not said: %+v, %q", m, logged)
|
return fmt.Errorf("pages' membership is %+v, want the assignments stream's", m)
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
})
|
||||||
|
if got, want := fmt.Sprint(conn.ServedOn("pages", "pages_list")), "[{mesh.mod.pages.tool.pages_list.anchor }]"; got != want {
|
||||||
|
t.Errorf("a held-back membership's wrapper is served: pages_list on %s, want the stream's %s", got, want)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -161,11 +161,11 @@ func TestAModuleHearsEveryStreamItsMembershipNames(t *testing.T) {
|
|||||||
return nil
|
return nil
|
||||||
})
|
})
|
||||||
|
|
||||||
// The membership as the grants part keeps it (ADR 0571 clause 7, 10): read first and applied live, and nothing
|
// The membership as the grants part keeps it (ADR 0571 clause 7): read first, applied live, and the assignments
|
||||||
// published on the controller's subject is heard.
|
// stream's copy passed over once the state has given one.
|
||||||
granted := reader
|
granted := reader
|
||||||
granted.Hears = []bus.HeardIssued{{Stream: "MAIL", Consumer: "from-grants"}}
|
granted.Hears = []bus.HeardIssued{{Stream: "MAIL", Consumer: "from-grants"}}
|
||||||
mesh.Issue(t, granted)
|
mesh.IssueInGrants(t, granted)
|
||||||
again, err := bus.Connect(bus.Credential{URL: mt.URL(t), Module: "node-tools", Node: "anchor"})
|
again, err := bus.Connect(bus.Credential{URL: mt.URL(t), Module: "node-tools", Node: "anchor"})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
@@ -183,45 +183,46 @@ func TestAModuleHearsEveryStreamItsMembershipNames(t *testing.T) {
|
|||||||
t.Fatalf("the membership read at the follow names %q, want the grants part's", got)
|
t.Fatalf("the membership read at the follow names %q, want the grants part's", got)
|
||||||
}
|
}
|
||||||
granted.Hears[0].Consumer = "from-grants-2"
|
granted.Hears[0].Consumer = "from-grants-2"
|
||||||
mesh.Issue(t, granted)
|
mesh.IssueInGrants(t, granted)
|
||||||
mt.Until(t, func() error {
|
mt.Until(t, func() error {
|
||||||
if got := consumerOf(); got != "from-grants-2" {
|
if got := consumerOf(); got != "from-grants-2" {
|
||||||
return fmt.Errorf("the membership names %q, want the grants part's change", got)
|
return fmt.Errorf("the membership names %q, want the grants part's change", got)
|
||||||
}
|
}
|
||||||
return nil
|
return nil
|
||||||
})
|
})
|
||||||
// 774/S32/AC1: a copy said on the controller's subject (mesh.assignment) changes nothing; there is no stream to read it from.
|
// A later, different copy on the assignments stream changes nothing once the state has given one.
|
||||||
streamed := reader
|
arrived := make(chan struct{}, 1)
|
||||||
streamed.Hears = []bus.HeardIssued{{Stream: "MAIL", Consumer: "from-subject"}}
|
probe, err := bus.Connect(bus.Credential{URL: mt.URL(t), Module: "probe", Node: "elsewhere"})
|
||||||
mesh.Publish(t, "mesh.assignment.anchor.reader", streamed)
|
|
||||||
again.Flush()
|
|
||||||
time.Sleep(200 * time.Millisecond)
|
|
||||||
if got := consumerOf(); got != "from-grants-2" {
|
|
||||||
t.Fatalf("a membership said on the controller's subject was applied: the membership names %q", got)
|
|
||||||
}
|
|
||||||
// Review g1 round 1: memberships that cannot be read at the follow (a bus not ready after its boot) are read again until
|
|
||||||
// they can be; the follow, which runs once per module, does not give the module up for good.
|
|
||||||
was := bus.GrantsRetry
|
|
||||||
bus.GrantsRetry = 50 * time.Millisecond
|
|
||||||
t.Cleanup(func() { bus.GrantsRetry = was })
|
|
||||||
mesh.DropGrants(t)
|
|
||||||
late, err := bus.Connect(bus.Credential{URL: mt.URL(t), Module: "node-tools", Node: "anchor"})
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
late.Logf = func(string, ...any) {}
|
probe.Logf = func(string, ...any) {}
|
||||||
t.Cleanup(late.Close)
|
t.Cleanup(probe.Close)
|
||||||
late.Follow("reader")
|
stopProbe, err := probe.Raw(bus.MembershipSubject("anchor", "reader"), func(string, []byte) []byte {
|
||||||
if m := late.Membership("reader"); m != nil {
|
select {
|
||||||
t.Fatalf("a membership was served from a state not there: %+v", m)
|
case arrived <- struct{}{}:
|
||||||
}
|
default:
|
||||||
mesh.Issue(t, granted)
|
|
||||||
mt.Until(t, func() error {
|
|
||||||
if m := late.Membership("reader"); m == nil || len(m.Hears) != 1 || m.Hears[0].Consumer != "from-grants-2" {
|
|
||||||
return fmt.Errorf("the membership is %+v, want the one written once the state is there", m)
|
|
||||||
}
|
}
|
||||||
return nil
|
return nil
|
||||||
})
|
})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
t.Cleanup(stopProbe)
|
||||||
|
probe.Flush()
|
||||||
|
streamed := reader
|
||||||
|
streamed.Hears = []bus.HeardIssued{{Stream: "MAIL", Consumer: "from-stream"}}
|
||||||
|
mesh.Issue(t, streamed)
|
||||||
|
select {
|
||||||
|
case <-arrived:
|
||||||
|
case <-time.After(5 * time.Second):
|
||||||
|
t.Fatal("the stream's copy was not published")
|
||||||
|
}
|
||||||
|
again.Flush()
|
||||||
|
time.Sleep(200 * time.Millisecond)
|
||||||
|
if got := consumerOf(); got != "from-grants-2" {
|
||||||
|
t.Fatalf("the stream's copy was applied over the grants part's: the membership names %q", got)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// novox/hq ADR 0363 §6 (issue 725): a consumer the membership names before the controller made it is bound once
|
// novox/hq ADR 0363 §6 (issue 725): a consumer the membership names before the controller made it is bound once
|
||||||
|
|||||||
@@ -86,10 +86,11 @@ func New(t *testing.T) *Mesh {
|
|||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
js, _ := nc.JetStream()
|
js, _ := nc.JetStream()
|
||||||
_ = js.DeleteStream("EVENTS")
|
for _, s := range []string{"ASSIGNMENTS", "EVENTS"} {
|
||||||
// The grants part's state memberships, where a runtime reads what is issued it (ADR 0571 clause 10).
|
_ = js.DeleteStream(s)
|
||||||
_ = js.DeleteKeyValue(buscompose.GrantsMembershipsBucket)
|
}
|
||||||
if _, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: buscompose.GrantsMembershipsBucket, History: 1}); err != nil {
|
if _, err := js.AddStream(&nats.StreamConfig{Name: "ASSIGNMENTS", Subjects: []string{"mesh.assignment.>"},
|
||||||
|
MaxMsgsPerSubject: 1, AllowDirect: true}); err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
if _, err := js.AddStream(&nats.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.mod.*.event.>"}}); err != nil {
|
if _, err := js.AddStream(&nats.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.mod.*.event.>"}}); err != nil {
|
||||||
@@ -99,11 +100,20 @@ func New(t *testing.T) *Mesh {
|
|||||||
return &Mesh{nc: nc, js: js}
|
return &Mesh{nc: nc, js: js}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Issue writes a membership where the grants part keeps it, its state memberships, key `<node>.<module>` (novox/hq ADR
|
// Issue publishes a membership.
|
||||||
// 0571 clause 7 and 10; mesh-sdk buscompose.GrantsMembershipsBucket), the one place a runtime reads it. It writes with
|
|
||||||
// mesh-sdk's encoder, the grants part's own, never json of this package's type: a value built here by hand agreed with
|
|
||||||
// the reader's mistake and hid that the part wrote another shape (issue 774).
|
|
||||||
func (m *Mesh) Issue(t *testing.T, mem bus.Membership) {
|
func (m *Mesh) Issue(t *testing.T, mem bus.Membership) {
|
||||||
|
t.Helper()
|
||||||
|
body, _ := json.Marshal(mem)
|
||||||
|
if _, err := m.js.Publish(bus.MembershipSubject(mem.Node, mem.Module), body); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// IssueInGrants writes a membership where the grants part keeps it, its state memberships, key `<node>.<module>`
|
||||||
|
// (novox/hq ADR 0571 clause 7; mesh-sdk buscompose.GrantsMembershipsBucket), making the bucket when it is not there.
|
||||||
|
// It writes with mesh-sdk's encoder, the grants part's own, never json of this package's type: a value built here by
|
||||||
|
// hand agreed with the reader's mistake and hid that the part wrote another shape (issue 774).
|
||||||
|
func (m *Mesh) IssueInGrants(t *testing.T, mem bus.Membership) {
|
||||||
t.Helper()
|
t.Helper()
|
||||||
m.putGrants(t, mem, membership.EncodeGrantsMembership(sdkMembership(t, mem)))
|
m.putGrants(t, mem, membership.EncodeGrantsMembership(sdkMembership(t, mem)))
|
||||||
}
|
}
|
||||||
@@ -131,14 +141,6 @@ func (m *Mesh) putGrants(t *testing.T, mem bus.Membership, body []byte) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// DropGrants deletes the grants part's state memberships, as a bus that has not made it yet holds none; Issue makes it again.
|
|
||||||
func (m *Mesh) DropGrants(t *testing.T) {
|
|
||||||
t.Helper()
|
|
||||||
if err := m.js.DeleteKeyValue(buscompose.GrantsMembershipsBucket); err != nil {
|
|
||||||
t.Fatal(err)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// sdkMembership is a membership of this package's type as mesh-sdk's, which the grants part composes and writes.
|
// sdkMembership is a membership of this package's type as mesh-sdk's, which the grants part composes and writes.
|
||||||
func sdkMembership(t *testing.T, mem bus.Membership) membership.Membership {
|
func sdkMembership(t *testing.T, mem bus.Membership) membership.Membership {
|
||||||
t.Helper()
|
t.Helper()
|
||||||
@@ -317,18 +319,6 @@ func (m *Mesh) Emit(t *testing.T, subject string, body any) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Publish says body on a plain subject, as a core publish with no stream behind it.
|
|
||||||
func (m *Mesh) Publish(t *testing.T, subject string, body any) {
|
|
||||||
t.Helper()
|
|
||||||
data, _ := json.Marshal(body)
|
|
||||||
if err := m.nc.Publish(subject, data); err != nil {
|
|
||||||
t.Fatal(err)
|
|
||||||
}
|
|
||||||
if err := m.nc.Flush(); err != nil {
|
|
||||||
t.Fatal(err)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// Pending is what a module's consumer still holds: delivered and not acknowledged, and not yet
|
// Pending is what a module's consumer still holds: delivered and not acknowledged, and not yet
|
||||||
// delivered.
|
// delivered.
|
||||||
func (m *Mesh) Pending(t *testing.T, node, module string) (ackPending, notDelivered uint64) {
|
func (m *Mesh) Pending(t *testing.T, node, module string) (ackPending, notDelivered uint64) {
|
||||||
|
|||||||
@@ -194,19 +194,3 @@ func TestOnlyTheMachinesRuntimeStartsTheHelper(t *testing.T) {
|
|||||||
Start(nil, c.module, c.node, time.Second, said.Logf)()
|
Start(nil, c.module, c.node, time.Second, said.Logf)()
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// 774/S33 (ADR 0571 clause 13): this manifest, read as the lease write's check reads it, is not refused at the write: it
|
|
||||||
// reads and invokes nothing of the controller and consumes nothing of it.
|
|
||||||
func TestTheManifestIsNotRefusedAtTheLeaseWrite(t *testing.T) {
|
|
||||||
raw, err := os.ReadFile("../../module.json")
|
|
||||||
if err != nil {
|
|
||||||
t.Fatal(err)
|
|
||||||
}
|
|
||||||
m, err := catalogue.Decode(raw)
|
|
||||||
if err != nil {
|
|
||||||
t.Fatal(err)
|
|
||||||
}
|
|
||||||
if refused := catalogue.ManifestRefusals(map[string]catalogue.Manifest{Module: m}, catalogue.PhaseWrite); len(refused) != 0 {
|
|
||||||
t.Errorf("the lease write's check refuses node-tools' manifest: %v", refused)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|||||||
Reference in New Issue
Block a user