mesh-catalog/modules/zfs/cmd/zfs-tools/bus.go

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"os"
	"strings"
	"sync"
	"time"

	"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
	stdio "git.novox.be/novox/mesh-sdk/go"
	"git.novox.be/novox/mesh-sdk/go/buscalls"
	"git.novox.be/novox/mesh-sdk/go/process"
	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"
)

// busStates is the module's states as this process reaches them (its bus account's `state`).
type busStates struct{ c *process.Conn }

func (b busStates) Put(state, key string, v any) error {
	_, err := b.c.State(state).Put(key, v)
	return err
}
func (b busStates) Delete(state, key string) error      { return b.c.State(state).Delete(key) }
func (b busStates) Keys(state string) ([]string, error) { return b.c.State(state).Keys() }
func (b busStates) Get(state, key string) (json.RawMessage, bool, error) {
	e, err := b.c.State(state).Get(key)
	if err != nil {
		return nil, false, err
	}
	if e == nil {
		return nil, false, nil
	}
	return e.Value, true, nil
}

// JoinBus gives a root process the module's events and states through the keeper's socket (A-37): the process holds no
// bus account of its own. Without the socket variable the machine keeps none, and what it changes is said as a warning (§18:
// with the bus down the apply converges from the config file it last got). It answers how to leave the bus.
func (m *Machine) JoinBus() func() {
	sock := strings.TrimSpace(os.Getenv(KeeperSocketEnv))
	if sock == "" {
		m.Warnings = append(m.Warnings, KeeperSocketEnv+" is not set, so the keeper is not reached and the bus is taken as down")
		return func() {}
	}
	// Every call reaches the socket anew, so a keeper that comes up after this process is used from then on; one that
	// is down at a call answers that call with an error, never the bus taken as down for the process's life.
	// A call the keeper does not answer is said once, as a warning: the root process then acts as with the bus down (A-37).
	var once sync.Once
	c := keeperClient{sock: sock, down: func(err error) {
		once.Do(func() {
			m.Warnings = append(m.Warnings, "the bus is taken as down: "+err.Error())
		})
	}}
	m.Emit = c.emit
	m.States = c
	m.Findings = meshconditionsclient.FindingWriter{State: findingState{c}}
	m.Call = c.call
	return func() {}
}

// RunKeeper is the process `zfs-keeper`: the module's one process bus account, serving the socket to root.
func RunKeeper(ctx context.Context) error {
	sock := strings.TrimSpace(os.Getenv(KeeperSocketEnv))
	if sock == "" {
		return fmt.Errorf("%s is not set", KeeperSocketEnv)
	}
	acct, err := process.ReadBrokerAccount(os.Getenv)
	if err != nil {
		return err
	}
	c, err := process.Connect(acct)
	if err != nil {
		return err
	}
	defer c.Close()
	l, err := Listen(sock)
	if err != nil {
		return err
	}
	nc, js, err := acct.Dial()
	if err != nil {
		return err
	}
	defer nc.Close()
	k := &Keeper{Emit: func(kind string, body any) error { return publishEvent(js, acct, kind, body) }, States: busStates{c}, Call: c.Call}
	return k.Serve(ctx, l)
}

// eventPublisher is the part of JetStream the keeper publishes with.
type eventPublisher interface {
	PublishMsg(ctx context.Context, m *nats.Msg, opts ...jetstream.PublishOpt) (*jetstream.PubAck, error)
}

// publishEvent puts one event on the module's events stream as a JetStream publish the bus acknowledges, on the
// subject the stream captures, mesh.bus.<module>.events.<kind> (A-41); never process.Conn.Emit, whose subject the
// keeper's account is not granted. The message id (nanotime-pid) is the time in nanoseconds and the process id, so two events of one process never share one.
func publishEvent(js eventPublisher, acct process.Account, kind string, body any) error {
	data, err := json.Marshal(body)
	if err != nil {
		return err
	}
	id := fmt.Sprintf("%d-%d", time.Now().UnixNano(), os.Getpid())
	msg := nats.NewMsg(acct.BusSubject(EventsStream, kind))
	msg.Header.Set("x-event-id", id)
	msg.Header.Set("x-source", acct.Module)
	msg.Header.Set("x-node", acct.Node)
	msg.Header.Set("x-time", time.Now().UTC().Format(time.RFC3339Nano))
	msg.Header.Set("content-type", "application/json")
	msg.Data = data
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	_, err = js.PublishMsg(ctx, msg, jetstream.WithMsgID(id))
	return err
}

// toolStates is the module's states as the tools bundle reaches them, through the tool runner.
type toolStates struct{}

