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

package holder

import (
	"bytes"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"fmt"
	"slices"
	"sort"
	"strings"
	"time"
)

// The bounds of incoming (design 51 *Publishing*, ADR 0611): a part is at most 1 MiB, the store 20 GiB, a transfer
// with no part for 24 hours is dropped, and past 80 % of the cap the holder writes a finding.
const (
	MaxPart         = 1 << 20
	IncomingCap     = 20 << 30
	TransferIdle    = 24 * time.Hour
	IncomingFullPct = 80
	// IncomingObjects and TransfersState are the object store and the state distribution asks as bus state.
	IncomingObjects = "incoming"
	TransfersState  = "transfers"
)

// The kinds a seat holds.
const (
	KindImage    = "image"
	KindBundle   = "bundle"
	KindArchive  = "archive"
	KindUpstream = "upstream"
)

// Part is one file of a publish: an image's layer, config or manifest, a bundle's or an archive's blob, by its digest.
type Part struct {
	Digest string `json:"digest"`
	Size   int64  `json:"size"`
	// MediaType says what the part is to the store; an image's manifest names its own.
	MediaType string `json:"media_type,omitempty"`
}

// PublishRequest is the body of `publish` (ADR 0611 §1).
type PublishRequest struct {
	Transfer    string `json:"transfer"`
	Name        string `json:"name"`
	Version     string `json:"version,omitempty"`
	Tag         string `json:"tag,omitempty"`
	Kind        string `json:"kind"`
	Fingerprint string `json:"fingerprint,omitempty"`
	Commit      string `json:"commit,omitempty"`
	// Digest is the digest the build recorded: an image's manifest, a bundle's or an archive's blob.
	Digest string `json:"digest"`
	Parts  []Part `json:"parts"`
}

// PublishAnswer is what `publish` answers: the parts the holder lacks, which the build module sends; none when it
// holds them all and the publish ended (Published names how).
type PublishAnswer struct {
	Transfer string   `json:"transfer"`
	Missing  []string `json:"missing"`
	// Ended says the publish ended at once: everything was held, `published` or `refused` is said.
	Ended bool `json:"ended,omitempty"`
}

// PartRequest is the body of `publish-part` (ADR 0611 §2): one chunk of one part.
type PartRequest struct {
	Transfer string `json:"transfer"`
	Digest   string `json:"digest"`
	Sequence int    `json:"sequence"`
	Last     bool   `json:"last"`
	Data     []byte `json:"data"`
}

// Transfer is one open publish, kept in the state transfers; its chunks are objects of incoming.
type Transfer struct {
	PublishRequest
	Seat   string `json:"seat"`
	Opener string `json:"opener"`
	// Held are the parts whose every chunk arrived and hashed right; Chunks the chunks heard per part.
	Held   []string       `json:"held,omitempty"`
	Chunks map[string]int `json:"chunks,omitempty"`
	Opened time.Time      `json:"opened"`
	Last   time.Time      `json:"last"`
}

func (t Transfer) missing() []string {
	var out []string
	for _, p := range t.Parts {
		if !slices.Contains(t.Held, p.Digest) && !slices.Contains(out, p.Digest) {
			out = append(out, p.Digest)
		}
	}
	sort.Strings(out)
	return out
}

// Event is the body of `published`, `refused` and `removed` on `<seat>.events` (design 51 *One contract*): the name,
// version, kind, tag, digest, fingerprint, commit and holder; `refused` its reason, `removed` its reason too.
type Event struct {
	Name        string `json:"name"`
	Version     string `json:"version,omitempty"`
	Kind        string `json:"kind"`
	Tag         string `json:"tag,omitempty"`
	Digest      string `json:"digest"`
	Fingerprint string `json:"fingerprint,omitempty"`
	Commit      string `json:"commit,omitempty"`
	Holder      string `json:"holder"`
	Transfer    string `json:"transfer,omitempty"`
	// Reference is the seat's answer: `<registry>/<module>/<artifact>@sha256:<digest>` (ADR 0605 §9).
	Reference string `json:"reference,omitempty"`
	Reason    string `json:"reason,omitempty"`
	// By is who asked a removal, hold or release: the caller the tool runner proved.
	By string `json:"by,omitempty"`
}

// The events of the seats.
const (
	EventPublished = "published"
	EventRefused   = "refused"
	EventRemoved   = "removed"
	EventHeld      = "held"
	EventReleased  = "released"
)

// Objects is distribution's own object store incoming.
type Objects interface {
	Put(name string, data []byte) error
	Get(name string) ([]byte, error)
	Delete(name string) error
	// Used is the bytes the store holds.
	Used() (uint64, error)
}

