Improve deployment stability by adding health check endpoints []

dev_tools/bin/edgy

@@ -255,15 +255,36 @@ def load_edgy_config(args: argparse.Namespace, test_home: Path) -> EdgyConfig:
     )
 
     test_suites = config_file_dict.get('test_suites', {})
+    primary_test_suites = test_suites.get('primary', [])
+    secondary_test_suites = test_suites.get('secondary', [])
+    _validate_configured_suites(
+      config_file_path,
+      [*primary_test_suites, *secondary_test_suites]
+    )
     config_sets = config_file_dict.get('config_sets', {})
     return EdgyConfig(
       test_home,
       loaded_configs,
       config_sets,
-      test_suites.get('primary', []),
-      test_suites.get('secondary', [])
+      primary_test_suites,
+      secondary_test_suites
     )
 
+def _validate_configured_suites(config_file_path: Path,
+                                test_suite_names: List[Any]) -> None:
+  '''Exit unless every suite ``config_file_path`` lists has a valid name.'''
+  invalid_test_suite_encountered = False
+  for test_suite in test_suite_names:
+    if not (isinstance(test_suite, str) and
+            edgtest.is_valid_suite_name(test_suite)):
+      edgutil.eprint(f"Invalid test suite name {test_suite!r} in "
+                     f"{config_file_path} (expected 1-64 of A-Z, a-z, "
+                     "0-9, _ and -).")
+      invalid_test_suite_encountered = True
+
+  if invalid_test_suite_encountered:
+    sys.exit(1)
+
 def get_expectation_path(run_config: TestRunConfig,
                          test_suite_name: str) -> Path:
   return Path(run_config.expectations_dir, f"{test_suite_name}.log")
@@ -627,7 +648,7 @@ def _diagnose_bad_expectations_file(file_path: Path) -> None:
       line_ctr += 1
       try:
         parse_test_status(line_text)
-      except:
+      except Exception:
         if len(line_text) != 0:
           edgutil.eprint(
             f"{file_path_str}, line {line_ctr}: "
@@ -668,7 +689,7 @@ def _construct_replay_filters(
         )
         for key in comparison:
           desired_tests.add(parsed_test_status_key_to_name(key))
-    except:
+    except Exception:
       edgutil.eprint(
         f"Failed to compare the previous run of the {test_suite.name} "
         'suite to its expectations.'
@@ -691,7 +712,7 @@ def _construct_replay_filters(
           elif (args.replay == 'output-mismatch' and
                 parsed_test_status.is_output_mismatch()):
             desired_tests.add(parsed_test_status_key_to_name(key))
-    except:
+    except Exception:
       edgutil.eprint(
         f"Failed to parse test statuses in previous run of {test_suite.name}."
       )
@@ -1233,7 +1254,7 @@ class TestStatusChecker:
                 encountered_lines: List[str]) -> Optional['TestStatusChecker']:
     try:
       return TestStatusChecker(expected_lines, encountered_lines)
-    except:
+    except Exception:
       return None
 
   def _get_status_for_comparison(

dev_tools/bin/ifc-qt

@@ -28,7 +28,11 @@ import edgifc.decoration
 from enum import IntEnum
 from pathlib import Path
 
-from PySide6 import QtCore, QtWidgets, QtGui  # type: ignore[import-not-found,import]
+from PySide6 import (  # type: ignore[import-not-found,import]
+  QtCore,
+  QtWidgets,
+  QtGui
+)
 
 from edgifc import (
   IFCReader,
@@ -383,7 +387,7 @@ class PartitionNodeFormatter:
         values.append(template_string.format(value.description))
       else:
         values.append(format_fn(value))
-    except:
+    except Exception:
       traceback.print_exc()
       values.append(edgifc.decoration.UNKNOWN_VALUE)
 
@@ -428,7 +432,7 @@ class PartitionNodeFormatter:
           stream_reader,
           tacit_field.name()
         )
-      except:
+      except Exception:
         traceback.print_exc()
         field_value = 0
 

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,25 @@ 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 BaseException:
+          # Failed or cancelled: not added.
+          self._review.discard_handler(handler)
+          raise
 
   async def remove_section(self, handler: ReviewClientHandler) -> None:
     '''Unsubscribe, then unregister.'''
@@ -513,8 +531,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)
@@ -637,7 +655,6 @@ def __init__(self,
     self._client = client
     self._review = _ClientReview(review_id, name, timestamp)
     self._section_tasks: Dict[int, asyncio.Task] = {}
-    self._section_first_ready: Dict[int, asyncio.Event] = {}
 
   @property
   def review_id(self) -> uuid.UUID:
@@ -659,7 +676,8 @@ async def add_section(self, handler: ReviewClientHandler) -> None:
     self._review.register_handler(handler)
     try:
       await self.subscribe_section(handler)
-    except Exception:
+    except BaseException:
+      # Failed or cancelled: not added.
       self._review.discard_handler(handler)
       raise
 
@@ -681,13 +699,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)
-
-  def _signal_section_ready(self, handler_key: int) -> None:
-    ready = self._section_first_ready.get(handler_key)
-    if ready is not None and not ready.is_set():
-      ready.set()
+      except Exception as ex:
+        await _reply_to_failed_request(channel, handler, ex)
 
   async def _section_connection_loop(self,
                                      handler: ReviewClientHandler) -> None:
@@ -710,34 +723,33 @@ async def _section_connection_loop(self,
       open_timeout = self._client._open_timeout,
       status_listener = self._client._status_fan_in
     )
-    try:
-      async for websocket in resilient:
-        if (self._client._closed or
-            handler_key not in self._section_tasks):
-          break
+    async for websocket in resilient:
+      if (self._client._closed or
+          handler_key not in self._section_tasks):
+        break
 
-        channel = _DirectChannel(websocket)
-        try:
-          await announce_section(channel, handler, self._review)
-          self._signal_section_ready(handler_key)
-          await self._run_handler_loop(handler, channel)
-        except RoleOccupied:
-          self._section_tasks.pop(handler_key, None)
-          self._review.discard_handler(handler)
-          break
-        except IncompatibleProtocolVersion as ex:
-          self._client.abort(ex)
-          break
-        except ConnectionClosed:
-          continue
+      channel = _DirectChannel(websocket)
+      try:
+        await announce_section(channel, handler, self._review)
+        await self._run_handler_loop(handler, channel)
+      except RoleOccupied:
+        self._section_tasks.pop(handler_key, None)
+        self._review.discard_handler(handler)
+        break
+      except IncompatibleProtocolVersion as ex:
+        self._client.abort(ex)
+        break
+      except ConnectionClosed:
+        continue
 
-        if (self._client._closed or
-            handler_key not in self._section_tasks):
-          break
-    finally:
-      self._signal_section_ready(handler_key)
+      if (self._client._closed or
+          handler_key not in self._section_tasks):
+        break
 
   async def subscribe_section(self, handler: ReviewClientHandler) -> None:
+    '''Start keeping a socket for handler's section; as with a batch
+    session that isn't connected, it is served once that socket connects.
+    '''
     self._client._require_connected()
     handler_key = id(handler)
     if handler_key in self._section_tasks:
@@ -746,29 +758,15 @@ async def subscribe_section(self, handler: ReviewClientHandler) -> None:
         handler.subscription_type
       )
 
-    ready = asyncio.Event()
-    self._section_first_ready[handler_key] = ready
     loop = get_running_event_loop()
     # Register the key before starting so the loop treats this section as
     # wanted; unsubscribe pops the key to stop reconnecting.
     task = loop.create_task(self._section_connection_loop(handler))
     self._section_tasks[handler_key] = task
-    try:
-      await ready.wait()
-      if handler_key not in self._section_tasks:
-        raise NotConnectedError(
-          f"section failed to connect: "
-          f"{handler.subscription_type}/{handler.slug}"
-        )
-    finally:
-      self._section_first_ready.pop(handler_key, None)
 
   async def unsubscribe_section(self, handler: ReviewClientHandler) -> None:
     handler_key = id(handler)
     task = self._section_tasks.pop(handler_key, None)
-    ready = self._section_first_ready.pop(handler_key, None)
-    if ready is not None and not ready.is_set():
-      ready.set()
     if task is not None:
       task.cancel()
       try:
@@ -1019,18 +1017,21 @@ async def _recv_loop(self, websocket: ClientConnection) -> None:
             subscription_id = uuid.UUID(str(value['subscription_id']))
           except (KeyError, ValueError, TypeError):
             continue
-          # Bind inbound routing before completing the subscribe future so
-          # early batch.event frames are not dropped.
-          if subscription_id not in self._inbounds:
-            self._inbounds[subscription_id] = asyncio.Queue()
           key = (
             str(value.get('id')),
             str(value.get('type')),
             str(value.get('slug', '')),
           )
           future = self._pending_subscribe.pop(key, None)
-          if future is not None and not future.done():
-            future.set_result(value)
+          if future is None or future.done():
+            # Whoever subscribed gave up: claimed by no one.
+            await self._release(subscription_id)
+            continue
+          # Bind inbound routing before completing the subscribe future so
+          # early batch.event frames are not dropped.
+          if subscription_id not in self._inbounds:
+            self._inbounds[subscription_id] = asyncio.Queue()
+          future.set_result(value)
 
         elif tag == RequestTags.BATCH_UNSUBSCRIBE:
           if not isinstance(value, dict):
@@ -1079,9 +1080,18 @@ async def _subscribe(self, payload: Dict[str, Any]) -> uuid.UUID:
         SUBSCRIBE_TIMEOUT_S,
         f"subscribe {key[1]}/{key[2]}"
       )
