diff --git a/modules/mesh-grants/cmd/mesh-grants/create.go b/modules/mesh-grants/cmd/mesh-grants/create.go index c155393..e48f458 100644 --- a/modules/mesh-grants/cmd/mesh-grants/create.go +++ b/modules/mesh-grants/cmd/mesh-grants/create.go @@ -18,6 +18,26 @@ type creating struct { js jetstream.JetStream } +// CreateConsumer creates one durable consumer an account needs, by its stream and name (ADR 0561 §6): the create alone; +// one the bus holds answers ErrHeld and is left as it is. +func (c creating) CreateConsumer(stream string, cfg jetstream.ConsumerConfig) error { + ctx, cancel := context.WithTimeout(c.ctx, requestWithin) + defer cancel() + if _, err := c.js.Consumer(ctx, stream, cfg.Durable); err == nil { + return grants.ErrHeld + } + _, err := c.js.CreateConsumer(ctx, stream, cfg) + switch { + case err == nil: + return nil + case errors.Is(err, jetstream.ErrConsumerExists): + return grants.ErrHeld + case errors.Is(err, context.DeadlineExceeded): + return fmt.Errorf("no answer in %s: a create the bus refuses is answered by none (%w)", requestWithin, err) + } + return err +} + func (c creating) CreateDeclared(cfg jetstream.StreamConfig) error { ctx, cancel := context.WithTimeout(c.ctx, requestWithin) defer cancel() diff --git a/modules/mesh-grants/cmd/mesh-grants/main.go b/modules/mesh-grants/cmd/mesh-grants/main.go index b6ff6a5..6ef39b6 100644 --- a/modules/mesh-grants/cmd/mesh-grants/main.go +++ b/modules/mesh-grants/cmd/mesh-grants/main.go @@ -112,7 +112,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } bus := newBusReader(ctx, js) f.Part = &grants.Part{Read: bus, Call: conn, Store: ownStore{conn}, Emit: emitting{conn}, - Publish: memberships{ctx: ctx, nc: nc, js: js}, Create: creating{ctx: ctx, js: js}, + Publish: memberships{ctx: ctx, nc: nc, js: js}, Create: creating{ctx: ctx, js: js}, Consumers: creating{ctx: ctx, js: js}, Objects: sealedObjects{ctx: ctx, js: js}, Logf: logger.Printf, Bus: grants.BusAddress{URL: account.URL, Fingerprint: account.Fingerprint}} f.Watch = bus diff --git a/modules/mesh-grants/grants_test.go b/modules/mesh-grants/grants_test.go index d275df7..dd00930 100644 --- a/modules/mesh-grants/grants_test.go +++ b/modules/mesh-grants/grants_test.go @@ -57,6 +57,31 @@ type fakeBus struct { refuseUsers bool // rootFree answers the controller's root-free verb: nil is no answerer at all (no responders). rootFree func(machines []string) (any, error) + // consumers are the durable consumers the bus holds, by stream, as nats_streams answers them for one stream; and + // consumerCreates each consumer the part asked to create, `.`. + consumers map[string][]map[string]any + consumerCreates []string +} + +// CreateConsumer is the bus's create of a durable consumer: held when the bus holds one of that name, else held from now. +func (b *fakeBus) CreateConsumer(stream string, cfg jetstream.ConsumerConfig) error { + b.mu.Lock() + defer b.mu.Unlock() + b.consumerCreates = append(b.consumerCreates, stream+"."+cfg.Durable) + for _, c := range b.consumers[stream] { + if c["name"] == cfg.Durable { + return grants.ErrHeld + } + } + filters := append([]string{}, cfg.FilterSubjects...) + if cfg.FilterSubject != "" { + filters = []string{cfg.FilterSubject} + } + if b.consumers == nil { + b.consumers = map[string][]map[string]any{} + } + b.consumers[stream] = append(b.consumers[stream], map[string]any{"name": cfg.Durable, "filters": filters, "max_deliver": cfg.MaxDeliver}) + return nil } func (b *fakeBus) LastMembership(node, module string) ([]byte, uint64, bool, error) { @@ -177,6 +202,9 @@ func (b *fakeBus) Call(key string, body any) (json.RawMessage, error) { } return json.Marshal(map[string]any{"users": out}) case grants.CallStreams: + if stream, _ := args["stream"].(string); stream != "" { + return json.Marshal(map[string]any{"detail": []map[string]any{{"name": stream, "consumer_detail": b.consumers[stream]}}}) + } return json.Marshal(map[string]any{"detail": b.streams}) case grants.CallRootFree: if b.rootFree == nil { @@ -1770,6 +1798,43 @@ func TestGrants(t *testing.T) { t.Errorf("the part hears %v", grants.Consumes) } }) + + // 774/AC6 (novox/hq ADR 0561 §6): the durable consumers an account needs, with the controller's configuration, are + // compared in shadow while the controller holds the lease, and created, only where the bus lacks them, from the write + // that hands the lease away; one the bus holds is never changed. + t.Run("account consumers: compared while the controller holds the lease, created once it is handed away", func(t *testing.T) { + const cManifest = `{"module":"mod-c","version":"1","own-secrets":{"broker":"/var/lib/c/broker"},"consumes":["mod-y.done"]}` + b := fixture() + b.manifestOf("mod-c", cManifest) + b.assign("b", "mod-c") + b.loaded = shadowOf(t, b) + b.put(grants.ReadLease, grants.LeaseKey, map[string]any{"instance": "novox/1/2026-10-27", "epoch": 57}) + p := partOn(b) + p.Consumers = b + res := p.Compose() + if res.Consumers == nil || res.Consumers.On || !slices.Contains(res.Consumers.Missing, "EVENTS.b_mod-c") || len(b.consumerCreates) != 0 { + t.Fatalf("while the controller holds the lease: %+v, created %v", res.Consumers, b.consumerCreates) + } + b.put(grants.ReadLease, grants.LeaseKey, map[string]any{"handed-to": "the operator's hand-over"}) + b.consumers = map[string][]map[string]any{"EVENTS": {{"name": "b_other", "filters": []string{"x"}, "max_deliver": 5}}} + res = p.Compose() + if !res.Consumers.On || !slices.Equal(res.Consumers.Created, []string{"EVENTS.b_mod-c"}) { + t.Fatalf("once handed away: %+v", res.Consumers) + } + var made map[string]any + for _, c := range b.consumers["EVENTS"] { + if c["name"] == "b_mod-c" { + made = c + } + } + if made == nil || made["max_deliver"] != 5 || !slices.Contains(made["filters"].([]string), "mesh.mod.mod-y.event.done") { + t.Errorf("b_mod-c was made as %v, not as the controller makes it", made) + } + res = p.Compose() + if len(res.Consumers.Created) != 0 || len(res.Consumers.Missing) != 0 || len(b.consumerCreates) != 1 { + t.Errorf("a consumer the bus holds was asked again: %+v, %v", res.Consumers, b.consumerCreates) + } + }) } type fakeAsker struct { diff --git a/modules/mesh-grants/internal/grants/consumers.go b/modules/mesh-grants/internal/grants/consumers.go new file mode 100644 index 0000000..3dc631f --- /dev/null +++ b/modules/mesh-grants/internal/grants/consumers.go @@ -0,0 +1,156 @@ +package grants + +import ( + "encoding/json" + "errors" + "fmt" + "slices" + "sort" + "strings" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/buscompose" +) + +// **The durable consumers an account needs** (novox/hq ADR 0561 §6, issue 774). The controller creates a module's +// consumer on EVENTS and each process's on its work queue when it issues the account. From the write that hands its +// lease away the grants part creates them, with the configuration the bus composer derives (buscompose.AccountConsumers), +// only where the bus lacks one; one the bus holds is never changed or deleted, a difference is only said. Before that +// write they are compared in shadow alone, so two parts never create one consumer. + +// ReadLease is the controller's lease as the part reads it (buscompose.NamedReaders names the vault and the part), and +// LeaseKey its one key. +const ( + ReadLease = buscompose.ControllerLease + LeaseKey = buscompose.ControllerLeaseKey +) + +// ConsumerCreator creates one durable consumer the bus does not hold: the part's process alone, whose account is granted +// the create of each account's consumer by stream and name. One the bus holds answers ErrHeld. +type ConsumerCreator interface { + CreateConsumer(stream string, cfg jetstream.ConsumerConfig) error +} + +// ConsumersState is what one pass found and did of the account consumers. +type ConsumersState struct { + // On says the lease is handed away, so the part creates what the bus lacks; Why says why it is on or off. + On bool `json:"on"` + Why string `json:"why"` + // Wanted is how many the records compose; Missing those the bus lacks (`.`), Differ those it holds + // otherwise, each with how. + Wanted int `json:"wanted"` + Missing []string `json:"missing,omitempty"` + Differ []string `json:"differ,omitempty"` + Created []string `json:"created,omitempty"` + NotCreated []string `json:"not_created,omitempty"` + // Unread is a stream whose consumers could not be read: nothing on it is compared or created. + Unread []string `json:"unread,omitempty"` +} + +// heldConsumer is one consumer as the bus module's nats_streams reads it for one stream. +type heldConsumer struct { + Name string `json:"name"` + Filters []string `json:"filters"` + MaxDeliver int `json:"max_deliver"` +} + +// heldConsumers reads the consumers the bus holds on stream, by name. +func heldConsumers(c Caller, stream string) (map[string]heldConsumer, error) { + raw, err := c.Call(CallStreams, map[string]any{"stream": stream}) + if err != nil { + return nil, fmt.Errorf("%s %s: %w", CallStreams, stream, err) + } + var got struct { + Detail []struct { + Name string `json:"name"` + Consumers []heldConsumer `json:"consumer_detail"` + } `json:"detail"` + } + if err := json.Unmarshal(unwrap(raw), &got); err != nil { + return nil, fmt.Errorf("%s %s answered no consumers: %w", CallStreams, stream, err) + } + out := map[string]heldConsumer{} + for _, d := range got.Detail { + if d.Name != stream { + continue + } + for _, c := range d.Consumers { + out[c.Name] = c + } + } + return out, nil +} + +// leaseHandedAway reads the lease as the part's reads give it. +func leaseHandedAway(r Reader) (bool, string) { + raw, found, err := r.Get(ReadLease, LeaseKey) + return buscompose.LeaseHandedAway(raw, found, err) +} + +// consumers compares the account consumers the records compose with those the bus holds, and creates the missing ones +// once the lease is handed away and the part holds a creator. +func (p *Part) consumers(res *Result, recs buscompose.Records) { + users, err := buscompose.Users(recs) + if err != nil { + res.Errors = append(res.Errors, "the account consumers could not be composed: "+err.Error()) + return + } + st := &ConsumersState{} + st.On, st.Why = leaseHandedAway(p.Read) + res.Consumers = st + wanted := buscompose.AccountConsumers(users) + st.Wanted = len(wanted) + byStream := map[string][]buscompose.Consumer{} + for _, mc := range wanted { + byStream[mc.Consumer.Stream] = append(byStream[mc.Consumer.Stream], mc.Consumer) + } + for _, stream := range sortedKeys(byStream) { + held, err := heldConsumers(p.Call, stream) + if err != nil { + st.Unread = append(st.Unread, stream+": "+err.Error()) + continue + } + for _, c := range byStream[stream] { + key := stream + "." + c.Name + h, ok := held[c.Name] + if ok { + if why := consumerDiffers(c, h); why != "" { + st.Differ = append(st.Differ, key+": "+why) + } + continue + } + if !st.On || p.Consumers == nil { + st.Missing = append(st.Missing, key) + continue + } + switch err := p.Consumers.CreateConsumer(stream, buscompose.ConsumerConfig(c)); { + case err == nil: + st.Created = append(st.Created, key) + p.say(res, EventAsserted, map[string]any{"consumer": c.Name, "stream": stream}) + case errors.Is(err, ErrHeld): + default: + st.Missing = append(st.Missing, key) + st.NotCreated = append(st.NotCreated, key+": "+err.Error()) + } + } + } +} + +// consumerDiffers says how a held consumer differs from the composed one in what nats_streams reads back, or "". +func consumerDiffers(c buscompose.Consumer, h heldConsumer) string { + var why []string + want, have := slices.Sorted(slices.Values(c.Filters)), slices.Sorted(slices.Values(h.Filters)) + if !slices.Equal(want, have) { + why = append(why, fmt.Sprintf("filters %v, composed %v", have, want)) + } + maxDeliver := c.MaxDeliver + if maxDeliver == 0 { + maxDeliver = -1 + } + if h.MaxDeliver != maxDeliver { + why = append(why, fmt.Sprintf("max deliver %d, composed %d", h.MaxDeliver, maxDeliver)) + } + sort.Strings(why) + return strings.Join(why, "; ") +} diff --git a/modules/mesh-grants/internal/grants/follow.go b/modules/mesh-grants/internal/grants/follow.go index 79ff31d..98ba29b 100644 --- a/modules/mesh-grants/internal/grants/follow.go +++ b/modules/mesh-grants/internal/grants/follow.go @@ -6,6 +6,7 @@ import ( "strings" "sync" + "git.novox.be/novox/mesh-sdk/go/buscompose" "git.novox.be/novox/mesh-sdk/go/delivery" ) @@ -18,7 +19,7 @@ type Watcher interface { // Followed are the states whose change composes again: placement, the indexer's manifests by module, the resolved // config values, the node agents' haves (a machine's first have, and the bus node's loaded list) and the assigner's // wants (the manifest each machine is wanted to run), and the vault's hashes of the bus accounts. -var Followed = []string{ReadAssignments, ReadModules, ReadResolved, ReadHave, ReadWants, ReadAccounts, ReadPlacement} +var Followed = []string{ReadAssignments, ReadModules, ReadResolved, ReadHave, ReadWants, ReadAccounts, ReadPlacement, ReadLease} // Follower composes again on every change it hears, never on a timer. It never stops for a refusal: a watch it cannot // open is said, composed without, and opened again at the next wake (any change it does hear, or a reconnect). @@ -46,6 +47,8 @@ type Follower struct { held string notAsserted string refusedMans string + // leaseOn is whether the lease was handed away at its last change heard, once leaseSeen. + leaseOn, leaseSeen bool } func (f *Follower) init() { @@ -85,6 +88,15 @@ func (f *Follower) logf(format string, args ...any) { // heard says whether a change composes again: every change but a have's, and of the haves a machine's first and the // bus module's loaded list moving. func (f *Follower) heard(state, key string, value []byte) bool { + if state == ReadLease { + // Renewed every few seconds by its holder: only the write that hands it away, or back, composes again (ADR 0561 §6). + on, _ := buscompose.LeaseHandedAway(value, len(value) > 0, nil) + f.mu.Lock() + defer f.mu.Unlock() + was, seen := f.leaseOn, f.leaseSeen + f.leaseOn, f.leaseSeen = on, true + return !seen || was != on + } if state != ReadHave { return true } diff --git a/modules/mesh-grants/internal/grants/inputs.go b/modules/mesh-grants/internal/grants/inputs.go index 0634fe5..d870f41 100644 --- a/modules/mesh-grants/internal/grants/inputs.go +++ b/modules/mesh-grants/internal/grants/inputs.go @@ -41,7 +41,7 @@ const ( // Reads is every state the part reads, in the manifest's order. var Reads = []string{ReadAssignments, ReadControllerGrant, ReadModules, ReadManifests, ReadResolved, ReadHave, ReadWants, ReadBuilds, - ReadAccounts, ReadPlacement} + ReadAccounts, ReadPlacement, ReadLease} // The bus module's tools the part calls, as its manifest's invokes name them: read-only, each. const ( diff --git a/modules/mesh-grants/internal/grants/part.go b/modules/mesh-grants/internal/grants/part.go index 257ccb2..6af7f4e 100644 --- a/modules/mesh-grants/internal/grants/part.go +++ b/modules/mesh-grants/internal/grants/part.go @@ -108,6 +108,8 @@ type Result struct { RootNotFree []string `json:"root_not_free,omitempty"` // Sealed is the list sealed to the bus node as the key `user-list.sealed` names it (ADR 0547 clause 7). Sealed *SealedState `json:"sealed,omitempty"` + // Consumers is what the pass found and created of the durable consumers an account needs (ADR 0561 §6). + Consumers *ConsumersState `json:"consumers,omitempty"` // Accounts is what the pass wrote of the wanted accounts the vault mints from (ADR 0561 §2). Accounts *AccountsWritten `json:"accounts,omitempty"` Errors []string `json:"errors,omitempty"` @@ -134,6 +136,9 @@ type Part struct { Logf func(string, ...any) // Objects is the part's object bucket of sealed lists: the process's connection; nil in the tools. Objects Objects + // Consumers creates an account's durable consumer from the lease write (consumers.go): the process's connection; + // nil in the tools, which compare only. + Consumers ConsumerCreator // Bus is where the nodes reach the bus, which each wanted account carries (accounts.go): the process's own credential's; // empty in the tools, which write no account. Bus BusAddress @@ -304,6 +309,7 @@ func (p *Part) Compose() Result { res.AssertionsWritten++ } } + p.consumers(&res, recs) if _, _, found, err := p.Store.Get(StateTakeover, KeySwitch); err == nil && !found { data, _ := json.Marshal(Switch{V: 1, State: Shadow, Acts: []Act{}}) if _, err := p.Store.Put(StateTakeover, KeySwitch, data); err != nil { diff --git a/modules/mesh-grants/module.json b/modules/mesh-grants/module.json index 189508d..ea541f3 100644 --- a/modules/mesh-grants/module.json +++ b/modules/mesh-grants/module.json @@ -12,7 +12,8 @@ "mesh-delivery-indexer.modules", "mesh-delivery-indexer.placement", "node-engine.have", - "mesh-vault.bus-accounts" + "mesh-vault.bus-accounts", + "mesh-controller.lease" ], "consumes": [ "mesh-controller.condition-raised",