// TransferState is the state transfers.
type TransferState interface {
	Get(id string) (Transfer, bool, error)
	Put(t Transfer) error
	Delete(id string) error
	List() ([]Transfer, error)
}

// Builder is the builder's tool builder_versions, asked before anything is stored (ADR 0605 §4).
type Builder interface {
	Recorded(name, version, digest string) (ok bool, why string, err error)
}

// Store is the artifact store behind the loopback, written by distribution's own process alone.
type Store interface {
	// Has says whether the store holds a blob of name.
	Has(name, digest string) (bool, error)
	// Put stores a transfer's parts under its name and version and answers the reference.
	Put(t Transfer, parts map[string][]byte) (reference string, err error)
	// Held is the reference of name at version when the store already holds digest under it: a rebuild of a held
	// fingerprint answers it and stores nothing.
	Held(name, version, digest string) (reference string, ok bool, err error)
}

// Say says an event on a seat's family `<seat>.events`.
type Say func(seat, event string, body Event) error

// Finding writes (raised true) or clears distribution's finding of a kind for a subject.
type Finding func(kind, subject, words string, raised bool)

// Holder answers the verbs.
type Holder struct {
	Callers   Callers
	Objects   Objects
	Transfers TransferState
	Builder   Builder
	Store     Store
	Say       Say
	Finding   Finding
	Now       func() time.Time
	// Cap is incoming's cap; IncomingCap when zero.
	Cap uint64
}

func (h *Holder) now() time.Time {
	if h.Now != nil {
		return h.Now().UTC()
	}
	return time.Now().UTC()
}

func (h *Holder) cap() uint64 {
	if h.Cap > 0 {
		return h.Cap
	}
	return IncomingCap
}

// Publish is `publish` on seat for the proved caller.
func (h *Holder) Publish(seat, caller string, r PublishRequest) (PublishAnswer, error) {
	if err := h.Callers.Allowed(VerbPublish, caller); err != nil {
		return PublishAnswer{}, err
	}
	if err := checkPublish(seat, r); err != nil {
		return PublishAnswer{}, err
	}
	if t, found, err := h.Transfers.Get(r.Transfer); err != nil {
		return PublishAnswer{}, err
	} else if found {
		// A publish opening a transfer already open answers the parts held: a cut transfer goes on from there.
		if t.Opener != caller {
			return PublishAnswer{}, fmt.Errorf("publish: refused for %s: transfer %s was opened by another caller", caller, r.Transfer)
		}
		return PublishAnswer{Transfer: t.Transfer, Missing: t.missing()}, nil
	}
	// A rebuild of a held fingerprint answers the held version and stores nothing.
	if r.Version != "" {
		if ref, ok, err := h.Store.Held(r.Name, r.Version, r.Digest); err != nil {
			return PublishAnswer{}, fmt.Errorf("publish: the store cannot say whether it holds %s@%s: %w", r.Name, r.Version, err)
		} else if ok {
			return PublishAnswer{Transfer: r.Transfer, Ended: true}, h.Say(seat, EventPublished, h.eventOf(seat, r, ref, ""))
		}
	}
	used, err := h.Objects.Used()
	if err != nil {
		return PublishAnswer{}, fmt.Errorf("publish: incoming's fill cannot be read: %w", err)
	}
	var size uint64
	for _, p := range r.Parts {
		size += uint64(max(p.Size, 0))
	}
	if used+size > h.cap() {
		reason := fmt.Sprintf("incoming is at its cap of %d bytes (%d held, %d asked)", h.cap(), used, size)
		_ = h.Say(seat, EventRefused, h.eventOf(seat, r, "", reason))
		return PublishAnswer{Transfer: r.Transfer, Ended: true}, errors.New("publish: refused: " + reason)
	}
	t := Transfer{PublishRequest: r, Seat: seat, Opener: caller, Opened: h.now(), Last: h.now(), Chunks: map[string]int{}}
	// Only the missing layers are sent: a part the store already holds under the name counts as held.
	for _, p := range r.Parts {
		if ok, err := h.Store.Has(r.Name, p.Digest); err == nil && ok {
			t.Held = append(t.Held, p.Digest)
		}
	}
	if err := h.Transfers.Put(t); err != nil {
		return PublishAnswer{}, fmt.Errorf("publish: transfer %s cannot be kept: %w", r.Transfer, err)
	}
	h.fill()
	if len(t.missing()) == 0 {
		return PublishAnswer{Transfer: r.Transfer, Ended: true}, h.finish(t)
	}
	return PublishAnswer{Transfer: r.Transfer, Missing: t.missing()}, nil
}