-    except Exception:
-      self._pending_subscribe.pop(key, None)
-      if not future.done():
+    except BaseException:
+      # Failed or cancelled: no longer awaited.
+      if self._pending_subscribe.get(key) is future:
+        del self._pending_subscribe[key]
+      if (future.done() and not future.cancelled() and
+          future.exception() is None):
+        # Answered all the same: claimed by no one.  Shielded, so being
+        # cancelled again can't keep the release from being sent.
+        await asyncio.shield(self._release(
+          uuid.UUID(str(future.result()['subscription_id']))
+        ))
+      elif not future.done():
         future.cancel()
       raise
     subscription_id = uuid.UUID(str(echo['subscription_id']))
@@ -1101,12 +1111,31 @@ async def _unsubscribe(self, subscription_id: uuid.UUID) -> None:
         UNSUBSCRIBE_TIMEOUT_S,
         f"unsubscribe {subscription_id}"
       )
-    except Exception:
+    except BaseException:
+      # Failed or cancelled: no longer awaited.
       self._pending_unsubscribe.pop(subscription_id, None)
       if not future.done():
         future.cancel()
       raise
 
+  async def _release(self, subscription_id: uuid.UUID) -> None:
+    '''Have the server end a subscription nothing here claimed; its answer
+    isn't awaited (``_recv_loop`` drops it).
+
+    Sent only while connected: the server ends a subscription with its
+    connection, so there's nothing to wait for a reconnect to release.
+    '''
+    self._inbounds.pop(subscription_id, None)
+    connection = self._connection
+    if connection is not None:
+      try:
+        await connection.send_if_connected(encode_message(message(
+          RequestTags.BATCH_UNSUBSCRIBE,
+          value = { 'subscription_id': str(subscription_id) }
+        )))
+      except ConnectionClosed:
+        pass
+
   def _make_channel(self, subscription_id: uuid.UUID) -> _SubscriptionChannel:
     inbound = self._inbounds.setdefault(subscription_id, asyncio.Queue())
     return _SubscriptionChannel(subscription_id, inbound, self)