mesh-delivery/modules/mesh-delivery-watchman/internal/watch/watch.go

// Package watch is the watchman's logic (novox/hq ADR 0433, ADR 0431, issue 1135): it turns the bus's connect and
// disconnect events, the connections request, the builds bucket and the checks-running bucket into presence, pulse and
// stuck keys. It holds no timer: every reaction is an event, a watch entry or a request the caller makes.
package watch

import (
	"fmt"
	"sync"
	"time"

	"git.novox.be/novox/mesh-sdk/go/delivery"
)

// Op is what happened to a watched key.
type Op int

const (
	// Put is a key written.
	Put Op = iota
	// Delete is a key deleted by a writer (the watchman's own delete, or a builder ending a check).
	Delete
	// Expired is the server's marker for a key whose time to live ran out.
	Expired
)

// Out is where the watchman writes. Every call is a write the caller makes on the bus.
type Out interface {
	PutPresence(key string, online bool) error
	// PutPulse writes the pulse key with a time to live, replacing a live one.
	PutPulse(key string, value []byte, ttl time.Duration) error
	// DeletePulse deletes the pulse key (a delete, not an expiry).
	DeletePulse(key string) error
	PutStuck(key string, value []byte) error
}

type presence struct {
	online bool
	since  time.Time
}

type pulse struct {
	bucket, key, waited string
	since               time.Time
	head                string
	token               string
	seeded              bool
}

// Watchman is the state the watchman keeps in memory; it is rebuilt from the buckets at every start. Its methods are
// safe to call from several watches at once.
type Watchman struct {
	mu       sync.Mutex
	out      Out
	now      func() time.Time
	logf     func(format string, args ...any)
	counts   map[string]int // open connections per presence key
	presence map[string]presence
	pulses   map[string]*pulse // live pulses, by pulse key
	stuck    map[string]delivery.Stuck
	bound    time.Duration
}

// New is a watchman writing to out, reading the time from now.
func New(out Out, now func() time.Time) *Watchman {
	return &Watchman{out: out, now: now, counts: map[string]int{}, presence: map[string]presence{},
		pulses: map[string]*pulse{}, stuck: map[string]delivery.Stuck{}, bound: delivery.PulseMoveBound}
}

// Logf sets where a write that failed is said; it is silent until set.
func (w *Watchman) Logf(f func(format string, args ...any)) { w.logf = f }

func (w *Watchman) say(format string, args ...any) {
	if w.logf != nil {
		w.logf(format, args...)
	}
}

// mirror makes the pulse of the move (bucket, key) that began at since, unless it stands already for the same move.
// A move that is already as old as the bound is stuck at once and has no pulse.
func (w *Watchman) mirror(bucket, key, waited, head string, since time.Time, revision uint64) {
	pk, err := delivery.PulseKey(bucket, key)
	if err != nil {
		w.say("no pulse for %s %s: %v", bucket, key, err)
		return
	}
	token := head + "|" + since.UTC().Format(time.RFC3339Nano)
	cur, tracked := w.pulses[pk]
	if tracked && !cur.seeded && cur.token == token {
		return
	}
	ttl, stalled := w.ttl(w.now().Sub(since))
	if stalled {
		if tracked {
			w.deletePulse(pk)
		}
		w.writeStuck(bucket, key, waited, since)
		return
	}
	value, err := delivery.EncodePulse(delivery.Pulse{Bucket: bucket, Key: key, Revision: revision, Owes: waited})
	if err != nil {
		w.say("no pulse for %s %s: %v", bucket, key, err)
		return
	}
	if err := w.out.PutPulse(pk, value, ttl); err != nil {
		w.say("cannot write the pulse %s: %v", pk, err)
		return
	}
	w.pulses[pk] = &pulse{bucket: bucket, key: key, waited: waited, since: since, head: head, token: token}
}

func (w *Watchman) deletePulse(pk string) {
	delete(w.pulses, pk)
	if err := w.out.DeletePulse(pk); err != nil {
		w.say("cannot delete the pulse %s: %v", pk, err)
	}
}

