diff --git a/modules/distribution/internal/holder/holder_test.go b/modules/distribution/internal/holder/holder_test.go index b9c152898..ec1ba9b66 100644 --- a/modules/distribution/internal/holder/holder_test.go +++ b/modules/distribution/internal/holder/holder_test.go @@ -463,6 +463,32 @@ func TestTheHolderOfTheTwoImageSeats(t *testing.T) { } } }) + t.Run("a late publish names a held bundle with no size: tagged at the store's size, nothing sent", func(t *testing.T) { + store := newFakeRegistry() + sts := httptest.NewServer(store) + defer sts.Close() + base, _ := url.Parse(sts.URL) + o := &OCI{Base: base} + blob := []byte(strings.Repeat("b", 4321)) + store.blobs["m/tools@"+digestOf(blob)] = blob + ref, err := o.Put(PublishRequest{Name: "m/tools", Version: "0.1.4", Kind: KindBundle, Digest: digestOf(blob), + Parts: []Part{{Digest: digestOf(blob)}}}, map[string][]byte{}) + if err != nil || ref != "m/tools@"+digestOf(blob) { + t.Fatalf("Put = %q, %v", ref, err) + } + var man struct { + Layers []struct { + Size int64 `json:"size"` + } `json:"layers"` + } + if err := json.Unmarshal(store.manifests["m/tools:0.1.4"], &man); err != nil || len(man.Layers) != 1 || man.Layers[0].Size != int64(len(blob)) { + t.Errorf("the version's manifest %s, want one layer of %d bytes", store.manifests["m/tools:0.1.4"], len(blob)) + } + if _, err := o.Put(PublishRequest{Name: "m/tools", Version: "0.1.5", Kind: KindBundle, Digest: "sha256:" + strings.Repeat("e", 64), + Parts: []Part{{Digest: "sha256:" + strings.Repeat("e", 64)}}}, map[string][]byte{}); err == nil { + t.Error("a digest the store does not hold was tagged") + } + }) t.Run("mirror answers at once and says the copy's end on the seat's events", func(t *testing.T) { b := newBed(true) d := "sha256:" + strings.Repeat("c", 64) diff --git a/modules/distribution/internal/holder/mirror_fakes_test.go b/modules/distribution/internal/holder/mirror_fakes_test.go index 33af195cf..6e51ebd0b 100644 --- a/modules/distribution/internal/holder/mirror_fakes_test.go +++ b/modules/distribution/internal/holder/mirror_fakes_test.go @@ -4,6 +4,7 @@ import ( "errors" "io" "net/http" + "strconv" "strings" "sync" ) @@ -63,6 +64,7 @@ func (f *fakeRegistry) ServeHTTP(w http.ResponseWriter, r *http.Request) { if r.Method == http.MethodGet { f.gets = append(f.gets, p[i+7:]) } + w.Header().Set("Content-Length", strconv.Itoa(len(b))) // a HEAD's too, as a registry answers it _, _ = w.Write(b) case strings.Contains(p, "/manifests/"): i := strings.Index(p, "/manifests/") diff --git a/modules/distribution/internal/holder/oci.go b/modules/distribution/internal/holder/oci.go index 6deee2222..c722d03b5 100644 --- a/modules/distribution/internal/holder/oci.go +++ b/modules/distribution/internal/holder/oci.go @@ -166,11 +166,9 @@ func (o *OCI) Put(t PublishRequest, parts map[string][]byte) (string, error) { if err := o.putBlob(t.Name, "sha256:"+sum([]byte(emptyConfig)), []byte(emptyConfig)); err != nil { return "", err } - size := int64(0) - for _, p := range t.Parts { - if p.Digest == t.Digest { - size = p.Size - } + size, err := o.layerSize(t, parts) + if err != nil { + return "", err } m, _ := json.Marshal(map[string]any{ "schemaVersion": 2, "mediaType": mediaManifest, @@ -187,6 +185,27 @@ func (o *OCI) Put(t PublishRequest, parts map[string][]byte) (string, error) { } } +// layerSize is the size of a bundle's or an archive's one blob: its bytes' when they travelled, else the size its part +// names, else (a late publish of a held digest names it with no size, its bytes having left the build node) the store's. +func (o *OCI) layerSize(t PublishRequest, parts map[string][]byte) (int64, error) { + if data, ok := parts[t.Digest]; ok { + return int64(len(data)), nil + } + for _, p := range t.Parts { + if p.Digest == t.Digest && p.Size > 0 { + return p.Size, nil + } + } + resp, _, err := o.do(http.MethodHead, "/v2/"+t.Name+"/blobs/"+t.Digest, "", nil) + if err != nil { + return 0, err + } + if resp.StatusCode != http.StatusOK || resp.ContentLength < 0 { + return 0, fmt.Errorf("the store holds no blob %s of %s to tag (%d)", t.Digest, t.Name, resp.StatusCode) + } + return resp.ContentLength, nil +} + // Held is the reference of name at version when the store holds digest under it. func (o *OCI) Held(name, version, digest string) (string, bool, error) { d, err := o.Digest(name, version) diff --git a/modules/mesh-build-go/README.md b/modules/mesh-build-go/README.md index d45552acb..391fcc89a 100644 --- a/modules/mesh-build-go/README.md +++ b/modules/mesh-build-go/README.md @@ -142,6 +142,9 @@ trunk build's, in chunks, only what the holder lacks; the holder asks the builde seat's events. The step answers the seat's form `oci-artifact-registry//@` (ADR 0605 §9); the builder writes the older form beside it while a counted node-engine reads only that (ADR 0624); a publish the holder refuses, or a seat nothing holds, fails the step `not stored: : `. The packed bytes never stay on the node. +A build the builder numbers later is published late: the module consumes the builder's `version.numbered` and, for a +bundle or an archive with a digest, publishes that held digest under its version, naming its one part by digest and +sending nothing (`latepublish.go`); a holder that no longer holds the digest fails the event, which is offered again. A contracts module (a package artifact naming `generated-from`) has its generated Go files regenerated with the seat generator this module is built with and compared (`buildstep/generated.go`): the unit level of `build_go_test` fails naming each file that differs, is missing or was not written by the generator (its `out-of-step`), and a build ask diff --git a/modules/mesh-build-go/cmd/mesh-build-go/latepublish.go b/modules/mesh-build-go/cmd/mesh-build-go/latepublish.go new file mode 100644 index 000000000..b0104c5bf --- /dev/null +++ b/modules/mesh-build-go/cmd/mesh-build-go/latepublish.go @@ -0,0 +1,89 @@ +package main + +// The late publish (novox/hq design 51 *The versions*, *Publishing*; CAT-BG2): a trunk build of a bundle or an archive +// publishes its digest unnumbered when the builder has not numbered it yet; the builder says version.numbered once it +// does, and this module publishes the held digest again under that version. Nothing travels: the publish names the one +// part by its digest, which the holder holds, so it answers held and tags the version (the holder asks builder_versions +// first, ADR 0605 §4). The bytes never stayed on the node, so a holder that no longer holds the digest is said and the +// event offered again, which ends in the controller's condition when it keeps failing. Every node's instance may hear +// the same event; the publish is the same call under the same transfer name, so a second one is answered held. + +import ( + "context" + "errors" + "fmt" + "io" + "time" + + registry "git.novox.be/novox/mesh-contracts/go/seats/containerimageregistry" + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/publish" +) + +// NumberedPattern is the event type this module consumes for the late publish, as its manifest's `consumes` names it. +const NumberedPattern = delivery.BuilderModule + "." + delivery.VersionNumberedEvent + +// errNotHere is a held part's Open: its bytes did not stay on the node. +var errNotHere = errors.New("the packed bytes did not stay on the node; the holder must hold the digest") + +// LatePublish publishes a numbered bundle or archive's held digest under its version, from the builder's +// version.numbered body. It answers false, nil for an event that is not this module's to publish: another kind (a Go +// package publishes at its build, an image is mesh-build-image's), another seat, or a version numbered with no digest. +func (m *Module) LatePublish(ctx context.Context, body []byte) (bool, error) { + n, err := delivery.DecodeVersionNumbered(body) + if err != nil { + return false, err + } + seat := n.Seat + if seat == "" { + seat = ArtifactSeat + } + if n.Digest == "" || seat != ArtifactSeat || (n.Kind != "bundle" && n.Kind != "archive") { + return false, nil + } + call := m.Call + if call == nil { + call = stdio.Ask + } + in := registry.PublishInput{Name: n.Name, Kind: n.Kind, Version: n.Version, Tag: n.Tag, Digest: n.Digest, + Fingerprint: n.Fingerprint, Commit: n.Commit, Transfer: transferName(n.Name, "v"+n.Version, n.Digest)} + sent, err := publish.Publish(ctx, publish.Call(call), seat, in, []publish.Part{{Digest: n.Digest, + Open: func() (io.ReadCloser, error) { return nil, errNotHere }}}) + if err != nil { + return false, fmt.Errorf("the late publish of %s@%s (%s): %w", n.Name, n.Version, n.Digest, err) + } + if len(sent.Parts) > 0 { + return false, fmt.Errorf("the late publish of %s@%s sent parts it should not have", n.Name, n.Version) + } + return true, nil +} + +// hearNumbered subscribes to the builder's version.numbered for the late publish, asking again until the runtime +// takes it, and says each publish and each failure. +func (m *Module) hearNumbered(ctx context.Context) { + handle := func(e stdio.Envelope) error { + did, err := m.LatePublish(ctx, e.Body) + if err != nil { + logf("%v", err) + return err + } + if did { + logf("published a numbered build's held digest under its version (%s)", e.Headers["x-event-id"]) + } + return nil + } + for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { + if err := stdio.Subscribe(NumberedPattern, handle); err == nil { + logf("listening for the builder's %s", NumberedPattern) + return + } else { + logf("not hearing the builder's %s yet (%v); asking again in %s", NumberedPattern, err, wait) + } + select { + case <-ctx.Done(): + return + case <-time.After(wait): + } + } +} diff --git a/modules/mesh-build-go/cmd/mesh-build-go/main.go b/modules/mesh-build-go/cmd/mesh-build-go/main.go index 92fe5780f..3218430fc 100644 --- a/modules/mesh-build-go/cmd/mesh-build-go/main.go +++ b/modules/mesh-build-go/cmd/mesh-build-go/main.go @@ -82,6 +82,8 @@ func main() { if m.Node != "" { // The seat's asks, pulled once Serve has the runtime's channel; a pull before it is said and tried again. go func() { _ = puller.Pull(ctx) }() + // The builder's version.numbered, for the late publish of a held digest (latepublish.go). + go m.hearNumbered(ctx) } if err := stdio.Serve("", append(m.Tools(steps), m.SeatTools(puller)...)); err != nil { logf("%v", err) diff --git a/modules/mesh-build-go/cmd/mesh-build-go/manifest_test.go b/modules/mesh-build-go/cmd/mesh-build-go/manifest_test.go index 71e81ed1f..ccf81f132 100644 --- a/modules/mesh-build-go/cmd/mesh-build-go/manifest_test.go +++ b/modules/mesh-build-go/cmd/mesh-build-go/manifest_test.go @@ -174,6 +174,10 @@ func TestTheModuleInvokesTheArtifactSeatsVerbsAndWritesNoStore(t *testing.T) { } } } + // The late publish hears the builder's version.numbered (latepublish.go). + if !slices.Contains(m.Consumes, NumberedPattern) { + t.Errorf("consumes %v lack %s", m.Consumes, NumberedPattern) + } for _, r := range m.Resources { if r["id"] == "registry" { t.Errorf("the store's address is still placed: %v", r) diff --git a/modules/mesh-build-go/cmd/mesh-build-go/publish_test.go b/modules/mesh-build-go/cmd/mesh-build-go/publish_test.go index e3969876e..221a9bd2e 100644 --- a/modules/mesh-build-go/cmd/mesh-build-go/publish_test.go +++ b/modules/mesh-build-go/cmd/mesh-build-go/publish_test.go @@ -56,6 +56,11 @@ func withStore(t *testing.T, m *Module) *fakeStore { t.Errorf("publish of kind %q, transfer %q", in.Kind, in.Transfer) } s.open[in.Transfer] = in + // A digest held under the name already travels no more (the late publish, CAT-BG2). + if _, held := s.blobs[in.Name+"@"+in.Digest]; held { + s.published = append(s.published, in) + return json.Marshal(registry.PublishAnswer{Transfer: in.Transfer, Held: true}) + } return json.Marshal(registry.PublishAnswer{Transfer: in.Transfer, Lacks: []string{in.Digest}}) case "publish-part": var c registry.PublishPartInput @@ -225,3 +230,57 @@ func TestAGoPackageIsBuiltAsAModuleZipAndPublishedUnderItsModulePath(t *testing. t.Errorf("no go.mod: %s %q", st.State, st.Reason) } } + +// CAT-BG2 (design 51 *The versions*): the builder's version.numbered for a bundle or an archive publishes the held +// digest under its version, sending nothing; a digest the holder no longer holds fails the event (offered again), +// and what is not this module's to publish is let pass. +func TestTheNumberedEventPublishesAHeldDigestUnderItsVersion(t *testing.T) { + digest := "sha256:" + strings.Repeat("ab", 32) + other := "sha256:" + strings.Repeat("cd", 32) + numbered := func(kind, seat, digest string) []byte { + raw, err := delivery.EncodeVersionNumbered(delivery.VersionNumbered{Name: "mod/tools", Kind: kind, Version: "0.1.4", + Module: "mod", Fingerprint: "fp", Commit: "c0", Digest: digest, Seat: seat}) + if err != nil { + t.Fatal(err) + } + return raw + } + for _, c := range []struct { + name string + body []byte + did, fail bool + }{ + {"a held bundle is tagged", numbered("bundle", ArtifactSeat, digest), true, false}, + {"an archive with no seat named goes to the artifact seat", numbered("archive", "", digest), true, false}, + {"a digest the holder lacks fails, sending nothing", numbered("bundle", ArtifactSeat, other), false, true}, + {"a Go package is not published late", numbered("package", GoPackageSeat, digest), false, false}, + {"an image is mesh-build-image's", numbered("image", "container-image-registry", digest), false, false}, + {"no digest, nothing to publish", numbered("bundle", ArtifactSeat, ""), false, false}, + {"an unreadable body fails", []byte(`{"name":"mod/tools"}`), false, true}, + } { + t.Run(c.name, func(t *testing.T) { + m := &Module{} + s := withStore(t, m) + s.blobs["mod/tools@"+digest] = []byte("held") + did, err := m.LatePublish(t.Context(), c.body) + if did != c.did || (err != nil) != c.fail { + t.Fatalf("did %v, err %v; want did %v, fail %v", did, err, c.did, c.fail) + } + for _, k := range s.asked { + if strings.HasSuffix(k, ".publish-part") { + t.Errorf("a late publish sent a part: %v", s.asked) + } + } + if !c.did { + return + } + if len(s.published) != 1 { + t.Fatalf("published %v", s.published) + } + p := s.published[0] + if p.Name != "mod/tools" || p.Version != "0.1.4" || p.Digest != digest || p.Fingerprint != "fp" || p.Commit != "c0" { + t.Errorf("published %+v", p) + } + }) + } +} diff --git a/modules/mesh-build-go/go.work.sum b/modules/mesh-build-go/go.work.sum new file mode 100644 index 000000000..d8beaa0a6 --- /dev/null +++ b/modules/mesh-build-go/go.work.sum @@ -0,0 +1,39 @@ +github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/go-jose/go-jose/v4 v4.1.4/go.mod h1:x4oUasVrzR7071A4TnHLGSPpNOm2a21K9Kf04k1rs08= +github.com/go-quicktest/qt v1.101.0/go.mod h1:14Bz/f7NwaXPtdYEgzsx46kqSxVwTbzVZsDC26tQJow= +github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= +github.com/google/go-tpm-tools v0.3.13-0.20230620182252-4639ecce2aba/go.mod h1:EFYHy8/1y2KfgTAsx7Luu7NGhoxtuVHnNo8jE7FikKc= +github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo= +github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= +github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= +github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c/go.mod h1:7rwL4CYBLnjLxUqIJNnCWiEdr3bn6IUYi15bNlnbCCU= +github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsKMEsNbx1VGcRFpLqf3715MtcvvzbA= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs= +github.com/rogpeppe/go-internal v1.11.0/go.mod h1:ddIwULY96R17DhadqLgMfk9H9tvdUzkipdSkR5nkCZA= +github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= +github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY= +go.uber.org/automaxprocs v1.6.0/go.mod h1:ifeIMSnPZuznNm6jmdzmU3/bfk01Fe2fotchwEFJ8r8= +golang.org/x/crypto v0.52.0/go.mod h1:1QgfPxDqh0T2M/elOJtp9RvuR95kVjir0e6/BvEmGbc= +golang.org/x/mod v0.21.0/go.mod h1:6SkKJ3Xj0I0BrPOZoBy3bdMptDDU9oJrpohJ3eWZ1fY= +golang.org/x/mod v0.33.0/go.mod h1:swjeQEj+6r7fODbD2cqrnje9PnziFuw4bmLbBZFrQ5w= +golang.org/x/mod v0.39.0/go.mod h1:bvIbwjQ0HUFFf5AKukeeYQG4ZBUG9yxQbR9aEweIwYY= +golang.org/x/mod v0.41.0 h1:qJmnOUb4YB+FsEuM3HcWucdZASCPGhsX6uljO6pog0c= +golang.org/x/mod v0.41.0/go.mod h1:Ek9pY8RKWXwsWvd3rQiHYtMqkjSUV+s1Rj7j4H5Ur6o= +golang.org/x/net v0.58.0/go.mod h1:YwCddHnFlT7eLQqVprV19OnhLGtc5xOKgE0RyqgfWAU= +golang.org/x/sync v0.19.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= +golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= +golang.org/x/sys v0.26.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +golang.org/x/sys v0.45.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/telemetry v0.0.0-20260811182544-a038080d80e5/go.mod h1:LVehoXe41cL5SCVQilsV7Gg6BNG+Js6P9PhSbYTIUkQ= +golang.org/x/telemetry v0.0.0-20260924152758-ed294f943157/go.mod h1:i+ivNqjDnTF3WTElsdk5g9V5DTSBYgdNo7xTU9SDwYA= +golang.org/x/term v0.46.0/go.mod h1:+K02xbkittuwc0Am4abfA3Fc+XRGXkvBXNO88NCXPoc= +golang.org/x/tools v0.26.0/go.mod h1:TPVVj70c7JJ3WCazhD8OdXcZg/og+b9+tH/KxylGwH0= +golang.org/x/tools v0.42.0/go.mod h1:Ma6lCIwGZvHK6XtgbswSoWroEkhugApmsXyrUmBhfr0= +golang.org/x/tools v0.49.0/go.mod h1:SJNXV9DBKT0UbdttsQjbfJlAE/q+y36++zo3uL3N0Oo= +golang.org/x/tools v0.51.0/go.mod h1:9eEncMayCV6zRMGhR5eZEC2iBx98qWcF1HZ9Z7wJOoA= +google.golang.org/genproto v0.0.0-20260128011058-8636f8732409 h1:VQZ/yAbAtjkHgH80teYd2em3xtIkkHd7ZhqfH2N9CsM= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/modules/mesh-build-go/module.json b/modules/mesh-build-go/module.json index b12732d26..f1c939412 100644 --- a/modules/mesh-build-go/module.json +++ b/modules/mesh-build-go/module.json @@ -19,6 +19,9 @@ "serves": ["current", "pause", "resume"] } ], + "consumes": [ + "mesh-delivery-builder.version.numbered" + ], "reads": [ "mesh-delivery-checker.kept", "mesh-delivery-indexer.manifests"