re-implement multipart. seems to be working

cmd/hold/main.go

@@ -191,6 +191,53 @@ type PutPresignedURLResponse struct {
 	ExpiresAt time.Time `json:"expires_at"`
 }
 
+// StartMultipartUploadRequest initiates a multipart upload
+type StartMultipartUploadRequest struct {
+	DID    string `json:"did"`
+	Digest string `json:"digest"`
+}
+
+// StartMultipartUploadResponse contains the multipart upload ID
+type StartMultipartUploadResponse struct {
+	UploadID  string    `json:"upload_id"`
+	ExpiresAt time.Time `json:"expires_at"`
+}
+
+// GetPartURLRequest requests a presigned URL for a specific part
+type GetPartURLRequest struct {
+	DID        string `json:"did"`
+	Digest     string `json:"digest"`
+	UploadID   string `json:"upload_id"`
+	PartNumber int    `json:"part_number"`
+}
+
+// GetPartURLResponse contains the presigned URL for a part
+type GetPartURLResponse struct {
+	URL       string    `json:"url"`
+	ExpiresAt time.Time `json:"expires_at"`
+}
+
+// CompleteMultipartRequest completes a multipart upload
+type CompleteMultipartRequest struct {
+	DID      string          `json:"did"`
+	Digest   string          `json:"digest"`
+	UploadID string          `json:"upload_id"`
+	Parts    []CompletedPart `json:"parts"`
+}
+
+// CompletedPart represents an uploaded part with its ETag
+type CompletedPart struct {
+	PartNumber int    `json:"part_number"`
+	ETag       string `json:"etag"`
+}
+
+// AbortMultipartRequest aborts an in-progress upload
+type AbortMultipartRequest struct {
+	DID      string `json:"did"`
+	Digest   string `json:"digest"`
+	UploadID string `json:"upload_id"`
+}
+
 // HandleGetPresignedURL handles requests for download URLs
 func (s *HoldService) HandleGetPresignedURL(w http.ResponseWriter, r *http.Request) {
 	if r.Method != http.MethodPost {
@@ -682,6 +729,282 @@ func (s *HoldService) getProxyUploadURL(digest, did string) string {
 	return fmt.Sprintf("%s/blobs/%s?did=%s", s.config.Server.PublicURL, digest, did)
 }
 
+// startMultipartUpload initiates a multipart upload and returns upload ID
+func (s *HoldService) startMultipartUpload(ctx context.Context, digest string) (string, error) {
+	if s.s3Client == nil {
+		return "", fmt.Errorf("S3 not configured")
+	}
+
+	path := blobPath(digest)
+	s3Key := strings.TrimPrefix(path, "/")
+	if s.s3PathPrefix != "" {
+		s3Key = s.s3PathPrefix + "/" + s3Key
+	}
+
+	result, err := s.s3Client.CreateMultipartUploadWithContext(ctx, &s3.CreateMultipartUploadInput{
+		Bucket: aws.String(s.bucket),
+		Key:    aws.String(s3Key),
+	})
+	if err != nil {
+		return "", err
+	}
+
+	log.Printf("Started multipart upload: digest=%s, uploadID=%s", digest, *result.UploadId)
+	return *result.UploadId, nil
+}
+
+// getPartPresignedURL generates presigned URL for a specific part
+func (s *HoldService) getPartPresignedURL(ctx context.Context, digest, uploadID string, partNumber int) (string, error) {
+	if s.s3Client == nil {
+		return "", fmt.Errorf("S3 not configured")
+	}
+
+	path := blobPath(digest)
+	s3Key := strings.TrimPrefix(path, "/")
+	if s.s3PathPrefix != "" {
+		s3Key = s.s3PathPrefix + "/" + s3Key
+	}
+
+	req, _ := s.s3Client.UploadPartRequest(&s3.UploadPartInput{
+		Bucket:     aws.String(s.bucket),
+		Key:        aws.String(s3Key),
+		UploadId:   aws.String(uploadID),
+		PartNumber: aws.Int64(int64(partNumber)),
+	})
+
+	url, err := req.Presign(15 * time.Minute)
+	if err != nil {
+		return "", err
+	}
+
+	log.Printf("Generated part presigned URL: digest=%s, uploadID=%s, part=%d", digest, uploadID, partNumber)
+	return url, nil
+}
+
+// completeMultipartUpload finalizes the multipart upload
+func (s *HoldService) completeMultipartUpload(ctx context.Context, digest, uploadID string, parts []CompletedPart) error {
+	if s.s3Client == nil {
+		return fmt.Errorf("S3 not configured")
+	}
+
+	path := blobPath(digest)
+	s3Key := strings.TrimPrefix(path, "/")
+	if s.s3PathPrefix != "" {
+		s3Key = s.s3PathPrefix + "/" + s3Key
+	}
+
+	// Convert to S3 CompletedPart format
+	s3Parts := make([]*s3.CompletedPart, len(parts))
+	for i, p := range parts {
+		s3Parts[i] = &s3.CompletedPart{
+			PartNumber: aws.Int64(int64(p.PartNumber)),
+			ETag:       aws.String(p.ETag),
+		}
+	}
+
+	_, err := s.s3Client.CompleteMultipartUploadWithContext(ctx, &s3.CompleteMultipartUploadInput{
+		Bucket:   aws.String(s.bucket),
+		Key:      aws.String(s3Key),
+		UploadId: aws.String(uploadID),
+		MultipartUpload: &s3.CompletedMultipartUpload{
+			Parts: s3Parts,
+		},
+	})
+
+	if err != nil {
+		log.Printf("Failed to complete multipart upload: digest=%s, uploadID=%s, err=%v", digest, uploadID, err)
+		return err
+	}
+
+	log.Printf("Completed multipart upload: digest=%s, uploadID=%s, parts=%d", digest, uploadID, len(parts))
+	return nil
+}
+
+// abortMultipartUpload aborts an in-progress multipart upload
+func (s *HoldService) abortMultipartUpload(ctx context.Context, digest, uploadID string) error {
+	if s.s3Client == nil {
+		return fmt.Errorf("S3 not configured")
+	}
+
+	path := blobPath(digest)
+	s3Key := strings.TrimPrefix(path, "/")
+	if s.s3PathPrefix != "" {
+		s3Key = s.s3PathPrefix + "/" + s3Key
+	}
+
+	_, err := s.s3Client.AbortMultipartUploadWithContext(ctx, &s3.AbortMultipartUploadInput{
+		Bucket:   aws.String(s.bucket),
+		Key:      aws.String(s3Key),
+		UploadId: aws.String(uploadID),
+	})
+
+	if err != nil {
+		log.Printf("Failed to abort multipart upload: digest=%s, uploadID=%s, err=%v", digest, uploadID, err)
+		return err
+	}
+
+	log.Printf("Aborted multipart upload: digest=%s, uploadID=%s", digest, uploadID)
+	return nil
+}
+
+// HandleStartMultipart initiates a multipart upload
+func (s *HoldService) HandleStartMultipart(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodPost {
+		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
+		return
+	}
+
+	var req StartMultipartUploadRequest
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		http.Error(w, fmt.Sprintf("invalid request: %v", err), http.StatusBadRequest)
+		return
+	}
+
+	// Validate DID authorization for WRITE
+	if !s.isAuthorizedWrite(req.DID) {
+		if req.DID == "" {
+			http.Error(w, "unauthorized: authentication required", http.StatusUnauthorized)
+		} else {
+			http.Error(w, "forbidden: write access denied", http.StatusForbidden)
+		}
+		return
+	}
+
+	// Start multipart upload
+	ctx := r.Context()
+	uploadID, err := s.startMultipartUpload(ctx, req.Digest)
+	if err != nil {
+		http.Error(w, fmt.Sprintf("failed to start multipart upload: %v", err), http.StatusInternalServerError)
+		return
+	}
+
+	expiry := time.Now().Add(24 * time.Hour) // Multipart uploads can take longer
+
+	resp := StartMultipartUploadResponse{
+		UploadID:  uploadID,
+		ExpiresAt: expiry,
+	}
+
+	w.Header().Set("Content-Type", "application/json")
+	json.NewEncoder(w).Encode(resp)
+}
+
+// HandleGetPartURL generates a presigned URL for uploading a specific part
+func (s *HoldService) HandleGetPartURL(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodPost {
+		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
+		return
+	}
+
+	var req GetPartURLRequest
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		http.Error(w, fmt.Sprintf("invalid request: %v", err), http.StatusBadRequest)
+		return
+	}
+
+	// Validate DID authorization for WRITE
+	if !s.isAuthorizedWrite(req.DID) {
+		if req.DID == "" {
+			http.Error(w, "unauthorized: authentication required", http.StatusUnauthorized)
+		} else {
+			http.Error(w, "forbidden: write access denied", http.StatusForbidden)
+		}
+		return
+	}
+
+	// Get presigned URL for this part
+	ctx := r.Context()
+	url, err := s.getPartPresignedURL(ctx, req.Digest, req.UploadID, req.PartNumber)
+	if err != nil {
+		http.Error(w, fmt.Sprintf("failed to generate part URL: %v", err), http.StatusInternalServerError)
+		return
+	}
+
+	expiry := time.Now().Add(15 * time.Minute)
+
+	resp := GetPartURLResponse{
+		URL:       url,
+		ExpiresAt: expiry,
+	}
+
+	w.Header().Set("Content-Type", "application/json")
+	json.NewEncoder(w).Encode(resp)
+}
+
+// HandleCompleteMultipart completes a multipart upload
+func (s *HoldService) HandleCompleteMultipart(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodPost {
+		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
+		return
+	}
+
+	var req CompleteMultipartRequest
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		http.Error(w, fmt.Sprintf("invalid request: %v", err), http.StatusBadRequest)
+		return
+	}
+
+	// Validate DID authorization for WRITE
+	if !s.isAuthorizedWrite(req.DID) {
+		if req.DID == "" {
+			http.Error(w, "unauthorized: authentication required", http.StatusUnauthorized)
+		} else {
+			http.Error(w, "forbidden: write access denied", http.StatusForbidden)
+		}
+		return
+	}
+
+	// Complete multipart upload
+	ctx := r.Context()
+	if err := s.completeMultipartUpload(ctx, req.Digest, req.UploadID, req.Parts); err != nil {
+		http.Error(w, fmt.Sprintf("failed to complete multipart upload: %v", err), http.StatusInternalServerError)
+		return
+	}
+
+	w.WriteHeader(http.StatusOK)
+	w.Header().Set("Content-Type", "application/json")
+	json.NewEncoder(w).Encode(map[string]string{
+		"status": "completed",
+	})
+}
+
+// HandleAbortMultipart aborts an in-progress multipart upload
+func (s *HoldService) HandleAbortMultipart(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodPost {
+		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
+		return
+	}
+
+	var req AbortMultipartRequest
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		http.Error(w, fmt.Sprintf("invalid request: %v", err), http.StatusBadRequest)
+		return
+	}
+
+	// Validate DID authorization for WRITE
+	if !s.isAuthorizedWrite(req.DID) {
+		if req.DID == "" {
+			http.Error(w, "unauthorized: authentication required", http.StatusUnauthorized)
+		} else {
+			http.Error(w, "forbidden: write access denied", http.StatusForbidden)
+		}
+		return
+	}
+
+	// Abort multipart upload
+	ctx := r.Context()
+	if err := s.abortMultipartUpload(ctx, req.Digest, req.UploadID); err != nil {
+		http.Error(w, fmt.Sprintf("failed to abort multipart upload: %v", err), http.StatusInternalServerError)
+		return
+	}
+
+	w.WriteHeader(http.StatusOK)
+	w.Header().Set("Content-Type", "application/json")
+	json.NewEncoder(w).Encode(map[string]string{
+		"status": "aborted",
+	})
+}
+
 // RegisterRequest represents a request to register this hold in a user's PDS
 type RegisterRequest struct {
 	DID         string `json:"did"`
@@ -800,6 +1123,12 @@ func main() {
 	mux.HandleFunc("/put-presigned-url", service.HandlePutPresignedURL)
 	mux.HandleFunc("/move", service.HandleMove)
 
+	// Multipart upload endpoints
+	mux.HandleFunc("/start-multipart", service.HandleStartMultipart)
+	mux.HandleFunc("/part-presigned-url", service.HandleGetPartURL)
+	mux.HandleFunc("/complete-multipart", service.HandleCompleteMultipart)
+	mux.HandleFunc("/abort-multipart", service.HandleAbortMultipart)
+
 	// Pre-register OAuth callback route (will be populated by auto-registration)
 	var oauthCallbackHandler http.HandlerFunc
 	mux.HandleFunc("/auth/oauth/callback", func(w http.ResponseWriter, r *http.Request) {

pkg/storage/proxy_blob_store.go

@@ -234,9 +234,9 @@ func (p *ProxyBlobStore) ServeBlob(ctx context.Context, w http.ResponseWriter, r
 	return nil
 }
 
-// Create returns a blob writer for uploading
+// Create returns a blob writer for uploading using multipart upload
 func (p *ProxyBlobStore) Create(ctx context.Context, options ...distribution.BlobCreateOption) (distribution.BlobWriter, error) {
-	fmt.Printf("🔧 [proxy_blob_store/Create] Starting streaming upload (NOT presigned URL)\n")
+	fmt.Printf("🔧 [proxy_blob_store/Create] Starting multipart upload\n")
 	fmt.Printf("   Storage endpoint: %s\n", p.storageEndpoint)
 	fmt.Printf("   Repository: %s\n", p.repository)
 
@@ -248,26 +248,28 @@ func (p *ProxyBlobStore) Create(ctx context.Context, options ...distribution.Blo
 		}
 	}
 
-	fmt.Printf("   Mount: %v\n", opts.Mount.ShouldMount)
-	if opts.Mount.ShouldMount {
-		fmt.Printf("   Mount from: %s\n", opts.Mount.From.Name())
-		fmt.Printf("   Mount digest: %s\n", opts.Mount.Stat.Digest)
+	// Generate unique writer ID
+	writerID := fmt.Sprintf("upload-%d", time.Now().UnixNano())
+
+	// Use temp digest for upload location (will be moved to final digest on commit)
+	tempDigest := fmt.Sprintf("uploads/temp-%s", writerID)
+
+	// Start multipart upload via hold service
+	uploadID, err := p.startMultipartUpload(ctx, tempDigest)
+	if err != nil {
+		return nil, fmt.Errorf("failed to start multipart upload: %w", err)
 	}
 
-	// Create pipe for streaming upload
-	pipeReader, pipeWriter := io.Pipe()
-	uploadErr := make(chan error, 1)
-	digestChan := make(chan string, 1)
+	fmt.Printf("   Started multipart upload: uploadID=%s\n", uploadID)
 
-	// Create writer
 	writer := &ProxyBlobWriter{
 		store:      p,
 		options:    opts,
-		pipeWriter: pipeWriter,
-		pipeReader: pipeReader,
-		digestChan: digestChan,
-		uploadErr:  uploadErr,
-		id:         fmt.Sprintf("upload-%d", time.Now().UnixNano()),
+		uploadID:   uploadID,
+		parts:      make([]CompletedPart, 0),
+		partNumber: 1,
+		buffer:     bytes.NewBuffer(make([]byte, 0, maxChunkSize)), // 5MB buffer
+		id:         writerID,
 		startedAt:  time.Now(),
 	}
 
@@ -276,68 +278,6 @@ func (p *ProxyBlobStore) Create(ctx context.Context, options ...distribution.Blo
 	globalUploads[writer.id] = writer
 	globalUploadsMu.Unlock()
 
-	// Start background goroutine that streams to temp location immediately
-	go func() {
-		defer pipeReader.Close()
-
-		// Stream to temp location immediately to avoid pipe deadlock
-		tempPath := fmt.Sprintf("uploads/temp-%s", writer.id) // No leading slash
-		url := fmt.Sprintf("%s/blobs/%s?did=%s", p.storageEndpoint, tempPath, p.did)
-
-		fmt.Printf("📦 [goroutine]: Starting streaming upload to temp location\n")
-		fmt.Printf("   Temp path: %s\n", tempPath)
-		fmt.Printf("   URL: %s\n", url)
-		fmt.Printf("   This is a PROXY upload (not presigned URL)\n")
-
-		// Use context with timeout to prevent hanging forever
-		uploadCtx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
-		defer cancel()
-
-		req, err := http.NewRequestWithContext(uploadCtx, "PUT", url, pipeReader)
-		if err != nil {
-			fmt.Printf("DEBUG [goroutine]: Failed to create request: %v\n", err)
-			// Consume digest channel even on error
-			<-digestChan
-			uploadErr <- fmt.Errorf("failed to create request: %w", err)
-			return
-		}
-		req.Header.Set("Content-Type", "application/octet-stream")
-
-		fmt.Printf("DEBUG [goroutine]: Sending PUT request...\n")
-		// Stream to temp location (this will block until all data is written)
-		resp, err := p.httpClient.Do(req)
-		if err != nil {
-			fmt.Printf("DEBUG [goroutine]: PUT failed: %v\n", err)
-			<-digestChan
-			uploadErr <- fmt.Errorf("failed to upload to temp: %w", err)
-			return
-		}
-		defer resp.Body.Close()
-
-		fmt.Printf("DEBUG [goroutine]: Got response status=%d\n", resp.StatusCode)
-
-		if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusCreated {
-			bodyBytes, _ := io.ReadAll(resp.Body)
-			fmt.Printf("DEBUG [goroutine]: Upload failed with status %d, body=%s\n", resp.StatusCode, string(bodyBytes))
-			<-digestChan
-			uploadErr <- fmt.Errorf("upload to temp failed: status %d, body: %s", resp.StatusCode, string(bodyBytes))
-			return
-		}
-
-		fmt.Printf("DEBUG [goroutine]: Upload to temp succeeded, waiting for digest...\n")
-		// Upload to temp succeeded, now wait for digest from Commit()
-		digest, ok := <-digestChan
-		if !ok {
-			uploadErr <- fmt.Errorf("upload cancelled after streaming to temp")
-			return
-		}
-
-		fmt.Printf("DEBUG [goroutine]: Got digest=%s, signaling completion\n", digest)
-		// Store digest for Commit() to use in move operation
-		writer.finalDigest = digest
-		uploadErr <- nil
-	}()
-
 	return writer, nil
 }
 
@@ -352,7 +292,7 @@ func (p *ProxyBlobStore) Resume(ctx context.Context, id string) (distribution.Bl
 		return nil, distribution.ErrBlobUploadUnknown
 	}
 
-	// With streaming, no flush needed - just return the writer
+	// Just return the writer - parts are buffered and flushed on demand
 	return writer, nil
 }
 
@@ -439,19 +379,176 @@ func (p *ProxyBlobStore) getUploadURL(ctx context.Context, dgst digest.Digest, s
 	return result.URL, nil
 }
 
-// ProxyBlobWriter implements distribution.BlobWriter for proxy uploads
+// startMultipartUpload initiates a multipart upload via hold service
+func (p *ProxyBlobStore) startMultipartUpload(ctx context.Context, digest string) (string, error) {
+	reqBody := map[string]any{
+		"did":    p.did,
+		"digest": digest,
+	}
+
+	body, err := json.Marshal(reqBody)
+	if err != nil {
+		return "", err
+	}
+
+	url := fmt.Sprintf("%s/start-multipart", p.storageEndpoint)
+	req, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewReader(body))
+	if err != nil {
+		return "", err
+	}
+	req.Header.Set("Content-Type", "application/json")
+
+	resp, err := p.httpClient.Do(req)
+	if err != nil {
+		return "", err
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		bodyBytes, _ := io.ReadAll(resp.Body)
+		return "", fmt.Errorf("start multipart failed: status %d, body: %s", resp.StatusCode, string(bodyBytes))
+	}
+
+	var result struct {
+		UploadID string `json:"upload_id"`
+	}
+	if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
+		return "", err
+	}
+
+	return result.UploadID, nil
+}
+
+// getPartPresignedURL gets a presigned URL for uploading a specific part
+func (p *ProxyBlobStore) getPartPresignedURL(ctx context.Context, digest, uploadID string, partNumber int) (string, error) {
+	reqBody := map[string]any{
+		"did":         p.did,
+		"digest":      digest,
+		"upload_id":   uploadID,
+		"part_number": partNumber,
+	}
+
+	body, err := json.Marshal(reqBody)
+	if err != nil {
+		return "", err
+	}
+
+	url := fmt.Sprintf("%s/part-presigned-url", p.storageEndpoint)
+	req, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewReader(body))
+	if err != nil {
+		return "", err
+	}
+	req.Header.Set("Content-Type", "application/json")
+
+	resp, err := p.httpClient.Do(req)
+	if err != nil {
+		return "", err
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		bodyBytes, _ := io.ReadAll(resp.Body)
+		return "", fmt.Errorf("get part URL failed: status %d, body: %s", resp.StatusCode, string(bodyBytes))
+	}
+
+	var result struct {
+		URL string `json:"url"`
+	}
+	if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
+		return "", err
+	}
+
+	return result.URL, nil
+}
+
+// completeMultipartUpload completes a multipart upload via hold service
+func (p *ProxyBlobStore) completeMultipartUpload(ctx context.Context, digest, uploadID string, parts []CompletedPart) error {
+	reqBody := map[string]any{
+		"did":       p.did,
+		"digest":    digest,
+		"upload_id": uploadID,
+		"parts":     parts,
+	}
+
+	body, err := json.Marshal(reqBody)
+	if err != nil {
+		return err
+	}
+
+	url := fmt.Sprintf("%s/complete-multipart", p.storageEndpoint)
+	req, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewReader(body))
+	if err != nil {
+		return err
+	}
+	req.Header.Set("Content-Type", "application/json")
+
+	resp, err := p.httpClient.Do(req)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		bodyBytes, _ := io.ReadAll(resp.Body)
+		return fmt.Errorf("complete multipart failed: status %d, body: %s", resp.StatusCode, string(bodyBytes))
+	}
+
+	return nil
+}
+
+// abortMultipartUpload aborts a multipart upload via hold service
+func (p *ProxyBlobStore) abortMultipartUpload(ctx context.Context, digest, uploadID string) error {
+	reqBody := map[string]any{
+		"did":       p.did,
+		"digest":    digest,
+		"upload_id": uploadID,
+	}
+
+	body, err := json.Marshal(reqBody)
+	if err != nil {
+		return err
+	}
+
+	url := fmt.Sprintf("%s/abort-multipart", p.storageEndpoint)
+	req, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewReader(body))
+	if err != nil {
+		return err
+	}
+	req.Header.Set("Content-Type", "application/json")
+
+	resp, err := p.httpClient.Do(req)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		bodyBytes, _ := io.ReadAll(resp.Body)
+		return fmt.Errorf("abort multipart failed: status %d, body: %s", resp.StatusCode, string(bodyBytes))
+	}
+
+	return nil
+}
+
+// CompletedPart represents an uploaded part with its ETag
+type CompletedPart struct {
+	PartNumber int    `json:"part_number"`
+	ETag       string `json:"etag"`
+}
+
+// ProxyBlobWriter implements distribution.BlobWriter for proxy uploads using multipart upload
 type ProxyBlobWriter struct {
 	store       *ProxyBlobStore
 	options     distribution.CreateOptions
-	pipeWriter  *io.PipeWriter // Streams directly to hold service
-	pipeReader  *io.PipeReader
-	digestChan  chan string // Sends digest to upload goroutine
-	uploadErr   chan error  // Receives upload result from goroutine
-	finalDigest string      // Final digest for move operation
-	size        int64
+	uploadID    string          // S3 multipart upload ID
+	parts       []CompletedPart // Track uploaded parts with ETags
+	partNumber  int             // Current part number (starts at 1)
+	buffer      *bytes.Buffer   // Buffer for current part
+	size        int64           // Total bytes written
 	closed      bool
-	id          string // Distribution's upload ID
+	id          string    // Distribution's upload ID (for state)
 	startedAt   time.Time
+	finalDigest string // Set on Commit
 }
 
 // ID returns the upload ID
