github: Look up whether a comment landed before resending a mutation

revup/github/endpoint.py

@@ -13,11 +13,17 @@
 # HTTP statuses worth retrying: gateway/timeout (5xx) and secondary-rate-limit (403).
 TRANSIENT_STATUSES = frozenset({500, 502, 503, 504})
 SECONDARY_LIMIT_STATUS = 403
+MAX_RETRIES = 3
 # Cap how long we'll auto-sleep waiting for a rate limit to reset. A longer wait
 # (primary budget exhausted) is surfaced to the user instead of hanging silently.
 MAX_BACKOFF_SECONDS = 60.0
 
 
+def is_transient_status(status: int) -> bool:
+    """Whether an http status means the request is worth sending again."""
+    return status in TRANSIENT_STATUSES or status == SECONDARY_LIMIT_STATUS
+
+
 def _backoff_delay(headers: Any, attempt: int, base_delay: float) -> float:
     """Seconds to wait before retrying, driven by GitHub's rate-limit headers.
 
@@ -138,7 +144,7 @@ async def graphql(
         self,
         query: str,
         *,
-        max_retries: int = 3,
+        max_retries: int = MAX_RETRIES,
         base_delay: float = 1.0,
         **kwargs: Any,
     ) -> GraphqlResponse:
@@ -161,8 +167,7 @@ async def graphql(
                     raise RevupForgeException(body["errors"])
                 return GraphqlResponse.parse(body)
 
-            retryable = status in TRANSIENT_STATUSES or status == SECONDARY_LIMIT_STATUS
-            if not retryable or attempt >= max_retries - 1:
+            if not is_transient_status(status) or attempt >= max_retries - 1:
                 raise RevupRequestException(status, body if body is not None else {})
 
             delay = min(_backoff_delay(headers, attempt, base_delay), MAX_BACKOFF_SECONDS)

revup/github/github.py

@@ -1,7 +1,7 @@
 import logging
 from typing import Any, Dict, List, Optional, Set, Tuple
 
-from revup.core_types import RevupForgeException
+from revup.core_types import RevupForgeException, RevupRequestException
 from revup.forge import (
     MAX_COMMENTS_TO_QUERY,
     Forge,
@@ -10,7 +10,7 @@
     PrInfo,
     PrUpdate,
 )
-from revup.github.endpoint import GitHubEndpoint
+from revup.github.endpoint import MAX_RETRIES, GitHubEndpoint, is_transient_status
 from revup.github.graphql import (
     ErrorClass,
     GraphqlError,
@@ -128,6 +128,9 @@
 # or a repeated timeout) before treating its errors as fatal.
 _MAX_STALLED_RETRIES = 2
 
+# How many of a pr's newest comments to read when checking whether an add landed.
+MAX_COMMENTS_TO_CHECK = 20
+
 
 def _merge_data(into: Dict[str, Any], src: Any) -> None:
     """Merge GraphQL `data` dict `src` into `into`, combining nested repository fields."""
@@ -193,6 +196,18 @@ def add_team_queries(self, teams: List[Tuple[str, str]]) -> None:
                 values=[org, slug],
             )
 
+    def add_comment_queries(self, subject_ids: List[str]) -> None:
+        for subject_id in subject_ids:
+            self.add(
+                prefix="coms",
+                scope="top",
+                field_template="{}: node(id: {}) {{... on PullRequest {{comments (last: "
+                + str(MAX_COMMENTS_TO_CHECK)
+                + ") {{nodes {{body}}}}}}}},",
+                var_types=["ID!"],
+                values=[subject_id],
+            )
+
     def parse_prs(self, result: Any, head_refs: List[str]) -> List[Optional[PrInfo]]:
         raw = self.extract(result, "pr")
         prs: List[Optional[PrInfo]] = []
@@ -327,6 +342,15 @@ def parse_labels(self, result: Any, labels: List[str]) -> Dict[str, str]:
                 logging.warning("Couldn't find an existing label named {}".format(label))
         return labels_to_ids
 
+    def parse_comment_bodies(
+        self, result: Any, subject_ids: List[str]
+    ) -> Dict[str, Optional[Set[str]]]:
+        """Bodies of each subject's newest comments, or None if github didn't return them."""
+        bodies: Dict[str, Optional[Set[str]]] = {}
+        for subject_id, node in zip(subject_ids, self.extract(result, "coms")):
+            bodies[subject_id] = {c["body"] for c in node["comments"]["nodes"]} if node else None
+        return bodies
+
     def parse_teams(
         self, result: Any, teams: List[Tuple[str, str]]
     ) -> Tuple[Dict[str, str], Dict[str, Optional[Set[str]]]]:
@@ -394,9 +418,48 @@ def _make_query_everything(
 
         return q
 
-    async def _run_once(self, q: GraphqlQuery) -> GraphqlResponse:
+    async def _run_once(self, q: GraphqlQuery, attempts: int = MAX_RETRIES) -> GraphqlResponse:
         query_str, variables = q.build()
-        return await self.endpoint.graphql(query_str, **variables)
+        if q.replay_safe:
+            return await self.endpoint.graphql(query_str, **variables)
+
+        try:
+            return await self.endpoint.graphql(query_str, max_retries=1, **variables)
+        except RevupRequestException as e:
+            if attempts <= 1 or not is_transient_status(e.status):
+                raise
+            logging.warning(
+                "GitHub returned {}, checking which comments it added before retrying".format(
+                    e.status
+                )
+            )
+
+        remaining = q.without(await self._comments_already_added(q))
+        if remaining.total_items() == 0:
+            # Every field landed before the failure.
+            return GraphqlResponse(data={})
+        return await self._run_once(remaining, attempts - 1)
+
+    async def _comments_already_added(self, q: GraphqlQuery) -> Set[str]:
+        """Aliases of the mutation's comment fields that must not be sent again.
+
+        A comment not positively shown as missing counts as added, since a duplicate is
+        worse than one the next upload adds.
+        """
+        # addComment is the only replay unsafe field, and takes a single input.
+        inputs = {alias: values[0] for alias, values in q.replay_unsafe_fields()}
+        subject_ids = list(dict.fromkeys(inp["subjectId"] for inp in inputs.values()))
+
+        check = GithubQuery(name="FindComments")
+        check.add_comment_queries(subject_ids)
+        bodies = check.parse_comment_bodies(await self._execute(check), subject_ids)
+
+        added = set()
+        for alias, inp in inputs.items():
+            on_pr = bodies[inp["subjectId"]]
+            if on_pr is None or inp["body"] in on_pr:
+                added.add(alias)
+        return added
 
     async def _execute(self, q: GraphqlQuery) -> Dict[str, Any]:
         """Run a query/mutation, salvaging partial results and re-transacting the rest.
