jochen
bcfcfaaaf3
A bus module older than its in_force answer read as not held, so a merge before mesh-catalog #166 is live would silence every ask until a person's bus upgrade. Such an answer now falls back to the check main makes (is the list sent to the bus's machine the one composed now), says the reload is not confirmed, and each case names its own remedy: push <node> for a list behind or not allowing it, a ~5 s wait for a reload, the bus upgrade when an older module's list cannot be judged.
536 lines
19 KiB
Go
536 lines
19 KiB
Go
package main
|
|
|
|
// The asker on the bus: its asks in the controller's bucket `asked`, its asks and cancels published on the
|
|
// seat under the controller's name, the verbs a warrant chooses called with the controller's grant, and the
|
|
// router's record of its asks read under its name (novox/hq ADR 0259).
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"sort"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/asks"
|
|
|
|
"github.com/novox/mesh-controller/internal/broker"
|
|
"github.com/novox/mesh-controller/internal/catalogue"
|
|
"github.com/novox/mesh-controller/internal/conditions"
|
|
"github.com/novox/mesh-controller/internal/inventory"
|
|
"github.com/novox/mesh-controller/internal/link"
|
|
)
|
|
|
|
// askerFrom is the serving controller's asker; nil in any other process.
|
|
var askerFrom *asker
|
|
|
|
// askWithin is how long a verb a warrant chose is given to answer.
|
|
const askWithin = time.Minute
|
|
|
|
type busAsked struct{ conn *nats.Conn }
|
|
|
|
func (b busAsked) kv(ctx context.Context) (jetstream.KeyValue, error) {
|
|
js, err := jetstream.New(b.conn)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return js.KeyValue(ctx, broker.AskedBucket)
|
|
}
|
|
|
|
func (b busAsked) Get(ctx context.Context, id string) (*asked, error) {
|
|
kv, err := b.kv(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
e, err := kv.Get(ctx, id)
|
|
if errors.Is(err, jetstream.ErrKeyNotFound) {
|
|
return nil, nil
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var r asked
|
|
return &r, json.Unmarshal(e.Value(), &r)
|
|
}
|
|
|
|
// Create keeps a new ask under its id, and only where none is kept: never over another.
|
|
func (b busAsked) Create(ctx context.Context, r asked) error {
|
|
kv, err := b.kv(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
body, err := json.Marshal(r)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, err = kv.Create(ctx, r.ID, body)
|
|
return err
|
|
}
|
|
|
|
// Change applies change to the ask kept under id by compare-and-set on its key's revision (the review of
|
|
// 2026-10-09, L2): read, changed, and written only over the revision read; when another write came between,
|
|
// read again and asked again, at most askChangeTries times. change says whether to write at all.
|
|
func (b busAsked) Change(ctx context.Context, id string, change func(*asked) bool) (bool, error) {
|
|
kv, err := b.kv(ctx)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
for try := 0; try < askChangeTries; try++ {
|
|
e, err := kv.Get(ctx, id)
|
|
if errors.Is(err, jetstream.ErrKeyNotFound) {
|
|
return false, nil
|
|
}
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
var r asked
|
|
if err := json.Unmarshal(e.Value(), &r); err != nil {
|
|
return false, err
|
|
}
|
|
if !change(&r) {
|
|
return false, nil
|
|
}
|
|
body, err := json.Marshal(r)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
if _, err := kv.Update(ctx, id, body, e.Revision()); err != nil {
|
|
var api *jetstream.APIError
|
|
if errors.Is(err, jetstream.ErrKeyExists) || (errors.As(err, &api) && api.ErrorCode == jetstream.JSErrCodeStreamWrongLastSequence) {
|
|
continue
|
|
}
|
|
return false, err
|
|
}
|
|
return true, nil
|
|
}
|
|
return false, fmt.Errorf("the ask %s changed under every one of %d tries", id, askChangeTries)
|
|
}
|
|
|
|
func (b busAsked) All(ctx context.Context) ([]asked, error) {
|
|
kv, err := b.kv(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
lister, err := kv.ListKeys(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer func() { _ = lister.Stop() }()
|
|
var out []asked
|
|
for k := range lister.Keys() {
|
|
e, err := kv.Get(ctx, k)
|
|
if err != nil {
|
|
continue
|
|
}
|
|
var r asked
|
|
if json.Unmarshal(e.Value(), &r) == nil {
|
|
out = append(out, r)
|
|
}
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// callAction performs an action's verb as the controller, through the grant that names it.
|
|
func callAction(conn *nats.Conn) func(ctx context.Context, a conditions.Action, args map[string]string) error {
|
|
return func(ctx context.Context, a conditions.Action, args map[string]string) error {
|
|
seat, verb, ok := strings.Cut(a.Verb, ".")
|
|
if !ok {
|
|
return fmt.Errorf("%q names no seat and verb", a.Verb)
|
|
}
|
|
body := map[string]any{}
|
|
for k, v := range args {
|
|
body[k] = v
|
|
}
|
|
if seat == catalogue.DeliverySeat {
|
|
_, err := askDeliveryOwner(ctx, conn, verb, body)
|
|
return err
|
|
}
|
|
granted := false
|
|
for _, v := range broker.VerbsTheControllerActsOnAWarrant {
|
|
granted = granted || (v.Seat == seat && v.Verb == verb)
|
|
}
|
|
if !granted {
|
|
return fmt.Errorf("%s: %w", a.Verb, errNotGranted)
|
|
}
|
|
var answer link.Answer
|
|
var err error
|
|
if a.Machine != "" {
|
|
answer, err = link.AskSeatTool(ctx, conn, seat, verb, a.Machine, body, askWithin)
|
|
} else {
|
|
answer, err = link.AskMeshSeatTool(ctx, conn, seat, verb, body, askWithin)
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if answer.Error != "" {
|
|
return fmt.Errorf("%s refused: %s", a.Verb, answer.Error)
|
|
}
|
|
return nil
|
|
}
|
|
}
|
|
|
|
// routerRecordOf reads the router's record of one of the controller's asks, under its name, and answers
|
|
// how it ended when it did: the bucket is the one the asks seat's declarer names as its records.
|
|
func routerRecordOf(conn *nats.Conn, inv *inventory.Inventory) func(ctx context.Context, id string) (*asks.Warrant, error) {
|
|
return func(ctx context.Context, id string) (*asks.Warrant, error) {
|
|
bucket, err := asksRecords(ctx, inv)
|
|
if err != nil || bucket == "" {
|
|
return nil, err
|
|
}
|
|
reply, err := conn.RequestWithContext(ctx, "$JS.API.DIRECT.GET.KV_"+bucket+".$KV."+bucket+"."+askerName+"."+id, nil)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if reply.Header.Get("Status") != "" {
|
|
return nil, nil // none, or not readable: the event says it
|
|
}
|
|
var rec struct {
|
|
State string `json:"state"`
|
|
Warrant *asks.Warrant `json:"warrant"`
|
|
}
|
|
if json.Unmarshal(reply.Data, &rec) != nil || rec.State == "open" || rec.Warrant == nil {
|
|
return nil, nil
|
|
}
|
|
return rec.Warrant, nil
|
|
}
|
|
}
|
|
|
|
// asksRecords is the bucket the asks seat's declarer keeps its record of asks in.
|
|
func asksRecords(ctx context.Context, inv *inventory.Inventory) (string, error) {
|
|
declared, err := inv.Catalogue(ctx)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
for _, m := range declared {
|
|
for _, s := range m.DefinesSeats {
|
|
if s.Name == broker.AsksSeat && len(s.Records) > 0 {
|
|
return broker.BucketName(m.Module, s.Records[0]), nil
|
|
}
|
|
}
|
|
}
|
|
return "", nil
|
|
}
|
|
|
|
// routerHereIn says whether a module declaring the asks seat, with its ask named by its caller, is assigned:
|
|
// without it nothing takes an ask, and asking would only fill a queue nobody reads.
|
|
func routerHereIn(inv *inventory.Inventory) func(ctx context.Context) (bool, error) {
|
|
return func(ctx context.Context) (bool, error) {
|
|
entries, err := inv.Catalogued(ctx)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
for _, e := range entries {
|
|
for _, s := range e.Manifest.DefinesSeats {
|
|
if s.Name == broker.AsksSeat && s.NamedByCaller("ask") && len(e.On) > 0 {
|
|
return true, nil
|
|
}
|
|
}
|
|
}
|
|
return false, nil
|
|
}
|
|
}
|
|
|
|
// grantHeldIn says whether the bus holds the controller's grant to ask (novox/hq issue 353), judged on the two
|
|
// facts that make it so, and failing closed on each:
|
|
//
|
|
// 1. **composed**: the user list the mesh composes now grants the controller itself to publish its ask
|
|
// (controllerMayAsk) — the controller's own grant, not whether any user of the list changed;
|
|
// 2. **in force**: the bus's own module, on the machine holding the bus, says the user list there allows it
|
|
// and the server reloaded after that list was written (the nats module's `nats_user_can`, `in_force`) — a
|
|
// list sent is not a list the bus enforces until it reloads.
|
|
//
|
|
// Anything that cannot be worked out is not held, with why, so the conditions that need the operator are said
|
|
// as undelivered rather than asked into a refusal.
|
|
func grantHeldIn(open *stores, conn *nats.Conn) func(ctx context.Context) (bool, string, error) {
|
|
return grantHeldBy(composedGrantIn(open), busGrantIn(open, conn), time.Now)
|
|
}
|
|
|
|
// grantJudge is one of the two facts: whether it holds, and why not.
|
|
type grantJudge func(ctx context.Context) (bool, string, error)
|
|
|
|
// grantHeldBy is the judgement over the two facts, remembered for grantLookEvery: composing the list resolves
|
|
// the bus's machine whole, and asking the bus's module is a call.
|
|
func grantHeldBy(composed, inForce grantJudge, now func() time.Time) func(ctx context.Context) (bool, string, error) {
|
|
var mu sync.Mutex
|
|
var at time.Time
|
|
var held bool
|
|
var why string
|
|
return func(ctx context.Context) (bool, string, error) {
|
|
mu.Lock()
|
|
defer mu.Unlock()
|
|
if !at.IsZero() && now().Sub(at) < grantLookEvery {
|
|
return held, why, nil
|
|
}
|
|
held, why = judgeGrant(ctx, composed, inForce)
|
|
at = now()
|
|
return held, why, nil
|
|
}
|
|
}
|
|
|
|
// judgeGrant is held only when both facts hold; an error is a fact not known, never a yes. A fact that holds
|
|
// with a caveat (a reload not confirmed) gives that caveat as the why of a yes.
|
|
func judgeGrant(ctx context.Context, composed, inForce grantJudge) (bool, string) {
|
|
var caveat []string
|
|
for _, fact := range []struct {
|
|
judge grantJudge
|
|
what string
|
|
}{{composed, "whether the mesh composes the controller a grant to ask"}, {inForce, "whether the bus enforces it"}} {
|
|
if fact.judge == nil {
|
|
return false, fact.what + " is not judged here"
|
|
}
|
|
ok, why, err := fact.judge(ctx)
|
|
if err != nil {
|
|
return false, fact.what + " could not be worked out: " + err.Error()
|
|
}
|
|
if !ok {
|
|
return false, why
|
|
}
|
|
if why != "" {
|
|
caveat = append(caveat, why)
|
|
}
|
|
}
|
|
return true, strings.Join(caveat, "; ")
|
|
}
|
|
|
|
// controllerMayAsk says whether the controller's own grant, as composed from these records, lets it publish its
|
|
// ask (ADR 0259 §3: composed while a router that takes asks under its asker's name is assigned).
|
|
func controllerMayAsk(records broker.Records) (bool, error) {
|
|
users, err := broker.Users(records)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
for _, u := range users {
|
|
if u.Kind != broker.KindController {
|
|
continue
|
|
}
|
|
perms, err := broker.PermissionsFor(u)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
return broker.MayPublish(perms, asks.AskSubject(askerName)), nil
|
|
}
|
|
return false, errors.New("the composition holds no controller")
|
|
}
|
|
|
|
// composedGrantIn is the first fact, from the mesh's records.
|
|
func composedGrantIn(open *stores) grantJudge {
|
|
return func(ctx context.Context) (bool, string, error) {
|
|
if open == nil || open.inventory == nil {
|
|
return false, "", errors.New("the mesh's records are not open")
|
|
}
|
|
records, err := open.inventory.BusRecords(ctx)
|
|
if err != nil {
|
|
return false, "", err
|
|
}
|
|
may, err := controllerMayAsk(records)
|
|
if err != nil || may {
|
|
return may, "", err
|
|
}
|
|
return false, "the mesh composes the controller no grant to ask: no router that takes asks under its " +
|
|
"asker's name is assigned", nil
|
|
}
|
|
}
|
|
|
|
// busGrantIn is the second fact, asked of the bus's own module on the machine holding the bus, with the
|
|
// judgement main made before it (novox/hq issue 353, PR 186: is the user list sent to the bus's machine the one
|
|
// composed now) as the floor for a bus module too old to say whether a grant is in force.
|
|
func busGrantIn(open *stores, conn *nats.Conn) grantJudge {
|
|
ask := func(ctx context.Context) (string, json.RawMessage, error) {
|
|
if open == nil || open.inventory == nil || conn == nil {
|
|
return "", nil, errors.New("the mesh's records or the bus are not open")
|
|
}
|
|
holders, err := seatHolders(ctx, open.inventory)
|
|
if err != nil {
|
|
return "", nil, err
|
|
}
|
|
h, held := holders[theBrokerSeat]
|
|
if !held || h.Node == "" || h.Module == "" {
|
|
return "", nil, nil
|
|
}
|
|
a, err := link.AskModuleToolOn(ctx, conn, h.Module, busUserCanTool, h.Node, map[string]any{
|
|
"user": broker.ControllerName, "action": "publish", "subject": asks.AskSubject(askerName)}, 15*time.Second)
|
|
if err != nil {
|
|
return h.Node, nil, err
|
|
}
|
|
if a.Error != "" {
|
|
return h.Node, nil, errors.New(a.Error)
|
|
}
|
|
return h.Node, a.Result, nil
|
|
}
|
|
behind := func(ctx context.Context) (string, bool, error) {
|
|
if open == nil || open.inventory == nil {
|
|
return "", false, errors.New("the mesh's records are not open")
|
|
}
|
|
return brokerBehind(ctx, open, nil)
|
|
}
|
|
return busGrantBy(ask, behind)
|
|
}
|
|
|
|
// busGrantBy judges the bus's word, and names for each way it is not held the one thing that makes it held:
|
|
//
|
|
// - **the list there does not grant it**: `push <node>` sends the list composed now;
|
|
// - **written, not reloaded**: the bus reloads it in place within about 5 s, so the remedy is to wait;
|
|
// - **the module is older than `in_force`**: the bus cannot confirm a reload, so the floor is main's check
|
|
// before it — held while the list sent there is the one composed now, `push <node>` while it is behind, and
|
|
// the bus upgrade (which carries a module that answers `in_force`) when even that cannot be worked out. A
|
|
// bus module too old is not a reason to ask nothing: an order of merges that leaves it so would silence
|
|
// every ask until a person's bus upgrade (the review of PR 188).
|
|
//
|
|
// ask answers the machine holding the bus and its module's answer ("" when nothing holds the bus); behind is
|
|
// whether the list sent to that machine is behind what the mesh composes.
|
|
func busGrantBy(ask func(ctx context.Context) (string, json.RawMessage, error),
|
|
behind func(ctx context.Context) (string, bool, error)) grantJudge {
|
|
return func(ctx context.Context) (bool, string, error) {
|
|
node, raw, err := ask(ctx)
|
|
if err != nil {
|
|
return false, "", err
|
|
}
|
|
if node == "" {
|
|
return false, "no module holds the bus, so nothing says what it enforces", nil
|
|
}
|
|
answer, err := readBusAnswer(raw)
|
|
if err != nil {
|
|
return false, "", fmt.Errorf("the bus module's answer on %s could not be read: %w", node, err)
|
|
}
|
|
pushIt := func(what string) string {
|
|
return fmt.Sprintf("%s, so the bus has not been given the controller's grant to ask; `push %s` carries it",
|
|
what, node)
|
|
}
|
|
switch {
|
|
case answer.InForce == nil:
|
|
const older = "the bus module there is older than the answer whether a grant is in force, so its " +
|
|
"reload is not confirmed"
|
|
if !answer.Allowed {
|
|
return false, pushIt(fmt.Sprintf("the user list on %s does not allow the controller to ask", node)) +
|
|
" (" + older + ")", nil
|
|
}
|
|
machine, late, err := behind(ctx)
|
|
if err != nil {
|
|
return false, "", fmt.Errorf("%s, and whether the list sent to %s is the one composed now could "+
|
|
"not be worked out (%v); the bus upgrade (`bus upgrade`) carries a module that says", older,
|
|
node, err)
|
|
}
|
|
if late {
|
|
if machine == "" {
|
|
machine = node
|
|
}
|
|
return false, fmt.Sprintf("the bus's user list on %s is behind what the mesh composes, so the bus has "+
|
|
"not been given the controller's grant to ask; `push %s` carries it (%s)", machine, machine, older), nil
|
|
}
|
|
return true, fmt.Sprintf("the user list on %s is the one composed now and allows the controller to ask; "+
|
|
"%s, and the bus upgrade (`bus upgrade`) carries a module that confirms it", node, older), nil
|
|
case !*answer.InForce && answer.Allowed:
|
|
why := "the server has not reloaded since it was written"
|
|
if answer.Why != "" {
|
|
why = answer.Why
|
|
}
|
|
return false, fmt.Sprintf("the bus on %s has the controller's grant to ask written but not in force yet "+
|
|
"(%s); the bus reloads it in place within about 5 s, so nothing is to be done but wait", node, why), nil
|
|
case !answer.Allowed:
|
|
return false, pushIt(fmt.Sprintf("the user list on %s does not allow the controller to ask", node)), nil
|
|
}
|
|
return true, "", nil
|
|
}
|
|
}
|
|
|
|
// busUserCanTool is the bus module's answer to whether a user may do something, and whether it is in force.
|
|
const busUserCanTool = "nats_user_can"
|
|
|
|
// busAnswer is the bus module's answer to busUserCanTool; InForce is nil from a module older than that field.
|
|
type busAnswer struct {
|
|
Allowed bool `json:"allowed"`
|
|
InForce *bool `json:"in_force"`
|
|
Why string `json:"why"`
|
|
}
|
|
|
|
func readBusAnswer(raw json.RawMessage) (busAnswer, error) {
|
|
var answer busAnswer
|
|
err := json.Unmarshal(raw, &answer)
|
|
return answer, err
|
|
}
|
|
|
|
// grantLookEvery is how often the bus's user list is judged against the one its machine was last sent.
|
|
const grantLookEvery = 30 * time.Second
|
|
|
|
// channelsIn is what the channels are now, as a fingerprint: each module claiming a kind of the channel
|
|
// bench, where, promising what, and whether of its own account. An ask the router refused is asked again
|
|
// once this changes.
|
|
func channelsIn(inv *inventory.Inventory) func(ctx context.Context) string {
|
|
return func(ctx context.Context) string {
|
|
entries, err := inv.Catalogued(ctx)
|
|
if err != nil {
|
|
return ""
|
|
}
|
|
var parts []string
|
|
for _, e := range entries {
|
|
for _, c := range e.Manifest.Claims {
|
|
if c.Kind == "" || !catalogue.KindedBenches[c.Name] {
|
|
continue
|
|
}
|
|
on := append([]string(nil), e.On...)
|
|
sort.Strings(on)
|
|
caps := append([]string(nil), c.Capabilities...)
|
|
sort.Strings(caps)
|
|
parts = append(parts, fmt.Sprintf("%s/%s=%s@%s[%s]own:%t", c.Name, c.Kind, e.Manifest.Module,
|
|
strings.Join(on, ","), strings.Join(caps, ","), e.Manifest.RunsAs != ""))
|
|
}
|
|
}
|
|
sort.Strings(parts)
|
|
return strings.Join(parts, ";")
|
|
}
|
|
}
|
|
|
|
// startAsking makes the serving controller's asker and hands it the router's words.
|
|
func startAsking(ctx context.Context, open *stores, server *link.Server, conn *nats.Conn, keeper *conditions.Keeper) {
|
|
js, err := jetstream.New(conn)
|
|
if err != nil {
|
|
fmt.Printf("the operator cannot be asked: %v\n", err)
|
|
return
|
|
}
|
|
a := servingAsker(open, conn, js, keeper)
|
|
if err := server.Decides(a); err != nil {
|
|
fmt.Printf("the operator's answers cannot be heard, so nothing is asked: %v\n", err)
|
|
return
|
|
}
|
|
askerFrom = a
|
|
go a.keep(ctx)
|
|
}
|
|
|
|
// servingAsker is the serving controller's asker, wired to the mesh: what it reads, publishes, performs and
|
|
// records, and the judgements it asks by — the bus's grant to ask among them (novox/hq issue 353).
|
|
func servingAsker(open *stores, conn *nats.Conn, js jetstream.JetStream, keeper *conditions.Keeper) *asker {
|
|
return &asker{
|
|
open: keeper.Open,
|
|
silence: func(ctx context.Context, key string, d time.Duration, by, why string) error {
|
|
_, err := keeper.Silence(ctx, key, d, by, why)
|
|
return err
|
|
},
|
|
store: busAsked{conn: conn},
|
|
publish: func(ctx context.Context, subject string, body []byte, id string) error {
|
|
_, err := js.Publish(ctx, subject, body, jetstream.WithMsgID(id))
|
|
return err
|
|
},
|
|
call: callAction(conn),
|
|
// A proposed settings layer is set by this controller itself on the warrant (novox/hq ADR 0277).
|
|
setLayer: setLayerIn(open),
|
|
record: func(ctx context.Context, act link.HandAct) error {
|
|
_, err := link.RecordHandAct(ctx, conn, act)
|
|
return err
|
|
},
|
|
routerRecord: routerRecordOf(conn, open.inventory),
|
|
routerHere: routerHereIn(open.inventory),
|
|
grantHeld: grantHeldIn(open, conn),
|
|
channels: channelsIn(open.inventory),
|
|
raise: func(ctx context.Context, obs []conditions.Observation) error {
|
|
return keeper.Reconcile(ctx, sourceAsker, obs)
|
|
},
|
|
now: time.Now,
|
|
logf: func(format string, args ...any) { fmt.Printf(format+"\n", args...) },
|
|
}
|
|
}
|