diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go index 56516c70..1a1292d4 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go @@ -19,6 +19,7 @@ import ( "git.novox.be/novox/mesh-sdk/go/delivery" "mesh-delivery-builder/internal/build" + "mesh-delivery-builder/internal/versions" ) // queueAsker writes a queues value and waits for the builder's answer to it. @@ -26,9 +27,17 @@ type queueAsker func(key string, value []byte) (json.RawMessage, error) // Tools are the builder's tools: stuck reads, askAgain writes the reasks value, queue writes a queues value and waits for // the builder's read, buildAgain writes a reasks value of mode build and waits for the builder's answer. -func Tools(r build.StuckReader, askAgain func(key string, value []byte) error, queue, buildAgain queueAsker, since build.LogSince, now func() time.Time) []stdio.Tool { +func Tools(r build.StuckReader, askAgain func(key string, value []byte) error, queue, buildAgain queueAsker, since build.LogSince, now func() time.Time, vs versions.Reader) []stdio.Tool { str := func(args map[string]any, k string) string { s, _ := args[k].(string); return strings.TrimSpace(s) } return []stdio.Tool{ + {Name: versions.VersionsTool, Description: versions.VersionsDescription, + Input: map[string]any{"type": "object", "required": []string{"name"}, "properties": map[string]any{ + "name": map[string]any{"type": "string", "description": "the package, image, bundle or archive name"}, + "version": map[string]any{"type": "string", "description": "one version, to ask whether it is recorded"}, + "digest": map[string]any{"type": "string", "description": "with version: the digest the holder received"}}}, + Run: func(args map[string]any) (any, error) { + return versions.AnswerVersions(context.Background(), vs, str(args, "name"), str(args, "version"), str(args, "digest")) + }}, {Name: build.StuckTool, Description: "Every build key that waits (building or queued), oldest first: its module, fingerprint, step words and since when, and each ask it " + "waits on by language with its id, when it was put, its asking (1 for the first; the builder asks an ask again when it can no longer be " + @@ -264,7 +273,15 @@ func main() { a, err := build.DecodeReAsk(v) return err == nil && !a.Answered.IsZero() }) - if err := stdio.Serve("", Tools(r, put, askQueue(stdio.State(build.QueuesState), 25*time.Second), buildAgain, stdio.Log(delivery.BuilderLog).Since, time.Now)); err != nil { + vs := stdio.State(versions.State) + vr := versions.Reader{Keys: vs.Keys, Get: func(key string) (json.RawMessage, bool, error) { + e, err := vs.Get(key) + if err != nil || e == nil { + return nil, false, err + } + return e.Value, true, nil + }} + if err := stdio.Serve("", Tools(r, put, askQueue(stdio.State(build.QueuesState), 25*time.Second), buildAgain, stdio.Log(delivery.BuilderLog).Since, time.Now, vr)); err != nil { fmt.Fprintf(os.Stderr, "[mesh-delivery-builder-tools] %v\n", err) os.Exit(1) } diff --git a/modules/mesh-delivery-builder/internal/build/claim_test.go b/modules/mesh-delivery-builder/internal/build/claim_test.go index ed359a31..54cf5e55 100644 --- a/modules/mesh-delivery-builder/internal/build/claim_test.go +++ b/modules/mesh-delivery-builder/internal/build/claim_test.go @@ -114,12 +114,11 @@ func TestClaimCreatesTheKeyBuildingAndASecondBuilderGetsNothing(t *testing.T) { if second, err := b.Claim(context.Background(), "m", fp1, hex64('a'), ClaimOptions{}); err != nil || second != nil { t.Errorf("second builder's claim = %v, %v, want none: the same source is never built twice", second, err) } -} -func TestClaimLosingTheCreateRaceGetsNothing(t *testing.T) { // 1607/AC3 - st := newMem() + // A claim losing the create race gets nothing (merged into this test for the test budget, design 51). + st = newMem() st.before = func() { seed(t, st, delivery.BuilderBuild{State: delivery.BuildBuilding, By: "novox.b"}) } - got, err := Claimer{Store: st, By: "novox.a"}.Claim(context.Background(), "m", fp1, "", ClaimOptions{}) + got, err = Claimer{Store: st, By: "novox.a"}.Claim(context.Background(), "m", fp1, "", ClaimOptions{}) if err != nil || got != nil { t.Errorf("claim = %v, %v, want none: the create failed because the key exists", got, err) } @@ -197,12 +196,11 @@ func TestFinishMovesTheKeyByCompareAndSetAndWritesOnlyADifferentValue(t *testing if st.writes != writes { t.Errorf("the same value was written again (%d writes, was %d)", st.writes, writes) } -} -func TestFinishAfterTheKeyChangedIsAConflictNotAnOverwrite(t *testing.T) { // 1607/AC3 - st := newMem() - c := Claimer{Store: st, By: "novox.a"} - got, _ := c.Claim(context.Background(), "m", fp1, "", ClaimOptions{}) + // Finish after the key changed is a conflict, not an overwrite (merged into this test for the test budget). + st = newMem() + c = Claimer{Store: st, By: "novox.a"} + got, _ = c.Claim(context.Background(), "m", fp1, "", ClaimOptions{}) st.put(got.Key, []byte(`{"v":1,"state":"failed","error":"someone else"}`)) err := c.Finish(context.Background(), got, delivery.BuilderBuild{State: delivery.BuildFailed, Error: "mine"}) if !errors.Is(err, ErrConflict) { diff --git a/modules/mesh-delivery-builder/internal/versions/store.go b/modules/mesh-delivery-builder/internal/versions/store.go new file mode 100644 index 00000000..4101d9b9 --- /dev/null +++ b/modules/mesh-delivery-builder/internal/versions/store.go @@ -0,0 +1,172 @@ +package versions + +import ( + "context" + "errors" + "fmt" + "time" +) + +// ErrConflict is what a Store answers when a Create finds the key or an Update finds another revision. +var ErrConflict = errors.New("the key exists or changed") + +// Store is the bucket: create and compare-and-set, as the builder's builds bucket (internal/build.Store). +type Store interface { + Get(ctx context.Context, key string) (value []byte, revision uint64, found bool, err error) + Create(ctx context.Context, key string, value []byte) (uint64, error) + Update(ctx context.Context, key string, value []byte, revision uint64) (uint64, error) + // KeysOf are the keys under a filter (NameFilter). + KeysOf(ctx context.Context, filter string) ([]string, error) +} + +// Versions is the builder's state. +type Versions struct { + Store Store + Now func() time.Time +} + +// Of is every entry of name. +func (v Versions) Of(ctx context.Context, name string) ([]Entry, error) { + keys, err := v.Store.KeysOf(ctx, NameFilter(name)) + if err != nil { + return nil, fmt.Errorf("the versions of %s cannot be listed: %w", name, err) + } + var out []Entry + for _, k := range keys { + raw, _, found, err := v.Store.Get(ctx, k) + if err != nil { + return nil, fmt.Errorf("%s: %w", k, err) + } + if !found { + continue + } + e, err := Decode(raw) + if err != nil { + return nil, fmt.Errorf("%s holds what is not a version: %w", k, err) + } + out = append(out, e) + } + return InOrder(out), nil +} + +// Claim is one numbering asked of the builder. +type Claim struct { + Name, Kind, Series, Module, Fingerprint, Commit, Revision, Digest, Seat string + // Tag asks a tagged version for this cleaned tag; empty for a trunk version. + Tag string + Merge uint64 + Floor Floor +} + +// Number numbers a claim under compare-and-set: two claims of one name never get one number, the second reading the +// first's key and taking the next (ADR 0606 §2). It answers the decision and, when it numbered or kept a version, its +// entry. Attempts are bounded by the claims the name can meet at once; a store that keeps conflicting is an error. +func (v Versions) Number(ctx context.Context, c Claim) (Decision, Entry, error) { + if c.Series == "" { + c.Series = DefaultSeries + } + for attempt := 0; attempt < 16; attempt++ { + entries, err := v.Of(ctx, c.Name) + if err != nil { + return Decision{}, Entry{}, err + } + var d Decision + if c.Tag != "" { + if !c.Floor.Known() && len(entries) == 0 { + return Decision{Wait: true, Why: "the tagged version waits for its floor"}, Entry{}, nil + } + ver := Tagged(entries, c.Series, c.Tag, c.Floor) + d = Decision{Version: ver, Publish: true, Why: "tagged " + ver} + } else { + d = Decide(entries, c.Kind, c.Series, c.Fingerprint, c.Revision, c.Floor) + } + if d.Version == "" { + return d, Entry{}, nil + } + if d.Keep { + for _, e := range entries { + if e.Version == d.Version { + return d, e, nil + } + } + } + e := Entry{Name: c.Name, Kind: c.Kind, Series: c.Series, Version: d.Version, Tag: c.Tag, Module: c.Module, + Fingerprint: c.Fingerprint, Commit: c.Commit, Revision: c.Revision, Digest: c.Digest, Seat: c.Seat, + State: Numbered, At: v.now(), Merge: c.Merge} + raw, err := Encode(e) + if err != nil { + return Decision{}, Entry{}, err + } + _, err = v.Store.Create(ctx, Key(c.Name, d.Version), raw) + if errors.Is(err, ErrConflict) { + continue // another claim took this number: read again and take the next + } + if err != nil { + return Decision{}, Entry{}, fmt.Errorf("%s@%s: %w", c.Name, d.Version, err) + } + return d, e, nil + } + return Decision{}, Entry{}, fmt.Errorf("%s: the versions kept changing under 16 claims", c.Name) +} + +// Mark moves one version to a new state by compare-and-set, as the holders' events say (published, refused, removed): +// change edits the entry read; a version the state does not hold is an error naming it. +func (v Versions) Mark(ctx context.Context, name, version string, change func(*Entry)) (Entry, error) { + key := Key(name, version) + for attempt := 0; attempt < 16; attempt++ { + raw, rev, found, err := v.Store.Get(ctx, key) + if err != nil { + return Entry{}, fmt.Errorf("%s: %w", key, err) + } + if !found { + return Entry{}, fmt.Errorf("%s@%s is not numbered by the builder", name, version) + } + e, err := Decode(raw) + if err != nil { + return Entry{}, fmt.Errorf("%s holds what is not a version: %w", key, err) + } + change(&e) + e.At = v.now() + out, err := Encode(e) + if err != nil { + return Entry{}, err + } + if _, err := v.Store.Update(ctx, key, out, rev); errors.Is(err, ErrConflict) { + continue + } else if err != nil { + return Entry{}, fmt.Errorf("%s: %w", key, err) + } + return e, nil + } + return Entry{}, fmt.Errorf("%s@%s kept changing under 16 marks", name, version) +} + +// Recorded is the holder's question through builder_versions (ADR 0605 §4): whether name, version and digest are a +// version the builder numbered. It answers the entry when they are, and why not otherwise. +func (v Versions) Recorded(ctx context.Context, name, version, digest string) (Entry, bool, string, error) { + raw, _, found, err := v.Store.Get(ctx, Key(name, version)) + if err != nil { + return Entry{}, false, "", err + } + if !found { + return Entry{}, false, fmt.Sprintf("%s@%s is not a version the builder numbered", name, version), nil + } + e, err := Decode(raw) + if err != nil { + return Entry{}, false, "", err + } + switch { + case e.State == Removed: + return e, false, fmt.Sprintf("%s@%s was removed (%s)", name, version, e.Reason), nil + case e.Digest != "" && digest != "" && e.Digest != digest: + return e, false, fmt.Sprintf("%s@%s is recorded with digest %s, not %s", name, version, e.Digest, digest), nil + } + return e, true, "", nil +} + +func (v Versions) now() time.Time { + if v.Now != nil { + return v.Now().UTC() + } + return time.Now().UTC() +} diff --git a/modules/mesh-delivery-builder/internal/versions/tools.go b/modules/mesh-delivery-builder/internal/versions/tools.go new file mode 100644 index 00000000..ece1d14b --- /dev/null +++ b/modules/mesh-delivery-builder/internal/versions/tools.go @@ -0,0 +1,123 @@ +package versions + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strings" +) + +// The builder's tools of its versions (design 51 *Events of the builder*, *Tools of the builder*). +const ( + VersionsTool = "builder_versions" + PublishAgainTool = "builder_publish_again" + FloorAgainTool = "builder_floor_again" + TagTool = "builder_tag" + TagAsksTool = "builder_tag_asks" + UntagTool = "builder_untag" + OldFormTool = "builder_old_form" + // NumberedEvent and OldFormClosedEvent are the builder's own family `version`, kept seven days, apart from its builds' + // ends, so a waiter on a build's end does not wake on them. + NumberedEvent = "version.numbered" + OldFormClosedEvent = "version.old-form-closed" +) + +// NumberedBody is the body of NumberedEvent: what a build module consumes to publish a build's version with the held digest, +// which stores nothing (design 51 *The floor*). +type NumberedBody struct { + Name string `json:"name"` + Kind string `json:"kind"` + Version string `json:"version"` + Tag string `json:"tag,omitempty"` + Module string `json:"module"` + Fingerprint string `json:"fingerprint"` + Commit string `json:"commit,omitempty"` + Digest string `json:"digest,omitempty"` + Seat string `json:"seat,omitempty"` + // Head is the head the version answers, when it answers a tag ask. + Head string `json:"head,omitempty"` +} + +// NumberedOf is the event of an entry. +func NumberedOf(e Entry, head string) NumberedBody { + return NumberedBody{Name: e.Name, Kind: e.Kind, Version: e.Version, Tag: e.Tag, Module: e.Module, Fingerprint: e.Fingerprint, + Commit: e.Commit, Digest: e.Digest, Seat: e.Seat, Head: head} +} + +// Reader reads the state as the tools bundle may: every key, and one. +type Reader struct { + Keys func() ([]string, error) + Get func(key string) (json.RawMessage, bool, error) +} + +// readStore is a Store over a Reader that refuses every write: the tools bundle only reads versions. +type readStore struct{ r Reader } + +var errReadOnly = errors.New("the versions are written by the builder alone") + +func (s readStore) Get(_ context.Context, key string) ([]byte, uint64, bool, error) { + v, ok, err := s.r.Get(key) + return v, 0, ok, err +} +func (readStore) Create(context.Context, string, []byte) (uint64, error) { return 0, errReadOnly } +func (readStore) Update(context.Context, string, []byte, uint64) (uint64, error) { + return 0, errReadOnly +} +func (s readStore) KeysOf(_ context.Context, filter string) ([]string, error) { + all, err := s.r.Keys() + if err != nil { + return nil, err + } + prefix := strings.TrimSuffix(filter, "*") + var out []string + for _, k := range all { + if strings.HasPrefix(k, prefix) && !strings.Contains(k[len(prefix):], ".") { + out = append(out, k) + } + } + return out, nil +} + +// VersionsAnswer is what builder_versions answers. +type VersionsAnswer struct { + Name string `json:"name"` + Versions []Entry `json:"versions"` + // Asked, Recorded and Why answer a holder's question for one version and digest (ADR 0605 §4). + Asked string `json:"asked,omitempty"` + Recorded *bool `json:"recorded,omitempty"` + Why string `json:"why,omitempty"` +} + +// AnswerVersions is builder_versions: a name's versions in order, in one call; with a version (and a digest), whether +// the builder numbered it with that digest, which the holder asks before it stores (ADR 0605 §4). +func AnswerVersions(ctx context.Context, r Reader, name, version, digest string) (VersionsAnswer, error) { + name = strings.TrimSpace(name) + if name == "" { + return VersionsAnswer{}, errors.New("name is required: the package, image, bundle or archive name") + } + v := Versions{Store: readStore{r}} + all, err := v.Of(ctx, name) + if err != nil { + return VersionsAnswer{}, err + } + out := VersionsAnswer{Name: name, Versions: all} + if out.Versions == nil { + out.Versions = []Entry{} + } + if version = strings.TrimSpace(version); version != "" { + _, ok, why, err := v.Recorded(ctx, name, version, strings.TrimSpace(digest)) + if err != nil { + return VersionsAnswer{}, fmt.Errorf("%s@%s: %w", name, version, err) + } + out.Asked, out.Recorded, out.Why = version, &ok, why + } + return out, nil +} + +// VersionsDescription is builder_versions' description. +const VersionsDescription = "A name's versions as the builder numbered them (its state versions, design 51): every trunk and tagged " + + "version of a package, image, bundle or archive in order, each with its kind, series, fingerprint, commit, source revision, digest, " + + "seat, state (numbered, published, refused with its reason, removed with its time, reason and the event that removed it) and when. " + + "With version, and digest, it also answers recorded: whether the builder numbered that version with that digest, and why not: the " + + "question a registry seat's holder asks before it stores a publish (ADR 0605 §4). Read-only." diff --git a/modules/mesh-delivery-builder/internal/versions/versions.go b/modules/mesh-delivery-builder/internal/versions/versions.go new file mode 100644 index 00000000..4fd501a9 --- /dev/null +++ b/modules/mesh-delivery-builder/internal/versions/versions.go @@ -0,0 +1,408 @@ +// Package versions is the builder's state `versions` (novox/hq design 51 *The versions*, ADR 0606): one key per name, +// tag and version, written only by the builder under compare-and-set, read by others only through the builder's tool +// builder_versions or its events (ADR 0596 §2). It holds what each version is (its fingerprint, commit, source revision +// and digest) and where it stands: numbered, published, refused with its reason, or removed with its time, reason and the +// event that removed it. +// +// The state is declared in the builder's old `state` section (ADR 0619: the builder is one of the foundation's nine +// parts), history one; its cap is 64 MiB, the most the template state allows (ADR 0595 §3), and past 80 % of it the +// builder writes a finding, cleared below it (Fill). +package versions + +import ( + "encoding/json" + "errors" + "fmt" + "sort" + "strconv" + "strings" + "time" +) + +// State is the builder state's name; Bucket the bucket the bus keeps it in. +const ( + State = "versions" + Bucket = "mesh-delivery-builder_versions" + // Cap is the bucket's cap in bytes (64 MiB); FullPercent the share of it past which the builder writes its finding. + Cap = 64 << 20 + FullPercent = 80 + // DefaultSeries is the series of a name whose manifest declares none (design 32). + DefaultSeries = "0.1" +) + +// The states a version is in. +const ( + Numbered = "numbered" + Published = "published" + Refused = "refused" + Removed = "removed" +) + +// The kinds the builder numbers: a package per source revision, the rest per fingerprint (design 51). +const ( + KindPackage = "package" + KindImage = "image" + KindBundle = "bundle" + KindArchive = "archive" +) + +// Entry is one key of the state: a version of a name, tagged or not. +type Entry struct { + Name string `json:"name"` + Kind string `json:"kind"` + Series string `json:"series"` + Version string `json:"version"` + // Tag is the cleaned tag of a tagged version (`feat-x`); empty for a trunk version. + Tag string `json:"tag,omitempty"` + Module string `json:"module"` + Fingerprint string `json:"fingerprint"` + Commit string `json:"commit,omitempty"` + // Revision is the source revision a package's version is numbered for (one version per revision, design 51). + Revision string `json:"revision,omitempty"` + // Digest is the digest the build recorded (an image's manifest, a blob's, Go's h1: hash). + Digest string `json:"digest,omitempty"` + // Seat is the registry seat it is published to. + Seat string `json:"seat,omitempty"` + State string `json:"state"` + At time.Time `json:"at"` + // Reason is why it was refused or removed: the holder's reason, a branch's end, retention, the sweep. + Reason string `json:"reason,omitempty"` + // RemovedBy is the id of the event that removed it. + RemovedBy string `json:"removed_by,omitempty"` + // Merge is the forge bridge's sequence of the merge that numbered it, for merge order. + Merge uint64 `json:"merge,omitempty"` +} + +// Valid says what is wrong with an entry, or nil. +func (e Entry) Valid() error { + switch { + case strings.TrimSpace(e.Name) == "": + return errors.New("a version names no name") + case e.Version == "": + return fmt.Errorf("%s: a version with no version", e.Name) + } + switch e.State { + case Numbered, Published, Refused, Removed: + default: + return fmt.Errorf("%s@%s: unknown state %q", e.Name, e.Version, e.State) + } + switch e.Kind { + case KindPackage, KindImage, KindBundle, KindArchive: + default: + return fmt.Errorf("%s@%s: unknown kind %q", e.Name, e.Version, e.Kind) + } + if _, _, err := SplitSeries(e.Series); err != nil { + return fmt.Errorf("%s@%s: %w", e.Name, e.Version, err) + } + return nil +} + +// Encode and Decode are the value's bytes. +func Encode(e Entry) ([]byte, error) { + if err := e.Valid(); err != nil { + return nil, err + } + return json.Marshal(e) +} + +func Decode(raw []byte) (Entry, error) { + var e Entry + if err := json.Unmarshal(raw, &e); err != nil { + return Entry{}, err + } + return e, e.Valid() +} + +// Key is the state's key of a name's version: two tokens, the name and the version each escaped (EscapeToken), so a +// name's versions are the keys under NameFilter and no name's keys fall under another's. +func Key(name, version string) string { return EscapeToken(name) + "." + EscapeToken(version) } + +// NameFilter is the key filter of every version of name. +func NameFilter(name string) string { return EscapeToken(name) + ".*" } + +// EscapeToken makes s one token of a bus key: letters, digits, '-', '_' and '/' stay; '=' and every other byte become +// '=' and two hex digits ('.' is "=2E", '@' "=40"). Two different strings never escape alike. +func EscapeToken(s string) string { + var b strings.Builder + for i := 0; i < len(s); i++ { + c := s[i] + switch { + case c >= 'a' && c <= 'z', c >= 'A' && c <= 'Z', c >= '0' && c <= '9', c == '-', c == '_', c == '/': + b.WriteByte(c) + default: + fmt.Fprintf(&b, "=%02X", c) + } + } + return b.String() +} + +// SplitSeries reads a series `.`. +func SplitSeries(series string) (major, minor int, err error) { + a, b, ok := strings.Cut(series, ".") + if !ok { + return 0, 0, fmt.Errorf("series %q is not .", series) + } + if major, err = strconv.Atoi(a); err != nil || major < 0 { + return 0, 0, fmt.Errorf("series %q is not .", series) + } + if minor, err = strconv.Atoi(b); err != nil || minor < 0 { + return 0, 0, fmt.Errorf("series %q is not .", series) + } + return major, minor, nil +} + +// Patch is the count of a trunk version of series (`0.1.14` → 14), read leniently: a leading 'v' and a path prefix +// (`go/v0.1.13`) are dropped; ok is false for a version of another series or a tagged one. +func Patch(series, version string) (int, bool) { + if i := strings.LastIndex(version, "/"); i >= 0 { + version = version[i+1:] + } + version = strings.TrimPrefix(version, "v") + if strings.ContainsAny(version, "-+") { + return 0, false + } + rest, ok := strings.CutPrefix(version, series+".") + if !ok { + return 0, false + } + n, err := strconv.Atoi(rest) + if err != nil || n < 0 { + return 0, false + } + return n, true +} + +// Floor is what the three sources answered for a name (design 51 *The floor*, ADR 0606 §3): the seat's versions, the +// package's git tags (the forge bridge's bridge_tags) and the pseudo-versions consumers require (the indexer's +// indexer_requirements). A source not answered is listed in Silent. +type Floor struct { + Seat []string `json:"seat,omitempty"` + Tags []string `json:"tags,omitempty"` + Requirements []string `json:"requirements,omitempty"` + // Silent names the sources that did not answer: "seat", "bridge_tags", "indexer_requirements". + Silent []string `json:"silent,omitempty"` +} + +// The floor's sources, as Floor.Silent names them. +const ( + SourceSeat = "seat" + SourceTags = "bridge_tags" + SourceRequirements = "indexer_requirements" +) + +// Known is whether every source answered. +func (f Floor) Known() bool { return len(f.Silent) == 0 } + +// Highest is the highest count of series the floor's three sources name, -1 when none names one. A pseudo-version +// (`v0.1.14-0.2026…-abc`) counts as the version below it was cut from, so its base (`0.1.13`) counts: Go orders it +// before `0.1.14`. +func (f Floor) Highest(series string) int { + high := -1 + for _, list := range [][]string{f.Seat, f.Tags, f.Requirements} { + for _, v := range list { + n, ok := Patch(series, v) + if !ok { + n, ok = pseudoBase(series, v) + } + if ok && n > high { + high = n + } + } + } + return high +} + +// pseudoBase is the count a Go pseudo-version stands on: `v0.1.14-0.