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

package holder

import (
	"fmt"
	"slices"
	"sort"
	"time"
)

// The two windows distribution keeps (design 51): the old seat name `mesh-artifact-store`, served until every node whose
// declaration names an artifact applied a build whose node-engine states it reads the new names (ADR 0610 §4, ADR 0614
// §4), and the direct writes to the store, accepted from the mesh until every node's have of every build module runs a
// build that publishes through the seats. Each wakes at distribution's start, on each `delivered` and on the node-engine
// seat's `manifest-reads-changed` (ADR 0618 §4), never on a timer; each reads every node, away ones included, through
// node_haves and `manifest-reads`, never a watch of another module's state (ADR 0596).
const (
	OldNameWindow      = "old-name"
	DirectWritesWindow = "direct-writes"
	// WindowFindingAfter is how long a node holds a window open before it is distribution's finding naming the node.
	WindowFindingAfter = 7 * 24 * time.Hour
	// The events a window's close says on distribution's own family distribution.events (7 days, 16 MiB, drop
	// oldest), never on a seat's `<seat>.events`.
	OldNameClosedEvent      = "old-name-closed"
	DirectWritesClosedEvent = "direct-writes-closed"
	// NewNamesCapability is the reader capability a node-engine states when it reads the new seat names (ADR 0614 §4).
	NewNamesCapability = "seat:" + ImageSeat
)

// Window is one window's record, kept in distribution's state windows: what registry_old_name and
// registry_direct_writes answer.
type Window struct {
	Name    string               `json:"name"`
	Open    bool                 `json:"open"`
	Holding []string             `json:"holding,omitempty"`
	Since   map[string]time.Time `json:"since,omitempty"`
	Dropped []DroppedNode        `json:"dropped,omitempty"`
	Closed  time.Time            `json:"closed,omitzero"`
	// LastHolding are the nodes that held it last, said when it closes.
	LastHolding []string `json:"last_holding,omitempty"`
}

// DroppedNode is one retired node dropped from a window by hand: a hand act.
type DroppedNode struct {
	Node string    `json:"node"`
	By   string    `json:"by"`
	Why  string    `json:"why"`
	At   time.Time `json:"at"`
}

// Closed is the body of a window's closing event.
type Closed struct {
	Window      string        `json:"window"`
	Seat        string        `json:"seat,omitempty"`
	At          time.Time     `json:"at"`
	LastHolding []string      `json:"last_holding"`
	Dropped     []DroppedNode `json:"dropped,omitempty"`
}

// NodeRead is what distribution read of one node: the reader capabilities its node-engine states through
// `manifest-reads` (Serves false while it does not serve the verb: it states nothing), whether its declaration names an
// artifact, and its build modules' haves: whether each runs a build that publishes through the seats.
type NodeRead struct {
	Node          string          `json:"node"`
	Serves        bool            `json:"serves"`
	Stated        []string        `json:"stated,omitempty"`
	NamesArtifact bool            `json:"names_artifact"`
	BuildModules  map[string]bool `json:"build_modules,omitempty"`
}

// HoldingOldName are the nodes holding the old name open: a node whose declaration names an artifact and whose
// node-engine does not state the new names (one not serving manifest-reads states nothing).
func HoldingOldName(nodes []NodeRead, dropped []DroppedNode) []string {
	var out []string
	for _, n := range nodes {
		if isDropped(dropped, n.Node) || !n.NamesArtifact {
			continue
		}
		if !n.Serves || !slices.Contains(n.Stated, NewNamesCapability) {
			out = append(out, n.Node)
		}
	}
	sort.Strings(out)
	return out
}

// HoldingDirectWrites are the nodes one of whose build modules' haves runs a build that does not yet publish through
// the seats.
func HoldingDirectWrites(nodes []NodeRead, dropped []DroppedNode) []string {
	var out []string
	for _, n := range nodes {
		if isDropped(dropped, n.Node) {
			continue
		}
		for _, publishes := range n.BuildModules {
			if !publishes {
				out = append(out, n.Node)
				break
			}
		}
	}
	sort.Strings(out)
	return out
}

func isDropped(dropped []DroppedNode, node string) bool {
	return slices.ContainsFunc(dropped, func(d DroppedNode) bool { return d.Node == node })
}

// Step moves the window to the nodes holding it at now. It answers the next record, the closing event when it closed in
// this step (said once), and the nodes holding it past WindowFindingAfter (the finding; cleared when it closes). A
// window once closed stays closed: a node that holds it later is no reason to serve the old name again.
func (w Window) Step(holding []string, now time.Time) (next Window, closed *Closed, overdue []string) {
	next = Window{Name: w.Name, Dropped: w.Dropped, Closed: w.Closed, LastHolding: w.LastHolding}
	if !w.Open && !w.Closed.IsZero() {
		return next, nil, nil
	}
	if len(holding) == 0 {
		next.Closed, next.LastHolding = now, w.Holding
		if next.LastHolding == nil {
			next.LastHolding = []string{}
		}
		return next, &Closed{Window: w.Name, At: now, LastHolding: next.LastHolding, Dropped: w.Dropped}, nil
	}
	next.Open, next.Holding, next.Since = true, holding, map[string]time.Time{}
	for _, n := range holding {
		since, had := w.Since[n]
		if !had {
			since = now
		}
		next.Since[n] = since
		if now.Sub(since) >= WindowFindingAfter {
			overdue = append(overdue, n)
		}
	}
	return next, nil, overdue
}

// Drop drops a retired node from the window by hand, a hand act with who and why.
func (w Window) Drop(node, by, why string, now time.Time) (Window, error) {
	if node == "" || why == "" {
		return w, fmt.Errorf("drop: a node and why are required (a hand act)")
	}
	if isDropped(w.Dropped, node) {
		return w, fmt.Errorf("drop: %s is dropped from %s already", node, w.Name)
	}
	w.Dropped = append(w.Dropped, DroppedNode{Node: node, By: by, Why: why, At: now})
	return w, nil
}