issue 974: a rollout's wait for a machine is a key on the bus that its report deletes and its expiry names silent

cmd/mesh-controller/rollout_silent_bus.go

@@ -0,0 +1,121 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/nats-io/nats.go/jetstream"
+
+	"github.com/novox/mesh-controller/internal/broker"
+)
+
+// A rollout's wait for a machine is a key on the bus (novox/hq ADR 0406 §5, issue 974): each send to a machine puts
+// a key `<plan>.<tier>.<node>` that lives as long as the machine may stay silent (ApplySilentAfter); the machine's
+// report on that send deletes it, so only a missing report lets it expire; and the expiry, heard by a watcher,
+// ends the wait for that machine, which the plan then names silent. No timer looks every so often whether a
+// machine has reported.
+
+// silentWait is one machine's wait on one send of a plan's tier.
+type silentWait struct {
+	Plan string
+	Tier int
+	Node string
+}
+
+// silentKey is a wait's key. A plan id may hold a dot, so it is read from the right (parseSilentKey).
+func silentKey(w silentWait) string { return w.Plan + "." + strconv.Itoa(w.Tier) + "." + w.Node }
+
+// parseSilentKey reads a key back: the node is the last token, the tier the one before, the plan what is left.
+func parseSilentKey(key string) (silentWait, bool) {
+	last := strings.LastIndexByte(key, '.')
+	if last < 1 || last == len(key)-1 {
+		return silentWait{}, false
+	}
+	rest, node := key[:last], key[last+1:]
+	mid := strings.LastIndexByte(rest, '.')
+	if mid < 1 || mid == len(rest)-1 {
+		return silentWait{}, false
+	}
+	tier, err := strconv.Atoi(rest[mid+1:])
+	if err != nil || tier < 0 {
+		return silentWait{}, false
+	}
+	return silentWait{Plan: rest[:mid], Tier: tier, Node: node}, true
+}
+
+// silentClock is the bus's side of the wait for a machine's report.
+type silentClock interface {
+	// Arm puts a wait's key, to live ttl: made again if the machine is sent again.
+	Arm(ctx context.Context, w silentWait, ttl time.Duration) error
+	// Reported takes a wait's key away: the machine reported on the send, and nothing is to wake the controller.
+	Reported(ctx context.Context, w silentWait) error
+}
+
+// busSilent is the clock on the controller's bucket.
+type busSilent struct {
+	kv jetstream.KeyValue
+}
+
+// newBusSilent opens the bucket and starts the watcher, which names each wait whose key expired until the context
+// ends. Only expiry wakes it: the marker of an expiry is a purge, and a key a report deleted, or one Arm made again,
+// leaves a delete (the vendored client reads the server's MaxAge marker as a purge, kv.go:971).
+func newBusSilent(ctx context.Context, api jetstream.JetStream, silent func(silentWait)) (*busSilent, error) {
+	kv, err := api.KeyValue(ctx, broker.RolloutSilentBucket)
+	if err != nil {
+		return nil, fmt.Errorf("the rollout's silent keys' bucket: %w", err)
+	}
+	w, err := kv.WatchAll(ctx, jetstream.UpdatesOnly())
+	if err != nil {
+		return nil, fmt.Errorf("watching the rollout's silent keys: %w", err)
+	}
+	go func() {
+		defer w.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case e, ok := <-w.Updates():
+				if !ok {
+					return
+				}
+				if e == nil || e.Operation() != jetstream.KeyValuePurge {
+					continue // a put, or a delete: a report's, or a send made again
+				}
+				if wait, ok := parseSilentKey(e.Key()); ok && silent != nil {
+					silent(wait)
+				}
+			}
+		}
+	}()
+	return &busSilent{kv: kv}, nil
+}
+
+func (s *busSilent) Arm(ctx context.Context, w silentWait, ttl time.Duration) error {
+	key := silentKey(w)
+	// Made again, not changed: a key's time to live is set when it is made.
+	if err := s.kv.Delete(ctx, key); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
+		return err
+	}
+	_, err := s.kv.Create(ctx, key, []byte(time.Now().UTC().Format(time.RFC3339)), jetstream.KeyTTL(leastSecond(ttl)))
+	return err
+}
+
+func (s *busSilent) Reported(ctx context.Context, w silentWait) error {
+	if err := s.kv.Delete(ctx, silentKey(w)); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
+		return err
+	}
+	return nil
+}
+
+// leastSecond rounds a time to live up to a whole second, at least one: the server rounds a key's time down to
+// whole seconds, so a fraction would close the key before the time it stands for.
+func leastSecond(d time.Duration) time.Duration {
+	if d < time.Second {
+		return time.Second
+	}
+	return (d + time.Second - 1) / time.Second * time.Second
+}