Improve deployment stability by adding health check endpoints []

dev_tools/pylibs/edgacknowledg/client/__init__.py

@@ -400,8 +400,12 @@ async def _send_reply_best_effort(channel: Channel,
 
 
 async def _reply_to_failed_request(channel: Channel,
-                                   handler: ReviewClientHandler) -> None:
-  '''Answer a request whose handler raised, and log why.
+                                   handler: ReviewClientHandler,
+                                   error: BaseException) -> None:
+  '''Tell the reviewer whose request failed with ``error`` (the handler
+  records them on it, in ``reviewer_id``) that it did, through the server,
+  and log why. With no reviewer to tell, it is only logged: the server
+  could forward the reply to no one.
 
   The section keeps serving afterwards: one bad request (which a reviewer
   can trigger) shouldn't take the whole section down.
@@ -410,7 +414,12 @@ async def _reply_to_failed_request(channel: Channel,
     f"Request to section {handler.subscription_type}/{handler.slug} failed:"
   )
   traceback.print_exc()
-  await _send_reply_best_effort(channel, message(RequestTags.INVALID_REQUEST))
+  reviewer_id = getattr(error, 'reviewer_id', None)
+  if reviewer_id is not None:
+    await _send_reply_best_effort(channel, message(
+      RequestTags.INVALID_REQUEST,
+      value = {'reviewer_id': reviewer_id}
+    ))
 
 
 async def _report_illegal_access(channel: Channel,
@@ -422,7 +431,7 @@ async def _report_illegal_access(channel: Channel,
   '''
   if error.reviewer_id is None:
     # Not traced to a reviewer; just a failed request.
-    await _reply_to_failed_request(channel, handler)
+    await _reply_to_failed_request(channel, handler, error)
   else:
     eprint(
       f"Reviewer {error.reviewer_id} of section "
@@ -476,16 +485,24 @@ def sections(self) -> List[ReviewClientHandler]:
     return self._review.sections()
 
   async def add_section(self, handler: ReviewClientHandler) -> None:
-    '''Register handler and subscribe it on the batch socket.'''
+    '''Register handler and subscribe it on the batch socket; while not
+    connected, it is subscribed once the session (re)connects, with every
+    other registered section.
+    '''
     # Serialized with reconnect restores so a section added mid-restore is
     # subscribed exactly once.
     async with self._client._subscriptions_lock:
       self._review.register_handler(handler)
-      try:
-        await self.subscribe_section(handler)
-      except Exception:
-        self._review.discard_handler(handler)
-        raise
+      # Not connected (sending would wait for it): the next session's
+      # restore subscribes it.
+      if self._client._session_ready.is_set():
+        try:
+          await self.subscribe_section(handler)
+        except NotConnectedError:
+          pass
+        except Exception:
+          self._review.discard_handler(handler)
+          raise
 
   async def remove_section(self, handler: ReviewClientHandler) -> None:
     '''Unsubscribe, then unregister.'''
@@ -513,8 +530,8 @@ async def _run_handler_loop(self,
           raise
         except IllegalFileAccessError as ex:
           await _report_illegal_access(channel, handler, ex)
-        except Exception:
-          await _reply_to_failed_request(channel, handler)
+        except Exception as ex:
+          await _reply_to_failed_request(channel, handler, ex)
     finally:
       self._client._inbounds.pop(channel.subscription_id, None)
       self._subscription_tasks.pop(channel.subscription_id, None)
@@ -681,8 +698,8 @@ async def _run_handler_loop(self,
         raise
       except IllegalFileAccessError as ex:
         await _report_illegal_access(channel, handler, ex)
-      except Exception:
-        await _reply_to_failed_request(channel, handler)
+      except Exception as ex:
+        await _reply_to_failed_request(channel, handler, ex)
 
   def _signal_section_ready(self, handler_key: int) -> None:
     ready = self._section_first_ready.get(handler_key)

dev_tools/pylibs/edgacknowledg/client/bench/__init__.py

@@ -33,6 +33,7 @@
   ConfinedPathComponent,
   IllegalFileAccessError,
   SandboxContext,
+  reviewer_of,
   trunc_display_content,
 )
 import edgutil
@@ -280,7 +281,9 @@ async def run_bench_source_handler(channel: Channel,
   request_tag, request_value, _section = decode_response(await channel.recv())
 
   if request_value is None or 'reviewer_id' not in request_value:
-    await channel.send(message(RequestTags.INVALID_REQUEST))
+    # Not a reviewer's request (none names one) but the server's, e.g. its
+    # error for a reply it refused: answering would only bounce back and
+    # forth.
     return
 
   reviewer_id = request_value['reviewer_id']
@@ -340,8 +343,8 @@ async def annotate_job() -> str:
         value = { 'reviewer_id': reviewer_id }
       ))
   except Exception as ex:
-    if isinstance(ex, IllegalFileAccessError):
-      ex.reviewer_id = reviewer_id
+    # Whom the failure reply (or the illegal-access report) is for.
+    setattr(ex, 'reviewer_id', reviewer_id)
     if sys.version_info >= (3, 11):
       ex.add_note(f"request triggered by reviewer: \"{reviewer_id}\"")
     raise ex
@@ -380,7 +383,10 @@ async def handle_bench_edit_request(session: BenchReviewSession,
         }
       )
 
-  return message(RequestTags.INVALID_REQUEST)
+  return message(
+    RequestTags.INVALID_REQUEST,
+    value = {'reviewer_id': request_value['reviewer_id']}
+  )
 
 
 class BenchSourceHandler:
@@ -441,6 +447,8 @@ def slug(self) -> str:
 
   async def handle_request(self, channel: Channel) -> None:
     request = await channel.recv()
+    if reviewer_of(request) is None:
+      return
 
     async def edit_job() -> None:
       reply = await handle_bench_edit_request(self._session, request)

dev_tools/pylibs/edgacknowledg/client/shared.py

@@ -40,6 +40,7 @@
   ProtocolMessage,
   RequestTags,
   RichName,
+  decode_response,
   message,
 )
 
@@ -176,6 +177,18 @@ def trunc_display_content(content: str) -> Dict[str, Union[str, bool]]:
   return { 'content': content }
 
 
+def reviewer_of(request: Any) -> Optional[str]:
+  '''The reviewer who made ``request``, as a handler receives it (the
+  server names them in every reviewer's request); ``None`` if it isn't a
+  reviewer's but the server's, e.g. its error for a reply it refused,
+  which a handler doesn't answer (an answer would only bounce back and
+  forth).
+  '''
+  _tag, value, _section = decode_response(request)
+  reviewer_id = value.get('reviewer_id') if isinstance(value, dict) else None
+  return None if reviewer_id is None else str(reviewer_id)
+
+
 
 class IllegalFileAccessError(Exception):
   '''Raised when a request asks for a file outside the allowed tree.

dev_tools/pylibs/edgacknowledg/client/test/__init__.py

@@ -40,6 +40,7 @@
   ConfinedPath,
   IllegalFileAccessError,
   SandboxContext,
+  reviewer_of,
   trunc_display_content,
 )
 from edgutil import eprint
@@ -206,6 +207,9 @@ def test_id_to_path(test_id: str) -> ConfinedPath:
 
     return test_home.joinpath('tests', *test_id_parts)
 
+  # Only a reviewer's request is answered: one naming none is the
+  # server's, e.g. its error for a reply it refused, and answering it would
+  # only bounce back and forth.
   if request_value is not None and 'reviewer_id' in request_value:
     reviewer_id = request_value['reviewer_id']
 
@@ -568,13 +572,11 @@ async def form_curr_diff_text(config_name: str,
           }
         ))
     except Exception as ex:
