mesh-tools/node-tools/internal/bus/per_node.go

package bus

import (
	"context"
	"errors"
	"fmt"
	"sort"
	"strings"
	"sync"

	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"
)

// A state its owner keeps one bucket per node (novox/hq ADR 0593 §6: each node's engine writes its own `composed-<node>`,
// and the provisioner, the assigner and the view read them all) is served to a bundle as one state, as mesh-sdk's
// process.NodeFamily serves it to a process: a key is read from the bucket its first part names, the keys and a watch
// go over every bucket of the prefix, and a watch takes in a bucket created later, heard by the bus's creation
// advisory, so a node added later is heard with no timer. This runtime holds its own connection, not mesh-sdk's, so
// the same rules are written here over it; the two subjects below are the ones the grants part grants a reader.
const (
	familyListSubject     = "$JS.API.STREAM.NAMES"
	familyCreatedAdvisory = "$JS.EVENT.ADVISORY.STREAM.CREATED."
)

func perNodeRefusal(module, name, what string) error {
	return fmt.Errorf("%s's state %s is kept one bucket per node and read only: no %s of it as one state (novox/hq ADR 0593 §6)",
		module, name, what)
}

// nodeBucketOf is the state of the one node a key names: its first part, which must be a node's name.
func nodeBucketOf(s StateIssued, key string) (StateIssued, error) {
	node, _, ok := strings.Cut(key, ".")
	if !ok || node == "" {
		return s, fmt.Errorf("%q names no node: a key of the per-node state %s starts <node>.", key, s.Name)
	}
	s.Bucket += node
	return s, nil
}

// perNodeNodes are the nodes whose bucket of the family the bus holds now, sorted.
func (c *Conn) perNodeNodes(s StateIssued) ([]string, error) {
	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
	defer cancel()
	js, err := jetstream.New(c.nc)
	if err != nil {
		return nil, err
	}
	names := js.KeyValueStoreNames(ctx)
	var nodes []string
	for name := range names.Name() {
		if node, ok := strings.CutPrefix(name, s.Bucket); ok && node != "" {
			nodes = append(nodes, node)
		}
	}
	if err := names.Error(); err != nil && !errors.Is(err, jetstream.ErrEndOfData) {
		return nil, fmt.Errorf("the state %s: listing the buckets %s*: %w", s.Name, s.Bucket, err)
	}
	sort.Strings(nodes)
	return nodes, nil
}

func (c *Conn) perNodeKeys(s StateIssued) ([]string, error) {
	nodes, err := c.perNodeNodes(s)
	if err != nil {
		return nil, err
	}
	keys := []string{}
	for _, node := range nodes {
		one := s
		one.Bucket += node
		ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
		kv, err := c.bucket(ctx, one)
		if err == nil {
			var k []string
			k, err = keysOf(ctx, kv)
			keys = append(keys, k...)
		}
		cancel()
		if err != nil {
			return nil, err
		}
	}
	sort.Strings(keys)
	return keys, nil
}

// perNodeWatch is StateWatch over every bucket of the family, and over a bucket created later. A pattern whose first
// part names a node watches that node's bucket only.
func (c *Conn) perNodeWatch(module string, s StateIssued, filter string, deliver func(StateChange) error) (func(), error) {
	var mu sync.Mutex
	one := func(ch StateChange) error {
		mu.Lock()
		defer mu.Unlock()
		return deliver(ch)
	}
	var (
		wmu     sync.Mutex
		watched = map[string]bool{}
		stops   []func()
		done    bool
		sub     *nats.Subscription
	)
	add := func(node string) error {
		wmu.Lock()
		if watched[node] || done {
			wmu.Unlock()
			return nil
		}
		watched[node] = true
		wmu.Unlock()
		bs := s
		bs.Bucket += node
		ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
		kv, err := c.bucket(ctx, bs)
		cancel()
		if err != nil {
			return err
		}
		stop, err := c.watchBucket(module, s.Name, kv, filter, one)
		if err != nil {
			return err
		}
		wmu.Lock()
		if done {
			wmu.Unlock()
			stop()
			return nil
		}
		stops = append(stops, stop)
		wmu.Unlock()
		return nil
	}
	stopAll := func() {
		wmu.Lock()
		done = true
		all := stops
		stops = nil
		s := sub
		wmu.Unlock()
		if s != nil {
			_ = s.Unsubscribe()
		}
		for _, stop := range all {
			stop()
		}
	}
	if first, _, _ := strings.Cut(filter, "."); filter != ">" && first != "*" && first != ">" {
		if err := add(first); err != nil {
			stopAll()
			return nil, err
		}
		return stopAll, nil
	}
	// Heard before the buckets are listed, so a bucket created between the two is not missed.
	subject := familyCreatedAdvisory + "*"
	sb, err := c.nc.Subscribe(subject, func(m *nats.Msg) {
		bucket, isKV := strings.CutPrefix(m.Subject, familyCreatedAdvisory+"KV_")
		if node, ok := strings.CutPrefix(bucket, s.Bucket); isKV && ok && node != "" {
			go func() {
				if err := add(node); err != nil {
					c.Logf("[mesh-tools] %s did not take in the new bucket %s%s of %s: %v", module, s.Bucket, node, s.Name, err)
				}
			}()
		}
	})
	if err != nil {
		return nil, fmt.Errorf("the state %s: hearing new buckets %s*: %w", s.Name, s.Bucket, err)
	}
	wmu.Lock()
	sub = sb
	wmu.Unlock()
	nodes, err := c.perNodeNodes(s)
	if err != nil {
		stopAll()
		return nil, err
	}
	for _, node := range nodes {
		if err := add(node); err != nil {
			stopAll()
			return nil, err
		}
	}
	return stopAll, nil
}