mesh-catalog/modules/mesh-build-image/cmd/mesh-build-image/copy.go

package main

// The copy step (novox/hq ADR 0498 §4; design 51: an upstream image is copied by `mirror` with its pinned digest;
// CAT-BI1): an upstream artifact is copied into the image seat by the seat's verb mirror, which the manifest
// `invokes`, by the digest its manifest pins. The holder copies it, every platform and only what it lacks, and says
// published or refused on the seat's events; the step waits on that outcome, heard through the manifest's `consumes`,
// not on the call, so a large base never outlasts a call's bound; the wait is kept in the state mirror-waits, so an
// outcome another instance heard is matched there. The copy runs no code of a change and reads no
// outside registry itself. An upstream without a digest is not copied: `not pinned:`. A copy the holder does not end
// with the pinned digest stored is `not stored:`.

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

	reg "git.novox.be/novox/mesh-contracts/go/seats/containerimageregistry"
	stdio "git.novox.be/novox/mesh-sdk/go"
	"git.novox.be/novox/mesh-sdk/go/delivery"

	"git.novox.be/novox/mesh-catalog/modules/mesh-build-go/buildstep"
)

// CopyTool is the copy step's tool.
const CopyTool = "build_image_copy"

// MirrorWait is how long a copy waits for the holder's outcome before it fails `not stored:`.
const MirrorWait = 30 * time.Minute

// CopyResult is what a passed copy step says: the pinned digest, now held by the image seat, and the seat's reference
// to it (container-image-registry/<path>@sha256:<digest>).
type CopyResult struct {
	Artifact  string `json:"artifact"`
	Digest    string `json:"digest"`
	Reference string `json:"reference"`
	Pushed    bool   `json:"pushed"`
}

var (
	nameOK   = regexp.MustCompile(`^[a-z0-9][a-z0-9._-]*(/[a-z0-9][a-z0-9._-]*)*$`)
	pinnedOK = regexp.MustCompile(`@(sha256:[0-9a-f]{64})$`)
)

func (m *Module) copyTool() StepTool {
	return StepTool{
		Name: CopyTool,
		Description: "Copy an upstream image into the seat container-image-registry by the digest the manifest pins, through " +
			"the seat's mirror (novox/hq ADR 0498 §4, design 51). Answers a step id at once; the step ends on its key with the " +
			"digest held and the seat's reference to it, when the holder says published. A reference without a digest fails " +
			"`not pinned:`, a refusal `not stored:`. Needs no workspace and runs no code of a change.",
		Input: map[string]any{
			"type": "object",
			"properties": map[string]any{
				"module":   map[string]any{"type": "string", "description": "the module that names the upstream artifact"},
				"artifact": map[string]any{"type": "string", "description": "the upstream artifact's name in the module's manifest"},
				"from":     map[string]any{"type": "string", "description": "the image reference the manifest gives, pinned by digest, such as nats:2.11.17-alpine@sha256:..."},
			},
			"required": []string{"module", "artifact", "from"},
		},
		Long: true,
		Run:  m.copy,
	}
}

// pinned is the digest a reference pins.
func pinned(from string) (string, error) {
	pin := pinnedOK.FindStringSubmatch(from)
	if pin == nil || strings.TrimSuffix(from, "@"+pin[1]) == "" {
		return "", fmt.Errorf("%q names no digest", from)
	}
	return pin[1], nil
}

