mesh-controller/internal/inventory/chain_assignments.go

package inventory

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"sort"
	"strings"
	"time"

	"github.com/jackc/pgx/v5"

	"github.com/novox/mesh-controller/internal/chain"
)

// Assignments on the bus (novox/hq ADR 0408, issue 1093): once the kind `assignments` is moved (chain.go) an assignment is
// a key `<node>.<module>` of the assignments bucket, written by compare-and-set by the controller that holds the lease,
// and every question the joins of the store answered is answered from the bucket. **No key is removed** (ADR 0408 §4): an
// unassignment writes the value again with `ended` set, and a value with `ended` or a `state` is no assignment to any reader
// here (chain.Assignment.Stands). No `pending` state is written yet (issue 1094).

// KindAssignments is the kind of the chain that is each module assigned to a node, with the build it runs there.
const KindAssignments = "assignments"

func assignmentKey(node, module string) string { return node + "." + module }

// assignRow is one assignment as the joins of the store read it.
type assignRow struct {
	Node, Module string
	Assigned     time.Time
}

// assignments is every assignment, by node then module: from the bucket when the kind is moved, else from the store.
// A key whose value names no time assigned is a module only pulled in by another (issue 765), and one that has ended or
// waits in a state is gone: no assignment. One on a machine the store does not know is none either, as the store took a
// removed machine's.
func (i *Inventory) assignments(ctx context.Context) ([]assignRow, error) {
	c, on, err := i.ChainReads(ctx, KindAssignments)
	if err != nil {
		return nil, err
	}
	if !on {
		rows, err := i.store.Pool().Query(ctx,
			`select n.name, a.module, a.assigned from assignment a join node n on n.id = a.node order by n.name, a.module`)
		if err != nil {
			return nil, err
		}
		defer rows.Close()
		var out []assignRow
		for rows.Next() {
			var a assignRow
			if err := rows.Scan(&a.Node, &a.Module, &a.Assigned); err != nil {
				return nil, err
			}
			out = append(out, a)
		}
		return out, rows.Err()
	}
	all, err := c.Assignments.All(ctx)
	if err != nil {
		return nil, fmt.Errorf("the assignments cannot be read from the bus: %w", err)
	}
	nodes, err := i.nodeNames(ctx)
	if err != nil {
		return nil, err
	}
	var out []assignRow
	for _, k := range all {
		if k.Value.Assigned.IsZero() || !k.Value.Stands() || !nodes[k.Value.Node] {
			continue
		}
		out = append(out, assignRow{k.Value.Node, k.Value.Module, k.Value.Assigned})
	}
	sort.Slice(out, func(a, b int) bool {
		if out[a].Node != out[b].Node {
			return out[a].Node < out[b].Node
		}
		return out[a].Module < out[b].Module
	})
	return out, nil
}

