Skip to content

Commit 458b3ab

Browse files
gbeloteclaude
gbelote
and
claude
committed
Address review: callback is a persistence path; bind mark_persisted to the saved set; timeout everywhere
Rivet's review of #2 (three P2s): 1. on_token_refresh is a persistence path, so it now runs BEFORE publication inside the same pending transition as store.save, for both managers; a raising/blocked callback keeps the set pending and surfaces WefunderTokenPersistenceError. Callback barrier/failure/retry tests, sync and async. 2. timeout now reaches the token managers and the re-mint closures in both constructors, and exchange_code / async_exchange_code forward it. Recorder tests cover expired-token refresh, the initial mint, and code exchange, sync and async. 3. mark_persisted(tokens) takes the exact set that was saved and is a no-op (False) for a stale acknowledgment, so an older r2 save can never publish an unsaved r3 — deterministic public-API test included. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
1 parent 7005bd6 commit 458b3ab

4 files changed

Lines changed: 221 additions & 88 deletions

File tree

‎src/wefunder/client.py‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -201,6 +201,7 @@ def __init__(
201201
token_base_url=token_base,
202202
transport=transport,
203203
now=now,
204+
timeout=timeout,
204205
)
205206
self._tm = TokenManager(
206207
self._initial_tokens,
@@ -212,6 +213,7 @@ def __init__(
212213
transport=transport,
213214
now=now,
214215
token_base_url=token_base,
216+
timeout=timeout,
215217
)
216218
now_ms = (lambda: now() * 1000) if now else None
217219
self._http = httpx.Client(
@@ -529,6 +531,7 @@ def __init__(
529531
token_base_url=token_base,
530532
transport=transport,
531533
now=now,
534+
timeout=timeout,
532535
)
533536
self._tm = AsyncTokenManager(
534537
self._initial_tokens,
@@ -540,6 +543,7 @@ def __init__(
540543
transport=transport,
541544
now=now,
542545
token_base_url=token_base,
546+
timeout=timeout,
543547
)
544548
now_ms = (lambda: now() * 1000) if now else None
545549
self._http = httpx.AsyncClient(

‎src/wefunder/oauth.py‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -226,7 +226,7 @@ def exchange_code(
226226
clients omit ``client_secret``; confidential clients pass it."""
227227
base = resolve_token_base(token_base_url=token_base_url, oauth_base_url=oauth_base_url)
228228
return _post_token(
229-
base, _exchange_params(client_id, client_secret, code, redirect_uri, code_verifier), transport, now
229+
base, _exchange_params(client_id, client_secret, code, redirect_uri, code_verifier), transport, now, timeout
230230
)
231231

232232

@@ -279,7 +279,7 @@ async def async_exchange_code(
279279
"""Async :func:`exchange_code`."""
280280
base = resolve_token_base(token_base_url=token_base_url, oauth_base_url=oauth_base_url)
281281
return await _apost_token(
282-
base, _exchange_params(client_id, client_secret, code, redirect_uri, code_verifier), transport, now
282+
base, _exchange_params(client_id, client_secret, code, redirect_uri, code_verifier), transport, now, timeout
283283
)
284284

285285

‎src/wefunder/token_manager.py‎

Lines changed: 23 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -57,6 +57,10 @@ async def _maybe_await(value: Any) -> Any:
5757
return value
5858

5959

60+
def _same_token_set(a: TokenSet, b: TokenSet) -> bool:
61+
return a.access_token == b.access_token and a.refresh_token == b.refresh_token
62+
63+
6064
class _TokenManagerBase:
6165
def __init__(
6266
self,
@@ -144,15 +148,16 @@ def get_access_token(self) -> str:
144148
self.refresh(stale_token=self._tokens.access_token)
145149
return self._tokens.access_token
146150

147-
def mark_persisted(self) -> TokenSet:
148-
"""Tell the manager you persisted ``pending_tokens`` yourself; publishes it."""
151+
def mark_persisted(self, tokens: TokenSet) -> bool:
152+
"""Tell the manager you persisted ``tokens`` (the set from a
153+
:class:`WefunderTokenPersistenceError`) yourself. Publishes it only if it is still the
154+
pending set; a stale acknowledgment (the manager has since rotated again) is a no-op and
155+
returns ``False``, so an older save can never publish a newer, unsaved set."""
149156
with self._lock:
150-
if self._pending is None:
151-
return self._tokens
157+
if self._pending is None or not _same_token_set(self._pending, tokens):
158+
return False
152159
self._tokens, self._pending = self._pending, None
153-
if self._on_token_refresh is not None:
154-
self._on_token_refresh(self._tokens)
155-
return self._tokens
160+
return True
156161

157162
def _publish_pending(self) -> TokenSet:
158163
# Persist BEFORE publishing (caller holds the lock): no thread may use the rotated token
@@ -164,12 +169,14 @@ def _publish_pending(self) -> TokenSet:
164169
try:
165170
if self._store is not None:
166171
self._store.save(tokens)
172+
# on_token_refresh is a persistence path too, so it runs BEFORE publication and a
173+
# failure keeps the set pending exactly like a store failure.
174+
if self._on_token_refresh is not None:
175+
self._on_token_refresh(tokens)
167176
except Exception as exc:
168177
raise WefunderTokenPersistenceError(tokens, exc) from exc
169178
self._pending = None
170179
self._tokens = tokens
171-
if self._on_token_refresh is not None:
172-
self._on_token_refresh(tokens)
173180
return tokens
174181

175182
def refresh(self, *, stale_token: str | None = None) -> TokenSet:
@@ -239,28 +246,26 @@ async def get_access_token(self) -> str:
239246
await self.refresh(stale_token=self._tokens.access_token)
240247
return self._tokens.access_token
241248

242-
async def mark_persisted(self) -> TokenSet:
243-
"""Tell the manager you persisted ``pending_tokens`` yourself; publishes it."""
249+
async def mark_persisted(self, tokens: TokenSet) -> bool:
250+
"""Async :meth:`TokenManager.mark_persisted`."""
244251
async with self._get_lock():
245-
if self._pending is None:
246-
return self._tokens
252+
if self._pending is None or not _same_token_set(self._pending, tokens):
253+
return False
247254
self._tokens, self._pending = self._pending, None
248-
if self._on_token_refresh is not None:
249-
await _maybe_await(self._on_token_refresh(self._tokens))
250-
return self._tokens
255+
return True
251256

252257
async def _publish_pending(self) -> TokenSet:
253258
tokens = self._pending
254259
assert tokens is not None
255260
try:
256261
if self._store is not None:
257262
await _maybe_await(self._store.save(tokens))
263+
if self._on_token_refresh is not None:
264+
await _maybe_await(self._on_token_refresh(tokens))
258265
except Exception as exc:
259266
raise WefunderTokenPersistenceError(tokens, exc) from exc
260267
self._pending = None
261268
self._tokens = tokens
262-
if self._on_token_refresh is not None:
263-
await _maybe_await(self._on_token_refresh(tokens))
264269
return tokens
265270

266271
async def refresh(self, *, stale_token: str | None = None) -> TokenSet:

0 commit comments

Comments
 (0)