mesh-controller/internal/inventory/chain_links.go

package inventory

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"maps"
	"slices"
	"sort"
	"time"

	"github.com/jackc/pgx/v5"

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

// The links of the chain (novox/hq ADR 0408 §1, design 47 acceptance 10, issue 1095): a send that puts a build on a machine
// writes, in the assignment of the module there, the build, what it published, its commit and the rollout that sent it, and
// since when; the rollout lists the assignments it changed with the build before and after (the controller's send, in
// release_plan.go). One read of an assignment then names the commit, the build and the rollout that put it there, and one read
// of the rollout lists its assignments.

type rolloutKey struct{}

// RolloutModule is what a rollout sends of one module: the commit and the build of it.
type RolloutModule struct{ Commit, Build string }

type rolloutOf struct {
	id      string
	modules map[string]RolloutModule
}

// WithRollout is a context for a send a rollout makes: what RecordSent writes in the assignments of the modules it carries
// names this rollout and the build it sent. A module the send carries beside these, that another walk is moving, is named
// by the build the mesh holds of its commit and by no rollout.
func WithRollout(ctx context.Context, id string, modules map[string]RolloutModule) context.Context {
	return context.WithValue(ctx, rolloutKey{}, rolloutOf{id: id, modules: modules})
}

// AssignmentRuns is what each module assigned to a machine runs there, as its assignment names it: the build and the
// commit. Empty when the assignments are not on the bus, and for a module not sent yet.
func (i *Inventory) AssignmentRuns(ctx context.Context, node string) (map[string]chain.Ref, error) {
	c, err := i.theChain(ctx)
	if err != nil {
		return nil, err
	}
	all, err := c.Assignments.All(ctx)
	if err != nil {
		return nil, err
	}
	out := map[string]chain.Ref{}
	for _, k := range all {
		a := k.Value
		if a.Node == node && a.Stands() && (a.Build != "" || a.Commit != "") {
			out[a.Module] = chain.Ref{Build: a.Build, Commit: a.Commit}
		}
	}
	return out, nil
}

// AssignmentsOn are the assignments of a machine as the chain names them, by module: the standing ones, with their links.
func (i *Inventory) AssignmentsOn(ctx context.Context, node string) ([]chain.Assignment, error) {
	c, err := i.theChain(ctx)
	if err != nil {
		return nil, err
	}
	all, err := c.Assignments.All(ctx)
	if err != nil {
		return nil, err
	}
	var out []chain.Assignment
	for _, k := range all {
		if a := k.Value; a.Node == node && !a.Assigned.IsZero() && a.Stands() {
			out = append(out, a)
		}
	}
	sort.Slice(out, func(a, b int) bool { return out[a].Module < out[b].Module })
	return out, nil
}

// recordRuns writes what a send put on a machine into the assignments, once the send is recorded, and it is how the mesh
// knows what a machine was sent (SentBuilds). **A send that names its builds** writes, for each module it carried a build of
// — by the commit — the build, what it published, the commit, the rollout and since when, on the module's key: filled where
// the key names nothing, replaced when the commit moved, never blanked by a push by hand; a module nothing assigns, carried
// because another requires it, gets a standing key marked pulled-in; and every other key of the machine loses what it named,
// for the machine is not sent it any more. **A send that names none** (a declaration by hand) makes what the machine runs
// unknown: every key of it says so (sent-unknown) and names no commit. **An error here fails the record of the send**, for
// what a machine was sent is read from these keys and from nowhere else.
func (i *Inventory) recordRuns(ctx context.Context, nodeID string, builds map[string]string, at time.Time) error {
	c, err := i.theChain(ctx)
	if err != nil {
		return fmt.Errorf("what the machine runs cannot be written: %w", err)
	}
	var node string
	if err := i.store.Pool().QueryRow(ctx, `select name from node where id = $1`, nodeID).Scan(&node); errors.Is(err, pgx.ErrNoRows) {
		return nil
	} else if err != nil {
		return err
	}
	rollout, _ := ctx.Value(rolloutKey{}).(rolloutOf)
	all, err := c.Assignments.All(ctx)
	if err != nil {
		return fmt.Errorf("the assignments of %s cannot be read: %w", node, err)
	}
	var failed []error
	for _, k := range sortedValues(all) {
		a := k.Value
		if a.Node != node || !a.Stands() {
			continue
		}
		if builds[a.Module] != "" {
			continue // carried, and its commit is written below
		}
		if err := i.forgetRun(ctx, c, k, builds == nil, at); err != nil {
			failed = append(failed, fmt.Errorf("%s on %s: %w", a.Module, node, err))
		}
	}
	modules := make([]string, 0, len(builds))
	for m := range builds {
		modules = append(modules, m)
	}
	sort.Strings(modules)
	for _, m := range modules {
		if builds[m] == "" {
			continue
		}
		if err := i.recordRun(ctx, c, node, m, builds[m], rollout, at); err != nil {
			failed = append(failed, fmt.Errorf("%s on %s: %w", m, node, err))
		}
	}
	return errors.Join(failed...)
}

