feat: add dynamic timeout retry with file-size-based parameters#468
feat: add dynamic timeout retry with file-size-based parameters#468
Conversation
Summary of ChangesHello, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This pull request significantly enhances the robustness of file transfer operations by introducing a dynamic and intelligent retry mechanism. It addresses previous limitations where fixed retry parameters and cascading cancellations led to frequent failures, especially with large files over unreliable networks. The new system adapts retry behavior based on file size and isolates failures, ensuring that individual transient errors do not disrupt the entire transfer process, thereby improving overall reliability and user experience. Highlights
Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for GitHub and other Google products, sign up here. Footnotes
|
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces a robust dynamic retry mechanism with file-size-based parameters, which is a significant improvement for handling transient network errors during large file transfers. The new pkg/retrypolicy is well-designed and thoroughly tested. A key architectural improvement is replacing cascading errgroup.WithContext with independent per-goroutine retries, preventing a single failure from terminating all concurrent operations. The addition of CLI flags for retry control and progress bar updates on retry significantly enhance usability. The changes are consistently applied across all data transfer paths. I have a couple of suggestions for improvement in pkg/backend/push.go to ensure a consistent user experience during retries.
| if err := retrypolicy.Do(ctx, func(rctx context.Context) error { | ||
| return pushIfNotExist(rctx, pb, internalpb.NormalizePrompt("Copying config"), src, dst, manifest.Config, repo, tag) | ||
| }, retrypolicy.DoOpts{ | ||
| FileSize: manifest.Config.Size, | ||
| FileName: "config", | ||
| Config: &cfg.RetryConfig, | ||
| }); err != nil { | ||
| return fmt.Errorf("failed to push config to remote: %w", err) | ||
| } |
There was a problem hiding this comment.
The retry logic for pushing the config object is missing an OnRetry handler. This is inconsistent with the retry logic for layers in this file and for all components in pkg/backend/pull.go. Adding an OnRetry handler will provide better visual feedback to the user during backoff periods, improving the user experience.
if err := retrypolicy.Do(ctx, func(rctx context.Context) error {
return pushIfNotExist(rctx, pb, internalpb.NormalizePrompt("Copying config"), src, dst, manifest.Config, repo, tag)
}, retrypolicy.DoOpts{
FileSize: manifest.Config.Size,
FileName: "config",
Config: &cfg.RetryConfig,
OnRetry: func(attempt uint, reason string, backoff time.Duration) {
prompt := fmt.Sprintf("%s (retry %d, %s, waiting %s)",
internalpb.NormalizePrompt("Copying config"), attempt, reason, backoff.Truncate(time.Second))
pb.Add(prompt, manifest.Config.Digest.String(), manifest.Config.Size, nil)
},
}); err != nil {
return fmt.Errorf("failed to push config to remote: %w", err)
}| // copy the manifest. | ||
| if err := retry.Do(func() error { | ||
| return pushIfNotExist(ctx, pb, internalpb.NormalizePrompt("Copying manifest"), src, dst, ocispec.Descriptor{ | ||
| if err := retrypolicy.Do(ctx, func(rctx context.Context) error { | ||
| return pushIfNotExist(rctx, pb, internalpb.NormalizePrompt("Copying manifest"), src, dst, ocispec.Descriptor{ | ||
| MediaType: manifest.MediaType, | ||
| Size: int64(len(manifestRaw)), | ||
| Digest: godigest.FromBytes(manifestRaw), | ||
| Data: manifestRaw, | ||
| }, repo, tag) | ||
| }, append(defaultRetryOpts, retry.Context(ctx))...); err != nil { | ||
| }, retrypolicy.DoOpts{ | ||
| FileSize: int64(len(manifestRaw)), | ||
| FileName: "manifest", | ||
| Config: &cfg.RetryConfig, | ||
| }); err != nil { | ||
| return fmt.Errorf("failed to push manifest to remote: %w", err) | ||
| } |
There was a problem hiding this comment.
Similar to the config push, the retry logic for pushing the manifest is missing an OnRetry handler. For consistency and better user experience, a handler should be added to update the progress bar during retries. I've also refactored the descriptor creation to make it accessible in the OnRetry callback.
// copy the manifest.
manifestDesc := ocispec.Descriptor{
MediaType: manifest.MediaType,
Size: int64(len(manifestRaw)),
Digest: godigest.FromBytes(manifestRaw),
Data: manifestRaw,
}
if err := retrypolicy.Do(ctx, func(rctx context.Context) error {
return pushIfNotExist(rctx, pb, internalpb.NormalizePrompt("Copying manifest"), src, dst, manifestDesc, repo, tag)
}, retrypolicy.DoOpts{
FileSize: manifestDesc.Size,
FileName: "manifest",
Config: &cfg.RetryConfig,
OnRetry: func(attempt uint, reason string, backoff time.Duration) {
prompt := fmt.Sprintf("%s (retry %d, %s, waiting %s)",
internalpb.NormalizePrompt("Copying manifest"), attempt, reason, backoff.Truncate(time.Second))
pb.Add(prompt, manifestDesc.Digest.String(), manifestDesc.Size, nil)
},
}); err != nil {
return fmt.Errorf("failed to push manifest to remote: %w", err)
}
aftersnow
force-pushed
the
worktree-timeout-retry
branch
from
9f96098 to
1369e47
Compare
|
/gemini review |
|
@codex-code-assist review |
There was a problem hiding this comment.
Code Review
This pull request introduces a significant improvement to the application's resilience by implementing a dynamic, file-size-based retry mechanism for data transfers. The new pkg/retrypolicy is well-designed and replaces the previous fixed retry logic. A key architectural change is the move away from cascading cancellations in errgroup, which will prevent a single failure from halting all concurrent operations. The addition of CLI flags for retry control is also a valuable feature. My review focuses on a critical race condition in the progress bar handling, a minor bug in the retry backoff calculation for logging, and opportunities to reduce code duplication for better maintainability.
| p.mu.RLock() | ||
| existing := p.bars[name] | ||
| p.mu.RUnlock() | ||
|
|
||
| // If the bar already exists, just reset its message. | ||
| if existing != nil { | ||
| existing.msg = fmt.Sprintf("%s %s", prompt, name) | ||
| existing.Bar.SetCurrent(0) | ||
| return | ||
| } | ||
|
|
||
| // Create a new placeholder bar. | ||
| p.Add(prompt, name, size, nil) |
There was a problem hiding this comment.
There is a data race condition here. The existing.msg field is being modified without a lock, while it can be concurrently read by the progress bar's rendering goroutine. This can lead to unpredictable behavior or crashes.
To fix this, you should use a write lock to protect both the read from the p.bars map and the subsequent write to the msg field. The lock should be released before calling p.Add to avoid deadlocks, as p.Add acquires its own locks.
| p.mu.RLock() | |
| existing := p.bars[name] | |
| p.mu.RUnlock() | |
| // If the bar already exists, just reset its message. | |
| if existing != nil { | |
| existing.msg = fmt.Sprintf("%s %s", prompt, name) | |
| existing.Bar.SetCurrent(0) | |
| return | |
| } | |
| // Create a new placeholder bar. | |
| p.Add(prompt, name, size, nil) | |
| p.mu.Lock() | |
| if existing, ok := p.bars[name]; ok { | |
| // If the bar already exists, just reset its message. | |
| existing.msg = fmt.Sprintf("%s %s", prompt, name) | |
| existing.Bar.SetCurrent(0) | |
| p.mu.Unlock() | |
| return | |
| } | |
| p.mu.Unlock() | |
| // Create a new placeholder bar. | |
| p.Add(prompt, name, size, nil) |
| var annoFilepath string | ||
| if layer.Annotations != nil { | ||
| if layer.Annotations[modelspec.AnnotationFilepath] != "" { | ||
| annoFilepath = layer.Annotations[modelspec.AnnotationFilepath] | ||
| } else { | ||
| annoFilepath = layer.Annotations[legacymodelspec.AnnotationFilepath] | ||
| } | ||
| } |
There was a problem hiding this comment.
This logic to extract the filepath from annotations is duplicated in pkg/backend/fetch_by_d7y.go and pkg/backend/pull_by_d7y.go. To improve maintainability and reduce code duplication, consider extracting this into a shared helper function within the backend package.
For example, you could define this function:
func getAnnotationFilepath(annotations map[string]string) string {
if annotations == nil {
return ""
}
if path := annotations[modelspec.AnnotationFilepath]; path != "" {
return path
}
return annotations[legacymodelspec.AnnotationFilepath]
}And then call it here.
| var annoFilepath string | |
| if layer.Annotations != nil { | |
| if layer.Annotations[modelspec.AnnotationFilepath] != "" { | |
| annoFilepath = layer.Annotations[modelspec.AnnotationFilepath] | |
| } else { | |
| annoFilepath = layer.Annotations[legacymodelspec.AnnotationFilepath] | |
| } | |
| } | |
| annoFilepath := getAnnotationFilepath(layer.Annotations) |
| var annoFilepath string | ||
| if desc.Annotations != nil { | ||
| if desc.Annotations[modelspec.AnnotationFilepath] != "" { | ||
| annoFilepath = desc.Annotations[modelspec.AnnotationFilepath] | ||
| } else { | ||
| annoFilepath = desc.Annotations[legacymodelspec.AnnotationFilepath] | ||
| } | ||
| } |
There was a problem hiding this comment.
This logic to extract the filepath from annotations is duplicated in pkg/backend/fetch.go and pkg/backend/pull_by_d7y.go. To improve maintainability and reduce code duplication, consider extracting this into a shared helper function within the backend package.
For example, you could define this function:
func getAnnotationFilepath(annotations map[string]string) string {
if annotations == nil {
return ""
}
if path := annotations[modelspec.AnnotationFilepath]; path != "" {
return path
}
return annotations[legacymodelspec.AnnotationFilepath]
}And then call it here.
| var annoFilepath string | |
| if desc.Annotations != nil { | |
| if desc.Annotations[modelspec.AnnotationFilepath] != "" { | |
| annoFilepath = desc.Annotations[modelspec.AnnotationFilepath] | |
| } else { | |
| annoFilepath = desc.Annotations[legacymodelspec.AnnotationFilepath] | |
| } | |
| } | |
| annoFilepath := getAnnotationFilepath(desc.Annotations) |
| var annoFilepath string | ||
| if desc.Annotations != nil { | ||
| if desc.Annotations[modelspec.AnnotationFilepath] != "" { | ||
| annoFilepath = desc.Annotations[modelspec.AnnotationFilepath] | ||
| } else { | ||
| annoFilepath = desc.Annotations[legacymodelspec.AnnotationFilepath] | ||
| } | ||
| } |
There was a problem hiding this comment.
This logic to extract the filepath from annotations is duplicated in pkg/backend/fetch.go and pkg/backend/fetch_by_d7y.go. To improve maintainability and reduce code duplication, consider extracting this into a shared helper function within the backend package.
For example, you could define this function:
func getAnnotationFilepath(annotations map[string]string) string {
if annotations == nil {
return ""
}
if path := annotations[modelspec.AnnotationFilepath]; path != "" {
return path
}
return annotations[legacymodelspec.AnnotationFilepath]
}And then call it here.
| var annoFilepath string | |
| if desc.Annotations != nil { | |
| if desc.Annotations[modelspec.AnnotationFilepath] != "" { | |
| annoFilepath = desc.Annotations[modelspec.AnnotationFilepath] | |
| } else { | |
| annoFilepath = desc.Annotations[legacymodelspec.AnnotationFilepath] | |
| } | |
| } | |
| annoFilepath := getAnnotationFilepath(desc.Annotations) |
| return retryable | ||
| }), | ||
| retry.OnRetry(func(n uint, err error) { | ||
| backoff := computeBackoff(n+1, initialDelay, maxBackoff) |
There was a problem hiding this comment.
The computeBackoff function is being called with n+1, but it seems to expect the 1-based retry attempt number, which is n. The retry-go library's OnRetry callback provides n as the 1-based retry attempt number. Passing n+1 causes the logged backoff duration to be one step ahead of the actual backoff being used by the retry library.
To ensure the logged backoff duration is accurate, you should call computeBackoff with n.
| backoff := computeBackoff(n+1, initialDelay, maxBackoff) | |
| backoff := computeBackoff(n, initialDelay, maxBackoff) |
| if attempt == 0 { | ||
| return initial | ||
| } |
There was a problem hiding this comment.
This check for attempt == 0 appears to be dead code. The retry-go library's OnRetry callback provides n as a 1-based retry attempt number, so n will always be 1 or greater. Since computeBackoff is called with n (or n+1), the attempt parameter will never be zero. This block can be safely removed.
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: eb1fc1ebc2
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
pkg/retrypolicy/retrypolicy.go
Outdated
| // Unknown errors default to retryable. | ||
| return true |
There was a problem hiding this comment.
This fallback makes every unexpected error retryable, and the new call sites now feed retrypolicy.Do with arbitrary local failures from paths like builder.BuildLayer and extractLayer. A permanent problem such as permission denied or no space left on device will now sit in the retry loop until the size-based deadline expires (10 minutes for small files, 60 minutes for blobs ≥10 GiB), so broken builds/pulls can appear hung for a long time instead of failing promptly.
Useful? React with 👍 / 👎.
| retryErr := retrypolicy.Do(ctx, func(retryCtx context.Context) error { | ||
| logrus.Debugf("pull: processing layer %s", layer.Digest) | ||
| // call the before hook. | ||
| cfg.Hooks.BeforePullLayer(layer, manifest) |
There was a problem hiding this comment.
The per-attempt retryCtx created by retrypolicy.Do is not actually used for the layer transfer here: fn closes over the outer command context, so both pullIfNotExist and pullAndExtractFromRemote still run with ctx. In the normal pull path, any registry read/write that only aborts on context cancellation can therefore ignore --retry-max-time (or the dynamic size-based limit) and block the whole layer instead of timing out and retrying.
Useful? React with 👍 / 👎.
|
/gemini review |
|
@codex-code-assist review |
There was a problem hiding this comment.
Code Review
This pull request introduces a robust and dynamic retry mechanism based on file size, significantly improving the reliability of data transfer operations. The refactoring to use a dedicated pkg/retrypolicy package centralizes retry logic, and the move to independent per-goroutine retries effectively addresses the cascading cancellation issue. The addition of CLI flags for retry configuration provides users with greater control. The comprehensive unit tests for the new retry policy are commendable, ensuring the correctness of the dynamic parameter calculations and error handling.
Overall, this is a well-designed and implemented feature that directly tackles a critical pain point for large file transfers.
| OnRetry: func(attempt uint, reason string, backoff time.Duration) { | ||
| prompt := fmt.Sprintf("%s (retry %d, %s, waiting %s)", | ||
| internalpb.NormalizePrompt("Copying blob"), attempt, reason, backoff.Truncate(time.Second)) | ||
| pb.Add(prompt, layer.Digest.String(), layer.Size, nil) | ||
| }, |
There was a problem hiding this comment.
For updating the progress bar during a retry, pb.Placeholder is more semantically appropriate than pb.Add. The Placeholder method is designed to update an existing bar's message and reset its current progress to zero, which is ideal for indicating a backoff period before the next attempt. This approach is already used consistently in pull.go and fetch.go.
Please apply this change to the OnRetry callbacks for layers, config, and manifest pushes for consistency and clearer UX.
For example, the OnRetry for layers could be updated as follows:
| OnRetry: func(attempt uint, reason string, backoff time.Duration) { | |
| prompt := fmt.Sprintf("%s (retry %d, %s, waiting %s)", | |
| internalpb.NormalizePrompt("Copying blob"), attempt, reason, backoff.Truncate(time.Second)) | |
| pb.Add(prompt, layer.Digest.String(), layer.Size, nil) | |
| }, | |
| OnRetry: func(attempt uint, reason string, backoff time.Duration) { | |
| prompt := fmt.Sprintf("%s (retry %d, %s, waiting %s)", | |
| internalpb.NormalizePrompt("Copying blob"), attempt, reason, backoff.Truncate(time.Second)) | |
| pb.Placeholder(layer.Digest.String(), prompt, layer.Size) | |
| }, |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: f3fbdfe46a
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| g.Wait() | ||
| if len(errs) > 0 { | ||
| return fmt.Errorf("failed to push blob to remote: %w", errors.Join(errs...)) |
There was a problem hiding this comment.
When ctx is cancelled after some layer uploads finish but before the remaining goroutines start (common when cfg.Concurrency < len(manifest.Layers)), those queued workers return ctx.Err() from the early select, but g.Wait() is ignored here. If the already-running uploads succeeded, errs stays empty and Push continues on to the config/manifest uploads, so the command can report success after skipping layers and may publish an incomplete artifact.
Useful? React with 👍 / 👎.
| _ = g.Wait() | ||
| if len(errs) > 0 { | ||
| return fmt.Errorf("failed to pull blob to local: %w", errors.Join(errs...)) |
There was a problem hiding this comment.
If the command context is cancelled between layer batches, the queued workers exit via the top-of-goroutine ctx.Done() check, but this _ = g.Wait() drops that cancellation and errs remains empty. In extract-from-remote mode that means pull can return success after extracting only a subset of files; in normal mode it can also proceed to fetch config/manifest after skipping some layers, leaving an incomplete local artifact.
Useful? React with 👍 / 👎.
| select { | ||
| case <-ctx.Done(): | ||
| return ctx.Err() | ||
| return nil |
There was a problem hiding this comment.
Returning nil here means any worker that never starts because the build context was already cancelled is treated as a success. Combined with the ignored eg.Wait() result below, cancelling a build after some files finish but before the rest start lets Process return no error with only a partial descriptors set, so Build can emit an apparently successful artifact that is missing files from the worktree.
Useful? React with 👍 / 👎.
|
@codex-code-assist review |
aftersnow
force-pushed
the
worktree-timeout-retry
branch
from
6ec310c to
1f6c0f5
Compare
…ameters Signed-off-by: Zhao Chen <zhaochen.zju@gmail.com>
…ld/fetch Signed-off-by: Zhao Chen <zhaochen.zju@gmail.com>
…ading cancellation Signed-off-by: Zhao Chen <zhaochen.zju@gmail.com>
Signed-off-by: Zhao Chen <zhaochen.zju@gmail.com>
Signed-off-by: Zhao Chen <zhaochen.zju@gmail.com>
Signed-off-by: Zhao Chen <zhaochen.zju@gmail.com>
…artifacts Signed-off-by: Zhao Chen <zhaochen.zju@gmail.com>
… tests Signed-off-by: Zhao Chen <zhaochen.zju@gmail.com>
aftersnow
force-pushed
the
worktree-timeout-retry
branch
from
1f6c0f5 to
fcdbcb0
Compare
1 participant
Summary
pkg/retrypolicypackage with dynamic retry parameters based on file size (small files: 10min/1min backoff, large files ≥10GB: 60min/10min backoff, linear interpolation between)errgroup.WithContextcancellation with independent per-goroutine retry — one layer failure no longer cancels all others--no-retryand--retry-max-timeCLI flags to push/pull/build/fetch commandsMotivation
When pushing large model files (multi-GB) to OCI registries backed by rate-limited storage (e.g., Harbor + OSS),
i/o timeouterrors frequently occur. The previous retry mechanism had two problems:errgroup.WithContextmeant one timeout killed all in-flight transfersChanges
pkg/retrypolicy/Do(),IsRetryable(),ShortReason(),computeDynamicParams()with 98.8% test coveragepkg/backend/push.goretrypolicy.Dofor config/manifestpkg/backend/pull.gopb.Placeholder()for retry progress displaypkg/backend/fetch.gopkg/backend/build.gopkg/backend/processor/base.gopkg/backend/pull_by_d7y.go,fetch_by_d7y.gocmd/{push,pull,build,fetch}.go--no-retry,--retry-max-timeflagsinternal/pb/pb.goPlaceholder()method for retry backoff displaypkg/backend/retry.godefaultRetryOpts)Test plan
pkg/retrypolicy/retrypolicy_test.go(98.8% coverage)go vet ./...passes clean🤖 Generated with Claude Code