···11+# Hold Service Multipart Upload Architecture
22+33+## Overview
44+55+The hold service supports multipart uploads through two modes:
66+1. **S3Native** - Uses S3's native multipart API with presigned URLs (optimal)
77+2. **Buffered** - Buffers parts in hold service memory, assembles on completion (fallback)
88+99+This dual-mode approach enables the hold service to work with:
1010+- S3-compatible storage with presigned URL support (S3, Storj, MinIO, etc.)
1111+- S3-compatible storage WITHOUT presigned URL support
1212+- Filesystem storage
1313+- Any storage driver supported by distribution
1414+1515+## Current State
1616+1717+### What Works ✅
1818+- **S3 Native Mode with presigned URLs**: Fully working! Direct uploads to S3 via presigned URLs
1919+- **Buffered mode with S3**: Tested and working with `DISABLE_PRESIGNED_URLS=true`
2020+- **Filesystem storage**: Tested and working! Buffered mode with filesystem driver
2121+- **AppView multipart client**: Implements chunked uploads via multipart API
2222+- **MultipartManager**: Session tracking, automatic cleanup, thread-safe operations
2323+- **Automatic fallback**: Falls back to buffered mode when S3 unavailable or disabled
2424+- **ETag normalization**: Handles quoted/unquoted ETags from S3
2525+- **Route handler**: `/multipart-parts/{uploadID}/{partNumber}` endpoint added and tested
2626+2727+### All Implementation Complete! 🎉
2828+All three multipart upload modes are fully implemented, tested, and working in production.
2929+3030+### Bugs Fixed 🔧
3131+- **Missing S3 parts in complete**: For S3Native mode, parts uploaded directly to S3 weren't being recorded. Fixed by storing parts from request in `HandleCompleteMultipart` before calling `CompleteMultipartUploadWithManager`.
3232+- **Malformed XML error from S3**: S3 requires ETags to be quoted in CompleteMultipartUpload XML. Added `normalizeETag()` function to ensure quotes are present.
3333+- **Route missing**: `/multipart-parts/{uploadID}/{partNumber}` not registered in cmd/hold/main.go. Fixed by adding route handler with path parsing.
3434+- **MultipartMgr access**: Field was private, preventing route handler access. Fixed by exporting as `MultipartMgr`.
3535+- **DISABLE_PRESIGNED_URLS not logged**: `initS3Client()` didn't check the flag before initializing. Fixed with early return check and proper logging.
3636+3737+## Architecture
3838+3939+### Three Modes of Operation
4040+4141+#### Mode 1: S3 Native Multipart ✅ WORKING
4242+```
4343+Docker → AppView → Hold → S3 (presigned URLs)
4444+ ↓
4545+ Returns presigned URL
4646+ ↓
4747+Docker ──────────→ S3 (direct upload)
4848+```
4949+5050+**Flow:**
5151+1. AppView: `POST /start-multipart` → Hold starts S3 multipart, returns uploadID
5252+2. AppView: `POST /part-presigned-url` → Hold returns S3 presigned URL
5353+3. Docker → S3: Direct upload via presigned URL
5454+4. AppView: `POST /complete-multipart` → Hold calls S3 CompleteMultipartUpload
5555+5656+**Advantages:**
5757+- No data flows through hold service
5858+- Minimal bandwidth usage
5959+- Fast uploads
6060+6161+#### Mode 2: S3 Proxy Mode (Buffered) ✅ WORKING
6262+```
6363+Docker → AppView → Hold → S3 (via driver)
6464+ ↓
6565+ Buffers & proxies
6666+ ↓
6767+ S3
6868+```
6969+7070+**Flow:**
7171+1. AppView: `POST /start-multipart` → Hold creates buffered session
7272+2. AppView: `POST /part-presigned-url` → Hold returns proxy URL
7373+3. Docker → Hold: `PUT /multipart-parts/{uploadID}/{part}` → Hold buffers
7474+4. AppView: `POST /complete-multipart` → Hold uploads to S3 via driver
7575+7676+**Use Cases:**
7777+- S3 provider doesn't support presigned URLs
7878+- S3 API fails to generate presigned URL
7979+- Fallback from Mode 1
8080+8181+#### Mode 3: Filesystem Mode ✅ WORKING
8282+```
8383+Docker → AppView → Hold (filesystem driver)
8484+ ↓
8585+ Buffers & writes
8686+ ↓
8787+ Local filesystem
8888+```
8989+9090+**Flow:**
9191+Same as Mode 2, but writes to filesystem driver instead of S3 driver.
9292+9393+**Use Cases:**
9494+- Development/testing with local filesystem
9595+- Small deployments without S3
9696+- Air-gapped environments
9797+9898+## Implementation: pkg/hold/multipart.go
9999+100100+### Core Components
101101+102102+#### MultipartManager
103103+```go
104104+type MultipartManager struct {
105105+ sessions map[string]*MultipartSession
106106+ mu sync.RWMutex
107107+}
108108+```
109109+110110+**Responsibilities:**
111111+- Track active multipart sessions
112112+- Clean up abandoned uploads (>24h inactive)
113113+- Thread-safe session access
114114+115115+#### MultipartSession
116116+```go
117117+type MultipartSession struct {
118118+ UploadID string // Unique ID for this upload
119119+ Digest string // Target blob digest
120120+ Mode MultipartMode // S3Native or Buffered
121121+ S3UploadID string // S3 upload ID (S3Native only)
122122+ Parts map[int]*MultipartPart // Buffered parts (Buffered only)
123123+ CreatedAt time.Time
124124+ LastActivity time.Time
125125+}
126126+```
127127+128128+**State Tracking:**
129129+- S3Native: Tracks S3 upload ID and part ETags
130130+- Buffered: Stores part data in memory
131131+132132+#### MultipartPart
133133+```go
134134+type MultipartPart struct {
135135+ PartNumber int // Part number (1-indexed)
136136+ Data []byte // Part data (Buffered mode only)
137137+ ETag string // S3 ETag or computed hash
138138+ Size int64
139139+}
140140+```
141141+142142+### Key Methods
143143+144144+#### StartMultipartUploadWithManager
145145+```go
146146+func (s *HoldService) StartMultipartUploadWithManager(
147147+ ctx context.Context,
148148+ digest string,
149149+ manager *MultipartManager,
150150+) (string, MultipartMode, error)
151151+```
152152+153153+**Logic:**
154154+1. Try S3 native multipart via `s.startMultipartUpload()`
155155+2. If successful → Create S3Native session
156156+3. If fails or no S3 client → Create Buffered session
157157+4. Return uploadID and mode
158158+159159+#### GetPartUploadURL
160160+```go
161161+func (s *HoldService) GetPartUploadURL(
162162+ ctx context.Context,
163163+ session *MultipartSession,
164164+ partNumber int,
165165+ did string,
166166+) (string, error)
167167+```
168168+169169+**Logic:**
170170+- S3Native mode: Generate S3 presigned URL via `s.getPartPresignedURL()`
171171+- Buffered mode: Return proxy endpoint `/multipart-parts/{uploadID}/{part}`
172172+173173+#### CompleteMultipartUploadWithManager
174174+```go
175175+func (s *HoldService) CompleteMultipartUploadWithManager(
176176+ ctx context.Context,
177177+ session *MultipartSession,
178178+ manager *MultipartManager,
179179+) error
180180+```
181181+182182+**Logic:**
183183+- S3Native: Call `s.completeMultipartUpload()` with S3 API
184184+- Buffered: Assemble parts in order, write via storage driver
185185+186186+#### HandleMultipartPartUpload (New Endpoint)
187187+```go
188188+func (s *HoldService) HandleMultipartPartUpload(
189189+ w http.ResponseWriter,
190190+ r *http.Request,
191191+ uploadID string,
192192+ partNumber int,
193193+ did string,
194194+ manager *MultipartManager,
195195+)
196196+```
197197+198198+**New HTTP endpoint:** `PUT /multipart-parts/{uploadID}/{partNumber}`
199199+200200+**Purpose:** Receive part uploads in Buffered mode
201201+202202+**Logic:**
203203+1. Validate session exists and is in Buffered mode
204204+2. Authorize write access
205205+3. Read part data from request body
206206+4. Store in session with computed ETag (SHA256)
207207+5. Return ETag in response header
208208+209209+## Integration Plan
210210+211211+### Phase 1: Migrate to pkg/hold (COMPLETE)
212212+- [x] Extract code from cmd/hold/main.go to pkg/hold/
213213+- [x] Create isolated multipart.go implementation
214214+- [x] Update cmd/hold/main.go to import pkg/hold
215215+- [x] Test existing functionality works
216216+217217+### Phase 2: Add Buffered Mode Support (COMPLETE ✅)
218218+- [x] Add MultipartManager to HoldService
219219+- [x] Update handlers to use `*WithManager` methods
220220+- [x] Add DISABLE_PRESIGNED_URLS environment variable for testing
221221+- [x] Implement presigned URL disable checks in all methods
222222+- [x] **Fixed: Record S3 parts from request in HandleCompleteMultipart**
223223+- [x] **Fixed: ETag normalization (add quotes for S3 XML)**
224224+- [x] **Test S3 native mode with presigned URLs** ✅ WORKING
225225+- [x] **Add route in cmd/hold/main.go** ✅ COMPLETE
226226+- [x] **Export MultipartMgr field for route handler access** ✅ COMPLETE
227227+- [x] **Test DISABLE_PRESIGNED_URLS=true with S3 storage** ✅ WORKING
228228+- [x] **Test filesystem storage with buffered multipart** ✅ WORKING
229229+230230+### Phase 3: Update AppView
231231+- [ ] Detect hold capabilities (presigned vs proxy)
232232+- [ ] Fallback to buffered mode when presigned fails
233233+- [ ] Handle `/multipart-parts/` proxy URLs
234234+235235+### Phase 4: Capability Discovery
236236+- [ ] Add capability endpoint: `GET /capabilities`
237237+- [ ] Return: `{"multipart": "native|buffered|both", "storage": "s3|filesystem"}`
238238+- [ ] AppView uses capabilities to choose upload strategy
239239+240240+## Testing Strategy
241241+242242+### Unit Tests
243243+- [ ] MultipartManager session lifecycle
244244+- [ ] Part buffering and assembly
245245+- [ ] Concurrent part uploads (thread safety)
246246+- [ ] Session cleanup (expired uploads)
247247+248248+### Integration Tests
249249+250250+**S3 Native Mode:**
251251+- [x] Start multipart → get presigned URLs → upload parts → complete ✅ WORKING
252252+- [x] Verify no data flows through hold service (only ~1KB API calls)
253253+- [ ] Test abort cleanup
254254+255255+**Buffered Mode (S3 with DISABLE_PRESIGNED_URLS):**
256256+- [x] Start multipart → get proxy URLs → upload parts → complete ✅ WORKING
257257+- [x] Verify parts assembled correctly
258258+- [ ] Test missing part detection
259259+- [ ] Test abort cleanup
260260+261261+**Buffered Mode (Filesystem):**
262262+- [x] Start multipart → get proxy URLs → upload parts → complete ✅ WORKING
263263+- [x] Verify parts assembled correctly ✅ WORKING
264264+- [x] Verify blobs written to filesystem ✅ WORKING
265265+- [ ] Test missing part detection
266266+- [ ] Test abort cleanup
267267+268268+### Load Tests
269269+- [ ] Concurrent multipart uploads (multiple sessions)
270270+- [ ] Large blobs (100MB+, many parts)
271271+- [ ] Memory usage with many buffered parts
272272+273273+## Performance Considerations
274274+275275+### Memory Usage (Buffered Mode)
276276+- Parts stored in memory until completion
277277+- Docker typically uses 5MB chunks (S3 minimum)
278278+- 100MB image = ~20 parts = ~100MB RAM during upload
279279+- Multiple concurrent uploads multiply memory usage
280280+281281+**Mitigation:**
282282+- Session cleanup (24h timeout)
283283+- Consider disk-backed buffering for large parts (future optimization)
284284+- Monitor memory usage and set limits
285285+286286+### Network Bandwidth
287287+- S3Native: Minimal (only API calls)
288288+- Buffered: Full blob data flows through hold service
289289+- Filesystem: Always buffered (no presigned URL option)
290290+291291+## Configuration
292292+293293+### Environment Variables
294294+295295+**Current (S3 only):**
296296+```bash
297297+STORAGE_DRIVER=s3
298298+S3_BUCKET=my-bucket
299299+S3_ENDPOINT=https://s3.amazonaws.com
300300+AWS_ACCESS_KEY_ID=...
301301+AWS_SECRET_ACCESS_KEY=...
302302+```
303303+304304+**Filesystem:**
305305+```bash
306306+STORAGE_DRIVER=filesystem
307307+STORAGE_ROOT_DIR=/var/lib/atcr/hold
308308+```
309309+310310+### Automatic Mode Selection
311311+No configuration needed - hold service automatically:
312312+1. Tries S3 native multipart if S3 client exists
313313+2. Falls back to buffered mode if S3 unavailable or fails
314314+3. Always uses buffered mode for filesystem driver
315315+316316+## Security Considerations
317317+318318+### Authorization
319319+- All multipart operations require write authorization
320320+- Buffered mode: Check auth on every part upload
321321+- S3Native: Auth only on start/complete (presigned URLs have embedded auth)
322322+323323+### Resource Limits
324324+- Max upload size: Controlled by storage backend
325325+- Max concurrent uploads: Limited by memory
326326+- Session timeout: 24 hours (configurable)
327327+328328+### Attack Vectors
329329+- **Memory exhaustion**: Attacker uploads many large parts
330330+ - Mitigation: Session limits, cleanup, auth
331331+- **Incomplete uploads**: Attacker starts but never completes
332332+ - Mitigation: 24h timeout, cleanup goroutine
333333+- **Part flooding**: Upload many tiny parts
334334+ - Mitigation: S3 has 10,000 part limit, could add to buffered mode
335335+336336+## Future Enhancements
337337+338338+### Disk-Backed Buffering
339339+Instead of memory, buffer parts to temporary disk location:
340340+- Reduces memory pressure
341341+- Supports larger uploads
342342+- Requires cleanup on completion/abort
343343+344344+### Parallel Part Assembly
345345+For large uploads, assemble parts in parallel:
346346+- Stream parts to writer as they arrive
347347+- Reduce memory footprint
348348+- Faster completion
349349+350350+### Chunked Completion
351351+For very large assembled blobs:
352352+- Stream to storage driver in chunks
353353+- Avoid loading entire blob in memory
354354+- Use `io.Copy()` with buffer
355355+356356+### Multi-Backend Support
357357+- Azure Blob Storage multipart
358358+- Google Cloud Storage resumable uploads
359359+- Backblaze B2 large file API
360360+361361+## Implementation Complete ✅
362362+363363+The buffered multipart mode is fully implemented with the following components:
364364+365365+**Route Handler** (`cmd/hold/main.go:47-73`):
366366+- Endpoint: `PUT /multipart-parts/{uploadID}/{partNumber}`
367367+- Parses URL path to extract uploadID and partNumber
368368+- Delegates to `service.HandleMultipartPartUpload()`
369369+370370+**Exported Manager** (`pkg/hold/service.go:20`):
371371+- Field `MultipartMgr` is now exported for route handler access
372372+- All handlers updated to use `s.MultipartMgr`
373373+374374+**Configuration Check** (`pkg/hold/s3.go:20-25`):
375375+- `initS3Client()` checks `DISABLE_PRESIGNED_URLS` flag before initializing
376376+- Logs clear message when presigned URLs are disabled
377377+- Prevents misleading "S3 presigned URLs enabled" message
378378+379379+## Testing Multipart Modes
380380+381381+### Test 1: S3 Native Mode (presigned URLs) ✅ TESTED
382382+```bash
383383+export STORAGE_DRIVER=s3
384384+export S3_BUCKET=your-bucket
385385+export AWS_ACCESS_KEY_ID=...
386386+export AWS_SECRET_ACCESS_KEY=...
387387+# Do NOT set DISABLE_PRESIGNED_URLS
388388+389389+# Start hold service
390390+./bin/atcr-hold
391391+392392+# Push an image
393393+docker push atcr.io/yourdid/test:latest
394394+395395+# Expected logs:
396396+# "✅ S3 presigned URLs enabled"
397397+# "Started S3 native multipart: uploadID=... s3UploadID=..."
398398+# "Completed multipart upload: digest=... uploadID=... parts=..."
399399+```
400400+401401+**Status**: ✅ Working - Direct uploads to S3, minimal bandwidth through hold service
402402+403403+### Test 2: Buffered Mode with S3 (forced proxy) ✅ TESTED
404404+```bash
405405+export STORAGE_DRIVER=s3
406406+export S3_BUCKET=your-bucket
407407+export AWS_ACCESS_KEY_ID=...
408408+export AWS_SECRET_ACCESS_KEY=...
409409+export DISABLE_PRESIGNED_URLS=true # Force buffered mode
410410+411411+# Start hold service
412412+./bin/atcr-hold
413413+414414+# Push an image
415415+docker push atcr.io/yourdid/test:latest
416416+417417+# Expected logs:
418418+# "⚠️ S3 presigned URLs DISABLED by config (DISABLE_PRESIGNED_URLS=true)"
419419+# "Presigned URLs disabled (DISABLE_PRESIGNED_URLS=true), using buffered mode"
420420+# "Stored part: uploadID=... part=1 size=..."
421421+# "Assembled buffered parts: uploadID=... parts=... totalSize=..."
422422+# "Completed buffered multipart: uploadID=... size=... written=..."
423423+```
424424+425425+**Status**: ✅ Working - Parts buffered in hold service memory, assembled and written to S3 via driver
426426+427427+### Test 3: Filesystem Mode (always buffered) ✅ TESTED
428428+```bash
429429+export STORAGE_DRIVER=filesystem
430430+export STORAGE_ROOT_DIR=/tmp/atcr-hold-test
431431+# DISABLE_PRESIGNED_URLS not needed (filesystem never has presigned URLs)
432432+433433+# Start hold service
434434+./bin/atcr-hold
435435+436436+# Push an image
437437+docker push atcr.io/yourdid/test:latest
438438+439439+# Expected logs:
440440+# "Storage driver is filesystem (not S3), presigned URLs disabled"
441441+# "Started buffered multipart: uploadID=..."
442442+# "Stored part: uploadID=... part=1 size=..."
443443+# "Assembled buffered parts: uploadID=... parts=... totalSize=..."
444444+# "Completed buffered multipart: uploadID=... size=... written=..."
445445+446446+# Verify blobs written to:
447447+ls -lh /var/lib/atcr/hold/docker/registry/v2/blobs/sha256/
448448+# Or from outside container:
449449+docker exec atcr-hold ls -lh /var/lib/atcr/hold/docker/registry/v2/blobs/sha256/
450450+```
451451+452452+**Status**: ✅ Working - Parts buffered in memory, assembled, and written to filesystem via driver
453453+454454+**Note**: Initial HEAD requests will show "Path not found" errors - this is normal! Docker checks if blobs exist before uploading. The errors occur for blobs that haven't been uploaded yet. After upload, subsequent HEAD checks succeed.
455455+456456+## References
457457+458458+- S3 Multipart Upload API: https://docs.aws.amazon.com/AmazonS3/latest/API/API_CreateMultipartUpload.html
459459+- Distribution Storage Driver Interface: https://github.com/distribution/distribution/blob/main/registry/storage/driver/storagedriver.go
460460+- OCI Distribution Spec (Blob Upload): https://github.com/opencontainers/distribution-spec/blob/main/spec.md#pushing-a-blob-in-chunks
+448
docs/MULTIPART_OLD.md
···11+S3 Multipart Upload Implementation Plan
22+ Problem Summary
33+ Current implementation uses a single presigned URL with a pipe for chunked uploads (PATCH). This causes:
44+ - Docker PATCH requests block waiting for pipe writes
55+ - S3 upload happens in background via single presigned URL
66+ - Docker times out → "client disconnected during blob PATCH"
77+ - Root cause: Single presigned URLs don't support OCI's chunked upload protocol
88+ Solution: S3 Multipart Upload API
99+ Implement proper S3 multipart upload to support Docker's chunked PATCH operations:
1010+ - Each PATCH → separate S3 part upload with its own presigned URL
1111+ - On Commit → complete multipart upload
1212+ - No buffering, no pipes, no blocking
1313+ ---
1414+ Architecture Changes
1515+ Current (Broken) Flow
1616+ POST /blobs/uploads/ → Create() → Single presigned URL to temp location
1717+ PATCH → Write to pipe → [blocks] → Background goroutine uploads via single URL
1818+ PATCH → [blocks on pipe] → Docker timeout → disconnect ❌
1919+ New (Multipart) Flow
2020+ POST /blobs/uploads/ → Create() → Initiate multipart upload, get upload ID
2121+ PATCH #1 → Get presigned URL for part 1 → Upload part 1 to S3 → Store ETag
2222+ PATCH #2 → Get presigned URL for part 2 → Upload part 2 to S3 → Store ETag
2323+ PUT (commit) → Complete multipart upload with ETags → Done ✅
2424+ ---
2525+ Implementation Details
2626+ 1. Hold Service: Add Multipart Upload Endpoints
2727+ File: cmd/hold/main.go
2828+ New Request/Response Types
2929+ // StartMultipartUploadRequest initiates a multipart upload
3030+ type StartMultipartUploadRequest struct {
3131+ DID string `json:"did"`
3232+ Digest string `json:"digest"`
3333+ }
3434+ type StartMultipartUploadResponse struct {
3535+ UploadID string `json:"upload_id"`
3636+ ExpiresAt time.Time `json:"expires_at"`
3737+ }
3838+ // GetPartURLRequest requests a presigned URL for a specific part
3939+ type GetPartURLRequest struct {
4040+ DID string `json:"did"`
4141+ Digest string `json:"digest"`
4242+ UploadID string `json:"upload_id"`
4343+ PartNumber int `json:"part_number"`
4444+ }
4545+ type GetPartURLResponse struct {
4646+ URL string `json:"url"`
4747+ ExpiresAt time.Time `json:"expires_at"`
4848+ }
4949+ // CompleteMultipartRequest completes a multipart upload
5050+ type CompleteMultipartRequest struct {
5151+ DID string `json:"did"`
5252+ Digest string `json:"digest"`
5353+ UploadID string `json:"upload_id"`
5454+ Parts []CompletedPart `json:"parts"`
5555+ }
5656+ type CompletedPart struct {
5757+ PartNumber int `json:"part_number"`
5858+ ETag string `json:"etag"`
5959+ }
6060+ // AbortMultipartRequest aborts an in-progress upload
6161+ type AbortMultipartRequest struct {
6262+ DID string `json:"did"`
6363+ Digest string `json:"digest"`
6464+ UploadID string `json:"upload_id"`
6565+ }
6666+ New Endpoints
6767+ POST /start-multipart
6868+ func (s *HoldService) HandleStartMultipart(w http.ResponseWriter, r *http.Request) {
6969+ // Validate DID authorization for WRITE
7070+ // Build S3 key from digest
7171+ // Call s3.CreateMultipartUploadRequest()
7272+ // Generate presigned URL if needed, or return upload ID
7373+ // Return upload ID to client
7474+ }
7575+ POST /part-presigned-url
7676+ func (s *HoldService) HandleGetPartURL(w http.ResponseWriter, r *http.Request) {
7777+ // Validate DID authorization for WRITE
7878+ // Build S3 key from digest
7979+ // Call s3.UploadPartRequest() with part number and upload ID
8080+ // Generate presigned URL
8181+ // Return presigned URL for this specific part
8282+ }
8383+ POST /complete-multipart
8484+ func (s *HoldService) HandleCompleteMultipart(w http.ResponseWriter, r *http.Request) {
8585+ // Validate DID authorization for WRITE
8686+ // Build S3 key from digest
8787+ // Prepare CompletedPart array with part numbers and ETags
8888+ // Call s3.CompleteMultipartUpload()
8989+ // Return success
9090+ }
9191+ POST /abort-multipart (for cleanup)
9292+ func (s *HoldService) HandleAbortMultipart(w http.ResponseWriter, r *http.Request) {
9393+ // Validate DID authorization for WRITE
9494+ // Call s3.AbortMultipartUpload()
9595+ // Return success
9696+ }
9797+ S3 Implementation
9898+ // startMultipartUpload initiates a multipart upload and returns upload ID
9999+ func (s *HoldService) startMultipartUpload(ctx context.Context, digest string) (string, error) {
100100+ if s.s3Client == nil {
101101+ return "", fmt.Errorf("S3 not configured")
102102+ }
103103+ path := blobPath(digest)
104104+ s3Key := strings.TrimPrefix(path, "/")
105105+ if s.s3PathPrefix != "" {
106106+ s3Key = s.s3PathPrefix + "/" + s3Key
107107+ }
108108+ result, err := s.s3Client.CreateMultipartUploadWithContext(ctx, &s3.CreateMultipartUploadInput{
109109+ Bucket: aws.String(s.bucket),
110110+ Key: aws.String(s3Key),
111111+ })
112112+ if err != nil {
113113+ return "", err
114114+ }
115115+ return *result.UploadId, nil
116116+ }
117117+ // getPartPresignedURL generates presigned URL for a specific part
118118+ func (s *HoldService) getPartPresignedURL(ctx context.Context, digest, uploadID string, partNumber int) (string, error) {
119119+ if s.s3Client == nil {
120120+ return "", fmt.Errorf("S3 not configured")
121121+ }
122122+ path := blobPath(digest)
123123+ s3Key := strings.TrimPrefix(path, "/")
124124+ if s.s3PathPrefix != "" {
125125+ s3Key = s.s3PathPrefix + "/" + s3Key
126126+ }
127127+ req, _ := s.s3Client.UploadPartRequest(&s3.UploadPartInput{
128128+ Bucket: aws.String(s.bucket),
129129+ Key: aws.String(s3Key),
130130+ UploadId: aws.String(uploadID),
131131+ PartNumber: aws.Int64(int64(partNumber)),
132132+ })
133133+ return req.Presign(15 * time.Minute)
134134+ }
135135+ // completeMultipartUpload finalizes the multipart upload
136136+ func (s *HoldService) completeMultipartUpload(ctx context.Context, digest, uploadID string, parts []CompletedPart) error {
137137+ if s.s3Client == nil {
138138+ return fmt.Errorf("S3 not configured")
139139+ }
140140+ path := blobPath(digest)
141141+ s3Key := strings.TrimPrefix(path, "/")
142142+ if s.s3PathPrefix != "" {
143143+ s3Key = s.s3PathPrefix + "/" + s3Key
144144+ }
145145+ // Convert to S3 CompletedPart format
146146+ s3Parts := make([]*s3.CompletedPart, len(parts))
147147+ for i, p := range parts {
148148+ s3Parts[i] = &s3.CompletedPart{
149149+ PartNumber: aws.Int64(int64(p.PartNumber)),
150150+ ETag: aws.String(p.ETag),
151151+ }
152152+ }
153153+ _, err := s.s3Client.CompleteMultipartUploadWithContext(ctx, &s3.CompleteMultipartUploadInput{
154154+ Bucket: aws.String(s.bucket),
155155+ Key: aws.String(s3Key),
156156+ UploadId: aws.String(uploadID),
157157+ MultipartUpload: &s3.CompletedMultipartUpload{
158158+ Parts: s3Parts,
159159+ },
160160+ })
161161+ return err
162162+ }
163163+ ---
164164+ 2. AppView: Rewrite ProxyBlobStore for Multipart
165165+ File: pkg/storage/proxy_blob_store.go
166166+ Remove Current Implementation
167167+ - Remove pipe-based streaming
168168+ - Remove background goroutine with single presigned URL
169169+ - Remove global upload tracking map
170170+ New ProxyBlobWriter Structure
171171+ type ProxyBlobWriter struct {
172172+ store *ProxyBlobStore
173173+ options distribution.CreateOptions
174174+ uploadID string // S3 multipart upload ID
175175+ parts []CompletedPart // Track uploaded parts with ETags
176176+ partNumber int // Current part number (starts at 1)
177177+ buffer *bytes.Buffer // Buffer for current part
178178+ size int64 // Total bytes written
179179+ closed bool
180180+ id string // Distribution's upload ID (for state)
181181+ startedAt time.Time
182182+ finalDigest string // Set on Commit
183183+ }
184184+ type CompletedPart struct {
185185+ PartNumber int
186186+ ETag string
187187+ }
188188+ New Create() - Initiate Multipart Upload
189189+ func (p *ProxyBlobStore) Create(ctx context.Context, options ...distribution.BlobCreateOption) (distribution.BlobWriter, error) {
190190+ var opts distribution.CreateOptions
191191+ for _, option := range options {
192192+ if err := option.Apply(&opts); err != nil {
193193+ return nil, err
194194+ }
195195+ }
196196+ // Use temp digest for upload location
197197+ writerID := fmt.Sprintf("upload-%d", time.Now().UnixNano())
198198+ tempDigest := digest.Digest(fmt.Sprintf("uploads/temp-%s", writerID))
199199+ // Start multipart upload via hold service
200200+ uploadID, err := p.startMultipartUpload(ctx, tempDigest)
201201+ if err != nil {
202202+ return nil, fmt.Errorf("failed to start multipart upload: %w", err)
203203+ }
204204+ writer := &ProxyBlobWriter{
205205+ store: p,
206206+ options: opts,
207207+ uploadID: uploadID,
208208+ parts: make([]CompletedPart, 0),
209209+ partNumber: 1,
210210+ buffer: bytes.NewBuffer(make([]byte, 0, 5*1024*1024)), // 5MB buffer
211211+ id: writerID,
212212+ startedAt: time.Now(),
213213+ }
214214+ // Store in global map for Resume()
215215+ globalUploadsMu.Lock()
216216+ globalUploads[writer.id] = writer
217217+ globalUploadsMu.Unlock()
218218+ return writer, nil
219219+ }
220220+ New Write() - Buffer and Flush Parts
221221+ func (w *ProxyBlobWriter) Write(p []byte) (int, error) {
222222+ if w.closed {
223223+ return 0, fmt.Errorf("writer closed")
224224+ }
225225+ n, err := w.buffer.Write(p)
226226+ w.size += int64(n)
227227+ // Flush if buffer reaches 5MB (S3 minimum part size)
228228+ if w.buffer.Len() >= 5*1024*1024 {
229229+ if err := w.flushPart(); err != nil {
230230+ return n, err
231231+ }
232232+ }
233233+ return n, err
234234+ }
235235+ func (w *ProxyBlobWriter) flushPart() error {
236236+ if w.buffer.Len() == 0 {
237237+ return nil
238238+ }
239239+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
240240+ defer cancel()
241241+ // Get presigned URL for this part
242242+ tempDigest := digest.Digest(fmt.Sprintf("uploads/temp-%s", w.id))
243243+ url, err := w.store.getPartPresignedURL(ctx, tempDigest, w.uploadID, w.partNumber)
244244+ if err != nil {
245245+ return fmt.Errorf("failed to get part presigned URL: %w", err)
246246+ }
247247+ // Upload part to S3
248248+ req, err := http.NewRequestWithContext(ctx, "PUT", url, bytes.NewReader(w.buffer.Bytes()))
249249+ if err != nil {
250250+ return err
251251+ }
252252+ resp, err := w.store.httpClient.Do(req)
253253+ if err != nil {
254254+ return err
255255+ }
256256+ defer resp.Body.Close()
257257+ if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusCreated {
258258+ return fmt.Errorf("part upload failed: status %d", resp.StatusCode)
259259+ }
260260+ // Store ETag for completion
261261+ etag := resp.Header.Get("ETag")
262262+ if etag == "" {
263263+ return fmt.Errorf("no ETag in response")
264264+ }
265265+ w.parts = append(w.parts, CompletedPart{
266266+ PartNumber: w.partNumber,
267267+ ETag: etag,
268268+ })
269269+ // Reset buffer and increment part number
270270+ w.buffer.Reset()
271271+ w.partNumber++
272272+ return nil
273273+ }
274274+ New Commit() - Complete Multipart and Move
275275+ func (w *ProxyBlobWriter) Commit(ctx context.Context, desc distribution.Descriptor) (distribution.Descriptor, error) {
276276+ if w.closed {
277277+ return distribution.Descriptor{}, fmt.Errorf("writer closed")
278278+ }
279279+ w.closed = true
280280+ // Flush any remaining buffered data
281281+ if w.buffer.Len() > 0 {
282282+ if err := w.flushPart(); err != nil {
283283+ // Try to abort multipart on error
284284+ w.store.abortMultipartUpload(ctx, w.uploadID)
285285+ return distribution.Descriptor{}, err
286286+ }
287287+ }
288288+ // Complete multipart upload at temp location
289289+ tempDigest := digest.Digest(fmt.Sprintf("uploads/temp-%s", w.id))
290290+ if err := w.store.completeMultipartUpload(ctx, tempDigest, w.uploadID, w.parts); err != nil {
291291+ return distribution.Descriptor{}, err
292292+ }
293293+ // Move from temp → final location (server-side S3 copy)
294294+ tempPath := fmt.Sprintf("uploads/temp-%s", w.id)
295295+ finalPath := desc.Digest.String()
296296+ moveURL := fmt.Sprintf("%s/move?from=%s&to=%s&did=%s",
297297+ w.store.storageEndpoint, tempPath, finalPath, w.store.did)
298298+ req, err := http.NewRequestWithContext(ctx, "POST", moveURL, nil)
299299+ if err != nil {
300300+ return distribution.Descriptor{}, err
301301+ }
302302+ resp, err := w.store.httpClient.Do(req)
303303+ if err != nil {
304304+ return distribution.Descriptor{}, err
305305+ }
306306+ defer resp.Body.Close()
307307+ if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusCreated {
308308+ bodyBytes, _ := io.ReadAll(resp.Body)
309309+ return distribution.Descriptor{}, fmt.Errorf("move failed: %d, %s", resp.StatusCode, bodyBytes)
310310+ }
311311+ // Remove from global map
312312+ globalUploadsMu.Lock()
313313+ delete(globalUploads, w.id)
314314+ globalUploadsMu.Unlock()
315315+ return distribution.Descriptor{
316316+ Digest: desc.Digest,
317317+ Size: w.size,
318318+ MediaType: desc.MediaType,
319319+ }, nil
320320+ }
321321+ Add Hold Service Client Methods
322322+ func (p *ProxyBlobStore) startMultipartUpload(ctx context.Context, dgst digest.Digest) (string, error) {
323323+ reqBody := map[string]any{
324324+ "did": p.did,
325325+ "digest": dgst.String(),
326326+ }
327327+ body, _ := json.Marshal(reqBody)
328328+ url := fmt.Sprintf("%s/start-multipart", p.storageEndpoint)
329329+ req, _ := http.NewRequestWithContext(ctx, "POST", url, bytes.NewReader(body))
330330+ req.Header.Set("Content-Type", "application/json")
331331+ resp, err := p.httpClient.Do(req)
332332+ if err != nil {
333333+ return "", err
334334+ }
335335+ defer resp.Body.Close()
336336+ var result struct {
337337+ UploadID string `json:"upload_id"`
338338+ }
339339+ if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
340340+ return "", err
341341+ }
342342+ return result.UploadID, nil
343343+ }
344344+ func (p *ProxyBlobStore) getPartPresignedURL(ctx context.Context, dgst digest.Digest, uploadID string, partNumber int) (string, error) {
345345+ reqBody := map[string]any{
346346+ "did": p.did,
347347+ "digest": dgst.String(),
348348+ "upload_id": uploadID,
349349+ "part_number": partNumber,
350350+ }
351351+ body, _ := json.Marshal(reqBody)
352352+ url := fmt.Sprintf("%s/part-presigned-url", p.storageEndpoint)
353353+ req, _ := http.NewRequestWithContext(ctx, "POST", url, bytes.NewReader(body))
354354+ req.Header.Set("Content-Type", "application/json")
355355+ resp, err := p.httpClient.Do(req)
356356+ if err != nil {
357357+ return "", err
358358+ }
359359+ defer resp.Body.Close()
360360+ var result struct {
361361+ URL string `json:"url"`
362362+ }
363363+ if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
364364+ return "", err
365365+ }
366366+ return result.URL, nil
367367+ }
368368+ func (p *ProxyBlobStore) completeMultipartUpload(ctx context.Context, dgst digest.Digest, uploadID string, parts []CompletedPart) error {
369369+ reqBody := map[string]any{
370370+ "did": p.did,
371371+ "digest": dgst.String(),
372372+ "upload_id": uploadID,
373373+ "parts": parts,
374374+ }
375375+ body, _ := json.Marshal(reqBody)
376376+ url := fmt.Sprintf("%s/complete-multipart", p.storageEndpoint)
377377+ req, _ := http.NewRequestWithContext(ctx, "POST", url, bytes.NewReader(body))
378378+ req.Header.Set("Content-Type", "application/json")
379379+ resp, err := p.httpClient.Do(req)
380380+ if err != nil {
381381+ return err
382382+ }
383383+ defer resp.Body.Close()
384384+ if resp.StatusCode != http.StatusOK {
385385+ return fmt.Errorf("complete multipart failed: status %d", resp.StatusCode)
386386+ }
387387+ return nil
388388+ }
389389+ ---
390390+ Testing Plan
391391+ 1. Unit Tests
392392+ - Test multipart upload initiation
393393+ - Test part upload with presigned URLs
394394+ - Test completion with ETags
395395+ - Test abort on errors
396396+ 2. Integration Tests
397397+ - Push small images (< 5MB, single part)
398398+ - Push medium images (10MB, 2 parts)
399399+ - Push large images (100MB, 20 parts)
400400+ - Test with Upcloud S3
401401+ - Test with Storj S3
402402+ 3. Validation
403403+ - Monitor logs for "client disconnected" errors (should be gone)
404404+ - Check Docker push success rate
405405+ - Verify blobs stored correctly in S3
406406+ - Check bandwidth usage on hold service (should be minimal)
407407+ ---
408408+ Migration & Deployment
409409+ Backward Compatibility
410410+ - Keep /put-presigned-url endpoint for fallback
411411+ - Keep /move endpoint (still needed)
412412+ - New multipart endpoints are additive
413413+ Deployment Steps
414414+ 1. Update hold service with new endpoints
415415+ 2. Update AppView ProxyBlobStore
416416+ 3. Deploy hold service first
417417+ 4. Deploy AppView
418418+ 5. Test with sample push
419419+ 6. Monitor logs
420420+ Rollback Plan
421421+ - Revert AppView to previous version (uses old presigned URL method)
422422+ - Hold service keeps both old and new endpoints
423423+ ---
424424+ Documentation Updates
425425+ Update docs/PRESIGNED_URLS.md
426426+ - Add section "Multipart Upload for Chunked Data"
427427+ - Explain why single presigned URLs don't work with PATCH
428428+ - Document new endpoints and flow
429429+ - Add S3 part size recommendations (5MB-64MB for Storj)
430430+ Add Troubleshooting Section
431431+ - "Client disconnected during PATCH" → resolved by multipart
432432+ - Storj-specific considerations (64MB parts recommended)
433433+ - Upcloud compatibility notes
434434+ ---
435435+ Performance Impact
436436+ Before (Broken)
437437+ - Docker PATCH → blocks on pipe → timeout → retry → fail
438438+ - Unable to push large images reliably
439439+ After (Multipart)
440440+ - Each PATCH → independent part upload → immediate response
441441+ - No blocking, no timeouts
442442+ - Parallel part uploads possible (future optimization)
443443+ - Reliable pushes for any image size
444444+ Bandwidth
445445+ - Hold service: Only API calls (~1KB per part)
446446+ - Direct S3 uploads: Full blob data
447447+ - S3 copy for move: Server-side (no hold bandwidth)
448448+ Estimated savings: 99.98% hold service bandwidth reduction (same as before, but now actually works!)
+1017
docs/PRESIGNED_UPLOADS.md
···11+# Presigned Upload URLs Implementation Guide
22+33+## Current Architecture (Proxy Mode)
44+55+### Upload Flow Today
66+1. **AppView** receives blob upload request from Docker
77+2. **ProxyBlobStore.Create()** creates streaming upload via pipe
88+3. Data streams to **Hold Service** temp location: `uploads/temp-{id}`
99+4. Hold service uploads to S3 via storage driver
1010+5. **ProxyBlobWriter.Commit()** moves blob: temp → final digest-based path
1111+6. Hold service performs S3 Move operation
1212+1313+### Why Uploads Don't Use Presigned URLs Today
1414+- `Create()` doesn't know the blob digest upfront
1515+- Presigned S3 URLs require the full object key (which includes digest)
1616+- Current approach streams to temp location, calculates digest, then moves
1717+1818+### Bandwidth Flow (Current)
1919+```
2020+Docker → AppView → Hold Service → S3/Storj
2121+ (proxy) (proxy)
2222+```
2323+2424+All upload bandwidth flows through Hold Service.
2525+2626+---
2727+2828+## Proposed Architecture (Presigned Uploads)
2929+3030+### New Upload Flow
3131+1. **AppView** receives blob upload request from Docker
3232+2. **ProxyBlobStore.Create()** creates buffered upload writer
3333+3. Data buffered in memory during `Write()` calls
3434+4. **ProxyBlobWriter.Commit()** calculates digest from buffer
3535+5. Request presigned PUT URL from Hold Service with digest
3636+6. Upload buffered data directly to S3 via presigned URL
3737+7. No move operation needed (uploaded to final path)
3838+3939+### Bandwidth Flow (Presigned)
4040+```
4141+Docker → AppView → S3/Storj (direct via presigned URL)
4242+ (buffer)
4343+4444+Hold Service only issues presigned URLs (minimal bandwidth)
4545+```
4646+4747+---
4848+4949+## Detailed Implementation
5050+5151+### Phase 1: Add Buffering to ProxyBlobWriter
5252+5353+**File:** `pkg/storage/proxy_blob_store.go`
5454+5555+#### Changes to ProxyBlobWriter struct
5656+5757+```go
5858+type ProxyBlobWriter struct {
5959+ store *ProxyBlobStore
6060+ options distribution.CreateOptions
6161+6262+ // Remove pipe-based streaming
6363+ // pipeWriter *io.PipeWriter
6464+ // pipeReader *io.PipeReader
6565+ // digestChan chan string
6666+ // uploadErr chan error
6767+6868+ // Add buffering
6969+ buffer *bytes.Buffer // In-memory buffer for blob data
7070+ hasher digest.Digester // Calculate digest while writing
7171+7272+ finalDigest string
7373+ size int64
7474+ closed bool
7575+ id string
7676+ startedAt time.Time
7777+}
7878+```
7979+8080+**Rationale:**
8181+- Remove pipe mechanism (no longer streaming to temp)
8282+- Add buffer to store blob data in memory
8383+- Add hasher to calculate digest incrementally
8484+8585+#### Modify Create() method
8686+8787+**Before (lines 208-312):**
8888+```go
8989+func (p *ProxyBlobStore) Create(ctx context.Context, options ...distribution.BlobCreateOption) (distribution.BlobWriter, error) {
9090+ // Creates pipe and starts background goroutine for streaming
9191+ pipeReader, pipeWriter := io.Pipe()
9292+ // ... streams to temp location
9393+}
9494+```
9595+9696+**After:**
9797+```go
9898+func (p *ProxyBlobStore) Create(ctx context.Context, options ...distribution.BlobCreateOption) (distribution.BlobWriter, error) {
9999+ fmt.Printf("🔧 [proxy_blob_store/Create] Starting buffered upload for presigned URL\n")
100100+101101+ // Parse options
102102+ var opts distribution.CreateOptions
103103+ for _, option := range options {
104104+ if err := option.Apply(&opts); err != nil {
105105+ return nil, err
106106+ }
107107+ }
108108+109109+ // Create buffered writer
110110+ writer := &ProxyBlobWriter{
111111+ store: p,
112112+ options: opts,
113113+ buffer: new(bytes.Buffer),
114114+ hasher: digest.Canonical.Digester(), // Usually SHA256
115115+ id: fmt.Sprintf("upload-%d", time.Now().UnixNano()),
116116+ startedAt: time.Now(),
117117+ }
118118+119119+ // Store in global uploads map for resume support
120120+ globalUploadsMu.Lock()
121121+ globalUploads[writer.id] = writer
122122+ globalUploadsMu.Unlock()
123123+124124+ fmt.Printf(" Upload ID: %s\n", writer.id)
125125+ fmt.Printf(" Repository: %s\n", p.repository)
126126+127127+ return writer, nil
128128+}
129129+```
130130+131131+**Key Changes:**
132132+- No more pipe creation
133133+- No background goroutine
134134+- Initialize buffer and hasher
135135+- Everything else stays synchronous
136136+137137+#### Modify Write() method
138138+139139+**Before (lines 440-455):**
140140+```go
141141+func (w *ProxyBlobWriter) Write(p []byte) (int, error) {
142142+ // Writes to pipe, streams to hold service
143143+ n, err := w.pipeWriter.Write(p)
144144+ w.size += int64(n)
145145+ return n, nil
146146+}
147147+```
148148+149149+**After:**
150150+```go
151151+func (w *ProxyBlobWriter) Write(p []byte) (int, error) {
152152+ if w.closed {
153153+ return 0, fmt.Errorf("writer closed")
154154+ }
155155+156156+ // Write to buffer
157157+ n, err := w.buffer.Write(p)
158158+ if err != nil {
159159+ return n, fmt.Errorf("failed to buffer data: %w", err)
160160+ }
161161+162162+ // Update hasher for digest calculation
163163+ w.hasher.Hash().Write(p)
164164+165165+ w.size += int64(n)
166166+167167+ // Memory pressure check (optional safety)
168168+ if w.buffer.Len() > 500*1024*1024 { // 500MB limit
169169+ return n, fmt.Errorf("blob too large for buffered upload: %d bytes", w.buffer.Len())
170170+ }
171171+172172+ return n, nil
173173+}
174174+```
175175+176176+**Key Changes:**
177177+- Write to in-memory buffer instead of pipe
178178+- Update hasher incrementally (efficient)
179179+- Add safety check for excessive memory usage
180180+- No streaming to hold service yet
181181+182182+#### Modify Commit() method
183183+184184+**Before (lines 493-548):**
185185+```go
186186+func (w *ProxyBlobWriter) Commit(ctx context.Context, desc distribution.Descriptor) (distribution.Descriptor, error) {
187187+ // Close pipe, send digest to goroutine
188188+ // Wait for temp upload
189189+ // Move temp → final
190190+}
191191+```
192192+193193+**After:**
194194+```go
195195+func (w *ProxyBlobWriter) Commit(ctx context.Context, desc distribution.Descriptor) (distribution.Descriptor, error) {
196196+ if w.closed {
197197+ return distribution.Descriptor{}, fmt.Errorf("writer closed")
198198+ }
199199+ w.closed = true
200200+201201+ // Remove from global uploads map
202202+ globalUploadsMu.Lock()
203203+ delete(globalUploads, w.id)
204204+ globalUploadsMu.Unlock()
205205+206206+ // Calculate digest from buffered data
207207+ calculatedDigest := w.hasher.Digest()
208208+209209+ // Verify digest matches if provided
210210+ if desc.Digest != "" && desc.Digest != calculatedDigest {
211211+ return distribution.Descriptor{}, fmt.Errorf(
212212+ "digest mismatch: expected %s, got %s",
213213+ desc.Digest, calculatedDigest,
214214+ )
215215+ }
216216+217217+ finalDigest := calculatedDigest
218218+ if desc.Digest != "" {
219219+ finalDigest = desc.Digest
220220+ }
221221+222222+ fmt.Printf("📤 [ProxyBlobWriter.Commit] Uploading via presigned URL\n")
223223+ fmt.Printf(" Digest: %s\n", finalDigest)
224224+ fmt.Printf(" Size: %d bytes\n", w.size)
225225+ fmt.Printf(" Buffered: %d bytes\n", w.buffer.Len())
226226+227227+ // Get presigned upload URL from hold service
228228+ url, err := w.store.getUploadURL(ctx, finalDigest, w.size)
229229+ if err != nil {
230230+ return distribution.Descriptor{}, fmt.Errorf("failed to get presigned upload URL: %w", err)
231231+ }
232232+233233+ fmt.Printf(" Presigned URL: %s\n", url)
234234+235235+ // Upload directly to S3 via presigned URL
236236+ req, err := http.NewRequestWithContext(ctx, "PUT", url, bytes.NewReader(w.buffer.Bytes()))
237237+ if err != nil {
238238+ return distribution.Descriptor{}, fmt.Errorf("failed to create upload request: %w", err)
239239+ }
240240+ req.Header.Set("Content-Type", "application/octet-stream")
241241+ req.ContentLength = w.size
242242+243243+ resp, err := w.store.httpClient.Do(req)
244244+ if err != nil {
245245+ return distribution.Descriptor{}, fmt.Errorf("presigned upload failed: %w", err)
246246+ }
247247+ defer resp.Body.Close()
248248+249249+ if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusCreated {
250250+ bodyBytes, _ := io.ReadAll(resp.Body)
251251+ return distribution.Descriptor{}, fmt.Errorf(
252252+ "presigned upload failed: status %d, body: %s",
253253+ resp.StatusCode, string(bodyBytes),
254254+ )
255255+ }
256256+257257+ fmt.Printf("✅ [ProxyBlobWriter.Commit] Upload successful\n")
258258+259259+ // Clear buffer to free memory
260260+ w.buffer = nil
261261+262262+ return distribution.Descriptor{
263263+ Digest: finalDigest,
264264+ Size: w.size,
265265+ MediaType: desc.MediaType,
266266+ }, nil
267267+}
268268+```
269269+270270+**Key Changes:**
271271+- Calculate digest from hasher (already computed incrementally)
272272+- Verify digest if provided by client
273273+- Get presigned upload URL with final digest
274274+- Upload buffer contents directly to S3
275275+- No temp location, no move operation
276276+- Clear buffer to free memory immediately
277277+278278+#### Modify Cancel() method
279279+280280+**Before (lines 551-572):**
281281+```go
282282+func (w *ProxyBlobWriter) Cancel(ctx context.Context) error {
283283+ // Close pipe, cancel temp upload
284284+}
285285+```
286286+287287+**After:**
288288+```go
289289+func (w *ProxyBlobWriter) Cancel(ctx context.Context) error {
290290+ w.closed = true
291291+292292+ // Remove from global uploads map
293293+ globalUploadsMu.Lock()
294294+ delete(globalUploads, w.id)
295295+ globalUploadsMu.Unlock()
296296+297297+ // Clear buffer to free memory
298298+ w.buffer = nil
299299+300300+ fmt.Printf("[ProxyBlobWriter.Cancel] Upload cancelled: id=%s\n", w.id)
301301+ return nil
302302+}
303303+```
304304+305305+**Key Changes:**
306306+- Simply clear buffer
307307+- No pipe cleanup needed
308308+- No temp cleanup needed (nothing uploaded yet)
309309+310310+---
311311+312312+### Phase 2: Update Hold Service (Optional Enhancement)
313313+314314+The current `getUploadURL()` implementation in `cmd/hold/main.go` (lines 528-587) already supports presigned uploads correctly. No changes needed unless you want to add additional logging.
315315+316316+**Optional logging enhancement at line 547:**
317317+318318+```go
319319+url, err := req.Presign(15 * time.Minute)
320320+if err != nil {
321321+ log.Printf("Failed to generate presigned upload URL: %v", err)
322322+ return s.getProxyUploadURL(digest, did), nil
323323+}
324324+325325+log.Printf("🔑 Generated presigned upload URL:")
326326+log.Printf(" Digest: %s", digest)
327327+log.Printf(" S3 Key: %s", s3Key)
328328+log.Printf(" Size: %d bytes", size)
329329+log.Printf(" URL length: %d chars", len(url))
330330+log.Printf(" Expires: 15min")
331331+332332+return url, nil
333333+```
334334+335335+---
336336+337337+### Phase 3: Memory Management Considerations
338338+339339+#### Add Configuration for Max Buffer Size
340340+341341+**File:** `pkg/storage/proxy_blob_store.go`
342342+343343+Add constants at top of file:
344344+345345+```go
346346+const (
347347+ maxChunkSize = 5 * 1024 * 1024 // 5MB (existing)
348348+349349+ // Maximum blob size for in-memory buffering
350350+ // Blobs larger than this will fail (alternative: fallback to proxy mode)
351351+ maxBufferedBlobSize = 500 * 1024 * 1024 // 500MB
352352+)
353353+```
354354+355355+#### Alternative: Disk-Based Buffering
356356+357357+For very large blobs, consider disk-based buffering:
358358+359359+```go
360360+type ProxyBlobWriter struct {
361361+ // ... existing fields ...
362362+363363+ // Choose one:
364364+ buffer *bytes.Buffer // Memory buffer (current)
365365+ // OR
366366+ tempFile *os.File // Disk buffer (for large blobs)
367367+ bufferSize int64
368368+}
369369+```
370370+371371+**Memory buffer (simple, fast):**
372372+- Pro: Fast, no disk I/O
373373+- Con: Limited by available RAM
374374+- Use for: Blobs < 500MB
375375+376376+**Disk buffer (scalable):**
377377+- Pro: No memory limit
378378+- Con: Slower, disk I/O overhead
379379+- Use for: Blobs > 500MB
380380+381381+#### Hybrid Approach (Recommended)
382382+383383+```go
384384+const (
385385+ memoryBufferThreshold = 50 * 1024 * 1024 // 50MB
386386+)
387387+388388+func (w *ProxyBlobWriter) Write(p []byte) (int, error) {
389389+ // If buffer exceeds threshold, switch to disk
390390+ if w.buffer != nil && w.buffer.Len() > memoryBufferThreshold {
391391+ return 0, fmt.Errorf("blob exceeds memory buffer threshold, disk buffering not implemented")
392392+ // TODO: Implement disk buffering or fallback to proxy mode
393393+ }
394394+395395+ // Otherwise use memory buffer
396396+ // ... existing Write() logic ...
397397+}
398398+```
399399+400400+---
401401+402402+## Optional Enhancement: Presigned HEAD URLs
403403+404404+### Motivation
405405+406406+Currently HEAD requests (blob verification) are proxied through the Hold Service. This is fine because HEAD bandwidth is negligible (~300 bytes per request), but we can eliminate this round-trip by using presigned HEAD URLs.
407407+408408+### Implementation
409409+410410+#### Step 1: Add getHeadURL() to Hold Service
411411+412412+**File:** `cmd/hold/main.go`
413413+414414+Add new function after `getDownloadURL()`:
415415+416416+```go
417417+// getHeadURL generates a presigned HEAD URL for blob verification
418418+func (s *HoldService) getHeadURL(ctx context.Context, digest string) (string, error) {
419419+ // Check if blob exists first
420420+ path := blobPath(digest)
421421+ _, err := s.driver.Stat(ctx, path)
422422+ if err != nil {
423423+ return "", fmt.Errorf("blob not found: %w", err)
424424+ }
425425+426426+ // If S3 client available, generate presigned HEAD URL
427427+ if s.s3Client != nil {
428428+ s3Key := strings.TrimPrefix(path, "/")
429429+ if s.s3PathPrefix != "" {
430430+ s3Key = s.s3PathPrefix + "/" + s3Key
431431+ }
432432+433433+ // Generate presigned HEAD URL (method-specific!)
434434+ req, _ := s.s3Client.HeadObjectRequest(&s3.HeadObjectInput{
435435+ Bucket: aws.String(s.bucket),
436436+ Key: aws.String(s3Key),
437437+ })
438438+439439+ log.Printf("🔍 [getHeadURL] Generating presigned HEAD URL:")
440440+ log.Printf(" Digest: %s", digest)
441441+ log.Printf(" S3 Key: %s", s3Key)
442442+443443+ url, err := req.Presign(15 * time.Minute)
444444+ if err != nil {
445445+ log.Printf("[getHeadURL] Presign failed: %v", err)
446446+ // Fallback to proxy URL
447447+ return s.getProxyHeadURL(digest), nil
448448+ }
449449+450450+ log.Printf("✅ [getHeadURL] Presigned HEAD URL generated")
451451+ return url, nil
452452+ }
453453+454454+ // Fallback: return proxy URL
455455+ return s.getProxyHeadURL(digest), nil
456456+}
457457+458458+// getProxyHeadURL returns a proxy URL for HEAD requests
459459+func (s *HoldService) getProxyHeadURL(digest string) string {
460460+ // HEAD requests don't need DID in query string (read-only check)
461461+ return fmt.Sprintf("%s/blobs/%s", s.config.Server.PublicURL, digest)
462462+}
463463+```
464464+465465+#### Step 2: Add HTTP endpoint for presigned HEAD URLs
466466+467467+**File:** `cmd/hold/main.go`
468468+469469+Add handler similar to `HandleGetPresignedURL()`:
470470+471471+```go
472472+// HeadPresignedURLRequest represents a request for a presigned HEAD URL
473473+type HeadPresignedURLRequest struct {
474474+ DID string `json:"did"`
475475+ Digest string `json:"digest"`
476476+}
477477+478478+// HeadPresignedURLResponse contains the presigned HEAD URL
479479+type HeadPresignedURLResponse struct {
480480+ URL string `json:"url"`
481481+ ExpiresAt time.Time `json:"expires_at"`
482482+}
483483+484484+// HandleHeadPresignedURL handles requests for HEAD URLs
485485+func (s *HoldService) HandleHeadPresignedURL(w http.ResponseWriter, r *http.Request) {
486486+ if r.Method != http.MethodPost {
487487+ http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
488488+ return
489489+ }
490490+491491+ var req HeadPresignedURLRequest
492492+ if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
493493+ http.Error(w, fmt.Sprintf("invalid request: %v", err), http.StatusBadRequest)
494494+ return
495495+ }
496496+497497+ // Validate DID authorization for READ
498498+ if !s.isAuthorizedRead(req.DID) {
499499+ if req.DID == "" {
500500+ http.Error(w, "unauthorized: authentication required", http.StatusUnauthorized)
501501+ } else {
502502+ http.Error(w, "forbidden: access denied", http.StatusForbidden)
503503+ }
504504+ return
505505+ }
506506+507507+ // Generate presigned HEAD URL
508508+ ctx := context.Background()
509509+ expiry := time.Now().Add(15 * time.Minute)
510510+511511+ url, err := s.getHeadURL(ctx, req.Digest)
512512+ if err != nil {
513513+ http.Error(w, fmt.Sprintf("failed to generate URL: %v", err), http.StatusInternalServerError)
514514+ return
515515+ }
516516+517517+ resp := HeadPresignedURLResponse{
518518+ URL: url,
519519+ ExpiresAt: expiry,
520520+ }
521521+522522+ w.Header().Set("Content-Type", "application/json")
523523+ json.NewEncoder(w).Encode(resp)
524524+}
525525+```
526526+527527+#### Step 3: Register endpoint in main()
528528+529529+**File:** `cmd/hold/main.go`
530530+531531+In `main()` function, add route:
532532+533533+```go
534534+mux.HandleFunc("/head-presigned-url", service.HandleHeadPresignedURL)
535535+```
536536+537537+#### Step 4: Update ProxyBlobStore.ServeBlob()
538538+539539+**File:** `pkg/storage/proxy_blob_store.go`
540540+541541+Modify HEAD handling (currently lines 197-224):
542542+543543+**Before:**
544544+```go
545545+if r.Method == http.MethodHead {
546546+ // Check if blob exists via hold service HEAD request
547547+ url := fmt.Sprintf("%s/blobs/%s?did=%s", p.storageEndpoint, dgst.String(), p.did)
548548+ req, err := http.NewRequestWithContext(ctx, "HEAD", url, nil)
549549+ // ... proxy through hold service ...
550550+}
551551+```
552552+553553+**After:**
554554+```go
555555+if r.Method == http.MethodHead {
556556+ // Get presigned HEAD URL from hold service
557557+ headURL, err := p.getHeadURL(ctx, dgst)
558558+ if err != nil {
559559+ return distribution.ErrBlobUnknown
560560+ }
561561+562562+ // Redirect to presigned HEAD URL
563563+ http.Redirect(w, r, headURL, http.StatusTemporaryRedirect)
564564+ return nil
565565+}
566566+```
567567+568568+#### Step 5: Add getHeadURL() to ProxyBlobStore
569569+570570+**File:** `pkg/storage/proxy_blob_store.go`
571571+572572+Add after `getDownloadURL()`:
573573+574574+```go
575575+// getHeadURL requests a presigned HEAD URL from the storage service
576576+func (p *ProxyBlobStore) getHeadURL(ctx context.Context, dgst digest.Digest) (string, error) {
577577+ reqBody := map[string]any{
578578+ "did": p.did,
579579+ "digest": dgst.String(),
580580+ }
581581+582582+ body, err := json.Marshal(reqBody)
583583+ if err != nil {
584584+ return "", err
585585+ }
586586+587587+ url := fmt.Sprintf("%s/head-presigned-url", p.storageEndpoint)
588588+ req, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewReader(body))
589589+ if err != nil {
590590+ return "", err
591591+ }
592592+ req.Header.Set("Content-Type", "application/json")
593593+594594+ resp, err := p.httpClient.Do(req)
595595+ if err != nil {
596596+ return "", err
597597+ }
598598+ defer resp.Body.Close()
599599+600600+ if resp.StatusCode != http.StatusOK {
601601+ return "", fmt.Errorf("failed to get HEAD URL: status %d", resp.StatusCode)
602602+ }
603603+604604+ var result struct {
605605+ URL string `json:"url"`
606606+ }
607607+ if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
608608+ return "", err
609609+ }
610610+611611+ return result.URL, nil
612612+}
613613+```
614614+615615+### Presigned HEAD URLs: Trade-offs
616616+617617+**Benefits:**
618618+- Offloads HEAD requests from Hold Service
619619+- Docker verifies blobs directly against S3
620620+- Slightly lower latency (one fewer hop)
621621+622622+**Costs:**
623623+- Requires round-trip to get presigned HEAD URL
624624+- More complex code
625625+- Two HTTP requests instead of one proxy request
626626+627627+**Bandwidth Analysis:**
628628+- Current: 1 HEAD request to Hold Service (~300 bytes)
629629+- Presigned: 1 POST to get URL (~200 bytes) + 1 HEAD to S3 (~300 bytes)
630630+- **Net difference: Adds ~200 bytes per verification**
631631+632632+**Recommendation:** Optional enhancement. The current proxied HEAD approach is simpler and bandwidth difference is negligible. Only implement if:
633633+- Hold Service is becoming a bottleneck
634634+- You want to minimize Hold Service load completely
635635+- Latency of HEAD requests becomes noticeable
636636+637637+---
638638+639639+## Testing & Validation
640640+641641+### Test Plan for Presigned Uploads
642642+643643+#### 1. Small Blob Upload (< 1MB)
644644+```bash
645645+# Build test image with small layers
646646+echo "FROM scratch" > Dockerfile
647647+echo "COPY small-file /" >> Dockerfile
648648+dd if=/dev/urandom of=small-file bs=1024 count=512 # 512KB
649649+650650+docker build -t atcr.io/youruser/test:small .
651651+docker push atcr.io/youruser/test:small
652652+```
653653+654654+**Expected behavior:**
655655+- Blob buffered in memory
656656+- Presigned upload URL requested with correct digest
657657+- Direct upload to S3 via presigned URL
658658+- No temp location, no move operation
659659+660660+**Verify in logs:**
661661+```
662662+📤 [ProxyBlobWriter.Commit] Uploading via presigned URL
663663+ Digest: sha256:...
664664+ Size: 524288 bytes
665665+ Presigned URL: https://gateway.storjshare.io/...
666666+✅ [ProxyBlobWriter.Commit] Upload successful
667667+```
668668+669669+#### 2. Medium Blob Upload (10-50MB)
670670+```bash
671671+dd if=/dev/urandom of=medium-file bs=1048576 count=25 # 25MB
672672+673673+docker build -t atcr.io/youruser/test:medium .
674674+docker push atcr.io/youruser/test:medium
675675+```
676676+677677+**Monitor memory usage:**
678678+```bash
679679+# While push is running
680680+docker stats atcr-appview
681681+```
682682+683683+Should see ~25MB spike during buffer + upload.
684684+685685+#### 3. Large Blob Upload (100-500MB)
686686+```bash
687687+dd if=/dev/urandom of=large-file bs=1048576 count=200 # 200MB
688688+689689+docker build -t atcr.io/youruser/test:large .
690690+docker push atcr.io/youruser/test:large
691691+```
692692+693693+**Monitor:**
694694+- Memory usage (should see ~200MB spike)
695695+- Upload completes successfully
696696+- S3 shows blob in correct location
697697+698698+#### 4. Concurrent Uploads
699699+```bash
700700+# Push multiple images in parallel
701701+docker push atcr.io/youruser/test1:tag &
702702+docker push atcr.io/youruser/test2:tag &
703703+docker push atcr.io/youruser/test3:tag &
704704+wait
705705+```
706706+707707+**Verify:**
708708+- All uploads complete successfully
709709+- Memory usage peaks but doesn't OOM
710710+- No data corruption (digests match)
711711+712712+#### 5. Error Handling Tests
713713+714714+**Test presigned URL failure:**
715715+- Temporarily break S3 credentials
716716+- Verify graceful error message
717717+- Check for memory leaks (buffer cleared on error)
718718+719719+**Test digest mismatch:**
720720+- This shouldn't happen in practice, but verify error handling
721721+- Buffer should be cleared even on error
722722+723723+**Test network interruption:**
724724+- Kill network during upload
725725+- Verify proper error propagation
726726+- Check for hanging goroutines
727727+728728+### Test Plan for Presigned HEAD URLs (Optional)
729729+730730+#### 1. HEAD Request Redirect
731731+```bash
732732+# Pull image (triggers HEAD verification)
733733+docker pull atcr.io/youruser/test:tag
734734+```
735735+736736+**Expected behavior:**
737737+- AppView redirects HEAD to presigned HEAD URL
738738+- Docker follows redirect to S3
739739+- S3 responds to HEAD request successfully
740740+741741+**Verify in logs:**
742742+```
743743+🔍 [getHeadURL] Generating presigned HEAD URL:
744744+ Digest: sha256:...
745745+✅ [getHeadURL] Presigned HEAD URL generated
746746+```
747747+748748+#### 2. Method Verification
749749+```bash
750750+# Manually verify presigned HEAD URL works
751751+curl -I "presigned-head-url-here"
752752+```
753753+754754+Should return 200 OK with Content-Length header.
755755+756756+```bash
757757+# Verify it ONLY works with HEAD (not GET)
758758+curl "presigned-head-url-here"
759759+```
760760+761761+Should return 403 Forbidden (method mismatch).
762762+763763+---
764764+765765+## Performance Comparison
766766+767767+### Current Architecture (Proxy Mode)
768768+769769+**Upload:**
770770+```
771771+Client → AppView (stream) → Hold Service (stream) → S3
772772+ ~0ms delay ~0ms delay ~100ms
773773+```
774774+- Total latency: ~100ms + upload time
775775+- Bandwidth: All through Hold Service
776776+777777+**Download:**
778778+```
779779+Client → AppView (redirect) → S3 (presigned GET)
780780+ ~5ms ~50ms
781781+```
782782+- Total latency: ~55ms + download time
783783+- Bandwidth: Direct from S3 ✅
784784+785785+**Verification (HEAD):**
786786+```
787787+Client → AppView (redirect) → Hold Service (proxy HEAD) → S3
788788+ ~5ms ~10ms ~50ms
789789+```
790790+- Total latency: ~65ms
791791+- Bandwidth: ~300 bytes through Hold Service
792792+793793+### Presigned Upload Architecture
794794+795795+**Upload:**
796796+```
797797+Client → AppView (buffer) → S3 (presigned PUT)
798798+ ~0ms ~100ms
799799+```
800800+- Total latency: ~100ms + upload time (same)
801801+- Bandwidth: Direct to S3 ✅
802802+- Memory: +blob_size during buffer
803803+804804+**Download:** (unchanged)
805805+```
806806+Client → AppView (redirect) → S3 (presigned GET)
807807+```
808808+809809+**Verification (HEAD):** (if presigned HEAD enabled)
810810+```
811811+Client → AppView (redirect) → S3 (presigned HEAD)
812812+ ~5ms ~50ms
813813+```
814814+- Total latency: ~55ms (10ms faster)
815815+- Bandwidth: Direct to S3 ✅
816816+817817+---
818818+819819+## Trade-offs Summary
820820+821821+### Presigned Uploads
822822+823823+| Aspect | Proxy Mode (Current) | Presigned URLs |
824824+|--------|---------------------|----------------|
825825+| **Upload Bandwidth** | Through Hold Service | Direct to S3 ✅ |
826826+| **Hold Service Load** | High (all upload traffic) | Low (only URL generation) ✅ |
827827+| **Memory Usage** | Low (streaming) | High (buffering) ⚠️ |
828828+| **Disk Usage** | None | Optional temp files for large blobs |
829829+| **Code Complexity** | Simple ✅ | Moderate |
830830+| **Max Blob Size** | Unlimited ✅ | Limited by memory (~500MB) ⚠️ |
831831+| **Latency** | Same | Same |
832832+| **Error Recovery** | Simple (cancel stream) | More complex (clear buffer) |
833833+834834+### Presigned HEAD URLs
835835+836836+| Aspect | Proxy Mode (Current) | Presigned HEAD |
837837+|--------|---------------------|----------------|
838838+| **Bandwidth** | 300 bytes (negligible) | 500 bytes (still negligible) |
839839+| **Hold Service Load** | Low (HEAD is tiny) | Lower (but minimal gain) |
840840+| **Latency** | 65ms | 55ms (10ms faster) |
841841+| **Code Complexity** | Simple ✅ | More complex |
842842+| **Reliability** | High (fewer moving parts) ✅ | Moderate (more failure modes) |
843843+844844+---
845845+846846+## Recommendations
847847+848848+### Presigned Uploads
849849+850850+**Implement if:**
851851+- ✅ Hold Service bandwidth is a concern
852852+- ✅ You want to minimize Hold Service load
853853+- ✅ Most blobs are < 100MB (typical Docker layers)
854854+- ✅ AppView has sufficient memory (2-4GB+ RAM)
855855+856856+**Skip if:**
857857+- ⚠️ Memory is constrained
858858+- ⚠️ You regularly push very large layers (> 500MB)
859859+- ⚠️ Current proxy mode is working fine
860860+- ⚠️ Simplicity is priority
861861+862862+### Presigned HEAD URLs
863863+864864+**Implement if:**
865865+- ✅ You want complete S3 offloading
866866+- ✅ You're already implementing presigned uploads
867867+- ✅ Hold Service is CPU/bandwidth constrained
868868+869869+**Skip if:**
870870+- ⚠️ Current HEAD proxying works fine (it does)
871871+- ⚠️ You want to minimize code complexity
872872+- ⚠️ 10ms latency difference doesn't matter
873873+874874+### Suggested Approach
875875+876876+**Phase 1:** Implement presigned uploads first
877877+- Bigger performance win (offloads upload bandwidth)
878878+- More valuable for write-heavy workflows
879879+- Test thoroughly with various blob sizes
880880+881881+**Phase 2:** Monitor and evaluate
882882+- Check Hold Service load after presigned uploads
883883+- Measure HEAD request impact
884884+- Assess if presigned HEAD is worth the complexity
885885+886886+**Phase 3:** Optionally add presigned HEAD
887887+- Only if Hold Service is still bottlenecked
888888+- Or if you want feature completeness
889889+890890+---
891891+892892+## Migration Path
893893+894894+### Step 1: Feature Flag
895895+Add configuration option to enable/disable presigned uploads:
896896+897897+```go
898898+// In AppView config
899899+type Config struct {
900900+ // ... existing fields ...
901901+902902+ UsePresignedUploads bool `yaml:"use_presigned_uploads"` // Default: false
903903+}
904904+```
905905+906906+### Step 2: Gradual Rollout
907907+1. Deploy with `use_presigned_uploads: false` (current behavior)
908908+2. Test in staging with `use_presigned_uploads: true`
909909+3. Roll out to production incrementally
910910+4. Monitor memory usage and error rates
911911+912912+### Step 3: Fallback Mechanism
913913+If presigned upload fails, fallback to proxy mode:
914914+915915+```go
916916+func (w *ProxyBlobWriter) Commit(...) {
917917+ // Try presigned upload
918918+ url, err := w.store.getUploadURL(ctx, finalDigest, w.size)
919919+ if err != nil {
920920+ // Fallback: use proxy mode
921921+ log.Printf("⚠️ Presigned upload unavailable, falling back to proxy")
922922+ return w.proxyUpload(ctx, desc)
923923+ }
924924+ // ... presigned upload ...
925925+}
926926+```
927927+928928+---
929929+930930+## Appendix: Memory Profiling
931931+932932+To monitor memory usage during development:
933933+934934+```bash
935935+# Enable Go memory profiling
936936+go tool pprof http://localhost:5000/debug/pprof/heap
937937+938938+# Or use runtime metrics
939939+import "runtime"
940940+941941+var m runtime.MemStats
942942+runtime.ReadMemStats(&m)
943943+fmt.Printf("Alloc = %v MB", m.Alloc / 1024 / 1024)
944944+```
945945+946946+Monitor these metrics:
947947+- `Alloc`: Current memory allocation
948948+- `TotalAlloc`: Cumulative allocation (detect leaks)
949949+- `Sys`: Total memory from OS
950950+- `NumGC`: Garbage collection count
951951+952952+Expected behavior with presigned uploads:
953953+- Memory spikes during `Write()` calls
954954+- Memory drops after `Commit()` completes
955955+- No memory leaks (TotalAlloc should plateau)
956956+957957+---
958958+959959+## Questions for Decision
960960+961961+Before implementing, answer:
962962+963963+1. **What's the typical size of your Docker layers?**
964964+ - < 50MB: Presigned uploads perfect fit
965965+ - 50-200MB: Acceptable with memory monitoring
966966+ - > 200MB: Consider disk buffering or stick with proxy
967967+968968+2. **What's your AppView's available memory?**
969969+ - 1GB: Skip presigned uploads
970970+ - 2-4GB: Fine for typical workloads
971971+ - 8GB+: No concerns
972972+973973+3. **Is Hold Service bandwidth currently a problem?**
974974+ - No: Current proxy mode is fine
975975+ - Yes: Presigned uploads will help significantly
976976+977977+4. **How important is code simplicity?**
978978+ - Very: Stick with proxy mode
979979+ - Moderate: Implement presigned uploads only
980980+ - Low: Implement both presigned uploads and HEAD
981981+982982+5. **What's your deployment model?**
983983+ - Single Hold Service: Bandwidth matters more
984984+ - Multiple Hold Services: Less critical
985985+986986+---
987987+988988+## Implementation Checklist
989989+990990+### Presigned Uploads
991991+- [ ] Modify `ProxyBlobWriter` struct (remove pipe, add buffer/hasher)
992992+- [ ] Update `Create()` to initialize buffer
993993+- [ ] Update `Write()` to buffer + hash data
994994+- [ ] Update `Commit()` to upload via presigned URL
995995+- [ ] Update `Cancel()` to clear buffer
996996+- [ ] Add memory usage monitoring
997997+- [ ] Add configuration flag
998998+- [ ] Test with small blobs (< 1MB)
999999+- [ ] Test with medium blobs (10-50MB)
10001000+- [ ] Test with large blobs (100-500MB)
10011001+- [ ] Test concurrent uploads
10021002+- [ ] Test error scenarios
10031003+- [ ] Update documentation
10041004+- [ ] Deploy to staging
10051005+- [ ] Monitor production rollout
10061006+10071007+### Presigned HEAD URLs (Optional)
10081008+- [ ] Add `getHeadURL()` to Hold Service
10091009+- [ ] Add `HandleHeadPresignedURL()` endpoint
10101010+- [ ] Register `/head-presigned-url` route
10111011+- [ ] Add `getHeadURL()` to ProxyBlobStore
10121012+- [ ] Update `ServeBlob()` to redirect HEAD requests
10131013+- [ ] Test HEAD redirects
10141014+- [ ] Verify method-specific signatures
10151015+- [ ] Test with Docker pull operations
10161016+- [ ] Deploy to staging
10171017+- [ ] Monitor production rollout
+49
docs/PRESIGNED_URLS.md
···718718719719The implementation has automatic fallbacks, so partial failures won't break functionality.
720720721721+## Testing with DISABLE_PRESIGNED_URLS
722722+723723+### Environment Variable
724724+725725+Set `DISABLE_PRESIGNED_URLS=true` to force proxy/buffered mode even when S3 is configured.
726726+727727+**Use cases:**
728728+- Testing proxy/buffered code paths with S3 storage
729729+- Debugging multipart uploads in buffered mode
730730+- Simulating S3 providers that don't support presigned URLs
731731+- Verifying fallback behavior works correctly
732732+733733+### How It Works
734734+735735+When `DISABLE_PRESIGNED_URLS=true`:
736736+737737+**Single blob operations:**
738738+- `getDownloadURL()` returns proxy URL instead of S3 presigned URL
739739+- `getHeadURL()` returns proxy URL instead of S3 presigned HEAD URL
740740+- `getUploadURL()` returns proxy URL instead of S3 presigned PUT URL
741741+- Client uses `/blobs/{digest}` endpoints (proxy through hold service)
742742+743743+**Multipart uploads:**
744744+- `StartMultipartUploadWithManager()` creates **Buffered** session instead of **S3Native**
745745+- `GetPartUploadURL()` returns `/multipart-parts/{uploadID}/{partNumber}` instead of S3 presigned URL
746746+- Parts are buffered in memory in the hold service
747747+- `CompleteMultipartUploadWithManager()` assembles parts and writes via storage driver
748748+749749+### Testing Example
750750+751751+```bash
752752+# Test S3 with forced proxy mode
753753+export STORAGE_DRIVER=s3
754754+export S3_BUCKET=my-bucket
755755+export AWS_ACCESS_KEY_ID=...
756756+export AWS_SECRET_ACCESS_KEY=...
757757+export DISABLE_PRESIGNED_URLS=true # Force buffered/proxy mode
758758+759759+./bin/atcr-hold
760760+761761+# Push an image - should use proxy mode
762762+docker push atcr.io/yourdid/test:latest
763763+764764+# Check logs for:
765765+# "Presigned URLs disabled, using proxy URL"
766766+# "Presigned URLs disabled (DISABLE_PRESIGNED_URLS=true), using buffered mode"
767767+# "Stored part: uploadID=... part=1 size=..."
768768+```
769769+721770## Future Enhancements
722771723772### 1. Configurable Expiration
+21
pkg/atproto/client.go
···33import (
44 "bytes"
55 "context"
66+ "encoding/base64"
67 "encoding/json"
78 "fmt"
89 "io"
···343344 return nil, fmt.Errorf("failed to read blob data: %w", err)
344345 }
345346347347+ // Check if PDS returned JSON-wrapped blob (Bluesky implementation)
348348+ // PDS may wrap blobs as JSON-encoded base64 strings
349349+ // Detection: Check if content starts with a quote (indicating JSON string)
350350+ if len(data) > 0 && data[0] == '"' {
351351+ // Blob is JSON-encoded - decode it
352352+ var base64Str string
353353+ if err := json.Unmarshal(data, &base64Str); err != nil {
354354+ return nil, fmt.Errorf("failed to unmarshal JSON-wrapped blob: %w", err)
355355+ }
356356+357357+ // Base64-decode the blob content
358358+ decoded, err := base64.StdEncoding.DecodeString(base64Str)
359359+ if err != nil {
360360+ return nil, fmt.Errorf("failed to base64-decode blob: %w", err)
361361+ }
362362+363363+ return decoded, nil
364364+ }
365365+366366+ // Raw blob response (expected ATProto behavior)
346367 return data, nil
347368}
348369
+7
pkg/auth/scope.go
···55 "strings"
66)
7788+// AccessEntry represents access permissions for a resource
99+type AccessEntry struct {
1010+ Type string `json:"type"` // "repository"
1111+ Name string `json:"name,omitempty"` // e.g., "alice/myapp"
1212+ Actions []string `json:"actions,omitempty"` // e.g., ["pull", "push"]
1313+}
1414+815// ParseScope parses Docker registry scope strings into AccessEntry structures
916// Scope format: "repository:alice/myapp:pull,push"
1017// Multiple scopes can be provided
-8
pkg/auth/types.go
···11-package auth
22-33-// AccessEntry represents access permissions for a resource
44-type AccessEntry struct {
55- Type string `json:"type"` // "repository"
66- Name string `json:"name,omitempty"` // e.g., "alice/myapp"
77- Actions []string `json:"actions,omitempty"` // e.g., ["pull", "push"]
88-}
+131
pkg/hold/authorization.go
···11+package hold
22+33+import (
44+ "context"
55+ "encoding/json"
66+ "fmt"
77+ "log"
88+99+ "atcr.io/pkg/atproto"
1010+ "github.com/bluesky-social/indigo/atproto/identity"
1111+ "github.com/bluesky-social/indigo/atproto/syntax"
1212+)
1313+1414+// isAuthorizedRead checks if a DID can read from this hold
1515+// Authorization:
1616+// - Public hold: allow anonymous (empty DID) or any authenticated user
1717+// - Private hold: require authentication (any user with sailor.profile)
1818+func (s *HoldService) isAuthorizedRead(did string) bool {
1919+ // Check hold public flag
2020+ isPublic, err := s.isHoldPublic()
2121+ if err != nil {
2222+ log.Printf("ERROR: Failed to check hold public flag: %v", err)
2323+ // Fail secure - deny access on error
2424+ return false
2525+ }
2626+2727+ if isPublic {
2828+ // Public hold - allow anyone (even anonymous)
2929+ return true
3030+ }
3131+3232+ // Private hold - require authentication
3333+ // Any authenticated user with sailor.profile can read
3434+ if did == "" {
3535+ // Anonymous user trying to access private hold
3636+ return false
3737+ }
3838+3939+ // For MVP: assume DID presence means they have sailor.profile
4040+ // Future: could query PDS to verify sailor.profile exists
4141+ return true
4242+}
4343+4444+// isAuthorizedWrite checks if a DID can write to this hold
4545+// Authorization: must be hold owner OR crew member
4646+func (s *HoldService) isAuthorizedWrite(did string) bool {
4747+ if did == "" {
4848+ // Anonymous writes not allowed
4949+ return false
5050+ }
5151+5252+ // Check if DID is the hold owner
5353+ ownerDID := s.config.Registration.OwnerDID
5454+ if ownerDID == "" {
5555+ log.Printf("ERROR: Hold owner DID not configured")
5656+ return false
5757+ }
5858+5959+ if did == ownerDID {
6060+ // Owner always has write access
6161+ return true
6262+ }
6363+6464+ // Check if DID is a crew member
6565+ isCrew, err := s.isCrewMember(did)
6666+ if err != nil {
6767+ log.Printf("ERROR: Failed to check crew membership: %v", err)
6868+ return false
6969+ }
7070+7171+ return isCrew
7272+}
7373+7474+// isHoldPublic checks if this hold allows public (anonymous) reads
7575+func (s *HoldService) isHoldPublic() (bool, error) {
7676+ // Use cached config value for now
7777+ // Future: could query PDS for hold record to get live value
7878+ return s.config.Server.Public, nil
7979+}
8080+8181+// isCrewMember checks if a DID is a crew member of this hold
8282+func (s *HoldService) isCrewMember(did string) (bool, error) {
8383+ ownerDID := s.config.Registration.OwnerDID
8484+ if ownerDID == "" {
8585+ return false, fmt.Errorf("hold owner DID not configured")
8686+ }
8787+8888+ ctx := context.Background()
8989+9090+ // Resolve owner's PDS endpoint using indigo
9191+ directory := identity.DefaultDirectory()
9292+ ownerDIDParsed, err := syntax.ParseDID(ownerDID)
9393+ if err != nil {
9494+ return false, fmt.Errorf("invalid owner DID: %w", err)
9595+ }
9696+9797+ ident, err := directory.LookupDID(ctx, ownerDIDParsed)
9898+ if err != nil {
9999+ return false, fmt.Errorf("failed to resolve owner PDS: %w", err)
100100+ }
101101+102102+ pdsEndpoint := ident.PDSEndpoint()
103103+ if pdsEndpoint == "" {
104104+ return false, fmt.Errorf("no PDS endpoint found for owner")
105105+ }
106106+107107+ // Create unauthenticated client to read public records
108108+ client := atproto.NewClient(pdsEndpoint, ownerDID, "")
109109+110110+ // List crew records for this hold
111111+ // Crew records are public, so we can read them without auth
112112+ records, err := client.ListRecords(ctx, atproto.HoldCrewCollection, 100)
113113+ if err != nil {
114114+ return false, fmt.Errorf("failed to list crew records: %w", err)
115115+ }
116116+117117+ // Check if DID is in crew list
118118+ for _, record := range records {
119119+ var crewRecord atproto.HoldCrewRecord
120120+ if err := json.Unmarshal(record.Value, &crewRecord); err != nil {
121121+ continue
122122+ }
123123+124124+ if crewRecord.Member == did {
125125+ // Found crew membership
126126+ return true, nil
127127+ }
128128+ }
129129+130130+ return false, nil
131131+}
+134
pkg/hold/config.go
···11+package hold
22+33+import (
44+ "fmt"
55+ "os"
66+ "time"
77+88+ "github.com/distribution/distribution/v3/configuration"
99+)
1010+1111+// Config represents the hold service configuration
1212+type Config struct {
1313+ Version string `yaml:"version"`
1414+ Storage StorageConfig `yaml:"storage"`
1515+ Server ServerConfig `yaml:"server"`
1616+ Registration RegistrationConfig `yaml:"registration"`
1717+}
1818+1919+// RegistrationConfig defines auto-registration settings
2020+type RegistrationConfig struct {
2121+ // OwnerDID is the owner's ATProto DID (from env: HOLD_OWNER)
2222+ // If set, auto-registration is enabled
2323+ OwnerDID string `yaml:"owner_did"`
2424+}
2525+2626+// StorageConfig wraps distribution's storage configuration
2727+type StorageConfig struct {
2828+ configuration.Storage `yaml:",inline"`
2929+}
3030+3131+// ServerConfig defines server settings
3232+type ServerConfig struct {
3333+ // Addr is the address to listen on (e.g., ":8080")
3434+ Addr string `yaml:"addr"`
3535+3636+ // PublicURL is the public URL of this hold service (e.g., "https://hold.example.com")
3737+ PublicURL string `yaml:"public_url"`
3838+3939+ // Public controls whether this hold allows public blob reads without auth (from env: HOLD_PUBLIC)
4040+ Public bool `yaml:"public"`
4141+4242+ // TestMode uses localhost for OAuth redirects while storing real URL in hold record (from env: TEST_MODE)
4343+ TestMode bool `yaml:"test_mode"`
4444+4545+ // DisablePresignedURLs forces proxy mode even with S3 configured (for testing) (from env: DISABLE_PRESIGNED_URLS)
4646+ DisablePresignedURLs bool `yaml:"disable_presigned_urls"`
4747+4848+ // ReadTimeout for HTTP requests
4949+ ReadTimeout time.Duration `yaml:"read_timeout"`
5050+5151+ // WriteTimeout for HTTP requests
5252+ WriteTimeout time.Duration `yaml:"write_timeout"`
5353+}
5454+5555+// LoadConfigFromEnv loads all configuration from environment variables
5656+func LoadConfigFromEnv() (*Config, error) {
5757+ cfg := &Config{
5858+ Version: "0.1",
5959+ }
6060+6161+ // Server configuration
6262+ cfg.Server.Addr = getEnvOrDefault("HOLD_SERVER_ADDR", ":8080")
6363+ cfg.Server.PublicURL = os.Getenv("HOLD_PUBLIC_URL")
6464+ if cfg.Server.PublicURL == "" {
6565+ return nil, fmt.Errorf("HOLD_PUBLIC_URL is required")
6666+ }
6767+ cfg.Server.Public = os.Getenv("HOLD_PUBLIC") == "true"
6868+ cfg.Server.TestMode = os.Getenv("TEST_MODE") == "true"
6969+ cfg.Server.DisablePresignedURLs = os.Getenv("DISABLE_PRESIGNED_URLS") == "true"
7070+ cfg.Server.ReadTimeout = 5 * time.Minute // Increased for large blob uploads
7171+ cfg.Server.WriteTimeout = 5 * time.Minute // Increased for large blob uploads
7272+7373+ // Registration configuration (optional)
7474+ cfg.Registration.OwnerDID = os.Getenv("HOLD_OWNER")
7575+7676+ // Storage configuration - build from env vars based on storage type
7777+ storageType := getEnvOrDefault("STORAGE_DRIVER", "s3")
7878+ var err error
7979+ cfg.Storage, err = buildStorageConfig(storageType)
8080+ if err != nil {
8181+ return nil, fmt.Errorf("failed to build storage config: %w", err)
8282+ }
8383+8484+ return cfg, nil
8585+}
8686+8787+// buildStorageConfig creates storage configuration based on driver type
8888+func buildStorageConfig(driver string) (StorageConfig, error) {
8989+ params := make(map[string]any)
9090+9191+ switch driver {
9292+ case "s3":
9393+ // S3/Storj/Minio configuration from standard AWS env vars
9494+ accessKey := os.Getenv("AWS_ACCESS_KEY_ID")
9595+ secretKey := os.Getenv("AWS_SECRET_ACCESS_KEY")
9696+ region := getEnvOrDefault("AWS_REGION", "us-east-1")
9797+ bucket := os.Getenv("S3_BUCKET")
9898+ endpoint := os.Getenv("S3_ENDPOINT") // For Storj/Minio
9999+100100+ if bucket == "" {
101101+ return StorageConfig{}, fmt.Errorf("S3_BUCKET is required for S3 storage")
102102+ }
103103+104104+ params["accesskey"] = accessKey
105105+ params["secretkey"] = secretKey
106106+ params["region"] = region
107107+ params["bucket"] = bucket
108108+ if endpoint != "" {
109109+ params["regionendpoint"] = endpoint
110110+ }
111111+112112+ case "filesystem":
113113+ // Filesystem configuration
114114+ rootDir := getEnvOrDefault("STORAGE_ROOT_DIR", "/var/lib/atcr/hold")
115115+ params["rootdirectory"] = rootDir
116116+117117+ default:
118118+ return StorageConfig{}, fmt.Errorf("unsupported storage driver: %s", driver)
119119+ }
120120+121121+ // Build distribution Storage config
122122+ storageCfg := configuration.Storage{}
123123+ storageCfg[driver] = configuration.Parameters(params)
124124+125125+ return StorageConfig{Storage: storageCfg}, nil
126126+}
127127+128128+// getEnvOrDefault gets an environment variable or returns a default value
129129+func getEnvOrDefault(key, defaultValue string) string {
130130+ if val := os.Getenv(key); val != "" {
131131+ return val
132132+ }
133133+ return defaultValue
134134+}
+587
pkg/hold/handlers.go
···11+package hold
22+33+import (
44+ "context"
55+ "encoding/json"
66+ "fmt"
77+ "io"
88+ "log"
99+ "net/http"
1010+ "time"
1111+1212+ "atcr.io/pkg/atproto"
1313+)
1414+1515+// PresignedURLOperation defines the type of presigned URL operation
1616+type PresignedURLOperation string
1717+1818+const (
1919+ OperationGet PresignedURLOperation = "GET"
2020+ OperationHead PresignedURLOperation = "HEAD"
2121+ OperationPut PresignedURLOperation = "PUT"
2222+)
2323+2424+// PresignedURLRequest represents a request for a presigned URL (GET, HEAD, or PUT)
2525+type PresignedURLRequest struct {
2626+ Operation PresignedURLOperation `json:"operation"`
2727+ DID string `json:"did"`
2828+ Digest string `json:"digest"`
2929+ Size int64 `json:"size,omitempty"` // Only required for PUT operations
3030+}
3131+3232+// PresignedURLResponse contains the presigned URL
3333+type PresignedURLResponse struct {
3434+ URL string `json:"url"`
3535+ ExpiresAt time.Time `json:"expires_at"`
3636+}
3737+3838+// HandlePresignedURL handles presigned URL requests (GET, HEAD, or PUT)
3939+// Operation type is specified in the request body
4040+func (s *HoldService) HandlePresignedURL(w http.ResponseWriter, r *http.Request) {
4141+ if r.Method != http.MethodPost {
4242+ http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
4343+ return
4444+ }
4545+4646+ var req PresignedURLRequest
4747+ if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
4848+ http.Error(w, fmt.Sprintf("invalid request: %v", err), http.StatusBadRequest)
4949+ return
5050+ }
5151+5252+ // Validate DID authorization based on operation type
5353+ var authorized bool
5454+ switch req.Operation {
5555+ case OperationGet, OperationHead:
5656+ authorized = s.isAuthorizedRead(req.DID)
5757+ case OperationPut:
5858+ authorized = s.isAuthorizedWrite(req.DID)
5959+ default:
6060+ http.Error(w, "unsupported operation", http.StatusBadRequest)
6161+ return
6262+ }
6363+6464+ if !authorized {
6565+ log.Printf("[HandlePresignedURL:%s] Authorization FAILED", req.Operation)
6666+ if req.DID == "" {
6767+ http.Error(w, "unauthorized: authentication required", http.StatusUnauthorized)
6868+ } else {
6969+ http.Error(w, "forbidden: access denied", http.StatusForbidden)
7070+ }
7171+ return
7272+ }
7373+7474+ // Generate presigned URL (15 minute expiry)
7575+ ctx := context.Background()
7676+ expiry := time.Now().Add(15 * time.Minute)
7777+7878+ url, err := s.getPresignedURL(ctx, req.Operation, req.Digest, req.DID)
7979+ if err != nil {
8080+ log.Printf("[HandlePresignedURL:%s] getPresignedURL failed: %v", req.Operation, err)
8181+ http.Error(w, fmt.Sprintf("failed to generate URL: %v", err), http.StatusInternalServerError)
8282+ return
8383+ }
8484+8585+ log.Printf("[HandlePresignedURL:%s] Returning URL to client", req.Operation)
8686+8787+ resp := PresignedURLResponse{
8888+ URL: url,
8989+ ExpiresAt: expiry,
9090+ }
9191+9292+ w.Header().Set("Content-Type", "application/json")
9393+ json.NewEncoder(w).Encode(resp)
9494+}
9595+9696+// HandleProxyGet proxies a blob download through the service
9797+func (s *HoldService) HandleProxyGet(w http.ResponseWriter, r *http.Request) {
9898+ if r.Method != http.MethodGet && r.Method != http.MethodHead {
9999+ http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
100100+ return
101101+ }
102102+103103+ // Extract digest from path (e.g., /blobs/sha256:abc123)
104104+ digest := r.URL.Path[len("/blobs/"):]
105105+ if digest == "" {
106106+ http.Error(w, "missing digest", http.StatusBadRequest)
107107+ return
108108+ }
109109+110110+ // Get DID from query param or header
111111+ did := r.URL.Query().Get("did")
112112+ if did == "" {
113113+ did = r.Header.Get("X-ATCR-DID")
114114+ }
115115+ log.Printf(" DID: %s", did)
116116+117117+ // Authorize READ access
118118+ if !s.isAuthorizedRead(did) {
119119+ log.Printf("[HandleProxyGet] Authorization FAILED")
120120+ if did == "" {
121121+ http.Error(w, "unauthorized: authentication required", http.StatusUnauthorized)
122122+ } else {
123123+ http.Error(w, "forbidden: access denied", http.StatusForbidden)
124124+ }
125125+ return
126126+ }
127127+128128+ ctx := r.Context()
129129+ path := blobPath(digest)
130130+131131+ // For HEAD requests, just check if blob exists
132132+ if r.Method == http.MethodHead {
133133+ stat, err := s.driver.Stat(ctx, path)
134134+ if err != nil {
135135+ http.Error(w, "blob not found", http.StatusNotFound)
136136+ return
137137+ }
138138+ w.Header().Set("Content-Type", "application/octet-stream")
139139+ w.Header().Set("Content-Length", fmt.Sprintf("%d", stat.Size()))
140140+ w.WriteHeader(http.StatusOK)
141141+ return
142142+ }
143143+144144+ // For GET requests, read and return the blob
145145+ content, err := s.driver.GetContent(ctx, path)
146146+ if err != nil {
147147+ http.Error(w, "blob not found", http.StatusNotFound)
148148+ return
149149+ }
150150+151151+ w.Header().Set("Content-Type", "application/octet-stream")
152152+ w.Write(content)
153153+}
154154+155155+// HandleMove moves a blob from one path to another
156156+// POST /move?from={path}&to={digest}&did={did}
157157+func (s *HoldService) HandleMove(w http.ResponseWriter, r *http.Request) {
158158+ if r.Method != http.MethodPost {
159159+ http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
160160+ return
161161+ }
162162+163163+ fromPath := r.URL.Query().Get("from")
164164+ toDigest := r.URL.Query().Get("to")
165165+ did := r.URL.Query().Get("did")
166166+167167+ if fromPath == "" || toDigest == "" {
168168+ http.Error(w, "missing from or to parameter", http.StatusBadRequest)
169169+ return
170170+ }
171171+172172+ // Authorize WRITE access
173173+ if !s.isAuthorizedWrite(did) {
174174+ if did == "" {
175175+ http.Error(w, "unauthorized: authentication required", http.StatusUnauthorized)
176176+ } else {
177177+ http.Error(w, "forbidden: write access denied", http.StatusForbidden)
178178+ }
179179+ return
180180+ }
181181+182182+ ctx := r.Context()
183183+ sourcePath := blobPath(fromPath)
184184+ destPath := blobPath(toDigest)
185185+186186+ // Try to move using driver's Move operation
187187+ if err := s.driver.Move(ctx, sourcePath, destPath); err != nil {
188188+ log.Printf("HandleMove: failed to move blob: %v", err)
189189+ http.Error(w, fmt.Sprintf("failed to move blob: %v", err), http.StatusInternalServerError)
190190+ return
191191+ }
192192+193193+ log.Printf("HandleMove: successfully moved blob from=%s to=%s", fromPath, toDigest)
194194+ w.WriteHeader(http.StatusOK)
195195+}
196196+197197+// HandleProxyPut proxies a blob upload through the service
198198+func (s *HoldService) HandleProxyPut(w http.ResponseWriter, r *http.Request) {
199199+ if r.Method != http.MethodPut {
200200+ http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
201201+ return
202202+ }
203203+204204+ digest := r.URL.Path[len("/blobs/"):]
205205+ if digest == "" {
206206+ http.Error(w, "missing digest", http.StatusBadRequest)
207207+ return
208208+ }
209209+210210+ did := r.URL.Query().Get("did")
211211+ if did == "" {
212212+ did = r.Header.Get("X-ATCR-DID")
213213+ }
214214+215215+ // Authorize WRITE access
216216+ if !s.isAuthorizedWrite(did) {
217217+ log.Printf("[HandleProxyPut] Authorization FAILED")
218218+ if did == "" {
219219+ http.Error(w, "unauthorized: authentication required", http.StatusUnauthorized)
220220+ } else {
221221+ http.Error(w, "forbidden: write access denied", http.StatusForbidden)
222222+ }
223223+ return
224224+ }
225225+226226+ // Stream blob to storage (no buffering)
227227+ ctx := r.Context()
228228+ path := blobPath(digest)
229229+230230+ // Create writer for streaming
231231+ writer, err := s.driver.Writer(ctx, path, false)
232232+ if err != nil {
233233+ log.Printf("HandleProxyPut: failed to create writer: %v", err)
234234+ http.Error(w, "failed to create writer", http.StatusInternalServerError)
235235+ return
236236+ }
237237+238238+ // Stream directly from request body to storage
239239+ written, err := io.Copy(writer, r.Body)
240240+ if err != nil {
241241+ writer.Cancel(ctx)
242242+ log.Printf("HandleProxyPut: failed to write blob: %v", err)
243243+ http.Error(w, "failed to write blob", http.StatusInternalServerError)
244244+ return
245245+ }
246246+247247+ // Commit the write
248248+ if err := writer.Commit(ctx); err != nil {
249249+ log.Printf("HandleProxyPut: failed to commit blob: %v", err)
250250+ http.Error(w, "failed to commit blob", http.StatusInternalServerError)
251251+ return
252252+ }
253253+254254+ log.Printf("HandleProxyPut: successfully stored blob path=%s, size=%d", digest, written)
255255+ w.WriteHeader(http.StatusCreated)
256256+}
257257+258258+// StartMultipartUploadRequest initiates a multipart upload
259259+type StartMultipartUploadRequest struct {
260260+ DID string `json:"did"`
261261+ Digest string `json:"digest"`
262262+}
263263+264264+// StartMultipartUploadResponse contains the multipart upload ID
265265+type StartMultipartUploadResponse struct {
266266+ UploadID string `json:"upload_id"`
267267+ ExpiresAt time.Time `json:"expires_at"`
268268+}
269269+270270+// HandleStartMultipart initiates a multipart upload
271271+func (s *HoldService) HandleStartMultipart(w http.ResponseWriter, r *http.Request) {
272272+ if r.Method != http.MethodPost {
273273+ http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
274274+ return
275275+ }
276276+277277+ var req StartMultipartUploadRequest
278278+ if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
279279+ http.Error(w, fmt.Sprintf("invalid request: %v", err), http.StatusBadRequest)
280280+ return
281281+ }
282282+283283+ // Validate DID authorization for WRITE
284284+ if !s.isAuthorizedWrite(req.DID) {
285285+ if req.DID == "" {
286286+ http.Error(w, "unauthorized: authentication required", http.StatusUnauthorized)
287287+ } else {
288288+ http.Error(w, "forbidden: write access denied", http.StatusForbidden)
289289+ }
290290+ return
291291+ }
292292+293293+ // Start multipart upload with manager (supports both S3Native and Buffered modes)
294294+ ctx := r.Context()
295295+ uploadID, mode, err := s.StartMultipartUploadWithManager(ctx, req.Digest, s.MultipartMgr)
296296+ if err != nil {
297297+ http.Error(w, fmt.Sprintf("failed to start multipart upload: %v", err), http.StatusInternalServerError)
298298+ return
299299+ }
300300+301301+ log.Printf("Started multipart upload: uploadID=%s, mode=%v, digest=%s", uploadID, mode, req.Digest)
302302+303303+ expiry := time.Now().Add(24 * time.Hour) // Multipart uploads can take longer
304304+305305+ resp := StartMultipartUploadResponse{
306306+ UploadID: uploadID,
307307+ ExpiresAt: expiry,
308308+ }
309309+310310+ w.Header().Set("Content-Type", "application/json")
311311+ json.NewEncoder(w).Encode(resp)
312312+}
313313+314314+// GetPartURLRequest requests a presigned URL for a specific part
315315+type GetPartURLRequest struct {
316316+ DID string `json:"did"`
317317+ Digest string `json:"digest"`
318318+ UploadID string `json:"upload_id"`
319319+ PartNumber int `json:"part_number"`
320320+}
321321+322322+// GetPartURLResponse contains the presigned URL for a part
323323+type GetPartURLResponse struct {
324324+ URL string `json:"url"`
325325+ ExpiresAt time.Time `json:"expires_at"`
326326+}
327327+328328+// HandleGetPartURL generates a presigned URL for uploading a specific part
329329+func (s *HoldService) HandleGetPartURL(w http.ResponseWriter, r *http.Request) {
330330+ if r.Method != http.MethodPost {
331331+ http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
332332+ return
333333+ }
334334+335335+ var req GetPartURLRequest
336336+ if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
337337+ http.Error(w, fmt.Sprintf("invalid request: %v", err), http.StatusBadRequest)
338338+ return
339339+ }
340340+341341+ // Validate DID authorization for WRITE
342342+ if !s.isAuthorizedWrite(req.DID) {
343343+ if req.DID == "" {
344344+ http.Error(w, "unauthorized: authentication required", http.StatusUnauthorized)
345345+ } else {
346346+ http.Error(w, "forbidden: write access denied", http.StatusForbidden)
347347+ }
348348+ return
349349+ }
350350+351351+ // Get multipart session
352352+ session, err := s.MultipartMgr.GetSession(req.UploadID)
353353+ if err != nil {
354354+ http.Error(w, fmt.Sprintf("session not found: %v", err), http.StatusNotFound)
355355+ return
356356+ }
357357+358358+ // Get part upload URL (presigned for S3Native, proxy for Buffered)
359359+ ctx := r.Context()
360360+ url, err := s.GetPartUploadURL(ctx, session, req.PartNumber, req.DID)
361361+ if err != nil {
362362+ http.Error(w, fmt.Sprintf("failed to generate part URL: %v", err), http.StatusInternalServerError)
363363+ return
364364+ }
365365+366366+ expiry := time.Now().Add(15 * time.Minute)
367367+368368+ resp := GetPartURLResponse{
369369+ URL: url,
370370+ ExpiresAt: expiry,
371371+ }
372372+373373+ w.Header().Set("Content-Type", "application/json")
374374+ json.NewEncoder(w).Encode(resp)
375375+}
376376+377377+// CompleteMultipartRequest completes a multipart upload
378378+type CompleteMultipartRequest struct {
379379+ DID string `json:"did"`
380380+ Digest string `json:"digest"`
381381+ UploadID string `json:"upload_id"`
382382+ Parts []CompletedPart `json:"parts"`
383383+}
384384+385385+// CompletedPart represents an uploaded part with its ETag
386386+type CompletedPart struct {
387387+ PartNumber int `json:"part_number"`
388388+ ETag string `json:"etag"`
389389+}
390390+391391+// HandleCompleteMultipart completes a multipart upload
392392+func (s *HoldService) HandleCompleteMultipart(w http.ResponseWriter, r *http.Request) {
393393+ if r.Method != http.MethodPost {
394394+ http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
395395+ return
396396+ }
397397+398398+ var req CompleteMultipartRequest
399399+ if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
400400+ http.Error(w, fmt.Sprintf("invalid request: %v", err), http.StatusBadRequest)
401401+ return
402402+ }
403403+404404+ // Validate DID authorization for WRITE
405405+ if !s.isAuthorizedWrite(req.DID) {
406406+ if req.DID == "" {
407407+ http.Error(w, "unauthorized: authentication required", http.StatusUnauthorized)
408408+ } else {
409409+ http.Error(w, "forbidden: write access denied", http.StatusForbidden)
410410+ }
411411+ return
412412+ }
413413+414414+ // Get multipart session
415415+ session, err := s.MultipartMgr.GetSession(req.UploadID)
416416+ if err != nil {
417417+ http.Error(w, fmt.Sprintf("session not found: %v", err), http.StatusNotFound)
418418+ return
419419+ }
420420+421421+ // For S3Native mode, use parts from request (uploaded directly to S3)
422422+ // For Buffered mode, parts are in the session
423423+ if session.Mode == S3Native {
424424+ // Record parts from AppView's request (they have ETags from S3)
425425+ for _, p := range req.Parts {
426426+ session.RecordS3Part(p.PartNumber, p.ETag, 0)
427427+ }
428428+ log.Printf("Recorded %d S3 parts from request for uploadID=%s", len(req.Parts), req.UploadID)
429429+ }
430430+431431+ // Complete multipart upload (handles both S3Native and Buffered modes)
432432+ ctx := r.Context()
433433+ if err := s.CompleteMultipartUploadWithManager(ctx, session, s.MultipartMgr); err != nil {
434434+ http.Error(w, fmt.Sprintf("failed to complete multipart upload: %v", err), http.StatusInternalServerError)
435435+ return
436436+ }
437437+438438+ log.Printf("Completed multipart upload: uploadID=%s, mode=%v", req.UploadID, session.Mode)
439439+440440+ w.WriteHeader(http.StatusOK)
441441+ w.Header().Set("Content-Type", "application/json")
442442+ json.NewEncoder(w).Encode(map[string]string{
443443+ "status": "completed",
444444+ })
445445+}
446446+447447+// AbortMultipartRequest aborts an in-progress upload
448448+type AbortMultipartRequest struct {
449449+ DID string `json:"did"`
450450+ Digest string `json:"digest"`
451451+ UploadID string `json:"upload_id"`
452452+}
453453+454454+// HandleAbortMultipart aborts an in-progress multipart upload
455455+func (s *HoldService) HandleAbortMultipart(w http.ResponseWriter, r *http.Request) {
456456+ if r.Method != http.MethodPost {
457457+ http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
458458+ return
459459+ }
460460+461461+ var req AbortMultipartRequest
462462+ if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
463463+ http.Error(w, fmt.Sprintf("invalid request: %v", err), http.StatusBadRequest)
464464+ return
465465+ }
466466+467467+ // Validate DID authorization for WRITE
468468+ if !s.isAuthorizedWrite(req.DID) {
469469+ if req.DID == "" {
470470+ http.Error(w, "unauthorized: authentication required", http.StatusUnauthorized)
471471+ } else {
472472+ http.Error(w, "forbidden: write access denied", http.StatusForbidden)
473473+ }
474474+ return
475475+ }
476476+477477+ // Get multipart session
478478+ session, err := s.MultipartMgr.GetSession(req.UploadID)
479479+ if err != nil {
480480+ http.Error(w, fmt.Sprintf("session not found: %v", err), http.StatusNotFound)
481481+ return
482482+ }
483483+484484+ // Abort multipart upload (handles both S3Native and Buffered modes)
485485+ ctx := r.Context()
486486+ if err := s.AbortMultipartUploadWithManager(ctx, session, s.MultipartMgr); err != nil {
487487+ http.Error(w, fmt.Sprintf("failed to abort multipart upload: %v", err), http.StatusInternalServerError)
488488+ return
489489+ }
490490+491491+ log.Printf("Aborted multipart upload: uploadID=%s, mode=%v", req.UploadID, session.Mode)
492492+493493+ w.WriteHeader(http.StatusOK)
494494+ w.Header().Set("Content-Type", "application/json")
495495+ json.NewEncoder(w).Encode(map[string]string{
496496+ "status": "aborted",
497497+ })
498498+}
499499+500500+// RegisterRequest represents a request to register this hold in a user's PDS
501501+type RegisterRequest struct {
502502+ DID string `json:"did"`
503503+ AccessToken string `json:"access_token"`
504504+ PDSEndpoint string `json:"pds_endpoint"`
505505+}
506506+507507+// RegisterResponse contains the registration result
508508+type RegisterResponse struct {
509509+ HoldURI string `json:"hold_uri"`
510510+ CrewURI string `json:"crew_uri"`
511511+ Message string `json:"message"`
512512+}
513513+514514+// HandleRegister registers this hold service in a user's PDS (manual endpoint)
515515+func (s *HoldService) HandleRegister(w http.ResponseWriter, r *http.Request) {
516516+ if r.Method != http.MethodPost {
517517+ http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
518518+ return
519519+ }
520520+521521+ var req RegisterRequest
522522+ if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
523523+ http.Error(w, fmt.Sprintf("invalid request: %v", err), http.StatusBadRequest)
524524+ return
525525+ }
526526+527527+ // Validate required fields
528528+ if req.DID == "" || req.AccessToken == "" || req.PDSEndpoint == "" {
529529+ http.Error(w, "missing required fields: did, access_token, pds_endpoint", http.StatusBadRequest)
530530+ return
531531+ }
532532+533533+ // Get public URL from config
534534+ publicURL := s.config.Server.PublicURL
535535+ if publicURL == "" {
536536+ // Fallback to constructing URL from request
537537+ scheme := "http"
538538+ if r.TLS != nil {
539539+ scheme = "https"
540540+ }
541541+ publicURL = fmt.Sprintf("%s://%s", scheme, r.Host)
542542+ }
543543+544544+ // Derive hold name from URL
545545+ holdName, err := extractHostname(publicURL)
546546+ if err != nil {
547547+ http.Error(w, fmt.Sprintf("failed to extract hostname: %v", err), http.StatusBadRequest)
548548+ return
549549+ }
550550+551551+ ctx := r.Context()
552552+553553+ // Create ATProto client with user's credentials
554554+ client := atproto.NewClient(req.PDSEndpoint, req.DID, req.AccessToken)
555555+556556+ // Create HoldRecord
557557+ holdRecord := atproto.NewHoldRecord(publicURL, req.DID, s.config.Server.Public)
558558+559559+ holdResult, err := client.PutRecord(ctx, atproto.HoldCollection, holdName, holdRecord)
560560+ if err != nil {
561561+ http.Error(w, fmt.Sprintf("failed to create hold record: %v", err), http.StatusInternalServerError)
562562+ return
563563+ }
564564+565565+ log.Printf("Created hold record: %s", holdResult.URI)
566566+567567+ // Create HoldCrewRecord for the owner
568568+ crewRecord := atproto.NewHoldCrewRecord(holdResult.URI, req.DID, "owner")
569569+570570+ crewRKey := fmt.Sprintf("%s-%s", holdName, req.DID)
571571+ crewResult, err := client.PutRecord(ctx, atproto.HoldCrewCollection, crewRKey, crewRecord)
572572+ if err != nil {
573573+ http.Error(w, fmt.Sprintf("failed to create crew record: %v", err), http.StatusInternalServerError)
574574+ return
575575+ }
576576+577577+ log.Printf("Created crew record: %s", crewResult.URI)
578578+579579+ resp := RegisterResponse{
580580+ HoldURI: holdResult.URI,
581581+ CrewURI: crewResult.URI,
582582+ Message: fmt.Sprintf("Successfully registered hold service. Storage endpoint: %s", publicURL),
583583+ }
584584+585585+ w.Header().Set("Content-Type", "application/json")
586586+ json.NewEncoder(w).Encode(resp)
587587+}
+381
pkg/hold/multipart.go
···11+package hold
22+33+import (
44+ "context"
55+ "crypto/sha256"
66+ "encoding/hex"
77+ "fmt"
88+ "io"
99+ "log"
1010+ "net/http"
1111+ "sync"
1212+ "time"
1313+1414+ "github.com/google/uuid"
1515+)
1616+1717+// MultipartMode indicates how multipart uploads are handled
1818+type MultipartMode int
1919+2020+const (
2121+ // S3Native uses S3's native multipart API with presigned URLs
2222+ S3Native MultipartMode = iota
2323+ // Buffered buffers parts in memory and assembles them in the hold service
2424+ Buffered
2525+)
2626+2727+// MultipartSession tracks an in-progress multipart upload
2828+type MultipartSession struct {
2929+ UploadID string // Unique upload ID
3030+ Digest string // Target digest path
3131+ Mode MultipartMode // Upload mode (S3Native or Buffered)
3232+ S3UploadID string // S3 upload ID (for S3Native mode)
3333+ Parts map[int]*MultipartPart // Buffered parts (for Buffered mode)
3434+ CreatedAt time.Time // When upload started
3535+ LastActivity time.Time // Last part upload
3636+ mu sync.RWMutex // Protects Parts map
3737+}
3838+3939+// MultipartPart represents a single part in a multipart upload
4040+type MultipartPart struct {
4141+ PartNumber int // Part number (1-indexed)
4242+ Data []byte // Part data (for Buffered mode)
4343+ ETag string // ETag from S3 or computed hash
4444+ Size int64 // Part size in bytes
4545+ UploadedAt time.Time // When part was uploaded
4646+}
4747+4848+// MultipartManager manages multipart upload sessions
4949+type MultipartManager struct {
5050+ sessions map[string]*MultipartSession // uploadID -> session
5151+ mu sync.RWMutex // Protects sessions map
5252+}
5353+5454+// NewMultipartManager creates a new multipart manager
5555+func NewMultipartManager() *MultipartManager {
5656+ mgr := &MultipartManager{
5757+ sessions: make(map[string]*MultipartSession),
5858+ }
5959+6060+ // Start cleanup goroutine for abandoned uploads
6161+ go mgr.cleanupLoop()
6262+6363+ return mgr
6464+}
6565+6666+// cleanupLoop periodically cleans up expired sessions
6767+func (m *MultipartManager) cleanupLoop() {
6868+ ticker := time.NewTicker(15 * time.Minute)
6969+ defer ticker.Stop()
7070+7171+ for range ticker.C {
7272+ m.cleanupExpiredSessions()
7373+ }
7474+}
7575+7676+// cleanupExpiredSessions removes sessions inactive for >24 hours
7777+func (m *MultipartManager) cleanupExpiredSessions() {
7878+ m.mu.Lock()
7979+ defer m.mu.Unlock()
8080+8181+ now := time.Now()
8282+ for uploadID, session := range m.sessions {
8383+ if now.Sub(session.LastActivity) > 24*time.Hour {
8484+ log.Printf("Cleaning up expired multipart session: uploadID=%s, age=%v", uploadID, now.Sub(session.CreatedAt))
8585+ delete(m.sessions, uploadID)
8686+ }
8787+ }
8888+}
8989+9090+// CreateSession creates a new multipart upload session
9191+func (m *MultipartManager) CreateSession(digest string, mode MultipartMode, s3UploadID string) *MultipartSession {
9292+ uploadID := uuid.New().String()
9393+9494+ session := &MultipartSession{
9595+ UploadID: uploadID,
9696+ Digest: digest,
9797+ Mode: mode,
9898+ S3UploadID: s3UploadID,
9999+ Parts: make(map[int]*MultipartPart),
100100+ CreatedAt: time.Now(),
101101+ LastActivity: time.Now(),
102102+ }
103103+104104+ m.mu.Lock()
105105+ m.sessions[uploadID] = session
106106+ m.mu.Unlock()
107107+108108+ log.Printf("Created multipart session: uploadID=%s, digest=%s, mode=%v", uploadID, digest, mode)
109109+ return session
110110+}
111111+112112+// GetSession retrieves a multipart session by upload ID
113113+func (m *MultipartManager) GetSession(uploadID string) (*MultipartSession, error) {
114114+ m.mu.RLock()
115115+ defer m.mu.RUnlock()
116116+117117+ session, ok := m.sessions[uploadID]
118118+ if !ok {
119119+ return nil, fmt.Errorf("multipart session not found: %s", uploadID)
120120+ }
121121+122122+ return session, nil
123123+}
124124+125125+// DeleteSession removes a multipart session
126126+func (m *MultipartManager) DeleteSession(uploadID string) {
127127+ m.mu.Lock()
128128+ defer m.mu.Unlock()
129129+130130+ delete(m.sessions, uploadID)
131131+ log.Printf("Deleted multipart session: uploadID=%s", uploadID)
132132+}
133133+134134+// StorePart stores a part in the session (for Buffered mode)
135135+func (s *MultipartSession) StorePart(partNumber int, data []byte) string {
136136+ s.mu.Lock()
137137+ defer s.mu.Unlock()
138138+139139+ // Compute ETag as SHA256 hash of part data
140140+ hash := sha256.Sum256(data)
141141+ etag := hex.EncodeToString(hash[:])
142142+143143+ part := &MultipartPart{
144144+ PartNumber: partNumber,
145145+ Data: data,
146146+ ETag: etag,
147147+ Size: int64(len(data)),
148148+ UploadedAt: time.Now(),
149149+ }
150150+151151+ s.Parts[partNumber] = part
152152+ s.LastActivity = time.Now()
153153+154154+ log.Printf("Stored part: uploadID=%s, part=%d, size=%d bytes, etag=%s", s.UploadID, partNumber, len(data), etag)
155155+ return etag
156156+}
157157+158158+// RecordS3Part records a part uploaded to S3 (for S3Native mode)
159159+func (s *MultipartSession) RecordS3Part(partNumber int, etag string, size int64) {
160160+ s.mu.Lock()
161161+ defer s.mu.Unlock()
162162+163163+ part := &MultipartPart{
164164+ PartNumber: partNumber,
165165+ ETag: etag,
166166+ Size: size,
167167+ UploadedAt: time.Now(),
168168+ }
169169+170170+ s.Parts[partNumber] = part
171171+ s.LastActivity = time.Now()
172172+173173+ log.Printf("Recorded S3 part: uploadID=%s, part=%d, size=%d bytes, etag=%s", s.UploadID, partNumber, size, etag)
174174+}
175175+176176+// AssembleBufferedParts assembles all buffered parts into a single blob
177177+// Returns the complete data and total size
178178+func (s *MultipartSession) AssembleBufferedParts() ([]byte, int64, error) {
179179+ s.mu.RLock()
180180+ defer s.mu.RUnlock()
181181+182182+ if s.Mode != Buffered {
183183+ return nil, 0, fmt.Errorf("session is not in buffered mode")
184184+ }
185185+186186+ // Calculate total size
187187+ var totalSize int64
188188+ maxPart := 0
189189+ for partNum, part := range s.Parts {
190190+ totalSize += part.Size
191191+ if partNum > maxPart {
192192+ maxPart = partNum
193193+ }
194194+ }
195195+196196+ // Check for missing parts
197197+ for i := 1; i <= maxPart; i++ {
198198+ if _, ok := s.Parts[i]; !ok {
199199+ return nil, 0, fmt.Errorf("missing part %d", i)
200200+ }
201201+ }
202202+203203+ // Assemble parts in order
204204+ assembled := make([]byte, 0, totalSize)
205205+ for i := 1; i <= maxPart; i++ {
206206+ part := s.Parts[i]
207207+ assembled = append(assembled, part.Data...)
208208+ }
209209+210210+ log.Printf("Assembled buffered parts: uploadID=%s, parts=%d, totalSize=%d bytes", s.UploadID, maxPart, totalSize)
211211+ return assembled, totalSize, nil
212212+}
213213+214214+// GetCompletedParts returns the list of completed parts for S3 multipart completion
215215+func (s *MultipartSession) GetCompletedParts() []CompletedPart {
216216+ s.mu.RLock()
217217+ defer s.mu.RUnlock()
218218+219219+ parts := make([]CompletedPart, 0, len(s.Parts))
220220+ for _, part := range s.Parts {
221221+ parts = append(parts, CompletedPart{
222222+ PartNumber: part.PartNumber,
223223+ ETag: part.ETag,
224224+ })
225225+ }
226226+227227+ return parts
228228+}
229229+230230+// StartMultipartUploadWithManager initiates a multipart upload using the manager
231231+// Returns uploadID and mode
232232+func (s *HoldService) StartMultipartUploadWithManager(ctx context.Context, digest string, manager *MultipartManager) (string, MultipartMode, error) {
233233+ // Check if presigned URLs are disabled for testing
234234+ if s.config.Server.DisablePresignedURLs {
235235+ log.Printf("Presigned URLs disabled (DISABLE_PRESIGNED_URLS=true), using buffered mode")
236236+ session := manager.CreateSession(digest, Buffered, "")
237237+ log.Printf("Started buffered multipart: uploadID=%s", session.UploadID)
238238+ return session.UploadID, Buffered, nil
239239+ }
240240+241241+ // Try S3 native multipart first
242242+ if s.s3Client != nil {
243243+ s3UploadID, err := s.startMultipartUpload(ctx, digest)
244244+ if err == nil {
245245+ // S3 native multipart succeeded
246246+ session := manager.CreateSession(digest, S3Native, s3UploadID)
247247+ log.Printf("Started S3 native multipart: uploadID=%s, s3UploadID=%s", session.UploadID, s3UploadID)
248248+ return session.UploadID, S3Native, nil
249249+ }
250250+ log.Printf("S3 native multipart failed, falling back to buffered mode: %v", err)
251251+ }
252252+253253+ // Fallback to buffered mode
254254+ session := manager.CreateSession(digest, Buffered, "")
255255+ log.Printf("Started buffered multipart: uploadID=%s", session.UploadID)
256256+ return session.UploadID, Buffered, nil
257257+}
258258+259259+// GetPartUploadURL generates a URL for uploading a part
260260+// For S3Native: returns presigned URL
261261+// For Buffered: returns proxy endpoint
262262+func (s *HoldService) GetPartUploadURL(ctx context.Context, session *MultipartSession, partNumber int, did string) (string, error) {
263263+ if session.Mode == S3Native {
264264+ // Generate S3 presigned URL for this part
265265+ url, err := s.getPartPresignedURL(ctx, session.Digest, session.S3UploadID, partNumber)
266266+ if err != nil {
267267+ return "", fmt.Errorf("failed to generate S3 part URL: %w", err)
268268+ }
269269+ return url, nil
270270+ }
271271+272272+ // Buffered mode: return proxy endpoint
273273+ url := fmt.Sprintf("%s/multipart-parts/%s/%d?did=%s",
274274+ s.config.Server.PublicURL, session.UploadID, partNumber, did)
275275+ return url, nil
276276+}
277277+278278+// CompleteMultipartUploadWithManager completes a multipart upload
279279+func (s *HoldService) CompleteMultipartUploadWithManager(ctx context.Context, session *MultipartSession, manager *MultipartManager) error {
280280+ defer manager.DeleteSession(session.UploadID)
281281+282282+ if session.Mode == S3Native {
283283+ // Complete S3 multipart upload
284284+ parts := session.GetCompletedParts()
285285+ if err := s.completeMultipartUpload(ctx, session.Digest, session.S3UploadID, parts); err != nil {
286286+ return fmt.Errorf("failed to complete S3 multipart: %w", err)
287287+ }
288288+ log.Printf("Completed S3 native multipart: uploadID=%s, parts=%d", session.UploadID, len(parts))
289289+ return nil
290290+ }
291291+292292+ // Buffered mode: assemble parts and write via driver
293293+ data, size, err := session.AssembleBufferedParts()
294294+ if err != nil {
295295+ return fmt.Errorf("failed to assemble parts: %w", err)
296296+ }
297297+298298+ // Write assembled blob to storage
299299+ path := blobPath(session.Digest)
300300+ writer, err := s.driver.Writer(ctx, path, false)
301301+ if err != nil {
302302+ return fmt.Errorf("failed to create writer: %w", err)
303303+ }
304304+305305+ written, err := writer.Write(data)
306306+ if err != nil {
307307+ writer.Cancel(ctx)
308308+ return fmt.Errorf("failed to write blob: %w", err)
309309+ }
310310+311311+ if err := writer.Commit(ctx); err != nil {
312312+ return fmt.Errorf("failed to commit blob: %w", err)
313313+ }
314314+315315+ log.Printf("Completed buffered multipart: uploadID=%s, size=%d bytes, written=%d", session.UploadID, size, written)
316316+ return nil
317317+}
318318+319319+// AbortMultipartUploadWithManager aborts a multipart upload
320320+func (s *HoldService) AbortMultipartUploadWithManager(ctx context.Context, session *MultipartSession, manager *MultipartManager) error {
321321+ defer manager.DeleteSession(session.UploadID)
322322+323323+ if session.Mode == S3Native {
324324+ // Abort S3 multipart upload
325325+ if err := s.abortMultipartUpload(ctx, session.Digest, session.S3UploadID); err != nil {
326326+ return fmt.Errorf("failed to abort S3 multipart: %w", err)
327327+ }
328328+ log.Printf("Aborted S3 native multipart: uploadID=%s", session.UploadID)
329329+ return nil
330330+ }
331331+332332+ // Buffered mode: just delete the session (parts are in memory)
333333+ log.Printf("Aborted buffered multipart: uploadID=%s", session.UploadID)
334334+ return nil
335335+}
336336+337337+// HandleMultipartPartUpload handles uploading a part in buffered mode
338338+// This is a new endpoint: PUT /multipart-parts/{uploadID}/{partNumber}
339339+func (s *HoldService) HandleMultipartPartUpload(w http.ResponseWriter, r *http.Request, uploadID string, partNumber int, did string, manager *MultipartManager) {
340340+ if r.Method != http.MethodPut {
341341+ http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
342342+ return
343343+ }
344344+345345+ // Get session
346346+ session, err := manager.GetSession(uploadID)
347347+ if err != nil {
348348+ http.Error(w, fmt.Sprintf("session not found: %v", err), http.StatusNotFound)
349349+ return
350350+ }
351351+352352+ // Verify authorization
353353+ if !s.isAuthorizedWrite(did) {
354354+ if did == "" {
355355+ http.Error(w, "unauthorized: authentication required", http.StatusUnauthorized)
356356+ } else {
357357+ http.Error(w, "forbidden: write access denied", http.StatusForbidden)
358358+ }
359359+ return
360360+ }
361361+362362+ // Verify session is in buffered mode
363363+ if session.Mode != Buffered {
364364+ http.Error(w, "session is not in buffered mode", http.StatusBadRequest)
365365+ return
366366+ }
367367+368368+ // Read part data
369369+ data, err := io.ReadAll(r.Body)
370370+ if err != nil {
371371+ http.Error(w, fmt.Sprintf("failed to read part data: %v", err), http.StatusInternalServerError)
372372+ return
373373+ }
374374+375375+ // Store part and get ETag
376376+ etag := session.StorePart(partNumber, data)
377377+378378+ // Return ETag in response
379379+ w.Header().Set("ETag", etag)
380380+ w.WriteHeader(http.StatusOK)
381381+}
+267
pkg/hold/registration.go
···11+package hold
22+33+import (
44+ "context"
55+ "encoding/json"
66+ "fmt"
77+ "log"
88+ "net/http"
99+ "net/url"
1010+ "strings"
1111+ "time"
1212+1313+ "atcr.io/pkg/atproto"
1414+ "atcr.io/pkg/auth/oauth"
1515+ "github.com/bluesky-social/indigo/atproto/identity"
1616+ "github.com/bluesky-social/indigo/atproto/syntax"
1717+)
1818+1919+// HealthHandler handles health check requests
2020+func (s *HoldService) HealthHandler(w http.ResponseWriter, r *http.Request) {
2121+ w.Header().Set("Content-Type", "application/json")
2222+ w.Write([]byte(`{"status":"ok"}`))
2323+}
2424+2525+// isHoldRegistered checks if a hold with the given public URL is already registered in the PDS
2626+func (s *HoldService) isHoldRegistered(ctx context.Context, did, pdsEndpoint, publicURL string) (bool, error) {
2727+ // We need to query the PDS without authentication to check public records
2828+ // ATProto records are publicly readable, so we can use an unauthenticated client
2929+ client := atproto.NewClient(pdsEndpoint, did, "")
3030+3131+ // List all hold records for this DID
3232+ records, err := client.ListRecords(ctx, atproto.HoldCollection, 100)
3333+ if err != nil {
3434+ return false, fmt.Errorf("failed to list hold records: %w", err)
3535+ }
3636+3737+ // Check if any hold record matches our public URL
3838+ for _, record := range records {
3939+ var holdRecord atproto.HoldRecord
4040+ if err := json.Unmarshal(record.Value, &holdRecord); err != nil {
4141+ continue
4242+ }
4343+4444+ if holdRecord.Endpoint == publicURL {
4545+ return true, nil
4646+ }
4747+ }
4848+4949+ return false, nil
5050+}
5151+5252+// AutoRegister registers this hold service in the owner's PDS
5353+// Checks if already registered first, then does OAuth if needed
5454+func (s *HoldService) AutoRegister(callbackHandler *http.HandlerFunc) error {
5555+ reg := &s.config.Registration
5656+ publicURL := s.config.Server.PublicURL
5757+5858+ if publicURL == "" {
5959+ return fmt.Errorf("HOLD_PUBLIC_URL not set")
6060+ }
6161+6262+ if reg.OwnerDID == "" {
6363+ return fmt.Errorf("HOLD_OWNER not set - required for registration")
6464+ }
6565+6666+ ctx := context.Background()
6767+6868+ log.Printf("Checking registration status for DID: %s", reg.OwnerDID)
6969+7070+ // Resolve DID to PDS endpoint using indigo
7171+ directory := identity.DefaultDirectory()
7272+ didParsed, err := syntax.ParseDID(reg.OwnerDID)
7373+ if err != nil {
7474+ return fmt.Errorf("invalid owner DID: %w", err)
7575+ }
7676+7777+ ident, err := directory.LookupDID(ctx, didParsed)
7878+ if err != nil {
7979+ return fmt.Errorf("failed to resolve PDS for DID: %w", err)
8080+ }
8181+8282+ pdsEndpoint := ident.PDSEndpoint()
8383+ if pdsEndpoint == "" {
8484+ return fmt.Errorf("no PDS endpoint found for DID")
8585+ }
8686+8787+ log.Printf("PDS endpoint: %s", pdsEndpoint)
8888+8989+ // Check if hold is already registered
9090+ isRegistered, err := s.isHoldRegistered(ctx, reg.OwnerDID, pdsEndpoint, publicURL)
9191+ if err != nil {
9292+ log.Printf("Warning: failed to check registration status: %v", err)
9393+ log.Printf("Proceeding with OAuth registration...")
9494+ } else if isRegistered {
9595+ log.Printf("✓ Hold service already registered in PDS")
9696+ log.Printf("Public URL: %s", publicURL)
9797+ return nil
9898+ }
9999+100100+ // Not registered, need to do OAuth
101101+ log.Printf("Hold not registered, starting OAuth flow...")
102102+103103+ // Get handle from DID document (already resolved above)
104104+ handle := ident.Handle.String()
105105+ if handle == "" || handle == "handle.invalid" {
106106+ return fmt.Errorf("no valid handle found for DID")
107107+ }
108108+109109+ log.Printf("Resolved handle: %s", handle)
110110+ log.Printf("Starting OAuth registration for hold service")
111111+ log.Printf("Public URL: %s", publicURL)
112112+113113+ return s.registerWithOAuth(publicURL, handle, reg.OwnerDID, pdsEndpoint, callbackHandler)
114114+}
115115+116116+// registerWithOAuth performs OAuth flow and registers the hold
117117+func (s *HoldService) registerWithOAuth(publicURL, handle, did, pdsEndpoint string, callbackHandler *http.HandlerFunc) error {
118118+ // Define the scopes we need for hold registration
119119+ holdScopes := []string{
120120+ "atproto",
121121+ fmt.Sprintf("repo:%s?action=create", atproto.HoldCollection),
122122+ fmt.Sprintf("repo:%s?action=update", atproto.HoldCollection),
123123+ fmt.Sprintf("repo:%s?action=create", atproto.HoldCrewCollection),
124124+ fmt.Sprintf("repo:%s?action=update", atproto.HoldCrewCollection),
125125+ fmt.Sprintf("repo:%s?action=create", atproto.SailorProfileCollection),
126126+ fmt.Sprintf("repo:%s?action=update", atproto.SailorProfileCollection),
127127+ }
128128+129129+ // Determine base URL based on mode
130130+ // Callback path standardized to /auth/oauth/callback across ATCR
131131+ var baseURL string
132132+133133+ if s.config.Server.TestMode {
134134+ // Test mode: Use localhost for OAuth (browser accessible) but store real URL in hold record
135135+ // Extract port from publicURL (e.g., "http://172.28.0.3:8080" -> ":8080")
136136+ parsedURL, err := url.Parse(publicURL)
137137+ if err != nil {
138138+ return fmt.Errorf("failed to parse public URL: %w", err)
139139+ }
140140+ port := parsedURL.Port()
141141+ if port == "" {
142142+ port = "8080" // default
143143+ }
144144+ baseURL = fmt.Sprintf("http://127.0.0.1:%s", port)
145145+ } else {
146146+ baseURL = publicURL
147147+ }
148148+149149+ // Run interactive OAuth flow with persistent server
150150+ ctx := context.Background()
151151+152152+ result, err := oauth.InteractiveFlowWithCallback(
153153+ ctx,
154154+ baseURL,
155155+ handle,
156156+ holdScopes, // Pass hold-specific scopes
157157+ func(handler http.HandlerFunc) error {
158158+ // Populate the pre-registered callback handler
159159+ *callbackHandler = handler
160160+ return nil
161161+ },
162162+ func(authURL string) error {
163163+ // Display OAuth URL for user to visit
164164+ log.Print("\n" + strings.Repeat("=", 80))
165165+ log.Printf("OAUTH AUTHORIZATION REQUIRED")
166166+ log.Print(strings.Repeat("=", 80))
167167+ log.Printf("\nPlease visit this URL to authorize the hold service:\n")
168168+ log.Printf(" %s\n", authURL)
169169+ log.Printf("Waiting for authorization...")
170170+ log.Print(strings.Repeat("=", 80) + "\n")
171171+ return nil
172172+ },
173173+ )
174174+ if err != nil {
175175+ return err
176176+ }
177177+178178+ log.Printf("Authorization received!")
179179+ log.Printf("OAuth session obtained successfully")
180180+ log.Printf("DID: %s", did)
181181+ log.Printf("PDS: %s", pdsEndpoint)
182182+183183+ // Create ATProto client with indigo's API client (handles DPoP automatically)
184184+ apiClient := result.Session.APIClient()
185185+ client := atproto.NewClientWithIndigoClient(pdsEndpoint, did, apiClient)
186186+187187+ return s.registerWithClient(publicURL, did, client)
188188+}
189189+190190+// registerWithClient registers the hold using an authenticated ATProto client
191191+func (s *HoldService) registerWithClient(publicURL, did string, client *atproto.Client) error {
192192+ // Derive hold name from URL (hostname)
193193+ holdName, err := extractHostname(publicURL)
194194+ if err != nil {
195195+ return fmt.Errorf("failed to extract hostname from URL: %w", err)
196196+ }
197197+198198+ log.Printf("Registering hold service: url=%s, name=%s, owner=%s", publicURL, holdName, did)
199199+200200+ ctx := context.Background()
201201+202202+ // Create HoldRecord
203203+ holdRecord := atproto.NewHoldRecord(publicURL, did, s.config.Server.Public)
204204+205205+ // Use hostname as record key
206206+ holdResult, err := client.PutRecord(ctx, atproto.HoldCollection, holdName, holdRecord)
207207+ if err != nil {
208208+ return fmt.Errorf("failed to create hold record: %w", err)
209209+ }
210210+211211+ log.Printf("✓ Created hold record: %s", holdResult.URI)
212212+213213+ // Create HoldCrewRecord for the owner
214214+ crewRecord := atproto.NewHoldCrewRecord(holdResult.URI, did, "owner")
215215+216216+ crewRKey := fmt.Sprintf("%s-%s", holdName, did)
217217+ crewResult, err := client.PutRecord(ctx, atproto.HoldCrewCollection, crewRKey, crewRecord)
218218+ if err != nil {
219219+ return fmt.Errorf("failed to create crew record: %w", err)
220220+ }
221221+222222+ log.Printf("✓ Created crew record: %s", crewResult.URI)
223223+224224+ // Update sailor profile to set this as the default hold
225225+ profile, err := atproto.GetProfile(ctx, client)
226226+ if err != nil {
227227+ log.Printf("Warning: failed to get sailor profile: %v", err)
228228+ } else {
229229+ if profile == nil {
230230+ // Create new profile with this hold as default
231231+ profile = atproto.NewSailorProfileRecord(publicURL)
232232+ } else {
233233+ // Update existing profile with new defaultHold
234234+ profile.DefaultHold = publicURL
235235+ profile.UpdatedAt = time.Now()
236236+ }
237237+238238+ err = atproto.UpdateProfile(ctx, client, profile)
239239+ if err != nil {
240240+ log.Printf("Warning: failed to update sailor profile: %v", err)
241241+ } else {
242242+ log.Printf("✓ Updated sailor profile defaultHold: %s", publicURL)
243243+ }
244244+ }
245245+246246+ log.Print("\n" + strings.Repeat("=", 80))
247247+ log.Printf("REGISTRATION COMPLETE")
248248+ log.Print(strings.Repeat("=", 80))
249249+ log.Printf("Hold service is now registered and ready to use!")
250250+ log.Print(strings.Repeat("=", 80) + "\n")
251251+252252+ return nil
253253+}
254254+255255+// extractHostname extracts the hostname from a URL to use as the hold name
256256+func extractHostname(urlStr string) (string, error) {
257257+ u, err := url.Parse(urlStr)
258258+ if err != nil {
259259+ return "", err
260260+ }
261261+ // Remove port if present
262262+ hostname := u.Hostname()
263263+ if hostname == "" {
264264+ return "", fmt.Errorf("no hostname in URL")
265265+ }
266266+ return hostname, nil
267267+}
+221
pkg/hold/s3.go
···11+package hold
22+33+import (
44+ "context"
55+ "fmt"
66+ "log"
77+ "sort"
88+ "strings"
99+ "time"
1010+1111+ "github.com/aws/aws-sdk-go/aws"
1212+ "github.com/aws/aws-sdk-go/aws/credentials"
1313+ "github.com/aws/aws-sdk-go/aws/session"
1414+ "github.com/aws/aws-sdk-go/service/s3"
1515+)
1616+1717+// initS3Client initializes the S3 client for presigned URL generation
1818+// Returns nil error if S3 client is successfully initialized
1919+// Returns error if storage is not S3 or if initialization fails (service will fall back to proxy mode)
2020+func (s *HoldService) initS3Client() error {
2121+ // Check if presigned URLs are explicitly disabled
2222+ if s.config.Server.DisablePresignedURLs {
2323+ log.Printf("⚠️ S3 presigned URLs DISABLED by config (DISABLE_PRESIGNED_URLS=true)")
2424+ log.Printf(" All uploads will use buffered mode (parts buffered in hold service)")
2525+ return nil // Not an error - just using buffered mode
2626+ }
2727+2828+ // Check if storage driver is S3
2929+ if s.config.Storage.Type() != "s3" {
3030+ log.Printf("Storage driver is %s (not S3), presigned URLs disabled", s.config.Storage.Type())
3131+ return nil // Not an error - just using different driver
3232+ }
3333+3434+ // Extract S3 configuration from storage parameters
3535+ params := s.config.Storage.Parameters()
3636+3737+ // Extract required S3 configuration
3838+ region, _ := params["region"].(string)
3939+ if region == "" {
4040+ region = "us-east-1" // Default region
4141+ }
4242+4343+ accessKey, _ := params["accesskey"].(string)
4444+ secretKey, _ := params["secretkey"].(string)
4545+ bucket, _ := params["bucket"].(string)
4646+4747+ if bucket == "" {
4848+ return fmt.Errorf("S3 bucket not configured")
4949+ }
5050+5151+ // Build AWS config
5252+ awsConfig := &aws.Config{
5353+ Region: aws.String(region),
5454+ }
5555+5656+ // Add credentials if provided (allow IAM role auth if not provided)
5757+ if accessKey != "" && secretKey != "" {
5858+ awsConfig.Credentials = credentials.NewStaticCredentials(accessKey, secretKey, "")
5959+ }
6060+6161+ // Add custom endpoint for S3-compatible services (Storj, MinIO, R2, etc.)
6262+ if endpoint, ok := params["regionendpoint"].(string); ok && endpoint != "" {
6363+ awsConfig.Endpoint = aws.String(endpoint)
6464+ awsConfig.S3ForcePathStyle = aws.Bool(true) // Required for MinIO, Storj
6565+ }
6666+6767+ // Create AWS session
6868+ sess, err := session.NewSession(awsConfig)
6969+ if err != nil {
7070+ return fmt.Errorf("failed to create AWS session: %w", err)
7171+ }
7272+7373+ // Create S3 client
7474+ s.s3Client = s3.New(sess)
7575+ s.bucket = bucket
7676+7777+ // Extract path prefix if configured (rootdirectory in S3 params)
7878+ if rootDir, ok := params["rootdirectory"].(string); ok && rootDir != "" {
7979+ s.s3PathPrefix = strings.TrimPrefix(rootDir, "/")
8080+ }
8181+8282+ log.Printf("✅ S3 presigned URLs enabled")
8383+8484+ return nil
8585+}
8686+8787+// startMultipartUpload initiates a multipart upload and returns upload ID
8888+func (s *HoldService) startMultipartUpload(ctx context.Context, digest string) (string, error) {
8989+ if s.s3Client == nil {
9090+ return "", fmt.Errorf("S3 not configured")
9191+ }
9292+9393+ path := blobPath(digest)
9494+ s3Key := strings.TrimPrefix(path, "/")
9595+ if s.s3PathPrefix != "" {
9696+ s3Key = s.s3PathPrefix + "/" + s3Key
9797+ }
9898+9999+ result, err := s.s3Client.CreateMultipartUploadWithContext(ctx, &s3.CreateMultipartUploadInput{
100100+ Bucket: aws.String(s.bucket),
101101+ Key: aws.String(s3Key),
102102+ })
103103+ if err != nil {
104104+ return "", err
105105+ }
106106+107107+ log.Printf("Started multipart upload: digest=%s, uploadID=%s", digest, *result.UploadId)
108108+ return *result.UploadId, nil
109109+}
110110+111111+// getPartPresignedURL generates presigned URL for a specific part
112112+func (s *HoldService) getPartPresignedURL(ctx context.Context, digest, uploadID string, partNumber int) (string, error) {
113113+ if s.s3Client == nil {
114114+ return "", fmt.Errorf("S3 not configured")
115115+ }
116116+117117+ path := blobPath(digest)
118118+ s3Key := strings.TrimPrefix(path, "/")
119119+ if s.s3PathPrefix != "" {
120120+ s3Key = s.s3PathPrefix + "/" + s3Key
121121+ }
122122+123123+ req, _ := s.s3Client.UploadPartRequest(&s3.UploadPartInput{
124124+ Bucket: aws.String(s.bucket),
125125+ Key: aws.String(s3Key),
126126+ UploadId: aws.String(uploadID),
127127+ PartNumber: aws.Int64(int64(partNumber)),
128128+ })
129129+130130+ url, err := req.Presign(15 * time.Minute)
131131+ if err != nil {
132132+ return "", err
133133+ }
134134+135135+ log.Printf("Generated part presigned URL: digest=%s, uploadID=%s, part=%d", digest, uploadID, partNumber)
136136+ return url, nil
137137+}
138138+139139+// normalizeETag ensures an ETag has quotes (required by S3 CompleteMultipartUpload)
140140+// S3 returns ETags with quotes, but HTTP clients may strip them
141141+func normalizeETag(etag string) string {
142142+ // Already has quotes
143143+ if strings.HasPrefix(etag, "\"") && strings.HasSuffix(etag, "\"") {
144144+ return etag
145145+ }
146146+ // Add quotes
147147+ return fmt.Sprintf("\"%s\"", etag)
148148+}
149149+150150+// completeMultipartUpload finalizes the multipart upload
151151+func (s *HoldService) completeMultipartUpload(ctx context.Context, digest, uploadID string, parts []CompletedPart) error {
152152+ if s.s3Client == nil {
153153+ return fmt.Errorf("S3 not configured")
154154+ }
155155+156156+ path := blobPath(digest)
157157+ s3Key := strings.TrimPrefix(path, "/")
158158+ if s.s3PathPrefix != "" {
159159+ s3Key = s.s3PathPrefix + "/" + s3Key
160160+ }
161161+162162+ // Sort parts by part number (S3 requires ascending order)
163163+ sort.Slice(parts, func(i, j int) bool {
164164+ return parts[i].PartNumber < parts[j].PartNumber
165165+ })
166166+167167+ // Convert to S3 CompletedPart format
168168+ // IMPORTANT: S3 requires ETags to be quoted in the CompleteMultipartUpload XML
169169+ s3Parts := make([]*s3.CompletedPart, len(parts))
170170+ for i, p := range parts {
171171+ etag := normalizeETag(p.ETag)
172172+ s3Parts[i] = &s3.CompletedPart{
173173+ PartNumber: aws.Int64(int64(p.PartNumber)),
174174+ ETag: aws.String(etag),
175175+ }
176176+ }
177177+178178+ _, err := s.s3Client.CompleteMultipartUploadWithContext(ctx, &s3.CompleteMultipartUploadInput{
179179+ Bucket: aws.String(s.bucket),
180180+ Key: aws.String(s3Key),
181181+ UploadId: aws.String(uploadID),
182182+ MultipartUpload: &s3.CompletedMultipartUpload{
183183+ Parts: s3Parts,
184184+ },
185185+ })
186186+187187+ if err != nil {
188188+ log.Printf("Failed to complete multipart upload: digest=%s, uploadID=%s, err=%v", digest, uploadID, err)
189189+ return err
190190+ }
191191+192192+ log.Printf("Completed multipart upload: digest=%s, uploadID=%s, parts=%d", digest, uploadID, len(parts))
193193+ return nil
194194+}
195195+196196+// abortMultipartUpload aborts an in-progress multipart upload
197197+func (s *HoldService) abortMultipartUpload(ctx context.Context, digest, uploadID string) error {
198198+ if s.s3Client == nil {
199199+ return fmt.Errorf("S3 not configured")
200200+ }
201201+202202+ path := blobPath(digest)
203203+ s3Key := strings.TrimPrefix(path, "/")
204204+ if s.s3PathPrefix != "" {
205205+ s3Key = s.s3PathPrefix + "/" + s3Key
206206+ }
207207+208208+ _, err := s.s3Client.AbortMultipartUploadWithContext(ctx, &s3.AbortMultipartUploadInput{
209209+ Bucket: aws.String(s.bucket),
210210+ Key: aws.String(s3Key),
211211+ UploadId: aws.String(uploadID),
212212+ })
213213+214214+ if err != nil {
215215+ log.Printf("Failed to abort multipart upload: digest=%s, uploadID=%s, err=%v", digest, uploadID, err)
216216+ return err
217217+ }
218218+219219+ log.Printf("Aborted multipart upload: digest=%s, uploadID=%s", digest, uploadID)
220220+ return nil
221221+}
+44
pkg/hold/service.go
···11+package hold
22+33+import (
44+ "context"
55+ "fmt"
66+ "log"
77+88+ "github.com/aws/aws-sdk-go/service/s3"
99+ storagedriver "github.com/distribution/distribution/v3/registry/storage/driver"
1010+ "github.com/distribution/distribution/v3/registry/storage/driver/factory"
1111+)
1212+1313+// HoldService provides presigned URLs for blob storage in a hold
1414+type HoldService struct {
1515+ driver storagedriver.StorageDriver
1616+ config *Config
1717+ s3Client *s3.S3 // S3 client for presigned URLs (nil if not S3 storage)
1818+ bucket string // S3 bucket name
1919+ s3PathPrefix string // S3 path prefix (if any)
2020+ MultipartMgr *MultipartManager // Exported for access in route handlers
2121+}
2222+2323+// NewHoldService creates a new hold service
2424+func NewHoldService(cfg *Config) (*HoldService, error) {
2525+ // Create storage driver from config
2626+ ctx := context.Background()
2727+ driver, err := factory.Create(ctx, cfg.Storage.Type(), cfg.Storage.Parameters())
2828+ if err != nil {
2929+ return nil, fmt.Errorf("failed to create storage driver: %w", err)
3030+ }
3131+3232+ service := &HoldService{
3333+ driver: driver,
3434+ config: cfg,
3535+ MultipartMgr: NewMultipartManager(),
3636+ }
3737+3838+ // Initialize S3 client for presigned URLs (if using S3 storage)
3939+ if err := service.initS3Client(); err != nil {
4040+ log.Printf("WARNING: S3 presigned URLs disabled: %v", err)
4141+ }
4242+4343+ return service, nil
4444+}
+115
pkg/hold/storage.go
···11+package hold
22+33+import (
44+ "context"
55+ "fmt"
66+ "log"
77+ "strings"
88+ "time"
99+1010+ "github.com/aws/aws-sdk-go/aws"
1111+ "github.com/aws/aws-sdk-go/service/s3"
1212+)
1313+1414+// blobPath converts a digest (e.g., "sha256:abc123...") or temp path to a storage path
1515+// Distribution stores blobs as: /docker/registry/v2/blobs/{algorithm}/{xx}/{hash}/data
1616+// where xx is the first 2 characters of the hash for directory sharding
1717+// NOTE: Path must start with / for filesystem driver
1818+func blobPath(digest string) string {
1919+ // Handle temp paths (start with uploads/temp-)
2020+ if strings.HasPrefix(digest, "uploads/temp-") {
2121+ return fmt.Sprintf("/docker/registry/v2/%s/data", digest)
2222+ }
2323+2424+ // Split digest into algorithm and hash
2525+ parts := strings.SplitN(digest, ":", 2)
2626+ if len(parts) != 2 {
2727+ // Fallback for malformed digest
2828+ return fmt.Sprintf("/docker/registry/v2/blobs/%s/data", digest)
2929+ }
3030+3131+ algorithm := parts[0]
3232+ hash := parts[1]
3333+3434+ // Use first 2 characters for sharding
3535+ if len(hash) < 2 {
3636+ return fmt.Sprintf("/docker/registry/v2/blobs/%s/%s/data", algorithm, hash)
3737+ }
3838+3939+ return fmt.Sprintf("/docker/registry/v2/blobs/%s/%s/%s/data", algorithm, hash[:2], hash)
4040+}
4141+4242+// getPresignedURL generates a presigned URL for GET, HEAD, or PUT operations
4343+func (s *HoldService) getPresignedURL(ctx context.Context, operation PresignedURLOperation, digest string, did string) (string, error) {
4444+ path := blobPath(digest)
4545+4646+ // Check blob exists for GET/HEAD operations (not for PUT since blob doesn't exist yet)
4747+ if operation == OperationGet || operation == OperationHead {
4848+ if _, err := s.driver.Stat(ctx, path); err != nil {
4949+ return "", fmt.Errorf("blob not found: %w", err)
5050+ }
5151+ }
5252+5353+ // Check if presigned URLs are disabled
5454+ if s.config.Server.DisablePresignedURLs {
5555+ log.Printf("Presigned URLs disabled, using proxy URL")
5656+ return s.getProxyURL(digest, did), nil
5757+ }
5858+5959+ // Generate presigned URL if S3 client is available
6060+ if s.s3Client != nil {
6161+ // Build S3 key from blob path
6262+ s3Key := strings.TrimPrefix(path, "/")
6363+ if s.s3PathPrefix != "" {
6464+ s3Key = s.s3PathPrefix + "/" + s3Key
6565+ }
6666+6767+ // Create appropriate S3 request based on operation
6868+ var req interface {
6969+ Presign(time.Duration) (string, error)
7070+ }
7171+ switch operation {
7272+ case OperationGet:
7373+ // Note: Don't use ResponseContentType - not supported by all S3-compatible services
7474+ req, _ = s.s3Client.GetObjectRequest(&s3.GetObjectInput{
7575+ Bucket: aws.String(s.bucket),
7676+ Key: aws.String(s3Key),
7777+ })
7878+7979+ case OperationHead:
8080+ req, _ = s.s3Client.HeadObjectRequest(&s3.HeadObjectInput{
8181+ Bucket: aws.String(s.bucket),
8282+ Key: aws.String(s3Key),
8383+ })
8484+8585+ case OperationPut:
8686+ req, _ = s.s3Client.PutObjectRequest(&s3.PutObjectInput{
8787+ Bucket: aws.String(s.bucket),
8888+ Key: aws.String(s3Key),
8989+ ContentType: aws.String("application/octet-stream"),
9090+ })
9191+9292+ default:
9393+ return "", fmt.Errorf("unsupported operation: %s", operation)
9494+ }
9595+9696+ // Generate presigned URL with 15 minute expiry
9797+ url, err := req.Presign(15 * time.Minute)
9898+ if err != nil {
9999+ log.Printf("[getPresignedURL] Presign FAILED for %s: %v", operation, err)
100100+ log.Printf(" Falling back to proxy URL")
101101+ return s.getProxyURL(digest, did), nil
102102+ }
103103+104104+ return url, nil
105105+ }
106106+107107+ // Fallback: return proxy URL through this service
108108+ return s.getProxyURL(digest, did), nil
109109+}
110110+111111+// getProxyURL returns a proxy URL for blob operations (fallback when presigned URLs unavailable)
112112+func (s *HoldService) getProxyURL(digest, did string) string {
113113+ // All operations use the same proxy endpoint
114114+ return fmt.Sprintf("%s/blobs/%s?did=%s", s.config.Server.PublicURL, digest, did)
115115+}