refactor(corrosion): reduce transport retry time for corrosion 10s->2s, move some logs to error level

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 bounds the transport retry of a single request. Kept short so it only absorbs transient
-	http2MaxRetryTime = 10 * time.Second
+	// blips. Retrying during a Corrosion outage is the job of the higher-level recovery loops.
-	// resubscribeMaxRetryTime is the maximum amount of time an API client will retry resubscribing to a query after
+	http2MaxRetryTime = 2 * time.Second
-	// an error occurs.
+	// 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.
+// Retries are split by failure mode: the transport briefly retries a single request on transient network errors, while
-// It automatically resubscribes to active subscriptions if an error occurs using an exponential backoff policy with a
+// subscriptions resubscribe from the last change ID when their stream breaks and own the wait while Corrosion is down.
-// maximum interval of 1 second and a maximum elapsed time of 60 seconds.
+//
 // 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.

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)
 				}
 			}

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