jschoubben
63602264ef
A module's containers that declare identity run as an identity the controller allocates once, mesh-wide, from 50000-59999; each access names its users and whether its folder is taken; an access wait is excused for a folder the mesh may not keep yet, and the operator is asked once per folder and machine, the folder taken only on the warrant. All of it is sent only to an engine at contract 5. A placement for an access a newer build brings is judged against that build (issue 575). mesh-host pinned to the ADR 0325 branch head.
611 lines
22 KiB
Go
611 lines
22 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 check
|
|
// main made before it (novox/hq issue 353, PR 186: is the user list sent to the bus's machine the one composed
|
|
// now) kept for the cases the module's word does not settle.
|
|
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
|
|
}
|
|
list := func(ctx context.Context) (busListState, error) {
|
|
if open == nil || open.inventory == nil {
|
|
return busListState{}, errors.New("the mesh's records are not open")
|
|
}
|
|
s, err := brokerListState(ctx, open, nil)
|
|
if err != nil || s.Machine == "" || !s.Behind {
|
|
return s, err
|
|
}
|
|
step, err := busHeld(ctx, open.inventory, []string{s.Machine})
|
|
if err != nil {
|
|
return busListState{}, err
|
|
}
|
|
s.BusStep = step[s.Machine]
|
|
return s, nil
|
|
}
|
|
return busGrantBy(ask, list)
|
|
}
|
|
|
|
// busListState is the user list on the bus's machine as the controller's record knows it: whether the list
|
|
// sent there is behind the one composed now, why it could not be composed, and the bus step a send there waits
|
|
// on (that machine cannot be pushed until a person's `bus upgrade`).
|
|
type busListState struct {
|
|
Machine string
|
|
Behind bool
|
|
Uncomposed error
|
|
BusStep string
|
|
}
|
|
|
|
// remedy is the one thing that sends the bus's machine the list composed now: the plan's own reason when it
|
|
// cannot be composed, the bus upgrade while sends there wait on it, and otherwise a push of that machine.
|
|
func (s busListState) remedy(node string) string {
|
|
if s.Machine != "" {
|
|
node = s.Machine
|
|
}
|
|
switch {
|
|
case s.Uncomposed != nil:
|
|
return fmt.Sprintf("the user list for %s cannot be composed (%v); `plan %s` says what to fix first",
|
|
node, s.Uncomposed, node)
|
|
case s.BusStep != "":
|
|
return fmt.Sprintf("sends to %s wait on the bus's planned step, so a push will not carry it: %s", node, s.BusStep)
|
|
}
|
|
return fmt.Sprintf("`push %s` carries it", node)
|
|
}
|
|
|
|
// busGrantBy judges the bus's word, and names for each way it is not held the one thing that makes it held:
|
|
//
|
|
// - **in force**: held;
|
|
// - **allowed, from a module older than `in_force`**: held, with the caveat that the reload is not confirmed.
|
|
// The file allows the publish, and the bus reloads within about 5 s of a change to it, so the worst case is
|
|
// one ask refused in those seconds, which is asked again (issue 369). The list check is not consulted:
|
|
// while a bus step is pending the bus's machine cannot be pushed, so any change to the composed list would
|
|
// read as behind for good and silence every ask until a person's bus upgrade (the review of PR 188);
|
|
// - **allowed, written, not reloaded** (the module gives both times): wait about 5 s;
|
|
// - **allowed, in force not known** (the module could not read a time): what is not known, said;
|
|
// - **not allowed, or the module cannot answer**: the list check decides — the remedy that sends the list
|
|
// when it is behind (busListState.remedy), and, for a module that cannot answer, held while the list sent
|
|
// is current, as main judged it.
|
|
//
|
|
// ask answers the machine holding the bus and its module's answer ("" when nothing holds the bus); list is the
|
|
// record's word on the list sent to that machine.
|
|
func busGrantBy(ask func(ctx context.Context) (string, json.RawMessage, error),
|
|
list func(ctx context.Context) (busListState, error)) grantJudge {
|
|
return func(ctx context.Context) (bool, string, error) {
|
|
node, raw, askErr := ask(ctx)
|
|
if askErr == nil && node == "" {
|
|
return false, "no module holds the bus, so nothing says what it enforces", nil
|
|
}
|
|
var answer busAnswer
|
|
if askErr == nil {
|
|
if answer, askErr = readBusAnswer(raw); askErr != nil {
|
|
askErr = fmt.Errorf("its answer could not be read: %w", askErr)
|
|
}
|
|
}
|
|
if askErr != nil {
|
|
if node == "" {
|
|
node = "the bus's machine"
|
|
}
|
|
s, err := list(ctx)
|
|
if err != nil {
|
|
return false, "", fmt.Errorf("the bus module on %s could not answer (%v), and whether the list sent "+
|
|
"there is the one composed now could not be worked out: %w", node, askErr, err)
|
|
}
|
|
if s.Behind {
|
|
return false, fmt.Sprintf("the bus module on %s could not answer (%v), and the user list sent there is "+
|
|
"behind what the mesh composes; %s", node, askErr, s.remedy(node)), nil
|
|
}
|
|
return true, fmt.Sprintf("the bus module on %s could not answer (%v); judged by the user list sent there, "+
|
|
"which is the one composed now", node, askErr), nil
|
|
}
|
|
notAllowed := func(older string) (bool, string, error) {
|
|
s, err := list(ctx)
|
|
if err != nil {
|
|
return false, "", fmt.Errorf("the user list on %s does not allow the controller to ask%s, and whether "+
|
|
"the list sent there is the one composed now could not be worked out: %w", node, older, err)
|
|
}
|
|
if s.Behind {
|
|
return false, fmt.Sprintf("the user list on %s does not allow the controller to ask%s: it is behind "+
|
|
"what the mesh composes; %s", node, older, s.remedy(node)), nil
|
|
}
|
|
return false, fmt.Sprintf("the user list on %s does not allow the controller to ask%s, though the list "+
|
|
"sent there is the one composed now; %s again", node, older, s.remedy(node)), nil
|
|
}
|
|
switch {
|
|
case answer.InForce == nil && answer.Allowed:
|
|
return true, fmt.Sprintf("the user list on %s allows the controller to ask, but its reload is not "+
|
|
"confirmed: the bus module there is older than the answer whether a grant is in force, and the bus "+
|
|
"upgrade (`bus upgrade`) carries one that confirms it", node), nil
|
|
case answer.InForce == nil:
|
|
return notAllowed(" (the bus module there is older than the answer whether a grant is in force)")
|
|
case !answer.Allowed:
|
|
return notAllowed("")
|
|
case *answer.InForce:
|
|
return true, "", nil
|
|
case answer.FileWritten != "" && answer.ServerLoaded != "":
|
|
return false, fmt.Sprintf("the bus on %s has the controller's grant to ask written (%s) but has not "+
|
|
"reloaded since (last load %s); it reloads in place within about 5 s, so nothing is to be done but wait",
|
|
node, answer.FileWritten, answer.ServerLoaded), nil
|
|
}
|
|
why := answer.Why
|
|
if why == "" {
|
|
why = "it gives no reason"
|
|
}
|
|
return false, fmt.Sprintf("the user list on %s allows the controller to ask, but whether the bus enforces it "+
|
|
"is not known: %s", node, why), 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,
|
|
// and FileWritten and ServerLoaded are given only when it could read both times (mesh-catalog's nats InForce).
|
|
type busAnswer struct {
|
|
Allowed bool `json:"allowed"`
|
|
InForce *bool `json:"in_force"`
|
|
Why string `json:"why"`
|
|
FileWritten string `json:"file_written"`
|
|
ServerLoaded string `json:"server_loaded"`
|
|
}
|
|
|
|
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),
|
|
// A shared folder waiting to be kept is asked about once, and taken on the warrant (novox/hq ADR 0325 §8).
|
|
folders: foldersIn(open),
|
|
take: takeFolderIn(open, func(ctx context.Context, act link.AccessActAsk) (link.HandOverAnswer, error) {
|
|
ident, err := open.Identity(ctx)
|
|
if err != nil {
|
|
return link.HandOverAnswer{}, fmt.Errorf("the mesh's signing key cannot be read, so nothing was asked of "+
|
|
"%s: %w", act.Node, err)
|
|
}
|
|
return link.AskTake(ctx, conn, ident, act, link.TakeWithin)
|
|
}),
|
|
machines: func(ctx context.Context) ([]string, error) {
|
|
nodes, err := open.inventory.Nodes(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
names := make([]string, 0, len(nodes))
|
|
for _, n := range nodes {
|
|
names = append(names, n.Name)
|
|
}
|
|
return names, nil
|
|
},
|
|
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...) },
|
|
}
|
|
}
|