mesh-controller/cmd/mesh-controller/states_announced.go

package main

import (
	"context"
	"errors"
	"fmt"
	"os"
	"strings"

	"git.novox.be/novox/mesh-sdk/go/announce"
	"github.com/nats-io/nats.go/jetstream"
	"github.com/novox/mesh-controller/internal/broker"
	"github.com/novox/mesh-controller/internal/inventory"
)

// What each module keeps, and who may read it (novox/hq ADR 0495 clauses 2 to 5, ADR 0513 §3; issue 1790, D1-D3).
//
// Every module says on the bus which states it keeps and which of them others may read (announce.Subject), from its
// code; the controller says its own (D1). The serving controller keeps the last word of each module in its bucket
// (D2), and the bus's users are composed from it: a reader's `reads: <owner>.<name>` is granted only when the owner
// announced that state readable; an owner that has announced nothing yet is read from `reads` as before (the
// transition), and `status` names it. A changed word composes the bus's users again: the machine holding the bus is
// sent its new user list (D3). No timer: the words wake it.

// announceOwnStates says what the controller keeps, on its own event subject (D1).
func announceOwnStates(ctx context.Context, js *broker.JetStream) error {
	host, _ := os.Hostname()
	body, err := announce.Encode(broker.ControllerAnnouncement(host))
	if err != nil {
		return err
	}
	api, err := jetstream.New(js.Conn())
	if err != nil {
		return err
	}
	if _, err := api.Publish(ctx, announce.Subject(broker.ControllerSeat), body); err != nil {
		return fmt.Errorf("saying what the controller keeps: %w", err)
	}
	return nil
}

// keepAnnouncements follows every module's word on what it keeps (D2) and keeps the last of each, from the last each
// module said: one module's word in the bucket under its name. A word whose module is not the subject's is refused,
// said; a changed word composes the bus's users again (D3).
func keepAnnouncements(ctx context.Context, js *broker.JetStream, open *stores) error {
	api, err := jetstream.New(js.Conn())
	if err != nil {
		return err
	}
	kv, err := api.KeyValue(ctx, broker.StatesAnnouncedBucket)
	if err != nil {
		return fmt.Errorf("the bucket of the states announced: %w", err)
	}
	stream, err := api.Stream(ctx, broker.EventsStream)
	if err != nil {
		return fmt.Errorf("the events stream: %w", err)
	}
	consumer, err := stream.OrderedConsumer(ctx, jetstream.OrderedConsumerConfig{
		FilterSubjects: []string{announce.SubjectAll}, DeliverPolicy: jetstream.DeliverLastPerSubjectPolicy})
	if err != nil {
		return fmt.Errorf("following the states announced: %w", err)
	}
	consuming, err := consumer.Consume(func(m jetstream.Msg) {
		changed, err := keepAnnouncement(ctx, kv, m.Subject(), m.Data())
		if err != nil {
			fmt.Printf("states: %v\n", err)
			return
		}
		if changed {
			inBackground(func() { composeAnnouncedReads(context.WithoutCancel(ctx), open) })
		}
	})
	if err != nil {
		return fmt.Errorf("following the states announced: %w", err)
	}
	go func() {
		<-ctx.Done()
		consuming.Stop()
	}()
	return nil
}

// announcedKeeper is the bucket of the states announced, as keepAnnouncement writes it.
type announcedKeeper interface {
	Get(ctx context.Context, key string) (jetstream.KeyValueEntry, error)
	Put(ctx context.Context, key string, value []byte) (uint64, error)
}

// keepAnnouncement keeps one module's word, and says whether it changed what the bucket held.
func keepAnnouncement(ctx context.Context, kv announcedKeeper, subject string, body []byte) (bool, error) {
	s, err := announce.Decode(body)
	if err != nil {
		return false, fmt.Errorf("a word on %s is not read: %w", subject, err)
	}
	if subject != announce.Subject(s.Module) {
		return false, fmt.Errorf("%s says what %s keeps, and a module says only what it keeps itself: refused", subject,
			s.Module)
	}
	key, err := announce.Key(s.Module)
	if err != nil {
		return false, err
	}
	stored, err := announce.Encode(s)
	if err != nil {
		return false, err
	}
	if was, err := kv.Get(ctx, key); err == nil {
		if string(was.Value()) == string(stored) {
			return false, nil
		}
	} else if !errors.Is(err, jetstream.ErrKeyNotFound) {
		return false, fmt.Errorf("the last word of %s cannot be read: %w", s.Module, err)
	}
	if _, err := kv.Put(ctx, key, stored); err != nil {
		return false, fmt.Errorf("the word of %s cannot be kept: %w", s.Module, err)
	}
	fmt.Printf("states: %s keeps %s\n", s.Module, statesWords(s))
	return true, nil
}