@@ -673,7 +736,13 @@ def _build_update_mutation(self, prs: List[PrUpdate]) -> GraphqlQuery:
 
         q = GraphqlQuery(operation=GraphqlOperation.MUTATION)
 
-        def add_all(prefix: str, mutation: str, var_type: str, items: List[Any]) -> None:
+        def add_all(
+            prefix: str,
+            mutation: str,
+            var_type: str,
+            items: List[Any],
+            replay_safe: bool = True,
+        ) -> None:
             for inp in items:
                 q.add(
                     prefix=prefix,
@@ -686,9 +755,11 @@ def add_all(prefix: str, mutation: str, var_type: str, items: List[Any]) -> None
             }},""",
                     var_types=[var_type],
                     values=[inp],
+                    replay_safe=replay_safe,
                 )
 
-        add_all("com", "addComment", "AddCommentInput!", comments)
+        # A resent addComment adds another identical comment; the rest are idempotent.
+        add_all("com", "addComment", "AddCommentInput!", comments, replay_safe=False)
         add_all("pr", "updatePullRequest", "UpdatePullRequestInput!", inputs)
         add_all("rev", "requestReviews", "RequestReviewsInput!", reviewers)
         add_all("asn", "addAssigneesToAssignable", "AddAssigneesToAssignableInput!", assignees)

revup/github/graphql.py

@@ -105,6 +105,8 @@ class SingleQuery:
     values: List[Any]
     fragment: str
     index: int
+    # False if re-sending this field would duplicate an effect it already applied.
+    replay_safe: bool = True
 
     @property
     def alias(self) -> str:
@@ -161,6 +163,7 @@ def add(
         var_types: List[str],
         values: List[Any],
         fragment: str = "",
+        replay_safe: bool = True,
     ) -> None:
         """Add a field. Its alias index is fixed here so results stay addressable
         by prefix even after the query is split and its results merged."""
@@ -176,9 +179,19 @@ def add(
                 values=list(values),
                 fragment=fragment,
                 index=idx,
+                replay_safe=replay_safe,
             )
         )
 
+    @property
+    def replay_safe(self) -> bool:
+        """Whether the whole request can be re-sent when its response never arrived."""
+        return all(q.replay_safe for q in self.queries)
+
+    def replay_unsafe_fields(self) -> List[Tuple[str, List[Any]]]:
+        """The alias and variable values of every field that can't be re-sent."""
+        return [(q.alias, list(q.values)) for q in self.queries if not q.replay_safe]
+
     def extract(self, result: Any, prefix: str) -> List[Any]:
         """Return the result node for every field with the given prefix, in add order."""
         return [q.extract(result) for q in self.queries if q.prefix == prefix]
