publish.py: retry a release call that ends without a result; print the release's step and code

fleet/coord/publish.py

@@ -8,16 +8,29 @@
 import time
 from pathlib import Path
 
+TOKEN = re.compile(r"[A-Za-z0-9.:_-]{1,64}")
+
+
+class CommandFailed(RuntimeError):
+    def __init__(self, name, returncode):
+        super().__init__("command failed: " + name)
+        self.returncode = returncode
+
 
 def command(args, *, body=None, env=None, allow_missing=False):
     result = subprocess.run(args, input=body, capture_output=True, text=True, env=env)
     if result.returncode:
         if allow_missing and "ImageNotFoundException" in result.stderr:
             return None
-        raise RuntimeError("command failed: " + args[0])
+        raise CommandFailed(args[0], result.returncode)
     return result.stdout
 
 
+def attribution(result):
+    """The release function's failed step and AWS error code: the only fields of its result repeated here."""
+    return {k: result[k] for k in ("step", "code") if isinstance(result.get(k), str) and TOKEN.fullmatch(result[k])}
+
+
 def main():
     commit, repo = os.environ["GITHUB_SHA"], os.environ["COORDINATOR_REPO"]
     if not re.fullmatch(r"[0-9a-f]{40}", commit):
@@ -49,9 +62,20 @@ def main():
     with tempfile.TemporaryDirectory(prefix="coordinator-release-") as tmp:
         output = Path(tmp) / "result.json"  # Only status, task ARN and image digest; no credentials.
         for attempt in range(11):
-            meta = json.loads(command(["aws", "lambda", "invoke", "--region", "us-west-1",
-                "--function-name", os.environ["COORDINATOR_RELEASE"], "--cli-binary-format", "raw-in-base64-out",
-                "--payload", "file:///dev/stdin", "--cli-read-timeout", "910", str(output)], body=payload))
+            try:
+                meta = json.loads(command(["aws", "lambda", "invoke", "--region", "us-west-1",
+                    "--function-name", os.environ["COORDINATOR_RELEASE"], "--cli-binary-format", "raw-in-base64-out",
+                    "--payload", "file:///dev/stdin", "--cli-read-timeout", "910", str(output)], body=payload))
+            except CommandFailed as error:
+                # The call can end while the function keeps running (its connection was dropped, then the
+                # CLI's own retries were throttled by the function's single reserved slot). The release is
+                # serialized and idempotent: a later invocation reports busy, then the finished release.
+                if attempt == 10:
+                    raise
+                print(f"The release invocation ended without a result (aws exit {error.returncode}); "
+                      "retrying in 60 seconds", flush=True)
+                time.sleep(60)
+                continue
             if meta.get("FunctionError"):
                 raise RuntimeError("release function failed")
             result = json.loads(output.read_text())
@@ -61,6 +85,8 @@ def main():
                 time.sleep(60)
                 continue
             if status not in ("deployed", "unchanged") or result.get("digest") != images[0]["imageDigest"]:
+                if attribution(result):
+                    print(json.dumps({"release": attribution(result)}), flush=True)
                 raise RuntimeError("staging release did not complete")
             print(json.dumps({"status": status, "commit": commit, "digest": result["digest"]}))
             return

fleet/coord/test_release.py

@@ -1,7 +1,9 @@
 #!/usr/bin/env python3
 """Offline checks for publishing and the live IPv4/name-regression probe."""
 import importlib.util
+import contextlib
 import fnmatch
+import io
 import copy
 import json
 import os
@@ -275,6 +277,61 @@ def test_malformed_commit_or_repository_never_runs_a_command(self):
             command.assert_not_called()
 
 
+class ReleaseInvocationTests(unittest.TestCase):
+    """The Lambda invocation: what a failed release prints, and a call that ends without a result."""
+
+    DIGEST = "sha256:" + "c" * 64
+
+    def run_publish(self, results):
+        """Run main() with each `aws lambda invoke` taking the next entry: a result dict, or an exit status."""
+        results, sleeps = list(results), []
+        def command(args, body=None, env=None, allow_missing=False):
+            if args[1] == "ecr":
+                return json.dumps({"imageDetails": [{"imageDigest": self.DIGEST}]})
+            outcome = results.pop(0)
+            if isinstance(outcome, int):
+                raise publish.CommandFailed(args[0], outcome)
+            Path(args[-1]).write_text(json.dumps(outcome))
+            return json.dumps({"StatusCode": 200})
+        env = {"GITHUB_SHA": "a" * 40, "COORDINATOR_REPO": "111122223333.dkr.ecr.us-east-2.amazonaws.com/fcvm-fleet/coordinator",
+               "COORDINATOR_RELEASE": "fcvm-coordinator-release"}
+        out = io.StringIO()
+        with patch.dict(os.environ, env), patch.object(publish, "command", command), \
+                patch.object(publish.time, "sleep", sleeps.append), contextlib.redirect_stdout(out):
+            try:
+                publish.main()
+                error = None
+            except Exception as caught:
+                error = caught
+        self.assertEqual(results, [], "not every scripted invocation ran")
+        return out.getvalue(), error, sleeps
+
+    def test_a_failed_release_prints_its_step_and_code_and_nothing_else(self):
+        printed, error, _ = self.run_publish([{
+            "status": "failed", "errorType": "ClientError", "step": "unlock",
+            "code": "ConditionalCheckFailedException", "message": "secret detail", "taskDefinition": "secret"}])
+        self.assertIsInstance(error, RuntimeError)
+        self.assertEqual([json.loads(line) for line in printed.splitlines()],
+                         [{"release": {"step": "unlock", "code": "ConditionalCheckFailedException"}}])
+
+    def test_a_field_that_is_not_a_plain_token_is_not_printed(self):
+        printed, error, _ = self.run_publish([{"status": "failed", "step": "rollout", "code": "has spaces: secret"}])
+        self.assertIsInstance(error, RuntimeError)
+        self.assertEqual([json.loads(line) for line in printed.splitlines()], [{"release": {"step": "rollout"}}])
+
+    def test_an_invocation_that_ends_without_a_result_is_retried_until_the_release_reports(self):
+        printed, error, sleeps = self.run_publish(
+            [255, 254, {"status": "busy"}, {"status": "unchanged", "digest": self.DIGEST}])
+        self.assertIsNone(error)
+        self.assertEqual(sleeps, [60, 60, 60])
+        self.assertEqual(json.loads(printed.splitlines()[-1])["status"], "unchanged")
+
+    def test_invocations_that_never_return_a_result_still_fail(self):
+        printed, error, sleeps = self.run_publish([255] * 11)
+        self.assertIsInstance(error, publish.CommandFailed)
+        self.assertEqual(len(sleeps), 10)
+
+
 class WorkflowTests(unittest.TestCase):
     def test_changes_to_the_actual_cli_files_trigger_a_release(self):
         workflow = (HERE.parent.parent / ".github/workflows/fleet-coordinator-deploy.yml").read_text()