func (i *Inventory) nodeNames(ctx context.Context) (map[string]bool, error) {
	rows, err := i.store.Pool().Query(ctx, `select name from node`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	out := map[string]bool{}
	for rows.Next() {
		var name string
		if err := rows.Scan(&name); err != nil {
			return nil, err
		}
		out[name] = true
	}
	return out, rows.Err()
}

// assignedNodes is the machines a module is assigned to, sorted.
func (i *Inventory) assignedNodes(ctx context.Context, module string) ([]string, error) {
	rows, err := i.assignments(ctx)
	if err != nil {
		return nil, err
	}
	var out []string
	for _, a := range rows {
		if a.Module == module {
			out = append(out, a.Node)
		}
	}
	return out, nil
}

// assignedModules is the modules assigned to a machine, sorted.
func (i *Inventory) assignedModules(ctx context.Context, node string) ([]string, error) {
	rows, err := i.assignments(ctx)
	if err != nil {
		return nil, err
	}
	var out []string
	for _, a := range rows {
		if a.Node == node {
			out = append(out, a.Module)
		}
	}
	return out, nil
}

// isAssigned says a module is assigned to a machine.
func (i *Inventory) isAssigned(ctx context.Context, node, module string) (bool, error) {
	rows, err := i.assignments(ctx)
	if err != nil {
		return false, err
	}
	for _, a := range rows {
		if a.Node == node && a.Module == module {
			return true, nil
		}
	}
	return false, nil
}

// raiseGeneration raises the assignment generation, as the store's triggers do for a change of its table (migration
// 0094): in the chain nothing in the store changes with an assignment, so the writer says it.
func (i *Inventory) raiseGeneration(ctx context.Context) error {
	_, err := i.store.Pool().Exec(ctx, `update assignment_generation set generation = generation + 1`)
	return err
}

// chainAssign puts a module on a machine, and says whether that is new.
func (i *Inventory) chainAssign(ctx context.Context, c *chain.Chain, node Node, module string) (bool, error) {
	var known bool
	if err := i.store.Pool().QueryRow(ctx, `select exists (select 1 from module where name = $1)`, module).Scan(&known); err != nil {
		return false, err
	}
	if !known {
		return false, fmt.Errorf("%w: %s", ErrNoSuchModule, module)
	}
	key := assignmentKey(node.Name, module)
	for try := 0; try < casTries; try++ {
		cur, found, err := c.Assignments.Get(ctx, key)
		if err != nil {
			return false, err
		}
		if found && !cur.Value.Assigned.IsZero() && cur.Value.Stands() {
			return false, nil // one assignment of a module per machine: a repeat is absorbed (novox/hq ADR 0115)
		}
		now := time.Now().UTC()
		fresh := chain.Assignment{Node: node.Name, Module: module, Assigned: now}
		switch {
		case found && cur.Value.Stands():
			// A module only pulled in by another becomes an assignment, keeping what it holds.
			cur.Value.Assigned = now
			_, err = c.Assignments.Update(ctx, key, cur.Value, cur.Revision)
		case found:
			// One that ended is assigned again, and one that was pending is made: a fresh value over it, the earlier one
			// staying in history. The wait is carried into the value that stands (issue 1094).
			fresh.Pending = carriedPending(cur.Value, now)
			_, err = c.Assignments.Update(ctx, key, fresh, cur.Revision)
		default:
			_, err = c.Assignments.Create(ctx, key, fresh)
		}
		if errors.Is(err, chain.ErrMoved) || errors.Is(err, chain.ErrExists) {
			continue
		}
		if err != nil {
			return false, fmt.Errorf("%s is not assigned to %s on the bus: %w", module, node.Name, err)
		}
		if err := i.raiseGeneration(ctx); err != nil {
			return true, err
		}
		return true, nil
	}
	return false, fmt.Errorf("the assignment of %s to %s kept moving under the write", module, node.Name)
}

// chainUnassign takes a module off a machine: the value is written again with `ended`, the key stays.
func (i *Inventory) chainUnassign(ctx context.Context, c *chain.Chain, node Node, module string) error {
	key := assignmentKey(node.Name, module)
	for try := 0; try < casTries; try++ {
		cur, found, err := c.Assignments.Get(ctx, key)
		if err != nil {
			return err
		}
		if !found || cur.Value.Assigned.IsZero() || !cur.Value.Stands() {
			return fmt.Errorf("%s is not assigned to %s", module, node.Name)
		}
		cur.Value.Ended = time.Now().UTC()
		if _, err := c.Assignments.Update(ctx, key, cur.Value, cur.Revision); errors.Is(err, chain.ErrMoved) {
			continue
		} else if err != nil {
			return fmt.Errorf("%s is not taken off %s on the bus: %w", module, node.Name, err)
		}
		return i.raiseGeneration(ctx)
	}
	return fmt.Errorf("the assignment of %s to %s kept moving under the write", module, node.Name)
}

// chainAwaitingSince is AwaitingSince over the bucket: the oldest assignment on the machine made after it was last sent.
func (i *Inventory) chainAwaitingSince(ctx context.Context, name string) (time.Time, error) {
	var sentAt *time.Time
	var created time.Time
	err := i.store.Pool().QueryRow(ctx, `select sent_at, created from node where name = $1`, name).Scan(&sentAt, &created)
	if errors.Is(err, pgx.ErrNoRows) {
		return time.Time{}, fmt.Errorf("%w: %s", ErrNoSuchNode, name)
	}
	if err != nil {
		return time.Time{}, err
	}
	rows, err := i.assignments(ctx)
	if err != nil {
		return time.Time{}, err
	}
	var oldest time.Time
	for _, a := range rows {
		if a.Node == name && (sentAt == nil || a.Assigned.After(*sentAt)) && (oldest.IsZero() || a.Assigned.Before(oldest)) {
			oldest = a.Assigned
		}
	}
	switch {
	case !oldest.IsZero():
		return oldest, nil
	case sentAt != nil:
		return *sentAt, nil
	}
	return created, nil
}

// KeepAssignmentPlans writes the provisioner's plan of each assignment into its value on the bus, and a value of its own for
// each module only pulled in by another on a machine (issue 765), and ends the value of one that is no longer pulled in. The
// plans are derived from what each machine was last sent, not part of the record of what is assigned, so they are read
// again on a tick and written where they differ: **by compare-and-set on the one field**, never laying over the build, commit
// and rollout another write put in the same value, and never ending an assignment. A plan the provisioner's rules refuse is
// refused by name and nothing is written (AssignmentValues).
func (i *Inventory) KeepAssignmentPlans(ctx context.Context) (written, removed int, err error) {
	c, on, err := i.ChainReads(ctx, KindAssignments)
	if err != nil || !on {
		return 0, 0, err
	}
	var plans map[string]Provisioning
	if i.provisioningPlans != nil {
		if plans, err = i.provisioningPlans(ctx); err != nil {
			return 0, 0, fmt.Errorf("composing the provisioner's plans: %w", err)
		}
	}
	assigned, err := i.assignments(ctx)
	if err != nil {
		return 0, 0, err
	}
	rows := make([]AssignmentMirrored, 0, len(assigned))
	for _, a := range assigned {
		rows = append(rows, AssignmentMirrored{Node: a.Node, Module: a.Module, Assigned: a.Assigned.UTC()})
	}
	want, err := AssignmentValues(rows, plans)
	if err != nil {
		return 0, 0, err
	}
	have, err := c.Assignments.All(ctx)
	if err != nil {
		return 0, 0, fmt.Errorf("the assignments cannot be read from the bus: %w", err)
	}
	for key, body := range want {
		var next chain.Assignment
		if err := json.Unmarshal(body, &next); err != nil {
			return written, removed, err
		}
		cur, has := have[key]
		switch keepStep(has, cur.Value, next) {
		case keepSkip:
			continue
		case keepCreate:
			next.Assigned = time.Time{}
			_, err = c.Assignments.Create(ctx, key, next)
		case keepFresh:
			// A value that ended, pulled in again: a fresh one over it, the ended one staying in history.
			next.Assigned = time.Time{}
			_, err = c.Assignments.Update(ctx, key, next, cur.Revision)
		default:
			cur.Value.RequiredBy, cur.Value.Provisioning = next.RequiredBy, next.Provisioning
			_, err = c.Assignments.Update(ctx, key, cur.Value, cur.Revision)
		}
		if errors.Is(err, chain.ErrMoved) || errors.Is(err, chain.ErrExists) {
			continue // written meanwhile; the next pass reads it
		}
		if err != nil {
			return written, removed, fmt.Errorf("the plan of %s is not written: %w", key, err)
		}
		written++
	}
	for key, cur := range have {
		if _, wanted := want[key]; wanted || !cur.Value.Assigned.IsZero() || !cur.Value.Stands() {
			continue
		}
		// A value of a module no longer pulled in, which no assignment names: it ends, and stays.
		cur.Value.Ended = time.Now().UTC()
		if _, err := c.Assignments.Update(ctx, key, cur.Value, cur.Revision); errors.Is(err, chain.ErrMoved) {
			continue
		} else if err != nil {
			return written, removed, fmt.Errorf("the plan of %s is not ended: %w", key, err)
		}
		removed++
	}
	return written, removed, nil
}

// sameJSON says two encoded values are the same bytes, an absent value being a null one.
func sameJSON(a, b json.RawMessage) bool {
	trim := func(x json.RawMessage) string {
		if s := strings.TrimSpace(string(x)); s != "null" {
			return s
		}
		return ""
	}
	return trim(a) == trim(b)
}

// KeptValue is an assignment's value as the bucket holds it, encoded, at its revision.
type KeptValue struct {
	Value    []byte
	Revision uint64
}

// KeptAssignments is every key of the assignments bucket that holds a value that stands, and false when the assignments are
// not on the bus.
// What the serving controller compares from one look to the next to say a key changed (`assignment-changed`).
func (i *Inventory) KeptAssignments(ctx context.Context) (map[string]KeptValue, bool, error) {
	c, on, err := i.ChainReads(ctx, KindAssignments)
	if err != nil || !on {
		return nil, false, err
	}
	all, err := c.Assignments.All(ctx)
	if err != nil {
		return nil, true, err
	}
	out := make(map[string]KeptValue, len(all))
	for key, k := range all {
		if !k.Value.Stands() {
			continue // an ended value is no assignment: for what is said of it, the key has gone
		}
		body, err := json.Marshal(k.Value)
		if err != nil {
			return nil, true, err
		}
		out[key] = KeptValue{Value: body, Revision: k.Revision}
	}
	return out, true, nil
}

// What a pass that keeps the plans does with one key.
const (
	keepSkip   = "skip"
	keepCreate = "create"
	keepFresh  = "fresh"
	keepField  = "field"
)

// keepStep decides what the pass does with a key from what it wants there (read from the assignments) and what the bucket
// holds (read a moment later). **It never makes or revives an assignment**: that is Assign's, and an unassignment between the
// two reads leaves a key that ended while the pass still wants it as an assignment, which would come back as a value
// pulled in by another. Only a key with no time assigned, on both sides, is made or made again here.
func keepStep(has bool, cur, next chain.Assignment) string {
	switch {
	case !has && !next.Assigned.IsZero():
		return keepSkip // assigned since the bucket was read: its own write is not here yet
	case !has:
		return keepCreate
	case cur.State == chain.StatePending:
		return keepSkip // a wait for a build is not pulled in by a plan
	case !cur.Stands():
		if !cur.Assigned.IsZero() || !next.Assigned.IsZero() {
			return keepSkip // an assignment that ended is not pulled in again by a plan
		}
		return keepFresh
	case sameJSON(cur.Provisioning, next.Provisioning) && cur.RequiredBy == next.RequiredBy:
		return keepSkip
	}
	return keepField
}

// carriedPending is the wait a value that does not stand carries into the assignment made over it: a wait that was applying
// stays applying, for the one making it settles it applied; one still waiting is made by hand, and says so. Nothing for a
// value that held no wait, or whose key is not one that was pending.
func carriedPending(was chain.Assignment, now time.Time) *chain.Pending {
	if was.Pending == nil || was.State != chain.StatePending {
		return nil
	}
	p := *was.Pending
	if p.State == chain.PendingWaiting {
		p.State, p.Note, p.Settled = chain.PendingApplied, "assigned by hand", &now
	}
	return &p
}

// AssignmentValue is one key of the assignments bucket as it is held, standing or not: one read of an assignment.
func (i *Inventory) AssignmentValue(ctx context.Context, node, module string) (chain.Assignment, bool, error) {
	c, err := i.theChain(ctx)
	if err != nil {
		return chain.Assignment{}, false, err
	}
	k, found, err := c.Assignments.Get(ctx, assignmentKey(node, module))
	return k.Value, found, err
}