jschoubben
a4ed4e6742
ADR 0571 clause 10: the grants part publishes memberships on mesh.assignment until the lease write and stops then. FollowMembership now reads and watches its own key of the state memberships and nothing else, and refuses to follow, naming the state, when it cannot read it (there is nothing else to follow, and a silent follow of nothing is the fault). A module's principal and the runtime lose their subscribe on mesh.assignment; the grants part's publish grant stays until the write (S21 stops the publishing). Tests: TestFollowMembershipHearsWhatIsIssuedToTheAccount (integration) edited: no state is a refusal, a membership put in the state is applied, one published on the subject is not; run locally against its embedded bus. Rows added to TestTheCopiedComposerComposesAsTheControllers: nobody subscribes mesh.assignment, and the controller's golden grant of it is the one row the copy now lacks. No new test function. (issue 774, 774/S5) Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL
64 lines
2.2 KiB
Go
64 lines
2.2 KiB
Go
package process
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/membership"
|
|
)
|
|
|
|
// FollowMembership reads this account's own membership and follows it (novox/hq ADR 0160): apply is called
|
|
// with every membership the mesh issues this account's node and module, the one already issued first and
|
|
// each later one as it is issued, so a change reaches a running process without a restart. A body that is
|
|
// 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. 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")
|
|
}
|
|
bad := func(err error) {
|
|
if onBad != nil {
|
|
onBad(err)
|
|
}
|
|
}
|
|
take := func(body []byte) {
|
|
issued, err := membership.DecodeGrantsMembership(body)
|
|
if err != nil {
|
|
bad(fmt.Errorf("a membership arrived that is not one: %w", err))
|
|
return
|
|
}
|
|
apply(issued)
|
|
}
|
|
key := membership.GrantsKey(c.account.Node, c.account.Module)
|
|
js, err := jetstream.New(c.nc)
|
|
if err != 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 {
|
|
return fmt.Errorf("cannot read %s of %s: %w", key, membership.GrantsBucket, err)
|
|
}
|
|
w, err := kv.Watch(context.Background(), key)
|
|
if err != nil {
|
|
return fmt.Errorf("cannot watch %s of %s: %w", key, membership.GrantsBucket, err)
|
|
}
|
|
go func() {
|
|
for e := range w.Updates() {
|
|
if e != nil && e.Operation() == jetstream.KeyValuePut {
|
|
take(e.Value())
|
|
}
|
|
}
|
|
}()
|
|
return nil
|
|
}
|