From af815fbc7d590b31b9d0bf544749d152d16e79db Mon Sep 17 00:00:00 2001 From: Evan Jarrett Date: Sun, 4 Jan 2026 22:39:48 -0600 Subject: [PATCH] use for range and wg.Go --- pkg/appview/config.go | 3 +- pkg/appview/db/schema.go | 7 +- pkg/appview/handlers/opengraph.go | 2 +- pkg/appview/handlers/repository.go | 25 +++-- pkg/appview/holdhealth/worker.go | 21 ++--- pkg/appview/licenses/licenses.go | 4 +- pkg/appview/middleware/auth_test.go | 13 +-- pkg/appview/storage/profile_test.go | 6 +- pkg/appview/storage/routing_repository.go | 91 +++++++++---------- .../storage/routing_repository_test.go | 16 ++-- pkg/atproto/directory_test.go | 14 +-- pkg/auth/token/issuer_test.go | 12 +-- pkg/hold/oci/helpers_test.go | 90 ------------------ pkg/hold/oci/multipart.go | 58 +++--------- pkg/hold/pds/xrpc.go | 4 +- 15 files changed, 107 insertions(+), 259 deletions(-) diff --git a/pkg/appview/config.go b/pkg/appview/config.go index 04c8773..f26d628 100644 --- a/pkg/appview/config.go +++ b/pkg/appview/config.go @@ -388,8 +388,7 @@ func parseChecksums(checksumsStr string) map[string]string { return checksums } - pairs := strings.Split(checksumsStr, ",") - for _, pair := range pairs { + for pair := range strings.SplitSeq(checksumsStr, ",") { parts := strings.SplitN(strings.TrimSpace(pair), ":", 2) if len(parts) == 2 { platform := strings.TrimSpace(parts[0]) diff --git a/pkg/appview/db/schema.go b/pkg/appview/db/schema.go index b08677b..bc3f92a 100644 --- a/pkg/appview/db/schema.go +++ b/pkg/appview/db/schema.go @@ -225,9 +225,7 @@ func splitSQLStatements(query string) []string { var statements []string // Split on semicolons - parts := strings.Split(query, ";") - - for _, part := range parts { + for part := range strings.SplitSeq(query, ";") { // Trim whitespace stmt := strings.TrimSpace(part) @@ -237,9 +235,8 @@ func splitSQLStatements(query string) []string { } // Skip comment-only statements - lines := strings.Split(stmt, "\n") hasCode := false - for _, line := range lines { + for line := range strings.SplitSeq(stmt, "\n") { trimmed := strings.TrimSpace(line) if trimmed != "" && !strings.HasPrefix(trimmed, "--") { hasCode = true diff --git a/pkg/appview/handlers/opengraph.go b/pkg/appview/handlers/opengraph.go index ce1c210..68b8010 100644 --- a/pkg/appview/handlers/opengraph.go +++ b/pkg/appview/handlers/opengraph.go @@ -105,7 +105,7 @@ func (h *RepoOGHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { if licenses != "" { // Show first license if multiple - license := strings.Split(licenses, ",")[0] + license, _, _ := strings.Cut(licenses, ",") license = strings.TrimSpace(license) card.DrawBadge(license, badgeX, badgeY, ogcard.FontBadge, ogcard.ColorBadgeBg, ogcard.ColorText) } diff --git a/pkg/appview/handlers/repository.go b/pkg/appview/handlers/repository.go index bfb417c..bc48f63 100644 --- a/pkg/appview/handlers/repository.go +++ b/pkg/appview/handlers/repository.go @@ -89,17 +89,14 @@ func (h *RepositoryPageHandler) ServeHTTP(w http.ResponseWriter, r *http.Request continue } - wg.Add(1) - go func(idx int) { - defer wg.Done() - - endpoint := manifests[idx].HoldEndpoint + wg.Go(func() { + endpoint := manifests[i].HoldEndpoint // Try to get cached status first (instant) if cached := h.HealthChecker.GetCachedStatus(endpoint); cached != nil { mu.Lock() - manifests[idx].Reachable = cached.Reachable - manifests[idx].Pending = false + manifests[i].Reachable = cached.Reachable + manifests[i].Pending = false mu.Unlock() return } @@ -110,19 +107,19 @@ func (h *RepositoryPageHandler) ServeHTTP(w http.ResponseWriter, r *http.Request mu.Lock() if ctx.Err() == context.DeadlineExceeded { // Timeout - mark as pending for HTMX polling - manifests[idx].Reachable = false - manifests[idx].Pending = true + manifests[i].Reachable = false + manifests[i].Pending = true } else if err != nil { // Error - mark as unreachable - manifests[idx].Reachable = false - manifests[idx].Pending = false + manifests[i].Reachable = false + manifests[i].Pending = false } else { // Success - manifests[idx].Reachable = reachable - manifests[idx].Pending = false + manifests[i].Reachable = reachable + manifests[i].Pending = false } mu.Unlock() - }(i) + }) } // Wait for all checks to complete or timeout diff --git a/pkg/appview/holdhealth/worker.go b/pkg/appview/holdhealth/worker.go index 9b2d24c..4e559e5 100644 --- a/pkg/appview/holdhealth/worker.go +++ b/pkg/appview/holdhealth/worker.go @@ -53,10 +53,7 @@ func NewWorkerWithStartupDelay(checker *Checker, db DBQuerier, refreshInterval, // Start begins the background worker func (w *Worker) Start(ctx context.Context) { - w.wg.Add(1) - go func() { - defer w.wg.Done() - + w.wg.Go(func() { slog.Info("Hold health worker starting background health checks") // Wait for services to be ready (Docker startup race condition) @@ -89,7 +86,7 @@ func (w *Worker) Start(ctx context.Context) { w.checker.Cleanup() } } - }() + }) } // Stop gracefully stops the worker @@ -154,20 +151,16 @@ func (w *Worker) refreshAllHolds(ctx context.Context) { var statsMu sync.Mutex for _, endpoint := range uniqueEndpoints { - wg.Add(1) - - go func(ep string) { - defer wg.Done() - + wg.Go(func() { // Acquire semaphore sem <- struct{}{} defer func() { <-sem }() // Check health - isReachable, err := w.checker.CheckHealth(ctx, ep) + isReachable, err := w.checker.CheckHealth(ctx, endpoint) // Update cache - w.checker.SetStatus(ep, isReachable, err) + w.checker.SetStatus(endpoint, isReachable, err) // Update stats statsMu.Lock() @@ -175,10 +168,10 @@ func (w *Worker) refreshAllHolds(ctx context.Context) { reachable++ } else { unreachable++ - slog.Warn("Hold health worker hold unreachable", "endpoint", ep, "error", err) + slog.Warn("Hold health worker hold unreachable", "endpoint", endpoint, "error", err) } statsMu.Unlock() - }(endpoint) + }) } // Wait for all checks to complete diff --git a/pkg/appview/licenses/licenses.go b/pkg/appview/licenses/licenses.go index aaa7d2a..0ce6ee9 100644 --- a/pkg/appview/licenses/licenses.go +++ b/pkg/appview/licenses/licenses.go @@ -129,12 +129,10 @@ func ParseLicenses(licensesStr string) []LicenseInfo { licensesStr = strings.ReplaceAll(licensesStr, " OR ", ",") licensesStr = strings.ReplaceAll(licensesStr, ";", ",") - parts := strings.Split(licensesStr, ",") - var result []LicenseInfo seen := make(map[string]bool) // Deduplicate - for _, part := range parts { + for part := range strings.SplitSeq(licensesStr, ",") { part = strings.TrimSpace(part) if part == "" { continue diff --git a/pkg/appview/middleware/auth_test.go b/pkg/appview/middleware/auth_test.go index c85a54e..3704bc6 100644 --- a/pkg/appview/middleware/auth_test.go +++ b/pkg/appview/middleware/auth_test.go @@ -358,24 +358,21 @@ func TestMiddleware_ConcurrentAccess(t *testing.T) { var wg sync.WaitGroup var mu sync.Mutex // Protect results map - for i := range 10 { - wg.Add(1) - go func(index int, sessionID string) { - defer wg.Done() - + for i := range results { + wg.Go(func() { req := httptest.NewRequest("GET", "/test", nil) req.AddCookie(&http.Cookie{ Name: "atcr_session", - Value: sessionID, + Value: sessionIDs[i], }) w := httptest.NewRecorder() wrappedHandler.ServeHTTP(w, req) mu.Lock() - results[index] = w.Code + results[i] = w.Code mu.Unlock() - }(i, sessionIDs[i]) + }) } wg.Wait() diff --git a/pkg/appview/storage/profile_test.go b/pkg/appview/storage/profile_test.go index 8c68db5..31410f5 100644 --- a/pkg/appview/storage/profile_test.go +++ b/pkg/appview/storage/profile_test.go @@ -341,14 +341,12 @@ func TestGetProfile_MigrationLocking(t *testing.T) { // Make 5 concurrent GetProfile calls var wg sync.WaitGroup for range 5 { - wg.Add(1) - go func() { - defer wg.Done() + wg.Go(func() { _, err := GetProfile(context.Background(), client) if err != nil { t.Errorf("GetProfile() error = %v", err) } - }() + }) } wg.Wait() diff --git a/pkg/appview/storage/routing_repository.go b/pkg/appview/storage/routing_repository.go index deba6b3..ec2aa2f 100644 --- a/pkg/appview/storage/routing_repository.go +++ b/pkg/appview/storage/routing_repository.go @@ -7,6 +7,7 @@ package storage import ( "context" "log/slog" + "sync" "github.com/distribution/distribution/v3" ) @@ -18,12 +19,13 @@ const HTTPRequestMethod contextKey = "http.request.method" // RoutingRepository routes manifests to ATProto and blobs to external hold service // The registry (AppView) is stateless and NEVER stores blobs locally // NOTE: A fresh instance is created per-request (see middleware/registry.go) -// so no mutex is needed - each request has its own instance type RoutingRepository struct { distribution.Repository - Ctx *RegistryContext // All context and services (exported for token updates) - manifestStore *ManifestStore // Manifest store instance (lazy-initialized) - blobStore *ProxyBlobStore // Blob store instance (lazy-initialized) + Ctx *RegistryContext // All context and services (exported for token updates) + manifestStore *ManifestStore // Manifest store instance (lazy-initialized) + manifestStoreOnce sync.Once // Ensures thread-safe lazy initialization + blobStore *ProxyBlobStore // Blob store instance (lazy-initialized) + blobStoreOnce sync.Once // Ensures thread-safe lazy initialization } // NewRoutingRepository creates a new routing repository @@ -36,63 +38,58 @@ func NewRoutingRepository(baseRepo distribution.Repository, ctx *RegistryContext // Manifests returns the ATProto-backed manifest service func (r *RoutingRepository) Manifests(ctx context.Context, options ...distribution.ManifestServiceOption) (distribution.ManifestService, error) { - // Lazy-initialize manifest store (no mutex needed - one instance per request) - if r.manifestStore == nil { + r.manifestStoreOnce.Do(func() { // Ensure blob store is created first (needed for label extraction during push) blobStore := r.Blobs(ctx) r.manifestStore = NewManifestStore(r.Ctx, blobStore) - } + }) return r.manifestStore, nil } // Blobs returns a proxy blob store that routes to external hold service // The registry (AppView) NEVER stores blobs locally - all blobs go through hold service func (r *RoutingRepository) Blobs(ctx context.Context) distribution.BlobStore { - // Return cached blob store if available (no mutex needed - one instance per request) - if r.blobStore != nil { - slog.Debug("Returning cached blob store", "component", "storage/blobs", "did", r.Ctx.DID, "repo", r.Ctx.Repository) - return r.blobStore - } - - // Determine if this is a pull (GET/HEAD) or push (PUT/POST/etc) operation - // Pull operations use the historical hold DID from the database (blobs are where they were pushed) - // Push operations use the discovery-based hold DID from user's profile/default - // This allows users to change their default hold and have new pushes go there - isPull := false - if method, ok := ctx.Value(HTTPRequestMethod).(string); ok { - isPull = method == "GET" || method == "HEAD" - } - - holdDID := r.Ctx.HoldDID // Default to discovery-based DID - holdSource := "discovery" - - // Only query database for pull operations - if isPull && r.Ctx.Database != nil { - // Query database for the latest manifest's hold DID - if dbHoldDID, err := r.Ctx.Database.GetLatestHoldDIDForRepo(r.Ctx.DID, r.Ctx.Repository); err == nil && dbHoldDID != "" { - // Use hold DID from database (pull case - use historical reference) - holdDID = dbHoldDID - holdSource = "database" - slog.Debug("Using hold from database manifest (pull)", "component", "storage/blobs", "did", r.Ctx.DID, "repo", r.Ctx.Repository, "hold", dbHoldDID) - } else if err != nil { - // Log error but don't fail - fall back to discovery-based DID - slog.Warn("Failed to query database for hold DID", "component", "storage/blobs", "error", err) + r.blobStoreOnce.Do(func() { + // Determine if this is a pull (GET/HEAD) or push (PUT/POST/etc) operation + // Pull operations use the historical hold DID from the database (blobs are where they were pushed) + // Push operations use the discovery-based hold DID from user's profile/default + // This allows users to change their default hold and have new pushes go there + isPull := false + if method, ok := ctx.Value(HTTPRequestMethod).(string); ok { + isPull = method == "GET" || method == "HEAD" } - // If dbHoldDID is empty (no manifests yet), fall through to use discovery-based DID - } - if holdDID == "" { - // This should never happen if middleware is configured correctly - panic("hold DID not set in RegistryContext - ensure default_hold_did is configured in middleware") - } + holdDID := r.Ctx.HoldDID // Default to discovery-based DID + holdSource := "discovery" - slog.Debug("Using hold DID for blobs", "component", "storage/blobs", "did", r.Ctx.DID, "repo", r.Ctx.Repository, "hold", holdDID, "source", holdSource) + // Only query database for pull operations + if isPull && r.Ctx.Database != nil { + // Query database for the latest manifest's hold DID + if dbHoldDID, err := r.Ctx.Database.GetLatestHoldDIDForRepo(r.Ctx.DID, r.Ctx.Repository); err == nil && dbHoldDID != "" { + // Use hold DID from database (pull case - use historical reference) + holdDID = dbHoldDID + holdSource = "database" + slog.Debug("Using hold from database manifest (pull)", "component", "storage/blobs", "did", r.Ctx.DID, "repo", r.Ctx.Repository, "hold", dbHoldDID) + } else if err != nil { + // Log error but don't fail - fall back to discovery-based DID + slog.Warn("Failed to query database for hold DID", "component", "storage/blobs", "error", err) + } + // If dbHoldDID is empty (no manifests yet), fall through to use discovery-based DID + } - // Update context with the correct hold DID (may be from database or discovered) - r.Ctx.HoldDID = holdDID + if holdDID == "" { + // This should never happen if middleware is configured correctly + panic("hold DID not set in RegistryContext - ensure default_hold_did is configured in middleware") + } - // Create and cache proxy blob store - r.blobStore = NewProxyBlobStore(r.Ctx) + slog.Debug("Using hold DID for blobs", "component", "storage/blobs", "did", r.Ctx.DID, "repo", r.Ctx.Repository, "hold", holdDID, "source", holdSource) + + // Update context with the correct hold DID (may be from database or discovered) + r.Ctx.HoldDID = holdDID + + // Create and cache proxy blob store + r.blobStore = NewProxyBlobStore(r.Ctx) + }) return r.blobStore } diff --git a/pkg/appview/storage/routing_repository_test.go b/pkg/appview/storage/routing_repository_test.go index 9206062..8933878 100644 --- a/pkg/appview/storage/routing_repository_test.go +++ b/pkg/appview/storage/routing_repository_test.go @@ -318,13 +318,11 @@ func TestRoutingRepository_ConcurrentAccess(t *testing.T) { // Concurrent access to Manifests() for i := 0; i < numGoroutines; i++ { - wg.Add(1) - go func(index int) { - defer wg.Done() + wg.Go(func() { store, err := repo.Manifests(context.Background()) require.NoError(t, err) - manifestStores[index] = store - }(i) + manifestStores[i] = store + }) } wg.Wait() @@ -341,11 +339,9 @@ func TestRoutingRepository_ConcurrentAccess(t *testing.T) { // Concurrent access to Blobs() for i := 0; i < numGoroutines; i++ { - wg.Add(1) - go func(index int) { - defer wg.Done() - blobStores[index] = repo.Blobs(context.Background()) - }(i) + wg.Go(func() { + blobStores[i] = repo.Blobs(context.Background()) + }) } wg.Wait() diff --git a/pkg/atproto/directory_test.go b/pkg/atproto/directory_test.go index 96d6bdb..98f3260 100644 --- a/pkg/atproto/directory_test.go +++ b/pkg/atproto/directory_test.go @@ -29,18 +29,16 @@ func TestGetDirectoryConcurrency(t *testing.T) { t.Run("concurrent access is thread-safe", func(t *testing.T) { const numGoroutines = 100 var wg sync.WaitGroup - wg.Add(numGoroutines) // Channel to collect all directory instances instances := make(chan any, numGoroutines) // Launch many goroutines concurrently accessing GetDirectory for range numGoroutines { - go func() { - defer wg.Done() + wg.Go(func() { dir := GetDirectory() instances <- dir - }() + }) } // Wait for all goroutines to complete @@ -120,20 +118,18 @@ func TestGetDirectoryRaceConditions(t *testing.T) { const numGoroutines = 50 var wg sync.WaitGroup - wg.Add(numGoroutines) instances := make([]any, numGoroutines) var mu sync.Mutex // Simulate many goroutines trying to get the directory simultaneously for i := 0; i < numGoroutines; i++ { - go func(idx int) { - defer wg.Done() + wg.Go(func() { dir := GetDirectory() mu.Lock() - instances[idx] = dir + instances[i] = dir mu.Unlock() - }(i) + }) } wg.Wait() diff --git a/pkg/auth/token/issuer_test.go b/pkg/auth/token/issuer_test.go index 31bb200..cd59a6e 100644 --- a/pkg/auth/token/issuer_test.go +++ b/pkg/auth/token/issuer_test.go @@ -378,19 +378,17 @@ func TestIssuer_ConcurrentIssue(t *testing.T) { // Issue tokens concurrently const numGoroutines = 10 var wg sync.WaitGroup - wg.Add(numGoroutines) tokens := make([]string, numGoroutines) errors := make([]error, numGoroutines) for i := 0; i < numGoroutines; i++ { - go func(idx int) { - defer wg.Done() - subject := "did:plc:user" + string(rune('0'+idx)) + wg.Go(func() { + subject := "did:plc:user" + string(rune('0'+i)) token, err := issuer.Issue(subject, nil, AuthMethodOAuth) - tokens[idx] = token - errors[idx] = err - }(i) + tokens[i] = token + errors[i] = err + }) } wg.Wait() diff --git a/pkg/hold/oci/helpers_test.go b/pkg/hold/oci/helpers_test.go index b4914f4..500f34d 100644 --- a/pkg/hold/oci/helpers_test.go +++ b/pkg/hold/oci/helpers_test.go @@ -5,96 +5,6 @@ import ( ) // Tests for helper functions - -func TestBlobPath_SHA256(t *testing.T) { - tests := []struct { - name string - digest string - expected string - }{ - { - name: "standard sha256 digest", - digest: "sha256:abc123def456", - expected: "/docker/registry/v2/blobs/sha256/ab/abc123def456/data", - }, - { - name: "short hash (less than 2 chars)", - digest: "sha256:a", - expected: "/docker/registry/v2/blobs/sha256/a/data", - }, - { - name: "exactly 2 char hash", - digest: "sha256:ab", - expected: "/docker/registry/v2/blobs/sha256/ab/ab/data", - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - result := blobPath(tt.digest) - if result != tt.expected { - t.Errorf("Expected %s, got %s", tt.expected, result) - } - }) - } -} - -func TestBlobPath_TempUpload(t *testing.T) { - tests := []struct { - name string - digest string - expected string - }{ - { - name: "temp upload path", - digest: "uploads/temp-uuid-123", - expected: "/docker/registry/v2/uploads/temp-uuid-123/data", - }, - { - name: "temp upload with different uuid", - digest: "uploads/temp-abc-def-456", - expected: "/docker/registry/v2/uploads/temp-abc-def-456/data", - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - result := blobPath(tt.digest) - if result != tt.expected { - t.Errorf("Expected %s, got %s", tt.expected, result) - } - }) - } -} - -func TestBlobPath_MalformedDigest(t *testing.T) { - tests := []struct { - name string - digest string - expected string - }{ - { - name: "no colon in digest", - digest: "malformed-digest", - expected: "/docker/registry/v2/blobs/malformed-digest/data", - }, - { - name: "empty digest", - digest: "", - expected: "/docker/registry/v2/blobs//data", - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - result := blobPath(tt.digest) - if result != tt.expected { - t.Errorf("Expected %s, got %s", tt.expected, result) - } - }) - } -} - func TestNormalizeETag(t *testing.T) { tests := []struct { name string diff --git a/pkg/hold/oci/multipart.go b/pkg/hold/oci/multipart.go index a2d066a..fe1d14b 100644 --- a/pkg/hold/oci/multipart.go +++ b/pkg/hold/oci/multipart.go @@ -12,7 +12,8 @@ import ( "time" "atcr.io/pkg/atproto" - "github.com/aws/aws-sdk-go/service/s3" + "atcr.io/pkg/s3" + awss3 "github.com/aws/aws-sdk-go/service/s3" "github.com/google/uuid" ) @@ -237,13 +238,13 @@ func (h *XRPCHandler) StartMultipartUploadWithManager(ctx context.Context, diges if h.s3Service.Client == nil { return "", S3Native, fmt.Errorf("S3 not configured") } - path := blobPath(digest) + path := s3.BlobPath(digest) s3Key := strings.TrimPrefix(path, "/") if h.s3Service.PathPrefix != "" { s3Key = h.s3Service.PathPrefix + "/" + s3Key } - result, err := h.s3Service.Client.CreateMultipartUploadWithContext(ctx, &s3.CreateMultipartUploadInput{ + result, err := h.s3Service.Client.CreateMultipartUploadWithContext(ctx, &awss3.CreateMultipartUploadInput{ Bucket: &h.s3Service.Bucket, Key: &s3Key, }) @@ -280,13 +281,13 @@ func (h *XRPCHandler) GetPartUploadURL(ctx context.Context, uploadID string, par return nil, fmt.Errorf("S3 not configured") } - path := blobPath(session.Digest) + path := s3.BlobPath(session.Digest) s3Key := strings.TrimPrefix(path, "/") if h.s3Service.PathPrefix != "" { s3Key = h.s3Service.PathPrefix + "/" + s3Key } pnum := int64(partNumber) - req, _ := h.s3Service.Client.UploadPartRequest(&s3.UploadPartInput{ + req, _ := h.s3Service.Client.UploadPartRequest(&awss3.UploadPartInput{ Bucket: &h.s3Service.Bucket, Key: &s3Key, UploadId: &session.S3UploadID, @@ -342,26 +343,26 @@ func (h *XRPCHandler) CompleteMultipartUploadWithManager(ctx context.Context, up // Convert to S3 CompletedPart format // IMPORTANT: S3 requires ETags to be quoted in the CompleteMultipartUpload XML - s3Parts := make([]*s3.CompletedPart, len(parts)) + s3Parts := make([]*awss3.CompletedPart, len(parts)) for i, p := range parts { etag := normalizeETag(p.ETag) pnum := int64(p.PartNumber) - s3Parts[i] = &s3.CompletedPart{ + s3Parts[i] = &awss3.CompletedPart{ PartNumber: &pnum, ETag: &etag, } } - sourcePath := blobPath(session.Digest) + sourcePath := s3.BlobPath(session.Digest) s3Key := strings.TrimPrefix(sourcePath, "/") if h.s3Service.PathPrefix != "" { s3Key = h.s3Service.PathPrefix + "/" + s3Key } - _, err = h.s3Service.Client.CompleteMultipartUploadWithContext(ctx, &s3.CompleteMultipartUploadInput{ + _, err = h.s3Service.Client.CompleteMultipartUploadWithContext(ctx, &awss3.CompleteMultipartUploadInput{ Bucket: &h.s3Service.Bucket, Key: &s3Key, UploadId: &session.S3UploadID, - MultipartUpload: &s3.CompletedMultipartUpload{ + MultipartUpload: &awss3.CompletedMultipartUpload{ Parts: s3Parts, }, }) @@ -374,7 +375,7 @@ func (h *XRPCHandler) CompleteMultipartUploadWithManager(ctx context.Context, up "parts", len(s3Parts)) // Verify the blob exists at temp location before moving - destPath := blobPath(finalDigest) + destPath := s3.BlobPath(finalDigest) slog.Debug("About to move blob", "source", sourcePath, "dest", destPath) @@ -412,7 +413,7 @@ func (h *XRPCHandler) CompleteMultipartUploadWithManager(ctx context.Context, up } // Write assembled blob to final digest location (not temp) - path := blobPath(finalDigest) + path := s3.BlobPath(finalDigest) writer, err := h.driver.Writer(ctx, path, false) if err != nil { return fmt.Errorf("failed to create writer: %w", err) @@ -448,13 +449,13 @@ func (h *XRPCHandler) AbortMultipartUploadWithManager(ctx context.Context, uploa if h.s3Service.Client == nil { return fmt.Errorf("S3 not configured") } - path := blobPath(session.Digest) + path := s3.BlobPath(session.Digest) s3Key := strings.TrimPrefix(path, "/") if h.s3Service.PathPrefix != "" { s3Key = h.s3Service.PathPrefix + "/" + s3Key } - _, err := h.s3Service.Client.AbortMultipartUploadWithContext(ctx, &s3.AbortMultipartUploadInput{ + _, err := h.s3Service.Client.AbortMultipartUploadWithContext(ctx, &awss3.AbortMultipartUploadInput{ Bucket: &h.s3Service.Bucket, Key: &s3Key, UploadId: &session.S3UploadID, @@ -499,32 +500,3 @@ func normalizeETag(etag string) string { // Add quotes return fmt.Sprintf("\"%s\"", etag) } - -// blobPath converts a digest (e.g., "sha256:abc123...") or temp path to a storage path -// Distribution stores blobs as: /docker/registry/v2/blobs/{algorithm}/{xx}/{hash}/data -// where xx is the first 2 characters of the hash for directory sharding -// NOTE: Path must start with / for filesystem driver -// This is used for OCI container layers (content-addressed, globally deduplicated) -func blobPath(digest string) string { - // Handle temp paths (start with uploads/temp-) - if strings.HasPrefix(digest, "uploads/temp-") { - return fmt.Sprintf("/docker/registry/v2/%s/data", digest) - } - - // Split digest into algorithm and hash - parts := strings.SplitN(digest, ":", 2) - if len(parts) != 2 { - // Fallback for malformed digest - return fmt.Sprintf("/docker/registry/v2/blobs/%s/data", digest) - } - - algorithm := parts[0] - hash := parts[1] - - // Use first 2 characters for sharding - if len(hash) < 2 { - return fmt.Sprintf("/docker/registry/v2/blobs/%s/%s/data", algorithm, hash) - } - - return fmt.Sprintf("/docker/registry/v2/blobs/%s/%s/%s/data", algorithm, hash[:2], hash) -} diff --git a/pkg/hold/pds/xrpc.go b/pkg/hold/pds/xrpc.go index 3689a1c..daaa257 100644 --- a/pkg/hold/pds/xrpc.go +++ b/pkg/hold/pds/xrpc.go @@ -217,8 +217,8 @@ func (h *XRPCHandler) HandleDescribeServer(w http.ResponseWriter, r *http.Reques hostname := h.pds.PublicURL hostname = strings.TrimPrefix(hostname, "http://") hostname = strings.TrimPrefix(hostname, "https://") - hostname = strings.Split(hostname, "/")[0] // Remove path - hostname = strings.Split(hostname, ":")[0] // Remove port + hostname, _, _ = strings.Cut(hostname, "/") // Remove path + hostname, _, _ = strings.Cut(hostname, ":") // Remove port response := map[string]any{ "did": h.pds.DID(),