novox/mesh-delivery
10
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

BL10: tagged versions and version.numbered go from the builder

Browse Source 
The state tag-asks and its bounds, builder_tag, builder_tag_asks and builder_untag, the checker's tag-wanted
consumer, the branch end's removal of tagged versions, the numbering of a check's build under a tag and the
event version.numbered (its contract in builds, its emit, its manifest line) are removed; a forge message still
wakes a floor. builder_publish_again refuses a held build, since nothing says a version numbered.

Builds: design 47 G §373, §321, §349, §413

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
jschoubben
2026-11-11 20:35:04 +01:00
parent 848c2b035e
commit bd3727a900
23 changed files with 103 additions and 1421 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-builder/builds/builds.go
+1 -1
View File
@@ -1,6 +1,6 @@
// Package builds holds what the builder module owns and no seat carries (novox/hq ADR 0634 §1): the layout of its state
// `builds` (a build per module and fingerprint) and the rules its writer keeps, the question a registry seat's holder asks
// it before it stores a publish, its event version.numbered and the entries of its build log. Other modules import it from
// it before it stores a publish and the entries of its build log. Other modules import it from
// the builder by its published version (ADR 0573 §1); the builder's manifest declares it as a package artifact.
package builds
modules/mesh-delivery-builder/builds/numbered.go
-57
View File
@@ -1,57 +0,0 @@
package builds
import (
"encoding/json"
"fmt"
)
// The builder's event version.numbered (design 51 *Events of the builder*, *The floor*): said once per version the builder
// numbers, on its own family `version` (seven days), apart from its builds' ends, so a waiter on a build's end does not
// wake on it. A build module consumes it to publish a build's version with the held digest, which stores nothing.
const (
// VersionNumberedEvent is the event's local name.
VersionNumberedEvent = "version.numbered"
// VersionNumberedSubject is its subject, as a consumer's filter names it.
VersionNumberedSubject = "mesh.mod." + BuilderModule + ".event." + VersionNumberedEvent
)
// VersionNumbered is the body of VersionNumberedEvent: a numbered version of a name, the build it was numbered for, and
// what the build module needs to publish it.
type VersionNumbered struct {
// Name is the numbered name, as builder_versions lists it; Kind its artifact's kind.
Name string `json:"name"`
Kind string `json:"kind"`
Version string `json:"version"`
Tag string `json:"tag,omitempty"`
// Module and Fingerprint are the build the version was numbered for; Commit its source commit.
Module string `json:"module"`
Fingerprint string `json:"fingerprint"`
Commit string `json:"commit,omitempty"`
// Digest is the built artifact's, when the build held one; Seat the registry seat it is published through.
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"`
// Again is when builder_publish_again said it once more, so the bus's dedup does not drop it.
Again string `json:"again,omitempty"`
}
// EncodeVersionNumbered writes the body, refusing one that names no name, version, module or fingerprint.
func EncodeVersionNumbered(n VersionNumbered) ([]byte, error) {
if n.Name == "" || n.Version == "" || n.Module == "" || n.Fingerprint == "" {
return nil, fmt.Errorf("builds: %s names its name, version, module and fingerprint", VersionNumberedEvent)
}
return json.Marshal(n)
}
// DecodeVersionNumbered reads the body, with the same refusal.
func DecodeVersionNumbered(data []byte) (VersionNumbered, error) {
var n VersionNumbered
if err := json.Unmarshal(data, &n); err != nil {
return VersionNumbered{}, fmt.Errorf("builds: %s is not readable: %w", VersionNumberedEvent, err)
}
if _, err := EncodeVersionNumbered(n); err != nil {
return VersionNumbered{}, err
}
return n, nil
}
modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go
+2 -49
View File
@@ -35,9 +35,8 @@ 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.
// 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, fw, of versions.Reader) []stdio.Tool {
fw, of 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) {
@@ -57,32 +56,10 @@ func Tools(r build.StuckReader, askAgain func(key string, value []byte) error, q
}
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, <module>/<artifact>, as builder_versions lists it"},
"fp": map[string]any{"type": "string", "description": "the fingerprint, src2:<hex>"},
"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": "<owner>/<name>"},
"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) {
@@ -95,18 +72,6 @@ func Tools(r build.StuckReader, askAgain func(key string, value []byte) error, q
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, <owner>/<name>"}}},
Run: func(args map[string]any) (any, error) { return versions.AnswerTagAsks(ta, str(args, "repository")) }},
{Name: versions.OldFormTool, Description: versions.OldFormDescription,
Input: map[string]any{"type": "object", "properties": map[string]any{
"drop": map[string]any{"type": "string", "description": "a retired node to drop from the count by hand"},
@@ -373,18 +338,6 @@ func main() {
}
return e.Value, true, 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
}}
fs := stdio.State(versions.FloorWaitsState)
fr := versions.Reader{Keys: fs.Keys, Get: func(key string) (json.RawMessage, bool, error) {
e, err := fs.Get(key)
@@ -417,7 +370,7 @@ func main() {
}
return e.Value, true, nil
}}
tools := Tools(r, put, askQueue(stdio.State(build.QueuesState), 25*time.Second), buildAgain, stdio.Log(builds.BuilderLog).Since, time.Now, vr, tags, tr, fr, ofr)
tools := Tools(r, put, askQueue(stdio.State(build.QueuesState), 25*time.Second), buildAgain, stdio.Log(builds.BuilderLog).Since, time.Now, vr, fr, ofr)
tools = append(tools, reservedTools(rr, buildAgain, time.Now, meshhandactsclient.Recorder{Ask: stdio.Ask})...)
// The builder's expiring keys (state bounds): what each bounds and when it ends.
tools = append(tools, stdio.Tool{Name: versions.BoundsTool, Description: versions.BoundsDescription,
modules/mesh-delivery-builder/cmd/mesh-delivery-builder/branchend.go
+8 -155
View File
@@ -2,173 +2,26 @@ package main
import (
"context"
"encoding/json"
"errors"
"strings"
"time"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg"
"mesh-delivery-builder/internal/build"
"mesh-delivery-builder/internal/versions"
)
// forgeStream is the forge bridge's stream `forge` as the bus names it, where the builder's durable consumer hears a
// branch's end (design 51 *Tagged versions*).
// forgeStream is the forge bridge's stream `forge` as the bus names it, where the builder's durable consumer hears the
// forge's messages: each wakes a floor whose bridge_tags was silent (design 51 *The floor*). No tagged version is
// removed on a branch's end, none being made (design 47 G §373).
const forgeStream = "FORGE"
// forgeBranchDeleted is the forge bridge's kind for a branch's end (forgeBranchDeleted in the group's mesh-sdk).
const forgeBranchDeleted = "branch-deleted"
// checkerAsksStream is the checker's family `asks` as the bus names it, and tagWantedSubject its event tag-wanted
// (design 51 *Tagged versions*, DLV-C3): on the checker's own stream, so its subject is mesh.stream.<module>.event.*.
const (
checkerAsksStream = "ASKS"
tagWantedSubject = "mesh.stream." + checkerModule + ".event.tag-wanted"
)
// forgeEnd is the part of a forge message the builder reads.
type forgeEnd struct {
Kind string `json:"kind"`
Repository string `json:"repository"`
Branch string `json:"branch"`
Action string `json:"action"`
// forgeWakes hears the forge bridge's stream: every message wakes a floor whose bridge_tags was silent, and is acknowledged.
type forgeWakes struct {
wake func(ctx context.Context)
logf func(string, ...any)
}
// ended says whether the message ends a branch: merged, a pull closed, or branch-deleted.
func (f forgeEnd) ended() bool {
switch f.Kind {
case forgemsg.ForgeMerged, forgeBranchDeleted:
return f.Branch != ""
case forgemsg.ForgePull:
return f.Action == forgemsg.PullClosed && f.Branch != ""
}
return false
}
// branchEnds hears the forge bridge's stream: every message wakes a floor whose bridge_tags was silent, and a branch's
// end removes its tagged versions, acknowledged only once all are removed.
type branchEnds struct {
acts versions.Acts
findings *findings
tagAsks func() jetstream.KeyValue
wake func(ctx context.Context)
logf func(string, ...any)
now func() time.Time
}
func (h *branchEnds) heard(ctx context.Context, m jetstream.Msg) {
func (h *forgeWakes) heard(ctx context.Context, m jetstream.Msg) {
if h.wake != nil {
h.wake(ctx)
}
var f forgeEnd
if err := json.Unmarshal(m.Data(), &f); err != nil || !f.ended() {
h.ack(m)
return
}
end, err := h.acts.BranchEnded(ctx, f.Repository, f.Branch)
if err != nil {
h.logf("the end of %s %s: the versions cannot be read, heard again: %v", f.Repository, f.Branch, err)
h.again(m)
return
}
id := f.Repository + "/" + f.Branch
if end.Done {
h.findings.clear(ctx, versions.RemovalNotDone, meshconditionsclient.ScopeModule, id)
if kv := h.tagAsks(); kv != nil {
if err := kv.Delete(ctx, versions.TagAskKey(f.Repository, f.Branch)); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
h.logf("the tag ask of %s %s stays in %s: %v", f.Repository, f.Branch, versions.TagAsksState, err)
}
}
h.ack(m)
return
}
since := h.now()
if md, err := m.Metadata(); err == nil {
since = md.Timestamp
}
if h.now().Sub(since) >= versions.RemovalFindingAfter {
h.findings.raise(ctx, finding{kind: versions.RemovalNotDone, scope: meshconditionsclient.ScopeModule, id: id,
summary: end.Finding(f.Repository, f.Branch, since),
headline: "A branch's tagged versions are not removed",
explanation: "The branch ended, and the registry seats did not say removed for every version tagged for it within 15 minutes.",
resolved: "Every seat said removed.",
needs: "Look at the seats named: " + strings.Join(end.Seats(), ", ")})
}
h.logf("the end of %s %s: remove asked of %d tagged versions, heard again in a minute", f.Repository, f.Branch, len(end.Left))
h.again(m)
}
func (h *branchEnds) ack(m jetstream.Msg) {
if err := m.Ack(); err != nil {
h.logf("acknowledging %s: %v", m.Subject(), err)
}
}
// again leaves the message for the bus to give again in a minute: the retry is the bus's redelivery, not a timer here.
func (h *branchEnds) again(m jetstream.Msg) {
if err := m.NakWithDelay(versions.RemoveAgainAfter); err != nil {
h.logf("leaving %s for the bus to give again: %v", m.Subject(), err)
}
}
// tagWanted takes the checker's tag-wanted in: written to tag-asks as builder_tag writes it, answered by its watch.
func tagWanted(ctx context.Context, kv jetstream.KeyValue, m jetstream.Msg, by string, now time.Time, logf func(string, ...any)) {
if m.Subject() != tagWantedSubject {
_ = m.Ack() // the family's other events (the verdicts) are not the builder's
return
}
var t versions.TagWanted
err := json.Unmarshal(m.Data(), &t)
var key string
var value []byte
if err == nil {
key, value, err = versions.TagWantedAsk(t, by, now)
}
if err != nil {
logf("%s holds what is no tag-wanted, left: %v", m.Subject(), err)
_ = m.Ack()
return
}
if kv == nil {
logf("tag-wanted of %s %s is heard again: %s is not on the bus yet", t.Repository, t.Branch, versions.TagAsksState)
_ = m.NakWithDelay(versions.RemoveAgainAfter)
return
}
if _, err := kv.Put(ctx, key, value); err != nil {
logf("tag-wanted of %s %s is heard again: %v", t.Repository, t.Branch, err)
_ = m.Nak()
return
}
_ = m.Ack()
}
// tagsVia is build.Builder.Tags over the state tag-asks: the asks of the repository, read when a check's build is
// looked at; none while the state is not bound.
func tagsVia(kv func() jetstream.KeyValue, logf func(string, ...any)) build.TagOf {
return func(ctx context.Context, repository, head string) (string, bool) {
store := kv()
if store == nil {
return "", false
}
lister, err := store.ListKeysFiltered(ctx, versions.EscapeToken(repository)+".>")
if err != nil {
logf("the tag asks of %s cannot be read, so its check builds no tagged version: %v", repository, err)
return "", false
}
var asks []versions.TagAsk
for key := range lister.Keys() {
e, err := store.Get(ctx, key)
if err != nil {
continue
}
if t, err := versions.DecodeTagAsk(e.Value()); err == nil {
asks = append(asks, t)
}
}
return versions.TagFor(asks, repository, head)
}
}
modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go
+25 -103
View File
@@ -29,13 +29,11 @@ import (
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/gopackageregistry"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/ociartifactregistry"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg"
sdkstep "git.novox.be/novox/mesh-delivery/sdk/buildstep"
"git.novox.be/novox/mesh-sdk/go/buscompose"
"git.novox.be/novox/mesh-sdk/go/keys"
@@ -217,8 +215,6 @@ type bed struct {
builds jetstream.KeyValue
tools map[string]jetstream.KeyValue
instances jetstream.KeyValue
tagAsks jetstream.KeyValue
tagBounds jetstream.KeyValue
packages jetstream.KeyValue // the indexer's packages, served by its read tool and views family
out *saidLog // the builder's process log, every instance's
stop func() // stops the builder started last
@@ -306,12 +302,6 @@ func startBedWith(t *testing.T, closed bool) *bed {
viewstest.Serve(t, ctx, js, ctl, assignerModule, "assignments", assignerModule+".assigner_read_assignments", readers...)
b.builds = mk(builds.BuilderBuildsBucket, builds.BuilderBuildsHistory)
b.instances = mk(instancesBucket, 1)
// The tag asks and their bounds (design 51 *Tagged versions*): the bounds declared key-ttl, so an ended bound leaves the
// bus's marker a watch hears.
b.tagAsks = mk(versions.TagAsksBucket, 1)
if b.tagBounds, err = js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: versions.TagAskBoundsBucket, History: 1, LimitMarkerTTL: time.Minute}); err != nil {
t.Fatal(err)
}
b.out = &saidLog{}
// The builder's module log, as the runtime raises it (ADR 0297): its stream on the log's subjects.
logStream := buscompose.LogStreamName(builds.BuilderModule, builds.BuilderLog)
@@ -525,16 +515,7 @@ func TestBuilderMakesADoneKeyWithTheDigests(t *testing.T) { // 1607/AC7
first := "src2:" + strings.Repeat("1", 64)
b.moduleFP(t, "netcheck", netcheck, "modules/netcheck", bothToolchains, commit, first, 1)
digest := b.moduleFP(t, "netcheck", netcheck, "modules/netcheck", bothToolchains, commit, fp, 2)
// DLV-B6: a tag ask kept and never numbered (no check builds its head) is narrated once its bound ends on the bus, with
// what it waits on, and its bound is written again; no timer runs in the builder.
key, ask, err := versions.NewTagAsk("novox/mesh-sdk", "feat/x", "abc", "a group member's check", "me@shanks", time.Now())
if err != nil {
t.Fatal(err)
}
if _, err := b.tagAsks.Put(b.ctx, key, ask); err != nil {
t.Fatal(err)
}
b.start(t, "n1", map[string]string{"MESH_BUILDER_TAG_ASK_BOUND": "1s", "MESH_BUILDER_STEP_BOUND": "1s"})
b.start(t, "n1", map[string]string{"MESH_BUILDER_STEP_BOUND": "1s"})
v := b.await(t, "netcheck", isState(builds.BuildDone))
for _, a := range []string{"tools-go", "tools-typescript"} {
es := reg.of(t, "netcheck/"+a)
@@ -577,39 +558,17 @@ func TestBuilderMakesADoneKeyWithTheDigests(t *testing.T) { // 1607/AC7
t.Errorf("%s called %d times, want %d (on one node, ADR 0478 clause 4)", tool, got, want)
}
}
said := b.awaitSaid(t, "the tag ask of novox/mesh-sdk feat/x at abc (feat-x) is not answered after")
if !strings.Contains(said, "the check's build of abc is not claimed yet") {
t.Errorf("the narration = %q, want what the ask waits on", said)
}
// And in the build log, under the builder's own key (the ruling of 13:17: build.said).
if !slices.ContainsFunc(b.logged(t, builds.BuilderModule), func(e builds.BuildLogEntry) bool {
return e.Event == builds.BuildSaidEvent && e.Why == versions.SaidUnanswered && e.FP == key && e.Commit == "abc" &&
strings.Contains(e.Line, "the check's build of abc is not claimed yet")
}) {
t.Errorf("the build log of %s = %+v, want the unanswered tag ask said", builds.BuilderModule, b.logged(t, builds.BuilderModule))
}
if e, err := b.tagAsks.Get(b.ctx, key); err != nil {
t.Fatal(err)
} else if a, _ := versions.DecodeTagAsk(e.Value()); a.Answered.IsZero() || !a.Unanswered() {
t.Errorf("the tag ask = %+v, want kept and answered, no version yet", a)
}
for deadline := time.Now().Add(10 * time.Second); ; time.Sleep(50 * time.Millisecond) {
if _, err := b.tagBounds.Get(b.ctx, key); err == nil {
break
} else if time.Now().After(deadline) {
t.Fatalf("the bound is not written again after its narration: %v", err)
}
}
// Design 32, ADR 0615 §1: a package's name the indexer writes later reaches the builder by its packages' views family
// (never their bucket), and looks at the module again; the done bundles are kept, not numbered twice.
if _, err := b.packages.Put(b.ctx, "netcheck.0.1", []byte(`{"v":1,"name":"netcheck","package":"netcheck/lib","series":"0.1"}`)); err != nil {
t.Fatal(err)
}
b.awaitSaid(t, "names a package of netcheck")
if got := reg.numberedOf("netcheck/tools-go"); len(got) != 1 || got[0] != "0.1.5" {
t.Errorf("version.numbered of netcheck/tools-go = %v, want 0.1.5 once", got)
// Design 47 G §349: version.numbered is no longer said, for a number or for a late name.
if got := reg.numberedOf("netcheck/tools-go"); len(got) != 0 {
t.Errorf("version.numbered of netcheck/tools-go said %v, want none", got)
}
// DLV-B10: builder_publish_again after a published takes a new number, and says it numbered.
// DLV-B10: builder_publish_again of a published, held build is refused: nothing says a version numbered (§349).
for _, a := range []string{"tools-go", "tools-typescript"} {
reg.published(t, "netcheck/"+a, "0.1.5", digestOf(a))
}
@@ -618,8 +577,14 @@ func TestBuilderMakesADoneKeyWithTheDigests(t *testing.T) { // 1607/AC7
if err != nil {
t.Fatal(err)
}
if a := reg.answered(t, key, again); a.Refused || a.Do != versions.AgainRenumber || a.Version != "0.1.6" {
t.Errorf("builder_publish_again = %+v, want a new number, 0.1.6", a)
if a := reg.answered(t, key, again); !a.Refused {
t.Errorf("builder_publish_again = %+v, want a refusal", a)
}
// A numbered version a holder will refuse below, written to the state as the builder numbers one.
refusable, _ := json.Marshal(versions.Entry{Name: "netcheck/tools-go", Kind: versions.KindBundle, Series: "0.1", Version: "0.1.6", Module: "netcheck",
Fingerprint: "src2:" + strings.Repeat("6", 64), Seat: ociartifactregistry.Seat, State: versions.Numbered, At: time.Now().UTC()})
if _, err := reg.versions.Put(b.ctx, versions.Key("netcheck/tools-go", "0.1.6"), refusable); err != nil {
t.Fatal(err)
}
// Running-testing note (review round 1): two claims never get one number on the bus's own compare-and-set, as
// the builder's state versions keeps them: eight claims of one name at once, eight versions.
@@ -644,8 +609,8 @@ func TestBuilderMakesADoneKeyWithTheDigests(t *testing.T) { // 1607/AC7
if len(got) != 8 {
t.Errorf("eight claims of one name at once were numbered %v, want eight versions", slices.Sorted(maps.Keys(got)))
}
if got := reg.numberedOf("netcheck/tools-go"); len(got) != 2 || got[1] != "0.1.6" {
t.Errorf("version.numbered of netcheck/tools-go = %v, want 0.1.6 said after 0.1.5", got)
if got := reg.numberedOf("netcheck/tools-go"); len(got) != 0 {
t.Errorf("version.numbered of netcheck/tools-go said %v, want none", got)
}
// DLV-B10: a build module's bound running out fails the trunk build, which is the builder's finding build-failed; the
// fingerprint built again keeps its number, said with why, and its done build clears the finding.
@@ -683,33 +648,6 @@ func TestBuilderMakesADoneKeyWithTheDigests(t *testing.T) { // 1607/AC7
return e.Version == "0.1.6" && e.State == versions.Refused && e.Reason == "the cap" && e.RefusedDigest == digestOf("other")
})
})
// DLV-B6: a branch's end (the forge bridge's merged) asks the seat to remove each version tagged for it, the reason
// tag-gone, and the seat's removed turns it removed.
tagged := versions.Entry{Name: "mesh-sdk/go", Kind: versions.KindPackage, Series: "0.1", Version: "0.1.15-feat-x.1", Tag: "feat-x", Module: "mesh-sdk",
Fingerprint: "src2:" + strings.Repeat("5", 64), Repository: "novox/mesh-sdk", Seat: gopackageregistry.Seat, State: versions.Published, At: time.Now().UTC()}
raw, _ := json.Marshal(tagged)
if _, err := reg.versions.Put(b.ctx, versions.Key(tagged.Name, tagged.Version), raw); err != nil {
t.Fatal(err)
}
if _, err := b.js.Publish(b.ctx, process.ModuleStreamSubject(forgeBridgeModule, "forge"),
[]byte(`{"kind":"`+forgemsg.ForgeMerged+`","repository":"novox/mesh-sdk","branch":"feat/x"}`)); err != nil {
t.Fatal(err)
}
for deadline := time.Now().Add(20 * time.Second); ; time.Sleep(50 * time.Millisecond) {
reg.mu.Lock()
asked := slices.ContainsFunc(reg.removes, func(a map[string]any) bool {
return a["name"] == tagged.Name && a["version"] == tagged.Version && a["reason"] == "tag-gone"
})
reg.mu.Unlock()
if asked {
break
}
if time.Now().After(deadline) {
t.Fatalf("the seat's remove of %s@%s was not asked at the branch's end: %v", tagged.Name, tagged.Version, reg.removes)
}
}
reg.outcome(t, gopackageregistry.SubjectRemoved, versions.Outcome{Name: tagged.Name, Version: tagged.Version, Kind: versions.KindPackage, Holder: "gitea", Reason: "tag-gone"})
reg.awaitOf(t, tagged.Name, func(es []versions.Entry) bool { return len(es) == 1 && es[0].State == versions.Removed })
}
// TestAnArchiveOnlyModuleAndARootModuleAreBuilt: the indexer gives an archive no language, so modules.ca-trust has no
@@ -1180,8 +1118,7 @@ type registry struct {
b *bed
versions, findings, reasks jetstream.KeyValue
mu sync.Mutex
numbered []builds.VersionNumbered
removes []map[string]any
numbered []string // the bodies said on the subject version.numbered was said on before design 47 G §349
}
func (b *bed) registry(t *testing.T, node string, seatVersions ...string) *registry {
@@ -1209,27 +1146,12 @@ func (b *bed) registry(t *testing.T, node string, seatVersions ...string) *regis
t.Fatal(err)
}
}
// The Go seat's remove, as its holder answers it; each call is kept.
rm, err := b.js.Conn().Subscribe("mesh.seat."+gopackageregistry.Seat+".call.remove.>", func(m *nats.Msg) {
var args map[string]any
_ = json.Unmarshal(m.Data, &args)
r.mu.Lock()
r.removes = append(r.removes, args)
r.mu.Unlock()
_ = m.Respond([]byte(`{"result":{},"node":"ctl"}`))
})
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = rm.Unsubscribe() })
nc := b.js.Conn()
// As a build module hears it: on the contract's subject, read by the contract's decoder (BG2's late publish).
sub, err := nc.Subscribe(builds.VersionNumberedSubject, func(m *nats.Msg) {
if nb, err := builds.DecodeVersionNumbered(m.Data); err == nil {
r.mu.Lock()
r.numbered = append(r.numbered, nb)
r.mu.Unlock()
}
// Nothing says version.numbered any more (§349): its old subject is watched, to say so if something does.
sub, err := nc.Subscribe("mesh.mod."+builds.BuilderModule+".event.version.numbered", func(m *nats.Msg) {
r.mu.Lock()
r.numbered = append(r.numbered, string(m.Data))
r.mu.Unlock()
})
if err != nil {
t.Fatal(err)
@@ -1335,14 +1257,14 @@ func (r *registry) awaitFinding(t *testing.T, key string, raised bool) {
t.Fatalf("the finding %s raised = %v within 20 s, want %v", key, !raised, raised)
}
// numberedOf are the version.numbered the builder said for name, in order.
// numberedOf are the bodies said as version.numbered for name: none since design 47 G §349.
func (r *registry) numberedOf(name string) []string {
r.mu.Lock()
defer r.mu.Unlock()
var out []string
for _, nb := range r.numbered {
if nb.Name == name {
out = append(out, nb.Version)
for _, body := range r.numbered {
if strings.Contains(body, `"`+name+`"`) {
out = append(out, body)
}
}
return out
modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go
+9 -67
View File
@@ -117,13 +117,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
return fmt.Errorf("MESH_BUILDER_STEP_BOUND: %w", err)
}
}
// An unanswered tag ask is narrated every TagAskBound (design 51: five minutes); a test gives a shorter one.
tagAskBound := versions.TagAskBound
if v := strings.TrimSpace(getenv("MESH_BUILDER_TAG_ASK_BOUND")); v != "" {
if tagAskBound, err = time.ParseDuration(v); err != nil {
return fmt.Errorf("MESH_BUILDER_TAG_ASK_BOUND: %w", err)
}
}
tools := &buildrun.ToolchainSet{}
// Each build module's part of a build is an ask on its build seat's stream, pulled by a build node with a free slot
// (novox/hq ADR 0525): the builder chooses no node, so it reads neither presence nor have and holds no capacity.
@@ -152,7 +145,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
// The builder asks the controller, as its own account, whether the artifact store still holds a done key's blobs
// (the seat verb held, novox/hq issue 1937) before it leaves the key done, on its calls connection.
// The builder numbers each trunk build at its claim (novox/hq design 51 *The versions*): its state versions, bound once
// the bus holds it; the floor's three sources asked as its own account; each new number said as version.numbered.
// the bus holds it; the floor's three sources asked as its own account.
versionsKV := process.AwaitBucket(ctx, js, conn, versions.Bucket, "no trunk build is numbered: each is built unnumbered and published by its digest", logger.Printf)
found := &findings{kv: findingsKV, logf: logger.Printf, now: time.Now}
floorWaits := &versions.FloorWaits{Logf: logger.Printf, Said: found.floorSaid(context.WithoutCancel(ctx))}
@@ -164,12 +157,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
})
// A package is numbered under its name in its ecosystem, read from the indexer's packages state (design 32).
numberer := &versions.Numberer{Versions: versions.Versions{Store: awaitedVersions{b: versionsKV}}, Floor: floorVia(calls.Call), Waits: floorWaits,
Names: versions.NamesIn(indexerPackages{packages}),
Said: func(nb builds.VersionNumbered) {
if err := emitNumbered(context.WithoutCancel(ctx), js, cred, nb); err != nil {
logger.Printf("%v", err)
}
}}
Names: versions.NamesIn(indexerPackages{packages})}
// A publish that ends badly is a finding (design 51 *Health and conditions*): a holder's refused, a version no holder
// answered within its bound (a key of the state bounds with its own time to live), a seat's outcome that does not read.
pubBoundsKV := process.AwaitBucket(ctx, js, conn, versions.BoundsBucket, "a publish no holder answers, and a node holding the old form seven days, are not raised", logger.Printf)
@@ -193,15 +181,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
unput := &unputs{bounds: pubBoundsKV.KV, store: b.Claimer.Store, looking: b.Looking, findings: found,
bound: meshconditionsclient.CallUnansweredAfter, now: time.Now, logf: logger.Printf}
b.Left, b.Ended = unput.left, unput.ended
tagAsksKV := process.AwaitBucket(ctx, js, conn, versions.TagAsksBucket, "builder_tag and builder_untag are not answered", logger.Printf)
// A check's build of a head a kept tag ask names numbers its packages tagged (design 51 *Tagged versions*), and the ask
// is marked numbered with its versions.
b.Tags = tagsVia(tagAsksKV.KV, logger.Printf)
b.Numbers = build.VersionNumbers{N: numberer, TagNumbered: tagNumbered(tagAsksKV.KV, logger.Printf, time.Now)}
// An ask not answered within five minutes is narrated with what it waits on: its bound's end on the bus says when.
boundsKV := process.AwaitBucket(ctx, js, conn, versions.TagAskBoundsBucket, "an unanswered tag ask is not narrated", logger.Printf)
bounds := &tagBounds{asks: tagAsksKV.KV, bounds: boundsKV.KV, bound: tagAskBound, waits: floorWaits.Waiting, logf: logger.Printf, now: time.Now,
log: func(e builds.BuildLogEntry) { b.Log(e) }}
b.Numbers = build.VersionNumbers{N: numberer}
// The versions a build reserves at its start (design 47 G §369): its state reserved, and the catalog's versions-filled
// and version-released heard; nothing is reserved before the fill.
reservedKV := process.AwaitBucket(ctx, js, conn, versions.ReservedBucket, "no version is reserved: builds are made unreserved", logger.Printf)
@@ -210,7 +190,7 @@ 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, DropReserved: reserved.dropped, Done: doneVia(b), Reconsider: b.Reconsider, Remove: removeVia(calls.Call),
acts := versions.Acts{Numberer: numberer, DropReserved: reserved.dropped, Done: doneVia(b), Reconsider: b.Reconsider,
BuildAgain: func(ctx context.Context, module, fp, why, by string) (string, bool) {
r := b.BuildAgain(ctx, module, fp, why, by, buildAgainBound)
if r.Refused != "" {
@@ -393,23 +373,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
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).
tagAsksKV.Ready(func(kv jetstream.KeyValue) {
each := func(e jetstream.KeyValueEntry) {
tagAsked(ctx, acts, kv, e, logger.Printf)
if !opOf(e) {
bounds.disarm(ctx, e.Key())
} else if t, err := versions.DecodeTagAsk(e.Value()); err == nil {
bounds.asked(ctx, e.Key(), t)
}
}
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) }()
}
})
pubBoundsKV.Ready(func(kv jetstream.KeyValue) {
each := func(e jetstream.KeyValueEntry) {
if reserved.ended(ctx, e) {
@@ -435,14 +398,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
}
watchBounds(ctx, kv, each, found.notHeard(ctx), errs, logger.Printf)
})
boundsKV.Ready(func(kv jetstream.KeyValue) {
each := func(e jetstream.KeyValueEntry) { bounds.ended(ctx, e) }
if bw, err := settle(ctx, kv, each); err != nil {
process.NotWatched(logger.Printf, versions.TagAskBoundsBucket, err, "an unanswered tag ask is not narrated until the builder starts again")
} else {
go func() { errs <- follow(ctx, bw, 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) {
@@ -453,24 +408,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
go func() { errs <- follow(ctx, qw, eachRead) }()
}
})
// A branch's end removes its tagged versions (design 51 *Tagged versions*, DLV-B6): the forge bridge's merged, a pull's
// closed or branch-deleted on the builder's durable consumer of its stream; each forge message also wakes a floor whose
// bridge_tags was silent. The checker's tag-wanted is kept in tag-asks. A consumer not there yet is said.
ends := &branchEnds{acts: acts, findings: found, tagAsks: tagAsksKV.KV, logf: logger.Printf, now: time.Now,
wake: func(ctx context.Context) {
(&floorWake{b: b, waits: floorWaits, logf: logger.Printf}).wake(ctx, versions.WakeForge, "", "")
}}
for _, c := range []struct {
stream, what string
hear func(jetstream.Msg)
}{
{forgeStream, "a branch's end", func(m jetstream.Msg) { ends.heard(ctx, m) }},
{checkerAsksStream, "the checker's tag-wanted", func(m jetstream.Msg) {
tagWanted(ctx, tagAsksKV.KV(), m, checkerModule, time.Now(), logger.Printf)
}},
} {
hear(c.stream, c.what+" is not heard until the builder starts again", c.hear)
}
// Each forge message wakes a floor whose bridge_tags was silent (design 51 *The floor*). A consumer not there yet is said.
ends := &forgeWakes{logf: logger.Printf, wake: func(ctx context.Context) {
(&floorWake{b: b, waits: floorWaits, logf: logger.Printf}).wake(ctx, versions.WakeForge, "", "")
}}
hear(forgeStream, "the forge's messages are not heard until the builder starts again", func(m jetstream.Msg) { ends.heard(ctx, m) })
if stopped != nil {
return stopped
}
modules/mesh-delivery-builder/cmd/mesh-delivery-builder/refused_test.go
+4 -35
View File
@@ -29,7 +29,6 @@ import (
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds/forms"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg"
"mesh-delivery-builder/internal/build"
"mesh-delivery-builder/internal/versions"
@@ -314,46 +313,16 @@ func (noVersions) Update(context.Context, string, []byte, uint64) (uint64, error
func (noVersions) KeysOf(context.Context, string) ([]string, error) { return nil, nil }
func aBranchsEndIsHeard(t *testing.T) {
for _, c := range []struct {
f forgeEnd
want bool
}{
{forgeEnd{Kind: forgemsg.ForgeMerged, Repository: "o/r", Branch: "feat/x"}, true},
{forgeEnd{Kind: forgeBranchDeleted, Repository: "o/r", Branch: "feat/x"}, true},
{forgeEnd{Kind: forgemsg.ForgePull, Repository: "o/r", Branch: "feat/x", Action: forgemsg.PullClosed}, true},
{forgeEnd{Kind: forgemsg.ForgePull, Repository: "o/r", Branch: "feat/x", Action: forgemsg.PullOpened}, false},
{forgeEnd{Kind: forgemsg.ForgeMerged, Repository: "o/r"}, false},
} {
if got := c.f.ended(); got != c.want {
t.Errorf("%+v ended = %v, want %v", c.f, got, c.want)
}
}
// A forge message wakes a floor and is acknowledged; no tagged version is removed on a branch's end (design 47 G §373).
var logs bytes.Buffer
woken := 0
h := &branchEnds{acts: versions.Acts{Numberer: &versions.Numberer{Versions: versions.Versions{Store: noVersions{}}}}, findings: &findings{logf: t.Logf},
tagAsks: func() jetstream.KeyValue { return nil }, wake: func(context.Context) { woken++ }, logf: log.New(&logs, "", 0).Printf, now: time.Now}
h := &forgeWakes{wake: func(context.Context) { woken++ }, logf: log.New(&logs, "", 0).Printf}
open := &wakeMsg{subject: "mesh.mod.mesh-delivery-forge-bridge.event.forge.o.r.pull", data: []byte(`{"kind":"pull","repository":"o/r","branch":"feat/x","action":"opened"}`)}
h.heard(context.Background(), open)
merged := &wakeMsg{subject: "mesh.mod.mesh-delivery-forge-bridge.event.forge.o.r.merged", data: []byte(`{"kind":"merged","repository":"o/r","branch":"feat/x"}`)}
h.heard(context.Background(), merged)
if !open.acked || !merged.acked || woken != 2 {
t.Errorf("an open pull acked %v, a merge with nothing tagged acked %v, floors woken %d times", open.acked, merged.acked, woken)
}
// The checker's family holds its verdicts beside tag-wanted: a verdict is acknowledged and not kept; a tag-wanted with
// tag-asks not bound is heard again.
verdict := &wakeMsg{subject: "mesh.stream.mesh-delivery-checker.event.verdict", data: []byte(`{"repository":"o/r"}`)}
wanted := &wakeMsg{subject: tagWantedSubject, data: []byte(`{"repository":"o/r","branch":"feat/x","head":"abc","tag":"feat-x"}`)}
for _, m := range []*wakeMsg{verdict, wanted} {
tagWanted(context.Background(), nil, m, checkerModule, time.Now(), t.Logf)
}
if !verdict.acked || verdict.delayed != 0 || wanted.acked || wanted.delayed != versions.RemoveAgainAfter || !strings.HasPrefix(tagWantedSubject, "mesh.stream.") {
t.Errorf("a verdict acked %v (delayed %v), a tag-wanted acked %v (delayed %v)", verdict.acked, verdict.delayed, wanted.acked, wanted.delayed)
}
unreadable := &wakeMsg{subject: merged.subject, data: merged.data}
h.acts.Numberer.Versions.Store = failingVersions{}
h.heard(context.Background(), unreadable)
if unreadable.acked || unreadable.delayed != versions.RemoveAgainAfter {
t.Errorf("a branch's end the versions cannot answer: acked %v, given again after %v", unreadable.acked, unreadable.delayed)
if !open.acked || !merged.acked || woken != 2 || open.delayed != 0 || merged.delayed != 0 {
t.Errorf("an open pull acked %v, a merge acked %v, floors woken %d times", open.acked, merged.acked, woken)
}
}
modules/mesh-delivery-builder/cmd/mesh-delivery-builder/tagbounds.go
-132
View File
@@ -1,132 +0,0 @@
package main
import (
"context"
"encoding/json"
"errors"
"time"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
"mesh-delivery-builder/internal/versions"
)
// tagBounds narrates a tag ask the builder has not answered within five minutes, with what it waits on (design 51
// *Tagged versions*): each unanswered ask has a key in the state tag-ask-bounds written with its own time to live, and
// the bus's marker of its end, heard on the watch, is what narrates it and writes the bound again. No timer.
type tagBounds struct {
asks func() jetstream.KeyValue
bounds func() jetstream.KeyValue
waits func() []versions.FloorWait
logf func(string, ...any)
// log writes a narration into the builder's build log (build.said); nil writes none.
log func(builds.BuildLogEntry)
now func() time.Time
bound time.Duration
}
// arm writes key's bound when none stands: a Create, so a standing bound keeps its time and a restart arms each
// unanswered ask whose bound ended while the builder was away.
func (tb *tagBounds) arm(ctx context.Context, key string) {
kv := tb.bounds()
if kv == nil {
return
}
if _, err := kv.Create(ctx, key, []byte(`{}`), jetstream.KeyTTL(tb.bound)); err != nil && !errors.Is(err, jetstream.ErrKeyExists) {
tb.logf("the bound of the tag ask %s is not written, so it is not narrated past five minutes: %v", key, err)
}
}
// disarm removes key's bound: the ask was answered or ended.
func (tb *tagBounds) disarm(ctx context.Context, key string) {
if kv := tb.bounds(); kv != nil {
_ = kv.Delete(ctx, key)
}
}
// asked is told each value of tag-asks: an unanswered ask is armed, any other disarmed.
func (tb *tagBounds) asked(ctx context.Context, key string, t versions.TagAsk) {
if t.Unanswered() {
tb.arm(ctx, key)
return
}
tb.disarm(ctx, key)
}
// ended is told each change of tag-ask-bounds: a bound whose time ended (the bus's marker, a purge on the watch) on an
// ask still unanswered narrates it and arms it again.
func (tb *tagBounds) ended(ctx context.Context, e jetstream.KeyValueEntry) {
if e.Operation() != jetstream.KeyValuePurge {
return
}
kv := tb.asks()
if kv == nil {
return
}
got, err := kv.Get(ctx, e.Key())
if err != nil {
return // the ask ended with its branch
}
t, err := versions.DecodeTagAsk(got.Value())
if err != nil || !t.Unanswered() {
return
}
var waits []versions.FloorWait
if tb.waits != nil {
waits = tb.waits()
}
line := versions.NarrateUnanswered(t, waits, tb.now())
tb.logf("%s", line)
// In the build log too (design 51 *Tagged versions*): under the builder's own key, no build being claimed yet, the
// ask's key standing for the fingerprint.
if tb.log != nil {
tb.log(builds.BuildLogEntry{Event: builds.BuildSaidEvent, Module: builds.BuilderModule, FP: e.Key(), Commit: t.Head,
At: tb.now().UTC(), Line: line, Why: versions.SaidUnanswered})
}
tb.arm(ctx, e.Key())
}
// tagNumbered marks the tag ask of a check's build's head numbered with its versions in tag-asks, by compare-and-set.
func tagNumbered(kv func() jetstream.KeyValue, logf func(string, ...any), now func() time.Time) func(context.Context, versions.BuildClaim, versions.Numbering) {
return func(ctx context.Context, c versions.BuildClaim, got versions.Numbering) {
store := kv()
if store == nil {
return
}
lister, err := store.ListKeysFiltered(ctx, versions.EscapeToken(c.Repository)+".>")
if err != nil {
logf("the tag ask of %s at %s is not marked numbered: %v", c.Repository, c.Head, err)
return
}
for key := range lister.Keys() {
for range 3 {
e, err := store.Get(ctx, key)
if err != nil {
break
}
t, err := versions.DecodeTagAsk(e.Value())
if err != nil {
break
}
t, changed := t.WithNumbered(c.Module, c.Head, got.Versions, got.Names, now())
if !changed {
break
}
data, err := json.Marshal(t)
if err != nil {
break
}
if _, err = store.Update(ctx, key, data, e.Revision()); err == nil {
logf("%s: tag ask of %s %s: %s", key, t.Repository, t.Branch, t.Answer)
break
}
if !errors.Is(err, jetstream.ErrKeyExists) && !errors.Is(err, jetstream.ErrKeyRevisionMismatch) {
logf("the tag ask %s is not marked numbered: %v", key, err)
break
}
}
}
}
}
modules/mesh-delivery-builder/cmd/mesh-delivery-builder/versions.go
-55
View File
@@ -13,7 +13,6 @@ import (
"git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/ociartifactregistry"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
"git.novox.be/novox/mesh-sdk/go/process"
@@ -208,19 +207,6 @@ 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 builds.VersionNumbered) error {
id := "numbered-" + versions.Key(b.Name, b.Version)
if b.Again != "" {
id += "-again-" + b.Again
}
if err := emitEvent(ctx, js, cred, builds.VersionNumberedEvent, id, b); err != nil {
return fmt.Errorf("saying %s for %s@%s: %w", builds.VersionNumberedEvent, b.Name, b.Version, err)
}
return nil
}
// emitEvent says one of the builder's events, mesh.mod.<module>.event.<event>, with the SDK's envelope headers and id as
// the bus's dedup id.
func emitEvent(ctx context.Context, js jetstream.JetStream, cred process.Account, event, id string, body any) error {
@@ -320,44 +306,3 @@ func doneVia(b *build.Builder) func(ctx context.Context, module, fp string) (boo
return cur.State == builds.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, or answers kept when another tag names the
// digest (the registry part's RemoveAnswer: kept, reason, tags), which ends the remove.
func removeVia(call toolCaller) func(ctx context.Context, seat string, e versions.Entry) (versions.RemoveOutcome, error) {
return func(_ context.Context, seat string, e versions.Entry) (versions.RemoveOutcome, error) {
raw, err := call("seat:"+seat+".remove", map[string]any{"name": e.Name, "version": e.Version, "tag": e.Tag, "reason": "tag-gone"})
if err != nil {
return versions.RemoveOutcome{}, err
}
var a ociartifactregistry.RemoveAnswer // the four registry seats share the part's answer
if len(raw) > 0 && json.Unmarshal(raw, &a) != nil {
return versions.RemoveOutcome{}, fmt.Errorf("%s's remove answered what is no answer of it: %s", seat, raw)
}
return versions.RemoveOutcome{Kept: a.Kept, Reason: a.Reason, Tags: a.Tags}, nil
}
}
// 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)
}
}
modules/mesh-delivery-builder/internal/build/numbers.go
+8 -27
View File
@@ -18,18 +18,14 @@ import (
// Numbers numbers a trunk build at its claim (novox/hq design 51 *The versions*, ADR 0606, ADR 0617 §1) and records the
// digests it was built with. Nil numbers nothing.
type Numbers interface {
// tag is a check's build's tag (TagOf), empty for a trunk build; done the digests of its key when it is done already
// (versions.BuildClaim.Done), nil otherwise.
Number(ctx context.Context, module string, m meshcatalog.ModulesByModuleValue, plan Plan, tag string, done map[string]string) (versions.Numbering, error)
// done is the digests of the build's key when it is done already (versions.BuildClaim.Done), nil otherwise.
Number(ctx context.Context, module string, m meshcatalog.ModulesByModuleValue, plan Plan, done map[string]string) (versions.Numbering, error)
Built(ctx context.Context, module string, m meshcatalog.ModulesByModuleValue, plan Plan, got versions.Numbering, made []builds.BuiltArtifact) error
}
// VersionNumbers is Numbers over the builder's state versions.
type VersionNumbers struct {
N *versions.Numberer
// TagNumbered is told the versions a check's build of a tag ask's head was numbered (the ask is then marked numbered
// in tag-asks, design 51 *Tagged versions*); nil tells nobody.
TagNumbered func(ctx context.Context, c versions.BuildClaim, got versions.Numbering)
}
// ClaimOf is the claim a module value's trunk build is numbered by: a package per source revision (Src), the rest per
@@ -46,18 +42,10 @@ func ClaimOf(module string, m meshcatalog.ModulesByModuleValue, plan Plan) versi
return c
}
func (v VersionNumbers) Number(ctx context.Context, module string, m meshcatalog.ModulesByModuleValue, plan Plan, tag string, done map[string]string) (versions.Numbering, error) {
func (v VersionNumbers) Number(ctx context.Context, module string, m meshcatalog.ModulesByModuleValue, plan Plan, done map[string]string) (versions.Numbering, error) {
c := ClaimOf(module, m, plan)
c.Done = done
if tag != "" {
// A check's build of a head a tag ask names: its packages tagged, below the next version (design 51 *Tagged versions*).
c.Tag, c.Head, c.Revision, c.Merge = tag, m.Commit, "", 0
}
got, err := v.N.Number(ctx, c)
if err == nil && tag != "" && len(got.Versions) > 0 && v.TagNumbered != nil {
v.TagNumbered(ctx, c, got)
}
return got, err
return v.N.Number(ctx, c)
}
func (v VersionNumbers) Built(ctx context.Context, module string, m meshcatalog.ModulesByModuleValue, plan Plan, got versions.Numbering, made []builds.BuiltArtifact) error {
@@ -68,22 +56,15 @@ func (v VersionNumbers) Built(ctx context.Context, module string, m meshcatalog.
return v.N.Built(ctx, ClaimOf(module, m, plan), got, digests)
}
// TagOf answers the cleaned tag a kept tag ask names for a repository's head (the builder's state tag-asks, design 51
// *Tagged versions*): a check's build of that head numbers its packages tagged at its claim.
type TagOf func(ctx context.Context, repository, head string) (tag string, ok bool)
// number numbers a build before its claim, a trunk build's artifacts or a check's build's packages under its tag: false
// number numbers a trunk build's artifacts before its claim: false
// when it waits for its floor (a package, design 51 *The floor*), and then nothing is claimed; the floor's wake looks at
// the module again.
func (b *Builder) number(ctx context.Context, module string, m meshcatalog.ModulesByModuleValue, plan Plan, tag string) (versions.Numbering, bool) {
func (b *Builder) number(ctx context.Context, module string, m meshcatalog.ModulesByModuleValue, plan Plan) (versions.Numbering, bool) {
if b.Numbers == nil {
return versions.Numbering{}, true
}
var done map[string]string
if tag == "" {
done = b.doneDigests(ctx, module, m.Fp2)
}
got, err := b.Numbers.Number(ctx, module, m, plan, tag, done)
done := b.doneDigests(ctx, module, m.Fp2)
got, err := b.Numbers.Number(ctx, module, m, plan, done)
for _, s := range got.Said {
b.logf("%s at %s: %s", module, m.Fp2, s.Line)
// And in the build log (design 51 *The versions*: it says a version was kept, and why), between the build's start and end.
modules/mesh-delivery-builder/internal/build/run.go
+3 -9
View File
@@ -97,8 +97,6 @@ type Builder struct {
Numbers Numbers
// Reserved reserves the version of a trunk build at its start (design 47 G §369); nil reserves nothing.
Reserved Reservations
// Tags answers a check's build's tag from the kept tag asks; nil numbers no check's build.
Tags TagOf
// Forms answers what each node's engine states it reads (ADR 0618 §2, ADR 0624 §1): present, the nodes whose engine
// answered when last asked, and counted, every node the assignments names, present or away, a node with no statement
// stating nothing (the baseline's first forms). A done key names each artifact in the forms these keep, the oldest in
@@ -485,17 +483,13 @@ func (b *Builder) try(ctx context.Context, module string, m meshcatalog.ModulesB
byHand := b.hands[key] != nil
b.mu.Unlock()
var numbering versions.Numbering
tag := ""
if !trunk && b.Tags != nil && b.refOf(module, m.Fp2) != "" {
tag, _ = b.Tags(ctx, repotrunk.RepositoryName(m.Repository), m.Commit)
}
if (trunk || tag != "") && planErr == nil && !nothing {
if trunk && planErr == nil && !nothing {
var goOn bool
if numbering, goOn = b.number(ctx, module, m, plan, tag); !goOn {
if numbering, goOn = b.number(ctx, module, m, plan); !goOn {
return
}
}
reserving := trunk && tag == "" && planErr == nil && !nothing && b.reserves(ctx, module, m, plan)
reserving := trunk && planErr == nil && !nothing && b.reserves(ctx, module, m, plan)
if reserving && b.waitFill(module, m) {
return
}
modules/mesh-delivery-builder/internal/build/run_test.go
+4 -11
View File
@@ -434,18 +434,13 @@ func TestACandidatesImageCallCarriesNoPublishAndItsKeyNoReference(t *testing.T)
r := newRig(t, imageArt)
r.puller = "ace"
trunkModule(t, r)
// DLV-B6: a check's build of a head a kept tag ask names is numbered under that tag at its claim.
// Design 47 G §373: a candidate's build is numbered under no tag, so it numbers nothing.
nums := &fakeNumbers{}
var tagAsked []string
r.b.Numbers = nums
r.b.Tags = func(_ context.Context, repository, head string) (string, bool) {
tagAsked = append(tagAsked, repository+"@"+head)
return "feat-x", head == headX
}
r.b.Candidate(context.Background(), "feat%2Fx", candidateValue(t, "feat/x", 7, headX))
r.b.Wait()
if !slices.Equal(nums.tags, []string{"feat-x"}) || len(tagAsked) != 1 || !strings.HasSuffix(tagAsked[0], "@"+headX) {
t.Errorf("the check's build numbered with tags %v (asked %v), want feat-x for its head", nums.tags, tagAsked)
if nums.numbered != 0 {
t.Errorf("a candidate's build was numbered %d times, want none", nums.numbered)
}
v := valueOf(t, r.st, r.key)
if v.State != builds.BuildDone || len(v.Artifacts) != 1 || v.Artifacts[0].Reference != "" {
@@ -514,13 +509,11 @@ type fakeNumbers struct {
versions map[string]string
numbered int
built map[string]string
tags []string
done map[string]string // the done key's digests the last look numbered with
refused string // answered as Numbering.Refused
}
func (f *fakeNumbers) Number(_ context.Context, _ string, _ meshcatalog.ModulesByModuleValue, _ Plan, tag string, done map[string]string) (versions.Numbering, error) {
f.tags = append(f.tags, tag)
func (f *fakeNumbers) Number(_ context.Context, _ string, _ meshcatalog.ModulesByModuleValue, _ Plan, done map[string]string) (versions.Numbering, error) {
f.done = done
f.numbered++
return versions.Numbering{Wait: f.wait, Versions: f.versions, Refused: f.refused}, nil
modules/mesh-delivery-builder/internal/versions/again.go
+6 -218
View File
@@ -2,19 +2,15 @@ package versions
import (
"context"
"encoding/json"
"errors"
"fmt"
"sort"
"strings"
"time"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
)
// 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 is a held build whose recorded version is published again: refused since design 47 G §349 (no version.numbered is said).
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).
@@ -51,141 +47,15 @@ func PublishAgain(entries []Entry, fingerprint string, built bool) (Again, error
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{Do: AgainPublish, Entry: e, Why: fmt.Sprintf("%s@%s (%s) is held", 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: <owner>/<name>")
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 && e.State != Kept {
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 " +
"nothing says a version numbered since design 47 G §349, so a held build is refused; 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."
@@ -193,15 +63,6 @@ const (
"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-<tag>.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).
@@ -213,66 +74,12 @@ type Acts struct {
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: what it answered, kept or asked.
Remove func(ctx context.Context, seat string, e Entry) (RemoveOutcome, error)
// DropOldForm drops a retired node from the old form's count by hand (builder_old_form): its answer, or why not.
DropOldForm func(ctx context.Context, node, by, why string, at time.Time) (string, error)
// DropReserved drops a module's reservation of one MAJOR.MINOR by hand (builder_reserved_drop): its answer, or why not.
DropReserved func(ctx context.Context, module, series, by, why string, at time.Time) (string, 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 " + "<next>-" + t.Tag + ".<n> 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 left, ok := a.removeOne(ctx, e); ok {
delete(t.Left, at)
} else if left != "" {
t.Left[at] = left
}
}
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
}
// removeOne asks e's seat to remove it: done when the seat kept it whole (recorded Kept, no retry); otherwise left, with
// the error of a call that failed.
func (a Acts) removeOne(ctx context.Context, e Entry) (left string, done bool) {
got, err := a.Remove(ctx, e.Seat, e)
if err != nil {
return e.Seat + ": " + err.Error(), false
}
if !got.Kept {
return "", false
}
if err := a.Numberer.keep(ctx, e, got.Reason, got.Tags); err != nil {
return e.Seat + ": kept, and not recorded so: " + err.Error(), false
}
return "", true
}
// 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, "/")
@@ -288,6 +95,9 @@ func (a Acts) PublishAgain(ctx context.Context, name, fp, why, by string, at tim
if err != nil {
return Again{}, "", err
}
if ag.Do == AgainPublish || (ag.Do == AgainRenumber && done) {
return Again{}, "", errors.New("nothing says a version numbered since design 47 G §349: the version is published under its reservation (§369)")
}
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,
@@ -305,14 +115,6 @@ func (a Acts) PublishAgain(ctx context.Context, name, fp, why, by string, at tim
}
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 += "; " + builds.VersionNumberedEvent + " said for " + ag.Entry.Version
}
return ag, answer, nil
}
@@ -337,17 +139,3 @@ func (a Acts) FloorAgain(ctx context.Context, name string) (string, error) {
a.Reconsider(ctx)
return answer + "; builder_versions shows its version once numbered", nil
}
// TagFor is the cleaned tag a kept tag ask names for repository's head: the ask a check's build of that head is
// numbered under (design 51 *Tagged versions*). An untag, or an ask whose branch ended, names none.
func TagFor(asks []TagAsk, repository, head string) (string, bool) {
for _, t := range asks {
if t.Untag || t.State == TagRemoving || t.State == TagRemoved || t.Head == "" {
continue
}
if t.Repository == repository && t.Head == head {
return t.Tag, t.Tag != ""
}
}
return "", false
}
modules/mesh-delivery-builder/internal/versions/branchend.go
-92
View File
@@ -1,92 +0,0 @@
package versions
import (
"context"
"fmt"
"sort"
"strings"
"time"
)
// A branch's end removes its tagged versions (design 51 *Tagged versions*, DLV-B6): on the forge bridge's `merged`, a
// pull's `closed` or `branch-deleted`, heard on the builder's durable consumer of the stream `forge`, the builder calls
// each seat's `remove` and acknowledges the message only once every seat said `removed` (Heard marks the entry); until
// then the bus gives it again each minute (RemoveAgainAfter) and, past RemovalFindingAfter, the builder's finding names
// the seats still owing.
const (
RemoveAgainAfter = time.Minute
RemovalFindingAfter = 15 * time.Minute
// RemovalNotDone is the finding's kind.
RemovalNotDone = "tag-removal-not-done"
)
// BranchEnd is what one hearing of a branch's end did: the versions still not removed (name@version to its seat, with
// the error of a remove that failed), and whether all are gone so the message is acknowledged.
type BranchEnd struct {
Left map[string]string
Done bool
}
// Seats are the seats still owing a removal, sorted, each once.
func (b BranchEnd) Seats() []string {
seen := map[string]bool{}
var out []string
for _, s := range b.Left {
seat, _, _ := strings.Cut(s, ":")
if !seen[seat] {
seen[seat] = true
out = append(out, seat)
}
}
sort.Strings(out)
return out
}
// Finding is the removal finding's words: what is left, since when.
func (b BranchEnd) Finding(repository, branch string, since time.Time) string {
var at []string
for v, s := range b.Left {
at = append(at, v+" on "+s)
}
sort.Strings(at)
return fmt.Sprintf("the tagged versions of %s %s are not all removed since its end at %s: %s; the builder asks remove again each minute",
repository, branch, since.UTC().Format(time.RFC3339), strings.Join(at, ", "))
}
// BranchEnded removes repository's tagged versions of branch: every one not yet removed is asked of its seat again.
// A store that cannot be read is the error: nothing is acknowledged then.
func (a Acts) BranchEnded(ctx context.Context, repository, branch string) (BranchEnd, error) {
all, err := a.Numberer.Versions.All(ctx)
if err != nil {
return BranchEnd{}, err
}
left := map[string]string{}
for _, e := range Untagged(all, repository, branch) {
at := e.Name + "@" + e.Version
left[at] = e.Seat
if a.Remove == nil {
continue
}
// A seat that kept it whole (another tag names its digest) ends it here: recorded kept, never asked again.
if l, done := a.removeOne(ctx, e); done {
delete(left, at)
} else if l != "" {
left[at] = l
}
}
return BranchEnd{Left: left, Done: len(left) == 0}, nil
}
// TagWanted is the body of the checker's `tag-wanted` (design 51 *Tagged versions*): a group member's head to number
// tagged versions for.
type TagWanted struct {
Repository string `json:"repository"`
Branch string `json:"branch"`
Head string `json:"head"`
Tag string `json:"tag,omitempty"`
}
// TagWantedAsk is the tag-asks value a tag-wanted becomes, as builder_tag writes it by hand: by the checker.
func TagWantedAsk(t TagWanted, by string, at time.Time) (string, []byte, error) {
return NewTagAsk(t.Repository, t.Branch, t.Head, "the checker's tag-wanted for a group member's check", by, at)
}
modules/mesh-delivery-builder/internal/versions/claim.go
+4 -16
View File
@@ -6,8 +6,6 @@ import (
"sort"
"strings"
"time"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
)
// The registry seats a version is published to (design 51 *The seats*), by what is published.
@@ -69,9 +67,6 @@ type BuildClaim struct {
Revision string
Merge uint64
Artifacts []ClaimArtifact
// Tag numbers a check's build of a head a tag ask names (design 51 *Tagged versions*): its packages only, each a
// tagged version kept with Repository for its branch's end; Head is the head, said in numbered. Empty for a trunk build.
Tag, Head string
// Done are the digests of the build's done key, by artifact, when the claim is a look at a build already done: what it
// left unnumbered (a package whose name was not known yet, a floor that was silent) is numbered with them, so
// version.numbered carries the digest its build module publishes late. Nil for a build still to make.
@@ -114,7 +109,6 @@ const (
SaidKept = "kept"
SaidWaiting = "waiting"
SaidUnnumbered = "unnumbered"
SaidUnanswered = "tag ask unanswered"
SaidOldFormClosed = "old-form-closed"
)
@@ -133,16 +127,13 @@ func narrationOf(name string, d Decision) Narration {
}
// Numberer numbers trunk builds at their claim (design 51 *The versions*, DLV-B2, B3): it asks a name's floor only
// when versions holds nothing for it, records a silent floor as a wait (FloorWaits, woken by its source's event), and
// says each new number (Said, the event version.numbered).
// when versions holds nothing for it, and records a silent floor as a wait (FloorWaits, woken by its source's event).
type Numberer struct {
Versions Versions
Floor FloorSources
Waits *FloorWaits
// Holder is the module holding a seat, whose health condition's clear wakes a silent seat source; nil names none.
Holder func(seat string) string
// Said is told each version newly numbered; nil tells nobody.
Said func(builds.VersionNumbered)
// Names answers a package's name in its ecosystem; nil numbers a package as NameOf names it.
Names PackageNames
// Publishing is told each version Built recorded that is still numbered, its holder's word not heard yet: the
@@ -159,7 +150,7 @@ func (n *Numberer) Number(ctx context.Context, c BuildClaim) (Numbering, error)
sort.Slice(arts, func(i, j int) bool { return arts[i].Artifact < arts[j].Artifact })
for _, a := range arts {
seat, kind, ok := SeatOf(a.Kind, a.Language)
if !ok || (c.Tag != "" && kind != KindPackage) {
if !ok {
continue
}
name := NameOf(c.Module, a.Artifact)
@@ -194,8 +185,8 @@ func (n *Numberer) Number(ctx context.Context, c BuildClaim) (Numbering, error)
if len(entries) == 0 {
floor, why = n.askFloor(ctx, seat, name, c)
}
d, e, err := n.Versions.Number(ctx, Claim{Name: name, Kind: kind, Series: c.Series, Module: c.Module, Fingerprint: c.Fingerprint,
Commit: c.Commit, Revision: c.Revision, Digest: c.Done[a.Artifact], Seat: seat, Merge: c.Merge, Floor: floor, Tag: c.Tag, Repository: c.Repository})
d, _, err := n.Versions.Number(ctx, Claim{Name: name, Kind: kind, Series: c.Series, Module: c.Module, Fingerprint: c.Fingerprint,
Commit: c.Commit, Revision: c.Revision, Digest: c.Done[a.Artifact], Seat: seat, Merge: c.Merge, Floor: floor})
if err != nil {
return out, err
}
@@ -214,9 +205,6 @@ func (n *Numberer) Number(ctx context.Context, c BuildClaim) (Numbering, error)
n.Waits.Answered(name)
}
out.Versions[a.Artifact] = d.Version
if !d.Keep && n.Said != nil {
n.Said(NumberedOf(e, c.Head))
}
}
return out, nil
}
modules/mesh-delivery-builder/internal/versions/heard.go
-19
View File
@@ -27,25 +27,6 @@ type Outcome struct {
Tags []string `json:"tags,omitempty"`
}
// RemoveOutcome is what a seat's remove answered: kept whole (another tag names its digest), with why and those tags; or
// asked, the seat saying removed on its events.
type RemoveOutcome struct {
Kept bool
Reason string
Tags []string
}
// keep marks a tagged version kept by its seat, so it is asked no more.
func (n *Numberer) keep(ctx context.Context, e Entry, reason string, tags []string) error {
if reason == "" {
reason = "shared-digest"
}
_, err := n.Versions.Mark(ctx, e.Name, e.Version, func(x *Entry) {
x.State, x.Reason, x.KeptWith = Kept, reason, append([]string(nil), tags...)
})
return err
}
// Heard takes one seat event in: a numbered version's entry turns published (with its digest when it held none),
// refused with the holder's reason, removed with its reason and the event's id, or kept (another tag names its digest)
// with the other tags; a by-digest publish (no version)
modules/mesh-delivery-builder/internal/versions/store.go
+5 -28
View File
@@ -55,12 +55,8 @@ 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
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
@@ -78,23 +74,7 @@ func (v Versions) Number(ctx context.Context, c Claim) (Decision, Entry, error)
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}
for _, e := range entries {
// The same fingerprint claimed again for its tag (a check's build looked at again) keeps its number.
if e.Tag == CleanTag(c.Tag) && e.Fingerprint == c.Fingerprint && e.State != Removed {
d = Decision{Version: e.Version, Keep: true, Publish: true, Why: "tagged " + e.Version + ", kept for this fingerprint"}
break
}
}
} else {
d = Decide(fresh(entries, c), 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
}
@@ -105,12 +85,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,
e := Entry{Name: c.Name, Kind: c.Kind, Series: c.Series, Version: d.Version, 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
@@ -205,7 +182,7 @@ func fresh(entries []Entry, c Claim) []Entry {
return out
}
// All is every entry of every name, by name in order: what builder_untag and the tag asks read.
// All is every entry of every name, by name in order: what the versions' readers read.
func (v Versions) All(ctx context.Context) ([]Entry, error) {
keys, err := v.Store.KeysOf(ctx, AllFilter)
if err != nil {
modules/mesh-delivery-builder/internal/versions/tagbound.go
-75
View File
@@ -1,75 +0,0 @@
package versions
import (
"fmt"
"maps"
"slices"
"strings"
"time"
)
// The bound of an unanswered tag ask (design 51 *Tagged versions*: "an ask the builder has not answered within five
// minutes is narrated in the build log with what it waits on"): a key of the state tag-ask-bounds, written with its own
// time to live, whose expiry the builder hears on its watch of the state. No timer: the bus ends the bound.
const (
TagAskBoundsState = "tag-ask-bounds"
TagAskBoundsBucket = "mesh-delivery-builder_tag-ask-bounds"
TagAskBound = 5 * time.Minute
)
// Unanswered says whether a tag ask still waits for its versions: kept, not an untag, none numbered yet.
func (t TagAsk) Unanswered() bool {
return !t.Untag && t.State == TagAsked && len(t.Versions) == 0
}
// WithNumbered is the tag ask once a check's build of its head numbered module's packages under its tag: numbered, each
// version by the name it was numbered under (names, by artifact; NameOf where it names none), the answer naming them. An
// untag, or an ask of another head, is left as it is.
func (t TagAsk) WithNumbered(module, head string, got, names map[string]string, at time.Time) (TagAsk, bool) {
if t.Untag || t.Head != head || len(got) == 0 || t.State == TagRemoving || t.State == TagRemoved {
return t, false
}
if t.Versions == nil {
t.Versions = map[string]string{}
}
changed := false
for artifact, v := range got {
name := Numbering{Names: names}.NameOf(module, artifact)
if t.Versions[name] != v {
t.Versions[name], changed = v, true
}
}
if !changed && t.State == TagNumbered {
return t, false
}
var said []string
for _, name := range slices.Sorted(maps.Keys(t.Versions)) {
said = append(said, name+" "+t.Versions[name])
}
t.State, t.Answered = TagNumbered, at.UTC()
t.Answer = "numbered: " + strings.Join(said, ", ")
return t, true
}
// NarrateUnanswered is the line said for a tag ask its bound found unanswered: since when, and what it waits on, the
// names waiting on their floor named with their silent sources.
func NarrateUnanswered(t TagAsk, waits []FloorWait, now time.Time) string {
on := "the check's build of " + short(t.Head) + " is not claimed yet (the checker builds a group member's head once it asks)"
var floors []string
for _, w := range waits {
floors = append(floors, w.Name+" waits on its floor ("+strings.Join(w.Silent, ", ")+" silent)")
}
if len(floors) > 0 {
slices.Sort(floors)
on += "; " + strings.Join(floors, "; ")
}
return fmt.Sprintf("the tag ask of %s %s at %s (%s) is not answered after %s: it waits on %s",
t.Repository, t.Branch, short(t.Head), t.Tag, now.Sub(t.At).Round(time.Second), on)
}
func short(head string) string {
if len(head) > 12 {
return head[:12]
}
return head
}
modules/mesh-delivery-builder/internal/versions/tools.go
+1 -12
View File
@@ -6,8 +6,6 @@ import (
"errors"
"fmt"
"strings"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
)
// The builder's tools of its versions (design 51 *Events of the builder*, *Tools of the builder*).
@@ -15,21 +13,12 @@ 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"
// OldFormClosedEvent is, with builds.VersionNumberedEvent, the builder's own family `version`, kept seven days,
// OldFormClosedEvent is 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.
OldFormClosedEvent = "version.old-form-closed"
)
// NumberedOf is the version.numbered body of an entry (builds.VersionNumbered, the contract a build module reads).
func NumberedOf(e Entry, head string) builds.VersionNumbered {
return builds.VersionNumbered{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)
modules/mesh-delivery-builder/internal/versions/versions.go
-41
View File
@@ -302,47 +302,6 @@ func Decide(entries []Entry, kind, series, fingerprint, revision string, floor F
return Decision{Version: v, Publish: true, Why: "numbered " + v}
}
// CleanTag makes a branch or tag name a semver prerelease identifier: lower case, every run of characters other than
// letters and digits one '-', none at either end; "tag" when nothing is left.
func CleanTag(tag string) string {
var b strings.Builder
dash := false
for _, r := range strings.ToLower(tag) {
if (r >= 'a' && r <= 'z') || (r >= '0' && r <= '9') {
b.WriteRune(r)
dash = false
continue
}
if !dash && b.Len() > 0 {
b.WriteByte('-')
dash = true
}
}
out := strings.TrimSuffix(b.String(), "-")
if out == "" {
return "tag"
}
return out
}
// Tagged is the next tagged version of name for tag: below the next trunk version, never the latest
// (`0.1.15-feat-x.2`, design 51 *Tagged versions*); a package new in a group is numbered from its floor
// (`0.1.0-feat-x.1`).
func Tagged(entries []Entry, series, tag string, floor Floor) string {
base := Next(entries, series, floor)
tag = CleanTag(tag)
n := 0
prefix := base + "-" + tag + "."
for _, e := range entries {
if rest, ok := strings.CutPrefix(e.Version, prefix); ok {
if k, err := strconv.Atoi(rest); err == nil && k > n {
n = k
}
}
}
return fmt.Sprintf("%s%d", prefix, n+1)
}
// Fill is whether a bucket holding bytes of cap is past its finding's mark.
func Fill(bytes, capacity uint64) (share float64, full bool) {
if capacity == 0 {
modules/mesh-delivery-builder/internal/versions/versions_test.go
+18 -207
View File
@@ -12,7 +12,6 @@ import (
"testing"
"time"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds/forms"
)
@@ -241,18 +240,6 @@ func TestTheBuildersVersions(t *testing.T) {
t.Error("builder_versions with no name answered")
}
})
t.Run("tagged", func(t *testing.T) {
es := []Entry{{Version: "0.1.14", Series: "0.1"}, {Version: "0.1.15-feat-x.1", Series: "0.1", Tag: "feat-x"}}
if got := Tagged(es, "0.1", "feat/X", known(nil, nil, nil)); got != "0.1.15-feat-x.2" {
t.Errorf("Tagged = %s, want 0.1.15-feat-x.2", got)
}
if got := Tagged(nil, "0.1", "feat-x", known(nil, nil, nil)); got != "0.1.0-feat-x.1" {
t.Errorf("a package new in a group = %s, want 0.1.0-feat-x.1", got)
}
if got := CleanTag("--Fix/ä_1965--"); got != "fix-1965" {
t.Errorf("CleanTag = %s", got)
}
})
t.Run("the floor wakes on its silent source, never a timer", func(t *testing.T) {
ctx := context.Background()
src := FloorSources{
@@ -400,17 +387,16 @@ func TestTheBuildersVersions(t *testing.T) {
},
Requirements: func(context.Context, string) ([]string, error) { return nil, nil },
}
var said []builds.VersionNumbered
fw := &FloorWaits{}
n := &Numberer{Versions: Versions{Store: newMem()}, Floor: src, Waits: fw, Said: func(b builds.VersionNumbered) { said = append(said, b) }}
n := &Numberer{Versions: Versions{Store: newMem()}, Floor: src, Waits: fw}
c := BuildClaim{Module: "m", Path: "go", Repository: "novox/m", Fingerprint: "fp1", Commit: "c1", Revision: "7", Merge: 7,
Artifacts: []ClaimArtifact{{Artifact: "program", Kind: KindBundle, Language: "go"}, {Artifact: "base", Kind: "upstream"}}}
got, err := n.Number(ctx, c)
if err != nil || len(got.Versions) != 0 || !reflect.DeepEqual(got.Unnumbered, []string{"program"}) || got.Wait != "" {
t.Fatalf("silent floor: %+v, %v; want program unnumbered, the upstream copy left out", got, err)
}
if !reflect.DeepEqual(asked, []string{"novox/m go/v"}) || len(fw.Waiting()) != 1 || len(said) != 0 {
t.Fatalf("asked %v, waits %+v, said %v", asked, fw.Waiting(), said)
if !reflect.DeepEqual(asked, []string{"novox/m go/v"}) || len(fw.Waiting()) != 1 {
t.Fatalf("asked %v, waits %+v", asked, fw.Waiting())
}
pkg := c
pkg.Artifacts = []ClaimArtifact{{Artifact: "lib", Kind: KindPackage, Language: "go"}}
@@ -419,27 +405,9 @@ func TestTheBuildersVersions(t *testing.T) {
}
tagsUp = true
got, err = n.Number(ctx, c)
if err != nil || got.Versions["program"] != "0.1.14" || len(said) != 1 || said[0].Version != "0.1.14" || len(fw.Waiting()) != 1 {
t.Fatalf("floor answered: %+v, %v, said %v, waits %+v; want 0.1.14 said once, lib still waiting", got, err, said, fw.Waiting())
if err != nil || got.Versions["program"] != "0.1.14" || len(fw.Waiting()) != 1 {
t.Fatalf("floor answered: %+v, %v, waits %+v; want 0.1.14, lib still waiting", got, err, fw.Waiting())
}
// DLV-B6: a check's build of a head a tag ask names numbers only its packages, tagged below the next version, kept
// with its repository for the branch's end, said with the head; claimed again it keeps its number.
tagged := c
tagged.Tag, tagged.Head, tagged.Revision, tagged.Merge = "feat-x", "h1", "", 0
tagged.Artifacts = append(tagged.Artifacts, ClaimArtifact{Artifact: "tlib", Kind: KindPackage, Language: "go"})
saidBefore, askedBefore := said, asked
said = nil
got, err = n.Number(ctx, tagged)
if err != nil || len(got.Versions) != 1 || !strings.Contains(got.Versions["tlib"], "-feat-x.1") || len(said) != 1 || said[0].Head != "h1" || said[0].Tag != "feat-x" {
t.Fatalf("tagged claim: %+v, %v, said %+v; want tlib alone tagged, said with its head", got, err, said)
}
if es, _ := n.Versions.Of(ctx, "m/tlib"); len(es) != 1 || es[0].Repository != "novox/m" {
t.Errorf("the tagged version kept = %+v, want its repository", es)
}
if again, _ := n.Number(ctx, tagged); again.Versions["tlib"] != got.Versions["tlib"] || len(said) != 1 {
t.Errorf("claimed again: %+v, said %d; want the same number, said once", again, len(said))
}
said, asked = saidBefore, askedBefore
got, _ = n.Number(ctx, c)
var publishing []string // review round 1: a version built and still numbered arms its publish bound
n.Publishing = func(_ context.Context, e Entry) { publishing = append(publishing, e.Name+"@"+e.Version) }
@@ -450,8 +418,8 @@ func TestTheBuildersVersions(t *testing.T) {
t.Fatalf("the publish bounds armed: %v, want m/program@0.1.14", publishing)
}
again, _ := n.Number(ctx, c)
if again.Versions["program"] != "0.1.14" || len(said) != 1 || len(asked) != 3 {
t.Fatalf("again: %+v, said %d, asked %v; want the number kept, nothing said, the floor not asked", again, len(said), asked)
if again.Versions["program"] != "0.1.14" || len(asked) != 3 {
t.Fatalf("again: %+v, asked %v; want the number kept, the floor not asked", again, asked)
}
if _, ok, why, _ := n.Versions.Recorded(ctx, "m/program", "0.1.14", "sha256:bb"); ok {
t.Fatalf("another digest recorded: %s", why)
@@ -510,8 +478,7 @@ func TestTheBuildersVersions(t *testing.T) {
return nil, nil
},
}
var said []builds.VersionNumbered
n := &Numberer{Versions: Versions{Store: newMem()}, Floor: src, Names: NamesIn(state), Said: func(b builds.VersionNumbered) { said = append(said, b) }}
n := &Numberer{Versions: Versions{Store: newMem()}, Floor: src, Names: NamesIn(state)}
c := BuildClaim{Module: "mesh-sdk", Path: "go", Repository: "novox/mesh-sdk", Fingerprint: "fp1", Commit: "c1", Revision: "3", Merge: 3,
Artifacts: []ClaimArtifact{{Artifact: "go", Kind: KindPackage, Language: "go"}, {Artifact: "ts", Kind: KindPackage, Language: "typescript"},
{Artifact: "new", Kind: KindPackage, Language: "go"}, {Artifact: "program", Kind: KindBundle, Language: "go"}}}
@@ -531,41 +498,26 @@ func TestTheBuildersVersions(t *testing.T) {
if es, _ := n.Versions.Of(ctx, "mesh-sdk/go"); len(es) != 0 {
t.Errorf("numbered under the builder's name too: %+v", es)
}
names := map[string]bool{}
for _, b := range said {
names[b.Name] = true
}
if !names["git.novox.be/novox/mesh-sdk/go"] || !names["@novox/mesh-sdk"] {
t.Errorf("version.numbered said %v, want the ecosystem names", names)
}
if err := n.Built(ctx, c, got, map[string]string{"go": "sha256:aa"}); err != nil {
t.Fatal(err)
}
if es, _ := n.Versions.Of(ctx, "git.novox.be/novox/mesh-sdk/go"); len(es) != 1 || es[0].Digest != "sha256:aa" {
t.Errorf("the built digest is not recorded under the name: %+v", es)
}
tagged, _ := TagAsk{Head: "h", State: TagAsked}.WithNumbered("mesh-sdk", "h", map[string]string{"go": "0.1.4-x.1"}, got.Names, time.Unix(1, 0))
if tagged.Versions["git.novox.be/novox/mesh-sdk/go"] != "0.1.4-x.1" {
t.Errorf("a tag ask names %v, want the Go module path", tagged.Versions)
}
// The indexer writes the name of `new` later: its series key names the module, and the look at the done build
// numbers it with the done key's digest, said in version.numbered for the late publish; the rest are kept.
// numbers it with the done key's digest; the rest are kept.
state["new.0.1"] = []byte(`{"v":1,"name":"new","package":"mesh-sdk/new","series":"0.1"}`)
for key, want := range map[string]string{"new.0.1": "mesh-sdk", "git=2enovox=2ebe/novox/mesh-sdk/go": "", "broken.0.1": ""} {
if module, name, ok := NamedModule(key, state[key]); module != want || ok != (want != "") || (ok && name != "new") {
t.Errorf("NamedModule(%s) = %q, %q, %v, want %q", key, module, name, ok, want)
}
}
said = nil
done := c
done.Done = map[string]string{"go": "sha256:aa", "new": "sha256:bb"}
late, err := n.Number(ctx, done)
if err != nil || late.Versions["new"] != "0.1.4" || late.Versions["go"] != "0.1.4" || len(late.Unnumbered) != 0 {
t.Fatalf("a look at the done build = %+v, %v; want new numbered and go kept", late, err)
}
if len(said) != 1 || said[0].Name != "new" || said[0].Digest != "sha256:bb" {
t.Errorf("version.numbered said %+v, want new once with the done build's digest", said)
}
if es, _ := n.Versions.Of(ctx, "new"); len(es) != 1 || es[0].Digest != "sha256:bb" {
t.Errorf("versions of new = %+v, want its done digest recorded", es)
}
@@ -657,23 +609,25 @@ func TestTheBuildersVersions(t *testing.T) {
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 []builds.VersionNumbered
var rebuilt []string
done := true
acts := Acts{Numberer: &Numberer{Versions: v, Said: func(b builds.VersionNumbered) { said = append(said, b) }, Waits: &FloorWaits{}},
acts := Acts{Numberer: &Numberer{Versions: v, 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)
// Nothing says a version numbered (design 47 G §349): a held build is refused before any number is taken.
if _, _, err := acts.PublishAgain(ctx, "m/p", "f1", "w", "me", time.Unix(5, 0)); err == nil || len(rebuilt) != 0 {
t.Errorf("publish again of a published, held build answered %v, rebuilt %v; want a refusal", err, rebuilt)
}
if es, _ := v.Of(ctx, "m/p"); len(es) != 2 {
t.Errorf("the refused publish again took a number: %+v", es)
}
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 ag, _, err := acts.PublishAgain(ctx, "m/p", "f1", "w", "me", time.Unix(6, 0)); err != nil || ag.Do != AgainRenumber || ag.Entry.Version != "0.1.5" || len(rebuilt) != 1 {
t.Errorf("publish again of a gone build: %+v %v rebuilt %v", ag, err, rebuilt)
}
if _, err := acts.FloorAgain(ctx, "m/p"); err == nil {
t.Error("a name with versions asks its floor again")
@@ -682,147 +636,4 @@ func TestTheBuildersVersions(t *testing.T) {
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"}}
for _, c := range []struct {
asks []TagAsk
repo, head string
want string
}{
{[]TagAsk{a}, "novox/mesh-sdk", "abc", "feat-x-y"},
{[]TagAsk{a}, "novox/mesh-sdk", "other", ""},
{[]TagAsk{{Repository: "novox/mesh-sdk", Head: "abc", Tag: "t", State: TagRemoved}}, "novox/mesh-sdk", "abc", ""},
} {
if got, _ := TagFor(c.asks, c.repo, c.head); got != c.want {
t.Errorf("TagFor(%s, %s) = %q, want %q", c.repo, c.head, got, c.want)
}
}
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
var kept map[string]bool // review round 2: the names whose seat keeps them whole (a shared digest)
acts := Acts{Numberer: &Numberer{Versions: Versions{Store: st}}, Remove: func(_ context.Context, seat string, e Entry) (RemoveOutcome, error) {
removed = append(removed, e.Name+"@"+e.Version)
if kept[e.Name] {
return RemoveOutcome{Kept: true, Reason: "shared-digest", Tags: []string{"0.1.2"}}, nil
}
if seat == SeatNPM {
return RemoveOutcome{}, errors.New("down")
}
return RemoveOutcome{}, 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)
}
// DLV-B6: a branch's end asks remove of every tagged version not removed, and is done once each seat said removed.
removed = nil
end, err := acts.BranchEnded(context.Background(), "o/r", "feat/x")
if err != nil || end.Done || len(removed) != 2 || !reflect.DeepEqual(end.Seats(), []string{"", SeatNPM}) || !strings.Contains(end.Finding("o/r", "feat/x", at), "n3@0.1.0-feat-x.1 on npm-package-registry: down") {
t.Fatalf("BranchEnded = %+v %v, removed %v", end, err, removed)
}
for _, v := range [][2]string{{"n", "0.1.2-feat-x.1"}, {"n3", "0.1.0-feat-x.1"}} {
if _, err := acts.Numberer.Heard(context.Background(), SeatNPM, HeardRemoved, "rm", Outcome{Name: v[0], Version: v[1], Reason: "tag-gone"}); err != nil {
t.Fatal(err)
}
}
if end, _ := acts.BranchEnded(context.Background(), "o/r", "feat/x"); !end.Done {
t.Errorf("every seat said removed, and the branch's end is not done: %+v", end)
}
// Review round 2: a remove its seat answers kept (another tag names the digest) ends there, with no retry: the
// version is recorded kept on the store with the reason and the other tags, which builder_versions shows.
raw, _ = Encode(Entry{Name: "n4", Kind: KindPackage, Series: "0.1", Version: "0.1.2-feat-x.1", Tag: "feat-x", Repository: "o/r", Seat: SeatNPM, State: Numbered})
_, _ = st.Create(context.Background(), Key("n4", "0.1.2-feat-x.1"), raw)
kept, removed = map[string]bool{"n4": true}, nil
if end, err := acts.BranchEnded(context.Background(), "o/r", "feat/x"); err != nil || !end.Done || len(removed) != 1 {
t.Fatalf("a kept remove: %+v %v, removed %v; want done at once", end, err, removed)
}
if es, _ := acts.Numberer.Versions.Of(context.Background(), "n4"); len(es) != 1 || es[0].State != Kept || es[0].Reason != "shared-digest" || !reflect.DeepEqual(es[0].KeptWith, []string{"0.1.2"}) {
t.Errorf("the kept version = %+v, want kept, shared-digest, with 0.1.2", es)
}
if removed = nil; func() bool {
end, _ := acts.BranchEnded(context.Background(), "o/r", "feat/x")
return !end.Done || len(removed) != 0
}() {
t.Errorf("a kept version is asked again: removed %v", removed)
}
if _, err := acts.Numberer.Heard(context.Background(), SeatNPM, HeardKept, "k", Outcome{Name: "n", Version: "0.1.2-feat-x.1", Tags: []string{"0.1.2"}}); err != nil {
t.Fatal(err)
}
if es, _ := acts.Numberer.Versions.Of(context.Background(), "n"); slices.ContainsFunc(es, func(e Entry) bool { return e.Version == "0.1.2-feat-x.1" && e.State == Kept }) {
t.Error("a seat's kept turned a removed version back to kept")
}
// B6: a check's build of the ask's head marks it numbered with its versions; another head, or an untag, is left.
if !a.Unanswered() {
t.Errorf("a kept ask with no version is answered: %+v", a)
}
if _, changed := a.WithNumbered("mesh-sdk", "other", map[string]string{"go": "0.1.15-feat-x-y.1"}, nil, at); changed {
t.Error("an ask is marked numbered by a build of another head")
}
if _, changed := u.WithNumbered("o", "", map[string]string{"go": "1"}, nil, at); changed {
t.Error("an untag is marked numbered")
}
n, changed := a.WithNumbered("mesh-sdk", "abc", map[string]string{"go": "0.1.15-feat-x-y.1"}, nil, at.Add(time.Minute))
if !changed || n.State != TagNumbered || n.Versions["mesh-sdk/go"] != "0.1.15-feat-x-y.1" || n.Unanswered() || !strings.Contains(n.Answer, "mesh-sdk/go 0.1.15-feat-x-y.1") {
t.Errorf("WithNumbered = %+v, %v", n, changed)
}
if _, again := n.WithNumbered("mesh-sdk", "abc", map[string]string{"go": "0.1.15-feat-x-y.1"}, nil, at); again {
t.Error("the same versions mark the ask again")
}
said := NarrateUnanswered(a, []FloorWait{{Name: "mesh-sdk/go", Silent: []string{"bridge_tags"}}}, at.Add(6*time.Minute))
for _, w := range []string{"novox/mesh-sdk feat/X y", "after 6m0s", "not claimed yet", "mesh-sdk/go waits on its floor (bridge_tags silent)"} {
if !strings.Contains(said, w) {
t.Errorf("NarrateUnanswered = %q, want %q in it", said, w)
}
}
if k, v, err := TagWantedAsk(TagWanted{Repository: "o/r", Branch: "feat/x", Head: "abc"}, "mesh-delivery-checker", at); err != nil || k != TagAskKey("o/r", "feat/x") || !strings.Contains(string(v), `"by":"mesh-delivery-checker"`) {
t.Errorf("TagWantedAsk = %q %s %v", k, v, err)
}
})
}
modules/mesh-delivery-builder/manifest_test.go
+5 -5
View File
@@ -102,7 +102,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: "engine-reads", History: 1}, {Name: "findings", History: 5}, {Name: "floor-waits", History: 1}, {Name: "hearing", History: 1}, {Name: "instances", History: 1}, {Name: "old-form", History: 1}, {Name: "bounds", History: 1, KeyTTL: true}, {Name: "queues", History: 1}, {Name: "reserved", History: 1}, {Name: "reasks", History: 1}, {Name: "tag-ask-bounds", History: 1, KeyTTL: true}, {Name: "tag-asks", History: 1}, {Name: "versions", History: 1}, {Name: "first-views", History: 1}}; !reflect.DeepEqual(m.State, want) { // builds.BuilderBuildsHistory
if want := []state{{Name: "builds", History: 5}, {Name: "engine-reads", History: 1}, {Name: "findings", History: 5}, {Name: "floor-waits", History: 1}, {Name: "hearing", History: 1}, {Name: "instances", History: 1}, {Name: "old-form", History: 1}, {Name: "bounds", History: 1, KeyTTL: true}, {Name: "queues", History: 1}, {Name: "reserved", History: 1}, {Name: "reasks", History: 1}, {Name: "versions", History: 1}, {Name: "first-views", History: 1}}; !reflect.DeepEqual(m.State, want) { // builds.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
@@ -146,7 +146,7 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1
if want := []string{"container-image-registry.kept", "container-image-registry.published", "container-image-registry.refused", "container-image-registry.removed",
"go-package-registry.kept", "go-package-registry.published", "go-package-registry.refused", "go-package-registry.removed",
"mesh-artifact-sweep.artifacts-gone", "mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported", "mesh-catalog.manifest-accepted", "mesh-catalog.version-released", "mesh-catalog.versions-filled",
"mesh-conditions.condition-cleared", "mesh-delivery-assigner.views.assignments", "mesh-delivery-checker.tag-wanted", "mesh-delivery-forge-bridge.forge.**",
"mesh-conditions.condition-cleared", "mesh-delivery-assigner.views.assignments", "mesh-delivery-forge-bridge.forge.**",
"mesh-delivery-indexer.views.candidates", "mesh-delivery-indexer.views.manifests", "mesh-delivery-indexer.views.modules",
"mesh-delivery-indexer.views.packages", "node-engine.manifest-reads-changed",
"npm-package-registry.kept", "npm-package-registry.published", "npm-package-registry.refused", "npm-package-registry.removed",
@@ -161,11 +161,11 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1
}
// 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_bounds", "builder_build_again", "builder_builds_since", "builder_engine_reads", "builder_first_view_again", "builder_first_views",
"builder_floor_again", "builder_floor_waits", "builder_old_form", "builder_publish_again", "builder_reserved", "builder_reserved_drop", "builder_seat_queue", "builder_stuck", "builder_tag", "builder_tag_asks", "builder_untag", "builder_versions", "reserved"}; !reflect.DeepEqual(sorted(m.Tools), want) {
"builder_floor_again", "builder_floor_waits", "builder_old_form", "builder_publish_again", "builder_reserved", "builder_reserved_drop", "builder_seat_queue", "builder_stuck", "builder_versions", "reserved"}; !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.
if want := []string{"build.asked-again", "version.numbered", "version.old-form-closed"}; !reflect.DeepEqual(m.Emits, want) {
// Design 47 G §349: version.numbered is no longer said; old-form-closed on the family version, apart from the builds' ends.
if want := []string{"build.asked-again", "version.old-form-closed"}; !reflect.DeepEqual(m.Emits, want) {
t.Errorf("emits = %v, want %v", m.Emits, want)
}
// 1955/L7 (ADR 0571 §1): its module log, each build's start and end, kept 30 days in a cap that holds them; no bare stream.
modules/mesh-delivery-builder/module.json
-7
View File
@@ -26,9 +26,6 @@
"builder_reserved_drop",
"builder_seat_queue",
"builder_stuck",
"builder_tag",
"builder_tag_asks",
"builder_untag",
"builder_versions",
"reserved",
"builder_first_views",
@@ -36,7 +33,6 @@
],
"emits": [
"build.asked-again",
"version.numbered",
"version.old-form-closed"
],
"consumes": [
@@ -55,7 +51,6 @@
"mesh-catalog.version-released",
"mesh-catalog.versions-filled",
"mesh-conditions.condition-cleared",
"mesh-delivery-checker.tag-wanted",
"mesh-delivery-forge-bridge.forge.**",
"node-engine.manifest-reads-changed",
"npm-package-registry.kept",
@@ -110,8 +105,6 @@
{ "name": "queues", "history": 1 },
{ "name": "reserved", "history": 1 },
{ "name": "reasks", "history": 1 },
{ "name": "tag-ask-bounds", "history": 1, "key-ttl": true },
{ "name": "tag-asks", "history": 1 },
{ "name": "versions", "history": 1 },
{ "name": "first-views", "history": 1 }
],