diff --git a/cmd/hold/main.go b/cmd/hold/main.go index a5212af..da28d4e 100644 --- a/cmd/hold/main.go +++ b/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) { diff --git a/pkg/storage/proxy_blob_store.go b/pkg/storage/proxy_blob_store.go index 6e8b6e6..5f18845 100644 --- a/pkg/storage/proxy_blob_store.go +++ b/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 }