mesh-catalog/modules/distribution/internal/holder/driver.go

package holder

import (
	"errors"
	"fmt"
	"maps"
	"slices"
	"sort"
	"strings"
	"time"
)

// PublishingKey is the key of distribution's state windows that records, per build module, the fingerprints of the
// builds heard publishing through the seats: the tools bundle adds the have's fingerprint of the proved caller of each
// accepted publish. A node's have of a build module running one of them counts as publishing through the seats, so a
// node whose build module runs the same build but builds nothing (a paused build agent) does not hold the window.
const PublishingKey = "publishing"

// Publishing is that record: module -> fingerprints.
type Publishing struct {
	Name    string              `json:"name"`
	Modules map[string][]string `json:"modules"`
}

// Learn adds fp for module; it answers whether it was new.
func (p *Publishing) Learn(module, fp string) bool {
	if module == "" || fp == "" {
		return false
	}
	if p.Modules == nil {
		p.Modules = map[string][]string{}
	}
	if slices.Contains(p.Modules[module], fp) {
		return false
	}
	p.Modules[module] = append(p.Modules[module], fp)
	sort.Strings(p.Modules[module])
	return true
}

// NodeHave is one key of the status view's node_haves: a node, a module, the fingerprint it runs.
type NodeHave struct {
	Node, Module, FP string
}

// Delivery is what the windows' driver asks of the delivery parts and the node-engine seat, through their tools and
// verbs on the bus (ADR 0596): every node's haves, away ones included, and one node-engine's stated reader
// capabilities. Reads answers served false when the node-engine does not serve `manifest-reads` (it states nothing),
// with the call's error, if any, saying why (an away node, an older node-engine, a timeout).
type Delivery interface {
	Haves() ([]NodeHave, error)
	Reads(node string) (served bool, stated []string, err error)
}

// WindowState is distribution's state windows.
type WindowState interface {
	Window(name string) (Window, error)
	PutWindow(w Window) error
	Publishing() (Publishing, error)
}

// Driver moves the old-name and the direct-write windows: at distribution's start, on each `delivered` and on each
// node-engine's `manifest-reads-changed` (ADR 0618 §4), never on a timer. Each wake reads every node, since a window
// closes only when no node holds it.
type Driver struct {
	Delivery Delivery
	State    WindowState
	// Callers is the caller list as it holds on this wake: publish's callers are the build modules counted.
	Callers func() (Callers, error)
	// Closed says a window's closing on distribution.events.
	Closed  func(c Closed) error
	Finding func(kind, subject, words string, raised bool)
	Now     func() time.Time
}

// FindingWindowHeld is the finding of a node holding a window open for seven days.
const FindingWindowHeld = "window-held"

// Read reads every node: its haves, its node-engine's stated capabilities, and whether each build module's have runs a
// build heard publishing through the seats.
func (d *Driver) Read() ([]NodeRead, error) {
	haves, err := d.Delivery.Haves()
	if err != nil {
		return nil, fmt.Errorf("node_haves cannot be read, so no window moves: %w", err)
	}
	pub, err := d.State.Publishing()
	if err != nil {
		return nil, fmt.Errorf("the publishing builds cannot be read, so no window moves: %w", err)
	}
	callers, err := d.Callers()
	if err != nil {
		return nil, fmt.Errorf("the caller list cannot be read, so no window moves: %w", err)
	}
	builds := callers[VerbPublish]
	byNode := map[string]*NodeRead{}
	var order []string
	for _, h := range haves {
		if h.Module == "node" || h.Node == "" {
			continue
		}
		n := byNode[h.Node]
		if n == nil {
			n = &NodeRead{Node: h.Node, NamesArtifact: true}
			byNode[h.Node] = n
			order = append(order, h.Node)
		}
		if slices.Contains(builds, h.Module) {
			if n.BuildModules == nil {
				n.BuildModules = map[string]bool{}
			}
			n.BuildModules[h.Module] = slices.Contains(pub.Modules[h.Module], h.FP)
		}
	}
	sort.Strings(order)
	out := make([]NodeRead, 0, len(order))
	for _, node := range order {
		n := byNode[node]
		served, stated, err := d.Delivery.Reads(node)
		if err != nil && served {
			return nil, fmt.Errorf("%s's manifest-reads cannot be read, so no window moves: %w", node, err)
		}
		n.Serves, n.Stated = served, stated
		if !served && err != nil {
			n.NotServed = err.Error()
		}
		out = append(out, *n)
	}
	return out, nil
}

