mesh-host/internal/seatwait/seatwait.go

// Package seatwait is the node-engine's word that its reader of the seat list has waited (novox/hq ADR 0636 §1): a node
// whose reader holds only the bootstrapped definitions for five minutes writes one finding in the engine's own state
// `findings`, which the condition keeper raises as a condition and clears at the first full read (ADR 0564 §2).
//
// Event-driven: the reader says each change (Changed), and the one bound is a timer started when the wait begins and
// stopped when it ends, never a loop. A timer that fires looks at the reader again before it writes, so a read that
// ended the wait without a change of the held set (the seat answered the list the reader already held) writes nothing.
package seatwait

import (
	"context"
	"fmt"
	"sync"
	"time"

	condseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
	"git.novox.be/novox/mesh-sdk/go/seatlist"
)

// Bound is how long the reader may wait before the node says it (ADR 0636 §1).
const Bound = 5 * time.Minute

// Kind is the finding's kind; its id is the node's.
const Kind = "seat-list-unread"

// Reader is what the Waiter asks of the seat list's reader.
type Reader interface {
	Status() seatlist.Status
}

// Timer is a started timer that can be stopped.
type Timer interface{ Stop() bool }

// Clock gives the time and starts a timer; the tests give their own, so nothing sleeps.
type Clock interface {
	Now() time.Time
	AfterFunc(d time.Duration, f func()) Timer
}

// Writer is the node-engine's finding writer as the keepers use it: mesh-sdk's held writer over the node's state `findings`
// (views.HeldWriter, design 47 G A-29, A-38, A-40), which holds what stands there from its start, so the Waiter keeps no
// copy and a new process of the node-engine clears what an earlier one wrote (ADR 0636 §1, issue 1966).
type Writer interface {
	Write(ctx context.Context, f condseat.Finding) error
	Clear(ctx context.Context, scope, id, kind, why string) error
	Held() []condseat.Finding
}

const (
	scopeMachine = "machine"
	scopeModule  = "module"
)

// stands is whether the writer holds the finding of that scope, id and kind.
func stands(w Writer, scope, id, kind string) bool {
	for _, f := range w.Held() {
		if f.Scope == scope && f.ID == id && f.Kind == kind {
			return true
		}
	}
	return false
}

// Waiter writes and clears the node's finding.
type Waiter struct {
	Node     string
	Observer string
	// ObserverOf, when set, answers the observer at each write: the node-engine's own module, which a first start learns
	// from its first composing (design 47 G §527, §634).
	ObserverOf func() string
	Reader     Reader
	Writer     Writer
	Clock      Clock
	// Bound is how long the reader may wait; zero is the default Bound (the manifest's config `view-waits-minutes`, §633).
	Bound time.Duration
	// Ctx is the run's context for the writer's acts; nil is the background.
	Ctx context.Context
	// Say tells the node's journal; nil says nothing.
	Say func(string)

	mu    sync.Mutex
	timer Timer
	since time.Time
}

func (w *Waiter) ctx() context.Context {
	if w.Ctx != nil {
		return w.Ctx
	}
	return context.Background()
}

func (w *Waiter) stands() bool { return stands(w.Writer, scopeMachine, w.Node, Kind) }

func waiting(st seatlist.Status) bool { return !st.Heard || len(st.Waiting) > 0 }

// Waits is whether a reader of that status holds only the definitions it started with, or waits for some.
func Waits(st seatlist.Status) bool { return waiting(st) }

// Changed is told when the reader's held set or its status moved: it starts the bound when the reader waits and has no
// bound running, stops it and clears the finding when the reader has read the whole list.
func (w *Waiter) Changed() {
	st := w.Reader.Status()
	w.mu.Lock()
	defer w.mu.Unlock()
	if !waiting(st) {
		if w.timer != nil {
			w.timer.Stop()
			w.timer = nil
		}
		w.since = time.Time{}
		if w.stands() {
			if err := w.Writer.Clear(w.ctx(), scopeMachine, w.Node, Kind, "the node-engine has read the whole seat list"); err != nil {
				w.say(fmt.Sprintf("the node's finding that its seat list is unread cannot be cleared: %v", err))
			}
		}
		return
	}
	if w.timer != nil {
		return
	}
	if w.stands() {
		return // an earlier process wrote it and the list is still unread: it stands until the first full read
	}
	w.since = w.Clock.Now()
	w.timer = w.Clock.AfterFunc(w.bound(), w.expired)
}

// Waiting says whether the bound is running or the finding stands, and since when.
func (w *Waiter) Waiting() (bool, time.Time) {
	w.mu.Lock()
	defer w.mu.Unlock()
	if w.timer != nil {
		return true, w.since
	}
	return w.stands(), w.since
}

func (w *Waiter) expired() {
	st := w.Reader.Status()
	w.mu.Lock()
	defer w.mu.Unlock()
	w.timer = nil
	if !waiting(st) {
		w.since = time.Time{}
		return
	}
	if w.stands() {
		return
	}
	f := condseat.Finding{Observer: observerOf(w.Observer, w.ObserverOf), Kind: Kind, Scope: scopeMachine, ID: w.Node, Severity: condseat.FindingSeverityWarning,
		Source: Kind, Subject: condseat.Subject{Scope: scopeMachine, ID: w.Node, Machine: w.Node},
		Summary:     fmt.Sprintf("%s has read no complete seat list since %s: %s", w.Node, w.since.UTC().Format(time.RFC3339), st.Health()),
		Headline:    fmt.Sprintf("%s waits for the seat list", w.Node),
		Explanation: "The node holds only the definitions it starts with, so it keeps running what it runs and starts, composes and grants nothing new. It reads the list again at the next change the seat mesh-catalog says, or when asked to.",
		At:          w.Clock.Now()}
	if err := w.Writer.Write(w.ctx(), f); err != nil {
		w.say(fmt.Sprintf("the node's finding that its seat list is unread cannot be written: %v", err))
		return
	}
	w.say(f.Summary)
}

func (w *Waiter) say(s string) {
	if w.Say != nil {
		w.Say(s)
	}
}

// observerOf is of() when it is given, else fixed.
func observerOf(fixed string, of func() string) string {
	if of != nil {
		return of()
	}
	return fixed
}

func (w *Waiter) bound() time.Duration {
	if w.Bound > 0 {
		return w.Bound
	}
	return Bound
}