mesh-host/internal/compose/frombus/twice.go

package frombus

// **A node's plans at two sets of builds, over one read** (issue 1954; novox/hq ADR 0570 clause 2, ADR 0580 clause 2):
// the seat mesh-provisioner's holder composes each node at the builds its wants name and at the newest build of each
// placed module. Both are composed from one read of the bus (placement, records, the indexer's modules and manifests,
// values: base); only the builds the node's modules are composed from differ, each build read once for both, and a
// build taken in once per content by a reader that keeps it (StatesReader). Where the time went is said by phase.

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"sync"
	"time"

	"git.novox.be/novox/mesh-sdk/go/catalogue"
	"git.novox.be/novox/mesh-sdk/go/delivery"
)

// HeldPlans is one scope's plans, as PlansHeld answers them: err a refusal that is no single module's.
type HeldPlans struct {
	Plans map[string]catalogue.Plan
	Held  map[string]string
	Err   error
}

// Phase is one step of a composition and how long it took.
type Phase struct {
	Name string        `json:"phase"`
	Took time.Duration `json:"took"`
}

// PlansHeldTwice is PlansHeld at s and at s with no kept builds (each placed module at its newest build), over one
// read of the bus: what each answers is what PlansHeld answers for that scope. A read that fails is both scopes' err.
// took names each phase in order: the read, then each scope's builds and its plans.
func PlansHeldTwice(ctx context.Context, r Reader, mesh Mesh, s Scope) (at, newest HeldPlans, took []Phase) {
	start := time.Now()
	mark := func(name string) {
		now := time.Now()
		took = append(took, Phase{Name: name, Took: now.Sub(start)})
		start = now
	}
	b, err := readBase(ctx, r, mesh, s)
	mark("read")
	builds := onceBuilds(r)
	newestScope := s
	newestScope.KeptAt = nil
	for _, sc := range []struct {
		scope Scope
		into  *HeldPlans
		name  string
	}{{s, &at, "wanted"}, {newestScope, &newest, "newest"}} {
		held := map[string]string{}
		sc.into.Held = held
		refuse := func(module string, err error) { held[module] = err.Error() }
		b.refuseOwn(refuse)
		if err != nil {
			sc.into.Err = err
			continue
		}
		in, err := b.finish(ctx, r, builds, mesh, sc.scope, b.cloned(), refuse, nil)
		mark("builds at " + sc.name)
		if err != nil {
			sc.into.Err = err
			continue
		}
		keepWhatRuns(&in, sc.scope.Node, sc.scope.Running) // what runs is kept, as Compose keeps it (ADR 0601 §4)
		sc.into.Plans, sc.into.Err = plansHeldOf(in, s.Node, held)
		mark("plans at " + sc.name)
	}
	return at, newest, took
}

// onceBuilds is r's Build, each module's build at a fingerprint read once: both scopes of one composition read the
// builds of the same moment.
func onceBuilds(r Reader) func(ctx context.Context, module, fp string) (*delivery.BuilderBuild, error) {
	type answer struct {
		b   *delivery.BuilderBuild
		err error
	}
	read := map[string]answer{}
	return func(ctx context.Context, module, fp string) (*delivery.BuilderBuild, error) {
		key := module + " " + fp
		if a, ok := read[key]; ok {
			return a.b, a.err
		}
		b, err := r.Build(ctx, module, fp)
		read[key] = answer{b, err}
		return b, err
	}
}

// takenInKeeper is a Reader that keeps what a build's take-in made (StatesReader does): the resolved manifest's bytes,
// by the content it was made from, so a composition made again on every change takes in only a build it has not seen.
type takenInKeeper interface {
	takenIn(key string) ([]byte, bool)
	keepTakenIn(key string, raw []byte)
}

// takenInKey is what a take-in is made from: the artifact store's address and the build's record (its manifest's digest, its artifacts by digest), read
// whole; an archive is checked against its digest on every read, so the same record takes in to the same bytes.
func takenInKey(module, fp, address string, b delivery.BuilderBuild) (string, bool) {
	raw, err := json.Marshal(b)
	if err != nil {
		return "", false
	}
	sum := sha256.Sum256(append([]byte(module+" "+fp+" "+address+" "), raw...))
	return hex.EncodeToString(sum[:]), true
}

// takenInAtMost bounds the take-ins a reader keeps; past it they are emptied, which costs only a take-in again.
const takenInAtMost = 512

// keptTakeIns is a reader's kept take-ins.
type keptTakeIns struct {
	mu   sync.Mutex
	kept map[string][]byte
}

func (k *keptTakeIns) takenIn(key string) ([]byte, bool) {
	k.mu.Lock()
	defer k.mu.Unlock()
	raw, ok := k.kept[key]
	return raw, ok
}

func (k *keptTakeIns) keepTakenIn(key string, raw []byte) {
	k.mu.Lock()
	defer k.mu.Unlock()
	if k.kept == nil || len(k.kept) >= takenInAtMost {
		k.kept = map[string][]byte{}
	}
	k.kept[key] = raw
}

// PhasesSaid says the phases in one line, each in milliseconds.
func PhasesSaid(took []Phase) string {
	out := ""
	for i, p := range took {
		if i > 0 {
			out += ", "
		}
		out += fmt.Sprintf("%s %d ms", p.Name, p.Took.Milliseconds())
	}
	return out
}