@@ -286,3 +299,7 @@ def subset(self, aliases: Set[str]) -> GraphqlQuery:
         collision. Used to re-transact only the fields that failed retryably.
         """
         return self._with_fields([q for q in self.queries if q.alias in aliases])
+
+    def without(self, aliases: Set[str]) -> GraphqlQuery:
+        """A query containing every field whose alias is not in `aliases`."""
+        return self._with_fields([q for q in self.queries if q.alias not in aliases])

tests/test_github.py

@@ -9,7 +9,7 @@
 
 from revup.core_types import RevupForgeException, RevupRequestException
 from revup.forge import ForgeRepoInfo, PrComment, PrInfo, PrUpdate
-from revup.github.endpoint import GitHubEndpoint, _backoff_delay
+from revup.github.endpoint import MAX_RETRIES, GitHubEndpoint, _backoff_delay
 from revup.github.github import _MAX_STALLED_RETRIES, Github, _merge_data
 from revup.github.graphql import GraphqlResponse
 
@@ -620,6 +620,87 @@ def test_ready_for_review_mutation(self):
         assert kwargs["from_d0"]["pullRequestId"] == "PR_1"
         assert "to_d0" not in kwargs
 
+    def test_transient_status_drops_a_comment_github_already_added(self):
+        """A comment already on the pr landed before the 502, so it isn't resent."""
+        ep = make_endpoint()
+        ep._post = commenting_post({"PR_1": ["hi"]})
+        gh = make_github(ep)
+
+        asyncio.run(gh.update_pull_requests([PrUpdate(id="PR_1", comments=[PrComment("hi")])]))
+
+        sent = ep._post.sent
+        assert sum("addComment" in q for q, _ in sent) == 1
+        # The rest of the mutation is idempotent, so it still gets resent.
+        assert sum("updatePullRequest" in q for q, _ in sent) == 2
+
+    def test_transient_status_resends_a_comment_github_never_added(self):
+        """A comment that isn't on the pr never landed, so the mutation is resent."""
+        ep = make_endpoint()
+        ep._post = commenting_post({"PR_1": []})
+        gh = make_github(ep)
+
+        asyncio.run(gh.update_pull_requests([PrUpdate(id="PR_1", comments=[PrComment("hi")])]))
+
+        resent = [kwargs for q, kwargs in ep._post.sent if "addComment" in q]
+        assert len(resent) == 2
+        assert resent[1]["com0"]["body"] == "hi"
+
+    def test_transient_status_resends_only_the_missing_comment(self):
+        """A lost comment is resent while one that landed in the same request is dropped."""
+        ep = make_endpoint()
+        ep._post = commenting_post({"PR_1": ["landed"]})
+        gh = make_github(ep)
+
+        asyncio.run(
+            gh.update_pull_requests(
+                [PrUpdate(id="PR_1", comments=[PrComment("landed"), PrComment("lost")])]
+            )
+        )
+
+        resent = [kwargs for q, kwargs in ep._post.sent if "addComment" in q][1]
+        assert resent["com1"]["body"] == "lost"
+        assert "com0" not in resent
+
+    def test_repeated_transient_status_stops_resending(self):
+        """A mutation that keeps failing gives up instead of resending forever."""
+        ep = make_endpoint()
+        ep._post = commenting_post({"PR_1": []}, fail_forever=True)
+        gh = make_github(ep)
+
+        with pytest.raises(RevupRequestException):
+            asyncio.run(gh.update_pull_requests([PrUpdate(id="PR_1", comments=[PrComment("hi")])]))
+
+        assert sum("addComment" in q for q, _ in ep._post.sent) == MAX_RETRIES
+
+    def test_transient_status_retries_mutation_without_added_comments(self):
+        """Every mutation field other than addComment is idempotent, so it still retries."""
+        ep = make_endpoint()
+        ep._post = scripted_post(
+            (502, {}, None),
+            (200, {}, {"data": {"pr_out0": {"clientMutationId": "revup"}}}),
+        )
+        gh = make_github(ep)
+
+        with patch("asyncio.sleep", new=AsyncMock()) as sleep:
+            asyncio.run(
+                gh.update_pull_requests(
+                    [PrUpdate(id="PR_1", title="t", comments=[PrComment("edit me", "C_1")])]
+                )
+            )
+        sleep.assert_awaited_once()
+
+    def test_transient_status_with_only_added_comments_sends_nothing_more(self):
+        """A mutation whose every comment already landed sends nothing more."""
+        ep = make_endpoint()
+        ep._post = commenting_post({"PR_1": ["hi"]})
+        gh = make_github(ep)
+
+        q = gh._build_update_mutation([PrUpdate(id="PR_1", comments=[PrComment("hi")])])
+        resp = asyncio.run(gh._run_once(q.subset({"com_out0"})))
+
+        assert resp.data == {}
+        assert sum("addComment" in query for query, _ in ep._post.sent) == 1
+
     def test_mutation_timeout_resubmits_only_unapplied(self):
         """On a whole-request timeout, fields that applied (non-null data) are kept and
         only the unapplied (null) fields are resubmitted — no duplicate side effects."""
