package holder import ( "crypto/sha256" "encoding/hex" "errors" "fmt" "slices" "strings" "time" "git.novox.be/novox/mesh-sdk/go/handacts" "git.novox.be/novox/mesh-sdk/go/parts" ) // The bounds of incoming (design 51 *Publishing*, ADR 0611): a part is at most 1 MiB, a transfer with no part for 24 // hours is dropped, and past 80 % of the cap the holder writes a finding. The cap is the one incoming's binding carries // (the bus's own bound of the object store, read through the runtime); IncomingCap is the manifest's 8 GiB, the most // the objects template allows, said only where no binding is read (a test), and the manifest's test holds the two equal. const ( MaxPart = 1 << 20 IncomingCap = 8 << 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" // IdleState is the state of one key per open transfer with its own time to live (a state declared key-ttl): the // bus's marker of its end is the event that drops a transfer left with no part for TransferIdle (ADR 0611 §3), never // a timer. IdleState = "idle" ) // 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"` } // Opening is what distribution keeps of a publish beside the receiving side's own fields: the request and its seat. type Opening struct { PublishRequest Seat string `json:"seat"` } // Transfer is one open publish, kept in the state transfers; its chunks are objects of incoming. The receiving side is // mesh-sdk's seat-neutral parts.Receiver (ADR 0627 §2); distribution binds it to the seats it holds. type Transfer = parts.Transfer[Opening] // 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"` // Tags are, for kept, the other tags still naming its digest. Tags []string `json:"tags,omitempty"` // By is who asked a removal, hold or release: the caller the tool runner proved. By string `json:"by,omitempty"` // ID is the event's id on the bus when the outcome has one of its own (a mirror ask's): the same outcome said // twice, by a redelivery or a second follower, is one message. Empty: the runtime makes one. ID string `json:"-"` } // The events of the seats. const ( EventPublished = "published" EventRefused = "refused" EventRemoved = "removed" EventKept = "kept" EventHeld = "held" EventReleased = "released" ) // Objects is distribution's own object store incoming. type Objects = parts.Objects // TransferState is the state transfers. type TransferState = parts.Transfers[Opening] // 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 publish's parts under its name and version and answers the reference's path (SeatReference). Put(r PublishRequest, parts map[string][]byte) (reference string, err error) // Held is the reference's path 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) } // SeatReference is a reference in the form the seat answers (novox/hq ADR 0605 §9): the seat's name, then the path the // store answers (`@sha256:`), as catalogue.InRegistrySeat reads it back. Nothing set on the process names // a reach: the seat's name is the registry part. func SeatReference(seat, path string) string { return seat + "/" + path } // 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 reads incoming's cap from its binding (parts.BusCap); IncomingCap when nil or when the binding names none. Cap func() (uint64, error) // Record records a hand act through the seat mesh-hand-acts (handacts.Recorder); a drop by hand is refused // without it. Record func(handacts.Request) error // Idle writes a transfer's idle key with its own time to live, once (parts.BusArm over IdleState); the receiver // arms it on each open that leaves a transfer open. Nil arms nothing. Idle func(id string, ttl time.Duration) error } func (h *Holder) now() time.Time { if h.Now != nil { return h.Now().UTC() } return time.Now().UTC() } // CapNow is incoming's cap as its binding carries it. func (h *Holder) CapNow() uint64 { return h.receiver().CapNow() } // receiver binds mesh-sdk's receiving side to distribution's incoming, its transfers state, the builder's check, its // store and its seats' events. func (h *Holder) receiver() *parts.Receiver[Opening] { return &parts.Receiver[Opening]{Objects: h.Objects, Transfers: h.Transfers, Cap: IncomingCap, Bound: h.Cap, MaxChunk: MaxPart, Idle: TransferIdle, Arm: h.Idle, Full: IncomingFullPct, Now: h.now, // Only the missing layers are sent: a part the store already holds under the name counts as held. Holds: func(t Transfer, digest string) bool { ok, err := h.Store.Has(t.Shape.Name, digest) return err == nil && ok }, // builder_versions is asked before anything is stored (ADR 0605 §4). Check: func(t Transfer) error { if t.Shape.Version == "" { return nil } ok, why, err := h.Builder.Recorded(t.Shape.Name, t.Shape.Version, t.Shape.Digest) if err != nil { return errors.New("builder_versions did not answer, so nothing is stored: " + err.Error()) } if !ok { return errors.New("the builder did not number it: " + why) } return nil }, Store: func(t Transfer, held map[string][]byte) error { path, err := h.Store.Put(t.Shape.PublishRequest, held) ref := SeatReference(t.Shape.Seat, path) if err != nil { reason := "the store did not take it: " + err.Error() return errors.Join(errors.New("refused: "+reason), h.refused(t, reason)) } h.Finding("publish-refused", t.Shape.Name+"@"+t.Shape.Version, "", false) // Room was found again: a refusal for want of it clears with the first publish the holder takes. h.Finding("publish-refused", IncomingObjects, "", false) return h.Say(t.Shape.Seat, EventPublished, h.eventOf(t.Shape.Seat, t.Shape.PublishRequest, ref, "")) }, Refuse: h.refused, Filled: func(used, cap uint64, full bool) { h.Finding("incoming-full", IncomingObjects, fmt.Sprintf("incoming holds %d of its %d bytes (%d %%); registry_incoming lists and drops transfers", used, cap, used*100/cap), full) }} } // refused writes the finding and says refused on the transfer's seat. A refusal for want of room is the holder's, // not the file's: its finding stands on incoming and clears when the holder takes a publish again. func (h *Holder) refused(t Transfer, reason string) error { subject := t.Shape.Name + "@" + t.Shape.Version if parts.RefusedAtCap(reason) { subject = IncomingObjects } h.Finding("publish-refused", subject, reason, true) return h.Say(t.Shape.Seat, EventRefused, h.eventOf(t.Shape.Seat, t.Shape.PublishRequest, "", reason)) } func answerOf(r parts.Received) PublishAnswer { return PublishAnswer{Transfer: r.ID, Missing: r.Lacks, Ended: r.Ended} } // 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 } // 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, SeatReference(seat, ref), "")) } } wants := make([]parts.Want, 0, len(r.Parts)) for _, p := range r.Parts { wants = append(wants, parts.Want{Digest: p.Digest, Size: p.Size}) } got, err := h.receiver().Open(r.Transfer, caller, wants, Opening{PublishRequest: r, Seat: seat}) if err != nil { return answerOf(got), fmt.Errorf("publish: %w", err) } return answerOf(got), 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 } got, err := h.receiver().Take(r.Transfer, caller, r.Digest, r.Sequence, r.Last, r.Data, func(t Transfer) error { if t.Shape.Seat != seat { return fmt.Errorf("transfer %s is %s's, not %s's", r.Transfer, t.Shape.Seat, seat) } return nil }) if err != nil { return answerOf(got), fmt.Errorf("publish-part: %w", err) } return answerOf(got), nil } 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} } // 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) { return h.receiver().Drop(id) } // DropByHand is registry_incoming's drop: by a person, with why, recorded as a hand act through mesh-hand-acts before // anything goes; an act not recorded leaves the transfer open. func (h *Holder) DropByHand(id, by, why string) (Transfer, error) { if strings.TrimSpace(why) == "" { return Transfer{}, errors.New("a drop is a hand act: say why") } if h.Record == nil { return Transfer{}, errors.New("no hand act can be recorded here, so nothing is dropped") } t, found, err := h.Transfers.Get(id) if err != nil { return Transfer{}, fmt.Errorf("transfer %s cannot be read, so nothing is dropped: %w", id, err) } if !found { return Transfer{}, fmt.Errorf("transfer %s is not open", id) } what := fmt.Sprintf("dropped the open transfer %s of incoming (%s@%s on %s, opened by %s, %d parts lacked), by %s", id, t.Shape.Name, t.Shape.Version, t.Shape.Seat, t.Opener, len(t.Lacks()), by) if err := h.Record(handacts.Request{What: what, Why: why, Cause: "registry_incoming"}); err != nil { return Transfer{}, fmt.Errorf("the hand act cannot be recorded, so transfer %s stays open: %w", id, err) } return h.Drop(id) } // DropIdle drops every transfer with no part for TransferIdle, answering them: registry_incoming's look, by hand. The // drop itself follows each transfer's idle key (Expired); the objects template keeps for good, so nothing else does. func (h *Holder) DropIdle() ([]string, error) { return h.receiver().DropIdle() } // Expired is told that a transfer's idle key ended (the bus's marker, parts.FollowIdle): the receiver drops a transfer // with no part for TransferIdle, its chunks with it, arms one that took a part since again for the rest, and leaves one // already gone. It answers whether it dropped the transfer. func (h *Holder) Expired(id string) (bool, error) { return h.receiver().Expired(id) } 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.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) } 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) } return nil } func sum(b []byte) string { s := sha256.Sum256(b) return hex.EncodeToString(s[:]) }