Skip to content

Commit d45dc04

Browse files
Improve deployment stability by adding health check endpoints []
This allows tests to know when services are ready/pauses "docker compose up"'s return until services are actually listening. This also refactors the deployment to use Quart for the servers (consistent with the AcknowlEDG API server) so that developers do not need to maintain knowledge of multiple WebSocket server patterns. Additionally, management of process pool jobs has been improved so that a task taking longer than they are allowed, responding to a review no longer being monitored, or similar, are properly cleaned up without negatively impacting other tasks. In a similar "reliability" vein, caps have been added to ensure input data stays within reasonable bounds to preserve service health. Finally, a number of cleanup passes have been taken to improve code quality.
1 parent d5b7c25 commit d45dc04

37 files changed

Lines changed: 2420 additions & 1091 deletions

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

Lines changed: 31 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -400,8 +400,12 @@ async def _send_reply_best_effort(channel: Channel,
400400

401401

402402
async def _reply_to_failed_request(channel: Channel,
403-
handler: ReviewClientHandler) -> None:
404-
'''Answer a request whose handler raised, and log why.
403+
handler: ReviewClientHandler,
404+
error: BaseException) -> None:
405+
'''Tell the reviewer whose request failed with ``error`` (the handler
406+
records them on it, in ``reviewer_id``) that it did, through the server,
407+
and log why. With no reviewer to tell, it is only logged: the server
408+
could forward the reply to no one.
405409
406410
The section keeps serving afterwards: one bad request (which a reviewer
407411
can trigger) shouldn't take the whole section down.
@@ -410,7 +414,12 @@ async def _reply_to_failed_request(channel: Channel,
410414
f"Request to section {handler.subscription_type}/{handler.slug} failed:"
411415
)
412416
traceback.print_exc()
413-
await _send_reply_best_effort(channel, message(RequestTags.INVALID_REQUEST))
417+
reviewer_id = getattr(error, 'reviewer_id', None)
418+
if reviewer_id is not None:
419+
await _send_reply_best_effort(channel, message(
420+
RequestTags.INVALID_REQUEST,
421+
value = {'reviewer_id': reviewer_id}
422+
))
414423

415424

416425
async def _report_illegal_access(channel: Channel,
@@ -422,7 +431,7 @@ async def _report_illegal_access(channel: Channel,
422431
'''
423432
if error.reviewer_id is None:
424433
# Not traced to a reviewer; just a failed request.
425-
await _reply_to_failed_request(channel, handler)
434+
await _reply_to_failed_request(channel, handler, error)
426435
else:
427436
eprint(
428437
f"Reviewer {error.reviewer_id} of section "
@@ -476,16 +485,24 @@ def sections(self) -> List[ReviewClientHandler]:
476485
return self._review.sections()
477486

478487
async def add_section(self, handler: ReviewClientHandler) -> None:
479-
'''Register handler and subscribe it on the batch socket.'''
488+
'''Register handler and subscribe it on the batch socket; while not
489+
connected, it is subscribed once the session (re)connects, with every
490+
other registered section.
491+
'''
480492
# Serialized with reconnect restores so a section added mid-restore is
481493
# subscribed exactly once.
482494
async with self._client._subscriptions_lock:
483495
self._review.register_handler(handler)
484-
try:
485-
await self.subscribe_section(handler)
486-
except Exception:
487-
self._review.discard_handler(handler)
488-
raise
496+
# Not connected (sending would wait for it): the next session's
497+
# restore subscribes it.
498+
if self._client._session_ready.is_set():
499+
try:
500+
await self.subscribe_section(handler)
501+
except NotConnectedError:
502+
pass
503+
except Exception:
504+
self._review.discard_handler(handler)
505+
raise
489506

490507
async def remove_section(self, handler: ReviewClientHandler) -> None:
491508
'''Unsubscribe, then unregister.'''
@@ -513,8 +530,8 @@ async def _run_handler_loop(self,
513530
raise
514531
except IllegalFileAccessError as ex:
515532
await _report_illegal_access(channel, handler, ex)
516-
except Exception:
517-
await _reply_to_failed_request(channel, handler)
533+
except Exception as ex:
534+
await _reply_to_failed_request(channel, handler, ex)
518535
finally:
519536
self._client._inbounds.pop(channel.subscription_id, None)
520537
self._subscription_tasks.pop(channel.subscription_id, None)
@@ -681,8 +698,8 @@ async def _run_handler_loop(self,
681698
raise
682699
except IllegalFileAccessError as ex:
683700
await _report_illegal_access(channel, handler, ex)
684-
except Exception:
685-
await _reply_to_failed_request(channel, handler)
701+
except Exception as ex:
702+
await _reply_to_failed_request(channel, handler, ex)
686703

687704
def _signal_section_ready(self, handler_key: int) -> None:
688705
ready = self._section_first_ready.get(handler_key)

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

Lines changed: 12 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@
3333
ConfinedPathComponent,
3434
IllegalFileAccessError,
3535
SandboxContext,
36+
reviewer_of,
3637
trunc_display_content,
3738
)
3839
import edgutil
@@ -280,7 +281,9 @@ async def run_bench_source_handler(channel: Channel,
280281
request_tag, request_value, _section = decode_response(await channel.recv())
281282

282283
if request_value is None or 'reviewer_id' not in request_value:
283-
await channel.send(message(RequestTags.INVALID_REQUEST))
284+
# Not a reviewer's request (none names one) but the server's, e.g. its
285+
# error for a reply it refused: answering would only bounce back and
286+
# forth.
284287
return
285288

286289
reviewer_id = request_value['reviewer_id']
@@ -340,8 +343,8 @@ async def annotate_job() -> str:
340343
value = { 'reviewer_id': reviewer_id }
341344
))
342345
except Exception as ex:
343-
if isinstance(ex, IllegalFileAccessError):
344-
ex.reviewer_id = reviewer_id
346+
# Whom the failure reply (or the illegal-access report) is for.
347+
setattr(ex, 'reviewer_id', reviewer_id)
345348
if sys.version_info >= (3, 11):
346349
ex.add_note(f"request triggered by reviewer: \"{reviewer_id}\"")
347350
raise ex
@@ -380,7 +383,10 @@ async def handle_bench_edit_request(session: BenchReviewSession,
380383
}
381384
)
382385

383-
return message(RequestTags.INVALID_REQUEST)
386+
return message(
387+
RequestTags.INVALID_REQUEST,
388+
value = {'reviewer_id': request_value['reviewer_id']}
389+
)
384390

385391

386392
class BenchSourceHandler:
@@ -441,6 +447,8 @@ def slug(self) -> str:
441447

442448
async def handle_request(self, channel: Channel) -> None:
443449
request = await channel.recv()
450+
if reviewer_of(request) is None:
451+
return
444452

445453
async def edit_job() -> None:
446454
reply = await handle_bench_edit_request(self._session, request)

‎dev_tools/pylibs/edgacknowledg/client/shared.py‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,7 @@
4040
ProtocolMessage,
4141
RequestTags,
4242
RichName,
43+
decode_response,
4344
message,
4445
)
4546

@@ -176,6 +177,18 @@ def trunc_display_content(content: str) -> Dict[str, Union[str, bool]]:
176177
return { 'content': content }
177178

178179

180+
def reviewer_of(request: Any) -> Optional[str]:
181+
'''The reviewer who made ``request``, as a handler receives it (the
182+
server names them in every reviewer's request); ``None`` if it isn't a
183+
reviewer's but the server's, e.g. its error for a reply it refused,
184+
which a handler doesn't answer (an answer would only bounce back and
185+
forth).
186+
'''
187+
_tag, value, _section = decode_response(request)
188+
reviewer_id = value.get('reviewer_id') if isinstance(value, dict) else None
189+
return None if reviewer_id is None else str(reviewer_id)
190+
191+
179192

180193
class IllegalFileAccessError(Exception):
181194
'''Raised when a request asks for a file outside the allowed tree.

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

Lines changed: 12 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,7 @@
4040
ConfinedPath,
4141
IllegalFileAccessError,
4242
SandboxContext,
43+
reviewer_of,
4344
trunc_display_content,
4445
)
4546
from edgutil import eprint
@@ -206,6 +207,9 @@ def test_id_to_path(test_id: str) -> ConfinedPath:
206207

207208
return test_home.joinpath('tests', *test_id_parts)
208209

210+
# Only a reviewer's request is answered: one naming none is the
211+
# server's, e.g. its error for a reply it refused, and answering it would
212+
# only bounce back and forth.
209213
if request_value is not None and 'reviewer_id' in request_value:
210214
reviewer_id = request_value['reviewer_id']
211215

@@ -568,13 +572,11 @@ async def form_curr_diff_text(config_name: str,
568572
}
569573
))
570574
except Exception as ex:
571-
if isinstance(ex, IllegalFileAccessError):
572-
ex.reviewer_id = reviewer_id
575+
# Whom the failure reply (or the illegal-access report) is for.
576+
setattr(ex, 'reviewer_id', reviewer_id)
573577
if sys.version_info >= (3, 11):
574578
ex.add_note(f"request triggered by reviewer: \"{reviewer_id}\"")
575579
raise ex
576-
else:
577-
await channel.send(message(RequestTags.INVALID_REQUEST))
578580

