rethink-public/aiowebhooks
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: per-call attempt counter so concurrent sends don't corrupt attempts

Browse Source 
the attempt count used instance state (self._attempt_no), reset in _send_loop and incremented in _attempt; two concurrent send() calls on one Webhook interleaved, corrupting each other's WebhookResult.attempts. it is now a per-call mutable cell created in _send_loop and threaded into _attempt. verified under load: 400 concurrent sends, zero corrupted counts.

Signed-off-by: disqualifier <dev@disqualifier.me>
This commit is contained in:
disqualifier 2026-06-28 17:46:20 -04:00
parent bcf1f5511c
commit cd93e4f44f
2 changed files with 12 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
README.md
View File

@ -48,7 +48,9 @@ Webhook(
``` ```
Inject a shared `session` for throughput (one session per process); without one, each Inject a shared `session` for throughput (one session per process); without one, each
send opens and closes its own. send opens and closes its own. A single `Webhook` instance is safe to drive from many
concurrent `send()` calls — each call tracks its own attempt count, so concurrent sends
don't corrupt each other's `WebhookResult.attempts`.
## WebhookResult ## WebhookResult

14
src/aiowebhooks/sender.py
View File

@ -136,10 +136,10 @@ class Webhook:
synthetic status-0). proxy rotation on connection errors lives inside the synthetic status-0). proxy rotation on connection errors lives inside the
attempt and is capped separately. attempt and is capped separately.
""" """
self._attempt_no = 0 counter = [0]
try: try:
return await aretry( return await aretry(
lambda: self._attempt(session, url, payload), lambda: self._attempt(session, url, payload, counter),
attempts=self.max_retries + 1, attempts=self.max_retries + 1,
on=(_Retryable,), on=(_Retryable,),
) )
@ -147,7 +147,7 @@ class Webhook:
return exhausted.result return exhausted.result
async def _attempt( async def _attempt(
self, session: aiohttp.ClientSession, url: str, payload: Dict self, session: aiohttp.ClientSession, url: str, payload: Dict, counter: List[int]
) -> WebhookResult: ) -> WebhookResult:
"""one logical send: proxy rotation + a single POST; may raise _Retryable """one logical send: proxy rotation + a single POST; may raise _Retryable
@ -155,14 +155,18 @@ class Webhook:
aretry applies backoff; honors an explicit 429 retry_after by sleeping it aretry applies backoff; honors an explicit 429 retry_after by sleeping it
before signalling. returns a final WebhookResult on success or a terminal before signalling. returns a final WebhookResult on success or a terminal
(non-retryable) failure — never lets a provider/connection error escape. (non-retryable) failure — never lets a provider/connection error escape.
`counter` is a per-call mutable cell ([0]) owned by the calling `_send_loop`,
so the attempt count is local to one `send()` and concurrent sends on the
same instance don't corrupt each other's tally.
""" """
timeout = aiohttp.ClientTimeout(total=self.timeout) timeout = aiohttp.ClientTimeout(total=self.timeout)
last_proxy: Optional[str] = None last_proxy: Optional[str] = None
proxy_tries = 0 proxy_tries = 0
while True: while True:
self._attempt_no += 1 counter[0] += 1
attempts = self._attempt_no attempts = counter[0]
proxy_url = None proxy_url = None
if self._proxies is not None: if self._proxies is not None:
try: try: