Files
mesh-catalog/modules/gitea/internal/packageseats/holder.go
T
jschoubben e97167c58f G-catalog-B 7: a transfer dropped by hand is recorded through mesh-hand-acts first
registry_incoming, npm_incoming and go_incoming dropped a transfer by hand with only a log line. Each
drop now records its hand act through the seat mesh-hand-acts before anything goes, as nats_requests
does; an act not recorded leaves the transfer open. distribution's manifest invokes the seat.

Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
2026-11-04 18:13:44 +01:00

416 lines
17 KiB
Go

// Package packageseats is the forge module's holding of the two package seats, `npm-package-registry` and
// `go-package-registry` (novox/hq design 51 *The four registry seats*, ADR 0605, ADR 0611). A build module publishes
// a package through `publish` and `publish-part`: the package's one file lands in the forge module's own object store
// `incoming`, is hashed, checked against the builder's numbering and then uploaded into the forge's package registry
// with the module's own write side. The receiving side is mesh-sdk's seat-neutral parts.Receiver and the caller list
// mesh-sdk's VerbCallers (ADR 0627 §2): this package binds both to the two seats and to the forge.
package packageseats
import (
"errors"
"fmt"
"strings"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/handacts"
"git.novox.be/novox/mesh-sdk/go/parts"
)
// The seats the forge module holds.
const (
NpmSeat = "npm-package-registry"
GoSeat = "go-package-registry"
)
// Seats are the package seats, each with the forge's own package type behind it.
var Seats = map[string]string{NpmSeat: "npm", GoSeat: "go"}
// The verbs both seats serve (design 51 *One contract*).
const (
VerbPublish = "publish"
VerbPublishPart = "publish-part"
VerbVersions = "versions"
VerbRemove = "remove"
)
// Verbs are the verbs of both seats, in order.
var Verbs = []string{VerbPublish, VerbPublishPart, VerbRemove, VerbVersions}
// The events of both seats.
const (
EventPublished = "published"
EventRefused = "refused"
EventRemoved = "removed"
)
// The bounds of incoming (design 51 *Publishing*, ADR 0611): a chunk at most 1 MiB, a transfer with no chunk for 24
// hours dropped, a finding past 80 % of the cap. The cap is the one incoming's binding carries; IncomingCap is the
// manifest's 2 GiB (a package is one file, far smaller than an image), said only where no binding is read.
const (
MaxChunk = 1 << 20
IncomingCap = 2 << 30
TransferIdle = 24 * time.Hour
IncomingFullPct = 80
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): its
// end is the event that drops a transfer left with no chunk for TransferIdle, never a timer.
IdleState = "idle"
// KindPackage is the one kind a package seat holds.
KindPackage = "package"
// Self is the forge module, the holder every outcome names.
Self = "gitea"
// ReasonRetention is the reason of the forge's own retention removing a version nothing keeps.
ReasonRetention = "retention"
)
// Callers is the trusted config value `callers`, read as mesh-sdk's list per verb.
type Callers = stdio.VerbCallers
// DefaultCallers is design 51's list for the package seats: publish and publish-part the two build modules that
// publish packages; remove the builder (a tagged version whose branch ended); versions anyone on the mesh, as the
// registry's reads are anonymous.
func DefaultCallers() Callers {
builds := []string{"mesh-build-go", "mesh-build-node"}
return Callers{VerbPublish: builds, VerbPublishPart: builds, VerbRemove: {"mesh-delivery-builder"},
VerbVersions: {stdio.AnyCaller}}
}
// ConfiguredCallers is the config value `callers` as mesh-config resolved it; unset is DefaultCallers.
func ConfiguredCallers() (Callers, error) {
return stdio.ConfiguredVerbCallers("callers", Verbs, DefaultCallers())
}
// Publish is a publish as the seats' contract names it (mesh-contracts npmpackageregistry and gopackageregistry share
// one shape): what the tools bundle reads from either seat's PublishInput.
type Publish 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 string `json:"digest"`
Parts []parts.Want `json:"parts,omitempty"`
}
// Opening is what the holder keeps of a publish beside the receiving side's own fields.
type Opening struct {
Publish
Seat string `json:"seat"`
}
// Transfer is one open publish in the state transfers.
type Transfer = parts.Transfer[Opening]
// Outcome is the body of published, refused and removed, as the contract's Outcome names it.
type Outcome struct {
Name, Version, Kind, Tag, Digest, Fingerprint, Commit, Transfer, Reference, Reason string
// Module is the Go module path a Go package is fetched under; empty on the npm seat.
Module string
}
// Answer is what publish and publish-part answer: the transfer, the parts lacked, and whether it ended.
type Answer struct {
Transfer string
Lacks []string
Ended bool
}
// Version is one version a seat holds of a name.
type Version struct {
Version string
Digest string
Published time.Time
}
// Builder answers whether the builder numbered a version with a digest (mesh-sdk delivery.BuilderVersions).
type Builder interface {
Recorded(name, version, digest string) (ok bool, why string, err error)
}
// Store is the forge's package registry, written with the module's own write side alone.
type Store interface {
// Held is the sha256 digest of name@version's file when the registry holds it.
Held(seat, name, version string) (digest string, ok bool, err error)
// Put uploads the package's file and answers the Go module path (empty for npm).
Put(seat string, p Publish, file []byte) (module string, err error)
// Versions is every version of name, newest first.
Versions(seat, name string) ([]Version, error)
// Remove deletes name@version.
Remove(seat, name, version string) error
}
// Holder answers the verbs of both seats.
type Holder struct {
Callers Callers
Objects parts.Objects
Transfers parts.Transfers[Opening]
Builder Builder
Store Store
// Say says an event on a seat's family.
Say func(seat, event string, o Outcome) error
// Finding writes (raised) or clears the module's finding of a kind for a subject.
Finding func(kind, subject, words string, raised bool)
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(key string, ttl time.Duration) error
}
// CapNow is incoming's cap as its binding carries it.
func (h *Holder) CapNow() uint64 { return h.receiver().CapNow() }
func (h *Holder) now() time.Time {
if h.Now != nil {
return h.Now().UTC()
}
return time.Now().UTC()
}
// Reference is the seat's answer for a package: <name>@<version> (ADR 0605 §9).
func Reference(name, version string) string { return name + "@" + version }
// key is a transfer's key: the caller's own name under its seat, so the two seats' transfers never meet.
func key(seat, transfer string) string { return seat + "_" + transfer }
func (h *Holder) receiver() *parts.Receiver[Opening] {
return &parts.Receiver[Opening]{Objects: h.Objects, Transfers: h.Transfers, Cap: IncomingCap, Bound: h.Cap,
MaxChunk: MaxChunk, Idle: TransferIdle, Arm: h.Idle, Full: IncomingFullPct, Now: h.now,
// The builder's numbering is asked before anything is uploaded (ADR 0605 §4).
Check: func(t Transfer) error {
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 {
file := held[t.Shape.Digest]
module, err := h.Store.Put(t.Shape.Seat, t.Shape.Publish, file)
if err != nil {
reason := "the forge 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)
h.Finding("publish-refused", IncomingObjects, "", false)
o := outcomeOf(t.Shape.Publish, "")
o.Reference, o.Module = Reference(t.Shape.Name, t.Shape.Version), module
return h.Say(t.Shape.Seat, EventPublished, o)
},
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 %%)", used, cap, used*100/cap), full)
}}
}
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, outcomeOf(t.Shape.Publish, reason))
}
func outcomeOf(p Publish, reason string) Outcome {
return Outcome{Name: p.Name, Version: p.Version, Kind: p.Kind, Tag: p.Tag, Digest: p.Digest, Fingerprint: p.Fingerprint,
Commit: p.Commit, Transfer: p.Transfer, Reason: reason}
}
func (h *Holder) answer(transfer string, r parts.Received) Answer {
return Answer{Transfer: transfer, Lacks: r.Lacks, Ended: r.Ended}
}
// Publish is `publish` on seat for the proved caller: a package is one file, its digest the file's sha256.
func (h *Holder) Publish(seat, caller string, p Publish) (Answer, error) {
if err := h.Callers.Allowed(VerbPublish, caller); err != nil {
return Answer{}, err
}
if err := checkPublish(seat, p); err != nil {
return Answer{}, err
}
// A rebuild of a published version answers it held and uploads nothing; the same version with other content is
// refused, as the registry would refuse it.
if d, ok, err := h.Store.Held(seat, p.Name, p.Version); err != nil {
return Answer{}, fmt.Errorf("publish: the forge cannot say whether it holds %s@%s: %w", p.Name, p.Version, err)
} else if ok {
if d != p.Digest {
reason := fmt.Sprintf("%s@%s is published with %s, not %s", p.Name, p.Version, d, p.Digest)
return Answer{Transfer: p.Transfer, Ended: true}, h.refused(Transfer{Shape: Opening{Publish: p, Seat: seat}}, reason)
}
o := outcomeOf(p, "")
o.Reference = Reference(p.Name, p.Version)
return Answer{Transfer: p.Transfer, Ended: true}, h.Say(seat, EventPublished, o)
}
// The transfer's name is the caller's own (the seat's contract): none would open the seat's one nameless transfer,
// which key() still makes a valid id.
if strings.TrimSpace(p.Transfer) == "" {
return Answer{}, errors.New("publish: no transfer named; the caller names its own")
}
got, err := h.receiver().Open(key(seat, p.Transfer), caller, p.Parts, Opening{Publish: p, Seat: seat})
if err != nil {
return h.answer(p.Transfer, got), fmt.Errorf("publish: %w", err)
}
return h.answer(p.Transfer, got), nil
}
// PublishPart is `publish-part`: one chunk, from the caller that opened the transfer alone.
func (h *Holder) PublishPart(seat, caller, transfer, part string, sequence int, last bool, data []byte) (Answer, error) {
if err := h.Callers.Allowed(VerbPublishPart, caller); err != nil {
return Answer{}, err
}
got, err := h.receiver().Take(key(seat, transfer), caller, part, sequence, last, data, func(t Transfer) error {
if t.Shape.Seat != seat {
return fmt.Errorf("transfer %s is %s's, not %s's", transfer, t.Shape.Seat, seat)
}
return nil
})
if err != nil {
return h.answer(transfer, got), fmt.Errorf("publish-part: %w", err)
}
return h.answer(transfer, got), nil
}
// Versions is `versions`: every version the seat holds of name, newest first.
func (h *Holder) Versions(seat, caller, name string) ([]Version, error) {
if err := h.Callers.Allowed(VerbVersions, caller); err != nil {
return nil, err
}
if _, ok := Seats[seat]; !ok {
return nil, fmt.Errorf("versions: the forge holds no seat %s", seat)
}
if strings.TrimSpace(name) == "" {
return nil, errors.New("versions: no name")
}
v, err := h.Store.Versions(seat, name)
if err != nil {
return nil, fmt.Errorf("versions: the forge cannot list %s: %w", name, err)
}
return v, nil
}
// Remove is `remove`: the builder removes a tagged version (its branch ended); an untagged version goes only by the
// forge's own retention, never by a call. A version already gone is said removed all the same, so a retry ends.
func (h *Holder) Remove(seat, caller, name, version, reason string) (Outcome, error) {
if err := h.Callers.Allowed(VerbRemove, caller); err != nil {
return Outcome{}, err
}
if _, ok := Seats[seat]; !ok {
return Outcome{}, fmt.Errorf("remove: the forge holds no seat %s", seat)
}
if strings.TrimSpace(name) == "" || strings.TrimSpace(reason) == "" {
return Outcome{}, errors.New("remove: a name and a reason are required")
}
if !strings.Contains(version, "-") {
return Outcome{}, fmt.Errorf("remove: refused: %s@%s has no tag; only a tagged version is removed, an untagged one by retention alone", name, version)
}
d, held, err := h.Store.Held(seat, name, version)
if err != nil {
return Outcome{}, fmt.Errorf("remove: the forge cannot read %s@%s: %w", name, version, err)
}
if held {
if err := h.Store.Remove(seat, name, version); err != nil {
return Outcome{}, fmt.Errorf("remove: the forge did not remove %s@%s: %w", name, version, err)
}
}
o := Outcome{Name: name, Version: version, Kind: KindPackage, Digest: d, Reason: reason}
return o, h.Say(seat, EventRemoved, o)
}
// Retained says `removed` for a version the forge's own retention deleted (design 51 *Retention*).
func (h *Holder) Retained(seat, name, version, digest string) error {
return h.Say(seat, EventRemoved, Outcome{Name: name, Version: version, Kind: KindPackage, Digest: digest, Reason: ReasonRetention})
}
// Drop drops a transfer by hand.
func (h *Holder) Drop(seat, transfer string) (Transfer, error) {
return h.receiver().Drop(key(seat, transfer))
}
// DropByHand is <type>_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(seat, transfer, 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(key(seat, transfer))
if err != nil {
return Transfer{}, fmt.Errorf("transfer %s cannot be read, so nothing is dropped: %w", transfer, err)
}
if !found {
return Transfer{}, fmt.Errorf("transfer %s of %s is not open", transfer, seat)
}
what := fmt.Sprintf("dropped the open transfer %s of %s's incoming (%s@%s, opened by %s), by %s", transfer, seat, t.Shape.Name, t.Shape.Version, t.Opener, by)
if err := h.Record(handacts.Request{What: what, Why: why, Cause: Seats[seat] + "_incoming"}); err != nil {
return Transfer{}, fmt.Errorf("the hand act cannot be recorded, so transfer %s stays open: %w", transfer, err)
}
return h.Drop(seat, transfer)
}
// Incoming is what npm_incoming and go_incoming answer of a seat: its open transfers, incoming's fill (both seats
// share the store) and the cap its binding carries.
type Incoming struct {
Seat string `json:"seat"`
Transfers []Transfer `json:"transfers"`
Used uint64 `json:"used"`
Cap uint64 `json:"cap"`
}
// IncomingOf reads a seat's open transfers and incoming's fill.
func (h *Holder) IncomingOf(seat string) (Incoming, error) {
if _, ok := Seats[seat]; !ok {
return Incoming{}, fmt.Errorf("the forge holds no seat %s", seat)
}
all, err := h.Transfers.List()
if err != nil {
return Incoming{}, fmt.Errorf("the transfers cannot be read: %w", err)
}
out := Incoming{Seat: seat, Transfers: []Transfer{}, Cap: h.CapNow()}
for _, t := range all {
if t.Shape.Seat == seat {
out.Transfers = append(out.Transfers, t)
}
}
if out.Used, err = h.Objects.Used(); err != nil {
return out, fmt.Errorf("incoming's fill cannot be read: %w", err)
}
return out, nil
}
// Expired is told that a transfer's idle key (its key in the state transfers) ended (the bus's marker,
// parts.FollowIdle): the receiver drops a transfer with no chunk for TransferIdle with its chunks, arms one that took a
// chunk since again for the rest, and leaves one gone. It answers whether it dropped it.
func (h *Holder) Expired(id string) (bool, error) { return h.receiver().Expired(id) }
func checkPublish(seat string, p Publish) error {
switch {
case Seats[seat] == "":
return fmt.Errorf("publish: the forge holds no seat %s", seat)
case strings.TrimSpace(p.Name) == "":
return errors.New("publish: no name")
case strings.TrimSpace(p.Version) == "":
return errors.New("publish: a package is published under the version the builder gave it")
case p.Kind != KindPackage:
return fmt.Errorf("publish: %s holds packages, not a %s", seat, p.Kind)
case !strings.HasPrefix(p.Digest, "sha256:"):
return fmt.Errorf("publish: digest %q is no sha256 digest", p.Digest)
case len(p.Parts) != 1 || p.Parts[0].Digest != p.Digest:
return errors.New("publish: a package is one file, its one part the package's digest")
}
return nil
}