diff --git a/go/buscompose/buscompose_test.go b/go/buscompose/buscompose_test.go index 112ee09..1075f8a 100644 --- a/go/buscompose/buscompose_test.go +++ b/go/buscompose/buscompose_test.go @@ -161,8 +161,11 @@ func grantsAsTheControllers(t *testing.T, got, want string) { intended := beyondTheControllers(user) for g := range was { // A runtime's direct get of its membership from the assignments stream went with the stream (research 072 - // row 33): it reads the grants part's state memberships instead (ADR 0571 clause 7). - if !now[g] && !strings.HasPrefix(g, "$JS.API.DIRECT.GET."+AssignmentsStream+".mesh.assignment."+strings.SplitN(user, ".", 2)[0]+".") { + // row 33): it reads the grants part's state memberships instead (ADR 0571 clause 7). And it no longer hears + // the subject, where the grants part stops publishing at the lease write (ADR 0571 clause 10, issue 774). + node := strings.SplitN(user, ".", 2)[0] + if !now[g] && !strings.HasPrefix(g, "$JS.API.DIRECT.GET."+AssignmentsStream+".mesh.assignment."+node+".") && + !strings.HasPrefix(g, "mesh.assignment."+node+".") { t.Errorf("%s: the copy does not grant %s, which the controller's does", user, g) } } @@ -880,6 +883,17 @@ func healthChecks(t *testing.T) { if u.Kind == KindModule && MayPublish(perms, "$KV.mesh-delivery-assigner_policies.x") { t.Errorf("%s writes the assigner's policies", u.Username()) } + // ADR 0571 clause 10 (issue 774, 774/S5/AC2): nobody hears memberships on mesh.assignment; they are read from + // the grants part's state. + if MaySubscribe(perms, "mesh.assignment.a.restic") { + t.Errorf("%s hears mesh.assignment", u.Username()) + } + } + // Nor the runtime carrying the node's modules, which reads each one's membership from the state. + if perms, err := PermissionsFor(Principal{Kind: KindNodeTools, Node: "a", Module: RuntimeModule, + Carries: []Declared{{Module: "restic"}}}); err != nil || MaySubscribe(perms, "mesh.assignment.a.restic") || + !MayPublish(perms, "$JS.API.DIRECT.GET.KV_mesh-grants_memberships.$KV.mesh-grants_memberships.a.restic") { + t.Errorf("the runtime hears mesh.assignment or reads no membership: %v", err) } // 1953 S-age (ADR 0565 §10): a log that says an age keeps its entries that long and no longer; one that says none, as // long as the log (ADR 0297). diff --git a/go/buscompose/nats.go b/go/buscompose/nats.go index c37be8f..8b56c96 100644 --- a/go/buscompose/nats.go +++ b/go/buscompose/nats.go @@ -1075,10 +1075,10 @@ func PermissionsFor(p Principal) (Permissions, error) { } } sub = append(sub, announcing(announced...)...) - // Its own membership (ADR 0160): the one subject a runtime derives for itself, heard live. Nothing else's. - sub = append(sub, MembershipSubject(p.Node, p.Module)) - // And from the grants part's state memberships, its own key (novox/hq ADR 0571 clause 7): its one read of it, - // since the assignments stream's direct get went with the stream (research 072 row 33). + // Its own membership (ADR 0160), from the grants part's state memberships, its own key (novox/hq ADR 0571 clause + // 7): its one read of it, since the assignments stream's direct get went with the stream (research 072 row 33), + // and no longer heard on its subject, where the grants part stops publishing at the lease write (ADR 0571 clause + // 10, issue 774). pub = append(pub, readGrants(GrantsMembershipsBucket, MembershipKey(p.Node, p.Module))...) // 1b. The tools it calls, if its manifest says it calls any (novox/hq ADR 0152). The same @@ -1240,10 +1240,10 @@ func PermissionsFor(p Principal) (Permissions, error) { } } } - // Every assigned module's membership on this node (ADR 0160): one per module, heard live and read from the - // grants part's state memberships (ADR 0571 clause 7). This node's and no other's — the one token - // that varies is the module, so the pattern is the machine's own assignments. - sub = append(sub, "mesh.assignment."+p.Node+".*") + // Every assigned module's membership on this node (ADR 0160): one per module, read and watched from the + // grants part's state memberships (ADR 0571 clause 7), and no longer heard on `mesh.assignment`, where the grants + // part stops publishing at the lease write (ADR 0571 clause 10, issue 774). This node's and no other's — the + // one token that varies is the module, so the pattern is the machine's own assignments. pub = append(pub, readGrants(GrantsMembershipsBucket, p.Node+".*")...) // And that it is there (novox/hq to-be 45 §3, S11): its own heartbeat, under its machine's // name and no other's, on core NATS like the host's. diff --git a/go/process/follow.go b/go/process/follow.go index 9c1fc33..73fc02c 100644 --- a/go/process/follow.go +++ b/go/process/follow.go @@ -5,7 +5,6 @@ import ( "fmt" "time" - "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-sdk/go/membership" @@ -17,10 +16,10 @@ import ( // not a membership is handed to onBad (nil to ignore it) and skipped. // // It reads the membership from the grants part's state memberships, its own key (novox/hq ADR 0571 clause 7), -// watched so the value there now comes first and each change after it; and it hears the subject the grants part -// publishes on until the lease write. The assignments stream is no longer read: the bus module's tools delete it -// at the lease write (research 072 row 33). A state not there yet, or not readable, is told to onBad and the -// subject is followed alone. The returned error is a refusal to follow at all. +// watched so the value there now comes first and each change after it. Nothing else: the subject the grants part +// published on is no longer heard (ADR 0571 clause 10, issue 774), since it stops publishing at the lease write, and +// the assignments stream is no longer read (research 072 row 33). A state not there or not readable is a refusal to +// follow, returned: there is nothing else to follow. func (c *Conn) FollowMembership(apply func(membership.Membership), onBad func(error)) error { if c.account.Node == "" || c.account.Module == "" { return fmt.Errorf("the account names no node or module, so it has no membership") @@ -38,27 +37,20 @@ func (c *Conn) FollowMembership(apply func(membership.Membership), onBad func(er } apply(issued) } - subject := membership.Subject(c.account.Node, c.account.Module) - if _, err := c.nc.Subscribe(subject, func(m *nats.Msg) { take(m.Data) }); err != nil { - return fmt.Errorf("cannot follow %s: %w", subject, err) - } key := membership.GrantsKey(c.account.Node, c.account.Module) js, err := jetstream.New(c.nc) if err != nil { - bad(fmt.Errorf("cannot read %s of %s (%v); following %s alone", key, membership.GrantsBucket, err, subject)) - return nil + return fmt.Errorf("cannot read %s of %s: %w", key, membership.GrantsBucket, err) } ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() kv, err := js.KeyValue(ctx, membership.GrantsBucket) if err != nil { - bad(fmt.Errorf("cannot read %s of %s yet (%v); following %s alone", key, membership.GrantsBucket, err, subject)) - return nil + return fmt.Errorf("cannot read %s of %s: %w", key, membership.GrantsBucket, err) } w, err := kv.Watch(context.Background(), key) if err != nil { - bad(fmt.Errorf("cannot watch %s of %s (%v); following %s alone", key, membership.GrantsBucket, err, subject)) - return nil + return fmt.Errorf("cannot watch %s of %s: %w", key, membership.GrantsBucket, err) } go func() { for e := range w.Updates() { diff --git a/go/process/follow_integration_test.go b/go/process/follow_integration_test.go index cdb420b..3041b4e 100644 --- a/go/process/follow_integration_test.go +++ b/go/process/follow_integration_test.go @@ -3,14 +3,20 @@ package process import ( + "context" + "strings" "testing" "time" + "github.com/nats-io/nats.go/jetstream" + "git.novox.be/novox/mesh-sdk/go/membership" ) -// A process connected pinned to a test bus hears the membership published for its node and module, and -// ignores what is not one (novox/hq issue 1633, AC2). +// A process connected pinned to a test bus follows the membership the grants part writes for its node and module in +// its state memberships, and reports what is not one (novox/hq issue 1633, AC2); it no longer hears mesh.assignment, +// where the grants part stops publishing at the lease write (ADR 0571 clause 10; issue 774, 774/S5/AC1). Without the +// state it refuses to follow, naming it. func TestFollowMembershipHearsWhatIsIssuedToTheAccount(t *testing.T) { url, fingerprint := selfSignedServer(t) account := mailSenderAccount(t, url, fingerprint) @@ -20,26 +26,52 @@ func TestFollowMembershipHearsWhatIsIssuedToTheAccount(t *testing.T) { } defer conn.Close() - got := make(chan membership.Membership, 1) + got := make(chan membership.Membership, 4) bad := make(chan error, 4) - if err := conn.FollowMembership(func(m membership.Membership) { got <- m }, func(e error) { bad <- e }); err != nil { + follow := func() error { + return conn.FollowMembership(func(m membership.Membership) { got <- m }, func(e error) { bad <- e }) + } + if err := follow(); err == nil || !strings.Contains(err.Error(), membership.GrantsBucket) { + t.Fatalf("followed with no state memberships: %v", err) + } + js, err := jetstream.New(conn.nc) + if err != nil { t.Fatal(err) } - if err := conn.nc.Publish(membership.Subject("novox", "mesh-mail"), []byte("not json")); err != nil { + ctx := context.Background() + kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: membership.GrantsBucket}) + if err != nil { t.Fatal(err) } - if err := conn.nc.Publish(membership.Subject("novox", "mesh-mail"), []byte(`{"node":"novox","module":"mesh-mail","mesh":["10.0.0.1"]}`)); err != nil { + key := membership.GrantsKey("novox", "mesh-mail") + if _, err := kv.Put(ctx, key, []byte("not json")); err != nil { + t.Fatal(err) + } + if err := follow(); err != nil { + t.Fatal(err) + } + // Published on the subject, as the grants part did until the write: not heard. + if err := conn.nc.Publish(membership.Subject("novox", "mesh-mail"), []byte(`{"node":"novox","module":"mesh-mail","mesh":["10.0.0.9"]}`)); err != nil { + t.Fatal(err) + } + if _, err := kv.Put(ctx, key, membership.EncodeGrantsMembership(membership.Membership{Node: "novox", Module: "mesh-mail", + Mesh: []string{"10.0.0.1"}})); err != nil { t.Fatal(err) } select { case m := <-got: - if m.Module != "mesh-mail" || len(m.Mesh) != 1 { + if m.Module != "mesh-mail" || len(m.Mesh) != 1 || m.Mesh[0] != "10.0.0.1" { t.Fatalf("membership: %+v", m) } case <-time.After(5 * time.Second): - t.Fatal("no membership heard") + t.Fatal("no membership followed") + } + select { + case m := <-got: + t.Errorf("a second membership was applied, from the subject: %+v", m) + case <-time.After(300 * time.Millisecond): } if len(bad) == 0 { - t.Error("the body that is not a membership was not reported") + t.Error("the value that is not a membership was not reported") } }