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 (
	"fmt"
	"sync"
	"time"

	"git.novox.be/novox/mesh-sdk/go/seatlist"
	"git.novox.be/novox/mesh-sdk/go/views"
)

// 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
}

// Findings is the node's own state `findings`: the writer, and the read of whether a finding stands there. The bus state
// is the one truth of whether the finding exists (ADR 0636 §1, issue 1966): the Waiter keeps no copy, so a new process
// of the node-engine clears what an earlier one wrote.
type Findings interface {
	views.FindingWriter
	Has(f views.Finding) (bool, error)
}

// Waiter writes and clears the node's finding.
type Waiter struct {
	Node     string
	Observer string
	Reader   Reader
	Findings Findings
	Clock    Clock
	// Say tells the node's journal; nil says nothing.
	Say func(string)

	mu    sync.Mutex
	timer Timer
	since time.Time
}

// key is the finding this node's Waiter writes: Kind and ID name it, the rest is its text.
func (w *Waiter) key() views.Finding {
	return views.Finding{Observer: w.Observer, Kind: Kind, ID: w.Node}
}

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

// 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{}
		stands, err := w.Findings.Has(w.key())
		if err != nil {
			w.say(fmt.Sprintf("the node's finding that its seat list is unread cannot be read: %v", err))
			return
		}
		if stands {
			if err := w.Findings.Delete(w.key()); 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 stands, err := w.Findings.Has(w.key()); err == nil && 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(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
	}
	stands, err := w.Findings.Has(w.key())
	return err == nil && 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 stands, err := w.Findings.Has(w.key()); err == nil && stands {
		return
	}
	f := views.Finding{Observer: w.Observer, Kind: Kind, ID: 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.Findings.Put(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)
	}
}