579581

580582
class TestSourceHandler:
@@ -702,7 +704,10 @@ async def handle_test_edit_request(
702704
}
703705
)
704706

705-
return message(RequestTags.INVALID_REQUEST)
707+
return message(
708+
RequestTags.INVALID_REQUEST,
709+
value = {'reviewer_id': request_value['reviewer_id']}
710+
)
706711

707712

708713
class TestEditHandler:
@@ -729,6 +734,8 @@ def slug(self) -> str:
729734

730735
async def handle_request(self, channel: Channel) -> None:
731736
request = await channel.recv()
737+
if reviewer_of(request) is None:
738+
return
732739

733740
async def edit_job() -> None:
734741
reply = await handle_test_edit_request(request)

‎dev_tools/pylibs/edgacknowledg/client/websocket.py‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,10 @@
2626
import websockets
2727

2828
try:
29-
from websockets.asyncio.client import ClientConnection, connect # type: ignore
29+
from websockets.asyncio.client import ( # type: ignore
30+
ClientConnection,
31+
connect,
32+
)
3033
_LEGACY_WEBSOCKETS = False
3134
except Exception:
3235
from websockets.client import connect # type: ignore

‎dev_tools/pylibs/edgacknowledg/forge/config.py‎

Lines changed: 23 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -87,8 +87,11 @@ class ForgeServiceConfig:
8787
api_url: Optional[str]
8888
# Seconds between polls; 0 polls once at startup only.
8989
poll_interval: float
90-
# 'host:port' to serve webhooks on; None runs no web server at all.
91-
webhook_bind: Optional[str]
90+
# 'host:port' to serve the web app (serving) on: health checks, and webhooks
91+
# if enabled; None serves none.
92+
service_bind: Optional[str]
93+
# Whether the web app receives webhooks.
94+
webhooks: bool
9295
comments: bool
9396
# CI workflow (GitHub workflow file or id; GitLab: unused) whose runs are
9497
# tracked; None disables CI tracking.
@@ -182,6 +185,22 @@ def get_list(name: str) -> tuple[str, ...]:
182185
if max_concurrency < 1:
183186
raise ConfigError(f"{ENV_PREFIX}MAX_CONCURRENCY must be at least 1")
184187

