The lead's ruling of 13:0x (registry milestone): mesh-broker's events are mesh.seat.mesh-broker.event.<event> on the shared events stream like every seat's. The grants part and the watchman consume mesh-broker.<event> and no longer hear the bus module's own stream, which is its history for its tool. The keeper's process account is composed the seat's event subjects, and the acknowledgement of a snapshot's chunks of each entry it keeps, which ADR 0626 §1 grants and the composition lacked; a row holds it to no subscription of a restricted stream's subjects and no consumer of one. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
251 lines
9.1 KiB
Go
251 lines
9.1 KiB
Go
package grants
|
|
|
|
import (
|
|
"encoding/json"
|
|
"fmt"
|
|
"sort"
|
|
"strings"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
|
"git.novox.be/novox/mesh-sdk/go/membership"
|
|
"mesh-grants/internal/buscompose"
|
|
)
|
|
|
|
// **The binding travels on the bus module's events** (novox/hq ADR 0595 §1, §7; design 25 § 12.2; DLV-G1). The bus
|
|
// module (the holder of the seat mesh-broker) says `created` when it created one entry of a module's bus state,
|
|
// `recreated` when it rebuilt one, and `dropped` when it removed one; each carries the entry's names on the bus. The
|
|
// grants part hears them on its durable consumer of the bus module's own stream (its manifest's `consumes`), writes the
|
|
// binding into the module's membership on every machine the module is assigned, rewrites it on `recreated` and removes
|
|
// it on `dropped`. What it heard is kept in its own state memberships, one key per module (BoundKey), so a binding
|
|
// outlives the 30 days its event is kept and reaches a machine the module is assigned to later. A `refused` create
|
|
// hands no binding: only these three events are read.
|
|
|
|
// The bus module's events the part hears, by their local names (ADR 0595 §6; ADR 0620 §2: the holder's own events,
|
|
// never a seat's subject).
|
|
const (
|
|
BrokerCreated = "created"
|
|
BrokerRecreated = "recreated"
|
|
BrokerDropped = "dropped"
|
|
)
|
|
|
|
// BoundPrefix starts the key of the part's own state memberships that keeps one module's bindings, `bound.<module>`:
|
|
// never a `<node>.<module>` key, since no machine is named `bound` (a membership's runtime reads its own key alone).
|
|
const BoundPrefix = "bound."
|
|
|
|
// BoundKey is the key keeping module's bindings.
|
|
func BoundKey(module string) string { return BoundPrefix + module }
|
|
|
|
// BrokerEvent is the body of the bus module's `created`, `recreated` and `dropped`, the fields the part reads.
|
|
//
|
|
// It stands until the seat mesh-broker's definition is generated in mesh-contracts (ADR 0613 §1, §5; design 26): this
|
|
// part then reads the generated type, with the same fields.
|
|
type BrokerEvent struct {
|
|
V int `json:"v,omitempty"`
|
|
Module string `json:"module"`
|
|
// LocalName and StorageType are the seat's contract's names (mesh-contracts meshbroker.Said); Local and Storage the
|
|
// same read where an older body names them so.
|
|
LocalName string `json:"local-name,omitempty"`
|
|
Local string `json:"local,omitempty"`
|
|
StorageType string `json:"storage-type,omitempty"`
|
|
Storage string `json:"storage,omitempty"`
|
|
Template string `json:"template,omitempty"`
|
|
// Binding is the entry's names on the bus: whole on `created` and `recreated`; on `dropped` only its local name counts.
|
|
Binding membership.BusBinding `json:"binding"`
|
|
}
|
|
|
|
// Bound is the value of BoundKey: every binding of one module the bus module said it created and has not dropped, by
|
|
// local name.
|
|
type Bound struct {
|
|
V int `json:"v"`
|
|
Module string `json:"module"`
|
|
Bindings map[string]membership.BusBinding `json:"bindings"`
|
|
}
|
|
|
|
// brokerEvent is the local name of an event heard from the holder of mesh-broker, by its bus subject, and whether it is
|
|
// one: the seat's own subject alone (`mesh.seat.mesh-broker.event.<event>`, the lead's ruling, registry milestone 13:0x),
|
|
// where only its holder publishes, and only while a module holds the seat. The bus module's own stream is its history
|
|
// for its tool, never what the part hears.
|
|
func brokerEvent(event string, brokers map[string]bool) (string, bool) {
|
|
name, onTheSeat := strings.CutPrefix(event, "mesh.seat."+catalogue.BrokerSeat+".event.")
|
|
if !onTheSeat || len(brokers) == 0 {
|
|
return "", false
|
|
}
|
|
switch name {
|
|
case BrokerCreated, BrokerRecreated, BrokerDropped:
|
|
return name, true
|
|
}
|
|
return "", false
|
|
}
|
|
|
|
// HearBroker applies one event heard to the kept bindings: a `created` or `recreated` writes the entry's binding, a
|
|
// `dropped` removes it. It answers whether the kept bindings changed (so a pass composes again), or why the event was
|
|
// not applied; an event that is no bus module's is no change and no error. brokers are the modules holding mesh-broker.
|
|
func (p *Part) HearBroker(event string, body []byte, brokers map[string]bool) (bool, error) {
|
|
name, ok := brokerEvent(event, brokers)
|
|
if !ok {
|
|
return false, nil
|
|
}
|
|
var e BrokerEvent
|
|
if err := json.Unmarshal(body, &e); err != nil {
|
|
return false, fmt.Errorf("the bus module's %s does not read: %w", name, err)
|
|
}
|
|
local := firstOf(e.LocalName, e.Local, e.Binding.Local)
|
|
storage := firstOf(e.StorageType, e.Storage)
|
|
if e.Module == "" || local == "" || strings.Contains(e.Module, ".") {
|
|
return false, fmt.Errorf("the bus module's %s names no module or no entry (%q, %q)", name, e.Module, local)
|
|
}
|
|
b, err := p.bound(e.Module)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
was, had := b.Bindings[local]
|
|
switch name {
|
|
case BrokerDropped:
|
|
if !had {
|
|
return false, nil
|
|
}
|
|
delete(b.Bindings, local)
|
|
default:
|
|
bind := e.Binding
|
|
bind.Local = local
|
|
if bind.Storage == "" {
|
|
bind.Storage = storage
|
|
}
|
|
if bind.Template == "" {
|
|
bind.Template = e.Template
|
|
}
|
|
if had && sameBinding(was, bind) {
|
|
return false, nil
|
|
}
|
|
b.Bindings[local] = bind
|
|
}
|
|
raw, _ := json.Marshal(b)
|
|
if _, err := p.Store.Put(StateMemberships, BoundKey(e.Module), raw); err != nil {
|
|
return false, fmt.Errorf("its own %s %s cannot be written: %w", StateMemberships, BoundKey(e.Module), err)
|
|
}
|
|
return true, nil
|
|
}
|
|
|
|
// bound reads the kept bindings of module; none kept is an empty value.
|
|
func (p *Part) bound(module string) (Bound, error) {
|
|
b := Bound{V: 1, Module: module, Bindings: map[string]membership.BusBinding{}}
|
|
raw, _, found, err := p.Store.Get(StateMemberships, BoundKey(module))
|
|
if err != nil {
|
|
return b, fmt.Errorf("its own %s %s cannot be read: %w", StateMemberships, BoundKey(module), err)
|
|
}
|
|
if found {
|
|
if err := json.Unmarshal(raw, &b); err != nil {
|
|
return b, fmt.Errorf("its own %s %s does not read: %w", StateMemberships, BoundKey(module), err)
|
|
}
|
|
if b.Bindings == nil {
|
|
b.Bindings = map[string]membership.BusBinding{}
|
|
}
|
|
}
|
|
return b, nil
|
|
}
|
|
|
|
// bindingsOf is module's kept bindings in local-name order, for its memberships; nil when none.
|
|
func (p *Part) bindingsOf(module string) ([]membership.BusBinding, error) {
|
|
b, err := p.bound(module)
|
|
if err != nil || len(b.Bindings) == 0 {
|
|
return nil, err
|
|
}
|
|
locals := make([]string, 0, len(b.Bindings))
|
|
for l := range b.Bindings {
|
|
locals = append(locals, l)
|
|
}
|
|
sort.Strings(locals)
|
|
out := make([]membership.BusBinding, 0, len(locals))
|
|
for _, l := range locals {
|
|
out = append(out, b.Bindings[l])
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
func sameBinding(a, b membership.BusBinding) bool {
|
|
ra, _ := json.Marshal(a)
|
|
rb, _ := json.Marshal(b)
|
|
return string(ra) == string(rb)
|
|
}
|
|
|
|
// OwnHears are the module streams the part's own membership on node names it hearing (its consumes of another module's
|
|
// stream, the bus module's among them), each with its durable consumer: read from its own state memberships, which the
|
|
// part writes; none yet is an error, so the hearing is opened at the next change.
|
|
func OwnHears(s Store, node string) ([]membership.HeardIssued, error) {
|
|
key := buscompose.MembershipKey(node, Module)
|
|
raw, _, found, err := s.Get(StateMemberships, key)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("its own membership %s cannot be read: %w", key, err)
|
|
}
|
|
if !found {
|
|
return nil, fmt.Errorf("its own membership %s is not composed yet, so the module streams it hears are not known", key)
|
|
}
|
|
bare, err := membership.BareGrantsMembership(raw)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("its own membership %s does not read: %w", key, err)
|
|
}
|
|
var m membership.Membership
|
|
if err := json.Unmarshal(bare, &m); err != nil {
|
|
return nil, fmt.Errorf("its own membership %s does not read: %w", key, err)
|
|
}
|
|
return m.Hears, nil
|
|
}
|
|
|
|
// boundStreams are, by module, the streams on the bus of every binding kept (BoundKey): what the bus module's process
|
|
// may still define, an orphan's among them (buscompose.BusStateNames).
|
|
func (p *Part) boundStreams() (map[string][]string, error) {
|
|
keys, err := p.Store.Keys(StateMemberships)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("its own %s cannot be listed: %w", StateMemberships, err)
|
|
}
|
|
out := map[string][]string{}
|
|
for _, key := range keys {
|
|
module, ok := strings.CutPrefix(key, BoundPrefix)
|
|
if !ok {
|
|
continue
|
|
}
|
|
b, err := p.bound(module)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
for _, bind := range b.Bindings {
|
|
out[module] = append(out[module], bindingStreams(bind)...)
|
|
}
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// bindingStreams are the streams on the bus a binding names: a stream's, a bucket's KV_, an object store's OBJ_, a ledger's
|
|
// history, its snapshot stream and its state and index buckets.
|
|
func bindingStreams(b membership.BusBinding) []string {
|
|
var out []string
|
|
if b.Stream != "" {
|
|
out = append(out, b.Stream)
|
|
}
|
|
if b.Bucket != "" {
|
|
out = append(out, "KV_"+b.Bucket)
|
|
}
|
|
if b.ObjectStore != "" {
|
|
out = append(out, "OBJ_"+b.ObjectStore)
|
|
}
|
|
if b.History != "" {
|
|
out = append(out, b.History, b.History+"_snapshot")
|
|
}
|
|
if b.State != "" {
|
|
out = append(out, "KV_"+b.State)
|
|
}
|
|
for _, ix := range b.Indexes {
|
|
out = append(out, "KV_"+ix)
|
|
}
|
|
return out
|
|
}
|
|
|
|
func firstOf(values ...string) string {
|
|
for _, v := range values {
|
|
if v != "" {
|
|
return v
|
|
}
|
|
}
|
|
return ""
|
|
}
|