-      if isinstance(ex, IllegalFileAccessError):
-        ex.reviewer_id = reviewer_id
+      # Whom the failure reply (or the illegal-access report) is for.
+      setattr(ex, 'reviewer_id', reviewer_id)
       if sys.version_info >= (3, 11):
         ex.add_note(f"request triggered by reviewer: \"{reviewer_id}\"")
       raise ex
-  else:
-    await channel.send(message(RequestTags.INVALID_REQUEST))
 
 
 class TestSourceHandler:
@@ -702,7 +704,10 @@ async def handle_test_edit_request(
         }
       )
 
-  return message(RequestTags.INVALID_REQUEST)
+  return message(
+    RequestTags.INVALID_REQUEST,
+    value = {'reviewer_id': request_value['reviewer_id']}
+  )
 
 
 class TestEditHandler:
@@ -729,6 +734,8 @@ def slug(self) -> str:
 
   async def handle_request(self, channel: Channel) -> None:
     request = await channel.recv()
+    if reviewer_of(request) is None:
+      return
 
     async def edit_job() -> None:
       reply = await handle_test_edit_request(request)

dev_tools/pylibs/edgacknowledg/client/websocket.py

@@ -26,7 +26,10 @@
 import websockets
 
 try:
-  from websockets.asyncio.client import ClientConnection, connect  # type: ignore
+  from websockets.asyncio.client import (  # type: ignore
+    ClientConnection,
+    connect,
+  )
   _LEGACY_WEBSOCKETS = False
 except Exception:
   from websockets.client import connect  # type: ignore