@@ -941,6 +1022,48 @@ async def _post(query, kwargs):
     return _post
 
 
+def _alias_of(var_name):
+    """The field alias that goes with a rendered variable name."""
+    prefix = var_name.rstrip("0123456789")
+    return f"{prefix}_out{var_name[len(prefix) :]}"
+
+
+def commenting_post(comments_by_pr, fail_forever=False):
+    """An async _post replacement that 502s a mutation adding comments, answers the
+    resulting lookup from `comments_by_pr`, and succeeds otherwise.
+
+    Records every (query, variables) it was sent as `.sent`.
+    """
+    sent = []
+
+    async def _post(query, kwargs):
+        sent.append((query, kwargs))
+        if "node(" in query:
+            return (
+                200,
+                {},
+                {
+                    "data": {
+                        _alias_of(name): {
+                            "comments": {"nodes": [{"body": b} for b in comments_by_pr[pr_id]]}
+                        }
+                        for name, pr_id in kwargs.items()
+                    }
+                },
+            )
+        first_add = not any("addComment" in q for q, _ in sent[:-1])
+        if "addComment" in query and (fail_forever or first_add):
+            return 502, {}, None
+        return (
+            200,
+            {},
+            {"data": {_alias_of(name): {"clientMutationId": "revup"} for name in kwargs}},
+        )
+
+    _post.sent = sent
+    return _post
+
+
 class TestEndpointGraphqlPolicy:
     """endpoint.graphql retry/error policy, mocking the _post network seam."""