estuary-cdk: add basic retry logic for config encryption requests

When rotating OAuth2 tokens, a task can end up with a broken config if
the token exchange succeeds, invalidates all previous tokens, then the
connector fails to emit a `configUpdate` event. To guard against this
failure mode a little, very basic retry logic was added for when the
config encryption service experiences transient failures.

I haven't observed any config encryption failures in the wild, but it's
feasible something could happen causing Google Cloud Run to temporarily
be down.

Author: Alex-Bair

estuary-cdk/estuary_cdk/capture/base_capture_connector.py

@@ -177,7 +177,30 @@ async def _encrypt_config(
             "schema": config.model_json_schema(),
         }
 
-        encrypted_config = await self.request(log, ENCRYPTION_URL, "POST", json=body, _with_token = False)
+        # Retry for up to an hour if the encryption service is unavailable. Tasks can end up with broken configs
+        # if the updated config isn't encrypted & emitted in a configUpdate event. Retrying transient errors
+        # helps avoid being left with broken configs due to temporary encryption service outages.
+        max_retries = 20
+        retry_count = 0
+        backoff_seconds = 300
+
+        while True:
+            try:
+                encrypted_config = await self.request(log, ENCRYPTION_URL, "POST", json=body, _with_token=False)
+                break
+            except HTTPError as err:
+                if err.code >= 500 and retry_count < max_retries:
+                    retry_count += 1
+
+                    log.warning(f"Request failed (attempt {retry_count} of {max_retries}). Retrying in {backoff_seconds} seconds.", {
+                        "error code": err.code,
+                    })
+                    await asyncio.sleep(backoff_seconds)
+                else:
+                    log.error("Failed to encrypt config.", {
+                        "error code": err.code,
+                    })
+                    raise
 
         return json.loads(encrypted_config.decode('utf-8'))