188+
service_bind = get('SERVICE_BIND')
189+
if service_bind is not None:
190+
host, _, port = service_bind.rpartition(':')
191+
if host == '' or not port.isdigit():
192+
raise ConfigError(f"{ENV_PREFIX}SERVICE_BIND must be host:port")
193+
webhooks = get_flag('WEBHOOKS')
194+
if get('WEBHOOK_BIND') is not None:
195+
raise ConfigError(
196+
f"{ENV_PREFIX}WEBHOOK_BIND is gone: set {ENV_PREFIX}SERVICE_BIND "
197+
f"(compose.yaml does) and {ENV_PREFIX}WEBHOOKS=1 instead"
198+
)
199+
if webhooks and service_bind is None:
200+
raise ConfigError(
201+
f"{ENV_PREFIX}WEBHOOKS is set, but {ENV_PREFIX}SERVICE_BIND is not"
202+
)
203+
185204
unsandboxed = get_flag('UNSANDBOXED')
186205
builder_service_url = service_url('BUILDER', environ = environ)
187206
publisher_service_url = service_url('PUBLISHER', environ = environ)
@@ -208,7 +227,8 @@ def get_list(name: str) -> tuple[str, ...]:
208227
base_branches = frozenset(get_list('BASE_BRANCHES') or ('main',)),
209228
api_url = get('API_URL'),
210229
poll_interval = get_number('POLL_INTERVAL', 300),
211-
webhook_bind = get('WEBHOOK_BIND'),
230+
service_bind = service_bind,
231+
webhooks = webhooks,
212232
comments = get_flag('COMMENTS'),
213233
ci_workflow = get('CI_WORKFLOW'),
214234
settle_timeout = get_number('SETTLE_TIMEOUT', 2 * 60 * 60),

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