// moved is the mirrored key moving: its pulse is deleted, so it never expires, and a stuck key it left is written as
// moved (never deleted).
func (w *Watchman) moved(bucket, key string) {
	pk, err := delivery.PulseKey(bucket, key)
	if err != nil {
		return
	}
	if _, ok := w.pulses[pk]; ok {
		w.deletePulse(pk)
	}
	w.stuckMoved(bucket, key)
}

// stuckMoved writes a stuck key of the key as moved, if it stands and is not yet.
func (w *Watchman) stuckMoved(bucket, key string) {
	sk, err := delivery.StuckKey(bucket, key)
	if err != nil {
		return
	}
	s, ok := w.stuck[sk]
	if !ok || s.Moved {
		return
	}
	s.Moved = true
	w.putStuck(sk, s)
}

func (w *Watchman) writeStuck(bucket, key, waited string, since time.Time) {
	sk, err := delivery.StuckKey(bucket, key)
	if err != nil {
		w.say("no stuck key for %s %s: %v", bucket, key, err)
		return
	}
	if cur, ok := w.stuck[sk]; ok && cur.Since.Equal(since) {
		return
	}
	w.putStuck(sk, delivery.Stuck{Bucket: bucket, Key: key, Waited: waited, Since: since})
}

func (w *Watchman) putStuck(sk string, s delivery.Stuck) {
	value, err := delivery.EncodeStuck(s)
	if err != nil {
		w.say("cannot encode the stuck key %s: %v", sk, err)
		return
	}
	if err := w.out.PutStuck(sk, value); err != nil {
		w.say("cannot write the stuck key %s: %v", sk, err)
		return
	}
	s.V = delivery.Version
	w.stuck[sk] = s
}

// Pulse is a change heard on the pulse bucket. Only the server's expiry marker of a pulse the watchman made writes a
// stuck key: the watchman's own delete, and a key it never made, write nothing.
func (w *Watchman) Pulse(key string, op Op, at time.Time) {
	w.mu.Lock()
	defer w.mu.Unlock()
	if op != Expired {
		return
	}
	p, ok := w.pulses[key]
	if !ok || p.seeded {
		return
	}
	delete(w.pulses, key)
	w.writeStuck(p.bucket, p.key, p.waited, p.since)
}

// SeedStuck tells the watchman a stuck key that stands in its bucket.
func (w *Watchman) SeedStuck(key string, value []byte) {
	w.mu.Lock()
	defer w.mu.Unlock()
	if s, err := delivery.DecodeStuck(value); err == nil {
		w.stuck[key] = s
	}
}

// SeedPulse tells the watchman a pulse key that stands in its bucket at its start; Reconcile deletes it unless a
// mirrored key made it again since.
func (w *Watchman) SeedPulse(key string) {
	w.mu.Lock()
	defer w.mu.Unlock()
	bucket, k, err := delivery.ParsePulseKey(key)
	if err != nil {
		return
	}
	w.pulses[key] = &pulse{bucket: bucket, key: k, seeded: true}
}

// Reconcile ends the start: a pulse that stood and that no mirrored key has made again is a move that happened while
// the watchman was away, so it is deleted and writes no stuck key.
func (w *Watchman) Reconcile() {
	w.mu.Lock()
	defer w.mu.Unlock()
	for pk, p := range w.pulses {
		if p.seeded {
			w.deletePulse(pk)
		}
	}
}

func describe(format string, args ...any) string { return fmt.Sprintf(format, args...) }

// SetBound sets how long a move may take; it is delivery.PulseMoveBound unless a test shortens it.
func (w *Watchman) SetBound(d time.Duration) {
	w.mu.Lock()
	defer w.mu.Unlock()
	w.bound = d
}

// ttl is the time to live of a pulse for a move of the given age, and whether the move is stalled already
// (delivery.PulseTTL, with the bound this watchman keeps).
func (w *Watchman) ttl(age time.Duration) (time.Duration, bool) {
	if w.bound == delivery.PulseMoveBound {
		return delivery.PulseTTL(age)
	}
	if age >= w.bound {
		return 0, true
	}
	return w.bound - age, false
}