func (toolStates) Put(state, key string, v any) error {
	_, err := stdio.State(state).Put(key, v)
	return err
}
func (toolStates) Delete(state, key string) error      { return stdio.State(state).Delete(key) }
func (toolStates) Keys(state string) ([]string, error) { return stdio.State(state).Keys() }
func (toolStates) Get(state, key string) (json.RawMessage, bool, error) {
	e, err := stdio.State(state).Get(key)
	if err != nil || e == nil {
		return nil, false, err
	}
	return e.Value, true, nil
}

// eventHub hands the module's events, heard once through the tool runner, to the hand acts waiting for one.
type eventHub struct {
	mu      sync.Mutex
	started bool
	waiters map[int]hubWaiter
	next    int
}

type hubWaiter struct {
	kinds []string
	ch    chan map[string]any
}

func (h *eventHub) await(kinds []string, bound time.Duration) (map[string]any, bool) {
	h.mu.Lock()
	if h.waiters == nil {
		h.waiters = map[int]hubWaiter{}
	}
	h.next++
	id, w := h.next, hubWaiter{kinds, make(chan map[string]any, 1)}
	h.waiters[id] = w
	first := !h.started
	h.started = true
	h.mu.Unlock()
	defer func() { h.mu.Lock(); delete(h.waiters, id); h.mu.Unlock() }()
	if first {
		_ = stdio.Subscribe("#", func(e stdio.Envelope) error {
			var body map[string]any
			_ = json.Unmarshal(e.Body, &body)
			h.mu.Lock()
			defer h.mu.Unlock()
			for _, w := range h.waiters {
				for _, k := range w.kinds {
					if e.Key == k || strings.HasSuffix(e.Key, "."+k) {
						select {
						case w.ch <- body:
						default:
						}
					}
				}
			}
			return nil
		})
	}
	select {
	case ev := <-w.ch:
		return ev, true
	case <-time.After(bound):
		return nil, false
	}
}

// busLedger is the ledger `snapshots` through the tool runner.
type busLedger struct{}

func (busLedger) Apply(instance, operation string, body any, call string) error {
	raw, err := json.Marshal(body)
	if err != nil {
		return err
	}
	_, err = stdio.BusState(buscalls.LedgerApply, buscalls.ApplyRequest{Ledger: LedgerName, Instance: instance,
		Operation: operation, Body: raw, Call: call})
	return err
}

// takenOf reads a ledger get's answer: an instance never written answers state null and sequence 0, which is empty (A-40).
func takenOf(raw []byte) (struct {
	Taken map[string]LedgerEntry `json:"taken"`
}, error) {
	var body struct {
		Taken map[string]LedgerEntry `json:"taken"`
	}
	var st buscalls.InstanceState
	if err := json.Unmarshal(raw, &st); err != nil {
		return body, err
	}
	if st.Sequence == 0 && (len(st.State) == 0 || string(st.State) == "null") {
		return body, nil
	}
	err := json.Unmarshal(st.State, &body)
	return body, err
}

func (busLedger) Entries(datasets []string) ([]LedgerEntry, error) {
	var out []LedgerEntry
	for _, ds := range datasets {
		for _, kind := range keepKinds {
			raw, err := stdio.BusState(buscalls.LedgerGet, buscalls.InstanceRequest{Ledger: LedgerName, Instance: LedgerInstance(kind, ds)})
			if err != nil {
				return nil, err
			}
			body, err := takenOf(raw)
			if err != nil {
				return nil, err
			}
			for _, name := range sortedKeys(body.Taken) {
				out = append(out, body.Taken[name])
			}
		}
	}
	return out, nil
}

// EventsStream is the module's events stream's local name in its `requires` (§25).
const EventsStream = "events"

// busStream is the module's events stream through the tool runner.
type busStream struct{}

// Last reads the stream from its first sequence on, page by page, and keeps the last n events of the kind.
func (busStream) Last(n int, kind string) ([]LoggedEvent, error) {
	raw, err := stdio.BusState(buscalls.StreamFill, buscalls.StreamRequest{Stream: EventsStream})
	if err != nil {
		return nil, err
	}
	var fill buscalls.StreamFilled
	if err := json.Unmarshal(raw, &fill); err != nil {
		return nil, err
	}
	var out []LoggedEvent
	from := fill.First
	for from != 0 && from <= fill.Last {
		raw, err := stdio.BusState(buscalls.StreamRead, buscalls.StreamReadRequest{Stream: EventsStream, From: from, Limit: 500})
		if err != nil {
			return nil, err
		}
		var page buscalls.StreamPage
		if err := json.Unmarshal(raw, &page); err != nil {
			return nil, err
		}
		for _, msg := range page.Messages {
			k := KindOfSubject(msg.Subject)
			if kind != "" && k != kind {
				continue
			}
			out = append(out, LoggedEvent{Sequence: msg.Sequence, Kind: k, At: msg.At, Body: msg.Body})
			if len(out) > n {
				out = out[1:]
			}
		}
		from = page.Next
	}
	return out, nil
}