dev_tools/pylibs/edgacknowledg/forge/config.py

@@ -87,8 +87,11 @@ class ForgeServiceConfig:
   api_url: Optional[str]
   # Seconds between polls; 0 polls once at startup only.
   poll_interval: float
-  # 'host:port' to serve webhooks on; None runs no web server at all.
-  webhook_bind: Optional[str]
+  # 'host:port' to serve the web app (serving) on: health checks, and webhooks
+  # if enabled; None serves none.
+  service_bind: Optional[str]
+  # Whether the web app receives webhooks.
+  webhooks: bool
   comments: bool
   # CI workflow (GitHub workflow file or id; GitLab: unused) whose runs are
   # tracked; None disables CI tracking.
@@ -182,6 +185,22 @@ def get_list(name: str) -> tuple[str, ...]:
     if max_concurrency < 1:
       raise ConfigError(f"{ENV_PREFIX}MAX_CONCURRENCY must be at least 1")
 
+    service_bind = get('SERVICE_BIND')
+    if service_bind is not None:
+      host, _, port = service_bind.rpartition(':')
+      if host == '' or not port.isdigit():
+        raise ConfigError(f"{ENV_PREFIX}SERVICE_BIND must be host:port")
+    webhooks = get_flag('WEBHOOKS')
+    if get('WEBHOOK_BIND') is not None:
+      raise ConfigError(
+        f"{ENV_PREFIX}WEBHOOK_BIND is gone: set {ENV_PREFIX}SERVICE_BIND "
+        f"(compose.yaml does) and {ENV_PREFIX}WEBHOOKS=1 instead"
+      )
+    if webhooks and service_bind is None:
+      raise ConfigError(
+        f"{ENV_PREFIX}WEBHOOKS is set, but {ENV_PREFIX}SERVICE_BIND is not"
+      )
+
     unsandboxed = get_flag('UNSANDBOXED')
     builder_service_url = service_url('BUILDER', environ = environ)
     publisher_service_url = service_url('PUBLISHER', environ = environ)
@@ -208,7 +227,8 @@ def get_list(name: str) -> tuple[str, ...]:
       base_branches = frozenset(get_list('BASE_BRANCHES') or ('main',)),
       api_url = get('API_URL'),
       poll_interval = get_number('POLL_INTERVAL', 300),
-      webhook_bind = get('WEBHOOK_BIND'),
+      service_bind = service_bind,
+      webhooks = webhooks,
       comments = get_flag('COMMENTS'),
       ci_workflow = get('CI_WORKFLOW'),
       settle_timeout = get_number('SETTLE_TIMEOUT', 2 * 60 * 60),

