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 `.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: `//@sha256:` (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 `.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 }