// PublishPart is `publish-part`: one chunk, from the caller that opened the transfer alone.
func (h *Holder) PublishPart(seat, caller string, r PartRequest) (PublishAnswer, error) {
	if err := h.Callers.Allowed(VerbPublishPart, caller); err != nil {
		return PublishAnswer{}, err
	}
	t, found, err := h.Transfers.Get(r.Transfer)
	switch {
	case err != nil:
		return PublishAnswer{}, err
	case !found:
		return PublishAnswer{}, fmt.Errorf("publish-part: transfer %s is not open (dropped after %s with no part, or never opened by publish)", r.Transfer, TransferIdle)
	case t.Opener != caller:
		return PublishAnswer{}, fmt.Errorf("publish-part: refused for %s: transfer %s was opened by another caller", caller, r.Transfer)
	case t.Seat != seat:
		return PublishAnswer{}, fmt.Errorf("publish-part: transfer %s is %s's, not %s's", r.Transfer, t.Seat, seat)
	case len(r.Data) > MaxPart:
		return PublishAnswer{}, fmt.Errorf("publish-part: a chunk of %d bytes is over the %d a part may carry", len(r.Data), MaxPart)
	case !slices.ContainsFunc(t.Parts, func(p Part) bool { return p.Digest == r.Digest }):
		return PublishAnswer{}, fmt.Errorf("publish-part: transfer %s names no part %s", r.Transfer, r.Digest)
	case slices.Contains(t.Held, r.Digest):
		return PublishAnswer{Transfer: t.Transfer, Missing: t.missing()}, nil
	case r.Sequence != t.Chunks[r.Digest]:
		return PublishAnswer{}, fmt.Errorf("publish-part: part %s of transfer %s waits for chunk %d, not %d", r.Digest, r.Transfer, t.Chunks[r.Digest], r.Sequence)
	}
	if err := h.Objects.Put(chunkName(t.Transfer, r.Digest, r.Sequence), r.Data); err != nil {
		return PublishAnswer{}, fmt.Errorf("publish-part: incoming did not take the chunk: %w", err)
	}
	t.Chunks[r.Digest] = r.Sequence + 1
	t.Last = h.now()
	if r.Last {
		whole, err := h.part(t, r.Digest)
		if err != nil {
			return PublishAnswer{}, err
		}
		if got := "sha256:" + sum(whole); got != r.Digest {
			h.dropChunks(t, r.Digest)
			delete(t.Chunks, r.Digest)
			_ = h.Transfers.Put(t)
			return PublishAnswer{}, fmt.Errorf("publish-part: refused: the bytes of part %s hash to %s; send the part again", r.Digest, got)
		}
		t.Held = append(t.Held, r.Digest)
	}
	if err := h.Transfers.Put(t); err != nil {
		return PublishAnswer{}, err
	}
	h.fill()
	if len(t.missing()) == 0 {
		return PublishAnswer{Transfer: t.Transfer, Ended: true}, h.finish(t)
	}
	return PublishAnswer{Transfer: t.Transfer, Missing: t.missing()}, nil
}

// finish asks the builder and stores, and says published or refused; the transfer's chunks go either way.
func (h *Holder) finish(t Transfer) error {
	defer func() {
		for _, p := range t.Parts {
			h.dropChunks(t, p.Digest)
		}
		_ = h.Transfers.Delete(t.Transfer)
		h.fill()
	}()
	refuse := func(reason string) error {
		h.Finding("publish-refused", t.Name+"@"+t.Version, reason, true)
		if err := h.Say(t.Seat, EventRefused, h.eventOf(t.Seat, t.PublishRequest, "", reason)); err != nil {
			return fmt.Errorf("%s; and refused could not be said: %w", reason, err)
		}
		return errors.New("publish: refused: " + reason)
	}
	if t.Version != "" {
		ok, why, err := h.Builder.Recorded(t.Name, t.Version, t.Digest)
		if err != nil {
			return refuse("builder_versions did not answer, so nothing is stored: " + err.Error())
		}
		if !ok {
			return refuse("the builder did not number it: " + why)
		}
	}
	parts := map[string][]byte{}
	for _, p := range t.Parts {
		if _, had := t.Chunks[p.Digest]; !had {
			continue // held by the store already
		}
		b, err := h.part(t, p.Digest)
		if err != nil {
			return refuse("part " + p.Digest + " cannot be read from incoming: " + err.Error())
		}
		parts[p.Digest] = b
	}
	ref, err := h.Store.Put(t, parts)
	if err != nil {
		return refuse("the store did not take it: " + err.Error())
	}
	h.Finding("publish-refused", t.Name+"@"+t.Version, "", false)
	return h.Say(t.Seat, EventPublished, h.eventOf(t.Seat, t.PublishRequest, ref, ""))
}

