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 1a1292d4..c40f7be1 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 @@ -27,9 +27,78 @@ 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, vs versions.Reader) []stdio.Tool { +// tags writes a tag-asks value and waits for the builder's answer; ta reads tag-asks. +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, + tags queueAsker, ta versions.Reader) []stdio.Tool { str := func(args map[string]any, k string) string { s, _ := args[k].(string); return strings.TrimSpace(s) } + // handAct writes a reasks value of a versions hand act and answers the builder's answer, or its refusal as an error. + handAct := func(key string, value []byte, err error) (any, error) { + if err != nil { + return nil, err + } + raw, err := buildAgain(key, value) + if err != nil { + return nil, err + } + a, err := build.DecodeReAsk(raw) + if err != nil { + return nil, err + } + if a.Refused { + return nil, errors.New(a.Answer) + } + return map[string]any{"key": key, "name": a.Name, "do": a.Do, "version": a.Version, "answer": a.Answer}, nil + } + tagAct := func(key string, value []byte, err error) (any, error) { + if err != nil { + return nil, err + } + raw, err := tags(key, value) + if err != nil { + return nil, err + } + return versions.DecodeTagAsk(raw) + } + nameFP := map[string]any{"type": "object", "required": []string{"name", "fp", "why"}, "properties": map[string]any{ + "name": map[string]any{"type": "string", "description": "the numbered name, /, as builder_versions lists it"}, + "fp": map[string]any{"type": "string", "description": "the fingerprint, src2:"}, + "why": map[string]any{"type": "string", "description": "why, in plain words"}}} + branch := func(head bool) map[string]any { + req := []string{"repository", "branch", "why"} + props := map[string]any{ + "repository": map[string]any{"type": "string", "description": "/"}, + "branch": map[string]any{"type": "string", "description": "the branch; its cleaned name is the tag"}, + "why": map[string]any{"type": "string", "description": "why, in plain words"}} + if head { + req = append(req, "head") + props["head"] = map[string]any{"type": "string", "description": "the commit the tagged version is built from"} + } + return map[string]any{"type": "object", "required": req, "properties": props} + } return []stdio.Tool{ + {Name: versions.PublishAgainTool, Description: versions.PublishAgainDescription, Input: nameFP, + Run: func(args map[string]any) (any, error) { + return handAct(build.NewPublishAgain(str(args, "name"), str(args, "fp"), str(args, "why"), stdio.CallerOf(args), now())) + }}, + {Name: versions.FloorAgainTool, Description: versions.FloorAgainDescription, + Input: map[string]any{"type": "object", "required": []string{"name", "why"}, "properties": map[string]any{ + "name": map[string]any{"type": "string", "description": "the numbered name, /"}, + "why": map[string]any{"type": "string", "description": "why, in plain words"}}}, + Run: func(args map[string]any) (any, error) { + return handAct(build.NewFloorAgain(str(args, "name"), str(args, "why"), stdio.CallerOf(args), now())) + }}, + {Name: versions.TagTool, Description: versions.TagDescription, Input: branch(true), + Run: func(args map[string]any) (any, error) { + return tagAct(versions.NewTagAsk(str(args, "repository"), str(args, "branch"), str(args, "head"), str(args, "why"), stdio.CallerOf(args), now())) + }}, + {Name: versions.UntagTool, Description: versions.UntagDescription, Input: branch(false), + Run: func(args map[string]any) (any, error) { + return tagAct(versions.NewUntag(str(args, "repository"), str(args, "branch"), str(args, "why"), stdio.CallerOf(args), now())) + }}, + {Name: versions.TagAsksTool, Description: versions.TagAsksDescription, + Input: map[string]any{"type": "object", "properties": map[string]any{ + "repository": map[string]any{"type": "string", "description": "only this repository's asks, /"}}}, + Run: func(args map[string]any) (any, error) { return versions.AnswerTagAsks(ta, str(args, "repository")) }}, {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"}, @@ -281,7 +350,19 @@ func main() { } 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 { + ts := stdio.State(versions.TagAsksState) + tags := askKept(ts, versions.TagAsksState, versions.TagAsksBucket, "the ask", 25*time.Second, func(v []byte) bool { + a, err := versions.DecodeTagAsk(v) + return err == nil && !a.Answered.IsZero() + }) + tr := versions.Reader{Keys: ts.Keys, Get: func(key string) (json.RawMessage, bool, error) { + e, err := ts.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, tags, tr)); err != nil { fmt.Fprintf(os.Stderr, "[mesh-delivery-builder-tools] %v\n", err) os.Exit(1) } diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go index 7bcbe48e..439037ce 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go @@ -136,11 +136,23 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } // A state bound later numbers the builds looked at without it. versionsKV.Ready(func(jetstream.KeyValue) { b.Reconsider(ctx) }) + // The versions' hand acts (builder_publish_again, builder_floor_again) come through reasks, as builder_build_again. + acts := versions.Acts{Numberer: numberer, Done: doneVia(b), Reconsider: b.Reconsider, Remove: removeVia(calls.Call), + BuildAgain: func(ctx context.Context, module, fp, why, by string) (string, bool) { + r := b.BuildAgain(ctx, module, fp, why, by, buildAgainBound) + if r.Refused != "" { + return r.Refused, true + } + if r.Failed != "" { + return "built again, and it failed before its asks were put: " + r.Failed, false + } + return fmt.Sprintf("built again as revision %d", r.Revision), false + }} // A candidate's preview is merged onto the trunk head the indexer previewed it onto: source.main at the candidate's base // revision (the gate's source state). The source's revisions are watched with their history (a direct get of one revision // is not what a read grants): a revision heard after the candidate that names it builds that candidate then. Where it // cannot be read the trunk's keys are built and the candidates wait (issue 1945). - errs := make(chan error, 6) + errs := make(chan error, 8) trunks := &trunkSource{ctx: ctx, js: js, heads: &sourceHeads{}, moved: func(ctx context.Context) { // A source revision heard is the forge's merge: it wakes a floor whose bridge_tags was silent (design 51 *The floor*). (&floorWake{b: b, waits: floorWaits, logf: logger.Printf}).wake(ctx, versions.WakeForge, "", "") @@ -263,13 +275,23 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // every other input. The builder asks again by itself without it. // A state not on the bus yet at this start is watched once it appears (issue 1790). process.AwaitBucket(ctx, js, conn, build.ReAsksBucket, "builder_ask_again is not heard; an ask that can no longer be answered is still asked again by itself", logger.Printf).Ready(func(kv jetstream.KeyValue) { - eachReAsk := func(e jetstream.KeyValueEntry) { reAsked(ctx, b, kv, e, logger.Printf) } + eachReAsk := func(e jetstream.KeyValueEntry) { reAsked(ctx, b, acts, kv, e, logger.Printf) } if rw, err := settle(ctx, kv, eachReAsk); err != nil { process.NotWatched(logger.Printf, build.ReAsksBucket, err, "builder_ask_again is not heard until the builder starts again") } else { go func() { errs <- follow(ctx, rw, eachReAsk) }() } }) + // A tag ask or a branch's removal by hand (builder_tag, builder_untag) is written to tag-asks and answered there + // (design 51 *Tagged versions*); the checker's tag-wanted and the branch's end join it on their consumers (DLV-B6). + process.AwaitBucket(ctx, js, conn, versions.TagAsksBucket, "builder_tag and builder_untag are not answered", logger.Printf).Ready(func(kv jetstream.KeyValue) { + each := func(e jetstream.KeyValueEntry) { tagAsked(ctx, acts, kv, e, logger.Printf) } + if tw, err := settle(ctx, kv, each); err != nil { + process.NotWatched(logger.Printf, versions.TagAsksBucket, err, "builder_tag and builder_untag are not answered until the builder starts again") + } else { + go func() { errs <- follow(ctx, tw, each) }() + } + }) // A person reads a build seat's stream with builder_seat_queue, which writes the queues state (issue 1945 note 62): the // builder alone may read the stream, so it answers there. process.AwaitBucket(ctx, js, conn, build.QueuesBucket, "builder_seat_queue is not answered", logger.Printf).Ready(func(kv jetstream.KeyValue) { diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reasks.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reasks.go index 024528d1..61b1c6a5 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reasks.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reasks.go @@ -15,6 +15,7 @@ import ( "github.com/nats-io/nats.go/jetstream" "mesh-delivery-builder/internal/build" + "mesh-delivery-builder/internal/versions" ) // askAgainer is the part of the builder a reasks value asks through. @@ -31,7 +32,7 @@ const buildAgainBound = 15 * time.Second // reAsked answers one value of the reasks state (the tool builder_ask_again, issue 1945): a value not yet answered asks the // key's waiting lanes again and is written back, by compare-and-set at its revision, with what the builder did. A value // answered already (this builder's own write) or deleted is left. -func reAsked(ctx context.Context, b askAgainer, kv jetstream.KeyValue, e jetstream.KeyValueEntry, logf func(string, ...any)) { +func reAsked(ctx context.Context, b askAgainer, acts versions.Acts, kv jetstream.KeyValue, e jetstream.KeyValueEntry, logf func(string, ...any)) { if e.Operation() != jetstream.KeyValuePut { return } @@ -52,6 +53,11 @@ func reAsked(ctx context.Context, b askAgainer, kv jetstream.KeyValue, e jetstre announcedAgain(ctx, b, kv, e, r, logf) return } + if r.Mode == build.ModePublishAgain || r.Mode == build.ModeFloorAgain { + // A version's hand act (design 51 *Tools of the builder*) may build again, so it is answered beside the watch. + go versionActed(ctx, acts, kv, e, r, logf) + return + } asked := b.AskAgain(e.Key(), r.Why) if len(asked) == 0 { r.Answer = "no ask of this key waits in this builder now (it is not building, or its build has not reached its ask yet): nothing was asked; builder_stuck shows where it stands" @@ -149,3 +155,27 @@ func appendLog(ctx context.Context, js jetstream.JetStream, cred process.Account } return nil } + +// versionActed answers a reasks value of builder_publish_again or builder_floor_again, a hand act said in the log. +func versionActed(ctx context.Context, acts versions.Acts, kv jetstream.KeyValue, e jetstream.KeyValueEntry, r build.ReAsk, logf func(string, ...any)) { + var err error + if r.Mode == build.ModePublishAgain { + var ag versions.Again + ag, r.Answer, err = acts.PublishAgain(ctx, r.Name, r.FP, r.Why, r.By, r.At) + r.Do, r.Version = ag.Do, ag.Entry.Version + } else { + r.Answer, err = acts.FloorAgain(ctx, r.Name) + } + if err != nil { + r.Answer, r.Refused = err.Error(), true + } + r.Answered = time.Now().UTC() + logf("%s: %s of %s by %s (%s): %s", e.Key(), r.Mode, r.Name, r.By, r.Why, r.Answer) + data, err := json.Marshal(r) + if err != nil { + return + } + if _, err := kv.Update(ctx, e.Key(), data, e.Revision()); err != nil && !errors.Is(err, jetstream.ErrKeyExists) && !errors.Is(err, jetstream.ErrKeyRevisionMismatch) { + logf("%s.%s: the answer could not be written: %v", build.ReAsksState, e.Key(), err) + } +} diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/versions.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/versions.go index 5d3b7706..1726c4e3 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/versions.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/versions.go @@ -153,7 +153,11 @@ func floorVia(call toolCaller) versions.FloorSources { // emitNumbered says the builder's event version.numbered (design 51 *Events of the builder*), on its own family apart // from its builds' ends; the name and version are the bus's dedup id. func emitNumbered(ctx context.Context, js jetstream.JetStream, cred process.Account, b versions.NumberedBody) error { - if err := emitEvent(ctx, js, cred, versions.NumberedEvent, "numbered-"+versions.Key(b.Name, b.Version), b); err != nil { + id := "numbered-" + versions.Key(b.Name, b.Version) + if b.Again != "" { + id += "-again-" + b.Again + } + if err := emitEvent(ctx, js, cred, versions.NumberedEvent, id, b); err != nil { return fmt.Errorf("saying %s for %s@%s: %w", versions.NumberedEvent, b.Name, b.Version, err) } return nil @@ -236,3 +240,55 @@ func (f *floorWake) wake(ctx context.Context, kind, seat, module string) { func conditionModules(id string) []string { return strings.FieldsFunc(id, func(r rune) bool { return r == '/' || r == '.' || r == ':' }) } + +// doneVia says whether a build key is done, read from the builder's own builds state: the build's file is held. +func doneVia(b *build.Builder) func(ctx context.Context, module, fp string) (bool, error) { + return func(ctx context.Context, module, fp string) (bool, error) { + key, err := delivery.BuilderBuildKey(module, fp) + if err != nil { + return false, err + } + raw, _, found, err := b.Claimer.Store.Get(ctx, key) + if err != nil || !found { + return false, err + } + cur, err := delivery.DecodeBuilderBuild(raw) + if err != nil { + return false, err + } + return cur.State == delivery.BuildDone, nil + } +} + +// removeVia calls a registry seat's remove for one tagged version, as the builder's own account (design 51 *Tagged +// versions*): the reason tag-gone; the seat says removed on its events. +func removeVia(call toolCaller) func(ctx context.Context, seat string, e versions.Entry) error { + return func(_ context.Context, seat string, e versions.Entry) error { + _, err := call("seat:"+seat+".remove", map[string]any{"name": e.Name, "version": e.Version, "tag": e.Tag, "reason": "tag-gone"}) + return err + } +} + +// tagAsked answers an unanswered value of tag-asks by compare-and-set at its revision, said in the log. +func tagAsked(ctx context.Context, acts versions.Acts, kv jetstream.KeyValue, e jetstream.KeyValueEntry, logf func(string, ...any)) { + if e.Operation() != jetstream.KeyValuePut { + return + } + t, err := versions.DecodeTagAsk(e.Value()) + if err != nil { + logf("%s.%s holds what is not a tag ask, left: %v", versions.TagAsksState, e.Key(), err) + return + } + if !t.Answered.IsZero() { + return + } + t = acts.TagAsked(ctx, t, time.Now()) + logf("%s: %s of %s %s by %s (%s): %s", e.Key(), map[bool]string{true: "untag", false: "tag ask"}[t.Untag], t.Repository, t.Branch, t.By, t.Why, t.Answer) + data, err := json.Marshal(t) + if err != nil { + return + } + if _, err := kv.Update(ctx, e.Key(), data, e.Revision()); err != nil && !errors.Is(err, jetstream.ErrKeyExists) && !errors.Is(err, jetstream.ErrKeyRevisionMismatch) { + logf("%s.%s: the answer could not be written: %v", versions.TagAsksState, e.Key(), err) + } +} diff --git a/modules/mesh-delivery-builder/internal/build/stuck.go b/modules/mesh-delivery-builder/internal/build/stuck.go index d6a713c0..167db4b6 100644 --- a/modules/mesh-delivery-builder/internal/build/stuck.go +++ b/modules/mesh-delivery-builder/internal/build/stuck.go @@ -1,6 +1,8 @@ package build import ( + "mesh-delivery-builder/internal/versions" + "encoding/json" "errors" "fmt" @@ -54,6 +56,49 @@ type ReAsk struct { Failed string `json:"failed,omitempty"` Revision uint64 `json:"revision,omitempty"` Asks map[string]string `json:"asks,omitempty"` + // Name is the numbered name (`/`) of a ModePublishAgain or ModeFloorAgain value; Do and Version the + // builder's answer: what it did (versions.Again*) and the version published. + Name string `json:"name,omitempty"` + Do string `json:"do,omitempty"` + Version string `json:"version,omitempty"` +} + +// ModePublishAgain is the Mode of a reasks value of builder_publish_again (design 51 *Tools of the builder*). +const ModePublishAgain = "publish-again" + +// ModeFloorAgain is the Mode of a reasks value of builder_floor_again (design 51 *The floor*). +const ModeFloorAgain = "floor-again" + +// NewPublishAgain is the value builder_publish_again writes for a name's fingerprint: the key of the name's module and +// fingerprint, and the value. +func NewPublishAgain(name, fp, why, by string, at time.Time) (string, []byte, error) { + module, artifact, ok := strings.Cut(strings.TrimSpace(name), "/") + if !ok || module == "" || artifact == "" { + return "", nil, fmt.Errorf("name %q is not a numbered name, /, as builder_versions lists it", name) + } + key, err := delivery.BuilderBuildKey(module, fp) + if err != nil { + return "", nil, err + } + if strings.TrimSpace(why) == "" { + return "", nil, errors.New("say why it is published again (why)") + } + data, err := json.Marshal(ReAsk{V: 1, Why: strings.TrimSpace(why), By: by, At: at.UTC(), Mode: ModePublishAgain, Module: module, FP: fp, Name: strings.TrimSpace(name)}) + return key, data, err +} + +// NewFloorAgain is the value builder_floor_again writes for a name: the key `floor.`, never a build key, +// and the value. +func NewFloorAgain(name, why, by string, at time.Time) (string, []byte, error) { + name = strings.TrimSpace(name) + if module, artifact, ok := strings.Cut(name, "/"); !ok || module == "" || artifact == "" { + return "", nil, fmt.Errorf("name %q is not a numbered name, /, as builder_versions lists it", name) + } + if strings.TrimSpace(why) == "" { + return "", nil, errors.New("say why the floor is asked again (why)") + } + data, err := json.Marshal(ReAsk{V: 1, Why: strings.TrimSpace(why), By: by, At: at.UTC(), Mode: ModeFloorAgain, Name: name}) + return "floor." + versions.EscapeToken(name), data, err } // ModeBuild is the Mode of a reasks value that asks the key to be built again, not its waiting asks to be asked again. diff --git a/modules/mesh-delivery-builder/internal/versions/again.go b/modules/mesh-delivery-builder/internal/versions/again.go new file mode 100644 index 00000000..9de9e48b --- /dev/null +++ b/modules/mesh-delivery-builder/internal/versions/again.go @@ -0,0 +1,315 @@ +package versions + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "sort" + "strings" + "time" +) + +// What builder_publish_again does with one fingerprint of a name (design 51 *Tools of the builder*), after a holder failed. +const ( + // AgainPublish says version.numbered again for the recorded version: the build module publishes the held digest. + AgainPublish = "publish" + // AgainBuild builds the fingerprint again under the same number: the build's file is gone and no published was said + // for the number, so the new digest replaces the recorded one (Numberer.Built). + AgainBuild = "build" + // AgainRenumber takes a new number: a published was said for the number, so other bytes never go under it. + AgainRenumber = "renumber" +) + +// Again is what builder_publish_again does, with the entry it acts on and why. +type Again struct { + Do string `json:"do"` + Entry Entry `json:"entry"` + Why string `json:"why"` +} + +// PublishAgain decides builder_publish_again for fingerprint among a name's entries; built says whether the builder still +// holds the build's file (its done key). It refuses a fingerprint not numbered, and a version refused for other bytes +// than the ones recorded under it. +func PublishAgain(entries []Entry, fingerprint string, built bool) (Again, error) { + fingerprint = strings.TrimSpace(fingerprint) + if fingerprint == "" { + return Again{}, errors.New("fp is required: the fingerprint whose version is published again") + } + for _, e := range latestFirst(entries) { + if e.Tag != "" || e.Fingerprint != fingerprint { + continue + } + switch { + case e.State == Refused && e.RefusedDigest != "" && e.Digest != "" && e.RefusedDigest != e.Digest: + return Again{}, fmt.Errorf("%s@%s was refused for other bytes (%s, recorded %s): it is not published again; a new build takes a new number", + e.Name, e.Version, e.RefusedDigest, e.Digest) + case e.State == Published || e.State == Removed: + return Again{Do: AgainRenumber, Entry: e, Why: fmt.Sprintf("%s@%s was %s, so the fingerprint takes a new number", e.Name, e.Version, e.State)}, nil + case !built: + return Again{Do: AgainBuild, Entry: e, Why: fmt.Sprintf("the build's file is gone and no published was said for %s@%s: built again under the same number", e.Name, e.Version)}, nil + } + return Again{Do: AgainPublish, Entry: e, Why: fmt.Sprintf("%s@%s (%s) is said numbered again, so its build module publishes the held digest", e.Name, e.Version, e.State)}, nil + } + return Again{}, fmt.Errorf("fingerprint %s is not numbered for this name: builder_versions lists what is", fingerprint) +} + +// The builder's state of tag asks (design 51 *Tagged versions*): one key per repository and branch, removed when the +// branch is merged or deleted or its pull request closed. +const ( + TagAsksState = "tag-asks" + TagAsksBucket = "mesh-delivery-builder_tag-asks" +) + +// The states of a tag ask. +const ( + TagAsked = "asked" + TagNumbered = "numbered" + TagRemoving = "removing" + TagRemoved = "removed" +) + +// TagAsk is one value of tag-asks: an ask of a tagged version for a branch's head, from the checker's tag-wanted or by +// hand (builder_tag), or by hand the removal of the branch's tagged versions (builder_untag, Untag). +type TagAsk struct { + V int `json:"v"` + Repository string `json:"repository"` + Branch string `json:"branch"` + Head string `json:"head,omitempty"` + Tag string `json:"tag"` + By string `json:"by,omitempty"` + Why string `json:"why,omitempty"` + At time.Time `json:"at"` + Untag bool `json:"untag,omitempty"` + // State, Versions (name to version), Left (what a removal still waits on, name@version to its seat) and Answer are + // the builder's. + State string `json:"state,omitempty"` + Versions map[string]string `json:"versions,omitempty"` + Left map[string]string `json:"left,omitempty"` + Answer string `json:"answer,omitempty"` + Answered time.Time `json:"answered,omitzero"` +} + +// TagAskKey is the key of a repository's branch in tag-asks. +func TagAskKey(repository, branch string) string { + return EscapeToken(repository) + "." + EscapeToken(branch) +} + +// NewTagAsk is the value builder_tag writes: its key and the value. The tag is the branch's, cleaned (CleanTag). +func NewTagAsk(repository, branch, head, why, by string, at time.Time) (string, []byte, error) { + return newTagAsk(repository, branch, head, why, by, at, false) +} + +// NewUntag is the value builder_untag writes: its key and the value. +func NewUntag(repository, branch, why, by string, at time.Time) (string, []byte, error) { + return newTagAsk(repository, branch, "", why, by, at, true) +} + +func newTagAsk(repository, branch, head, why, by string, at time.Time, untag bool) (string, []byte, error) { + repository, branch, head, why = strings.TrimSpace(repository), strings.TrimSpace(branch), strings.TrimSpace(head), strings.TrimSpace(why) + switch { + case repository == "" || !strings.Contains(repository, "/"): + return "", nil, errors.New("repository is required: /") + case branch == "": + return "", nil, errors.New("branch is required") + case !untag && head == "": + return "", nil, errors.New("head is required: the commit the tagged version is built from") + case why == "": + return "", nil, errors.New("say why (why): a hand act is recorded") + } + data, err := json.Marshal(TagAsk{V: 1, Repository: repository, Branch: branch, Head: head, Tag: CleanTag(branch), By: by, Why: why, + At: at.UTC(), Untag: untag, State: TagAsked}) + return TagAskKey(repository, branch), data, err +} + +// DecodeTagAsk reads a tag-asks value. +func DecodeTagAsk(data []byte) (TagAsk, error) { + var a TagAsk + if err := json.Unmarshal(data, &a); err != nil { + return a, err + } + if a.V != 1 { + return a, fmt.Errorf("a tag-asks value of version %d, not 1", a.V) + } + return a, nil +} + +// TagAsksAnswer is what builder_tag_asks answers: the open asks, oldest first. +type TagAsksAnswer struct { + Asks []TagAsk `json:"asks"` +} + +// AnswerTagAsks is builder_tag_asks over the state read, only repository's when it is given. +func AnswerTagAsks(r Reader, repository string) (TagAsksAnswer, error) { + keys, err := r.Keys() + if err != nil { + return TagAsksAnswer{}, fmt.Errorf("%s cannot be listed: %w", TagAsksState, err) + } + out := TagAsksAnswer{Asks: []TagAsk{}} + for _, k := range keys { + raw, ok, err := r.Get(k) + if err != nil { + return TagAsksAnswer{}, fmt.Errorf("%s.%s: %w", TagAsksState, k, err) + } + if !ok { + continue + } + a, err := DecodeTagAsk(raw) + if err != nil { + return TagAsksAnswer{}, fmt.Errorf("%s.%s holds what is not a tag ask: %w", TagAsksState, k, err) + } + if repository == "" || a.Repository == repository { + out.Asks = append(out.Asks, a) + } + } + sort.Slice(out.Asks, func(i, j int) bool { return out.Asks[i].At.Before(out.Asks[j].At) }) + return out, nil +} + +// Untagged are repository's tagged versions of tag that are not removed (a group's branch spans repositories, each +// ending apart): what a branch's end, or builder_untag, removes through +// each one's seat (design 51 *Tagged versions*). A version with no tag is never among them. +func Untagged(entries []Entry, repository, tag string) []Entry { + tag = CleanTag(tag) + var out []Entry + for _, e := range entries { + if e.Tag != "" && e.Repository == repository && CleanTag(e.Tag) == tag && e.State != Removed { + out = append(out, e) + } + } + return InOrder(out) +} + +// The tools' descriptions. +const ( + PublishAgainDescription = "Publish one fingerprint's version again after its holder failed (design 51): with the build's file held, the " + + "builder says version.numbered again and the build module publishes the held digest; with the file gone, it builds the fingerprint " + + "again under the same number, the new digest replacing the recorded one, since no published was said for it; after a published (or " + + "a removal) it takes a new number. It refuses a fingerprint not numbered and a version the holder refused for other bytes. A hand " + + "act, recorded with its caller and why in the builder's log." + FloorAgainDescription = "Ask the floor of one name again now, by hand (design 51 *The floor*): the seat's versions, the forge bridge's " + + "bridge_tags and the indexer's indexer_requirements, as the builder asks them by itself when a silent source says something. " + + "Answers which sources were silent and whether the floor is now known. A name the builder's versions already hold needs no floor: " + + "refused, saying so. A hand act." + TagDescription = "Ask a tagged version for a branch's head by hand, as the checker's tag-wanted asks it (design 51 *Tagged versions*): " + + "the builder keeps the ask in its state tag-asks until the branch is merged or deleted, or its pull request closed, and numbers " + + "each package the head builds below the next trunk version (0.1.15-.2). A hand act." + TagAsksDescription = "The open tag asks of the builder's state tag-asks, oldest first: repository, branch, head, tag, who asked and why, " + + "its state (asked, numbered, removing, removed), the versions numbered for it and, while its tagged versions are removed, the " + + "seats that have not yet said removed. With repository, that repository's only. Read-only." + UntagDescription = "Remove a branch's tagged versions by hand, as its merge, its pull request's close or its deletion does (design 51 " + + "*Tagged versions*): the builder calls each seat's remove with the reason tag-gone and marks each version removed once its seat says " + + "removed. Answers the versions it removes and the seats still waited on. A version with no tag is never removed. A hand act." +) + +// Acts are the builder's hand acts on its versions, answered by its program (the tools bundle writes them to reasks). +type Acts struct { + Numberer *Numberer + // Done says whether the build key of module and fingerprint is done: the build's file is held. + Done func(ctx context.Context, module, fp string) (bool, error) + // BuildAgain builds a failed key again (builder_build_again's path): its answer, or why it refused. + BuildAgain func(ctx context.Context, module, fp, why, by string) (answer string, refused bool) + // Reconsider looks at every module again, so a claim asks a name's floor. + Reconsider func(ctx context.Context) + // Remove calls a seat's verb remove for one tagged version, reason tag-gone. + Remove func(ctx context.Context, seat string, e Entry) error +} + +// TagAsked answers one unanswered value of tag-asks. An untag calls remove on each of the branch's tagged versions' +// seats and leaves each in Left until its seat says removed (Heard, on the seat's events); a call that fails stays in +// Left with its error, asked again by the branch's end. A tag ask is kept: its versions are numbered when the head's +// packages are claimed for the group's check. +func (a Acts) TagAsked(ctx context.Context, t TagAsk, now time.Time) TagAsk { + t.Answered = now.UTC() + if !t.Untag { + t.State = TagAsked + t.Answer = "kept: each package the head builds is numbered " + "-" + t.Tag + ". when its build is claimed for the check" + return t + } + all, err := a.Numberer.Versions.All(ctx) + if err != nil { + t.Answer = "nothing removed: " + err.Error() + return t + } + t.Left, t.Versions = map[string]string{}, map[string]string{} + for _, e := range Untagged(all, t.Repository, t.Tag) { + t.Versions[e.Name] = e.Version + at := e.Name + "@" + e.Version + t.Left[at] = e.Seat + if err := a.Remove(ctx, e.Seat, e); err != nil { + t.Left[at] = e.Seat + ": " + err.Error() + } + } + if len(t.Left) == 0 { + t.State, t.Left, t.Answer = TagRemoved, nil, "the branch holds no tagged version: nothing to remove" + return t + } + t.State = TagRemoving + t.Answer = fmt.Sprintf("remove asked of %d tagged versions; each is marked removed when its seat says removed", len(t.Left)) + return t +} + +// PublishAgain is builder_publish_again of name's fingerprint: what it did, the version, and its answer; an error refuses. +func (a Acts) PublishAgain(ctx context.Context, name, fp, why, by string, at time.Time) (Again, string, error) { + module, _, _ := strings.Cut(name, "/") + entries, err := a.Numberer.Versions.Of(ctx, name) + if err != nil { + return Again{}, "", err + } + done, err := a.Done(ctx, module, fp) + if err != nil { + return Again{}, "", fmt.Errorf("the build key of %s at %s cannot be read: %w", module, fp, err) + } + ag, err := PublishAgain(entries, fp, done) + if err != nil { + return Again{}, "", err + } + e := ag.Entry + if ag.Do == AgainRenumber { + _, ne, err := a.Numberer.Versions.Number(ctx, Claim{Name: e.Name, Kind: e.Kind, Series: e.Series, Module: e.Module, Fingerprint: e.Fingerprint, + Commit: e.Commit, Revision: e.Revision, Seat: e.Seat, Merge: e.Merge, Fresh: true}) + if err != nil { + return Again{}, "", err + } + ag.Entry = ne + } + answer := ag.Why + if ag.Do == AgainBuild || (ag.Do == AgainRenumber && !done) { + said, refused := a.BuildAgain(ctx, module, fp, why, by) + if refused { + return Again{}, "", fmt.Errorf("%s; and the build again was refused: %s", ag.Why, said) + } + answer += "; " + said + } + if ag.Do == AgainPublish || (ag.Do == AgainRenumber && done) { + if a.Numberer.Said != nil { + b := NumberedOf(ag.Entry, "") + b.Again = at.UTC().Format(time.RFC3339Nano) + a.Numberer.Said(b) + } + answer += "; " + NumberedEvent + " said for " + ag.Entry.Version + } + return ag, answer, nil +} + +// FloorAgain is builder_floor_again of name: the floor counts only for a name the versions hold nothing of. +func (a Acts) FloorAgain(ctx context.Context, name string) (string, error) { + entries, err := a.Numberer.Versions.Of(ctx, name) + if err != nil { + return "", err + } + if len(entries) > 0 { + return "", fmt.Errorf("%s holds %d versions: its floor counts only for its first number, so nothing is asked", name, len(entries)) + } + answer := name + " waits on no floor in this builder: the floor is asked at its next claim" + if a.Numberer.Waits != nil { + for _, w := range a.Numberer.Waits.Waiting() { + if w.Name == name { + answer = fmt.Sprintf("the floor of %s is asked again (%s were silent)", name, strings.Join(w.Silent, ", ")) + } + } + a.Numberer.Waits.Wake(WakeByHand, "", "", name) + } + a.Reconsider(ctx) + return answer + "; builder_versions shows its version once numbered", nil +} diff --git a/modules/mesh-delivery-builder/internal/versions/claim.go b/modules/mesh-delivery-builder/internal/versions/claim.go index f16ac6d0..04b3bb27 100644 --- a/modules/mesh-delivery-builder/internal/versions/claim.go +++ b/modules/mesh-delivery-builder/internal/versions/claim.go @@ -156,7 +156,7 @@ func (n *Numberer) wait(name, seat string, f Floor, why map[string]string) { n.Waits.Wait(name, seat, holder, f, why, now) } -// Built records the digest each numbered artifact was built with, on an entry that holds none yet: what the holder's +// Built records the digest each numbered artifact was built with, on an entry that holds none yet or was never published: what the holder's // builder_versions question compares (ADR 0605 §4). func (n *Numberer) Built(ctx context.Context, c BuildClaim, got Numbering, digests map[string]string) error { var errs []string @@ -166,7 +166,9 @@ func (n *Numberer) Built(ctx context.Context, c BuildClaim, got Numbering, diges continue } if _, err := n.Versions.Mark(ctx, NameOf(c.Module, artifact), version, func(e *Entry) { - if e.Digest == "" { + // The new digest replaces the recorded one only while no published was said for the number (design 51 + // *Tools of the builder*): a rebuild of a gone file under the same number. + if e.Digest == "" || e.State == Numbered || e.State == Refused { e.Digest = d } }); err != nil { diff --git a/modules/mesh-delivery-builder/internal/versions/heard.go b/modules/mesh-delivery-builder/internal/versions/heard.go index e4cb830b..d5223107 100644 --- a/modules/mesh-delivery-builder/internal/versions/heard.go +++ b/modules/mesh-delivery-builder/internal/versions/heard.go @@ -48,7 +48,7 @@ func (n *Numberer) Heard(ctx context.Context, seat, event, id string, o Outcome) case HeardRefused: change = func(e *Entry) { if e.State != Published { - e.State, e.Reason = Refused, o.Reason + e.State, e.Reason, e.RefusedDigest = Refused, o.Reason, o.Digest } } case HeardRemoved: diff --git a/modules/mesh-delivery-builder/internal/versions/store.go b/modules/mesh-delivery-builder/internal/versions/store.go index 4101d9b9..b29eff1a 100644 --- a/modules/mesh-delivery-builder/internal/versions/store.go +++ b/modules/mesh-delivery-builder/internal/versions/store.go @@ -7,6 +7,9 @@ import ( "time" ) +// AllFilter is the key filter of every version of every name: two tokens. +const AllFilter = "*.*" + // 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") @@ -15,7 +18,7 @@ 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 are the keys under a filter (NameFilter, or AllFilter for every key). KeysOf(ctx context.Context, filter string) ([]string, error) } @@ -52,10 +55,15 @@ func (v Versions) Of(ctx context.Context, name string) ([]Entry, error) { // Claim is one numbering asked of the builder. type Claim struct { Name, Kind, Series, Module, Fingerprint, Commit, Revision, Digest, Seat string + // Repository is the module's repository, kept on a tagged version for its branch's end. + Repository string // Tag asks a tagged version for this cleaned tag; empty for a trunk version. Tag string Merge uint64 Floor Floor + // Fresh numbers anew a fingerprint or revision already numbered: builder_publish_again after a published (design 51 + // *Tools of the builder*). + Fresh bool } // Number numbers a claim under compare-and-set: two claims of one name never get one number, the second reading the @@ -78,7 +86,7 @@ func (v Versions) Number(ctx context.Context, c Claim) (Decision, Entry, error) 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) + d = Decide(fresh(entries, c), c.Kind, c.Series, c.Fingerprint, c.Revision, c.Floor) } if d.Version == "" { return d, Entry{}, nil @@ -93,6 +101,9 @@ func (v Versions) Number(ctx context.Context, c Claim) (Decision, Entry, error) 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} + if c.Tag != "" { + e.Repository = c.Repository + } raw, err := Encode(e) if err != nil { return Decision{}, Entry{}, err @@ -170,3 +181,43 @@ func (v Versions) now() time.Time { } return time.Now().UTC() } + +// fresh is entries as Decide reads them for c: with Fresh, the entries of c's fingerprint and revision still count for the +// next number but keep none. +func fresh(entries []Entry, c Claim) []Entry { + if !c.Fresh { + return entries + } + out := make([]Entry, len(entries)) + for i, e := range entries { + if e.Tag == "" && (e.Fingerprint == c.Fingerprint || (c.Revision != "" && e.Revision == c.Revision)) { + e.Fingerprint, e.Revision = "", "" + } + out[i] = e + } + return out +} + +// All is every entry of every name, by name in order: what builder_untag and the tag asks read. +func (v Versions) All(ctx context.Context) ([]Entry, error) { + keys, err := v.Store.KeysOf(ctx, AllFilter) + if err != nil { + return nil, fmt.Errorf("the versions cannot be listed: %w", 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 out, nil +} diff --git a/modules/mesh-delivery-builder/internal/versions/tools.go b/modules/mesh-delivery-builder/internal/versions/tools.go index ece1d14b..bfe00f39 100644 --- a/modules/mesh-delivery-builder/internal/versions/tools.go +++ b/modules/mesh-delivery-builder/internal/versions/tools.go @@ -37,6 +37,8 @@ type NumberedBody struct { Seat string `json:"seat,omitempty"` // Head is the head the version answers, when it answers a tag ask. Head string `json:"head,omitempty"` + // Again is when builder_publish_again said it once more, so the bus's dedup does not drop it. + Again string `json:"again,omitempty"` } // NumberedOf is the event of an entry. @@ -69,6 +71,9 @@ func (s readStore) KeysOf(_ context.Context, filter string) ([]string, error) { if err != nil { return nil, err } + if filter == AllFilter { + return all, nil + } prefix := strings.TrimSuffix(filter, "*") var out []string for _, k := range all { diff --git a/modules/mesh-delivery-builder/internal/versions/versions.go b/modules/mesh-delivery-builder/internal/versions/versions.go index 4fd501a9..42c07212 100644 --- a/modules/mesh-delivery-builder/internal/versions/versions.go +++ b/modules/mesh-delivery-builder/internal/versions/versions.go @@ -67,8 +67,13 @@ type Entry struct { 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"` + // Repository is a tagged version's repository: its branch's end removes it (Untagged). + Repository string `json:"repository,omitempty"` // RemovedBy is the id of the event that removed it. RemovedBy string `json:"removed_by,omitempty"` + // RefusedDigest is the digest the holder refused: one other than Digest means the seat was offered other bytes under + // this number, which builder_publish_again refuses to publish again (design 51 *Tools of the builder*). + RefusedDigest string `json:"refused_digest,omitempty"` // Merge is the forge bridge's sequence of the merge that numbered it, for merge order. Merge uint64 `json:"merge,omitempty"` } diff --git a/modules/mesh-delivery-builder/internal/versions/versions_test.go b/modules/mesh-delivery-builder/internal/versions/versions_test.go index 42fe6254..a2eef4bd 100644 --- a/modules/mesh-delivery-builder/internal/versions/versions_test.go +++ b/modules/mesh-delivery-builder/internal/versions/versions_test.go @@ -61,6 +61,10 @@ func (m *mem) KeysOf(_ context.Context, filter string) ([]string, error) { prefix := strings.TrimSuffix(filter, "*") var out []string for k := range m.vals { + if filter == AllFilter { + out = append(out, k) + continue + } if strings.HasPrefix(k, prefix) && !strings.Contains(k[len(prefix):], ".") { out = append(out, k) } @@ -394,4 +398,161 @@ func TestTheBuildersVersions(t *testing.T) { t.Error("80 % is not full") } }) + t.Run("publish again", func(t *testing.T) { + num := Entry{Name: "m/p", Kind: KindBundle, Version: "0.1.3", Fingerprint: "f1", Digest: "sha256:aa", State: Numbered} + ref := num + ref.State, ref.RefusedDigest = Refused, "sha256:aa" + other := ref + other.RefusedDigest = "sha256:bb" + pub := num + pub.State = Published + gone := pub + gone.State = Removed + tagged := num + tagged.Tag, tagged.Version = "feat-x", "0.1.3-feat-x.1" + for _, c := range []struct { + name string + entries []Entry + fp string + built bool + do string + }{ + {"a held build is said numbered again", []Entry{num}, "f1", true, AgainPublish}, + {"refused for its own bytes is published again", []Entry{ref}, "f1", true, AgainPublish}, + {"a gone file is built again under the same number", []Entry{ref}, "f1", false, AgainBuild}, + {"after a published it takes a new number", []Entry{pub}, "f1", true, AgainRenumber}, + {"after a removal it takes a new number", []Entry{gone}, "f1", false, AgainRenumber}, + {"refused for other bytes is refused", []Entry{other}, "f1", true, ""}, + {"a fingerprint not numbered is refused", []Entry{num}, "f2", true, ""}, + {"a tagged version is not a trunk fingerprint's", []Entry{tagged}, "f1", true, ""}, + } { + a, err := PublishAgain(c.entries, c.fp, c.built) + if (err != nil) != (c.do == "") || a.Do != c.do { + t.Errorf("%s: PublishAgain = %+v, %v; want %q", c.name, a, err, c.do) + } + } + ctx := context.Background() + // A fresh claim of a published fingerprint takes the next number; the old one stays. + st := newMem() + v := Versions{Store: st} + raw, _ := Encode(Entry{Name: "m/p", Kind: KindBundle, Series: "0.1", Version: "0.1.3", Fingerprint: "f1", State: Published, At: time.Unix(1, 0)}) + _, _ = st.Create(ctx, Key("m/p", "0.1.3"), raw) + if d, _, err := v.Number(ctx, Claim{Name: "m/p", Kind: KindBundle, Fingerprint: "f1"}); err != nil || d.Version != "0.1.3" || !d.Keep { + t.Fatalf("a published fingerprint keeps its number: %+v %v", d, err) + } + if d, e, err := v.Number(ctx, Claim{Name: "m/p", Kind: KindBundle, Fingerprint: "f1", Fresh: true}); err != nil || d.Version != "0.1.4" || d.Keep || e.Fingerprint != "f1" { + t.Fatalf("a fresh claim takes the next number: %+v %+v %v", d, e, err) + } + // A refusal records the digest refused; Built replaces a digest no published was said for, never a published one. + n := &Numberer{Versions: v} + _, _ = n.Heard(ctx, SeatArtifacts, HeardRefused, "r1", Outcome{Name: "m/p", Version: "0.1.4", Digest: "sha256:old", Reason: "x"}) + if es, _ := v.Of(ctx, "m/p"); es[1].RefusedDigest != "sha256:old" { + t.Errorf("refused records no digest: %+v", es[1]) + } + _, _ = v.Mark(ctx, "m/p", "0.1.4", func(e *Entry) { e.Digest = "sha256:old" }) + bc := BuildClaim{Module: "m"} + _ = n.Built(ctx, bc, Numbering{Versions: map[string]string{"p": "0.1.4"}}, map[string]string{"p": "sha256:new"}) + _, _ = v.Mark(ctx, "m/p", "0.1.3", func(e *Entry) { e.Digest = "sha256:held" }) + _ = n.Built(ctx, bc, Numbering{Versions: map[string]string{"p": "0.1.3"}}, map[string]string{"p": "sha256:other"}) + es, _ := v.Of(ctx, "m/p") + if es[0].Digest != "sha256:held" || es[1].Digest != "sha256:new" { + t.Errorf("Built: published %s (want sha256:held), refused %s (want sha256:new)", es[0].Digest, es[1].Digest) + } + if all, err := v.All(ctx); err != nil || len(all) != 2 { + t.Errorf("All = %d entries, %v", len(all), err) + } + // The hand acts: a held build is said numbered once more under a fresh dedup id; a gone one is built again; a + // published one takes a new number; a name with versions asks no floor. + if _, _, err := (Acts{Numberer: &Numberer{Versions: v}, Done: func(context.Context, string, string) (bool, error) { return true, nil }}).PublishAgain(ctx, "m/p", "f1", "w", "", time.Unix(4, 0)); err == nil { + t.Error("a version refused for other bytes is published again") + } + _, _ = v.Mark(ctx, "m/p", "0.1.4", func(e *Entry) { e.State = Published }) + var said []NumberedBody + var rebuilt []string + done := true + acts := Acts{Numberer: &Numberer{Versions: v, Said: func(b NumberedBody) { said = append(said, b) }, Waits: &FloorWaits{}}, + Done: func(context.Context, string, string) (bool, error) { return done, nil }, + BuildAgain: func(_ context.Context, m, fp, _, _ string) (string, bool) { + rebuilt = append(rebuilt, m+" "+fp) + return "built again", false + }, + Reconsider: func(context.Context) {}} + if ag, _, err := acts.PublishAgain(ctx, "m/p", "f1", "w", "me", time.Unix(5, 0)); err != nil || ag.Do != AgainRenumber || ag.Entry.Version != "0.1.5" || + len(said) != 1 || said[0].Version != "0.1.5" || said[0].Again == "" || len(rebuilt) != 0 { + t.Errorf("publish again of a published, held build: %+v %v said %+v rebuilt %v", ag, err, said, rebuilt) + } + done = false + if ag, _, err := acts.PublishAgain(ctx, "m/p", "f1", "w", "me", time.Unix(6, 0)); err != nil || ag.Do != AgainBuild || ag.Entry.Version != "0.1.5" || len(rebuilt) != 1 || len(said) != 1 { + t.Errorf("publish again of a gone build: %+v %v said %d rebuilt %v", ag, err, len(said), rebuilt) + } + if _, err := acts.FloorAgain(ctx, "m/p"); err == nil { + t.Error("a name with versions asks its floor again") + } + if ans, err := acts.FloorAgain(ctx, "m/new"); err != nil || ans == "" { + t.Errorf("floor again of a new name: %q %v", ans, err) + } + }) + t.Run("tag asks", func(t *testing.T) { + at := time.Unix(100, 0) + k, raw, err := NewTagAsk("novox/mesh-sdk", "feat/X y", "abc", "a check", "me@shanks", at) + if err != nil || k != TagAskKey("novox/mesh-sdk", "feat/X y") || strings.Count(k, ".") != 1 { + t.Fatalf("NewTagAsk = %q, %v", k, err) + } + a, err := DecodeTagAsk(raw) + if err != nil || a.Tag != "feat-x-y" || a.State != TagAsked || a.Untag { + t.Fatalf("DecodeTagAsk = %+v, %v", a, err) + } + for _, bad := range [][4]string{{"mesh-sdk", "b", "h", "w"}, {"o/r", "", "h", "w"}, {"o/r", "b", "", "w"}, {"o/r", "b", "h", ""}} { + if _, _, err := NewTagAsk(bad[0], bad[1], bad[2], bad[3], "", at); err == nil { + t.Errorf("NewTagAsk%v is not refused", bad) + } + } + _, uraw, err := NewUntag("o/r", "feat/x", "branch gone by hand", "", at.Add(time.Second)) + if u, _ := DecodeTagAsk(uraw); err != nil || !u.Untag { + t.Fatalf("NewUntag = %+v, %v", u, err) + } + vals := map[string][]byte{"a": uraw, "b": raw} + r := Reader{Keys: func() ([]string, error) { return []string{"a", "b"}, nil }, + Get: func(k string) (json.RawMessage, bool, error) { return vals[k], true, nil }} + if got, err := AnswerTagAsks(r, ""); err != nil || len(got.Asks) != 2 || got.Asks[0].Repository != "novox/mesh-sdk" { + t.Errorf("AnswerTagAsks = %+v, %v", got, err) + } + if got, _ := AnswerTagAsks(r, "o/r"); len(got.Asks) != 1 || !got.Asks[0].Untag { + t.Errorf("AnswerTagAsks(o/r) = %+v", got) + } + es := []Entry{{Name: "n", Version: "0.1.2-feat-x.1", Tag: "feat-x", Series: "0.1", Repository: "o/r"}, {Name: "n", Version: "0.1.2-feat-x.2", Tag: "feat-x", Series: "0.1", State: Removed, Repository: "o/r"}, + {Name: "n2", Version: "0.1.2-feat-x.1", Tag: "feat-x", Series: "0.1", Repository: "o/other"}, + {Name: "n", Version: "0.1.1", Series: "0.1"}, {Name: "n", Version: "0.1.2-feat-y.1", Tag: "feat-y", Series: "0.1"}} + if got := Untagged(es, "o/r", "feat/x"); len(got) != 1 || got[0].Version != "0.1.2-feat-x.1" { + t.Errorf("Untagged = %+v", got) + } + st := newMem() + for _, e := range append(es, Entry{Name: "n3", Version: "0.1.0-feat-x.1", Tag: "feat-x", Series: "0.1", Repository: "o/r", Seat: SeatNPM}) { + e.Kind, e.Module, e.Fingerprint, e.At = KindPackage, "m", "f", at + if e.State == "" { + e.State = Numbered + } + raw, err := Encode(e) + if err != nil { + t.Fatal(err) + } + _, _ = st.Create(context.Background(), Key(e.Name, e.Version), raw) + } + var removed []string + acts := Acts{Numberer: &Numberer{Versions: Versions{Store: st}}, Remove: func(_ context.Context, seat string, e Entry) error { + removed = append(removed, e.Name+"@"+e.Version) + if seat == SeatNPM { + return errors.New("down") + } + return nil + }} + u, _ := DecodeTagAsk(uraw) + got := acts.TagAsked(context.Background(), u, at) + if got.State != TagRemoving || len(removed) != 2 || len(got.Left) != 2 || !strings.Contains(got.Left["n3@0.1.0-feat-x.1"], "down") || got.Answered.IsZero() { + t.Errorf("untag: %+v removed %v", got, removed) + } + if k := acts.TagAsked(context.Background(), a, at); k.State != TagAsked || k.Answer == "" { + t.Errorf("a tag ask is not kept: %+v", k) + } + }) } diff --git a/modules/mesh-delivery-builder/manifest_test.go b/modules/mesh-delivery-builder/manifest_test.go index 99e63787..cc78f53c 100644 --- a/modules/mesh-delivery-builder/manifest_test.go +++ b/modules/mesh-delivery-builder/manifest_test.go @@ -93,7 +93,7 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1 if m.Module != "mesh-delivery-builder" { t.Errorf("module = %q", m.Module) } - if want := []state{{Name: "builds", History: 5}, {Name: "hearing", History: 1}, {Name: "instances", History: 1}, {Name: "queues", History: 1}, {Name: "reasks", History: 1}, {Name: "versions", History: 1}}; !reflect.DeepEqual(m.State, want) { // delivery.BuilderBuildsHistory + if want := []state{{Name: "builds", History: 5}, {Name: "hearing", History: 1}, {Name: "instances", History: 1}, {Name: "queues", History: 1}, {Name: "reasks", History: 1}, {Name: "tag-asks", History: 1}, {Name: "versions", History: 1}}; !reflect.DeepEqual(m.State, want) { // delivery.BuilderBuildsHistory t.Errorf("state = %+v, want %+v", m.State, want) } // 1887/A4c: the builder chooses no node, so it reads neither presence nor have, invokes no build module's tool and @@ -109,8 +109,9 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1 // other tool; it holds no capacity. // 1955/L7 (ADR 0571 §2): the seat mesh-artifact-sweep's held, the controller's `held` is no longer asked (issue 1955). // Design 51 *The floor*: the floor's three sources, the registry seats' versions, bridge_tags and indexer_requirements. - if want := []string{"mesh-delivery-forge-bridge.bridge_tags", "mesh-delivery-indexer.indexer_requirements", "seat:container-image-registry.versions", - "seat:go-package-registry.versions", "seat:mesh-artifact-sweep.held", "seat:npm-package-registry.versions", "seat:oci-artifact-registry.versions"}; !reflect.DeepEqual(sorted(m.Invokes), want) || len(m.Config) != 0 { + if want := []string{"mesh-delivery-forge-bridge.bridge_tags", "mesh-delivery-indexer.indexer_requirements", "seat:container-image-registry.remove", "seat:container-image-registry.versions", + "seat:go-package-registry.remove", "seat:go-package-registry.versions", "seat:mesh-artifact-sweep.held", "seat:npm-package-registry.remove", + "seat:npm-package-registry.versions", "seat:oci-artifact-registry.remove", "seat:oci-artifact-registry.versions"}; !reflect.DeepEqual(sorted(m.Invokes), want) || len(m.Config) != 0 { t.Errorf("invokes = %v, config = %v, want %v and no config", m.Invokes, m.Config, want) } // The builder is woken by a build module's report of a toolchain, as the indexer is: the bucket may not exist when it @@ -128,7 +129,8 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1 t.Errorf("secrets-owner = %q", m.SecretsOwner) } // 1945: the builder's tools (builder_seat_queue: note 62), served by its tools bundle, and the event it says when it asks an ask again. - if want := []string{"builder_announce_again", "builder_ask_again", "builder_build_again", "builder_builds_since", "builder_seat_queue", "builder_stuck", "builder_versions"}; !reflect.DeepEqual(sorted(m.Tools), want) { + if want := []string{"builder_announce_again", "builder_ask_again", "builder_build_again", "builder_builds_since", "builder_floor_again", + "builder_publish_again", "builder_seat_queue", "builder_stuck", "builder_tag", "builder_tag_asks", "builder_untag", "builder_versions"}; !reflect.DeepEqual(sorted(m.Tools), want) { t.Errorf("tools = %v, want %v", m.Tools, want) } // Design 51 *Events of the builder*: numbered and old-form-closed on the family version, apart from the builds' ends. diff --git a/modules/mesh-delivery-builder/module.json b/modules/mesh-delivery-builder/module.json index 0c960725..87c4b61a 100644 --- a/modules/mesh-delivery-builder/module.json +++ b/modules/mesh-delivery-builder/module.json @@ -7,8 +7,13 @@ "builder_ask_again", "builder_build_again", "builder_builds_since", + "builder_floor_again", + "builder_publish_again", "builder_seat_queue", "builder_stuck", + "builder_tag", + "builder_tag_asks", + "builder_untag", "builder_versions" ], "emits": [ @@ -26,10 +31,14 @@ "invokes": [ "mesh-delivery-forge-bridge.bridge_tags", "mesh-delivery-indexer.indexer_requirements", + "seat:container-image-registry.remove", "seat:container-image-registry.versions", + "seat:go-package-registry.remove", "seat:go-package-registry.versions", "seat:mesh-artifact-sweep.held", + "seat:npm-package-registry.remove", "seat:npm-package-registry.versions", + "seat:oci-artifact-registry.remove", "seat:oci-artifact-registry.versions" ], "state": [ @@ -38,6 +47,7 @@ { "name": "instances", "history": 1 }, { "name": "queues", "history": 1 }, { "name": "reasks", "history": 1 }, + { "name": "tag-asks", "history": 1 }, { "name": "versions", "history": 1 } ], "logs": [