package link import ( "context" "encoding/json" "errors" "fmt" "sync" "time" "git.novox.be/novox/mesh-sdk/go/health" ) // Writer is what the node-engine calls itself on the facts it says (novox/hq ADR 0227 §2, ADR 0539 §9): the program // that wrote them, beside its epoch and sequence. const Writer = "node-engine" // Session is the node-engine's hold on the bus (novox/hq ADR 0539 §9): the link open now, what the controller asks // of this machine over it, and what the machine says on it unasked: its health, its haves and its facts. // // **It applies nothing and orders nothing.** A module is applied by the node agent from its want, and the facts are an // event of the node, ordered by the node-engine's own epoch and sequence. Nothing here receives, orders or reconciles // a whole declaration. type Session struct { // Membership is the node and the mesh's key a controller's ask is verified with. Membership Membership // OnLink is run, off the link's own loop, each time a link opens and when the controller asks the node to say its // facts again. It says them with SayFacts. Nil says nothing. OnLink func(ctx context.Context) // Heard is told every facts event the mesh has taken. Nil is allowed. Heard func(Report) // HaveWritten is told each time a have key is written, or found already saying what this node-engine would write: // the first proves a node-engine on trial (novox/hq ADR 0539 §9). Nil is allowed. HaveWritten func() // HandOver judges and records a directory used as found handed to the mesh by the operator, asked by // the controller over the link (novox/hq issue 356). Nil answers every such ask with a refusal. HandOver HandOverFunc // SetuidSearch throws away the last search for setuid programs and starts a full one, asked by the // controller's terminal over the link (novox/hq issue 361). Nil answers every such ask with a refusal. SetuidSearch SetuidSearchFunc // JudgeDirectories judges this node's declared directories read-only, asked by the controller over the // link (novox/hq issue 358). Nil answers every such ask with a refusal. JudgeDirectories JudgeDirectoriesFunc // AccessAct judges and runs the take act on an access's folder, asked by the controller on an approved // warrant (novox/hq ADR 0328 §7). Nil answers every such ask with a refusal. AccessAct AccessActFunc // MoveSecret moves one value sealed to this machine into the vault, asked by the controller during the move of // the secrets (design 57's V3, issue 676). Nil answers every such ask with a refusal. MoveSecret MoveSecretFunc Say Announce Timeout time.Duration once sync.Once mu sync.Mutex bus Bus // the link open now; nil while there is none // gone is closed when the link open now closes, and changed whenever a link opens or closes (WhenLinked). gone chan struct{} changed chan struct{} // saying is held while a facts event is made and said, so they leave in the order they are made. saying sync.Mutex // epoch is this process's own: the time it started, in milliseconds, so a later process orders after an earlier one // without anything kept on disk; sequence is one higher for every facts event it says (ADR 0227 §2). epoch int64 sequence int64 // haves is what this node-engine writes beside its facts (novox/hq ADR 0429 §4, issue 1132). haves haveState } func (q *Session) init() { q.once.Do(func() { if q.Say == nil { q.Say = func(string) {} } q.epoch = time.Now().UnixMilli() q.changed = make(chan struct{}) }) } // SayFacts says this node's facts on the link open now, in the node-engine's own order, and whether the bus took them. // A facts event made with no link open is not kept: the next link says the facts as they are then. func (q *Session) SayFacts(ctx context.Context, r Report) bool { q.init() q.saying.Lock() defer q.saying.Unlock() q.mu.Lock() bus := q.bus q.mu.Unlock() if bus == nil { return false } q.sequence++ r.Node = q.Membership.Node r.Writer = Writer r.Order = Order{Epoch: q.epoch, Sequence: q.sequence} if !publishReport(context.WithoutCancel(ctx), bus, q.Membership, r, q.Say, q.Timeout) { return false } // After the facts, which are sent first and unchanged; a write the bucket refuses never undoes them (issue 1132). if q.noteApply(r) { q.writeHaves(ctx, bus) } if q.Heard != nil && r.Refused == "" { q.Heard(r) } return true } // FactsAsked is the mesh asking this node to say its facts again: run on the link open now. func (q *Session) FactsAsked(ctx context.Context) { q.init() if q.OnLink != nil { go q.OnLink(ctx) } } // Attach is a link opening: it is the link facts are said on from now. func (q *Session) Attach(ctx context.Context, bus Bus) { q.init() q.mu.Lock() q.bus = bus q.gone = make(chan struct{}) close(q.changed) q.changed = make(chan struct{}) q.mu.Unlock() q.FactsAsked(ctx) } // WithLink is ctx carrying the link now open, so an apply can fetch a composition's files given by digest (novox/hq ADR // 0486, issue 1695). A session with no link open carries none. func (q *Session) WithLink(ctx context.Context) context.Context { q.mu.Lock() bus := q.bus q.mu.Unlock() if bus == nil { return ctx } return WithFiles(ctx, bus) } // Detach is the link closing: nothing more is said on it. func (q *Session) Detach(bus Bus) { q.init() q.saying.Lock() defer q.saying.Unlock() q.mu.Lock() if q.bus == bus { q.bus = nil close(q.gone) close(q.changed) q.changed = make(chan struct{}) } q.mu.Unlock() } // Linked says whether a link to the bus is open now (novox/hq ADR 0241: the bus is one part of the // machine's networking its engine judges). func (q *Session) Linked() bool { q.init() q.mu.Lock() defer q.mu.Unlock() return q.bus != nil } // SayHealth says a health statement on the link open now (novox/hq ADR 0240, to-be 48 §4), and whether // the bus took it. A statement is a word about the machine as it is, and it must not wait behind an apply: a // container crash-looping while a long apply runs is said while it runs. False with no link open, or one that cannot // say it; the caller says it again on its next look, and the next facts event carries it whatever happens. func (q *Session) SayHealth(ctx context.Context, h Health) bool { q.init() q.mu.Lock() bus := q.bus q.mu.Unlock() sayer, ok := bus.(HealthBus) if bus == nil || !ok { return false } body, err := json.Marshal(HealthSaid{Node: q.Membership.Node, Health: h}) if err != nil { return false } saying, cancel := context.WithTimeout(ctx, q.timeout()) defer cancel() if err := sayer.Health(saying, q.Membership.Node, body); err != nil { q.Say("could not say this machine's health: " + err.Error()) return false } // A look that changed a module's health changes its have (issue 1132). if q.noteHealth(h) { q.writeHaves(ctx, bus) } return true } func (q *Session) timeout() time.Duration { if q.Timeout <= 0 { return 10 * time.Second } return q.Timeout } // AskTool asks a module's tool on this machine's own node tools, as a declared health check (novox/hq ADR // 0240, to-be 48 §3), and reads its answer: healthy or not, with why. An error when there is no link open // or nothing answered — said as the check's failure, never as healthy. func (q *Session) AskTool(ctx context.Context, module, tool string) (health.Answer, error) { q.init() q.mu.Lock() bus := q.bus q.mu.Unlock() if bus == nil { return health.Answer{}, errors.New("no link to the bus is open") } asker, ok := bus.(ToolBus) if !ok { return health.Answer{}, errors.New("the link to the bus open now cannot ask a tool") } reply, err := asker.Ask(ctx, ToolSubject(module, tool, q.Membership.Node), []byte("{}")) if err != nil { return health.Answer{}, err } return ReadToolAnswer(reply) } // AskController asks the controller a line mesh-cli was given on this machine, on this node's own subject, over // the link open now (novox/hq ADR 0272 §3), and returns its answer as it came. No link open is said, not waited for. func (q *Session) AskController(ctx context.Context, body []byte) ([]byte, error) { q.init() q.mu.Lock() bus := q.bus q.mu.Unlock() if bus == nil { return nil, fmt.Errorf("%w: this machine has no link to the bus open now", ErrNotAsked) } asker, ok := bus.(ToolBus) if !ok { return nil, fmt.Errorf("%w: the link to the bus open now cannot ask the controller", ErrNotAsked) } return asker.Ask(ctx, CLISubject(q.Membership.Node), body) } // ErrNotAsked is a request the link knows it never delivered: no link open, or nobody listening on its subject. Any // other failure of a request may come after it was delivered, so what came of it is not known (novox/hq ADR 0272). var ErrNotAsked = errors.New("not asked") // ReadToolAnswer reads a tool's reply envelope — `{result, error}`, as the node tools answer every call — // as a health answer, through the SDK's reader (novox/hq ADR 0283): healthy, unhealthy, or waits for the // operator with what it waits for. An error, and a result that is not an answer, are unhealthy; a wait the // SDK cannot check is unhealthy too, so a malformed one never excuses anything. func ReadToolAnswer(reply []byte) (health.Answer, error) { var env struct { Result json.RawMessage `json:"result"` Error string `json:"error"` } if err := json.Unmarshal(reply, &env); err != nil { return health.Answer{}, fmt.Errorf("its answer is not one: %w", err) } if env.Error != "" { return health.Unhealthy(env.Error), nil } a, err := health.Read(env.Result) if err != nil { return health.Unhealthy("it answered something other than healthy or not"), nil } return a, nil } // SeatToolSubject is where a node seat's verb is asked of one machine's holder (design 33 §4): the seat's verb subject // with the machine as its last token. func SeatToolSubject(seat, verb, node string) string { return "mesh.seat." + seat + ".tool." + verb + "." + node } // AskSeat asks the holder of a seat on this node a verb, over the link open now, and answers its result, or its refusal // as an error. The one call the node agent makes of another part: the backup before a data module moves (ADR 0429 §14). func (q *Session) AskSeat(ctx context.Context, seat, verb string, args any, within time.Duration) (json.RawMessage, error) { q.init() q.mu.Lock() bus := q.bus q.mu.Unlock() if bus == nil { return nil, fmt.Errorf("%w: this machine has no link to the bus open now", ErrNotAsked) } asker, ok := bus.(ToolBus) if !ok { return nil, fmt.Errorf("%w: the link to the bus open now cannot ask a seat", ErrNotAsked) } body, err := json.Marshal(args) if err != nil { return nil, err } asking, cancel := context.WithTimeout(ctx, within) defer cancel() reply, err := asker.Ask(asking, SeatToolSubject(seat, verb, q.Membership.Node), body) if err != nil { return nil, fmt.Errorf("%s.%s could not be asked: %w", seat, verb, err) } var env struct { Result json.RawMessage `json:"result"` Error string `json:"error"` } if err := json.Unmarshal(reply, &env); err != nil { return nil, fmt.Errorf("%s.%s answered with something that is not a tool's answer", seat, verb) } if env.Error != "" { return nil, errors.New(env.Error) } return env.Result, nil }