diff --git a/internal/corrosion/client.go b/internal/corrosion/client.go index 543e47ee..1e114c2b 100644 --- a/internal/corrosion/client.go +++ b/internal/corrosion/client.go @@ -19,10 +19,10 @@ import ( const ( // http2ConnectTimeout is the maximum amount of time an HTTP2 client will wait for a connection to be established. http2ConnectTimeout = 3 * time.Second - // http2MaxRetryTime is the maximum amount of time an HTTP2 client will retry a request. - http2MaxRetryTime = 10 * time.Second - // resubscribeMaxRetryTime is the maximum amount of time an API client will retry resubscribing to a query after - // an error occurs. + // http2MaxRetryTime bounds the transport retry of a single request. Kept short so it only absorbs transient + // blips. Retrying during a Corrosion outage is the job of the higher-level recovery loops. + http2MaxRetryTime = 2 * time.Second + // resubscribeMaxRetryTime bounds resubscribing to a query after an error. resubscribeMaxRetryTime = 60 * time.Second ) @@ -35,10 +35,10 @@ type APIClient struct { // NewAPIClient creates a new Corrosion API client. The bearerToken is sent in the Authorization header of every // request to authenticate against Corrosion API. -// The client retries on network errors using an exponential backoff policy with a maximum interval of 1 second and -// a maximum elapsed time of 10 seconds. -// It automatically resubscribes to active subscriptions if an error occurs using an exponential backoff policy with a -// maximum interval of 1 second and a maximum elapsed time of 60 seconds. +// +// Retries are split by failure mode: the transport briefly retries a single request on transient network errors, while +// subscriptions resubscribe from the last change ID when their stream breaks and own the wait while Corrosion is down. +// // Use the WithHTTP2Client option to provide a custom HTTP client and the WithResubscribeBackoff option to change the // backoff policy for resubscribing to a query. func NewAPIClient(addr netip.AddrPort, bearerToken string, opts ...APIClientOption) (*APIClient, error) { @@ -61,7 +61,6 @@ func NewAPIClient(addr netip.AddrPort, bearerToken string, opts ...APIClientOpti NewBackoff: func() backoff.BackOff { return backoff.NewExponentialBackOff( backoff.WithInitialInterval(100*time.Millisecond), - backoff.WithMaxInterval(1*time.Second), backoff.WithMaxElapsedTime(http2MaxRetryTime), ) }, @@ -120,6 +119,7 @@ func (rt *AuthRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) return rt.Base.RoundTrip(req) } +// RetryRoundTripper retries a single HTTP request on transient network errors using the backoff returned by NewBackoff. type RetryRoundTripper struct { Base http.RoundTripper // NewBackoff creates a new backoff policy for each request. diff --git a/internal/corrosion/subscribe.go b/internal/corrosion/subscribe.go index 7c76b9b9..2bcb648c 100644 --- a/internal/corrosion/subscribe.go +++ b/internal/corrosion/subscribe.go @@ -297,13 +297,13 @@ func (c *APIClient) resubscribeWithBackoffFn(id string) func(context.Context, ui // A gone subscription can never be resubscribed, so stop retrying immediately and let the caller // recover by creating a fresh subscription. if errors.Is(err, ErrSubscriptionNotFound) { - slog.Debug("Corrosion subscription no longer exists, giving up resubscribing.", + slog.Error("Corrosion subscription no longer exists, giving up resubscribing.", "id", id, "from_change", fromChange) return nil, backoff.Permanent(fmt.Errorf("resubscribe to %s: %w", id, err)) } // Don't log retries triggered by context cancellation, the backoff will stop immediately. if ctx.Err() == nil { - slog.Debug("Failed to resubscribe to Corrosion query. Retrying with backoff.", + slog.Error("Failed to resubscribe to Corrosion query. Retrying with backoff.", "id", id, "from_change", fromChange, "err", err) } } diff --git a/internal/machine/cluster.go b/internal/machine/cluster.go index 9ae742fe..20638414 100644 --- a/internal/machine/cluster.go +++ b/internal/machine/cluster.go @@ -606,7 +606,7 @@ func (cc *clusterController) backfillMachineState(ctx context.Context) error { // syncDockerContainers watches local Docker containers and syncs them to the cluster store. // TODO: move this to the Docker controller. func (cc *clusterController) syncDockerContainers(ctx context.Context) error { - // Retry to watch and sync containers until the context is done. + // Supervise the watch-and-sync pipeline until the context is done. boff := backoff.WithContext(backoff.NewExponentialBackOff( backoff.WithInitialInterval(100*time.Millisecond), backoff.WithMaxInterval(5*time.Second), @@ -614,7 +614,7 @@ func (cc *clusterController) syncDockerContainers(ctx context.Context) error { ), ctx) watchAndSync := func() error { if wErr := cc.dockerCtrl.WatchAndSyncContainers(ctx); wErr != nil { - slog.Debug("Failed to watch and sync containers to cluster store, retrying.", "err", wErr) + slog.Error("Failed to watch and sync containers to cluster store, retrying.", "err", wErr) return wErr } return nil