dev_tools/pylibs/edgacknowledg/forge/db/store.py

@@ -364,6 +364,16 @@ async def find_pull_request(self,
       row = result.scalar_one_or_none()
       return _pull_request_record(row) if row is not None else None
 
+  async def find_pull_request_by_review(self,
+                                        review_id: uuid.UUID
+                                        ) -> Optional[PullRequestRecord]:
+    async with self._session() as session:
+      result = await session.execute(
+        select(PullRequest).where(PullRequest.review_id == review_id)
+      )
+      row = result.scalar_one_or_none()
+      return _pull_request_record(row) if row is not None else None
+
   async def open_pull_requests(self,
                                forge: str,
                                repository: Optional[str] = None
@@ -451,36 +461,66 @@ async def mark_processed(self,
       row.next_attempt_at = None
       row.last_error = None
 
-  async def mark_failed(self,
-                        pull_request_id: int,
-                        generation: int,
-                        error: str,
-                        *,
-                        now: datetime,
-                        max_attempts: int,
-                        base_delay: float,
-                        max_delay: float) -> bool:
-    '''Record a processing failure and schedule a retry with exponential
-    backoff. After ``max_attempts`` the service gives up until something
-    about the pull request changes. Returns whether it gave up.
+  async def begin_attempt(self,
+                          pull_request_id: int,
+                          generation: int,
+                          *,
+                          now: datetime,
+                          max_attempts: int,
+                          base_delay: float,
+                          max_delay: float) -> bool:
+    '''Count an attempt at processing ``generation``, before it starts,
+    as a failure: so one that never finishes (the service killed part-way,
+    say by a pull request that exhausts its memory) counts too. It
+    schedules the retry, with exponential backoff, and records why, until
+    ``mark_processed`` or ``mark_failed`` says how it went (or
+    ``refund_attempt`` takes it back). After ``max_attempts`` the service
+    gives up until something about the pull request changes. Returns
+    whether it gave up (then there is no attempt).
     '''
     async with self._session() as session, session.begin():
       row = await session.get(PullRequest, pull_request_id)
       if row is None:
         return True
-      row.last_error = error
       if row.generation != generation:
-        # Changed while processing; retry right away with the new state.
+        # Changed since; the attempt is at the new state.
         return False
-      row.error_count += 1
       if row.error_count >= max_attempts:
         row.processed_generation = generation
         row.next_attempt_at = None
         return True
+      row.error_count += 1
+      row.last_error = 'processing never finished (the service stopped)'
       delay = min(max_delay, base_delay * 2 ** (row.error_count - 1))
       row.next_attempt_at = now + timedelta(seconds = delay)
       return False
 
+  async def mark_failed(self,
+                        pull_request_id: int,
+                        generation: int,
+                        error: str) -> None:
+    '''Record why the attempt at ``generation`` (counted already, by
+    ``begin_attempt``) failed.
+    '''
+    async with self._session() as session, session.begin():
+      row = await session.get(PullRequest, pull_request_id)
+      # Changed while processing: its counts were reset, to retry at once.
+      if row is not None and row.generation == generation:
+        row.last_error = error
+
+  async def refund_attempt(self, record: PullRequestRecord) -> None:
+    '''Take back the attempt ``begin_attempt`` counted at ``record``'s
+    generation, which didn't fail through the pull request's fault (the
+    publisher couldn't be reached, or the service is stopping): as it was
+    in ``record``, read before.
+    '''
+    async with self._session() as session, session.begin():
+      row = await session.get(PullRequest, record.id)
+      if row is not None and row.generation == record.generation:
+        row.error_count = record.error_count
+        row.next_attempt_at = record.next_attempt_at
+        row.last_error = record.last_error
+
   async def sections(self, pull_request_id: int) -> list[SectionRecord]:
     async with self._session() as session:
       result = await session.execute(