diff --git a/cmd/appview/serve.go b/cmd/appview/serve.go index 461b2bc..bc0e0d5 100644 --- a/cmd/appview/serve.go +++ b/cmd/appview/serve.go @@ -411,8 +411,7 @@ func initializeDatabase() (*sql.DB, *sql.DB, *db.SessionStore) { return nil, nil, nil } - fmt.Printf("UI database initialized at %s\n", dbPath) - fmt.Printf("Read-only connection with authorizer created (blocks: oauth_sessions, ui_sessions, devices, etc.)\n") + fmt.Printf("UI database (readonly) initialized at %s\n", dbPath) // Create SQLite-backed session store sessionStore := db.NewSessionStore(database) diff --git a/cmd/hold/main.go b/cmd/hold/main.go index 9099fbe..d7e923d 100644 --- a/cmd/hold/main.go +++ b/cmd/hold/main.go @@ -194,6 +194,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 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 the part +type GetPartURLResponse struct { + URL string `json:"url"` + ExpiresAt time.Time `json:"expires_at"` +} + +// CompletedPart represents a completed multipart upload part +type CompletedPart struct { + PartNumber int `json:"part_number"` + ETag string `json:"etag"` +} + +// 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"` +} + +// 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 { @@ -642,6 +689,266 @@ 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 for multipart uploads") + } + + 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 "", fmt.Errorf("failed to create multipart upload: %w", err) + } + + log.Printf("Started multipart upload: key=%s, uploadId=%s", s3Key, *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 for multipart uploads") + } + + 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 "", fmt.Errorf("failed to presign part URL: %w", err) + } + + log.Printf("Generated presigned URL for part %d: key=%s, uploadId=%s", partNumber, s3Key, uploadID) + 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 for multipart uploads") + } + + 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 { + return fmt.Errorf("failed to complete multipart upload: %w", err) + } + + log.Printf("Completed multipart upload: key=%s, uploadId=%s, parts=%d", s3Key, uploadID, len(parts)) + return nil +} + +// abortMultipartUpload cancels 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 for multipart uploads") + } + + 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 { + return fmt.Errorf("failed to abort multipart upload: %w", err) + } + + log.Printf("Aborted multipart upload: key=%s, uploadId=%s", s3Key, 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 + } + + 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 + } + + resp := StartMultipartUploadResponse{ + UploadID: uploadID, + ExpiresAt: time.Now().Add(24 * time.Hour), // Multipart uploads expire in 24h + } + + 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 + } + + 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 + } + + resp := GetPartURLResponse{ + URL: url, + ExpiresAt: time.Now().Add(15 * time.Minute), + } + + 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 + } + + 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) + json.NewEncoder(w).Encode(map[string]string{"status": "completed"}) +} + +// HandleAbortMultipart aborts a 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 + } + + 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) + 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"` @@ -760,6 +1067,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/docs/MULTIPART.md b/docs/MULTIPART.md new file mode 100644 index 0000000..34d9718 --- /dev/null +++ b/docs/MULTIPART.md @@ -0,0 +1,570 @@ +S3 Multipart Upload Implementation Plan + + Problem Summary + + Current implementation uses a single presigned URL with a pipe for chunked uploads (PATCH). This causes: + - Docker PATCH requests block waiting for pipe writes + - S3 upload happens in background via single presigned URL + - Docker times out → "client disconnected during blob PATCH" + - Root cause: Single presigned URLs don't support OCI's chunked upload protocol + + Solution: S3 Multipart Upload API + + Implement proper S3 multipart upload to support Docker's chunked PATCH operations: + - Each PATCH → separate S3 part upload with its own presigned URL + - On Commit → complete multipart upload + - No buffering, no pipes, no blocking + + --- + Architecture Changes + + Current (Broken) Flow + + POST /blobs/uploads/ → Create() → Single presigned URL to temp location + PATCH → Write to pipe → [blocks] → Background goroutine uploads via single URL + PATCH → [blocks on pipe] → Docker timeout → disconnect ❌ + + New (Multipart) Flow + + POST /blobs/uploads/ → Create() → Initiate multipart upload, get upload ID + PATCH #1 → Get presigned URL for part 1 → Upload part 1 to S3 → Store ETag + PATCH #2 → Get presigned URL for part 2 → Upload part 2 to S3 → Store ETag + PUT (commit) → Complete multipart upload with ETags → Done ✅ + + --- + Implementation Details + + 1. Hold Service: Add Multipart Upload Endpoints + + File: cmd/hold/main.go + + New Request/Response Types + + // StartMultipartUploadRequest initiates a multipart upload + type StartMultipartUploadRequest struct { + DID string `json:"did"` + Digest string `json:"digest"` + } + + 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"` + } + + 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"` + } + + 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"` + } + + New Endpoints + + POST /start-multipart + func (s *HoldService) HandleStartMultipart(w http.ResponseWriter, r *http.Request) { + // Validate DID authorization for WRITE + // Build S3 key from digest + // Call s3.CreateMultipartUploadRequest() + // Generate presigned URL if needed, or return upload ID + // Return upload ID to client + } + + POST /part-presigned-url + func (s *HoldService) HandleGetPartURL(w http.ResponseWriter, r *http.Request) { + // Validate DID authorization for WRITE + // Build S3 key from digest + // Call s3.UploadPartRequest() with part number and upload ID + // Generate presigned URL + // Return presigned URL for this specific part + } + + POST /complete-multipart + func (s *HoldService) HandleCompleteMultipart(w http.ResponseWriter, r *http.Request) { + // Validate DID authorization for WRITE + // Build S3 key from digest + // Prepare CompletedPart array with part numbers and ETags + // Call s3.CompleteMultipartUpload() + // Return success + } + + POST /abort-multipart (for cleanup) + func (s *HoldService) HandleAbortMultipart(w http.ResponseWriter, r *http.Request) { + // Validate DID authorization for WRITE + // Call s3.AbortMultipartUpload() + // Return success + } + + S3 Implementation + + // 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 + } + + 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)), + }) + + return req.Presign(15 * time.Minute) + } + + // 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, + }, + }) + + return err + } + + --- + 2. AppView: Rewrite ProxyBlobStore for Multipart + + File: pkg/storage/proxy_blob_store.go + + Remove Current Implementation + + - Remove pipe-based streaming + - Remove background goroutine with single presigned URL + - Remove global upload tracking map + + New ProxyBlobWriter Structure + + type ProxyBlobWriter struct { + store *ProxyBlobStore + options distribution.CreateOptions + 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 (for state) + startedAt time.Time + finalDigest string // Set on Commit + } + + type CompletedPart struct { + PartNumber int + ETag string + } + + New Create() - Initiate Multipart Upload + + func (p *ProxyBlobStore) Create(ctx context.Context, options ...distribution.BlobCreateOption) (distribution.BlobWriter, error) { + var opts distribution.CreateOptions + for _, option := range options { + if err := option.Apply(&opts); err != nil { + return nil, err + } + } + + // Use temp digest for upload location + writerID := fmt.Sprintf("upload-%d", time.Now().UnixNano()) + tempDigest := digest.Digest(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) + } + + writer := &ProxyBlobWriter{ + store: p, + options: opts, + uploadID: uploadID, + parts: make([]CompletedPart, 0), + partNumber: 1, + buffer: bytes.NewBuffer(make([]byte, 0, 5*1024*1024)), // 5MB buffer + id: writerID, + startedAt: time.Now(), + } + + // Store in global map for Resume() + globalUploadsMu.Lock() + globalUploads[writer.id] = writer + globalUploadsMu.Unlock() + + return writer, nil + } + + New Write() - Buffer and Flush Parts + + func (w *ProxyBlobWriter) Write(p []byte) (int, error) { + if w.closed { + return 0, fmt.Errorf("writer closed") + } + + n, err := w.buffer.Write(p) + w.size += int64(n) + + // Flush if buffer reaches 5MB (S3 minimum part size) + if w.buffer.Len() >= 5*1024*1024 { + if err := w.flushPart(); err != nil { + return n, err + } + } + + return n, err + } + + 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 := digest.Digest(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) + } + + // Upload part to S3 + req, err := http.NewRequestWithContext(ctx, "PUT", url, bytes.NewReader(w.buffer.Bytes())) + if err != nil { + return err + } + + 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 { + return fmt.Errorf("part upload failed: status %d", resp.StatusCode) + } + + // 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, + }) + + // Reset buffer and increment part number + w.buffer.Reset() + w.partNumber++ + + return nil + } + + New Commit() - Complete Multipart and Move + + 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 + + // Flush any remaining buffered data + if w.buffer.Len() > 0 { + if err := w.flushPart(); err != nil { + // Try to abort multipart on error + w.store.abortMultipartUpload(ctx, w.uploadID) + return distribution.Descriptor{}, err + } + } + + // Complete multipart upload at temp location + tempDigest := digest.Digest(fmt.Sprintf("uploads/temp-%s", w.id)) + if err := w.store.completeMultipartUpload(ctx, tempDigest, w.uploadID, w.parts); err != nil { + return distribution.Descriptor{}, err + } + + // Move from temp → final location (server-side S3 copy) + tempPath := fmt.Sprintf("uploads/temp-%s", w.id) + finalPath := desc.Digest.String() + + moveURL := fmt.Sprintf("%s/move?from=%s&to=%s&did=%s", + w.store.storageEndpoint, tempPath, finalPath, w.store.did) + + req, err := http.NewRequestWithContext(ctx, "POST", moveURL, nil) + if err != nil { + return distribution.Descriptor{}, err + } + + resp, err := w.store.httpClient.Do(req) + if err != nil { + return distribution.Descriptor{}, err + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusCreated { + bodyBytes, _ := io.ReadAll(resp.Body) + return distribution.Descriptor{}, fmt.Errorf("move failed: %d, %s", resp.StatusCode, bodyBytes) + } + + // Remove from global map + globalUploadsMu.Lock() + delete(globalUploads, w.id) + globalUploadsMu.Unlock() + + return distribution.Descriptor{ + Digest: desc.Digest, + Size: w.size, + MediaType: desc.MediaType, + }, nil + } + + Add Hold Service Client Methods + + func (p *ProxyBlobStore) startMultipartUpload(ctx context.Context, dgst digest.Digest) (string, error) { + reqBody := map[string]any{ + "did": p.did, + "digest": dgst.String(), + } + body, _ := json.Marshal(reqBody) + + url := fmt.Sprintf("%s/start-multipart", p.storageEndpoint) + req, _ := http.NewRequestWithContext(ctx, "POST", url, bytes.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + + resp, err := p.httpClient.Do(req) + if err != nil { + return "", err + } + defer resp.Body.Close() + + var result struct { + UploadID string `json:"upload_id"` + } + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return "", err + } + + return result.UploadID, nil + } + + func (p *ProxyBlobStore) getPartPresignedURL(ctx context.Context, dgst digest.Digest, uploadID string, partNumber int) (string, error) { + reqBody := map[string]any{ + "did": p.did, + "digest": dgst.String(), + "upload_id": uploadID, + "part_number": partNumber, + } + body, _ := json.Marshal(reqBody) + + url := fmt.Sprintf("%s/part-presigned-url", p.storageEndpoint) + req, _ := http.NewRequestWithContext(ctx, "POST", url, bytes.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + + resp, err := p.httpClient.Do(req) + if err != nil { + return "", err + } + defer resp.Body.Close() + + var result struct { + URL string `json:"url"` + } + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return "", err + } + + return result.URL, nil + } + + func (p *ProxyBlobStore) completeMultipartUpload(ctx context.Context, dgst digest.Digest, uploadID string, parts []CompletedPart) error { + reqBody := map[string]any{ + "did": p.did, + "digest": dgst.String(), + "upload_id": uploadID, + "parts": parts, + } + body, _ := json.Marshal(reqBody) + + url := fmt.Sprintf("%s/complete-multipart", p.storageEndpoint) + req, _ := http.NewRequestWithContext(ctx, "POST", url, bytes.NewReader(body)) + 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 { + return fmt.Errorf("complete multipart failed: status %d", resp.StatusCode) + } + + return nil + } + + --- + Testing Plan + + 1. Unit Tests + + - Test multipart upload initiation + - Test part upload with presigned URLs + - Test completion with ETags + - Test abort on errors + + 2. Integration Tests + + - Push small images (< 5MB, single part) + - Push medium images (10MB, 2 parts) + - Push large images (100MB, 20 parts) + - Test with Upcloud S3 + - Test with Storj S3 + + 3. Validation + + - Monitor logs for "client disconnected" errors (should be gone) + - Check Docker push success rate + - Verify blobs stored correctly in S3 + - Check bandwidth usage on hold service (should be minimal) + + --- + Migration & Deployment + + Backward Compatibility + + - Keep /put-presigned-url endpoint for fallback + - Keep /move endpoint (still needed) + - New multipart endpoints are additive + + Deployment Steps + + 1. Update hold service with new endpoints + 2. Update AppView ProxyBlobStore + 3. Deploy hold service first + 4. Deploy AppView + 5. Test with sample push + 6. Monitor logs + + Rollback Plan + + - Revert AppView to previous version (uses old presigned URL method) + - Hold service keeps both old and new endpoints + + --- + Documentation Updates + + Update docs/PRESIGNED_URLS.md + + - Add section "Multipart Upload for Chunked Data" + - Explain why single presigned URLs don't work with PATCH + - Document new endpoints and flow + - Add S3 part size recommendations (5MB-64MB for Storj) + + Add Troubleshooting Section + + - "Client disconnected during PATCH" → resolved by multipart + - Storj-specific considerations (64MB parts recommended) + - Upcloud compatibility notes + + --- + Performance Impact + + Before (Broken) + + - Docker PATCH → blocks on pipe → timeout → retry → fail + - Unable to push large images reliably + + After (Multipart) + + - Each PATCH → independent part upload → immediate response + - No blocking, no timeouts + - Parallel part uploads possible (future optimization) + - Reliable pushes for any image size + + Bandwidth + + - Hold service: Only API calls (~1KB per part) + - Direct S3 uploads: Full blob data + - S3 copy for move: Server-side (no hold bandwidth) + + Estimated savings: 99.98% hold service bandwidth reduction (same as before, but now actually works!) diff --git a/docs/PRESIGNED_URLS.md b/docs/PRESIGNED_URLS.md index d9cfca3..a262405 100644 --- a/docs/PRESIGNED_URLS.md +++ b/docs/PRESIGNED_URLS.md @@ -110,6 +110,47 @@ Move: AppView → Hold Service → S3 (server-side CopyObject API) **Move path:** S3 internal copy (no data transfer!) **Hold service bandwidth:** ~2KB (presigned URL + CopyObject API) +### For Chunked Uploads (Multipart Upload) + +**Large blobs with OCI chunked protocol (Docker PATCH requests):** + +The OCI Distribution Spec uses chunked uploads via multiple PATCH requests. Single presigned URLs don't support this - we need **S3 Multipart Upload**. + +1. **Docker starts upload:** `POST /v2/alice/myapp/blobs/uploads/` +2. **AppView initiates multipart:** + ```json + POST /start-multipart + {"did": "...", "digest": "uploads/temp-{uuid}"} + → Returns: {"upload_id": "xyz123"} + ``` +3. **Docker sends chunk 1:** `PATCH /v2/.../uploads/{uuid}` (5MB data) +4. **AppView gets part URL:** + ```json + POST /part-presigned-url + {"did": "...", "digest": "uploads/temp-{uuid}", "upload_id": "xyz123", "part_number": 1} + → Returns: {"url": "https://s3.../part?uploadId=xyz123&partNumber=1&..."} + ``` +5. **AppView uploads part 1** using presigned URL → Gets ETag +6. **Docker sends chunk 2:** `PATCH /v2/.../uploads/{uuid}` (5MB data) +7. **Repeat steps 4-5** for part 2 (and subsequent parts) +8. **Docker finalizes:** `PUT /v2/.../uploads/{uuid}?digest=sha256:abc123` +9. **AppView completes multipart:** + ```json + POST /complete-multipart + {"did": "...", "digest": "uploads/temp-{uuid}", "upload_id": "xyz123", + "parts": [{"part_number": 1, "etag": "..."}, {"part_number": 2, "etag": "..."}]} + ``` +10. **AppView requests move:** `POST /move?from=uploads/temp-{uuid}&to=sha256:abc123` +11. **Hold service executes S3 server-side copy** (same as above) + +**Data path:** Docker → AppView (buffers 5MB) → S3 (via presigned URL per part) +**Each PATCH:** Independent, non-blocking, immediate response +**Hold service bandwidth:** ~1KB per part + ~1KB for completion + +**Why This Fixes "Client Disconnected" Errors:** +- Previous implementation: Single presigned URL + pipe → PATCH blocks → Docker timeout +- New implementation: Each PATCH → separate part upload → immediate response → no blocking + ## Why the Temp → Final Move is Required This is **not an ATCR implementation detail** — it's required by the [OCI Distribution Specification](https://github.com/opencontainers/distribution-spec/blob/main/spec.md#push). @@ -305,14 +346,111 @@ func (s *HoldService) getProxyUploadURL(digest, did string) string { } ``` -### 3. No Changes Needed for Move Operation +### 3. Multipart Upload Endpoints (Required for Chunked Uploads) + +**File: `cmd/hold/main.go`** + +#### Start Multipart Upload + +```go +func (s *HoldService) HandleStartMultipart(w http.ResponseWriter, r *http.Request) { + var req StartMultipartUploadRequest // {did, digest} + + // Validate DID authorization for WRITE + if !s.isAuthorizedWrite(req.DID) { + // Return 403 Forbidden + } + + // Initiate S3 multipart upload + result, err := s.s3Client.CreateMultipartUploadWithContext(ctx, &s3.CreateMultipartUploadInput{ + Bucket: aws.String(s.bucket), + Key: aws.String(s3Key), + }) + + // Return upload ID + json.NewEncoder(w).Encode(StartMultipartUploadResponse{ + UploadID: *result.UploadId, + ExpiresAt: time.Now().Add(24 * time.Hour), + }) +} +``` + +**Route:** `POST /start-multipart` + +#### Get Part Presigned URL + +```go +func (s *HoldService) HandleGetPartURL(w http.ResponseWriter, r *http.Request) { + var req GetPartURLRequest // {did, digest, upload_id, part_number} + + // Generate presigned URL for specific part + 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) + + json.NewEncoder(w).Encode(GetPartURLResponse{URL: url}) +} +``` + +**Route:** `POST /part-presigned-url` + +#### Complete Multipart Upload + +```go +func (s *HoldService) HandleCompleteMultipart(w http.ResponseWriter, r *http.Request) { + var req CompleteMultipartRequest // {did, digest, upload_id, parts: [{part_number, etag}]} + + // Convert parts to S3 format + s3Parts := make([]*s3.CompletedPart, len(req.Parts)) + for i, p := range req.Parts { + s3Parts[i] = &s3.CompletedPart{ + PartNumber: aws.Int64(int64(p.PartNumber)), + ETag: aws.String(p.ETag), + } + } + + // Complete multipart upload + _, 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}, + }) +} +``` + +**Route:** `POST /complete-multipart` + +#### Abort Multipart Upload + +```go +func (s *HoldService) HandleAbortMultipart(w http.ResponseWriter, r *http.Request) { + var req AbortMultipartRequest // {did, digest, upload_id} + + // Abort and cleanup parts + _, err := s.s3Client.AbortMultipartUploadWithContext(ctx, &s3.AbortMultipartUploadInput{ + Bucket: aws.String(s.bucket), + Key: aws.String(s3Key), + UploadId: aws.String(uploadID), + }) +} +``` + +**Route:** `POST /abort-multipart` + +### 4. Move Operation (No Changes) The existing `/move` endpoint already uses `driver.Move()`, which for S3: - Calls `s3.CopyObject()` (server-side copy) - Calls `s3.DeleteObject()` (delete source) - No data transfer through hold service! -**File: `cmd/hold/main.go:296` (already exists, no changes needed)** +**File: `cmd/hold/main.go:393` (already exists, no changes needed)** ```go func (s *HoldService) HandleMove(w http.ResponseWriter, r *http.Request) { @@ -328,7 +466,7 @@ func (s *HoldService) HandleMove(w http.ResponseWriter, r *http.Request) { } ``` -### 4. AppView Changes (Optional Optimization) +### 5. AppView Changes (Multipart Upload Implementation) **File: `pkg/storage/proxy_blob_store.go:228`** diff --git a/pkg/auth/token/handler.go b/pkg/auth/token/handler.go index d04fa5d..41cdb1b 100644 --- a/pkg/auth/token/handler.go +++ b/pkg/auth/token/handler.go @@ -95,8 +95,6 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { did = device.DID handle = device.Handle - fmt.Printf("DEBUG [token/handler]: Device secret validated for DID=%s, handle=%s\n", did, handle) - // Device is linked to OAuth session via DID // OAuth refresher will provide access token when needed via middleware } else { @@ -150,8 +148,6 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { return } - fmt.Printf("DEBUG [token/handler]: Access validated for DID=%s\n", did) - // Issue JWT token tokenString, err := h.issuer.Issue(did, access) if err != nil { @@ -161,7 +157,6 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { } fmt.Printf("DEBUG [token/handler]: Issued JWT token (length=%d) for DID=%s\n", len(tokenString), did) - fmt.Printf("DEBUG [token/handler]: JWT Token: %s\n", tokenString) // Return token response now := time.Now() diff --git a/pkg/middleware/registry.go b/pkg/middleware/registry.go index a0e9660..720c569 100644 --- a/pkg/middleware/registry.go +++ b/pkg/middleware/registry.go @@ -117,7 +117,6 @@ func (nr *NamespaceResolver) Repository(ctx context.Context, name reference.Name return nil, fmt.Errorf("no storage endpoint configured: ensure default_storage_endpoint is set in middleware config") } ctx = context.WithValue(ctx, "storage.endpoint", storageEndpoint) - fmt.Printf("DEBUG [registry/middleware]: Using storage endpoint: %s\n", storageEndpoint) // Create a new reference with identity/image format // Use the identity (or DID) as the namespace to ensure canonical format @@ -145,7 +144,6 @@ func (nr *NamespaceResolver) Repository(ctx context.Context, name reference.Name if err == nil { // OAuth session available - use indigo's API client (handles DPoP automatically) apiClient := session.APIClient() - fmt.Printf("DEBUG [registry/middleware]: Using OAuth session with indigo API client for DID=%s\n", did) atprotoClient = atproto.NewClientWithIndigoClient(pdsEndpoint, did, apiClient) } else { fmt.Printf("DEBUG [registry/middleware]: OAuth refresh failed for DID=%s: %v, falling back to Basic Auth\n", did, err) @@ -174,12 +172,9 @@ func (nr *NamespaceResolver) Repository(ctx context.Context, name reference.Name // Check cache first if cached, ok := nr.repositories.Load(cacheKey); ok { - fmt.Printf("DEBUG [registry/middleware]: Using cached RoutingRepository for %s\n", cacheKey) return cached.(*storage.RoutingRepository), nil } - fmt.Printf("DEBUG [registry/middleware]: Creating new RoutingRepository for image=%s (ATProto repo name)\n", repositoryName) - // Create routing repository - routes manifests to ATProto, blobs to hold service // The registry is stateless - no local storage is used // Pass storage endpoint and DID as parameters (can't use context as it gets lost) diff --git a/pkg/storage/proxy_blob_store.go b/pkg/storage/proxy_blob_store.go index 5d8f08c..cab1160 100644 --- a/pkg/storage/proxy_blob_store.go +++ b/pkg/storage/proxy_blob_store.go @@ -7,6 +7,7 @@ import ( "fmt" "io" "net/http" + "strings" "sync" "time" @@ -15,11 +16,17 @@ import ( ) const ( - // maxChunkSize is the maximum buffer size before flushing to hold service - // Matches S3's minimum multipart upload size - maxChunkSize = 5 * 1024 * 1024 // 5MB + // minPartSize is S3's minimum part size for multipart uploads + // Parts must be at least 5MB (except the last part) + minPartSize = 5 * 1024 * 1024 // 5MB ) +// CompletedPart represents a completed multipart upload part +type CompletedPart struct { + PartNumber int `json:"part_number"` + ETag string `json:"etag"` +} + // Global upload tracking (shared across all ProxyBlobStore instances) // This is necessary because distribution creates new repository/blob store instances per request var ( @@ -198,87 +205,36 @@ func (p *ProxyBlobStore) Create(ctx context.Context, options ...distribution.Blo } } - // Create pipe for streaming upload - pipeReader, pipeWriter := io.Pipe() - uploadErr := make(chan error, 1) - digestChan := make(chan string, 1) + // Use temp digest for upload location + writerID := fmt.Sprintf("upload-%d", time.Now().UnixNano()) + tempPath := fmt.Sprintf("uploads/temp-%s", writerID) + tempDigest := digest.Digest(tempPath) + + // 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) + } + + fmt.Printf("DEBUG [proxy_blob_store/Create]: Started multipart upload: id=%s, uploadID=%s\n", writerID, 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, minPartSize)), + id: writerID, startedAt: time.Now(), + tempDigest: tempDigest, } - // Store in global uploads map for resume support + // Store in global map for Resume() globalUploadsMu.Lock() 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("DEBUG [goroutine]: Starting upload to temp: url=%s\n", url) - - // 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 } @@ -293,7 +249,6 @@ func (p *ProxyBlobStore) Resume(ctx context.Context, id string) (distribution.Bl return nil, distribution.ErrBlobUploadUnknown } - // With streaming, no flush needed - just return the writer return writer, nil } @@ -384,15 +339,16 @@ func (p *ProxyBlobStore) getUploadURL(ctx context.Context, dgst digest.Digest, s 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 + tempDigest digest.Digest // Temp location digest } // ID returns the upload ID @@ -406,22 +362,81 @@ func (w *ProxyBlobWriter) StartedAt() time.Time { } // Write writes data to the upload -// Streams directly to hold service via pipe +// Buffers data and flushes parts when buffer reaches minPartSize 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 minimum part size (5MB) + if w.buffer.Len() >= minPartSize { + if err := w.flushPart(); err != nil { + return n, fmt.Errorf("failed to flush part: %w", err) + } + } + + return n, err +} + +// flushPart uploads the current buffer as a multipart upload 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 + url, err := w.store.getPartPresignedURL(ctx, w.tempDigest, w.uploadID, w.partNumber) + if err != nil { + return fmt.Errorf("failed to get part presigned URL: %w", err) + } + + // 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") + + fmt.Printf("DEBUG [proxy_blob_store/flushPart]: Uploading part %d, size=%d bytes\n", w.partNumber, w.buffer.Len()) + + resp, err := w.store.httpClient.Do(req) + if err != nil { + return fmt.Errorf("part upload failed: %w", 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") + } + + // Remove quotes from ETag if present (S3 sometimes adds them) + etag = strings.Trim(etag, "\"") + + w.parts = append(w.parts, CompletedPart{ + PartNumber: w.partNumber, + ETag: etag, + }) + + fmt.Printf("DEBUG [proxy_blob_store/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 @@ -466,33 +481,30 @@ func (w *ProxyBlobWriter) Commit(ctx context.Context, desc distribution.Descript } w.closed = true - // 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 as the final part + if w.buffer.Len() > 0 { + if err := w.flushPart(); err != nil { + // Try to abort multipart on error + w.store.abortMultipartUpload(ctx, w.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 + if err := w.store.completeMultipartUpload(ctx, w.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 + fmt.Printf("DEBUG [proxy_blob_store/Commit]: Completed multipart upload with %d parts, total size=%d\n", len(w.parts), w.size) + + // Move from temp → final location (server-side S3 copy) + tempPath := fmt.Sprintf("uploads/temp-%s", w.id) finalPath := desc.Digest.String() 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) } @@ -505,10 +517,15 @@ func (w *ProxyBlobWriter) Commit(ctx context.Context, desc distribution.Descript if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusCreated { bodyBytes, _ := io.ReadAll(resp.Body) - return distribution.Descriptor{}, fmt.Errorf("move blob failed: status %d, body: %s", resp.StatusCode, string(bodyBytes)) + return distribution.Descriptor{}, fmt.Errorf("move 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) + // Remove from global map + globalUploadsMu.Lock() + delete(globalUploads, w.id) + globalUploadsMu.Unlock() + + fmt.Printf("DEBUG [proxy_blob_store/Commit]: Successfully committed: digest=%s, size=%d\n", desc.Digest, w.size) return distribution.Descriptor{ Digest: desc.Digest, @@ -526,29 +543,144 @@ func (w *ProxyBlobWriter) Cancel(ctx context.Context) error { 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 on S3 + if err := w.store.abortMultipartUpload(ctx, w.tempDigest, w.uploadID); err != nil { + fmt.Printf("DEBUG [proxy_blob_store/Cancel]: Failed to abort multipart upload: %v\n", err) + // Continue anyway - we still want to clean up } - // Wait for goroutine to finish - <-w.uploadErr - - fmt.Printf("DEBUG [proxy_blob_store]: Cancelled upload: id=%s\n", w.id) + fmt.Printf("DEBUG [proxy_blob_store/Cancel]: Cancelled upload: id=%s, uploadID=%s\n", w.id, w.uploadID) return nil } // Close closes the writer -// Just returns - streaming continues via pipe +// Does nothing - actual completion happens in Commit() or Cancel() 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 } +// startMultipartUpload initiates a multipart upload via hold service +func (p *ProxyBlobStore) startMultipartUpload(ctx context.Context, dgst digest.Digest) (string, error) { + reqBody := map[string]any{ + "did": p.did, + "digest": dgst.String(), + } + body, _ := json.Marshal(reqBody) + + url := fmt.Sprintf("%s/start-multipart", p.storageEndpoint) + req, _ := http.NewRequestWithContext(ctx, "POST", url, bytes.NewReader(body)) + 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 { + return "", fmt.Errorf("failed to start multipart upload: status %d", resp.StatusCode) + } + + 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, dgst digest.Digest, uploadID string, partNumber int) (string, error) { + reqBody := map[string]any{ + "did": p.did, + "digest": dgst.String(), + "upload_id": uploadID, + "part_number": partNumber, + } + body, _ := json.Marshal(reqBody) + + url := fmt.Sprintf("%s/part-presigned-url", p.storageEndpoint) + req, _ := http.NewRequestWithContext(ctx, "POST", url, bytes.NewReader(body)) + 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 { + return "", fmt.Errorf("failed to get part presigned URL: status %d", resp.StatusCode) + } + + 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 +func (p *ProxyBlobStore) completeMultipartUpload(ctx context.Context, dgst digest.Digest, uploadID string, parts []CompletedPart) error { + reqBody := map[string]any{ + "did": p.did, + "digest": dgst.String(), + "upload_id": uploadID, + "parts": parts, + } + body, _ := json.Marshal(reqBody) + + url := fmt.Sprintf("%s/complete-multipart", p.storageEndpoint) + req, _ := http.NewRequestWithContext(ctx, "POST", url, bytes.NewReader(body)) + 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 +func (p *ProxyBlobStore) abortMultipartUpload(ctx context.Context, dgst digest.Digest, uploadID string) error { + reqBody := map[string]any{ + "did": p.did, + "digest": dgst.String(), + "upload_id": uploadID, + } + body, _ := json.Marshal(reqBody) + + url := fmt.Sprintf("%s/abort-multipart", p.storageEndpoint) + req, _ := http.NewRequestWithContext(ctx, "POST", url, bytes.NewReader(body)) + 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 +} + // readSeekCloser wraps an io.ReadCloser to implement ReadSeekCloser type readSeekCloser struct { io.ReadCloser