mesh-sdk/go/watched.go

package stdio

import (
	"fmt"
	"sort"
	"sync"
	"time"
)

// WatchedStates is a copy of the states a module reads, each filled by its own watch and kept by its changes (novox/hq
// ADR 0580 clause 1): a read of a whole state is one watch's current values, never a get per key, which took a
// composition past 30 seconds and a compare past its call bound. A state is watched on its first read, or on Follow,
// and the watch hands every current value over before the read is answered, so the copy is complete from the first
// answer. A key whose value never changes once written (a manifest, by its digest) is read once with Once and kept.
// A state helper, no rule: what a module reads is its own.
type WatchedStates struct {
	watch func(state string, handle func(StateChange) error) error
	get   func(state, key string) ([]byte, bool, error)

	mu     sync.Mutex
	states map[string]*watchedState
	// once holds each value read with Once, by state and key, for the process's life: a few kilobytes per manifest
	// digest. Past onceAtMost keys it is emptied, which costs only a read again.
	once map[string][]byte
}

// onceAtMost bounds the values Once keeps.
const onceAtMost = 4096

type watchedState struct {
	ready chan struct{} // closed once the watch's current values are all in the copy, or it was refused
	err   error

	mu     sync.Mutex
	values map[string][]byte
	// revs is, by key, the revision its value in the copy is at, and latest the highest revision the copy has seen: a
	// change the watch hands at a revision not newer than its key's is one the copy already took (Wrote, Removed).
	revs    map[string]uint64
	latest  uint64
	then    []func(StateChange) error
	started bool          // the current values are in, so a change runs then at once
	pending []StateChange // the current values, run through then once all are in
	thenMu  sync.Mutex    // then runs one change at a time
}

// NewWatchedStates is the copy over this bundle's runtime: each state by KeptState.Watch, Once by KeptState.Get.
func NewWatchedStates() *WatchedStates {
	return WatchedStatesOver(
		func(state string, handle func(StateChange) error) error { return State(state).Watch("", handle) },
		func(state, key string) ([]byte, bool, error) {
			e, err := State(state).Get(key)
			if err != nil || e == nil {
				return nil, false, err
			}
			return e.Value, true, nil
		})
}

// WatchedStatesOver is the copy over a watch and a get of one's own (a process's bus, a test's): watch hands handle
// every current value and returns once they are handled, then hands every change; get reads one key.
func WatchedStatesOver(watch func(state string, handle func(StateChange) error) error,
	get func(state, key string) ([]byte, bool, error)) *WatchedStates {
	return &WatchedStates{watch: watch, get: get, states: map[string]*watchedState{}, once: map[string][]byte{}}
}

// watched is the state's copy, its watch started now when it has none, the caller waiting until its current values
// are in. then, when given, runs after the copy took each change: the current values once they are all in (a then
// that reads the copy reads it whole), each later change as it comes. No lock is held while the watch starts, so a
// then may read any state of the copy.
func (w *WatchedStates) watched(state string, then func(StateChange) error) (*watchedState, error) {
	w.mu.Lock()
	s, ok := w.states[state]
	if ok {
		w.mu.Unlock()
		<-s.ready
		if s.err != nil {
			return nil, s.err
		}
		if then != nil {
			s.join(state, then)
		}
		return s, nil
	}
	s = &watchedState{ready: make(chan struct{}), values: map[string][]byte{}, revs: map[string]uint64{}}
	if then != nil {
		s.then = []func(StateChange) error{then}
	}
	w.states[state] = s
	w.mu.Unlock()
	err := w.watch(state, func(c StateChange) error {
		s.mu.Lock()
		switch {
		case c.Revision != 0 && c.Revision <= s.revs[c.Key]:
			// Already in the copy, or older than what is: this process wrote the key since (Wrote, Removed).
		case c.Op == OpPut:
			s.values[c.Key] = append([]byte(nil), c.Value...)
			s.revs[c.Key] = c.Revision
		default:
			delete(s.values, c.Key)
			s.revs[c.Key] = c.Revision
		}
		s.latest = max(s.latest, c.Revision)
		if !s.started {
			s.pending = append(s.pending, c)
			s.mu.Unlock()
			return nil
		}
		s.mu.Unlock()
		return s.run(state, c)
	})
	if err != nil {
		s.err = fmt.Errorf("the state %s is not watched: %w", state, err)
		w.mu.Lock()
		delete(w.states, state) // asked again on the next read
		w.mu.Unlock()
		close(s.ready)
		return nil, s.err
	}
	s.mu.Lock()
	s.started = true
	pending := s.pending
	s.pending = nil
	s.mu.Unlock()
	close(s.ready)
	for _, c := range pending {
		if err := s.run(state, c); err != nil {
			sayf("%v", err)
		}
	}
	return s, nil
}