func sortedValues(all map[string]chain.Kept[chain.Assignment]) []chain.Kept[chain.Assignment] {
	out := make([]chain.Kept[chain.Assignment], 0, len(all))
	for key, k := range all {
		k.Key = key
		out = append(out, k)
	}
	sort.Slice(out, func(a, b int) bool { return out[a].Key < out[b].Key })
	return out
}

// forgetRun writes that a machine is no longer sent what a key named: its links are cleared, and for a send by hand the key
// says what the machine runs is unknown.
func (i *Inventory) forgetRun(ctx context.Context, c *chain.Chain, k chain.Kept[chain.Assignment], hand bool, at time.Time) error {
	for try := 0; try < casTries; try++ {
		next := k.Value
		next.Build, next.Artifact, next.Package, next.Commit, next.Rollout, next.Since = "", "", nil, "", "", time.Time{}
		next.SentUnknown = time.Time{}
		if hand {
			next.SentUnknown = at.UTC()
		}
		if next.Build == k.Value.Build && next.Commit == k.Value.Commit && next.Rollout == k.Value.Rollout &&
			next.Artifact == k.Value.Artifact && next.Package == k.Value.Package && next.Since.Equal(k.Value.Since) &&
			next.SentUnknown.Equal(k.Value.SentUnknown) {
			return nil
		}
		if _, err := c.Assignments.Update(ctx, k.Key, next, k.Revision); errors.Is(err, chain.ErrMoved) {
			cur, found, gerr := c.Assignments.Get(ctx, k.Key)
			if gerr != nil {
				return gerr
			}
			if !found || !cur.Value.Stands() {
				return nil
			}
			k = chain.Kept[chain.Assignment]{Key: k.Key, Value: cur.Value, Revision: cur.Revision}
			continue
		} else {
			return err
		}
	}
	return fmt.Errorf("the assignment kept moving under the write")
}

func (i *Inventory) recordRun(ctx context.Context, c *chain.Chain, node, module, commit string, rollout rolloutOf, at time.Time) error {
	key := assignmentKey(node, module)
	var build, rolloutID string
	if sent, ours := rollout.modules[module]; ours && sent.Commit == commit {
		build, rolloutID = sent.Build, rollout.id
	}
	var published *chain.Published
	resolved := false
	for try := 0; try < casTries; try++ {
		cur, found, err := c.Assignments.Get(ctx, key)
		if err != nil {
			return err
		}
		if found && cur.Value.State != "" {
			return nil // a wait for a build: the module is not on the machine by it
		}
		standing := found && cur.Value.Stands()
		same := standing && cur.Value.Commit == commit
		if same && cur.Value.Build != "" && cur.Value.SentUnknown.IsZero() && (rolloutID == "" || cur.Value.Rollout != "") {
			return nil // a send that carries what the machine runs already says nothing new
		}
		if !resolved {
			// The build of the commit is looked up once, and only for a module that moved or names no build.
			if build == "" {
				if build, err = i.BuildOf(ctx, module, commit); err != nil {
					return err
				}
			}
			if build != "" {
				if b, found, err := i.BuildByID(ctx, build); err != nil {
					return err
				} else if found {
					published = publishedOf(b.Made)
				}
			}
			resolved = true
		}
		next := cur.Value
		switch {
		case !standing:
			// Carried and assigned to nobody here: a standing key, no time assigned, marked pulled in. An ended one is
			// written over, the ended one staying in history.
			next = chain.Assignment{Node: node, Module: module, PulledIn: true, Commit: commit, Build: build, Rollout: rolloutID,
				Since: at.UTC()}
		case same:
			// Already running it: fill what is missing, blank nothing.
			if next.Build == "" {
				next.Build = build
			}
			if next.Rollout == "" {
				next.Rollout = rolloutID
			}
			if next.Since.IsZero() {
				next.Since = at.UTC()
			}
		default:
			next.Commit, next.Build, next.Rollout, next.Since = commit, build, rolloutID, at.UTC()
			next.Artifact, next.Package = "", nil
		}
		next.SentUnknown = time.Time{}
		if published != nil && next.Artifact == "" && next.Package == nil && next.Build == build {
			next.Artifact, next.Package = published.Artifact, published.Package
		}
		if standing && next.Build == cur.Value.Build && next.Commit == cur.Value.Commit && next.Rollout == cur.Value.Rollout &&
			next.Artifact == cur.Value.Artifact && (next.Package == nil) == (cur.Value.Package == nil) &&
			next.Since.Equal(cur.Value.Since) && next.SentUnknown.Equal(cur.Value.SentUnknown) {
			return nil
		}
		if !found {
			if _, err := c.Assignments.Create(ctx, key, next); errors.Is(err, chain.ErrExists) {
				continue
			} else {
				return err
			}
		}
		if _, err := c.Assignments.Update(ctx, key, next, cur.Revision); errors.Is(err, chain.ErrMoved) {
			continue
		} else {
			return err
		}
	}
	return fmt.Errorf("the assignment kept moving under the write")
}