// statesWords is an announcement as a person reads it: each state, readable ones marked.
func statesWords(s announce.States) string {
	if len(s.States) == 0 {
		return "nothing"
	}
	var out []string
	for _, k := range s.States {
		if k.Readable {
			out = append(out, k.Name+" (readable)")
		} else {
			out = append(out, k.Name)
		}
	}
	return strings.Join(out, ", ")
}

// readAnnouncements is the last word of every module, by module (what inventory.Announcements reads): none when the
// bucket is not there yet, which reads every owner as not yet announced.
func readAnnouncements(ctx context.Context) (map[string]announce.States, error) {
	js, err := aBus()
	if err != nil {
		return nil, err
	}
	defer js.Close()
	api, err := jetstream.New(js.Conn())
	if err != nil {
		return nil, err
	}
	kv, err := api.KeyValue(ctx, broker.StatesAnnouncedBucket)
	if errors.Is(err, jetstream.ErrBucketNotFound) {
		// empty-on-error: not an error; no module announced yet, so every read is granted from `reads` as before
		return map[string]announce.States{}, nil
	}
	if err != nil {
		return nil, err
	}
	keys, err := kv.Keys(ctx)
	if errors.Is(err, jetstream.ErrNoKeysFound) {
		// empty-on-error: not an error; no module announced yet
		return map[string]announce.States{}, nil
	}
	if err != nil {
		return nil, err
	}
	out := map[string]announce.States{}
	for _, key := range keys {
		e, err := kv.Get(ctx, key)
		if errors.Is(err, jetstream.ErrKeyNotFound) {
			continue
		}
		if err != nil {
			return nil, err
		}
		s, err := announce.Decode(e.Value())
		if err != nil {
			return nil, fmt.Errorf("the kept word of %s is not read: %w", key, err)
		}
		out[s.Module] = s
	}
	return out, nil
}

// composeAnnouncedReads sends the machine holding the bus its user list when a changed word changed it (D3): as a new
// controller's user list is carried (carryUserList), not when that machine is held.
func composeAnnouncedReads(ctx context.Context, open *stores) {
	holder, behind, err := brokerBehind(ctx, open, nil)
	if err != nil {
		fmt.Printf("states: whether the bus's user list is behind the states announced cannot be read: %v\n", err)
		return
	}
	if holder == "" || !behind {
		return
	}
	held, err := heldMachines(ctx, open, []string{holder})
	if err != nil {
		fmt.Printf("states: whether %s may be sent its new user list cannot be read: %v\n", holder, err)
		return
	}
	if why, isHeld := held[holder]; isHeld {
		fmt.Printf("states: the user list the states announced compose is not sent to %s, which holds the bus: %s — "+
			"`push %s` sends it\n", holder, strings.Join(why, "; "), holder)
		return
	}
	if _, err := sendRollout(ctx, open, []string{holder}); err != nil {
		fmt.Printf("states: the user list the states announced compose could not be sent to %s: %v\n", holder, err)
		return
	}
	fmt.Printf("states: sent %s, which holds the bus, the user list the states announced compose\n", holder)
}

// notYetAnnounced are the owners some module reads the state of and that have announced nothing yet: what `status`
// names while the transition lasts.
func notYetAnnounced(ctx context.Context, inv *inventory.Inventory) ([]string, error) {
	if inventory.Announcements == nil {
		return nil, nil
	}
	records, err := inv.BusRecords(ctx)
	if err != nil {
		return nil, err
	}
	if records.Announced == nil {
		return nil, errors.New("the states announced cannot be read")
	}
	return broker.NotAnnounced(records), nil
}