CAT-BG2: a bundle or archive the builder numbers later is published late, sending nothing
mesh-build-go consumes the builder's version.numbered and publishes the held digest under its version, naming its one part by digest only, since the packed bytes left the node; distribution tags it at the size its store holds, so the version's manifest names the blob truly. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
This commit is contained in:
@@ -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)
|
||||
|
||||
@@ -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/")
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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/<module>/<artifact>@<digest>` (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: <artifact>: <the holder's words>`. 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
|
||||
|
||||
@@ -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):
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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=
|
||||
@@ -19,6 +19,9 @@
|
||||
"serves": ["current", "pause", "resume"]
|
||||
}
|
||||
],
|
||||
"consumes": [
|
||||
"mesh-delivery-builder.version.numbered"
|
||||
],
|
||||
"reads": [
|
||||
"mesh-delivery-checker.kept",
|
||||
"mesh-delivery-indexer.manifests"
|
||||
|
||||
Reference in New Issue
Block a user