@@ -465,22 +562,79 @@ func (w *ProxyBlobWriter) StartedAt() time.Time {
 }
 
 // Write writes data to the upload
-// Streams directly to hold service via pipe
+// Buffers data and flushes when buffer reaches 5MB
 func (w *ProxyBlobWriter) Write(p []byte) (int, error) {
 	if w.closed {
 		return 0, fmt.Errorf("writer closed")
 	}
 
-	// Write to pipe - streams immediately to hold service
-	n, err := w.pipeWriter.Write(p)
-	if err != nil {
-		// If write fails (client disconnected), close pipe to unblock goroutine
-		w.pipeWriter.CloseWithError(err)
-		return n, err
-	}
+	n, err := w.buffer.Write(p)
 	w.size += int64(n)
 
-	return n, nil
+	// Flush if buffer reaches 5MB (S3 minimum part size)
+	if w.buffer.Len() >= maxChunkSize {
+		if err := w.flushPart(); err != nil {
+			return n, err
+		}
+	}
+
+	return n, err
+}
+
+// flushPart uploads the current buffer as a part
+func (w *ProxyBlobWriter) flushPart() error {
+	if w.buffer.Len() == 0 {
+		return nil
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
+	defer cancel()
+
+	// Get presigned URL for this part
+	tempDigest := fmt.Sprintf("uploads/temp-%s", w.id)
+	url, err := w.store.getPartPresignedURL(ctx, tempDigest, w.uploadID, w.partNumber)
+	if err != nil {
+		return fmt.Errorf("failed to get part presigned URL: %w", err)
+	}
+
+	fmt.Printf("📤 [flushPart] Uploading part %d: size=%d bytes\n", w.partNumber, w.buffer.Len())
+
+	// Upload part to S3
+	req, err := http.NewRequestWithContext(ctx, "PUT", url, bytes.NewReader(w.buffer.Bytes()))
+	if err != nil {
+		return err
+	}
+	req.Header.Set("Content-Type", "application/octet-stream")
+
+	resp, err := w.store.httpClient.Do(req)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusCreated {
+		bodyBytes, _ := io.ReadAll(resp.Body)
+		return fmt.Errorf("part upload failed: status %d, body: %s", resp.StatusCode, string(bodyBytes))
+	}
+
+	// Store ETag for completion
+	etag := resp.Header.Get("ETag")
+	if etag == "" {
+		return fmt.Errorf("no ETag in response")
+	}
+
+	w.parts = append(w.parts, CompletedPart{
+		PartNumber: w.partNumber,
+		ETag:       etag,
+	})
+
+	fmt.Printf("✅ [flushPart] Part %d uploaded successfully: ETag=%s\n", w.partNumber, etag)
+
+	// Reset buffer and increment part number
+	w.buffer.Reset()
+	w.partNumber++
+
+	return nil
 }
 
 // ReadFrom reads from a reader
@@ -518,40 +672,47 @@ func (w *ProxyBlobWriter) Size() int64 {
 	return w.size
 }
 
-// Commit finalizes the upload
+// Commit finalizes the upload by completing multipart upload and moving to final location
 func (w *ProxyBlobWriter) Commit(ctx context.Context, desc distribution.Descriptor) (distribution.Descriptor, error) {
 	if w.closed {
 		return distribution.Descriptor{}, fmt.Errorf("writer closed")
 	}
 	w.closed = true
 
+	fmt.Printf("📝 [Commit] Starting commit: digest=%s, size=%d\n", desc.Digest, w.size)
+
 	// Remove from global uploads map
 	globalUploadsMu.Lock()
 	delete(globalUploads, w.id)
 	globalUploadsMu.Unlock()
 
-	// Close pipe to signal EOF to upload goroutine
-	if err := w.pipeWriter.Close(); err != nil {
-		return distribution.Descriptor{}, fmt.Errorf("failed to close pipe: %w", err)
+	// Flush any remaining buffered data
+	if w.buffer.Len() > 0 {
+		fmt.Printf("📤 [Commit] Flushing final buffer: %d bytes\n", w.buffer.Len())
+		if err := w.flushPart(); err != nil {
+			// Try to abort multipart on error
+			tempDigest := fmt.Sprintf("uploads/temp-%s", w.id)
+			w.store.abortMultipartUpload(ctx, tempDigest, w.uploadID)
+			return distribution.Descriptor{}, fmt.Errorf("failed to flush final part: %w", err)
+		}
 	}
 
-	// Send digest to upload goroutine (it's waiting after temp upload completes)
-	w.digestChan <- desc.Digest.String()
-	close(w.digestChan)
-
-	// Wait for upload goroutine to complete
-	if err := <-w.uploadErr; err != nil {
-		return distribution.Descriptor{}, fmt.Errorf("upload to temp failed: %w", err)
+	// Complete multipart upload at temp location
+	tempDigest := fmt.Sprintf("uploads/temp-%s", w.id)
+	fmt.Printf("🔒 [Commit] Completing multipart upload: uploadID=%s, parts=%d\n", w.uploadID, len(w.parts))
+	if err := w.store.completeMultipartUpload(ctx, tempDigest, w.uploadID, w.parts); err != nil {
+		return distribution.Descriptor{}, fmt.Errorf("failed to complete multipart upload: %w", err)
 	}
 
-	// Now move temp → final location
-	tempPath := fmt.Sprintf("uploads/temp-%s", w.id) // No leading slash
+	// Move from temp → final location (server-side S3 copy)
+	tempPath := fmt.Sprintf("uploads/temp-%s", w.id)
 	finalPath := desc.Digest.String()
 
+	fmt.Printf("🚚 [Commit] Moving blob: %s → %s\n", tempPath, finalPath)
 	moveURL := fmt.Sprintf("%s/move?from=%s&to=%s&did=%s",
 		w.store.storageEndpoint, tempPath, finalPath, w.store.did)
 
-	req, err := http.NewRequestWithContext(context.Background(), "POST", moveURL, nil)
+	req, err := http.NewRequestWithContext(ctx, "POST", moveURL, nil)
 	if err != nil {
 		return distribution.Descriptor{}, fmt.Errorf("failed to create move request: %w", err)
 	}
@@ -567,7 +728,7 @@ func (w *ProxyBlobWriter) Commit(ctx context.Context, desc distribution.Descript
 		return distribution.Descriptor{}, fmt.Errorf("move blob failed: status %d, body: %s", resp.StatusCode, string(bodyBytes))
 	}
 
-	fmt.Printf("DEBUG [proxy_blob_store]: Committed upload: digest=%s, size=%d (moved from temp)\n", desc.Digest, w.size)
+	fmt.Printf("✅ [Commit] Upload completed successfully: digest=%s, size=%d, parts=%d\n", desc.Digest, w.size, len(w.parts))
 
 	return distribution.Descriptor{
 		Digest:    desc.Digest,
@@ -576,34 +737,31 @@ func (w *ProxyBlobWriter) Commit(ctx context.Context, desc distribution.Descript
 	}, nil
 }
 
-// Cancel cancels the upload
+// Cancel cancels the upload by aborting the multipart upload
 func (w *ProxyBlobWriter) Cancel(ctx context.Context) error {
 	w.closed = true
 
+	fmt.Printf("❌ [Cancel] Cancelling upload: id=%s\n", w.id)
+
 	// Remove from global uploads map
 	globalUploadsMu.Lock()
 	delete(globalUploads, w.id)
 	globalUploadsMu.Unlock()
 
-	// Close digest channel without sending digest
-	close(w.digestChan)
-
-	// Close pipe with error to stop streaming
-	if w.pipeWriter != nil {
-		w.pipeWriter.CloseWithError(fmt.Errorf("upload cancelled"))
+	// Abort multipart upload
+	tempDigest := fmt.Sprintf("uploads/temp-%s", w.id)
+	if err := w.store.abortMultipartUpload(ctx, tempDigest, w.uploadID); err != nil {
+		fmt.Printf("⚠️  [Cancel] Failed to abort multipart upload: %v\n", err)
+		// Continue anyway - we want to mark upload as cancelled
 	}
 
-	// Wait for goroutine to finish
-	<-w.uploadErr
-
-	fmt.Printf("DEBUG [proxy_blob_store]: Cancelled upload: id=%s\n", w.id)
+	fmt.Printf("✅ [Cancel] Upload cancelled: id=%s\n", w.id)
 	return nil
 }
 
 // Close closes the writer
-// Just returns - streaming continues via pipe
+// Parts are flushed on demand, so this is a no-op
 func (w *ProxyBlobWriter) Close() error {
-	// Don't close pipe here - that happens in Commit() or Cancel()
 	// Don't set w.closed = true - allow resuming for next PATCH
 	return nil
 }