func (m *Module) copy(ctx context.Context, args map[string]any) (any, error) {
	module, _ := args["module"].(string)
	name, _ := args["artifact"].(string)
	from, _ := args["from"].(string)
	if !nameOK.MatchString(module) || !nameOK.MatchString(name) {
		return nil, fmt.Errorf("%q / %q: a module and an artifact are named in lower case letters, digits, dots, dashes and underscores", module, name)
	}
	digest, err := pinned(from)
	if err != nil {
		return nil, fmt.Errorf("not pinned: %s: %v; an upstream artifact is copied by the digest its manifest pins", name, err)
	}
	if m.Outcomes == nil || m.Outcomes.Waits == nil {
		return nil, fmt.Errorf("%s%s: this process hears no outcome of the image seat", delivery.NotStored, name)
	}
	key := MirrorWaitKey(digest)
	now := time.Now
	if m.Now != nil {
		now = m.Now
	}
	asked := CopyWait{Digest: digest, Module: module, Artifact: name, From: from, Node: m.Node,
		Asked: now().UTC(), Token: strconv.FormatInt(now().UnixNano(), 36) + "-" + m.Node}
	// Waited for before the ask, in memory and in the state mirror-waits, so an outcome said at once, or heard by another
	// instance of the module (the seat's events reach one of them), is not missed.
	heard, stop := m.Outcomes.Expect(digest)
	defer stop()
	if _, err := m.Outcomes.Waits.Put(key, asked); err != nil {
		return nil, fmt.Errorf("%s%s: keeping the copy's wait in %s: %v", delivery.NotStored, name, MirrorWaitsState, err)
	}
	defer m.Outcomes.drop(key, asked.Token)
	said := buildstep.NarrateSeat(ctx, buildstep.ImageSeat, "mirror of "+from)
	raw, err := m.call()("seat:"+buildstep.ImageSeat+"."+reg.VerbMirror, reg.MirrorInput{Name: module + "/" + name, Source: from, Digest: digest})
	if err != nil {
		said.Refused(err)
		return nil, fmt.Errorf("%s%s: %s's mirror: %v", delivery.NotStored, name, buildstep.ImageSeat, err)
	}
	var answer reg.MirrorAnswer
	_ = json.Unmarshal(raw, &answer)
	m.say("asked %s to copy %s (%s); waiting for its outcome", buildstep.ImageSeat, from, answer.Reference)
	wait := m.MirrorWait
	if wait == 0 {
		wait = MirrorWait
	}
	bound := time.After(wait)
	for {
		// The state first: an outcome written there while this copy did not hear it is this copy's.
		if o, ok := m.Outcomes.kept(key); ok {
			res, err := copyEnd(name, o, answer.Reference)
			if err != nil {
				said.Refused(errors.New(o.Reason))
			} else {
				said.Stored(res.(CopyResult).Reference)
			}
			return res, err
		}
		select {
		case <-heard:
		case <-said.Waiting():
			said.StillWaiting()
		case <-bound:
			err := fmt.Errorf("%s said no outcome of the copy of %s in %s", buildstep.ImageSeat, from, wait)
			said.Refused(err)
			return nil, fmt.Errorf("%s%s: %w", delivery.NotStored, name, err)
		case <-ctx.Done():
			said.Refused(ctx.Err())
			return nil, fmt.Errorf("%s%s: stopped while waiting for the copy of %s: %w", delivery.NotStored, name, from, ctx.Err())
		}
	}
}

// copyEnd is a copy's end on the outcome kept for it.
func copyEnd(name string, o MirrorOutcome, asked string) (any, error) {
	if o.Event == reg.EventRefused {
		return nil, fmt.Errorf("%s%s: %s refused the copy: %s", delivery.NotStored, name, buildstep.ImageSeat, o.Reason)
	}
	ref := o.Reference
	if ref == "" {
		ref = asked
	}
	return CopyResult{Artifact: name, Digest: o.Digest, Reference: ref, Pushed: true}, nil
}

// The state mirror-waits (CAT-BI1, the lead's ruling on A3r): each upstream copy waiting on the image seat's outcome,
// keyed by its digest, with the outcome once an instance of the module heard it. The seat's published and refused reach
// one instance of the module, not necessarily the one whose copy waits, and none while no instance listens; the
// instance that hears it writes it to the copy's key, and the waiting copy reads it there, woken by its watch of the
// state. A copy removes its key when it ends; one a stopped process left goes with the state's time to live.
const MirrorWaitsState = "mirror-waits"

