CAT-BI1: mesh-build-image publishes and copies through container-image-registry

A trunk image is written as an OCI layout and published by the seat's
publish and publish-part, only the parts the holder lacks; an upstream copy is
the seat's mirror, ended on the holder's published or refused. The module
holds no registry address: artifact-store, files/registry and the copier go,
and the step answers the seat's reference.

Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
This commit is contained in:
2026-11-04 15:48:55 +01:00
parent e40ace34a6
commit d47d231ec3
16 changed files with 389 additions and 903 deletions
@@ -11,6 +11,7 @@ import (
"context"
"encoding/json"
"io"
"os"
"strings"
reg "git.novox.be/novox/mesh-contracts/go/seats/containerimageregistry"
@@ -20,6 +21,7 @@ import (
// The registry seats a build module publishes on.
const (
ArtifactSeat = "oci-artifact-registry"
ImageSeat = reg.Seat
GoSeat = "go-package-registry"
NpmSeat = "npm-package-registry"
)
@@ -44,3 +46,20 @@ func PublishOne(ctx context.Context, call SeatCall, seat, name, kind, version, d
Open: func() (io.ReadCloser, error) { return io.NopCloser(bytes.NewReader(body)), nil }}})
return err
}
// File is one part of a publish kept on disk: an image's config, layer or manifest, by its digest and size.
type File struct {
Digest string
Size int64
Path string
}
// PublishFiles publishes in with files as its parts (an image's config, layers and manifest, CAT-BI1): the holder
// answers the parts it lacks, and only those are read and sent.
func PublishFiles(ctx context.Context, call SeatCall, seat string, in reg.PublishInput, files []File) (publish.Sent, error) {
ps := make([]publish.Part, 0, len(files))
for _, f := range files {
ps = append(ps, publish.Part{Digest: f.Digest, Size: f.Size, Open: func() (io.ReadCloser, error) { return os.Open(f.Path) }})
}
return publish.Publish(ctx, publish.Call(call), seat, in, ps)
}
+13 -9
View File
@@ -34,14 +34,17 @@ The Dockerfile's steps may use the network (ADR 0487 §6); nothing a step can re
- `build_image_build` builds one image artifact: `buildah bud` with the Dockerfile `from` names (in the module's
directory or a subdirectory of it) as file and the module's own directory as context, so a COPY names paths from the
module's root (issue 1703), `--target`, and the images `build.on` pins as `--build-arg`. Answers the image's digest. With
`publish: true` (a merged trunk commit only) the module's own process then pushes it to the registry under
`<module>/<artifact>:<tag of its own>` (the first 12 hex digits of the digest unless the call gives one), waiting out
a registry held still, and the key carries the digest the registry serves and the reference pinned by it. Without
`publish` (a pull request) nothing is pushed.
- `build_image_copy` copies an upstream artifact (`module`, `artifact`, `from`) registry to registry into the artifact store
over the registry API, the whole index and every blob, by the digest the manifest pins (ADR 0498 §4), and verifies the
store serves that digest. Answers the digest and the reference `<registry>/<module>/<artifact>@sha256:<digest>`. A
`from` without a digest fails `not pinned:`. It needs no workspace and no buildah.
`publish: true` (a merged trunk commit only) the module's own process then writes it as an OCI layout in the
workspace and publishes it through the seat `container-image-registry` (`publish`, `publish-part`, which the manifest
`invokes`; design 51 *Publishing*) as `<module>/<artifact>` at the ask's `version` (and a `tag` only when the call gives
one): its config, layers and manifest by digest, of which only the parts the holder lacks travel. The key carries the
layout's manifest digest and the seat's reference `container-image-registry/<module>/<artifact>@sha256:<digest>`; a
publish the holder does not take fails `not stored:`. Without `publish` (a pull request) nothing is published.
- `build_image_copy` copies an upstream artifact (`module`, `artifact`, `from`) into the image seat by its `mirror`, by the
digest the manifest pins (ADR 0498 §4): the holder copies it and says `published` or `refused` on the seat's events,
which the manifest `consumes`; the step ends on that outcome (or fails `not stored:` after 30 minutes without one) and
answers the digest and the seat's reference. A `from` without a digest fails `not pinned:` before the seat is asked. It
needs no workspace, no buildah, and reads no outside registry itself.
- It refuses, naming why, an image whose build context is a whole repository (`context`) or whose `build.on` names
another module's artifact (`unsupportedImage` in `build.go`): those forms belong to the images build-agent builds
until it is removed at R4g.
@@ -53,4 +56,5 @@ The Dockerfile's steps may use the network (ADR 0487 §6); nothing a step can re
(issue 1839). A copy writes nothing to the disk and is not refused.
- `build_image_step`, `build_image_steps`, `build_image_status` and `build_image_health` read.
The registry's address is placed into `<home>/registry` as `${seat:mesh-artifact-store:reach}` (host:port): the provisioner's client fills a moved module's placement, and the artifact store has not moved, so a `${bound:artifact-store:...}` is refused at composition (issue 1703).
The module holds no registry address and no registry credential: every write goes through the image seat's verbs, and
the holder checks each publish with the builder.
@@ -3,8 +3,8 @@ package main
// The build step (novox/hq ADR 0487 §1 §2 §5 §6; issue 1703c): one `image` artifact of a module of a fetched workspace,
// built from the Dockerfile in the module's own directory with buildah, rootless: as the module's own account, in a
// user namespace, with its image store outside the home, no mount of the host, no registry credential and an
// environment built from scratch. The step runs the Dockerfile and nothing else; the push of a trunk build is made
// after the build has ended (push.go).
// environment built from scratch. The step runs the Dockerfile and nothing else; the publish of a trunk build through
// the image seat is made after the build has ended (publish.go).
import (
"context"
@@ -21,7 +21,7 @@ import (
)
// BuildResult is what a passed build step says: the image's manifest digest as buildah holds it, and, for a trunk
// build that published, the digest the registry serves and the reference pinned by it.
// build that published, the digest it was published by and the seat's reference pinned by it.
type BuildResult struct {
Artifact string `json:"artifact"`
Digest string `json:"digest"`
@@ -36,16 +36,19 @@ func (m *Module) buildTool() StepTool {
Name: BuildTool,
Description: "Build one image artifact of a module of a fetched workspace from the Dockerfile in the module's own directory, " +
"rootless with buildah. Answers a step id at once; the step ends on its key with the image's digest. With publish " +
"true (a merged trunk commit) the image is then pushed to the registry from this module's own process and the key " +
"carries the digest the registry serves and the reference pinned by it; without it (a pull request) nothing is pushed.",
"true (a merged trunk commit) the image is then published through the seat container-image-registry from this " +
"module's own process, only the layers the holder lacks, and the key carries the digest it was published by and the " +
"seat's reference, container-image-registry/<module>/<artifact>@sha256:<digest>; without it (a pull request) nothing is published.",
Input: map[string]any{
"type": "object",
"properties": map[string]any{
"workspace": map[string]any{"type": "string", "description": "the workspace id build_image_fetch answered"},
"module": map[string]any{"type": "string", "description": "the module's directory in the workspace, such as modules/n8n"},
"artifact": map[string]any{"type": "string", "description": "the image artifact's name in the module's manifest"},
"publish": map[string]any{"type": "boolean", "description": "push the image to the registry after the build: only for a merged trunk commit"},
"tag": map[string]any{"type": "string", "description": "the registry tag of a published image; the first 12 hex digits of the digest when left out"},
"publish": map[string]any{"type": "boolean", "description": "publish the image through the image seat after the build: only for a merged trunk commit"},
"seat": map[string]any{"type": "string", "description": "the seat it is published through, container-image-registry (the ask's own)"},
"version": map[string]any{"type": "string", "description": "the version the builder gave it, named in the seat's publish"},
"tag": map[string]any{"type": "string", "description": "a tag the published version carries, by hand; none when left out"},
},
"required": []string{"workspace", "module", "artifact"},
},
@@ -111,6 +114,10 @@ func (m *Module) build(ctx context.Context, args map[string]any) (any, error) {
name, _ := args["artifact"].(string)
publish, _ := args["publish"].(bool)
tag, _ := args["tag"].(string)
version, _ := args["version"].(string)
if seat, _ := args["seat"].(string); seat != "" && seat != buildstep.ImageSeat {
return nil, &buildstep.Refusal{Text: fmt.Sprintf("an image is published through %s, not %s", buildstep.ImageSeat, seat)}
}
if !workspaceOK.MatchString(workspace) {
return nil, fmt.Errorf("%q is not a workspace id", workspace)
}
@@ -177,10 +184,7 @@ func (m *Module) build(ctx context.Context, args map[string]any) (any, error) {
if !publish {
return res, nil
}
if tag == "" {
tag = strings.TrimPrefix(digest, "sha256:")[:12]
}
return m.push(ctx, workspace, manifest.Module, local, res, tag)
return m.publish(ctx, workspace, manifest.Module, local, res, version, tag)
}
// dockerfileOf is the Dockerfile an artifact's `from` names and the build context: the module's own directory, the
@@ -151,7 +151,7 @@ func TestTheStepsAnswerThroughARealBusWithTheCapturedShapes(t *testing.T) {
}
compareCaptured(t, name, normalized(t, st, "", "", f.address()))
}
if len(f.registry) != 1 || f.pushes != 1 {
t.Errorf("the registry holds %v after %d pushes, want only the trunk build's one", f.registry, f.pushes)
if len(f.published) != 1 || f.pushes != 1 {
t.Errorf("the image seat was asked %v after %d layouts, want only the trunk build's one", f.published, f.pushes)
}
}
@@ -1,25 +1,44 @@
package main
import (
"encoding/json"
"os"
"path/filepath"
"slices"
"strings"
"testing"
)
// Issue 1703 (the live failure of the first assignment): the module is moved, so the provisioner's client on the node
// fills its templates, and it fills only config, secrets, collections, placement and what a moved provider told it. The
// registry's address therefore cannot be a ${bound:artifact-store:...} (the artifact store has not moved): it is the
// placement ${seat:mesh-artifact-store:reach}, the form the container runtime's daemon.json uses. The module's own part
// of that is what this holds: its registry file is that placement and none of its files or its manifest binds the store.
// That a node composes it is the node-engine's composition test (.work/sdk1/proofs-for-node-engine.md).
func TestTheRegistryIsThePlacementAndNoFileBindsTheStore(t *testing.T) {
raw, err := os.ReadFile(filepath.Join(moduleDir, "files", "registry"))
if err != nil {
// CAT-BI1 (design 51 *Publishing*): the module publishes and copies through the image seat's verbs, which its manifest
// invokes, hears the seat's outcomes through its consumes, and holds no registry: no registry file, no artifact-store
// requirement, and none of its files names the store's placement or binds it (issue 1703: the module is moved, and only
// what the mesh fills reaches its files).
func TestTheModuleInvokesTheImageSeatAndHoldsNoRegistry(t *testing.T) {
var m struct {
Requires []string `json:"requires"`
Invokes []string `json:"invokes"`
Consumes []string `json:"consumes"`
}
raw, err := os.ReadFile(filepath.Join(moduleDir, "module.json"))
if err != nil || json.Unmarshal(raw, &m) != nil {
t.Fatal(err)
}
if got := strings.TrimSpace(string(raw)); got != "${seat:mesh-artifact-store:reach}" {
t.Errorf("the registry file is %q, want the placement ${seat:mesh-artifact-store:reach}", got)
if want := []string{"seat:container-image-registry.publish", "seat:container-image-registry.publish-part",
"seat:container-image-registry.mirror"}; !slices.Equal(m.Invokes, want) {
t.Errorf("invokes = %v, want %v", m.Invokes, want)
}
var consumes []string
for _, e := range OutcomeEvents {
consumes = append(consumes, "container-image-registry."+e)
}
if !slices.Equal(m.Consumes, consumes) {
t.Errorf("consumes = %v, want %v", m.Consumes, consumes)
}
if slices.Contains(m.Requires, "artifact-store") {
t.Errorf("requires = %v: the module holds no store", m.Requires)
}
if _, err := os.Stat(filepath.Join(moduleDir, "files", "registry")); !os.IsNotExist(err) {
t.Errorf("files/registry is still there: %v", err)
}
files := []string{filepath.Join(moduleDir, "module.json")}
entries, err := os.ReadDir(filepath.Join(moduleDir, "files"))
@@ -34,8 +53,8 @@ func TestTheRegistryIsThePlacementAndNoFileBindsTheStore(t *testing.T) {
if err != nil {
t.Fatal(err)
}
if strings.Contains(string(body), "${bound:artifact-store") {
t.Errorf("%s binds the artifact store, which has not moved: a moved module's client cannot fill it", f)
if strings.Contains(string(body), "${bound:artifact-store") || strings.Contains(string(body), "mesh-artifact-store") {
t.Errorf("%s names the artifact store", f)
}
}
}
@@ -1,34 +1,36 @@
package main
// The copy step (novox/hq ADR 0498 §4; issue 1587): an upstream artifact is copied registry to registry into the
// artifact store, whole (an index with every manifest it lists, every blob), by the digest its manifest pins. The copy
// runs no code of a change, so it needs neither buildah nor an isolation; it is made over the registry API, which
// keeps every digest as it is (a pull and a push through buildah may rewrite a manifest). An upstream without a
// digest is not copied: `not pinned:`. A copy that does not end with the pinned digest in the store is `not stored:`.
// The copy step (novox/hq ADR 0498 §4; design 51: an upstream image is copied by `mirror` with its pinned digest;
// CAT-BI1): an upstream artifact is copied into the image seat by the seat's verb mirror, which the manifest
// `invokes`, by the digest its manifest pins. The holder copies it, every platform and only what it lacks, and says
// published or refused on the seat's events; the step waits on that outcome, heard through the manifest's `consumes`,
// not on the call, so a large base never outlasts a call's bound. The copy runs no code of a change and reads no
// outside registry itself. An upstream without a digest is not copied: `not pinned:`. A copy the holder does not end
// with the pinned digest stored is `not stored:`.
import (
"bytes"
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"fmt"
"io"
"net/http"
"net/url"
"regexp"
"strings"
"sync"
"time"
"git.novox.be/novox/mesh-catalog/modules/mesh-build-go/buildstep"
reg "git.novox.be/novox/mesh-contracts/go/seats/containerimageregistry"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-catalog/modules/mesh-build-go/buildstep"
)
// CopyTool is the copy step's tool.
const CopyTool = "build_image_copy"
// CopyResult is what a passed copy step says: the pinned digest, now in the store, and the reference to it in the
// form an image's is (<registry>/<module>/<artifact>@sha256:<digest>).
// MirrorWait is how long a copy waits for the holder's outcome before it fails `not stored:`.
const MirrorWait = 30 * time.Minute
// CopyResult is what a passed copy step says: the pinned digest, now held by the image seat, and the seat's reference
// to it (container-image-registry/<path>@sha256:<digest>).
type CopyResult struct {
Artifact string `json:"artifact"`
Digest string `json:"digest"`
@@ -37,19 +39,17 @@ type CopyResult struct {
}
var (
nameOK = regexp.MustCompile(`^[a-z0-9][a-z0-9._-]*(/[a-z0-9][a-z0-9._-]*)*$`)
pinnedOK = regexp.MustCompile(`@(sha256:[0-9a-f]{64})$`)
manifestKind = strings.Join([]string{
"application/vnd.oci.image.index.v1+json", "application/vnd.docker.distribution.manifest.list.v2+json",
"application/vnd.oci.image.manifest.v1+json", "application/vnd.docker.distribution.manifest.v2+json"}, ", ")
nameOK = regexp.MustCompile(`^[a-z0-9][a-z0-9._-]*(/[a-z0-9][a-z0-9._-]*)*$`)
pinnedOK = regexp.MustCompile(`@(sha256:[0-9a-f]{64})$`)
)
func (m *Module) copyTool() StepTool {
return StepTool{
Name: CopyTool,
Description: "Copy an upstream image, registry to registry, into the artifact store by the digest the manifest pins " +
"(novox/hq ADR 0498 §4). Answers a step id at once; the step ends on its key with the digest in the store and the " +
"reference to it. A reference without a digest fails `not pinned:`. Needs no workspace and runs no code of a change.",
Description: "Copy an upstream image into the seat container-image-registry by the digest the manifest pins, through " +
"the seat's mirror (novox/hq ADR 0498 §4, design 51). Answers a step id at once; the step ends on its key with the " +
"digest held and the seat's reference to it, when the holder says published. A reference without a digest fails " +
"`not pinned:`, a refusal `not stored:`. Needs no workspace and runs no code of a change.",
Input: map[string]any{
"type": "object",
"properties": map[string]any{
@@ -64,38 +64,13 @@ func (m *Module) copyTool() StepTool {
}
}
// upstreamRef is an image reference split the way a container runtime does: no host means the public hub.
type upstreamRef struct {
base, repository, digest string
}
func parseUpstream(from string) (upstreamRef, error) {
// pinned is the digest a reference pins.
func pinned(from string) (string, error) {
pin := pinnedOK.FindStringSubmatch(from)
if pin == nil {
return upstreamRef{}, fmt.Errorf("%q names no digest", from)
if pin == nil || strings.TrimSuffix(from, "@"+pin[1]) == "" {
return "", fmt.Errorf("%q names no digest", from)
}
name := strings.TrimSuffix(from, "@"+pin[1])
if colon := strings.LastIndex(name, ":"); colon > strings.LastIndex(name, "/") {
name = name[:colon] // the tag is only what it was called; the digest names the image
}
if name == "" {
return upstreamRef{}, fmt.Errorf("%q is not an image reference", from)
}
host, repository := "docker.io", name
if slash := strings.Index(name, "/"); slash >= 0 && (strings.ContainsAny(name[:slash], ".:") || name[:slash] == "localhost") {
host, repository = name[:slash], name[slash+1:]
}
if host == "docker.io" {
host = "registry-1.docker.io"
if !strings.Contains(repository, "/") {
repository = "library/" + repository
}
}
scheme := "https://"
if strings.HasPrefix(host, "localhost") || strings.HasPrefix(host, "127.") {
scheme = "http://"
}
return upstreamRef{base: scheme + host, repository: repository, digest: pin[1]}, nil
return pin[1], nil
}
func (m *Module) copy(ctx context.Context, args map[string]any) (any, error) {
@@ -105,259 +80,100 @@ func (m *Module) copy(ctx context.Context, args map[string]any) (any, error) {
if !nameOK.MatchString(module) || !nameOK.MatchString(name) {
return nil, fmt.Errorf("%q / %q: a module and an artifact are named in lower case letters, digits, dots, dashes and underscores", module, name)
}
src, err := parseUpstream(from)
digest, err := pinned(from)
if err != nil {
return nil, fmt.Errorf("not pinned: %s: %v; an upstream artifact is copied by the digest its manifest pins", name, err)
}
addr, err := m.registry()
if m.Outcomes == nil {
return nil, fmt.Errorf("%s%s: this process hears no outcome of the image seat", delivery.NotStored, name)
}
// Waited for before the ask, so an outcome said at once is not missed.
heard, stop := m.Outcomes.Expect(digest)
defer stop()
raw, err := m.call()("seat:"+buildstep.ImageSeat+"."+reg.VerbMirror, reg.MirrorInput{Name: module + "/" + name, Source: from, Digest: digest})
if err != nil {
return nil, fmt.Errorf("not stored: %s: %w", name, err)
return nil, fmt.Errorf("%s%s: %s's mirror: %v", delivery.NotStored, name, buildstep.ImageSeat, err)
}
repository := module + "/" + name
c := &copier{m: m, src: src, dstBase: "http://" + addr, dstRepo: repository, done: map[string]bool{}}
if err := m.waitForRegistry(ctx, addr, "copying "+from, func() error { return c.copyManifest(ctx, src.digest) }); err != nil {
return nil, fmt.Errorf("not stored: %s: %w", name, err)
var asked reg.MirrorAnswer
_ = json.Unmarshal(raw, &asked)
m.say("asked %s to copy %s (%s); waiting for its outcome", buildstep.ImageSeat, from, asked.Reference)
wait := m.MirrorWait
if wait == 0 {
wait = MirrorWait
}
if err := c.verify(ctx); err != nil {
return nil, fmt.Errorf("not stored: %s: %w", name, err)
select {
case o := <-heard:
if o.event == reg.EventRefused {
return nil, fmt.Errorf("%s%s: %s refused the copy: %s", delivery.NotStored, name, buildstep.ImageSeat, o.Reason)
}
ref := o.Reference
if ref == "" {
ref = asked.Reference
}
return CopyResult{Artifact: name, Digest: digest, Reference: ref, Pushed: true}, nil
case <-time.After(wait):
return nil, fmt.Errorf("%s%s: %s said no outcome of the copy of %s in %s", delivery.NotStored, name, buildstep.ImageSeat, from, wait)
case <-ctx.Done():
return nil, fmt.Errorf("%s%s: stopped while waiting for the copy of %s: %w", delivery.NotStored, name, from, ctx.Err())
}
return CopyResult{Artifact: name, Digest: src.digest, Reference: addr + "/" + repository + "@" + src.digest, Pushed: true}, nil
}
// copier copies one image from its upstream registry into the store's repository.
type copier struct {
m *Module
src upstreamRef
dstBase string
dstRepo string
token string
done map[string]bool
// outcome is one of the image seat's published or refused, as heard.
type outcome struct {
reg.Outcome
event string
}
func (c *copier) client() *http.Client {
if c.m.HTTP != nil {
return c.m.HTTP
// Outcomes hands the image seat's published and refused of an upstream copy to the copies waiting on its digest.
type Outcomes struct {
mu sync.Mutex
waiting map[string][]chan outcome
}
// Expect waits for the outcome of digest's copy; stop ends the wait.
func (o *Outcomes) Expect(digest string) (<-chan outcome, func()) {
ch := make(chan outcome, 1)
o.mu.Lock()
if o.waiting == nil {
o.waiting = map[string][]chan outcome{}
}
return &http.Client{Timeout: 30 * time.Minute}
}
// get reads from the upstream registry, answering one bearer challenge with an anonymous token.
func (c *copier) get(ctx context.Context, path, accept string) (*http.Response, error) {
for attempt := 0; attempt < 2; attempt++ {
req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.src.base+"/v2/"+c.src.repository+path, nil)
if err != nil {
return nil, err
}
if accept != "" {
req.Header.Set("Accept", accept)
}
if c.token != "" {
req.Header.Set("Authorization", "Bearer "+c.token)
}
resp, err := c.client().Do(req)
if err != nil {
return nil, err
}
if resp.StatusCode == http.StatusUnauthorized && attempt == 0 {
challenge := resp.Header.Get("Www-Authenticate")
resp.Body.Close()
if c.token, err = c.fetchToken(ctx, challenge); err != nil {
return nil, err
o.waiting[digest] = append(o.waiting[digest], ch)
o.mu.Unlock()
return ch, func() {
o.mu.Lock()
defer o.mu.Unlock()
left := o.waiting[digest][:0]
for _, c := range o.waiting[digest] {
if c != ch {
left = append(left, c)
}
continue
}
if resp.StatusCode != http.StatusOK {
resp.Body.Close()
return nil, fmt.Errorf("upstream answered %s for %s%s", resp.Status, c.src.repository, path)
if len(left) == 0 {
delete(o.waiting, digest)
} else {
o.waiting[digest] = left
}
return resp, nil
}
return nil, fmt.Errorf("upstream refused %s%s after a token", c.src.repository, path)
}
var challengeParam = regexp.MustCompile(`(\w+)="([^"]*)"`)
func (c *copier) fetchToken(ctx context.Context, challenge string) (string, error) {
if !strings.HasPrefix(strings.ToLower(challenge), "bearer ") {
return "", fmt.Errorf("upstream asks for credentials (%q) and this copy holds none", challenge)
// Heard takes one event of the seat (published or refused) in: an upstream copy's outcome reaches every copy waiting
// on its digest; any other is passed.
func (o *Outcomes) Heard(event string, body json.RawMessage) error {
var got reg.Outcome
if err := json.Unmarshal(body, &got); err != nil {
return fmt.Errorf("the image seat's %s is not an outcome: %w", event, err)
}
p := map[string]string{}
for _, kv := range challengeParam.FindAllStringSubmatch(challenge, -1) {
p[kv[1]] = kv[2]
}
realm := p["realm"]
if realm == "" {
return "", fmt.Errorf("upstream's challenge %q names no realm", challenge)
}
q := url.Values{"service": {p["service"]}, "scope": {"repository:" + c.src.repository + ":pull"}}
req, err := http.NewRequestWithContext(ctx, http.MethodGet, realm+"?"+q.Encode(), nil)
if err != nil {
return "", err
}
resp, err := c.client().Do(req)
if err != nil {
return "", err
}
defer resp.Body.Close()
var tok struct {
Token string `json:"token"`
Access string `json:"access_token"`
}
if resp.StatusCode != http.StatusOK || json.NewDecoder(resp.Body).Decode(&tok) != nil || (tok.Token == "" && tok.Access == "") {
return "", fmt.Errorf("upstream's token service answered %s", resp.Status)
}
if tok.Token != "" {
return tok.Token, nil
}
return tok.Access, nil
}
// copyManifest copies the manifest of digest and everything it names, children before the manifest itself.
func (c *copier) copyManifest(ctx context.Context, digest string) error {
if c.done[digest] {
if got.Kind != "upstream" || got.Digest == "" {
return nil
}
resp, err := c.get(ctx, "/manifests/"+digest, manifestKind)
if err != nil {
return err
}
body, err := io.ReadAll(io.LimitReader(resp.Body, 16<<20))
resp.Body.Close()
if err != nil {
return err
}
sum := sha256.Sum256(body)
if got := "sha256:" + hex.EncodeToString(sum[:]); got != digest {
return fmt.Errorf("upstream served %s for the manifest pinned as %s", got, digest)
}
var doc struct {
MediaType string `json:"mediaType"`
Config struct{ Digest string } `json:"config"`
Layers []struct{ Digest string } `json:"layers"`
Manifests []struct{ Digest string } `json:"manifests"`
}
if err := json.Unmarshal(body, &doc); err != nil {
return fmt.Errorf("the manifest %s is not JSON: %w", digest, err)
}
for _, child := range doc.Manifests {
if err := c.copyManifest(ctx, child.Digest); err != nil {
return err
o.mu.Lock()
defer o.mu.Unlock()
for _, ch := range o.waiting[got.Digest] {
select {
case ch <- outcome{Outcome: got, event: event}:
default:
}
}
blobs := []string{}
if doc.Config.Digest != "" {
blobs = append(blobs, doc.Config.Digest)
}
for _, l := range doc.Layers {
blobs = append(blobs, l.Digest)
}
for _, b := range blobs {
if err := c.copyBlob(ctx, b); err != nil {
return err
}
}
mediaType := resp.Header.Get("Content-Type")
if mediaType == "" {
mediaType = doc.MediaType
}
if err := c.putManifest(ctx, digest, mediaType, body); err != nil {
return err
}
c.done[digest] = true
return nil
}
func (c *copier) copyBlob(ctx context.Context, digest string) error {
if !digestOK.MatchString(digest) {
return fmt.Errorf("%q is not a blob digest", digest)
}
if c.done[digest] {
return nil
}
head, err := http.NewRequestWithContext(ctx, http.MethodHead, c.dstBase+"/v2/"+c.dstRepo+"/blobs/"+digest, nil)
if err != nil {
return err
}
if r, err := c.client().Do(head); err != nil {
return err
} else {
r.Body.Close()
if r.StatusCode == http.StatusOK {
c.done[digest] = true
return nil
}
}
post, err := http.NewRequestWithContext(ctx, http.MethodPost, c.dstBase+"/v2/"+c.dstRepo+"/blobs/uploads/", nil)
if err != nil {
return err
}
pr, err := c.client().Do(post)
if err != nil {
return err
}
pr.Body.Close()
loc, err := pr.Location()
if pr.StatusCode != http.StatusAccepted || err != nil {
return fmt.Errorf("the store answered %s to a blob upload of %s", pr.Status, digest)
}
q := loc.Query()
q.Set("digest", digest)
loc.RawQuery = q.Encode()
blob, err := c.get(ctx, "/blobs/"+digest, "")
if err != nil {
return err
}
defer blob.Body.Close()
put, err := http.NewRequestWithContext(ctx, http.MethodPut, loc.String(), blob.Body)
if err != nil {
return err
}
put.ContentLength = blob.ContentLength
put.Header.Set("Content-Type", "application/octet-stream")
resp, err := c.client().Do(put)
if err != nil {
return err
}
resp.Body.Close()
if resp.StatusCode != http.StatusCreated {
return fmt.Errorf("the store answered %s to the blob %s", resp.Status, digest)
}
c.done[digest] = true
return nil
}
func (c *copier) putManifest(ctx context.Context, digest, mediaType string, body []byte) error {
req, err := http.NewRequestWithContext(ctx, http.MethodPut, c.dstBase+"/v2/"+c.dstRepo+"/manifests/"+digest, bytes.NewReader(body))
if err != nil {
return err
}
req.Header.Set("Content-Type", mediaType)
resp, err := c.client().Do(req)
if err != nil {
return err
}
resp.Body.Close()
if resp.StatusCode != http.StatusCreated && resp.StatusCode != http.StatusOK {
return fmt.Errorf("the store answered %s to the manifest %s", resp.Status, digest)
}
return nil
}
// verify reads back what the store serves for the pinned digest: only the store can say.
func (c *copier) verify(ctx context.Context) error {
req, err := http.NewRequestWithContext(ctx, http.MethodHead, c.dstBase+"/v2/"+c.dstRepo+"/manifests/"+c.src.digest, nil)
if err != nil {
return err
}
req.Header.Set("Accept", manifestKind)
resp, err := c.client().Do(req)
if err != nil {
return err
}
resp.Body.Close()
if got := resp.Header.Get("Docker-Content-Digest"); resp.StatusCode != http.StatusOK || got != c.src.digest {
return fmt.Errorf("the store answered %s with digest %q for the pinned %s", resp.Status, got, c.src.digest)
}
return nil
}
@@ -1,116 +1,15 @@
package main
import (
"crypto/sha256"
"encoding/hex"
"encoding/json"
"io"
"net/http"
"net/http/httptest"
"strings"
"sync"
"testing"
"time"
reg "git.novox.be/novox/mesh-contracts/go/seats/containerimageregistry"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
func digestOf(b []byte) string { s := sha256.Sum256(b); return "sha256:" + hex.EncodeToString(s[:]) }
// memRegistry is a registry in memory: blobs and manifests by digest, per repository.
type memRegistry struct {
mu sync.Mutex
blobs map[string][]byte // repo|digest
manifests map[string][]byte // repo|digest
types map[string]string
puts []string
srv *httptest.Server
}
func newMem(t *testing.T) *memRegistry {
r := &memRegistry{blobs: map[string][]byte{}, manifests: map[string][]byte{}, types: map[string]string{}}
r.srv = httptest.NewServer(http.HandlerFunc(r.serve))
t.Cleanup(r.srv.Close)
return r
}
func (r *memRegistry) addr() string { return r.srv.Listener.Addr().String() }
func (r *memRegistry) serve(w http.ResponseWriter, req *http.Request) {
r.mu.Lock()
defer r.mu.Unlock()
path := strings.TrimPrefix(req.URL.Path, "/v2/")
if i := strings.Index(path, "/blobs/uploads/"); i >= 0 {
if req.Method == http.MethodPost {
w.Header().Set("Location", "/v2/"+path[:i]+"/blobs/uploads/u1")
w.WriteHeader(http.StatusAccepted)
return
}
body, _ := io.ReadAll(req.Body)
d := req.URL.Query().Get("digest")
if digestOf(body) != d {
w.WriteHeader(http.StatusBadRequest)
return
}
r.blobs[path[:i]+"|"+d] = body
r.puts = append(r.puts, "blob "+d)
w.WriteHeader(http.StatusCreated)
return
}
for _, kind := range []string{"blobs", "manifests"} {
if i := strings.Index(path, "/"+kind+"/"); i >= 0 {
key := path[:i] + "|" + path[i+len(kind)+2:]
store := r.blobs
if kind == "manifests" {
store = r.manifests
}
if req.Method == http.MethodPut {
body, _ := io.ReadAll(req.Body)
if digestOf(body) != path[i+len(kind)+2:] {
w.WriteHeader(http.StatusBadRequest)
return
}
r.manifests[key] = body
r.types[key] = req.Header.Get("Content-Type")
r.puts = append(r.puts, "manifest "+path[i+len(kind)+2:])
w.WriteHeader(http.StatusCreated)
return
}
body, ok := store[key]
if !ok {
http.NotFound(w, req)
return
}
w.Header().Set("Docker-Content-Digest", path[i+len(kind)+2:])
w.Header().Set("Content-Type", r.types[key])
if req.Method == http.MethodHead {
return
}
w.Write(body)
return
}
}
http.NotFound(w, req)
}
// upstreamImage puts a two-platform index (two manifests, each with a config and a layer) into r and answers the index digest.
func upstreamImage(r *memRegistry, repo string) string {
var children []map[string]string
for _, tag := range []string{"amd64", "arm64"} {
cfg, layer := []byte(`{"arch":"`+tag+`"}`), []byte("layer-"+tag)
r.blobs[repo+"|"+digestOf(cfg)], r.blobs[repo+"|"+digestOf(layer)] = cfg, layer
mf, _ := json.Marshal(map[string]any{"schemaVersion": 2, "mediaType": "application/vnd.oci.image.manifest.v1+json",
"config": map[string]string{"digest": digestOf(cfg)}, "layers": []map[string]string{{"digest": digestOf(layer)}}})
r.manifests[repo+"|"+digestOf(mf)] = mf
r.types[repo+"|"+digestOf(mf)] = "application/vnd.oci.image.manifest.v1+json"
children = append(children, map[string]string{"digest": digestOf(mf)})
}
idx, _ := json.Marshal(map[string]any{"schemaVersion": 2, "mediaType": "application/vnd.oci.image.index.v1+json", "manifests": children})
d := digestOf(idx)
r.manifests[repo+"|"+d] = idx
r.types[repo+"|"+d] = "application/vnd.oci.image.index.v1+json"
return d
}
func runCopy(t *testing.T, m *Module, states *keptStates, args map[string]any) (CopyResult, delivery.Step) {
t.Helper()
if _, err := tool(t, m, CopyTool)(args); err != nil {
@@ -126,93 +25,55 @@ func runCopy(t *testing.T, m *Module, states *keptStates, args map[string]any) (
return res, st
}
// ADR 0498 §4 / AC: the copy leaves the pinned digest in the store and answers it, with the whole index (every
// manifest it lists, every blob), and runs no buildah.
func TestTheCopyLeavesThePinnedDigestInTheStoreAndAnswersIt(t *testing.T) {
up, store := newMem(t), newMem(t)
digest := upstreamImage(up, "library/nats")
f := newFake(t)
states := newStates()
m := moduleWith(t, f, states)
m.HTTP = nil
reg := store.addr()
writeFile(t, m.RegistryFile, reg+"\n")
from := up.addr() + "/library/nats:2.11.17-alpine@" + digest
res, st := runCopy(t, m, states, map[string]any{"module": "nats", "artifact": "server", "from": from})
if st.State != delivery.StepPassed {
t.Fatalf("step = %+v", st)
}
if res.Digest != digest || res.Reference != reg+"/nats/server@"+digest || !res.Pushed || res.Artifact != "server" {
t.Errorf("result = %+v", res)
}
if _, ok := store.manifests["nats/server|"+digest]; !ok {
t.Fatal("the index is not in the store")
}
if len(store.manifests) != 3 || len(store.blobs) != 4 {
t.Errorf("store holds %d manifests and %d blobs, want 3 and 4", len(store.manifests), len(store.blobs))
}
if len(f.lines()) != 0 {
t.Errorf("the copy ran commands: %v", f.lines())
}
// A second copy writes nothing new but the manifests.
store.puts = nil
if _, st := runCopy(t, m, states, map[string]any{"module": "nats", "artifact": "server", "from": from}); st.State != delivery.StepPassed {
t.Fatalf("again: %+v", st)
}
for _, p := range store.puts {
if strings.HasPrefix(p, "blob ") {
t.Errorf("a blob the store holds was sent again: %s", p)
}
}
}
// ADR 0498 §4 / AC: an upstream without a digest fails `not pinned:`, and nothing is read from the upstream.
func TestAnUpstreamWithoutADigestFailsNotPinned(t *testing.T) {
up, store := newMem(t), newMem(t)
upstreamImage(up, "library/nats")
f := newFake(t)
states := newStates()
m := moduleWith(t, f, states)
writeFile(t, m.RegistryFile, store.addr()+"\n")
_, st := runCopy(t, m, states, map[string]any{"module": "nats", "artifact": "server", "from": up.addr() + "/library/nats:2.11.17-alpine"})
if st.State != delivery.StepFailed || !strings.Contains(st.Reason, "not pinned: server") {
t.Fatalf("step = %+v", st)
}
if len(store.puts) != 0 {
t.Errorf("the store was written: %v", store.puts)
}
}
// A manifest the upstream serves under another digest than the pin is not copied; the failure is `not stored:`.
func TestAnUpstreamServingAnotherManifestThanPinnedIsNotStored(t *testing.T) {
up, store := newMem(t), newMem(t)
d := upstreamImage(up, "library/nats")
up.manifests["library/nats|"+d] = []byte(`{"tampered":true}`)
f := newFake(t)
states := newStates()
m := moduleWith(t, f, states)
writeFile(t, m.RegistryFile, store.addr()+"\n")
_, st := runCopy(t, m, states, map[string]any{"module": "nats", "artifact": "server", "from": up.addr() + "/library/nats@" + d})
if st.State != delivery.StepFailed || !strings.HasPrefix(st.Reason, "not stored: server:") {
t.Fatalf("step = %+v", st)
}
if len(store.manifests) != 0 {
t.Errorf("the store holds %v", store.manifests)
}
}
func TestAShortNameIsQualifiedAsTheHubsAndAHostIsKept(t *testing.T) {
// ADR 0498 §4, design 51 (CAT-BI1): an upstream copy is the image seat's mirror by the pinned digest; the step ends on
// the holder's outcome on the seat's events, answering its reference, and runs no command. A refusal, no outcome in the
// bound and no pin each fail the step, the last before the seat is asked.
func TestTheCopyIsTheImageSeatsMirrorAndEndsOnItsOutcome(t *testing.T) {
d := "sha256:" + strings.Repeat("a", 64)
for from, want := range map[string]upstreamRef{
"alpine@" + d: {"https://registry-1.docker.io", "library/alpine", d},
"nats:2.11.17-alpine@" + d: {"https://registry-1.docker.io", "library/nats", d},
"grafana/grafana:11@" + d: {"https://registry-1.docker.io", "grafana/grafana", d},
"ghcr.io/org/app:1@" + d: {"https://ghcr.io", "org/app", d},
"localhost:5000/x/y@" + d: {"http://localhost:5000", "x/y", d},
from := "nats:2.11.17-alpine@" + d
for name, c := range map[string]struct {
from string
mirrored func(reg.MirrorInput) (string, reg.Outcome)
reason string
}{
"published": {from: from},
"refused": {from: from, mirrored: func(in reg.MirrorInput) (string, reg.Outcome) {
return reg.EventRefused, reg.Outcome{Kind: "upstream", Digest: in.Digest, Reason: "the upstream answered 404"}
}, reason: "not stored: server: container-image-registry refused the copy: the upstream answered 404"},
"another digest's outcome is not this copy's": {from: from, mirrored: func(in reg.MirrorInput) (string, reg.Outcome) {
return reg.EventPublished, reg.Outcome{Kind: "upstream", Digest: "sha256:" + strings.Repeat("b", 64)}
}, reason: "said no outcome of the copy"},
"a published image is no copy's outcome": {from: from, mirrored: func(in reg.MirrorInput) (string, reg.Outcome) {
return reg.EventPublished, reg.Outcome{Kind: "image", Digest: in.Digest}
}, reason: "said no outcome of the copy"},
"no pin": {from: "nats:2.11.17-alpine", reason: "not pinned: server"},
} {
got, err := parseUpstream(from)
if err != nil || got != want {
t.Errorf("%s = %+v, %v; want %+v", from, got, err, want)
}
t.Run(name, func(t *testing.T) {
f := newFake(t)
f.mirrored = c.mirrored
states := newStates()
m := moduleWith(t, f, states)
m.MirrorWait = 200 * time.Millisecond
res, st := runCopy(t, m, states, map[string]any{"module": "nats", "artifact": "server", "from": c.from})
if c.reason != "" {
if st.State != delivery.StepFailed || !strings.Contains(st.Reason, c.reason) {
t.Fatalf("step = %+v, want failed with %q", st, c.reason)
}
if c.reason == "not pinned: server" && len(f.mirrors) != 0 {
t.Errorf("an unpinned copy asked the seat: %v", f.mirrors)
}
return
}
if st.State != delivery.StepPassed || res.Digest != d || !res.Pushed ||
res.Reference != reg.Seat+"/upstream/nats/server@"+d {
t.Fatalf("step = %+v result = %+v", st, res)
}
if len(f.mirrors) != 1 || f.mirrors[0] != (reg.MirrorInput{Name: "nats/server", Source: from, Digest: d}) {
t.Errorf("mirrors = %+v", f.mirrors)
}
if len(f.lines()) != 0 {
t.Errorf("the copy ran commands: %v", f.lines())
}
})
}
}
@@ -144,7 +144,7 @@ func TestForgeFilesRenderFromTheBoundValuesInTheModulesOwnHome(t *testing.T) {
// template in the files/ archive (novox/hq ADR 0349 §2).
func TestManifestRequiresForgeAccountKeepsKeysAndTakesFilesFromTemplates(t *testing.T) {
m := readForgeManifest(t)
if want := []any{"forge-account", "rootless-image-build", "artifact-store", "node-tools"}; !reflect.DeepEqual(m.Requires, want) {
if want := []any{"forge-account", "rootless-image-build", "node-tools"}; !reflect.DeepEqual(m.Requires, want) {
t.Errorf("requires = %v, want %v", m.Requires, want)
}
keeps := false
@@ -2,9 +2,10 @@ package main
import (
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"net/http"
"net/http/httptest"
"fmt"
"os"
"path/filepath"
"strings"
@@ -12,6 +13,7 @@ import (
"testing"
"time"
reg "git.novox.be/novox/mesh-contracts/go/seats/containerimageregistry"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/forgekey"
)
@@ -64,46 +66,117 @@ type cmdRecord struct {
line string
}
// fake stands in for git and buildah at the process boundary and for the registry over HTTP.
// fake stands in for git and buildah at the process boundary and for the image seat's holder at the seat's verbs.
type fake struct {
mu sync.Mutex
calls []cmdRecord
digest string // what `buildah images` prints
pushes int
refuse int // the first n pushes answer "connection refused"
digest string // what `buildah images` prints
pushes int // layouts written
failBud forgekey.Ran // what a failing buildah bud answers; a zero Status is a passing bud
// digestFile: "" writes the digest, "none" writes no file, "empty" writes an empty one.
// digestFile: "" writes the layout's digest, "none" writes no file, "other" writes another digest.
digestFile string
// registry: repository:tag -> digest
registry map[string]string
server *httptest.Server
// pushed is the manifest digest of the last layout written; parts its parts' digests, the manifest last.
pushed string
parts []string
// The holder: held are the parts it holds already; refuse fails every publish; published the publishes it was
// asked, sent the parts it was sent whole; mirrors the mirrors asked, answered by mirrored (published when nil).
held map[string]bool
refuse error
published []reg.PublishInput
sent []string
mirrors []reg.MirrorInput
mirrored func(reg.MirrorInput) (string, reg.Outcome)
outcomes *Outcomes
}
const imageDigest = "sha256:1111111111111111111111111111111111111111111111111111111111111111"
func newFake(t *testing.T) *fake {
f := &fake{digest: imageDigest, registry: map[string]string{}}
f.server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
f.mu.Lock()
defer f.mu.Unlock()
key := strings.TrimPrefix(r.URL.Path, "/v2/")
i := strings.Index(key, "/manifests/")
if r.Method != http.MethodHead || i < 0 {
http.NotFound(w, r)
return
}
d, ok := f.registry[key[:i]+":"+key[i+len("/manifests/"):]]
if !ok {
http.NotFound(w, r)
return
}
w.Header().Set("Docker-Content-Digest", d)
}))
t.Cleanup(f.server.Close)
return f
return &fake{digest: imageDigest, held: map[string]bool{}}
}
func (f *fake) address() string { return f.server.Listener.Addr().String() }
// address is the registry part of the seat's reference: the seat's name.
func (f *fake) address() string { return reg.Seat }
func sha(b []byte) string {
s := sha256.Sum256(b)
return "sha256:" + hex.EncodeToString(s[:])
}
// writeLayout writes an OCI layout of one config, one layer and their manifest into dir, as buildah push oci: does.
func (f *fake) writeLayout(dir, digestFile string) {
blobs := filepath.Join(dir, "blobs", "sha256")
_ = os.MkdirAll(blobs, 0o755)
put := func(b []byte) map[string]any {
d := sha(b)
_ = os.WriteFile(filepath.Join(blobs, strings.TrimPrefix(d, "sha256:")), b, 0o644)
return map[string]any{"digest": d, "size": len(b)}
}
config := put([]byte(`{"architecture":"amd64","os":"linux"}`))
layer := put([]byte("the layer's bytes"))
config["mediaType"], layer["mediaType"] = "application/vnd.oci.image.config.v1+json", "application/vnd.oci.image.layer.v1.tar+gzip"
raw, _ := json.Marshal(map[string]any{"schemaVersion": 2, "mediaType": "application/vnd.oci.image.manifest.v1+json",
"config": config, "layers": []any{layer}})
manifest := put(raw)
manifest["mediaType"] = "application/vnd.oci.image.manifest.v1+json"
index, _ := json.Marshal(map[string]any{"schemaVersion": 2, "manifests": []any{manifest}})
_ = os.WriteFile(filepath.Join(dir, "index.json"), index, 0o644)
f.pushed = manifest["digest"].(string)
f.parts = []string{config["digest"].(string), layer["digest"].(string), f.pushed}
switch f.digestFile {
case "":
_ = os.WriteFile(digestFile, []byte(f.pushed), 0o644)
case "other":
_ = os.WriteFile(digestFile, []byte(imageDigest), 0o644)
}
}
// call is the holder's side of the seat's verbs.
func (f *fake) call(key string, body any) (json.RawMessage, error) {
raw, _ := json.Marshal(body)
f.mu.Lock()
defer f.mu.Unlock()
switch key {
case "seat:" + reg.Seat + "." + reg.VerbPublish:
var in reg.PublishInput
_ = json.Unmarshal(raw, &in)
f.published = append(f.published, in)
if f.refuse != nil {
return nil, f.refuse
}
a := reg.PublishAnswer{Transfer: in.Transfer}
for _, p := range in.Parts {
if !f.held[p.Digest] {
a.Lacks = append(a.Lacks, p.Digest)
}
}
a.Held = len(a.Lacks) == 0
return json.Marshal(a)
case "seat:" + reg.Seat + "." + reg.VerbPublishPart:
var in reg.PublishPartInput
_ = json.Unmarshal(raw, &in)
if in.Last {
f.sent = append(f.sent, in.Part)
}
return json.Marshal(reg.PublishPartAnswer{Transfer: in.Transfer, Part: in.Part, Complete: in.Last})
case "seat:" + reg.Seat + "." + reg.VerbMirror:
var in reg.MirrorInput
_ = json.Unmarshal(raw, &in)
f.mirrors = append(f.mirrors, in)
event, o := reg.EventPublished, reg.Outcome{Kind: "upstream", Name: "upstream/" + in.Name, Digest: in.Digest,
Holder: "distribution", Reference: reg.Seat + "/upstream/" + in.Name + "@" + in.Digest}
if f.mirrored != nil {
event, o = f.mirrored(in)
}
if event != "" && f.outcomes != nil {
// Said after the answer, on the seat's events, as the holder's watch says it.
b, _ := json.Marshal(o)
go func() { time.Sleep(time.Millisecond); _ = f.outcomes.Heard(event, b) }()
}
return json.Marshal(reg.MirrorAnswer{Digest: in.Digest, Reference: reg.Seat + "/upstream/" + in.Name + "@" + in.Digest})
}
return nil, fmt.Errorf("no verb %s", key)
}
func (f *fake) run(ctx context.Context, env []string, name string, args ...string) forgekey.Ran {
line := name + " " + strings.Join(args, " ")
@@ -124,28 +197,17 @@ func (f *fake) run(ctx context.Context, env []string, name string, args ...strin
case name == "buildah" && strings.Contains(line, " push "):
f.mu.Lock()
defer f.mu.Unlock()
if f.refuse > 0 {
f.refuse--
return forgekey.Ran{Status: 125, Stderr: "Error: dial tcp 127.0.0.1:5000: connect: connection refused\n"}
}
f.pushes++
var digestFile, dest string
var digestFile, dir string
for i, a := range args {
if a == "--digestfile" {
digestFile = args[i+1]
}
if strings.HasPrefix(a, "docker://") {
dest = strings.TrimPrefix(a, "docker://")
if strings.HasPrefix(a, "oci:") {
dir = strings.TrimPrefix(a, "oci:")
}
}
dest = strings.TrimPrefix(dest, f.address()+"/")
f.registry[dest] = f.digest
switch f.digestFile {
case "":
os.WriteFile(digestFile, []byte(f.digest), 0o644)
case "empty":
os.WriteFile(digestFile, nil, 0o644)
}
f.writeLayout(dir, digestFile)
}
return forgekey.Ran{}
}
@@ -186,14 +248,13 @@ func moduleWith(t *testing.T, f *fake, states *keptStates) *Module {
}
m := &Module{
Node: "ace", Account: "mesh-build-image", UID: "1234", Home: t.TempDir(), Work: t.TempDir(), Store: t.TempDir(),
Steps: states.putStep, Ctx: context.Background(), Run: f.run, HTTP: f.server.Client(),
SubUIDFile: write("subuid", "someone:100000:65536\nmesh-build-image:624288:65536\n"),
SubGIDFile: write("subgid", "mesh-build-image:624288:65536\n"),
RegistryFile: write("registry", f.address()+"\n"),
Tun: write("tun", ""),
RegistryWait: 2 * time.Second,
RegistryPause: 5 * time.Millisecond,
Steps: states.putStep, Ctx: context.Background(), Run: f.run, Call: f.call, Outcomes: &Outcomes{},
SubUIDFile: write("subuid", "someone:100000:65536\nmesh-build-image:624288:65536\n"),
SubGIDFile: write("subgid", "mesh-build-image:624288:65536\n"),
Tun: write("tun", ""),
MirrorWait: 5 * time.Second,
}
f.outcomes = m.Outcomes
return m
}
@@ -9,6 +9,7 @@ package main
import (
"context"
"fmt"
reg "git.novox.be/novox/mesh-contracts/go/seats/containerimageregistry"
"os"
"os/user"
"strings"
@@ -44,7 +45,7 @@ func main() {
if u, err := user.Current(); err == nil {
m.Home, m.Account, m.UID = u.HomeDir, u.Username, u.Uid
}
m.RegistryFile = m.Home + "/registry"
m.Outcomes = &Outcomes{}
m.Settings = m.Home + "/settings.json"
m.Run = StepRunner(m.Home)
// Outside a launch by the runtime — a build, a check — there is no node and no state to write.
@@ -66,6 +67,10 @@ func main() {
// started before booted.
go m.endLeftRunning(ctx, m.stepScan(), booted, logf)
}
if m.Node != "" {
// The image seat's outcomes, which a copy waits on (the manifest's consumes).
go hearOutcomes(ctx, m.Outcomes)
}
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) }()
@@ -75,3 +80,28 @@ func main() {
os.Exit(1)
}
}
// OutcomeEvents are the image seat's events a copy's end is said by, as the manifest consumes them.
var OutcomeEvents = []string{reg.EventPublished, reg.EventRefused}
// hearOutcomes subscribes to the image seat's published and refused, asking again until the runtime takes each.
func hearOutcomes(ctx context.Context, o *Outcomes) {
for _, event := range OutcomeEvents {
pattern := reg.Seat + "." + event
go func() {
for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
err := stdio.Subscribe(pattern, func(e stdio.Envelope) error { return o.Heard(event, e.Body) })
if err == nil {
logf("listening for %s", pattern)
return
}
logf("not hearing %s yet (%v); asking again in %s", pattern, err, wait)
select {
case <-ctx.Done():
return
case <-time.After(wait):
}
}
}()
}
}
@@ -123,14 +123,6 @@ func TestManifestKeepsItsStatesClaimsItsBuildSeatAndDeclaresItsHealthTool(t *tes
if !strings.Contains(string(raw), `"tool": "`+HealthTool+`"`) {
t.Errorf("runtime-health does not name %s", HealthTool)
}
reg := resource(t, m, "registry")
if reg["path"] != "${dir:state}/registry" {
t.Errorf("registry file = %v, want ${dir:state}/registry (the program reads <home>/registry)", reg)
}
tpl, _ := os.ReadFile(filepath.Join(moduleDir, "files", "registry"))
if strings.TrimSpace(string(tpl)) != "${seat:mesh-artifact-store:reach}" {
t.Errorf("registry template = %q", tpl)
}
raw, _ = os.ReadFile(filepath.Join(moduleDir, "module.json"))
var doc struct {
Config map[string]struct {
@@ -4,7 +4,6 @@ import (
"context"
"errors"
"fmt"
"net/http"
"sync"
"time"
@@ -42,17 +41,16 @@ type Module struct {
SubUIDFile, SubGIDFile string
// Tun is the tun device buildah's network for a RUN line (pasta) opens; TunDevice when empty.
Tun string
// RegistryFile holds the registry's address, host:port, which the manifest binds from the artifact-store provision.
RegistryFile string
// Call asks a seat's verb (the image seat's publish, publish-part and mirror); the runtime's stdio.Ask when nil.
Call buildstep.SeatCall
// Outcomes are the image seat's published and refused as this process hears them, which a copy waits on.
Outcomes *Outcomes
// RemoteBase is where repositories are fetched from, "ssh://mesh-forge/" when empty; FetchOptions are the options of
// the SDK's fetch and merge. A test points them at a local repository (buildstep.AllowFileRemote).
RemoteBase string
FetchOptions []buildstep.Option
// HTTP talks to the registry; a client of its own when nil. RegistryWait and RegistryPause bound the wait for a
// registry held still (RegistryWait and RegistryFirstPause when zero).
HTTP *http.Client
RegistryWait time.Duration
RegistryPause time.Duration
// MirrorWait bounds a copy's wait for its outcome on the image seat's events (MirrorWait when zero).
MirrorWait time.Duration
// Now is the clock; time.Now when nil.
Now func() time.Time
// Floor refuses every fetch and build while the work or the image store's filesystem has less free than it (issue
@@ -1,161 +0,0 @@
package main
// The trunk push (novox/hq ADR 0487 §5; issue 1703d): after the build has ended, from the module's own process and not
// from a Dockerfile step, the image is pushed to the registry under a tag of its own and the digest is read back from
// the registry, which is the only one that can say what it serves for the reference (issue 1742). The push and the
// read wait out a registry held still by the store's nightly collection (issue 457).
import (
"context"
"errors"
"fmt"
"net/http"
"os"
"path/filepath"
"strings"
"syscall"
"time"
"git.novox.be/novox/mesh-sdk/go/forgekey"
)
// RegistryWait is how long a call waits for a registry that refuses connections, longer than the nightly collection
// holds it still (about a minute and a half); RegistryFirstPause is the first pause, doubling to RegistryMostPause.
const (
RegistryWait = 5 * time.Minute
RegistryFirstPause = time.Second
RegistryMostPause = 10 * time.Second
)
// registry is the registry's address, host:port, as the manifest bound it into RegistryFile.
func (m *Module) registry() (string, error) {
raw, err := os.ReadFile(m.RegistryFile)
if err != nil {
return "", fmt.Errorf("the registry's address cannot be read: %w", err)
}
addr := strings.TrimSpace(string(raw))
if addr == "" || strings.Contains(addr, "${") || strings.HasPrefix(addr, ":") || strings.HasSuffix(addr, ":") {
return "", fmt.Errorf("the registry's address in %s is %q, not host:port", m.RegistryFile, addr)
}
return addr, nil
}
func (m *Module) push(ctx context.Context, workspace, module, local string, res BuildResult, tag string) (any, error) {
addr, err := m.registry()
if err != nil {
return nil, err
}
repository := module + "/" + res.Artifact
dest := addr + "/" + repository + ":" + tag
digestFile := filepath.Join(m.Work, workspace, "pushed."+tag)
argv := append(m.storeFlags(), "push", "--tls-verify=false", "--digestfile", digestFile, local, "docker://"+dest)
var last forgekey.Ran
if err := m.waitForRegistry(ctx, addr, "buildah push "+dest, func() error {
last = m.Run(ctx, m.env(), "buildah", argv...)
if last.Status != 0 || last.Err != "" {
return errors.New(cause(last))
}
return nil
}); err != nil {
m.sayFailure("buildah push "+dest, last)
return nil, fmt.Errorf("pushing %s: %w", dest, err)
}
var served string
if err := m.waitForRegistry(ctx, addr, "reading "+dest, func() error {
var err error
served, err = m.registryDigest(ctx, addr, repository, tag)
return err
}); err != nil {
return nil, fmt.Errorf("pushed %s and cannot read back what the registry serves for it: %w", dest, err)
}
// A push that succeeded wrote the digest it made: a file missing, unreadable or empty is a fault, and the registry's
// digest is never answered unchecked.
raw, err := os.ReadFile(digestFile)
if err != nil {
return nil, fmt.Errorf("pushed %s, but the digest buildah pushed cannot be read from %s, so the registry's %s cannot be checked: %w", dest, digestFile, served, err)
}
pushed := strings.TrimSpace(string(raw))
if !digestOK.MatchString(pushed) {
return nil, fmt.Errorf("pushed %s, but the digest file %s holds %q, not a digest, so the registry's %s cannot be checked", dest, digestFile, pushed, served)
}
if pushed != served {
return nil, fmt.Errorf("pushed %s as %s, but the registry serves %s for it", dest, pushed, served)
}
res.Digest, res.Pushed = served, true
res.Reference = addr + "/" + repository + "@" + served
return res, nil
}
// registryDigest is the digest the registry answers for repository:tag (Docker-Content-Digest of a HEAD).
func (m *Module) registryDigest(ctx context.Context, addr, repository, tag string) (string, error) {
req, err := http.NewRequestWithContext(ctx, http.MethodHead, "http://"+addr+"/v2/"+repository+"/manifests/"+tag, nil)
if err != nil {
return "", err
}
req.Header.Set("Accept", strings.Join([]string{
"application/vnd.oci.image.manifest.v1+json", "application/vnd.oci.image.index.v1+json",
"application/vnd.docker.distribution.manifest.v2+json", "application/vnd.docker.distribution.manifest.list.v2+json"}, ", "))
client := m.HTTP
if client == nil {
client = &http.Client{Timeout: time.Minute}
}
resp, err := client.Do(req)
if err != nil {
return "", err
}
resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return "", fmt.Errorf("the registry answered %s for %s:%s", resp.Status, repository, tag)
}
d := resp.Header.Get("Docker-Content-Digest")
if !digestOK.MatchString(d) {
return "", fmt.Errorf("the registry named no digest for %s:%s (Docker-Content-Digest %q)", repository, tag, d)
}
return d, nil
}
// refused is whether an error is a connection refused: from a dial here, or as buildah said it in its output.
func refused(err error) bool {
return err != nil && (errors.Is(err, syscall.ECONNREFUSED) || strings.Contains(err.Error(), "connection refused"))
}
// waitForRegistry runs try, and while it fails because the registry at addr refuses connections, tries again with a
// growing pause until RegistryWait has passed (novox/hq issue 457). A refusal is the one error waited for: nothing
// was sent, so trying again does nothing twice. what names the call, for what the module says.
func (m *Module) waitForRegistry(ctx context.Context, addr, what string, try func() error) error {
err := try()
if !refused(err) {
return err
}
limit, pause := m.RegistryWait, m.RegistryPause
if limit == 0 {
limit = RegistryWait
}
if pause == 0 {
pause = RegistryFirstPause
}
started := time.Now()
m.say("%s: refused; waiting for the registry at %s for up to %s — the store's nightly collection holds it still for about a minute and a half (issue 457)", what, addr, limit)
for {
left := limit - time.Since(started)
if left <= 0 {
return fmt.Errorf("the registry at %s refused every connection for %s, longer than its nightly collection holds it still, so it is down, not paused: %w", addr, time.Since(started).Round(time.Millisecond), err)
}
select {
case <-ctx.Done():
return fmt.Errorf("stopped while waiting for the registry at %s: %w (last: %v)", addr, ctx.Err(), err)
case <-time.After(min(pause, left)):
}
pause = min(pause*2, RegistryMostPause)
if err = try(); !refused(err) {
m.say("%s: the registry at %s answers again after %s", what, addr, time.Since(started).Round(time.Millisecond))
return err
}
}
}
func (m *Module) say(format string, a ...any) {
if m.Say != nil {
m.Say(format, a...)
}
}
@@ -1,156 +0,0 @@
package main
import (
"context"
"os"
"strings"
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/forgekey"
)
// 1703d/AC (ADR 0487 §5): a candidate's build leaves the registry unchanged: nothing is pushed, nothing is asked of it.
func TestACandidatesBuildLeavesTheRegistryUnchanged(t *testing.T) {
m, f, states := setup(t, same)
res, st := runBuild(t, m, states, nil) // no publish
if st.State != delivery.StepPassed || res.Pushed {
t.Fatalf("step = %+v result = %+v", st, res)
}
if len(f.called("push")) != 0 || len(f.registry) != 0 || f.pushes != 0 {
t.Errorf("a candidate's build reached the registry: pushes %d, registry %v, calls %v", f.pushes, f.registry, f.lines())
}
}
// 1703d/AC: a trunk build adds exactly one digest, under <module>/<artifact> and a tag of its own (the digest's first 12
// hex digits), after the build has ended, with TLS verification off (plain HTTP) and no credential; the key carries the
// digest the registry serves and the reference pinned by it.
func TestATrunkBuildPushesOnceAndAnswersTheRegistrysDigest(t *testing.T) {
m, f, states := setup(t, same)
res, st := runBuild(t, m, states, map[string]any{"publish": true})
if st.State != delivery.StepPassed {
t.Fatalf("step = %+v", st)
}
tag := "111111111111"
if len(f.registry) != 1 || f.registry["fixture-image/server:"+tag] != imageDigest || f.pushes != 1 {
t.Fatalf("registry = %v after %d pushes, want exactly fixture-image/server:%s", f.registry, f.pushes, tag)
}
if !res.Pushed || res.Digest != imageDigest || res.Reference != f.address()+"/fixture-image/server@"+imageDigest {
t.Errorf("result = %+v", res)
}
push := f.called("push")
if len(push) != 1 || !strings.Contains(push[0].line, "--tls-verify=false") || !strings.Contains(push[0].line, "docker://"+f.address()+"/fixture-image/server:"+tag) {
t.Fatalf("push = %v", push)
}
for _, bad := range []string{"--creds", "--authfile", "--sign"} {
if strings.Contains(push[0].line, bad) {
t.Errorf("the push carries %s", bad)
}
}
// The push comes after the build has ended: bud, then images, then push.
order := []string{}
for _, l := range f.lines() {
for _, v := range []string{" bud ", " images ", " push ", " rmi "} {
if strings.Contains(l, "buildah") && strings.Contains(l, v) {
order = append(order, strings.TrimSpace(v))
}
}
}
if strings.Join(order, ",") != "bud,images,push,rmi" {
t.Errorf("order = %v", order)
}
}
// A tag the call gives is the one used; one that is not a tag is refused.
func TestATagOfTheCallsOwn(t *testing.T) {
m, f, states := setup(t, same)
if _, st := runBuild(t, m, states, map[string]any{"publish": true, "tag": "fp-abc123"}); st.State != delivery.StepPassed {
t.Fatalf("step = %+v", st)
}
if _, ok := f.registry["fixture-image/server:fp-abc123"]; !ok || len(f.registry) != 1 {
t.Errorf("registry = %v", f.registry)
}
m2, f2, states2 := setup(t, same)
if _, st := runBuild(t, m2, states2, map[string]any{"publish": true, "tag": "a/b:c"}); st.State != delivery.StepFailed || len(f2.called("bud")) != 0 {
t.Errorf("step = %+v", st)
}
}
// issue 457: a registry held still (connection refused) is waited for, and the push then goes through once.
func TestAPushWaitsForARegistryHeldStill(t *testing.T) {
m, f, states := setup(t, same)
f.refuse = 2
res, st := runBuild(t, m, states, map[string]any{"publish": true})
if st.State != delivery.StepPassed || !res.Pushed {
t.Fatalf("step = %+v result = %+v", st, res)
}
if f.pushes != 1 || len(f.registry) != 1 || len(f.called("push")) != 3 {
t.Errorf("pushes %d, registry %v, push calls %d; want two refused and one made", f.pushes, f.registry, len(f.called("push")))
}
}
// A registry that stays down past the wait fails the step saying how long it was refused; nothing is claimed pushed.
func TestARegistryDownPastTheWaitFailsTheStep(t *testing.T) {
m, f, states := setup(t, same)
f.refuse = 1 << 20
m.RegistryWait = 50 * time.Millisecond
_, st := runBuild(t, m, states, map[string]any{"publish": true})
if st.State != delivery.StepFailed || !strings.Contains(st.Reason, "refused every connection") {
t.Errorf("step = %+v", st)
}
if len(f.registry) != 0 {
t.Errorf("registry = %v", f.registry)
}
}
// The digest the registry serves is the answer: when it differs from the one the push reported, the step fails.
func TestAnAnswerThatDisagreesWithThePushFails(t *testing.T) {
m, f, states := setup(t, same)
run := m.Run
m.Run = func(ctx context.Context, env []string, name string, args ...string) forgekey.Ran {
r := run(ctx, env, name, args...)
if name == "buildah" && strings.Contains(strings.Join(args, " "), " push ") {
f.mu.Lock()
for k := range f.registry {
f.registry[k] = "sha256:2222222222222222222222222222222222222222222222222222222222222222"
}
f.mu.Unlock()
}
return r
}
_, st := runBuild(t, m, states, map[string]any{"publish": true})
if st.State != delivery.StepFailed || !strings.Contains(st.Reason, "the registry serves") {
t.Errorf("step = %+v", st)
}
}
func TestTheRegistrysAddressMustBeBound(t *testing.T) {
for _, bad := range []string{"", "${bound:artifact-store:at}:5000", ":5000"} {
m, _, _ := setup(t, same)
writeFile(t, m.RegistryFile, bad)
if _, err := m.registry(); err == nil {
t.Errorf("%q was taken as the registry's address", bad)
}
}
}
func writeFile(t *testing.T, path, body string) {
t.Helper()
if err := os.WriteFile(path, []byte(body), 0o644); err != nil {
t.Fatal(err)
}
}
// A push that succeeded wrote its digest file: one missing, unreadable or empty is a fault, and the registry's digest is
// never answered unchecked (review round 1).
func TestAPushWithoutAReadableDigestFileFailsTheStep(t *testing.T) {
for _, kind := range []string{"none", "empty"} {
m, f, states := setup(t, same)
f.digestFile = kind
_, st := runBuild(t, m, states, map[string]any{"publish": true})
if st.State != delivery.StepFailed || !strings.Contains(st.Reason, "digest") {
t.Errorf("%s: step = %+v, want failed naming the digest file", kind, st)
}
}
}
-1
View File
@@ -1 +0,0 @@
${seat:mesh-artifact-store:reach}
+9 -9
View File
@@ -52,9 +52,17 @@
"requires": [
"forge-account",
"rootless-image-build",
"artifact-store",
"node-tools"
],
"invokes": [
"seat:container-image-registry.publish",
"seat:container-image-registry.publish-part",
"seat:container-image-registry.mirror"
],
"consumes": [
"container-image-registry.published",
"container-image-registry.refused"
],
"own-secrets": {
"broker": "${dir:state}/broker"
},
@@ -142,14 +150,6 @@
"mode": "0644",
"template": "files/gitconfig"
},
{
"id": "registry",
"type": "file",
"path": "${dir:state}/registry",
"owner": "mesh-build-image",
"mode": "0644",
"template": "files/registry"
},
{
"id": "git",
"type": "package",