// join adds then to a state already watched (a read started its watch first) and runs it over the copy's current values
// before any later change, so no change that landed before it joined is lost to it (review of ADR 0580 §2: a read
// beat the follower to the watch at start-up). Under thenMu, so a later change waits for these; a change the copy took
// just before the snapshot may also be handed again, which a then that wakes work takes as a second wake.
func (s *watchedState) join(state string, then func(StateChange) error) {
	s.thenMu.Lock()
	defer s.thenMu.Unlock()
	s.mu.Lock()
	s.then = append(s.then, then)
	current := make([]StateChange, 0, len(s.values))
	for k, v := range s.values {
		current = append(current, StateChange{State: state, Key: k, Op: OpPut, Value: v, Current: true})
	}
	s.mu.Unlock()
	sort.Slice(current, func(i, j int) bool { return current[i].Key < current[j].Key })
	for _, c := range current {
		if err := then(c); err != nil {
			sayf("%s %s: %v", state, c.Key, err)
		}
	}
}

// run hands one change, already in the copy, to every then.
func (s *watchedState) run(state string, c StateChange) error {
	s.thenMu.Lock()
	defer s.thenMu.Unlock()
	s.mu.Lock()
	then := append([]func(StateChange) error(nil), s.then...)
	s.mu.Unlock()
	var errs []error
	for _, f := range then {
		if err := f(c); err != nil {
			errs = append(errs, err)
		}
	}
	if len(errs) > 0 {
		return fmt.Errorf("%s %s: %v", state, c.Key, errs)
	}
	return nil
}

// Follow watches the state into the copy and runs then on each change after the copy took it, so a reader then wakes
// never reads a copy older than its change; it asks again until the watch is taken (2 s, doubled to a minute), as
// KeptState.Follow does. On a state already watched then hears the changes from now on.
func (w *WatchedStates) Follow(state string, then func(StateChange) error) {
	for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
		if _, err := w.watched(state, then); err == nil {
			return
		} else {
			sayf("%v; asking again in %s", err, wait)
		}
		time.Sleep(wait)
	}
}

// Get is a key's value in the copy, false when it has none.
func (w *WatchedStates) Get(state, key string) ([]byte, bool, error) {
	s, err := w.watched(state, nil)
	if err != nil {
		return nil, false, err
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	v, ok := s.values[key]
	return v, ok, nil
}

// Keys are the state's keys in the copy, sorted.
func (w *WatchedStates) Keys(state string) ([]string, error) {
	s, err := w.watched(state, nil)
	if err != nil {
		return nil, err
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	keys := make([]string, 0, len(s.values))
	for k := range s.values {
		keys = append(keys, k)
	}
	sort.Strings(keys)
	return keys, nil
}

// Values is every key's value in the copy.
func (w *WatchedStates) Values(state string) (map[string][]byte, error) {
	s, err := w.watched(state, nil)
	if err != nil {
		return nil, err
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	out := make(map[string][]byte, len(s.values))
	for k, v := range s.values {
		out[k] = v
	}
	return out, nil
}

// Once is a key whose value never changes once written, read by key the first time it is found and kept after: a
// manifest by its digest. A key not found is not kept, so it is read again on the next ask.
func (w *WatchedStates) Once(state, key string) ([]byte, bool, error) {
	w.mu.Lock()
	v, ok := w.once[state+" "+key]
	w.mu.Unlock()
	if ok {
		return v, true, nil
	}
	v, found, err := w.get(state, key)
	if err != nil || !found {
		return nil, false, err
	}
	w.mu.Lock()
	if len(w.once) >= onceAtMost {
		clear(w.once)
	}
	w.once[state+" "+key] = v
	w.mu.Unlock()
	return v, true, nil
}

// Wrote takes a put this process made into the copy at once, at the revision the bus answered (0 when not known): a
// state the process alone writes (a holder's own) is read back from the copy right after its own write, before the
// watch hands the change back, and the watch's older changes of the key are not taken over it (issue 1954). The state
// is watched first when it is not yet.
func (w *WatchedStates) Wrote(state, key string, value []byte, revision uint64) error {
	return w.took(state, key, value, revision, true)
}

// Removed takes a delete this process made into the copy at once, as Wrote takes a put.
func (w *WatchedStates) Removed(state, key string) error { return w.took(state, key, nil, 0, false) }

func (w *WatchedStates) took(state, key string, value []byte, revision uint64, put bool) error {
	s, err := w.watched(state, nil)
	if err != nil {
		return err
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	switch {
	case put && revision != 0:
		if revision <= s.revs[key] {
			return nil // the watch handed a newer change of the key already
		}
		s.values[key] = append([]byte(nil), value...)
		s.revs[key] = revision
		s.latest = max(s.latest, revision)
	default:
		// A delete's revision is not answered, nor a put's at times: it is newer than every revision the copy has seen,
		// so the watch's changes of the key up to that one are not taken over it.
		if put {
			s.values[key] = append([]byte(nil), value...)
		} else {
			delete(s.values, key)
		}
		s.revs[key] = s.latest
	}
	return nil
}

// Direct is a key read by key now, not from the copy: a value that changes and is read one at a time (a build).
func (w *WatchedStates) Direct(state, key string) ([]byte, bool, error) { return w.get(state, key) }