// MirrorWaitKey is a copy's key in mirror-waits: its digest, with the colon a key cannot hold as a dash.
func MirrorWaitKey(digest string) string { return strings.Replace(digest, ":", "-", 1) }

// CopyWait is one value of mirror-waits: the copy asked, and its outcome once heard.
type CopyWait struct {
	Digest   string         `json:"digest"`
	Module   string         `json:"module"`
	Artifact string         `json:"artifact"`
	From     string         `json:"from"`
	Node     string         `json:"node"`
	Asked    time.Time      `json:"asked"`
	Token    string         `json:"token"`
	Outcome  *MirrorOutcome `json:"outcome,omitempty"`
}

// MirrorOutcome is the image seat's published or refused of a copy, as kept.
type MirrorOutcome struct {
	reg.Outcome
	Event string    `json:"event"`
	Heard time.Time `json:"heard"`
	By    string    `json:"by,omitempty"`
}

// WaitState is the state mirror-waits: a key's value (nil for none), a write, a removal.
type WaitState interface {
	Get(key string) (json.RawMessage, error)
	Put(key string, value any) (uint64, error)
	Delete(key string) error
}

// stateWaits is mirror-waits as the process keeps it on the bus.
type stateWaits struct{ s stdio.KeptState }

func (w stateWaits) Get(key string) (json.RawMessage, error) {
	e, err := w.s.Get(key)
	if err != nil || e == nil {
		return nil, err
	}
	return e.Value, nil
}
func (w stateWaits) Put(key string, value any) (uint64, error) { return w.s.Put(key, value) }
func (w stateWaits) Delete(key string) error                   { return w.s.Delete(key) }

// Outcomes hands the image seat's published and refused of an upstream copy to the copy waiting on its digest: written
// into the copy's key in Waits, and a wake for the copies of this process waiting on it.
type Outcomes struct {
	// Waits is the state mirror-waits; Node names this instance in an outcome it writes; Now is the clock.
	Waits WaitState
	Node  string
	Now   func() time.Time

	mu      sync.Mutex
	waiting map[string][]chan struct{}
}

// Expect wakes on every change of digest's wait; stop ends it.
func (o *Outcomes) Expect(digest string) (<-chan struct{}, func()) {
	ch := make(chan struct{}, 1)
	o.mu.Lock()
	if o.waiting == nil {
		o.waiting = map[string][]chan struct{}{}
	}
	o.waiting[digest] = append(o.waiting[digest], ch)
	o.mu.Unlock()
	return ch, func() {
		o.mu.Lock()
		defer o.mu.Unlock()
		left := o.waiting[digest][:0]
		for _, c := range o.waiting[digest] {
			if c != ch {
				left = append(left, c)
			}
		}
		if len(left) == 0 {
			delete(o.waiting, digest)
		} else {
			o.waiting[digest] = left
		}
	}
}

func (o *Outcomes) wake(digest string) {
	o.mu.Lock()
	defer o.mu.Unlock()
	for _, ch := range o.waiting[digest] {
		select {
		case ch <- struct{}{}:
		default:
		}
	}
}

// read is the wait kept at key, nil for none or one this module cannot read.
func (o *Outcomes) read(key string) (*CopyWait, error) {
	raw, err := o.Waits.Get(key)
	if err != nil || len(raw) == 0 || string(raw) == "null" {
		return nil, err
	}
	var w CopyWait
	if json.Unmarshal(raw, &w) != nil || w.Digest == "" {
		return nil, nil
	}
	return &w, nil
}

// kept is the outcome kept for key's copy, when one was heard.
func (o *Outcomes) kept(key string) (MirrorOutcome, bool) {
	w, err := o.read(key)
	if err != nil || w == nil || w.Outcome == nil {
		return MirrorOutcome{}, false
	}
	return *w.Outcome, true
}

// drop removes key's wait when it is still the copy's own (token): a later copy of the same digest keeps its own.
func (o *Outcomes) drop(key, token string) {
	if w, err := o.read(key); err == nil && w != nil && w.Token == token {
		_ = o.Waits.Delete(key)
	}
}

