Merge f8fc960e8f into 62c7e845d1

pkg/registry/client.go

@@ -30,8 +30,10 @@ import (
 	"os"
 	"sort"
 	"strings"
+	"sync"
 
 	"github.com/Masterminds/semver/v3"
+	"github.com/opencontainers/go-digest"
 	"github.com/opencontainers/image-spec/specs-go"
 	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
 	"oras.land/oras-go/v2"
@@ -670,33 +672,53 @@ func (c *Client) Push(data []byte, ref string, options ...PushOption) (*PushResu
 		}
 	}
 
-	ctx := context.Background()
+	repository, err := remote.NewRepository(parsedRef.String())
-
-	memoryStore := memory.New()
-	chartDescriptor, err := oras.PushBytes(ctx, memoryStore, ChartLayerMediaType, data)
 	if err != nil {
 		return nil, err
 	}
+	repository.PlainHTTP = c.plainHTTP
+	repository.Client = c.authorizer
 
-	configData, err := json.Marshal(meta)
+	ctx := context.Background()
+	ctx = auth.AppendRepositoryScope(ctx, repository.Reference, auth.ActionPull, auth.ActionPush)
+
+	chartBlob := newBlob(repository, ChartLayerMediaType, data)
+	exists, err := chartBlob.exists(ctx)
 	if err != nil {
 		return nil, err
 	}
 
-	configDescriptor, err := oras.PushBytes(ctx, memoryStore, ConfigMediaType, configData)
+	layers := []ocispec.Descriptor{chartBlob.descriptor}
+	var wg sync.WaitGroup
+	if !exists {
+		wg.Go(func() { chartBlob.push(ctx) })
+	}
+
+	configData, err := json.Marshal(meta)
 	if err != nil {
 		return nil, err
 	}
+	configBlob := newBlob(repository, ConfigMediaType, configData)
+	wg.Go(func() { configBlob.pushNew(ctx) })
 
-	layers := []ocispec.Descriptor{chartDescriptor}
+	var provBlob blob
-	var provDescriptor ocispec.Descriptor
 	if operation.provData != nil {
-		provDescriptor, err = oras.PushBytes(ctx, memoryStore, ProvLayerMediaType, operation.provData)
+		provBlob = newBlob(repository, ProvLayerMediaType, operation.provData)
-		if err != nil {
+		wg.Go(func() { provBlob.pushNew(ctx) })
-			return nil, err
+	}
-		}
+	wg.Wait()
 
-		layers = append(layers, provDescriptor)
+	if chartBlob.err != nil {
+		return nil, chartBlob.err
+	}
+	if configBlob.err != nil {
+		return nil, configBlob.err
+	}
+	if provBlob.err != nil {
+		return nil, provBlob.err
+	}
+	if operation.provData != nil {
+		layers = append(layers, provBlob.descriptor)
 	}
 
 	// sort layers for determinism, similar to how ORAS v1 does it
