Hear what providers retire, and let a person approve, reject and delete (hq ADR 0230)

A provider now waits for a person before retiring more than three consumers
or half of what it holds, and deletes only when asked. The controller is that
person's way in: it keeps waiting and rejected sets as conditions, answers them
with retire approve|reject, lists and deletes retired consumers through the
provider's own tools on its machine, records each act in the hand-act log, and
probes for anything retired longer than thirty days (D11).

internal/link/contracts.go

@@ -42,6 +42,10 @@ var Contracts = map[string]Contract{
 		"catalogue's record, which is the order, so a late one reads the same record"},
 	KindCatchUp: {Unordered: "a catalogue asking what it missed: answered from the record, whenever asked"},
 	KindProvisioner: {Unordered: "a provider's newest word about a consumer, said again every fifteen minutes " +
-		"while it holds (ADR 0224): the condition keeps the last observed, and S8 says when the words stop",
-		Tests: []string{"TestAProviderFailingAConsumerBreaksAllWellUntilItRecovers"}},
+		"while it holds (ADR 0224): the condition keeps the last observed, and S8 says when the words stop. " +
+		"Its retirement word (ADR 0230) is unordered too: a waiting set is said again every fifteen minutes, " +
+		"and every other change is checked against the provider itself — `retire` and `cleanup` ask it, " +
+		"and the self-check's D11 asks it every run — so an older word read late is corrected by the next",
+		Tests: []string{"TestAProviderFailingAConsumerBreaksAllWellUntilItRecovers",
+			"TestARetirementWordIsKeptAsItsConditionNamingTheEmitterFromTheSubject"}},
 }

internal/link/receive_nats.go

@@ -262,7 +262,7 @@ func kindOfSubject(subject string) (string, bool) {
 }
 
 // ProvisionerEmitter is the module a provider's standing event came from, read from its subject
-// (`mesh.mod.<module>.event.provisioner.<failing|recovered>`); false for any other subject. The
+// (`mesh.mod.<module>.event.provisioner.<failing|recovered|retirement>`); false for any other subject. The
 // controller's own follow pattern, with `*` for the module, decodes too.
 func ProvisionerEmitter(subject string) (string, bool) {
 	rest, ok := strings.CutPrefix(subject, "mesh.mod.")
@@ -273,7 +273,8 @@ func ProvisionerEmitter(subject string) (string, bool) {
 	if !ok || module == "" || strings.Contains(module, ".") {
 		return "", false
 	}
-	if event != broker.ProvisionerFailing && event != broker.ProvisionerRecovered {
+	if event != broker.ProvisionerFailing && event != broker.ProvisionerRecovered &&
+		event != broker.ProvisionerRetirement {
 		return "", false
 	}
 	return module, true

internal/link/retirement.go

@@ -0,0 +1,241 @@
+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"`
+}
+
+// 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) {
+	body, err := json.Marshal(args)
+	if err != nil {
+		return Answer{}, err
+	}
+	asking, cancel := context.WithTimeout(ctx, timeout)
+	defer cancel()
+	subject := ModuleToolOn(module, tool, node)
+	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"`
+	StablePasses int                 `json:"stable_passes"`
+	Bound        string              `json:"bound"`
+	Waiting      *RetirementWaiting  `json:"waiting"`
+	Rejected     *RetirementRejected `json:"rejected"`
+	Retired      []RetiredConsumer   `json:"retired"`
+}

internal/link/serve.go

@@ -81,6 +81,8 @@ type Server struct {
 	replayer Replayer
 	// standings keeps what providers say about their consumers (novox/hq ADR 0224).
 	standings Standings
+	// retirements keeps what providers say about consumers the mesh stopped asking for (ADR 0230).
+	retirements Retirements
 
 	log *log.Logger
 	// giveUp is how long one message is held for the store; zero means GiveUpAfter.

internal/link/standing.go

@@ -81,6 +81,10 @@ func ReadStanding(subject string, body []byte) (Standing, error) {
 // it lasts, so one dropped is replaced; a recovery is said once, and dropping it would leave status
 // naming a consumer that is fine. So a store that is away holds the message, as a report is held.
 func (s *Server) provisioner(ctx context.Context, m Control) {
+	if IsRetirement(m.Subject()) {
+		s.retirement(ctx, m)
+		return
+	}
 	if s.standings == nil {
 		// Delivered because the consumer's filter names it, with nothing here keeping it: taken,
 		// because handing it back would not give it anywhere to go.

internal/link/standing_test.go

@@ -37,6 +37,7 @@ func TestTheControllerFollowsEveryProvidersStandingAndNothingElse(t *testing.T)
 	for subject, want := range map[string]string{
 		"mesh.mod.keycloak.event.provisioner.failing":   "keycloak",
 		"mesh.mod.postgres.event.provisioner.recovered": "postgres",
+		"mesh.mod.minio.event.provisioner.retirement":   "minio",
 		"mesh.mod.*.event.provisioner.failing":          "*",
 	} {
 		got, ok := ProvisionerEmitter(subject)
@@ -63,8 +64,11 @@ func TestTheControllerFollowsEveryProvidersStandingAndNothingElse(t *testing.T)
 			follows++
 		}
 	}
-	if follows != 2 {
-		t.Fatalf("the controller follows %d standing subjects, want failing and recovered", follows)
+	if follows != 3 {
+		t.Fatalf("the controller follows %d provider subjects, want failing, recovered and retirement (ADR 0230)", follows)
+	}
+	if !IsRetirement("mesh.mod.postgres.event.provisioner.retirement") || IsRetirement("mesh.mod.postgres.event.provisioner.failing") {
+		t.Fatal("a retirement word is not told from a standing")
 	}
 }