// Heard takes one event of the seat (published or refused) in: an upstream copy's outcome is written into its wait in
// mirror-waits and wakes the copies of this process waiting on its digest; any other, and one no copy waits on, is
// passed. A wait the state cannot take is an error, so the event is offered again.
func (o *Outcomes) Heard(event string, body json.RawMessage) error {
	var got reg.Outcome
	if err := json.Unmarshal(body, &got); err != nil {
		return fmt.Errorf("the image seat's %s is not an outcome: %w", event, err)
	}
	if got.Kind != "upstream" || got.Digest == "" || o.Waits == nil {
		return nil
	}
	key := MirrorWaitKey(got.Digest)
	w, err := o.read(key)
	if err != nil {
		return fmt.Errorf("reading the copy's wait %s: %w", key, err)
	}
	if w == nil {
		return nil
	}
	if w.Outcome == nil {
		now := time.Now
		if o.Now != nil {
			now = o.Now
		}
		w.Outcome = &MirrorOutcome{Outcome: got, Event: event, Heard: now().UTC(), By: o.Node}
		if _, err := o.Waits.Put(key, w); err != nil {
			return fmt.Errorf("keeping the outcome of %s in %s: %w", key, MirrorWaitsState, err)
		}
	}
	o.wake(got.Digest)
	return nil
}

// Changed takes one change of mirror-waits in, as the module's watch of the state hands it: a wait whose outcome
// another instance wrote wakes this process's copies of its digest.
func (o *Outcomes) Changed(change stdio.StateChange) error {
	var w CopyWait
	if len(change.Value) == 0 || json.Unmarshal(change.Value, &w) != nil || w.Outcome == nil {
		return nil
	}
	o.wake(w.Digest)
	return nil
}

// MirrorWaitsTool reads mirror-waits.
const MirrorWaitsTool = "build_image_mirror_waits"

// Waiting reads every copy waiting in mirror-waits, oldest first, with its outcome once heard.
func (m *Module) Waiting() ([]CopyWait, error) {
	if m.ReadWaits.Keys == nil || m.ReadWaits.Get == nil {
		return nil, fmt.Errorf("the state %s cannot be read here", MirrorWaitsState)
	}
	keys, err := m.ReadWaits.Keys()
	if err != nil {
		return nil, err
	}
	raws, err := readValues(keys, m.ReadWaits.Get)
	if err != nil {
		return nil, err
	}
	out := []CopyWait{}
	for _, raw := range raws {
		var w CopyWait
		if len(raw) == 0 || json.Unmarshal(raw, &w) != nil || w.Digest == "" {
			continue
		}
		out = append(out, w)
	}
	sort.Slice(out, func(i, j int) bool { return out[i].Asked.Before(out[j].Asked) })
	return out, nil
}

func (m *Module) mirrorWaitsTool() stdio.Tool {
	return stdio.Tool{Name: MirrorWaitsTool,
		Description: "The upstream copies waiting on the image seat's outcome of their mirror (the module's state " + MirrorWaitsState +
			"), oldest first: the digest, the module and artifact, the reference, the node whose copy asked and when, and the " +
			"seat's published or refused once an instance heard it. A copy removes its wait when it ends. Read-only.",
		Run: func(map[string]any) (any, error) {
			w, err := m.Waiting()
			return map[string]any{"waits": w}, err
		}}
}

// Plain is a long step that needs nothing of the machine: no guard. The copy runs no buildah.
func (m *Module) Plain(tool string, run func(ctx context.Context, args map[string]any) (any, error)) func(map[string]any) (any, error) {
	return func(args map[string]any) (any, error) {
		id, err := buildstep.Start(m.Ctx, m.Steps, m.Node, tool, func(ctx context.Context) (any, error) { return run(ctx, args) })
		if err != nil {
			return nil, err
		}
		key, _ := delivery.StepKey(m.Node, id)
		return map[string]any{"step": id, "key": key, "state": delivery.StepRunning}, nil
	}
}