mesh-sdk/go/buildstep/blob.go

package buildstep

import (
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"strings"
	"syscall"
	"time"
)

// RegistryWait is how long a push waits for an artifact store that refuses connections, longer than its nightly
// collection holds it still (about a minute and a half, issue 457); RegistryFirstPause is the first pause, doubling
// to RegistryMostPause. Variables so a test shortens them.
var (
	RegistryWait       = 5 * time.Minute
	RegistryFirstPause = time.Second
	RegistryMostPause  = 10 * time.Second
)

// BlobReference is where a pushed bundle or archive is, as a done key names it and the controller delivers from
// (novox/hq ADR 0498 clause 1): artifact-store://<module>/<artifact>/blobs/<digest>, the store's address left out
// (issue 102).
func BlobReference(module, artifact, digest string) string {
	return "artifact-store://" + module + "/" + artifact + "/blobs/" + digest
}

// PushBlob puts body into the artifact store at registry (host:port) as a blob of repository (<module>/<artifact>) by
// its digest (`sha256:` and hex), from the build module's own process, after the build step ended (novox/hq ADR 0498
// clauses 1 and 2): a blob the store already holds is done without an upload; otherwise it is uploaded (POST
// blobs/uploads/, then PUT with the digest). A store refusing connections is waited out up to RegistryWait; any other
// refusal fails, naming the store's status and its own words, for the caller's `not stored:` error. Bytes that are not
// their digest are refused before anything is sent.
func PushBlob(ctx context.Context, client *http.Client, registry, repository, digest string, body []byte) error {
	sum := sha256.Sum256(body)
	if got := "sha256:" + hex.EncodeToString(sum[:]); got != digest {
		return fmt.Errorf("the blob is %s, not the digest %s it was to be pushed as", got, digest)
	}
	if client == nil {
		client = http.DefaultClient
	}
	base := "http://" + registry + "/v2/" + repository + "/blobs/"
	var present bool
	if err := waitForStore(ctx, registry, func() error {
		var err error
		present, err = blobPresent(ctx, client, base+digest)
		return err
	}); err != nil {
		return err
	}
	if present {
		return nil
	}
	var location string
	if err := waitForStore(ctx, registry, func() error {
		var err error
		location, err = startUpload(ctx, client, base+"uploads/")
		return err
	}); err != nil {
		return err
	}
	target, err := url.Parse(base)
	if err != nil {
		return err
	}
	loc, err := target.Parse(location)
	if err != nil {
		return fmt.Errorf("the store answered the upload's location %q, which is not a URL: %w", location, err)
	}
	q := loc.Query()
	q.Set("digest", digest)
	loc.RawQuery = q.Encode()
	req, err := http.NewRequestWithContext(ctx, http.MethodPut, loc.String(), bytes.NewReader(body))
	if err != nil {
		return err
	}
	req.Header.Set("Content-Type", "application/octet-stream")
	resp, err := client.Do(req)
	if err != nil {
		return fmt.Errorf("putting %s of %s: %w", digest, repository, err)
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusCreated {
		return storeSaid(resp, "putting "+digest+" of "+repository)
	}
	return nil
}

func blobPresent(ctx context.Context, client *http.Client, at string) (bool, error) {
	req, err := http.NewRequestWithContext(ctx, http.MethodHead, at, nil)
	if err != nil {
		return false, err
	}
	resp, err := client.Do(req)
	if err != nil {
		return false, err
	}
	resp.Body.Close()
	switch resp.StatusCode {
	case http.StatusOK:
		return true, nil
	case http.StatusNotFound:
		return false, nil
	}
	return false, fmt.Errorf("asking the store for %s: it answered %s", at, resp.Status)
}

func startUpload(ctx context.Context, client *http.Client, at string) (string, error) {
	req, err := http.NewRequestWithContext(ctx, http.MethodPost, at, nil)
	if err != nil {
		return "", err
	}
	resp, err := client.Do(req)
	if err != nil {
		return "", err
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusAccepted {
		return "", storeSaid(resp, "starting an upload at "+at)
	}
	location := resp.Header.Get("Location")
	if location == "" {
		return "", fmt.Errorf("starting an upload at %s: the store named no location", at)
	}
	return location, nil
}

// storeSaid is the store's refusal: its status and the first of its own words.
func storeSaid(resp *http.Response, what string) error {
	words, _ := io.ReadAll(io.LimitReader(resp.Body, 2048))
	return fmt.Errorf("%s: the store answered %s: %s", what, resp.Status, strings.TrimSpace(string(words)))
}

// refused is whether an error is a connection refused.
func refused(err error) bool {
	return err != nil && (errors.Is(err, syscall.ECONNREFUSED) || strings.Contains(err.Error(), "connection refused"))
}

// waitForStore runs try, and while it fails because the store refuses connections, tries again with a growing pause
// until RegistryWait has passed. A refusal is the one error waited for: nothing was sent, so trying again does nothing
// twice.
func waitForStore(ctx context.Context, registry string, try func() error) error {
	err := try()
	if !refused(err) {
		return err
	}
	started, pause := time.Now(), RegistryFirstPause
	for {
		left := RegistryWait - time.Since(started)
		if left <= 0 {
			return fmt.Errorf("the artifact store at %s refused every connection for %s, longer than its nightly collection holds it still, so it is down, not paused: %w",
				registry, time.Since(started).Round(time.Millisecond), err)
		}
		select {
		case <-ctx.Done():
			return fmt.Errorf("stopped while waiting for the artifact store at %s: %w (last: %v)", registry, ctx.Err(), err)
		case <-time.After(min(pause, left)):
		}
		pause = min(pause*2, RegistryMostPause)
		if err = try(); !refused(err) {
			return err
		}
	}
}