From 3edecb9d3caaf8ed743b8f9a906c2405fc5b4acf Mon Sep 17 00:00:00 2001 From: Sameeksha Malav Date: Mon, 21 Sep 2026 19:08:35 +0530 Subject: [PATCH] feat: [AH-4984]: (har) add multipart upload support --- modules/har/pkg/har/harutil.go | 72 +++ modules/har/pkg/har/multipart.go | 790 ++++++++++++++++++++++++++ modules/har/pkg/har/multipart_test.go | 720 +++++++++++++++++++++++ modules/har/pkg/har/push_generic.go | 79 ++- pkg/spec/har.spec.yaml | 6 + 5 files changed, 1658 insertions(+), 9 deletions(-) create mode 100644 modules/har/pkg/har/multipart.go create mode 100644 modules/har/pkg/har/multipart_test.go diff --git a/modules/har/pkg/har/harutil.go b/modules/har/pkg/har/harutil.go index b61451a..7a3198f 100644 --- a/modules/har/pkg/har/harutil.go +++ b/modules/har/pkg/har/harutil.go @@ -8,6 +8,7 @@ import ( "archive/zip" "compress/gzip" "encoding/json" + "errors" "fmt" "io" "net/http" @@ -98,6 +99,77 @@ func doRequest(c *http.Client, req *http.Request) ([]byte, error) { return body, nil } +// apiError is a non-2xx response from the registry, preserving the status code and the +// machine-readable error code the backend puts in values.code. Callers switch on Code rather +// than matching error strings. +type apiError struct { + StatusCode int + Code string + Message string + Values map[string]any +} + +func (e *apiError) Error() string { + if e.Code != "" { + return fmt.Sprintf("HTTP %d (%s): %s", e.StatusCode, e.Code, e.Message) + } + return fmt.Sprintf("HTTP %d: %s", e.StatusCode, e.Message) +} + +// apiErrorCode returns the backend values.code carried by err, or "" if err is not an apiError. +func apiErrorCode(err error) string { + var apiErr *apiError + if errors.As(err, &apiErr) { + return apiErr.Code + } + return "" +} + +// apiErrorStatus returns the HTTP status carried by err, or 0 if err is not an apiError. +func apiErrorStatus(err error) int { + var apiErr *apiError + if errors.As(err, &apiErr) { + return apiErr.StatusCode + } + return 0 +} + +// doJSONRequest executes req and returns the status code and body. Unlike doRequest it reports +// the status code on success too (callers need to tell 200 from 201/202) and parses a non-2xx +// body into an *apiError so the backend's values.code is preserved. +func doJSONRequest(c *http.Client, req *http.Request) (int, []byte, error) { + resp, err := c.Do(req) + if err != nil { + return 0, nil, err + } + defer resp.Body.Close() + + body, readErr := io.ReadAll(resp.Body) + if readErr != nil { + return resp.StatusCode, nil, fmt.Errorf("reading response body: %w", readErr) + } + if resp.StatusCode >= 200 && resp.StatusCode < 300 { + return resp.StatusCode, body, nil + } + + apiErr := &apiError{StatusCode: resp.StatusCode, Message: strings.TrimSpace(string(body))} + + var parsed struct { + Message string `json:"message"` + Values map[string]any `json:"values"` + } + if json.Unmarshal(body, &parsed) == nil { + if parsed.Message != "" { + apiErr.Message = parsed.Message + } + apiErr.Values = parsed.Values + if code, ok := parsed.Values["code"].(string); ok { + apiErr.Code = code + } + } + return resp.StatusCode, body, apiErr +} + // buildPkgURL constructs a registry URL of the form: // // {registryURL}/pkg/{accountID}/{subpath}?accountIdentifier={accountID} diff --git a/modules/har/pkg/har/multipart.go b/modules/har/pkg/har/multipart.go new file mode 100644 index 0000000..8de37f3 --- /dev/null +++ b/modules/har/pkg/har/multipart.go @@ -0,0 +1,790 @@ +// Copyright © 2026 Harness Inc. +// SPDX-License-Identifier: Apache-2.0 + +package har + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "os" + "sync" + "time" + + "github.com/harness/cli/v3/pkg/cmdctx" +) + +const ( + // multipartThreshold is the size at or above which a file is uploaded in parallel parts. It + // matches the registry's default part size: below it the server returns partCount == 1, so + // multipart would add four round trips (start, put, complete, poll) for no parallelism at all. + multipartThreshold = 16 * 1024 * 1024 + + // defaultMaxConcurrentParts bounds part uploads in flight across every file in one command. + defaultMaxConcurrentParts = 8 + + // maxPresignBatch mirrors the server's cap on how many part URLs a single response carries. + // Uploads with more parts than this fetch the remainder in batches. + maxPresignBatch = 100 + + // presignRefreshWindow is the validity a presigned URL must have left to be worth using. A URL + // closer than this to expiry is re-fetched first, so a slow upload does not race the signature. + presignRefreshWindow = 60 * time.Second + + // partUploadMaxAttempts bounds retries of a single part, including the first attempt. + partUploadMaxAttempts = 4 + + // partRetryInitialDelay and partRetryMaxDelay bound the exponential backoff between attempts at + // one part. They are deliberately separate from the poll intervals below: the two schedules + // start at the same values today, but retuning how often the CLI polls for finalization should + // not silently change how hard it retries a failed part. + partRetryInitialDelay = 500 * time.Millisecond + partRetryMaxDelay = 5 * time.Second + + // completeMaxAttempts bounds how many times the CLI re-uploads missing parts and retries + // completion before giving up. + completeMaxAttempts = 3 + + // Finalization happens server-side after completion is accepted, so the CLI polls for it. + pollInitialInterval = 500 * time.Millisecond + pollMaxInterval = 5 * time.Second + pollTimeout = 30 * time.Minute + + // abortTimeout bounds the best-effort abort issued when an upload fails or is cancelled. + abortTimeout = 15 * time.Second + + // presignExpiryLayout is the timestamp format the registry uses for a part URL's expiry. + presignExpiryLayout = time.RFC3339 +) + +// Multipart session statuses reported by the registry. completed, failed and aborted are terminal. +const ( + uploadStatusOpen = "open" + uploadStatusFinalizing = "finalizing" + uploadStatusCompleted = "completed" + uploadStatusFailed = "failed" + uploadStatusAborted = "aborted" +) + +// Machine-readable error codes the registry returns in values.code. These are a stable contract +// between the registry and this client, so switch on them rather than on message text. +// +// Only the first four change what the client does. The rest are declared to document the contract +// and are surfaced to the user as-is: the registry's own message for them is already actionable +// (e.g. PATH_CONFLICT reports which path is busy, OPEN_UPLOAD_LIMIT says to abort one and retry), +// and there is no recovery the client could attempt on its own. +const ( + codeMultipartUnsupported = "MULTIPART_UNSUPPORTED" + codeIncompleteUpload = "INCOMPLETE_UPLOAD" + codeUploadNotFound = "UPLOAD_NOT_FOUND" + codeUploadNotOpen = "UPLOAD_NOT_OPEN" + codeDigestConflict = "DIGEST_CONFLICT" + codePathConflict = "PATH_CONFLICT" + codeOpenUploadLimit = "OPEN_UPLOAD_LIMIT" + codeSizeMismatch = "SIZE_MISMATCH" + codeInvalidRequest = "INVALID_REQUEST" +) + +// errMultipartUnsupported reports that this registry cannot do a multipart upload, either because +// the feature is off or because the endpoint does not exist on this server version. It is only ever +// returned before any file bytes are sent, so the caller can fall back to a single upload for free. +var errMultipartUnsupported = errors.New("multipart upload is not supported by this registry") + +// startUploadRequest is the body of a start-upload call. +// +// The server rejects unknown fields, so this struct must carry exactly the documented set. +type startUploadRequest struct { + Path string `json:"path"` + Size int64 `json:"size"` + SHA256 string `json:"sha256"` + SHA1 string `json:"sha1,omitempty"` + MD5 string `json:"md5,omitempty"` + SHA512 string `json:"sha512,omitempty"` +} + +// partURL is one presigned part destination. +type partURL struct { + PartNumber int `json:"partNumber"` + URL string `json:"url"` + ExpiresAt string `json:"expiresAt"` +} + +// expired reports whether this URL has less than presignRefreshWindow of validity left. An +// unparseable or absent expiry is treated as expired so the URL is refreshed rather than trusted. +func (p partURL) expired() bool { + if p.ExpiresAt == "" { + return true + } + expiresAt, err := time.Parse(presignExpiryLayout, p.ExpiresAt) + if err != nil { + return true + } + return time.Until(expiresAt) < presignRefreshWindow +} + +// startUploadResponse describes the session the server opened, or reports that the content already +// exists. partSize and partCount are authoritative: the client must never compute its own, because +// the object store rejects completion unless every non-final part is exactly partSize bytes. +type startUploadResponse struct { + UploadID string `json:"uploadId"` + Status string `json:"status"` + PartSize int64 `json:"partSize"` + PartCount int `json:"partCount"` + StatusURL string `json:"statusUrl"` + Parts []partURL `json:"parts"` +} + +type partsResponse struct { + Parts []partURL `json:"parts"` +} + +type uploadStatusResponse struct { + UploadID string `json:"uploadId"` + Status string `json:"status"` + Error *string `json:"error"` +} + +type completeUploadResponse struct { + UploadID string `json:"uploadId"` + Status string `json:"status"` + Error *string `json:"error"` +} + +// multipartUploader uploads files in parallel parts. One instance is shared by every file in a +// command so that partSem bounds the total number of part uploads in flight: a per-file semaphore +// would let file concurrency multiply by part fan-out and open (files x parts) connections at once. +type multipartUploader struct { + ctx *cmdctx.Ctx + registry string + + // control carries the start/status/parts/complete/abort calls, which are authenticated. + control *http.Client + + // part uploads go to presigned URLs, where the signature in the URL *is* the credential. + // Sending an Authorization or x-api-key header alongside it makes the object store reject the + // request, so this client is deliberately separate and has no auth wiring. + part *http.Client + + partSem chan struct{} +} + +func newMultipartUploader( + ctx *cmdctx.Ctx, control *http.Client, registry string, maxConcurrentParts int, +) *multipartUploader { + if maxConcurrentParts <= 0 { + maxConcurrentParts = defaultMaxConcurrentParts + } + return &multipartUploader{ + ctx: ctx, + registry: registry, + control: control, + part: &http.Client{Timeout: 10 * time.Minute}, + partSem: make(chan struct{}, maxConcurrentParts), + } +} + +// upload uploads localPath as relPath using a multipart session. +// +// sums is supplied by the caller rather than computed here because the single-upload fallback needs +// the same digests: hashing in both places would read a file that is large by definition twice. +// +// It returns errMultipartUnsupported if this registry cannot do multipart, in which case no bytes +// were sent and the caller should fall back to a single upload. +func (u *multipartUploader) upload( + name, version, relPath, localPath string, size int64, sums fileChecksums, +) error { + session, err := u.startUpload(name, version, relPath, size, sums) + if err != nil { + return err + } + + // The server already holds this content for the account, so there is nothing to upload. + if session.Status == uploadStatusCompleted { + fmt.Fprintf(os.Stderr, "%s already exists in the registry, skipped upload\n", relPath) + return nil + } + + if err := u.validateSession(session, size); err != nil { + return err + } + + // Abort the session on any failure so it does not hold the one-in-flight-upload-per-path slot + // until it expires, which would make an immediate retry fail with a path conflict. + done := false + defer func() { + if !done { + u.abortUpload(session.UploadID) + } + }() + + if err := u.uploadPartSet(localPath, size, session, allPartNumbers(session.PartCount)); err != nil { + return err + } + if err := u.completeUpload(localPath, size, session); err != nil { + return err + } + if err := u.pollUntilTerminal(relPath, session.UploadID); err != nil { + return err + } + + done = true + return nil +} + +// validateSession checks the sizing the server handed back is self-consistent before any bytes are +// sent, so a server bug surfaces as a clear error rather than a corrupted object at completion. +func (u *multipartUploader) validateSession(session *startUploadResponse, size int64) error { + switch { + case session.UploadID == "": + return fmt.Errorf("registry opened a multipart upload without an upload id") + case session.PartSize <= 0: + return fmt.Errorf("registry returned an invalid part size %d", session.PartSize) + case session.PartCount <= 0: + return fmt.Errorf("registry returned an invalid part count %d", session.PartCount) + } + + // Every part but the last is exactly partSize, so the declared sizing must cover the file with + // less than one part of slack. + if got, want := int64(session.PartCount)*session.PartSize, size; got < want || got-want >= session.PartSize { + return fmt.Errorf( + "registry part sizing does not match file: partCount=%d partSize=%d for %d bytes", + session.PartCount, session.PartSize, size, + ) + } + return nil +} + +// startUpload opens a multipart session, or reports that the content is already stored. +// +// path is the same string the single-upload route carries after /files/, namely +// {package}/{version}/{file...}. The server applies the same layout rules to it as it does to that +// route's remainder: for a GENERIC registry it splits out the package, version and file path and +// requires at least those three segments, while the other /files package types (RAW, HELM_HTTP, +// CRAN) treat it as a flat file path. Sending the identical string either way is what makes the two +// upload flows land the artifact in the same place for every one of those package types. +func (u *multipartUploader) startUpload( + name, version, relPath string, size int64, sums fileChecksums, +) (*startUploadResponse, error) { + body, err := json.Marshal(startUploadRequest{ + Path: fmt.Sprintf("%s/%s/%s", name, version, relPath), + Size: size, + SHA256: sums.SHA256, + SHA1: sums.SHA1, + MD5: sums.MD5, + SHA512: sums.SHA512, + }) + if err != nil { + return nil, fmt.Errorf("encoding start upload request: %w", err) + } + + req, err := u.newControlRequest(http.MethodPost, "", bytes.NewReader(body)) + if err != nil { + return nil, err + } + req.Header.Set("Content-Type", "application/json") + + status, respBody, err := doJSONRequest(u.control, req) + if err != nil { + if startFailureMeansUnsupported(u.ctx.Context, err) { + return nil, errMultipartUnsupported + } + return nil, fmt.Errorf("starting multipart upload: %w", err) + } + + var session startUploadResponse + if err := json.Unmarshal(respBody, &session); err != nil { + return nil, fmt.Errorf("decoding start upload response (HTTP %d): %w", status, err) + } + return &session, nil +} + +// startFailureMeansUnsupported reports whether a failed start should fall back to a single upload +// instead of failing the push. +// +// A 501 with MULTIPART_UNSUPPORTED says the server knows the endpoint but has the feature disabled, +// and a 404 says this server version does not have the endpoint at all. A transport failure counts +// too: no response arrived, so multipart availability is simply unknown, and a single PUT is a +// strictly simpler request that deserves the attempt rather than failing a push outright. +// +// Every other HTTP status is a real answer from the registry and must surface. Falling back on a +// 401, 403 or 5xx would mask the cause behind a second attempt that hits the same wall, and the +// user would see a confusing error from the fallback instead of the actual one. +func startFailureMeansUnsupported(ctx context.Context, err error) bool { + // Whether to fall back is decided from the command's own context, not from the error text. A + // cancelled or expired command says nothing about multipart support, and retrying under a message + // claiming the registry lacks the feature would be wrong. But the error alone cannot tell the two + // apart: http.Client.Timeout also reports context.DeadlineExceeded, so matching on the error would + // refuse to fall back in exactly the case that needs it - an unreachable registry, where the + // retries are exhausted and the client's own timeout fires while awaiting headers. + if ctx.Err() != nil { + return false + } + + var apiErr *apiError + if !errors.As(err, &apiErr) { + return true + } + return apiErr.Code == codeMultipartUnsupported || apiErr.StatusCode == http.StatusNotFound +} + +// uploadPartSet uploads the given part numbers in parallel, bounded by the shared part semaphore. +// It serves both the initial upload of every part and the repair of the specific parts the object +// store turned out to be missing, because the two differ only in which numbers they cover. +func (u *multipartUploader) uploadPartSet( + localPath string, size int64, session *startUploadResponse, partNumbers []int, +) error { + f, err := os.Open(localPath) + if err != nil { + return fmt.Errorf("opening %q: %w", localPath, err) + } + defer f.Close() + + urls := newPartURLCache(u, session) + + var wg sync.WaitGroup + errs := make([]error, len(partNumbers)) + + // ctx is cancelled as soon as one part fails so the remaining parts stop early instead of + // uploading bytes for a session that is already doomed. + ctx, cancel := context.WithCancel(u.ctx.Context) + defer cancel() + + for i, partNumber := range partNumbers { + wg.Add(1) + go func(i, partNumber int) { + defer wg.Done() + + select { + case u.partSem <- struct{}{}: + case <-ctx.Done(): + errs[i] = ctx.Err() + return + } + defer func() { <-u.partSem }() + + if err := u.uploadPart(ctx, f, size, session, urls, partNumber); err != nil { + errs[i] = fmt.Errorf("part %d: %w", partNumber, err) + cancel() + } + }(i, partNumber) + } + wg.Wait() + + return joinPartErrors(errs) +} + +// allPartNumbers returns every part number of a session, 1-based. +func allPartNumbers(partCount int) []int { + numbers := make([]int, partCount) + for i := range numbers { + numbers[i] = i + 1 + } + return numbers +} + +// uploadPart PUTs one part, refreshing its presigned URL and retrying on transient failures. +func (u *multipartUploader) uploadPart( + ctx context.Context, + f *os.File, + size int64, + session *startUploadResponse, + urls *partURLCache, + partNumber int, +) error { + offset := int64(partNumber-1) * session.PartSize + length := session.PartSize + if remaining := size - offset; remaining < length { + length = remaining + } + if length <= 0 { + return fmt.Errorf("computed a zero-length part at offset %d of %d bytes", offset, size) + } + + var lastErr error + for attempt := 1; attempt <= partUploadMaxAttempts; attempt++ { + if err := ctx.Err(); err != nil { + return err + } + + url, err := urls.get(ctx, partNumber) + if err != nil { + return err + } + + // A SectionReader is re-created per attempt so a retry re-reads the same bytes. Handing the + // *os.File itself to the request would advance the shared file offset and, because parts + // upload concurrently, read the wrong region. + body := io.NewSectionReader(f, offset, length) + + req, err := http.NewRequestWithContext(ctx, http.MethodPut, url, body) + if err != nil { + return fmt.Errorf("building part request: %w", err) + } + // The presigned URL carries its own credentials; no auth header is set here on purpose. + // Content-Length is mandatory because object stores reject an unsized part body. + req.ContentLength = length + req.Header.Set("Content-Type", "application/octet-stream") + + resp, err := u.part.Do(req) + if err != nil { + lastErr = err + } else { + // The part ETag is deliberately ignored: completion asks the object store itself which + // parts it holds, so the client never sends a part list back. + _, _ = io.Copy(io.Discard, resp.Body) + resp.Body.Close() + + if resp.StatusCode >= 200 && resp.StatusCode < 300 { + return nil + } + lastErr = fmt.Errorf("HTTP %d uploading part to object storage", resp.StatusCode) + + // 403 from an object store almost always means the signature expired rather than a + // permission change, so drop the cached URL and sign a fresh one for the next attempt. + if resp.StatusCode == http.StatusForbidden { + urls.invalidate(partNumber) + } + if !retryablePartStatus(resp.StatusCode) { + return lastErr + } + } + + if attempt < partUploadMaxAttempts { + if err := sleepCtx(ctx, backoffDelay(attempt)); err != nil { + return err + } + } + } + return fmt.Errorf("giving up after %d attempts: %w", partUploadMaxAttempts, lastErr) +} + +// retryablePartStatus reports whether a failed part PUT is worth retrying. A 403 is included +// because an expired signature presents as one and is fixed by re-signing. +func retryablePartStatus(status int) bool { + switch { + case status == http.StatusRequestTimeout, + status == http.StatusTooManyRequests, + status == http.StatusForbidden: + return true + case status >= 500: + return true + default: + return false + } +} + +// backoffDelay returns an exponential delay for retrying a part, given a 1-based attempt number. +func backoffDelay(attempt int) time.Duration { + delay := partRetryInitialDelay << (attempt - 1) + if delay > partRetryMaxDelay { + delay = partRetryMaxDelay + } + return delay +} + +func sleepCtx(ctx context.Context, d time.Duration) error { + timer := time.NewTimer(d) + defer timer.Stop() + select { + case <-timer.C: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} + +// partURLCache hands out presigned part URLs, fetching them from the registry in batches and +// re-signing any that are expired or near expiry. The server caps a single response at +// maxPresignBatch parts, so an upload with more parts than that necessarily fetches more than once. +type partURLCache struct { + uploader *multipartUploader + uploadID string + total int + + mu sync.Mutex + byNum map[int]partURL +} + +func newPartURLCache(uploader *multipartUploader, session *startUploadResponse) *partURLCache { + cache := &partURLCache{ + uploader: uploader, + uploadID: session.UploadID, + total: session.PartCount, + byNum: make(map[int]partURL, session.PartCount), + } + for _, part := range session.Parts { + cache.byNum[part.PartNumber] = part + } + return cache +} + +// get returns a usable URL for partNumber, fetching a fresh batch if the cached one is missing or +// close to expiry. +func (c *partURLCache) get(ctx context.Context, partNumber int) (string, error) { + c.mu.Lock() + cached, ok := c.byNum[partNumber] + c.mu.Unlock() + if ok && !cached.expired() { + return cached.URL, nil + } + + // The lock is released before fetching on purpose: holding it across a network call would + // serialise every part behind one re-sign. The cost is that two parts needing the same window + // may both fetch it, which is harmless — signing is idempotent and the results are identical. + // + // Fetch a window starting at this part rather than just this one part: parts are uploaded in + // ascending order, so the neighbours are about to be needed too. + from := partNumber + to := from + maxPresignBatch - 1 + if to > c.total { + to = c.total + } + + fetched, err := c.uploader.fetchParts(ctx, c.uploadID, from, to) + if err != nil { + return "", err + } + + c.mu.Lock() + for _, part := range fetched { + c.byNum[part.PartNumber] = part + } + refreshed, ok := c.byNum[partNumber] + c.mu.Unlock() + + if !ok { + return "", fmt.Errorf("registry did not return a URL for part %d", partNumber) + } + return refreshed.URL, nil +} + +// invalidate drops the cached URL for partNumber so the next get re-signs it. +func (c *partURLCache) invalidate(partNumber int) { + c.mu.Lock() + delete(c.byNum, partNumber) + c.mu.Unlock() +} + +// fetchParts asks the registry for presigned URLs for parts from..to inclusive. +func (u *multipartUploader) fetchParts(ctx context.Context, uploadID string, from, to int) ([]partURL, error) { + req, err := u.newControlRequest(http.MethodGet, uploadID+"/parts", nil) + if err != nil { + return nil, err + } + req = req.WithContext(ctx) + + q := req.URL.Query() + q.Set("from", fmt.Sprint(from)) + q.Set("to", fmt.Sprint(to)) + req.URL.RawQuery = q.Encode() + + _, body, err := doJSONRequest(u.control, req) + if err != nil { + return nil, fmt.Errorf("fetching part URLs %d-%d: %w", from, to, err) + } + + var parsed partsResponse + if err := json.Unmarshal(body, &parsed); err != nil { + return nil, fmt.Errorf("decoding part URLs response: %w", err) + } + return parsed.Parts, nil +} + +// completeUpload asks the registry to assemble the parts into the final object. +// +// The body is empty by design: the server lists the parts from the object store itself, so it needs +// neither ETags nor a part list. If the store is missing parts the server reports which ones, and +// the session stays open so they can be re-uploaded and completion retried. +func (u *multipartUploader) completeUpload(localPath string, size int64, session *startUploadResponse) error { + for attempt := 1; ; attempt++ { + req, err := u.newControlRequest(http.MethodPost, session.UploadID+"/complete", nil) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/json") + + _, body, err := doJSONRequest(u.control, req) + if err == nil { + var parsed completeUploadResponse + // A malformed body is not fatal here: the poll that follows is the real source of truth. + if jsonErr := json.Unmarshal(body, &parsed); jsonErr == nil && parsed.Status == uploadStatusFailed { + return fmt.Errorf("registry failed to finalize the upload: %s", derefOr(parsed.Error, "unknown error")) + } + return nil + } + + missing := missingParts(err) + if apiErrorCode(err) != codeIncompleteUpload || len(missing) == 0 || attempt >= completeMaxAttempts { + return fmt.Errorf("completing multipart upload: %w", err) + } + + fmt.Fprintf(os.Stderr, "Re-uploading %d missing part(s) ...\n", len(missing)) + if err := u.uploadPartSet(localPath, size, session, missing); err != nil { + return err + } + } +} + +// missingParts extracts the part numbers an INCOMPLETE_UPLOAD error reports as absent. +func missingParts(err error) []int { + var apiErr *apiError + if !errors.As(err, &apiErr) { + return nil + } + raw, ok := apiErr.Values["missingParts"].([]any) + if !ok { + return nil + } + + parts := make([]int, 0, len(raw)) + for _, v := range raw { + // JSON numbers decode to float64 through an any-typed map. + if n, ok := v.(float64); ok { + parts = append(parts, int(n)) + } + } + return parts +} + +// pollUntilTerminal waits for the registry to finish finalizing the upload. +// +// Completion is accepted asynchronously: the registry assembles the object and a background job +// verifies the digest before the artifact becomes visible, so the upload is only really done once +// the session reaches a terminal status. +func (u *multipartUploader) pollUntilTerminal(relPath, uploadID string) error { + ctx, cancel := context.WithTimeout(u.ctx.Context, pollTimeout) + defer cancel() + + fmt.Fprintf(os.Stderr, "Finalizing %s ...\n", relPath) + + interval := pollInitialInterval + for { + status, err := u.getUploadStatus(ctx, uploadID) + if err != nil { + return err + } + + switch status.Status { + case uploadStatusCompleted: + return nil + case uploadStatusFailed: + return fmt.Errorf("registry failed to finalize the upload: %s", derefOr(status.Error, "unknown error")) + case uploadStatusAborted: + return fmt.Errorf("the upload was aborted before it finished") + case uploadStatusOpen, uploadStatusFinalizing: + // Still working; fall through to wait. + default: + return fmt.Errorf("registry reported an unknown upload status %q", status.Status) + } + + if err := sleepCtx(ctx, interval); err != nil { + if errors.Is(err, context.DeadlineExceeded) { + return fmt.Errorf("upload did not finish finalizing within %s", pollTimeout) + } + return err + } + if interval < pollMaxInterval { + interval *= 2 + if interval > pollMaxInterval { + interval = pollMaxInterval + } + } + } +} + +func (u *multipartUploader) getUploadStatus(ctx context.Context, uploadID string) (*uploadStatusResponse, error) { + req, err := u.newControlRequest(http.MethodGet, uploadID, nil) + if err != nil { + return nil, err + } + req = req.WithContext(ctx) + + _, body, err := doJSONRequest(u.control, req) + if err != nil { + return nil, fmt.Errorf("checking upload status: %w", err) + } + + var status uploadStatusResponse + if err := json.Unmarshal(body, &status); err != nil { + return nil, fmt.Errorf("decoding upload status response: %w", err) + } + return &status, nil +} + +// abortUpload releases a session that will not be completed. It is best effort: the caller is +// already returning an error, and the server expires abandoned sessions anyway. +// +// It deliberately does not use the command context, which may already be cancelled — that is one of +// the main reasons an abort is needed. +func (u *multipartUploader) abortUpload(uploadID string) { + if uploadID == "" { + return + } + + ctx, cancel := context.WithTimeout(context.Background(), abortTimeout) + defer cancel() + + req, err := u.newControlRequest(http.MethodDelete, uploadID, nil) + if err != nil { + return + } + req = req.WithContext(ctx) + + if _, _, err := doJSONRequest(u.control, req); err != nil { + // An already-gone session is the expected outcome of racing the server's own cleanup. + if code := apiErrorCode(err); code == codeUploadNotFound || code == codeUploadNotOpen { + return + } + fmt.Fprintf(os.Stderr, "Warning: could not abort multipart upload %s: %v\n", uploadID, err) + } +} + +// newControlRequest builds an authenticated request against the registry's uploads endpoints. +// suffix is appended to the collection path, e.g. "" for the collection itself, +// "{uploadID}" for one session, or "{uploadID}/complete". +func (u *multipartUploader) newControlRequest(method, suffix string, body io.Reader) (*http.Request, error) { + subpath := u.registry + "/uploads" + if suffix != "" { + subpath += "/" + suffix + } + + url, err := buildPkgURL(u.ctx.Auth.RegistryURL, u.ctx.Auth.AccountID, subpath) + if err != nil { + return nil, err + } + + req, err := http.NewRequestWithContext(u.ctx.Context, method, url, body) + if err != nil { + return nil, fmt.Errorf("building %s %s request: %w", method, subpath, err) + } + setAuthHeader(req, u.ctx.Auth) + return req, nil +} + +// joinPartErrors combines per-part errors into one. Cancellations are dropped when a real failure +// is present, because the first failing part cancels the others and their context.Canceled errors +// would otherwise bury the actual cause. +func joinPartErrors(errs []error) error { + real := make([]error, 0, len(errs)) + for _, err := range errs { + if err != nil && !errors.Is(err, context.Canceled) { + real = append(real, err) + } + } + if len(real) > 0 { + return errors.Join(real...) + } + return errors.Join(errs...) +} + +// derefOr returns *s, or fallback when s is nil or empty. +func derefOr(s *string, fallback string) string { + if s == nil || *s == "" { + return fallback + } + return *s +} diff --git a/modules/har/pkg/har/multipart_test.go b/modules/har/pkg/har/multipart_test.go new file mode 100644 index 0000000..db2ed5b --- /dev/null +++ b/modules/har/pkg/har/multipart_test.go @@ -0,0 +1,720 @@ +// Copyright © 2026 Harness Inc. +// SPDX-License-Identifier: Apache-2.0 + +package har + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "sort" + "strings" + "sync" + "testing" + "time" + + "github.com/harness/cli/v3/pkg/auth" + "github.com/harness/cli/v3/pkg/cmdctx" +) + +// --- Test helpers --- + +func multipartTestCtx(registryURL string) *cmdctx.Ctx { + return &cmdctx.Ctx{ + Context: context.Background(), + Auth: &auth.ResolvedAuth{ + AuthType: auth.AuthTypePAT, + APIUrl: registryURL, + RegistryURL: registryURL, + AccountID: "acct", + PATToken: "test-token", + }, + } +} + +// fakeRegistry is an httptest-backed stand-in for the registry's multipart endpoints plus the +// object store the presigned part URLs point at. Everything it records is guarded by mu because +// parts are uploaded concurrently. +type fakeRegistry struct { + t *testing.T + srv *httptest.Server + + mu sync.Mutex + parts map[int][]byte + // startBodies records every decoded Start request body, for contract assertions. + startBodies []map[string]any + // partFetches records each (from,to) range requested from the parts endpoint. + partFetches [][2]int + aborted bool + completes int + + // Knobs the tests set before calling upload. + partSize int64 + partCount int + presignInStart int // how many part URLs the Start response carries + startStatus int // HTTP status Start replies with + startCode string // values.code Start replies with on a non-2xx + dedup bool // Start replies 200 {"status":"completed"} + expiredURLs bool // Start's part URLs are already expired + failPartsOnce map[int]bool + partFailStatus int + missingOnce []int // parts the first Complete reports as missing + finalStatus string // status the poll endpoint reports once complete has been accepted + finalError string +} + +func newFakeRegistry(t *testing.T) *fakeRegistry { + f := &fakeRegistry{ + t: t, + parts: map[int][]byte{}, + startStatus: http.StatusCreated, + partFailStatus: http.StatusInternalServerError, + failPartsOnce: map[int]bool{}, + finalStatus: uploadStatusCompleted, + } + f.srv = httptest.NewServer(f) + t.Cleanup(f.srv.Close) + return f +} + +func (f *fakeRegistry) ServeHTTP(w http.ResponseWriter, r *http.Request) { + switch { + case strings.HasPrefix(r.URL.Path, "/object/"): + f.servePart(w, r) + case r.Method == http.MethodPost && r.URL.Path == "/pkg/acct/reg/uploads": + f.serveStart(w, r) + case r.Method == http.MethodGet && r.URL.Path == "/pkg/acct/reg/uploads/up-1/parts": + f.serveParts(w, r) + case r.Method == http.MethodPost && r.URL.Path == "/pkg/acct/reg/uploads/up-1/complete": + f.serveComplete(w, r) + case r.Method == http.MethodGet && r.URL.Path == "/pkg/acct/reg/uploads/up-1": + f.serveStatus(w, r) + case r.Method == http.MethodDelete && r.URL.Path == "/pkg/acct/reg/uploads/up-1": + f.mu.Lock() + f.aborted = true + f.mu.Unlock() + w.WriteHeader(http.StatusNoContent) + default: + f.t.Errorf("unexpected request %s %s", r.Method, r.URL.Path) + w.WriteHeader(http.StatusNotFound) + } +} + +// writeAPIError mirrors the registry's error envelope: {"message":..,"values":{"code":..}}. +func writeAPIError(w http.ResponseWriter, status int, values map[string]any) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(map[string]any{"message": "boom", "values": values}) +} + +func (f *fakeRegistry) serveStart(w http.ResponseWriter, r *http.Request) { + // Control calls must be authenticated. + if r.Header.Get("x-api-key") == "" && r.Header.Get("Authorization") == "" { + f.t.Error("Start request carried no auth header") + } + + var body map[string]any + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + f.t.Fatalf("decoding start body: %v", err) + } + f.mu.Lock() + f.startBodies = append(f.startBodies, body) + f.mu.Unlock() + + if f.startStatus != http.StatusCreated { + writeAPIError(w, f.startStatus, map[string]any{"code": f.startCode}) + return + } + if f.dedup { + w.WriteHeader(http.StatusOK) + _ = json.NewEncoder(w).Encode(startUploadResponse{Status: uploadStatusCompleted}) + return + } + + presign := f.presignInStart + if presign == 0 || presign > f.partCount { + presign = f.partCount + } + w.WriteHeader(http.StatusCreated) + _ = json.NewEncoder(w).Encode(startUploadResponse{ + UploadID: "up-1", + Status: uploadStatusOpen, + PartSize: f.partSize, + PartCount: f.partCount, + Parts: f.presign(1, presign), + }) +} + +// presign builds part URLs for from..to inclusive, pointing at this server's object-store route. +func (f *fakeRegistry) presign(from, to int) []partURL { + expiry := time.Now().Add(time.Hour) + if f.expiredURLs { + expiry = time.Now().Add(10 * time.Second) // inside presignRefreshWindow + } + var urls []partURL + for n := from; n <= to; n++ { + urls = append(urls, partURL{ + PartNumber: n, + URL: fmt.Sprintf("%s/object/%d", f.srv.URL, n), + ExpiresAt: expiry.Format(presignExpiryLayout), + }) + } + return urls +} + +func (f *fakeRegistry) serveParts(w http.ResponseWriter, r *http.Request) { + var from, to int + fmt.Sscanf(r.URL.Query().Get("from"), "%d", &from) + fmt.Sscanf(r.URL.Query().Get("to"), "%d", &to) + + f.mu.Lock() + f.partFetches = append(f.partFetches, [2]int{from, to}) + // Refreshed URLs are always long-lived, so a refresh terminates rather than looping. + f.expiredURLs = false + f.mu.Unlock() + + _ = json.NewEncoder(w).Encode(partsResponse{Parts: f.presign(from, to)}) +} + +func (f *fakeRegistry) servePart(w http.ResponseWriter, r *http.Request) { + // A presigned URL is self-authenticating; an auth header alongside it makes real object + // stores reject the request, so assert the client does not send one. + if r.Header.Get("Authorization") != "" || r.Header.Get("x-api-key") != "" { + f.t.Error("part upload carried an auth header") + } + if r.ContentLength <= 0 { + f.t.Errorf("part upload had ContentLength %d, want > 0", r.ContentLength) + } + + var partNumber int + fmt.Sscanf(strings.TrimPrefix(r.URL.Path, "/object/"), "%d", &partNumber) + + f.mu.Lock() + if f.failPartsOnce[partNumber] { + delete(f.failPartsOnce, partNumber) + f.mu.Unlock() + w.WriteHeader(f.partFailStatus) + return + } + f.mu.Unlock() + + var buf bytes.Buffer + if _, err := buf.ReadFrom(r.Body); err != nil { + f.t.Errorf("reading part %d: %v", partNumber, err) + } + + f.mu.Lock() + f.parts[partNumber] = buf.Bytes() + f.mu.Unlock() + + w.Header().Set("ETag", fmt.Sprintf("%q", fmt.Sprintf("etag-%d", partNumber))) + w.WriteHeader(http.StatusOK) +} + +func (f *fakeRegistry) serveComplete(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + if len(body) != 0 { + f.t.Errorf("Complete carried a body %q, want empty", body) + } + + f.mu.Lock() + f.completes++ + missing := f.missingOnce + if len(missing) > 0 { + f.missingOnce = nil + for _, n := range missing { + delete(f.parts, n) + } + } + f.mu.Unlock() + + if len(missing) > 0 { + asAny := make([]any, len(missing)) + for i, n := range missing { + asAny[i] = float64(n) + } + writeAPIError(w, http.StatusConflict, map[string]any{ + "code": codeIncompleteUpload, "missingParts": asAny, + }) + return + } + + w.WriteHeader(http.StatusAccepted) + _ = json.NewEncoder(w).Encode(completeUploadResponse{UploadID: "up-1", Status: uploadStatusFinalizing}) +} + +func (f *fakeRegistry) serveStatus(w http.ResponseWriter, _ *http.Request) { + resp := uploadStatusResponse{UploadID: "up-1", Status: f.finalStatus} + if f.finalError != "" { + resp.Error = &f.finalError + } + _ = json.NewEncoder(w).Encode(resp) +} + +// assembled returns the parts concatenated in part-number order. +func (f *fakeRegistry) assembled() []byte { + f.mu.Lock() + defer f.mu.Unlock() + + nums := make([]int, 0, len(f.parts)) + for n := range f.parts { + nums = append(nums, n) + } + sort.Ints(nums) + + var out []byte + for _, n := range nums { + out = append(out, f.parts[n]...) + } + return out +} + +// writeTempFile writes size bytes of deterministic, non-repeating-per-part content. +func writeTempFile(t *testing.T, size int) string { + t.Helper() + data := make([]byte, size) + for i := range data { + data[i] = byte(i % 251) + } + path := filepath.Join(t.TempDir(), "big.bin") + if err := os.WriteFile(path, data, 0o600); err != nil { + t.Fatalf("writing temp file: %v", err) + } + return path +} + +// testSums hashes path the way the push command does before choosing an upload path. +func testSums(t *testing.T, path string) fileChecksums { + t.Helper() + sums, err := computeFileChecksums(path) + if err != nil { + t.Fatalf("computing checksums for %s: %v", path, err) + } + return sums +} + +func readFile(t *testing.T, path string) []byte { + t.Helper() + data, err := os.ReadFile(path) + if err != nil { + t.Fatalf("reading %s: %v", path, err) + } + return data +} + +// --- Tests --- + +func TestMultipartUploadRoundTrip(t *testing.T) { + const size = 1000 + f := newFakeRegistry(t) + f.partSize = 300 // 4 parts: 300, 300, 300, 100 + f.partCount = 4 + + path := writeTempFile(t, size) + u := newMultipartUploader(multipartTestCtx(f.srv.URL), newHTTPClient(), "reg", 3) + + if err := u.upload("mypkg", "1.0.0", "sub/dir/big.bin", path, size, testSums(t, path)); err != nil { + t.Fatalf("upload: %v", err) + } + + if got := f.assembled(); !bytes.Equal(got, readFile(t, path)) { + t.Errorf("assembled object differs from source: got %d bytes, want %d", len(got), size) + } + if len(f.parts) != 4 { + t.Errorf("uploaded %d parts, want 4", len(f.parts)) + } + // The final part must be the remainder, not a full partSize. + if got := len(f.parts[4]); got != 100 { + t.Errorf("final part was %d bytes, want 100", got) + } + + // The Start body must carry exactly the documented fields; the server rejects unknown ones. + body := f.startBodies[0] + wantKeys := map[string]bool{"path": true, "size": true, "sha256": true, "sha1": true, "md5": true, "sha512": true} + for k := range body { + if !wantKeys[k] { + t.Errorf("Start body carried unexpected field %q", k) + } + } + if got, want := body["path"], "mypkg/1.0.0/sub/dir/big.bin"; got != want { + t.Errorf("Start path = %q, want %q", got, want) + } + if got, want := body["size"], float64(size); got != want { + t.Errorf("Start size = %v, want %v", got, want) + } + if body["sha256"] == "" { + t.Error("Start body had an empty sha256") + } +} + +func TestMultipartUploadDedupSkipsUpload(t *testing.T) { + f := newFakeRegistry(t) + f.dedup = true + + path := writeTempFile(t, 100) + u := newMultipartUploader(multipartTestCtx(f.srv.URL), newHTTPClient(), "reg", 2) + + if err := u.upload("mypkg", "1.0.0", "big.bin", path, 100, testSums(t, path)); err != nil { + t.Fatalf("upload: %v", err) + } + if len(f.parts) != 0 { + t.Errorf("dedup path uploaded %d parts, want 0", len(f.parts)) + } + if f.completes != 0 { + t.Errorf("dedup path called Complete %d times, want 0", f.completes) + } +} + +// noRetryHTTPClient returns a client that does not retry, unlike newHTTPClient. Tests that make +// Start fail on a retryable condition (a 5xx, or nothing listening at all) use it so they assert +// the fallback decision without also sitting through five backoff waits. +func noRetryHTTPClient() *http.Client { + return &http.Client{Timeout: 10 * time.Second} +} + +// TestMultipartUploadStartFailures pins down which Start failures hand the push back to the +// single-PUT fallback and which ones must surface. Falling back on an auth or server error would +// mask the real cause behind a second attempt that hits the same wall. +func TestMultipartUploadStartFailures(t *testing.T) { + tests := []struct { + name string + status int + code string + wantFallback bool + }{ + {"feature disabled", http.StatusNotImplemented, codeMultipartUnsupported, true}, + {"endpoint absent", http.StatusNotFound, "", true}, + {"unauthenticated", http.StatusUnauthorized, "", false}, + {"forbidden", http.StatusForbidden, "", false}, + {"server error", http.StatusInternalServerError, "", false}, + {"path conflict", http.StatusConflict, codePathConflict, false}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + f := newFakeRegistry(t) + f.startStatus = tc.status + f.startCode = tc.code + + path := writeTempFile(t, 100) + u := newMultipartUploader(multipartTestCtx(f.srv.URL), noRetryHTTPClient(), "reg", 2) + + err := u.upload("mypkg", "1.0.0", "big.bin", path, 100, testSums(t, path)) + if err == nil { + t.Fatal("upload succeeded, want an error") + } + if got := errors.Is(err, errMultipartUnsupported); got != tc.wantFallback { + t.Fatalf("errors.Is(err, errMultipartUnsupported) = %t, want %t (err = %v)", + got, tc.wantFallback, err) + } + if len(f.parts) != 0 { + t.Errorf("uploaded %d parts before giving up, want 0", len(f.parts)) + } + }) + } +} + +// TestMultipartUploadStartUnreachableFallsBack covers a Start that never gets a reply: no response +// means multipart availability is unknown, and a single PUT is a strictly simpler request that +// deserves the attempt rather than failing the push outright. +func TestMultipartUploadStartUnreachableFallsBack(t *testing.T) { + // A server that is closed before the call: its address is real but nothing is listening, so + // Start fails in transport and produces no apiError to inspect. + dead := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {})) + url := dead.URL + dead.Close() + + path := writeTempFile(t, 100) + u := newMultipartUploader(multipartTestCtx(url), noRetryHTTPClient(), "reg", 2) + + err := u.upload("mypkg", "1.0.0", "big.bin", path, 100, testSums(t, path)) + if !errors.Is(err, errMultipartUnsupported) { + t.Fatalf("upload error = %v, want errMultipartUnsupported", err) + } +} + +// TestStartFailureMeansUnsupportedIgnoresCancellation checks that a cancelled or expired command is +// not reported as a registry lacking multipart: the fallback would just repeat the failure. +func TestStartFailureMeansUnsupportedIgnoresCancellation(t *testing.T) { + cancelled, cancel := context.WithCancel(context.Background()) + cancel() + expired, cancelExpired := context.WithDeadline(context.Background(), time.Now().Add(-time.Second)) + defer cancelExpired() + + for name, ctx := range map[string]context.Context{"cancelled": cancelled, "expired": expired} { + t.Run(name, func(t *testing.T) { + err := fmt.Errorf("starting multipart upload: %w", ctx.Err()) + if startFailureMeansUnsupported(ctx, err) { + t.Error("startFailureMeansUnsupported = true, want false for an ended command context") + } + }) + } +} + +// TestStartFailureMeansUnsupportedFallsBackOnClientTimeout guards a regression found on QA: an +// unreachable registry exhausts the HTTP client's retries and then trips http.Client.Timeout, which +// reports context.DeadlineExceeded even though the command's own context is still live. Deciding +// from the error rather than the context made this refuse to fall back - the exact opposite of what +// an unreachable Start should do. +func TestStartFailureMeansUnsupportedFallsBackOnClientTimeout(t *testing.T) { + err := fmt.Errorf(`starting multipart upload: Post "http://127.0.0.1:9/pkg/a/r/uploads": %w`+ + " (Client.Timeout exceeded while awaiting headers)", context.DeadlineExceeded) + if !startFailureMeansUnsupported(context.Background(), err) { + t.Error("startFailureMeansUnsupported = false, want true: the command context is still live, " + + "so a client-side timeout means multipart availability is unknown") + } +} + +func TestMultipartUploadFetchesPartsBeyondStartBatch(t *testing.T) { + const size = 500 + f := newFakeRegistry(t) + f.partSize = 100 + f.partCount = 5 + f.presignInStart = 2 // parts 3..5 must be fetched separately + + path := writeTempFile(t, size) + u := newMultipartUploader(multipartTestCtx(f.srv.URL), newHTTPClient(), "reg", 1) + + if err := u.upload("mypkg", "1.0.0", "big.bin", path, size, testSums(t, path)); err != nil { + t.Fatalf("upload: %v", err) + } + if got := f.assembled(); !bytes.Equal(got, readFile(t, path)) { + t.Error("assembled object differs from source") + } + if len(f.partFetches) == 0 { + t.Fatal("expected at least one parts fetch for the un-presigned parts") + } + // The fetch window must be clamped to the real part count, not extended past it. + for _, fetch := range f.partFetches { + if fetch[1] > f.partCount { + t.Errorf("parts fetch range %v exceeds partCount %d", fetch, f.partCount) + } + } +} + +func TestMultipartUploadRefreshesNearlyExpiredURLs(t *testing.T) { + const size = 200 + f := newFakeRegistry(t) + f.partSize = 100 + f.partCount = 2 + f.expiredURLs = true // Start's URLs are inside presignRefreshWindow + + path := writeTempFile(t, size) + u := newMultipartUploader(multipartTestCtx(f.srv.URL), newHTTPClient(), "reg", 1) + + if err := u.upload("mypkg", "1.0.0", "big.bin", path, size, testSums(t, path)); err != nil { + t.Fatalf("upload: %v", err) + } + if len(f.partFetches) == 0 { + t.Error("expected the near-expiry URLs to be refreshed before use") + } + if got := f.assembled(); !bytes.Equal(got, readFile(t, path)) { + t.Error("assembled object differs from source") + } +} + +func TestMultipartUploadRetriesMissingParts(t *testing.T) { + const size = 300 + f := newFakeRegistry(t) + f.partSize = 100 + f.partCount = 3 + f.missingOnce = []int{2} + + path := writeTempFile(t, size) + u := newMultipartUploader(multipartTestCtx(f.srv.URL), newHTTPClient(), "reg", 2) + + if err := u.upload("mypkg", "1.0.0", "big.bin", path, size, testSums(t, path)); err != nil { + t.Fatalf("upload: %v", err) + } + if f.completes != 2 { + t.Errorf("Complete called %d times, want 2 (one rejection then success)", f.completes) + } + if got := f.assembled(); !bytes.Equal(got, readFile(t, path)) { + t.Error("assembled object differs from source after re-uploading the missing part") + } +} + +func TestMultipartUploadRetriesTransientPartFailure(t *testing.T) { + const size = 200 + f := newFakeRegistry(t) + f.partSize = 100 + f.partCount = 2 + f.failPartsOnce = map[int]bool{2: true} + f.partFailStatus = http.StatusServiceUnavailable + + path := writeTempFile(t, size) + u := newMultipartUploader(multipartTestCtx(f.srv.URL), newHTTPClient(), "reg", 2) + + if err := u.upload("mypkg", "1.0.0", "big.bin", path, size, testSums(t, path)); err != nil { + t.Fatalf("upload: %v", err) + } + if got := f.assembled(); !bytes.Equal(got, readFile(t, path)) { + t.Error("assembled object differs from source after retrying a part") + } +} + +func TestMultipartUploadAbortsSessionOnFatalPartFailure(t *testing.T) { + const size = 200 + f := newFakeRegistry(t) + f.partSize = 100 + f.partCount = 2 + // 400 is not retryable, so the part fails on its first attempt and the upload gives up fast. + f.failPartsOnce = map[int]bool{1: true, 2: true} + f.partFailStatus = http.StatusBadRequest + + path := writeTempFile(t, size) + u := newMultipartUploader(multipartTestCtx(f.srv.URL), newHTTPClient(), "reg", 2) + + if err := u.upload("mypkg", "1.0.0", "big.bin", path, size, testSums(t, path)); err == nil { + t.Fatal("upload succeeded, want an error") + } + if !f.aborted { + t.Error("session was not aborted after a fatal part failure") + } + if f.completes != 0 { + t.Errorf("Complete called %d times after a part failure, want 0", f.completes) + } +} + +func TestMultipartUploadSurfacesFinalizeFailure(t *testing.T) { + const size = 100 + f := newFakeRegistry(t) + f.partSize = 100 + f.partCount = 1 + f.finalStatus = uploadStatusFailed + f.finalError = "sha256 mismatch" + + path := writeTempFile(t, size) + u := newMultipartUploader(multipartTestCtx(f.srv.URL), newHTTPClient(), "reg", 2) + + err := u.upload("mypkg", "1.0.0", "big.bin", path, size, testSums(t, path)) + if err == nil { + t.Fatal("upload succeeded, want a finalize failure") + } + if !strings.Contains(err.Error(), "sha256 mismatch") { + t.Errorf("error = %v, want it to mention the server's reason", err) + } + if !f.aborted { + t.Error("expected a best-effort abort after a failed finalize") + } +} + +func TestValidateSession(t *testing.T) { + u := &multipartUploader{} + tests := []struct { + name string + session startUploadResponse + size int64 + wantErr bool + }{ + {"exact multiple", startUploadResponse{UploadID: "x", PartSize: 100, PartCount: 2}, 200, false}, + {"partial final part", startUploadResponse{UploadID: "x", PartSize: 100, PartCount: 3}, 250, false}, + {"missing upload id", startUploadResponse{PartSize: 100, PartCount: 2}, 200, true}, + {"zero part size", startUploadResponse{UploadID: "x", PartSize: 0, PartCount: 2}, 200, true}, + {"zero part count", startUploadResponse{UploadID: "x", PartSize: 100, PartCount: 0}, 200, true}, + {"too few parts", startUploadResponse{UploadID: "x", PartSize: 100, PartCount: 1}, 250, true}, + {"one part too many", startUploadResponse{UploadID: "x", PartSize: 100, PartCount: 4}, 250, true}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + err := u.validateSession(&tc.session, tc.size) + if (err != nil) != tc.wantErr { + t.Errorf("validateSession(%d bytes) error = %v, wantErr %v", tc.size, err, tc.wantErr) + } + }) + } +} + +func TestPartURLExpired(t *testing.T) { + tests := []struct { + name string + expiresAt string + want bool + }{ + {"absent expiry is treated as expired", "", true}, + {"unparseable expiry is treated as expired", "not-a-time", true}, + {"inside refresh window", time.Now().Add(10 * time.Second).Format(presignExpiryLayout), true}, + {"already past", time.Now().Add(-time.Minute).Format(presignExpiryLayout), true}, + {"plenty of validity left", time.Now().Add(time.Hour).Format(presignExpiryLayout), false}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + if got := (partURL{ExpiresAt: tc.expiresAt}).expired(); got != tc.want { + t.Errorf("expired() = %v, want %v", got, tc.want) + } + }) + } +} + +func TestMissingParts(t *testing.T) { + err := &apiError{ + StatusCode: http.StatusConflict, + Code: codeIncompleteUpload, + Values: map[string]any{"code": codeIncompleteUpload, "missingParts": []any{float64(2), float64(5)}}, + } + got := missingParts(err) + if len(got) != 2 || got[0] != 2 || got[1] != 5 { + t.Errorf("missingParts = %v, want [2 5]", got) + } + + if got := missingParts(fmt.Errorf("plain error")); got != nil { + t.Errorf("missingParts on a non-apiError = %v, want nil", got) + } + // SIZE_MISMATCH shares the 409 status but carries no part list. + sizeErr := &apiError{StatusCode: http.StatusConflict, Code: codeSizeMismatch, Values: map[string]any{}} + if got := missingParts(sizeErr); got != nil { + t.Errorf("missingParts without the key = %v, want nil", got) + } +} + +func TestDoJSONRequestParsesErrorCode(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + writeAPIError(w, http.StatusConflict, map[string]any{"code": codePathConflict}) + })) + defer srv.Close() + + req, err := http.NewRequest(http.MethodGet, srv.URL, nil) + if err != nil { + t.Fatalf("building request: %v", err) + } + + status, _, err := doJSONRequest(&http.Client{}, req) + if status != http.StatusConflict { + t.Errorf("status = %d, want 409", status) + } + if got := apiErrorCode(err); got != codePathConflict { + t.Errorf("apiErrorCode = %q, want %q", got, codePathConflict) + } + if got := apiErrorStatus(err); got != http.StatusConflict { + t.Errorf("apiErrorStatus = %d, want 409", got) + } + if got := apiErrorCode(fmt.Errorf("plain")); got != "" { + t.Errorf("apiErrorCode on a plain error = %q, want empty", got) + } +} + +func TestDoJSONRequestReportsSuccessStatus(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusAccepted) + })) + defer srv.Close() + + req, _ := http.NewRequest(http.MethodPost, srv.URL, nil) + status, body, err := doJSONRequest(&http.Client{}, req) + if err != nil { + t.Fatalf("doJSONRequest: %v", err) + } + // 202 must be distinguishable from 200: the two mean different things at Complete. + if status != http.StatusAccepted { + t.Errorf("status = %d, want 202", status) + } + if len(body) != 0 { + t.Errorf("body = %q, want empty", body) + } +} diff --git a/modules/har/pkg/har/push_generic.go b/modules/har/pkg/har/push_generic.go index 9808dc1..4c844f6 100644 --- a/modules/har/pkg/har/push_generic.go +++ b/modules/har/pkg/har/push_generic.go @@ -4,6 +4,7 @@ package har import ( + "errors" "fmt" "net/http" "os" @@ -21,7 +22,9 @@ import ( // harness push artifact:generic [...] --name [--version v] // // Each may be a file or a directory. Directories are walked recursively. -// Files are uploaded to: {registryURL}/pkg/{accountID}/{registry}/generic/{name}/{version}/{relPath} +// Files are uploaded to: {registryURL}/pkg/{accountID}/{registry}/files/{name}/{version}/{relPath} +// Files at or above multipartThreshold are uploaded in parallel parts instead, falling back to the +// single-request upload when the registry does not support multipart. func pushGenericArtifact(ctx *cmdctx.Ctx) error { if len(ctx.Args) == 0 { return fmt.Errorf("push generic artifact requires at least one file or directory path") @@ -39,6 +42,16 @@ func pushGenericArtifact(ctx *cmdctx.Ctx) error { includeHidden := cmdctx.GetBool(ctx.FlagValues, "include-hidden") + maxConcurrentFiles := cmdctx.GetInt(ctx.FlagValues, "max-concurrent-uploads") + if maxConcurrentFiles <= 0 { + maxConcurrentFiles = defaultMaxConcurrentUploads + } + maxConcurrentParts := cmdctx.GetInt(ctx.FlagValues, "max-concurrent-parts") + if maxConcurrentParts <= 0 { + maxConcurrentParts = defaultMaxConcurrentParts + } + noMultipart := cmdctx.GetBool(ctx.FlagValues, "no-multipart") + fmt.Fprintf(os.Stderr, "Scanning %d input(s) ...\n", len(ctx.Args)) jobs, totalFiles, totalBytes, err := collectGenericJobs(ctx.Args, includeHidden) @@ -54,7 +67,11 @@ func pushGenericArtifact(ctx *cmdctx.Ctx) error { client := newHTTPClient() - sem := make(chan struct{}, defaultMaxConcurrentUploads) + // The part semaphore is created once here, not per file: file-level concurrency multiplied by + // per-file part fan-out would otherwise open (files x parts) connections at the same time. + uploader := newMultipartUploader(ctx, client, registry, maxConcurrentParts) + + sem := make(chan struct{}, maxConcurrentFiles) var wg sync.WaitGroup errs := make([]error, len(jobs)) @@ -65,8 +82,10 @@ func pushGenericArtifact(ctx *cmdctx.Ctx) error { sem <- struct{}{} defer func() { <-sem }() - fmt.Fprintf(os.Stderr, "Uploading %s ...\n", job.relPath) - if uploadErr := genericPutFile(ctx, client, registry, name, version, job.relPath, job.localPath); uploadErr != nil { + fmt.Fprintf(os.Stderr, "Uploading %s (%s) ...\n", job.relPath, formatBytes(job.size)) + if uploadErr := uploadGenericFile( + ctx, uploader, client, registry, name, version, job, noMultipart, + ); uploadErr != nil { errs[i] = fmt.Errorf("failed to upload %s: %w", job.relPath, uploadErr) } }(i, job) @@ -202,10 +221,46 @@ func walkDir(srcDir string, includeHidden bool) ([]genericUploadJob, int64, erro return jobs, totalBytes, nil } -// genericPutFile uploads a single file via PUT. +// uploadGenericFile uploads one file, choosing multipart for large files and a single PUT +// otherwise. If the registry does not support multipart (feature disabled or endpoint absent) the +// multipart attempt reports that before sending any bytes, so falling back here costs nothing. +func uploadGenericFile( + ctx *cmdctx.Ctx, + uploader *multipartUploader, + client *http.Client, + registry, name, version string, + job genericUploadJob, + noMultipart bool, +) error { + // Hashed once here and handed to whichever path runs. Both need the same digests, and the + // multipart-then-fallback route would otherwise read a large file twice just to hash it. + sums, err := computeFileChecksums(job.localPath) + if err != nil { + return fmt.Errorf("computing checksums for %q: %w", job.localPath, err) + } + + if !noMultipart && job.size >= multipartThreshold { + mpErr := uploader.upload(name, version, job.relPath, job.localPath, job.size, sums) + if mpErr == nil { + return nil + } + if !errors.Is(mpErr, errMultipartUnsupported) { + return mpErr + } + fmt.Fprintf(os.Stderr, + "Multipart upload unavailable for %s, falling back to a single upload ...\n", job.relPath) + } + return genericPutFile(ctx, client, registry, name, version, job.relPath, job.localPath, sums) +} + +// genericPutFile uploads a single file via PUT. sums are the file's digests, already computed by the +// caller so that a multipart attempt and this fallback do not each hash the same file. // -// URL: {registryURL}/pkg/{accountID}/{registry}/generic/{name}/{version}/{relPath} -func genericPutFile(ctx *cmdctx.Ctx, client *http.Client, registry, name, version, relPath, localPath string) error { +// URL: {registryURL}/pkg/{accountID}/{registry}/files/{name}/{version}/{relPath} +func genericPutFile( + ctx *cmdctx.Ctx, client *http.Client, registry, name, version, relPath, localPath string, + sums fileChecksums, +) error { f, err := os.Open(localPath) if err != nil { return fmt.Errorf("opening %q: %w", localPath, err) @@ -217,8 +272,7 @@ func genericPutFile(ctx *cmdctx.Ctx, client *http.Client, registry, name, versio return fmt.Errorf("stat %q: %w", localPath, err) } - subpath := fmt.Sprintf("%s/files/%s/%s/%s", registry, name, version, relPath) - uploadURL, err := buildPkgURL(ctx.Auth.RegistryURL, ctx.Auth.AccountID, subpath) + uploadURL, err := genericFileURL(ctx, registry, name, version, relPath) if err != nil { return err } @@ -228,6 +282,7 @@ func genericPutFile(ctx *cmdctx.Ctx, client *http.Client, registry, name, versio return fmt.Errorf("building request: %w", err) } setAuthHeader(req, ctx.Auth) + setChecksumHeaders(req.Header, sums) req.Header.Set("Content-Type", "application/octet-stream") req.ContentLength = fi.Size() @@ -237,6 +292,12 @@ func genericPutFile(ctx *cmdctx.Ctx, client *http.Client, registry, name, versio return nil } +// genericFileURL builds the single-PUT upload URL for one generic file. +func genericFileURL(ctx *cmdctx.Ctx, registry, name, version, relPath string) (string, error) { + subpath := fmt.Sprintf("%s/files/%s/%s/%s", registry, name, version, relPath) + return buildPkgURL(ctx.Auth.RegistryURL, ctx.Auth.AccountID, subpath) +} + // formatBytes returns a human-readable byte size string. func formatBytes(b int64) string { const unit = 1024 diff --git a/pkg/spec/har.spec.yaml b/pkg/spec/har.spec.yaml index 335d5be..c475e32 100644 --- a/pkg/spec/har.spec.yaml +++ b/pkg/spec/har.spec.yaml @@ -212,6 +212,12 @@ commands: description: "Artifact version (default: 1.0.0)" - name: include-hidden description: "Include hidden files and directories (names starting with '.') when walking directory inputs" + - name: max-concurrent-uploads + description: "Maximum number of files uploaded concurrently (default: 4, use 1 for sequential)" + - name: max-concurrent-parts + description: "Maximum number of multipart parts uploaded concurrently across all files (default: 8)" + - name: no-multipart + description: "Disable parallel multipart upload and always use a single upload request per file" - command: push artifact:maven verb: push