// CopySentBuilds copies what delivery 1 recorded in node.sent_builds onto the keys of the assignments, once, at the start of
// the controller that holds the lease and has attached the chain (issue 1109, review round 1): delivery 1 wrote no commit onto
// the key of a module an assignment pulled in, so without it those modules would be missing from SentBuilds although the
// machine is known to have been sent. A key that names a commit is never overwritten, so it is idempotent; once a node's keys
// are written its column is emptied, so a later restart cannot lay a stale commit over a key a later send cleared or ended.
// It answers how many keys it wrote; an error is raised by the caller as a condition.
func (i *Inventory) CopySentBuilds(ctx context.Context) (int, error) {
	c, err := i.theChain(ctx)
	if err != nil {
		return 0, err
	}
	rows, err := i.store.Pool().Query(ctx, `select name, sent_builds from node where sent_builds is not null`)
	if err != nil {
		return 0, err
	}
	type kept struct {
		name string
		sent map[string]string
	}
	var all []kept
	for rows.Next() {
		var name string
		var raw []byte
		if err := rows.Scan(&name, &raw); err != nil {
			rows.Close()
			return 0, err
		}
		sent := map[string]string{}
		if err := json.Unmarshal(raw, &sent); err != nil {
			rows.Close()
			return 0, fmt.Errorf("what %s was sent, in the store, cannot be read: %w", name, err)
		}
		all = append(all, kept{name, sent})
	}
	rows.Close()
	if err := rows.Err(); err != nil {
		return 0, err
	}
	now := time.Now().UTC()
	for _, k := range all {
		if err := i.copySentBuildsOf(ctx, c, k.name, k.sent, now); err != nil {
			return 0, fmt.Errorf("what %s was sent cannot be copied onto its assignments: %w", k.name, err)
		}
		if _, err := i.store.Pool().Exec(ctx, `update node set sent_builds = null where name = $1`, k.name); err != nil {
			return 0, err
		}
	}
	return len(all), nil
}

// copySentBuildsOf is CopySentBuilds for one node, over a chain: for each module the node was sent, a standing key that
// names no commit gets it (and the build, when one is known; no rollout), and where no key stands a standing pulled-in key is
// made. A key with a commit, a hand send's unknown and a wait for a build are left as they are.
func (i *Inventory) copySentBuildsOf(ctx context.Context, c *chain.Chain, node string, sent map[string]string, at time.Time) error {
	for _, module := range slices.Sorted(maps.Keys(sent)) {
		commit := sent[module]
		if commit == "" {
			continue
		}
		key := assignmentKey(node, module)
		build, err := i.chainBuildOf(ctx, c, module, commit)
		if err != nil {
			return err
		}
		for try := 0; ; try++ {
			if try == casTries {
				return fmt.Errorf("the assignment %s kept moving under the write", key)
			}
			cur, found, err := c.Assignments.Get(ctx, key)
			if err != nil {
				return err
			}
			if found && (cur.Value.State != "" || cur.Value.Commit != "" || !cur.Value.SentUnknown.IsZero()) {
				break
			}
			next := cur.Value
			if !found || !cur.Value.Stands() {
				next = chain.Assignment{Node: node, Module: module, PulledIn: true, Since: at}
			}
			next.Commit, next.Build = commit, build
			if next.Since.IsZero() {
				next.Since = at
			}
			if !found {
				_, err = c.Assignments.Create(ctx, key, next)
				if errors.Is(err, chain.ErrExists) {
					continue
				}
			} else {
				_, err = c.Assignments.Update(ctx, key, next, cur.Revision)
				if errors.Is(err, chain.ErrMoved) {
					continue
				}
			}
			if err != nil {
				return err
			}
			break
		}
	}
	return nil
}