jschoubben
db86dd52c8
mesh/delivery-group group feat/526-a-provider-delivers-what-its-software-issued rejected: a member's own check failed
mesh/delivery superseded: a newer head of the same pull request
mesh/merge-gate pass: builds build-agent, mesh-controller → ace, g14, novox, shanks; no bus step; every machine composes with the change as it did without …
mesh/repo-check pass: its merge-check.sh passed
A token a media server got from its vendor reached its consumers only as a value the mesh minted, which the server refused, or as a copy typed by hand. An offer may now say its software issues the credential; the new issued verb carries no value and asks the provider's tool, in the controller's own name, for the value sealed to one call, then seals it to every pair that does not hold it yet.
270 lines
10 KiB
Go
270 lines
10 KiB
Go
package link
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
|
|
"github.com/novox/mesh-controller/internal/broker"
|
|
)
|
|
|
|
// What becomes of a consumer the mesh stopped asking for (novox/hq ADR 0230).
|
|
//
|
|
// **A consumer is retired, not withdrawn, and deleted only by a person.** A provider that has seen the
|
|
// same consumers go unasked for in five passes disables their access and keeps their data, marked in
|
|
// its backend with when and why; asked for again, it enables them as they were. When more would go
|
|
// than its bound allows — more than three, or more than half of those it holds — it retires nothing and
|
|
// waits for `retire approve` or `retire reject`. A retired consumer is deleted only by `cleanup delete`,
|
|
// which the provider executes. Each of these is the provider's `provisioner.retirement` event, its
|
|
// `change` saying which; the controller keeps the ones that need a person as conditions.
|
|
|
|
// The changes a provider says.
|
|
const (
|
|
RetireWaiting = "waiting"
|
|
RetireSettled = "settled"
|
|
RetireApproved = "approved"
|
|
RetireRejected = "rejected"
|
|
RetireRetired = "retired"
|
|
RetireReenabled = "reenabled"
|
|
RetireDeleted = "deleted"
|
|
RetireAdopted = "adopted"
|
|
)
|
|
|
|
// ViaController is what the controller's verbs say they came through, so an act a provider was asked
|
|
// for some other way is told apart and recorded by hand.
|
|
const ViaController = "mesh-controller"
|
|
|
|
// RetiredConsumer is one consumer in a retirement word, or in a provider's answer.
|
|
type RetiredConsumer struct {
|
|
Consumer string `json:"consumer"`
|
|
Node string `json:"node,omitempty"`
|
|
RetiredAt string `json:"retired_at,omitempty"`
|
|
Why string `json:"why,omitempty"`
|
|
// SizeBytes is what it keeps on the backend; -1 or absent where the backend cannot say.
|
|
SizeBytes *int64 `json:"size_bytes,omitempty"`
|
|
// Kind is "consumer", or a backend's own word for something set aside ("set-aside-database").
|
|
Kind string `json:"kind,omitempty"`
|
|
// Access is "disabled" when the provider disabled it, "kept" for a mark-only provider whose consumer
|
|
// keeps its access until a person deletes it (ADR 0230); absent is disabled.
|
|
Access string `json:"access,omitempty"`
|
|
}
|
|
|
|
// Retirement is one provider's word about consumers the mesh stopped asking for.
|
|
type Retirement struct {
|
|
// Module is the emitter, read from the subject the bus let it publish on — never from the body.
|
|
Module string `json:"-"`
|
|
|
|
Provider string `json:"provider"`
|
|
ProviderNode string `json:"provider-node"`
|
|
Change string `json:"change"`
|
|
Consumers []RetiredConsumer `json:"consumers"`
|
|
Held int `json:"held"`
|
|
Bound string `json:"bound,omitempty"`
|
|
Why string `json:"why,omitempty"`
|
|
By string `json:"by,omitempty"`
|
|
Via string `json:"via,omitempty"`
|
|
At time.Time `json:"at"`
|
|
Since time.Time `json:"since,omitempty"`
|
|
}
|
|
|
|
// Names are the consumers it names, in its order.
|
|
func (r Retirement) Names() []string {
|
|
out := make([]string, 0, len(r.Consumers))
|
|
for _, c := range r.Consumers {
|
|
out = append(out, c.Consumer)
|
|
}
|
|
return out
|
|
}
|
|
|
|
// Retirements keeps what providers say about consumers they retire.
|
|
type Retirements interface {
|
|
// Retired records one word. An error the store is away for is held and asked again, like a report.
|
|
Retired(ctx context.Context, r Retirement) error
|
|
}
|
|
|
|
// KeepsRetirements says where retirement words are kept, and asks for them to be delivered.
|
|
func (s *Server) KeepsRetirements(r Retirements) error {
|
|
if err := s.inbound.Also(KindProvisioner); err != nil {
|
|
return err
|
|
}
|
|
s.retirements = r
|
|
return nil
|
|
}
|
|
|
|
// IsRetirement says a subject is a provider's retirement word.
|
|
func IsRetirement(subject string) bool {
|
|
_, ok := ProvisionerEmitter(subject)
|
|
return ok && strings.HasSuffix(subject, ".event."+broker.ProvisionerRetirement)
|
|
}
|
|
|
|
// ReadRetirement is one retirement word as the controller understands it.
|
|
func ReadRetirement(subject string, body []byte) (Retirement, error) {
|
|
module, ok := ProvisionerEmitter(subject)
|
|
if !ok || !IsRetirement(subject) {
|
|
return Retirement{}, fmt.Errorf("%s is not a provider's retirement word", subject)
|
|
}
|
|
var r Retirement
|
|
if err := json.Unmarshal(body, &r); err != nil {
|
|
return Retirement{}, fmt.Errorf("%s's retirement word could not be read: %w", module, err)
|
|
}
|
|
switch r.Change {
|
|
case RetireWaiting, RetireSettled, RetireApproved, RetireRejected, RetireRetired, RetireReenabled,
|
|
RetireDeleted, RetireAdopted:
|
|
default:
|
|
return Retirement{}, fmt.Errorf("%s said a retirement change the mesh has no name for: %q", module, r.Change)
|
|
}
|
|
if r.ProviderNode == "" {
|
|
return Retirement{}, fmt.Errorf("%s's retirement word named no machine", module)
|
|
}
|
|
r.Module = module
|
|
return r, nil
|
|
}
|
|
|
|
// retirement acts on one retirement word. Like a recovery, a word that clears a condition is said
|
|
// once, so a store that is away holds the message rather than dropping it.
|
|
func (s *Server) retirement(ctx context.Context, m Control) {
|
|
if s.retirements == nil {
|
|
_ = m.Took()
|
|
return
|
|
}
|
|
r, err := ReadRetirement(m.Subject(), m.Body())
|
|
if err != nil {
|
|
s.log.Printf("%v; ignored", err)
|
|
_ = m.Took()
|
|
return
|
|
}
|
|
err = s.retirements.Retired(ctx, r)
|
|
what := fmt.Sprintf("%s's retirement word (%s)", r.Module, r.Change)
|
|
switch s.decide(ctx, m, what, "", "", err) {
|
|
case Hold:
|
|
return
|
|
case Stale, GiveUp:
|
|
_ = m.Took()
|
|
return
|
|
}
|
|
if err != nil {
|
|
s.log.Printf("%s could not be kept: %v", what, err)
|
|
} else {
|
|
s.log.Printf("%s on %s %s %s", r.Module, r.ProviderNode, r.Change, strings.Join(r.Names(), ", "))
|
|
}
|
|
_ = m.Took()
|
|
}
|
|
|
|
// The tools every provider serves about its retired consumers (ADR 0230), asked of one machine.
|
|
const (
|
|
ToolRetirement = "provisioner_retirement"
|
|
ToolRetireApprove = "provisioner_retire_approve"
|
|
ToolRetireReject = "provisioner_retire_reject"
|
|
ToolRetiredDelete = "provisioner_delete"
|
|
)
|
|
|
|
// ErrNothingServes is a tool nothing on that machine serves: the module is not running there, or
|
|
// is older than the tool.
|
|
var ErrNothingServes = errors.New("nothing serves it")
|
|
|
|
// ModuleToolOn is where one machine's instance of a module answers a tool: the module's tool subject
|
|
// with the machine as its last token — the subject the node tools bind beside the shared one.
|
|
func ModuleToolOn(module, tool, node string) string { return ToolSubject(module, tool) + "." + node }
|
|
|
|
// AskModuleToolOn asks one machine's instance of a module one of its tools and reads its answer.
|
|
func AskModuleToolOn(ctx context.Context, conn *nats.Conn, module, tool, node string, args any,
|
|
timeout time.Duration) (Answer, error) {
|
|
return askModuleToolAt(ctx, conn, ModuleToolOn(module, tool, node), module, tool, node, args, timeout)
|
|
}
|
|
|
|
// AskModuleToolOnAs is AskModuleToolOn in the caller's own name: on the subject that names its caller (novox/hq
|
|
// issue 365), which the bus lets only that user publish, so the tool knows who asks from the subject alone. The
|
|
// one way a tool can answer the controller and nobody else (novox/hq ADR 0305 §2).
|
|
func AskModuleToolOnAs(ctx context.Context, conn *nats.Conn, caller, module, tool, node string, args any,
|
|
timeout time.Duration) (Answer, error) {
|
|
subject := broker.CalledSubject(ModuleToolOn(module, tool, node), caller)
|
|
if subject == "" {
|
|
return Answer{}, fmt.Errorf("%q cannot ask %s.%s on %s in its own name: it is not a bus user's name", caller,
|
|
module, tool, node)
|
|
}
|
|
return askModuleToolAt(ctx, conn, subject, module, tool, node, args, timeout)
|
|
}
|
|
|
|
// askModuleToolAt asks one machine's instance of a module one of its tools on subject and reads its answer.
|
|
func askModuleToolAt(ctx context.Context, conn *nats.Conn, subject, module, tool, node string, args any,
|
|
timeout time.Duration) (Answer, error) {
|
|
body, err := json.Marshal(args)
|
|
if err != nil {
|
|
return Answer{}, err
|
|
}
|
|
asking, cancel := context.WithTimeout(ctx, timeout)
|
|
defer cancel()
|
|
refused, stop := refusalsOf(conn, subject)
|
|
defer stop()
|
|
type replied struct {
|
|
msg *nats.Msg
|
|
err error
|
|
}
|
|
done := make(chan replied, 1)
|
|
go func() {
|
|
msg, err := conn.RequestWithContext(asking, subject, body)
|
|
done <- replied{msg, err}
|
|
}()
|
|
var reply *nats.Msg
|
|
select {
|
|
case r := <-done:
|
|
reply, err = r.msg, r.err
|
|
case why := <-refused:
|
|
cancel()
|
|
return Answer{}, fmt.Errorf("the bus refused the controller asking %s.%s on %s: %v", module, tool, node, why)
|
|
}
|
|
switch {
|
|
case errors.Is(err, nats.ErrNoResponders):
|
|
return Answer{}, fmt.Errorf("%w: %s on %s does not answer %s — it is not running there, or is "+
|
|
"older than the tool", ErrNothingServes, module, node, tool)
|
|
case errors.Is(err, context.DeadlineExceeded), errors.Is(err, nats.ErrTimeout):
|
|
return Answer{}, fmt.Errorf("%s on %s did not answer %s within %s", module, node, tool, timeout)
|
|
case err != nil:
|
|
return Answer{}, err
|
|
}
|
|
var answer Answer
|
|
if err := json.Unmarshal(reply.Data, &answer); err != nil {
|
|
return Answer{}, fmt.Errorf("%s on %s answered %s with something unreadable: %w", module, node, tool, err)
|
|
}
|
|
return answer, nil
|
|
}
|
|
|
|
// RetirementWaiting is the set a provider waits with for a person.
|
|
type RetirementWaiting struct {
|
|
Consumers []RetiredConsumer `json:"consumers"`
|
|
Since string `json:"since"`
|
|
Held int `json:"held"`
|
|
}
|
|
|
|
// RetirementRejected is the set a person's rejection keeps active.
|
|
type RetirementRejected struct {
|
|
Consumers []RetiredConsumer `json:"consumers"`
|
|
By string `json:"by"`
|
|
Why string `json:"why"`
|
|
At string `json:"at"`
|
|
}
|
|
|
|
// RetirementState is a provider's answer to provisioner_retirement.
|
|
type RetirementState struct {
|
|
Resource string `json:"resource"`
|
|
Node string `json:"node"`
|
|
Held []string `json:"held"`
|
|
// HeldSizes is what each held consumer keeps on the backend, in bytes, where the backend can say
|
|
// (novox/hq ADR 0233): what an empty replacement of a consumer's data is told by.
|
|
HeldSizes map[string]int64 `json:"held_sizes,omitempty"`
|
|
StablePasses int `json:"stable_passes"`
|
|
// StableForSeconds is how long the same unasked set must hold as well (ADR 0230: ten minutes).
|
|
StableForSeconds int `json:"stable_for_seconds,omitempty"`
|
|
// RetiresBy is "disable", or "mark-only" for a provider that cannot disable a consumer.
|
|
RetiresBy string `json:"retires_by,omitempty"`
|
|
Bound string `json:"bound"`
|
|
Waiting *RetirementWaiting `json:"waiting"`
|
|
Rejected *RetirementRejected `json:"rejected"`
|
|
Retired []RetiredConsumer `json:"retired"`
|
|
}
|