Lines changed: 55 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -364,6 +364,16 @@ async def find_pull_request(self,
364364
row = result.scalar_one_or_none()
365365
return _pull_request_record(row) if row is not None else None
366366

367+
async def find_pull_request_by_review(self,
368+
review_id: uuid.UUID
369+
) -> Optional[PullRequestRecord]:
370+
async with self._session() as session:
371+
result = await session.execute(
372+
select(PullRequest).where(PullRequest.review_id == review_id)
373+
)
374+
row = result.scalar_one_or_none()
375+
return _pull_request_record(row) if row is not None else None
376+
367377
async def open_pull_requests(self,
368378
forge: str,
369379
repository: Optional[str] = None
@@ -451,36 +461,66 @@ async def mark_processed(self,
451461
row.next_attempt_at = None
452462
row.last_error = None
453463

454-
async def mark_failed(self,
455-
pull_request_id: int,
456-
generation: int,
457-
error: str,
458-
*,
459-
now: datetime,
460-
max_attempts: int,
461-
base_delay: float,
462-
max_delay: float) -> bool:
463-
'''Record a processing failure and schedule a retry with exponential
464-
backoff. After ``max_attempts`` the service gives up until something
465-
about the pull request changes. Returns whether it gave up.
464+
async def begin_attempt(self,
465+
pull_request_id: int,
466+
generation: int,
467+
*,
468+
now: datetime,
469+
max_attempts: int,
470+
base_delay: float,
471+
max_delay: float) -> bool:
472+
'''Count an attempt at processing ``generation``, before it starts,
473+
as a failure: so one that never finishes (the service killed part-way,
474+
say by a pull request that exhausts its memory) counts too. It
475+
schedules the retry, with exponential backoff, and records why, until
476+
``mark_processed`` or ``mark_failed`` says how it went (or
477+
``refund_attempt`` takes it back). After ``max_attempts`` the service
478+
gives up until something about the pull request changes. Returns
479+
whether it gave up (then there is no attempt).
466480
'''
467481
async with self._session() as session, session.begin():
468482
row = await session.get(PullRequest, pull_request_id)
469483
if row is None:
470484
return True
471-
row.last_error = error
472485
if row.generation != generation:
473-
# Changed while processing; retry right away with the new state.
486+
# Changed since; the attempt is at the new state.
474487
return False
475-
row.error_count += 1
476488
if row.error_count >= max_attempts:
477489
row.processed_generation = generation
478490
row.next_attempt_at = None
479491
return True
492+
row.error_count += 1
493+
row.last_error = 'processing never finished (the service stopped)'
480494
delay = min(max_delay, base_delay * 2 ** (row.error_count - 1))
481495
row.next_attempt_at = now + timedelta(seconds = delay)
482496
return False
483497

498+
async def mark_failed(self,
499+
pull_request_id: int,
500+
generation: int,
501+
error: str) -> None:
502+
'''Record why the attempt at ``generation`` (counted already, by
503+
``begin_attempt``) failed.
504+
'''
505+
async with self._session() as session, session.begin():
506+
row = await session.get(PullRequest, pull_request_id)
507+
# Changed while processing: its counts were reset, to retry at once.
508+
if row is not None and row.generation == generation:
509+
row.last_error = error
510+
511+
async def refund_attempt(self, record: PullRequestRecord) -> None:
512+
'''Take back the attempt ``begin_attempt`` counted at ``record``'s
513+
generation, which didn't fail through the pull request's fault (the
514+
publisher couldn't be reached, or the service is stopping): as it was
515+
in ``record``, read before.
516+
'''
517+
async with self._session() as session, session.begin():
518+
row = await session.get(PullRequest, record.id)
519+
if row is not None and row.generation == record.generation:
520+
row.error_count = record.error_count
521+
row.next_attempt_at = record.next_attempt_at
522+
row.last_error = record.last_error
523+
484524
async def sections(self, pull_request_id: int) -> list[SectionRecord]:
485525
async with self._session() as session:
486526
result = await session.execute(

0 commit comments

Comments
 (0)