func (h *Holder) eventOf(seat string, r PublishRequest, ref, reason string) Event {
	return Event{Name: r.Name, Version: r.Version, Kind: r.Kind, Tag: r.Tag, Digest: r.Digest, Fingerprint: r.Fingerprint,
		Commit: r.Commit, Holder: Self, Transfer: r.Transfer, Reference: ref, Reason: reason}
}

func (h *Holder) part(t Transfer, digest string) ([]byte, error) {
	var buf bytes.Buffer
	for seq := 0; seq < t.Chunks[digest]; seq++ {
		b, err := h.Objects.Get(chunkName(t.Transfer, digest, seq))
		if err != nil {
			return nil, err
		}
		buf.Write(b)
	}
	return buf.Bytes(), nil
}

func (h *Holder) dropChunks(t Transfer, digest string) {
	for seq := 0; seq < t.Chunks[digest]; seq++ {
		_ = h.Objects.Delete(chunkName(t.Transfer, digest, seq))
	}
}

// fill writes incoming's finding at 80 % of the cap and clears it below.
func (h *Holder) fill() {
	used, err := h.Objects.Used()
	if err != nil {
		return
	}
	full := used*100 >= h.cap()*IncomingFullPct
	h.Finding("incoming-full", IncomingObjects, fmt.Sprintf("incoming holds %d of its %d bytes (%d %%); registry_incoming lists and drops transfers", used, h.cap(), used*100/h.cap()), full)
}

// Drop drops a transfer by hand (registry_incoming) or because it was idle for TransferIdle: its chunks and its key.
func (h *Holder) Drop(id string) (Transfer, error) {
	t, found, err := h.Transfers.Get(id)
	if err != nil {
		return Transfer{}, err
	}
	if !found {
		return Transfer{}, fmt.Errorf("transfer %s is not open", id)
	}
	for _, p := range t.Parts {
		h.dropChunks(t, p.Digest)
	}
	if err := h.Transfers.Delete(id); err != nil {
		return Transfer{}, err
	}
	h.fill()
	return t, nil
}

// DropIdle drops every transfer with no part for TransferIdle, answering them; it runs on each publish's open and each
// registry_incoming call, never on a timer (the object store's own lifetime of 24 hours drops the bytes too).
func (h *Holder) DropIdle() ([]string, error) {
	all, err := h.Transfers.List()
	if err != nil {
		return nil, err
	}
	var out []string
	for _, t := range all {
		if h.now().Sub(t.Last) >= TransferIdle {
			if _, err := h.Drop(t.Transfer); err == nil {
				out = append(out, t.Transfer)
			}
		}
	}
	return out, nil
}

func chunkName(transfer, digest string, seq int) string {
	return fmt.Sprintf("%s/%s/%d", transfer, strings.TrimPrefix(digest, "sha256:"), seq)
}

func sum(b []byte) string {
	s := sha256.Sum256(b)
	return hex.EncodeToString(s[:])
}

func checkPublish(seat string, r PublishRequest) error {
	switch {
	case !slices.Contains(Seats, seat):
		return fmt.Errorf("publish: distribution holds no seat %s", seat)
	case strings.TrimSpace(r.Transfer) == "" || strings.ContainsAny(r.Transfer, "/. *>"):
		return fmt.Errorf("publish: transfer %q is no transfer id", r.Transfer)
	case strings.TrimSpace(r.Name) == "":
		return errors.New("publish: no name")
	case !strings.HasPrefix(r.Digest, "sha256:"):
		return fmt.Errorf("publish: digest %q is no sha256 digest", r.Digest)
	case len(r.Parts) == 0:
		return errors.New("publish: no parts")
	}
	switch {
	case seat == ImageSeat && r.Kind != KindImage && r.Kind != KindUpstream:
		return fmt.Errorf("publish: %s holds images, not a %s", seat, r.Kind)
	case seat == ArtifactSeat && r.Kind != KindBundle && r.Kind != KindArchive:
		return fmt.Errorf("publish: %s holds bundles and archives, not a %s", seat, r.Kind)
	}
	for _, p := range r.Parts {
		if !strings.HasPrefix(p.Digest, "sha256:") || len(p.Digest) != len("sha256:")+64 {
			return fmt.Errorf("publish: part digest %q is no sha256 digest", p.Digest)
		}
	}
	return nil
}