// Wake reads every node and steps both windows; it answers a line for each thing it did, and an error when nothing
// could be judged.
func (d *Driver) Wake(why string) ([]string, error) {
	nodes, err := d.Read()
	if err != nil {
		return nil, err
	}
	now := d.now()
	named := make([]string, 0, len(nodes))
	byNode := map[string]NodeRead{}
	for _, n := range nodes {
		named = append(named, n.Node)
		byNode[n.Node] = n
	}
	var said, errs []string
	for _, c := range []struct {
		name, event, seat string
		holding           func([]NodeRead, []DroppedNode) []string
		why               func(NodeRead) string
	}{
		{OldNameWindow, OldNameClosedEvent, OldSeat, HoldingOldName, WhyOldName},
		{DirectWritesWindow, DirectWritesClosedEvent, "", HoldingDirectWrites, WhyDirectWrites},
	} {
		cur, err := d.State.Window(c.name)
		if err != nil {
			errs = append(errs, err.Error())
			continue
		}
		if cur.Name == "" {
			cur = Window{Name: c.name}
			cur.Open = true
		}
		next, closed, overdue := cur.Step(named, c.holding(nodes, cur.Dropped), now)
		if next.Open && len(named) > 0 {
			next.Why = map[string]string{}
			for _, n := range next.Holding {
				next.Why[n] = c.why(byNode[n])
			}
		}
		if !windowsEqual(cur, next) {
			if err := d.State.PutWindow(next); err != nil {
				errs = append(errs, fmt.Sprintf("the %s window is not kept: %v", c.name, err))
				continue
			}
		}
		for _, n := range cur.Holding {
			if !slices.Contains(overdue, n) && d.Finding != nil {
				d.Finding(FindingWindowHeld, c.name+"."+n, "", false)
			}
		}
		for _, n := range overdue {
			if d.Finding != nil {
				d.Finding(FindingWindowHeld, c.name+"."+n, fmt.Sprintf("%s has held distribution's %s window open since %s", n, c.name, next.Since[n].Format(time.RFC3339)), true)
			}
		}
		if closed != nil {
			closed.Seat = c.seat
			if d.Closed != nil {
				if err := d.Closed(*closed); err != nil {
					errs = append(errs, fmt.Sprintf("%s was not said: %v", c.event, err))
				}
			}
			said = append(said, fmt.Sprintf("the %s window is closed (%s); it was held last by %v", c.name, why, closed.LastHolding))
		} else if len(named) == 0 {
			said = append(said, fmt.Sprintf("the %s window is not judged (%s): the read named no node", c.name, why))
		} else if next.Open {
			said = append(said, fmt.Sprintf("the %s window stays open (%s): held by %v", c.name, why, next.Holding))
		}
	}
	if len(errs) > 0 {
		return said, errors.New(strings.Join(errs, "; "))
	}
	return said, nil
}

func (d *Driver) now() time.Time {
	if d.Now != nil {
		return d.Now()
	}
	return time.Now().UTC()
}

func windowsEqual(a, b Window) bool {
	return a.Open == b.Open && slices.Equal(a.Holding, b.Holding) && a.Closed.Equal(b.Closed) &&
		len(a.Since) == len(b.Since) && len(a.Dropped) == len(b.Dropped) && maps.Equal(a.Why, b.Why)
}