@@ -706,22 +728,20 @@ func (c *Client) Push(data []byte, ref string, options ...PushOption) (*PushResu
 
 	ociAnnotations := generateOCIAnnotations(meta, operation.creationTime)
 
-	manifestDescriptor, err := c.tagManifest(ctx, memoryStore, configDescriptor,
+	manifest := ocispec.Manifest{
-		layers, ociAnnotations, parsedRef)
+		Versioned:   specs.Versioned{SchemaVersion: 2},
-	if err != nil {
+		Config:      configBlob.descriptor,
-		return nil, err
+		Layers:      layers,
+		Annotations: ociAnnotations,
 	}
 
-	repository, err := remote.NewRepository(parsedRef.String())
+	manifestData, err := json.Marshal(manifest)
 	if err != nil {
 		return nil, err
 	}
-	repository.PlainHTTP = c.plainHTTP
-	repository.Client = c.authorizer
 
-	ctx = withScopeHint(ctx, repository, auth.ActionPull, auth.ActionPush)
+	manifestDescriptor, err := oras.TagBytes(ctx, repository, ocispec.MediaTypeImageManifest,
-
+		manifestData, parsedRef.String())
-	manifestDescriptor, err = oras.ExtendedCopy(ctx, memoryStore, parsedRef.String(), repository, parsedRef.String(), oras.DefaultExtendedCopyOptions)
 	if err != nil {
 		return nil, err
 	}
@@ -729,16 +749,16 @@ func (c *Client) Push(data []byte, ref string, options ...PushOption) (*PushResu
 	chartSummary := &descriptorPushSummaryWithMeta{
 		Meta: meta,
 	}
-	chartSummary.Digest = chartDescriptor.Digest.String()
+	chartSummary.Digest = chartBlob.descriptor.Digest.String()
-	chartSummary.Size = chartDescriptor.Size
+	chartSummary.Size = chartBlob.descriptor.Size
 	result := &PushResult{
 		Manifest: &descriptorPushSummary{
 			Digest: manifestDescriptor.Digest.String(),
 			Size:   manifestDescriptor.Size,
 		},
 		Config: &descriptorPushSummary{
-			Digest: configDescriptor.Digest.String(),
+			Digest: configBlob.descriptor.Digest.String(),
-			Size:   configDescriptor.Size,
+			Size:   configBlob.descriptor.Size,
 		},
 		Chart: chartSummary,
 		Prov:  &descriptorPushSummary{}, // prevent nil references
@@ -746,8 +766,8 @@ func (c *Client) Push(data []byte, ref string, options ...PushOption) (*PushResu
 	}
 	if operation.provData != nil {
 		result.Prov = &descriptorPushSummary{
-			Digest: provDescriptor.Digest.String(),
+			Digest: provBlob.descriptor.Digest.String(),
-			Size:   provDescriptor.Size,
+			Size:   provBlob.descriptor.Size,
 		}
 	}
 	_, _ = fmt.Fprintf(c.out, "Pushed: %s\n", result.Ref)
@@ -931,13 +951,53 @@ func (c *Client) tagManifest(ctx context.Context, memoryStore *memory.Store,
 		manifestData, parsedRef.String())
 }
 
-// add actions when request a registry authentication token(jwt)
+// blob represents a content-addressable blob to be pushed to an OCI registry.
-// example1. when we want to pull 'testrepo/local-subchart' we can send below url, and 'pull' is the action
+// It encapsulates the data, media type, and destination repository, and tracks
-// auth?scope=repository%3Atestrepo%2Flocal-subchart%3Apull&service=testservice
+// the resulting descriptor and any error from push operations.
-// example2. when we want to push 'testrepo/local-subchart' we can send below url, and 'pull%2Cpush' are the actions
+type blob struct {
-// auth?scope=repository%3Atestrepo%2Flocal-subchart%3Apull%2Cpush&service=testservice
+	dst        *remote.Repository
-// we can set the actions like below
+	mediaType  string
-// example) ctx = withScopeHint(ctx, repository, auth.ActionPush, auth.ActionPull)
+	data       []byte
-func withScopeHint(ctx context.Context, repo *remote.Repository, actions ...string) context.Context {
+	descriptor ocispec.Descriptor
-	return auth.AppendRepositoryScope(ctx, repo.Reference, actions...)
+	err        error
+}
+
+// newBlob creates a new blob with the given repository, media type, and data.
+func newBlob(dst *remote.Repository, mediaType string, data []byte) blob {
+	return blob{
+		dst:       dst,
+		mediaType: mediaType,
+		data:      data,
+	}
+}
+
+// exists checks if the blob already exists in the registry by computing its
+// digest and querying the repository. It also populates the blob's descriptor
+// with size, media type, and digest information.
+func (b *blob) exists(ctx context.Context) (bool, error) {
+	b.descriptor.Size = int64(len(b.data))
+	b.descriptor.MediaType = b.mediaType
+	b.descriptor.Digest = digest.FromBytes(b.data)
+	return b.dst.Exists(ctx, b.descriptor)
+}
+
+// pushNew checks if the blob exists in the registry first, and only pushes
+// if it doesn't exist. This avoids redundant uploads for blobs that are
+// already present. Any error is stored in b.err.
+func (b *blob) pushNew(ctx context.Context) {
+	var exists bool
+	exists, b.err = b.exists(ctx)
+	if b.err != nil {
+		return
+	}
+	if exists {
+		return
+	}
+	b.descriptor, b.err = oras.PushBytes(ctx, b.dst, b.mediaType, b.data)
+}
+
+// push unconditionally pushes the blob to the registry without checking
+// for existence first. Any error is stored in b.err.
+func (b *blob) push(ctx context.Context) {
+	b.descriptor, b.err = oras.PushBytes(ctx, b.dst, b.mediaType, b.data)
 }

pkg/registry/client_test.go

@@ -17,11 +17,15 @@ limitations under the License.
 package registry
 
 import (
+	"crypto/sha256"
+	"fmt"
 	"io"
 	"net/http"
 	"net/http/httptest"
+	"os"
 	"path/filepath"
 	"strings"
+	"sync"
 	"testing"
 
 	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
@@ -153,3 +157,114 @@ func TestWarnIfHostHasPath(t *testing.T) {
 		})
 	}
 }
+
+// TestPushConcurrent verifies that concurrent Push operations on the same Client
+// do not interfere with each other. This test is designed to catch race conditions
+// when run with -race flag.
+func TestPushConcurrent(t *testing.T) {
+	t.Parallel()
+
+	// Create a mock registry server that accepts pushes
+	var mu sync.Mutex
+	uploads := make(map[string][]byte)
+	var uploadCounter int
+
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		switch {
+		case r.Method == http.MethodHead && strings.Contains(r.URL.Path, "/blobs/"):
+			// Blob existence check - return 404 to force upload
+			w.WriteHeader(http.StatusNotFound)
+
+		case r.Method == http.MethodPost && strings.Contains(r.URL.Path, "/blobs/uploads/"):
+			// Start upload - return upload URL with unique ID
+			mu.Lock()
+			uploadCounter++
+			uploadID := fmt.Sprintf("upload-%d", uploadCounter)
+			mu.Unlock()
+			w.Header().Set("Location", fmt.Sprintf("%s%s", r.URL.Path, uploadID))
+			w.WriteHeader(http.StatusAccepted)
+
+		case r.Method == http.MethodPut && strings.Contains(r.URL.Path, "/blobs/uploads/"):
+			// Complete upload - extract digest from query param
+			body, err := io.ReadAll(r.Body)
+			if err != nil {
+				w.WriteHeader(http.StatusInternalServerError)
+				return
+			}
+			digest := r.URL.Query().Get("digest")
+			mu.Lock()
+			uploads[r.URL.Path] = body
+			mu.Unlock()
+			w.Header().Set("Docker-Content-Digest", digest)
+			w.WriteHeader(http.StatusCreated)
+
+		case r.Method == http.MethodPut && strings.Contains(r.URL.Path, "/manifests/"):
+			// The optimization pushes the manifest exactly once, by tag. A PUT to a
+			// digest reference (/manifests/sha256:...) means the manifest is being
+			// uploaded a second time by digest, which this test guards against.
+			if strings.Contains(r.URL.Path, "/manifests/sha256:") {
+				t.Errorf("unexpected manifest push by digest %q; manifest should be pushed once by tag", r.URL.Path)
+				w.WriteHeader(http.StatusBadRequest)
+				return
+			}
+			// Manifest push - compute actual sha256 digest of the body
+			body, err := io.ReadAll(r.Body)
+			if err != nil {
+				w.WriteHeader(http.StatusInternalServerError)
+				return
+			}
+			hash := sha256.Sum256(body)
+			digest := fmt.Sprintf("sha256:%x", hash)
+			w.Header().Set("Docker-Content-Digest", digest)
+			w.WriteHeader(http.StatusCreated)
+
+		default:
+			w.WriteHeader(http.StatusNotFound)
+		}
+	}))
+	defer srv.Close()
+
+	host := strings.TrimPrefix(srv.URL, "http://")
+
+	// Create client
+	credFile := filepath.Join(t.TempDir(), "config.json")
+	client, err := NewClient(
+		ClientOptWriter(io.Discard),
+		ClientOptCredentialsFile(credFile),
+		ClientOptPlainHTTP(),
+	)
+	require.NoError(t, err)
+
+	// Load test chart
+	chartData, err := os.ReadFile("../downloader/testdata/local-subchart-0.1.0.tgz")
+	require.NoError(t, err, "no error loading test chart")
+
+	meta, err := extractChartMeta(chartData)
+	require.NoError(t, err, "no error extracting chart meta")
+
+	// Run concurrent pushes
+	const numGoroutines = 10
+	var wg sync.WaitGroup
+	errs := make(chan error, numGoroutines)
+
+	for i := range numGoroutines {
+		wg.Add(1)
+		go func(idx int) {
+			defer wg.Done()
+			// Each goroutine pushes to a different tag to avoid conflicts
+			ref := fmt.Sprintf("%s/testrepo/%s:%s-%d", host, meta.Name, meta.Version, idx)
+			_, err := client.Push(chartData, ref, PushOptStrictMode(false))
+			if err != nil {
+				errs <- fmt.Errorf("goroutine %d: %w", idx, err)
+			}
+		}(i)
+	}
+
+	wg.Wait()
+	close(errs)
+
+	// Check for errors
+	for err := range errs {
+		t.Error(err)
+	}
+}