feather-reader 0.4.6

A minimalist, atproto-native RSS/Atom reader in Rust — your feed subscriptions live in your own PDS.
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
//! Holding a session valid: refresh-on-read, rotation, and what a failure means.
//!
//! Three things here are easy to get wrong and all three present as *random*
//! logouts rather than as a bug:
//!
//! * **Refresh tokens rotate and are single-use.** The new one must be stored
//!   atomically with the new access token, and a response that omits it means
//!   keep the old one — not store an empty string.
//! * **Concurrent refreshes must be serialized.** Two readers of the same
//!   session both presenting the same single-use token means the second gets
//!   `invalid_grant`, and most authorization servers revoke the first one's
//!   freshly issued tokens along with it. The scheduler plus one user request is
//!   already enough concurrency.
//! * **Only `invalid_grant` invalidates.** Deleting on a network blip or a 5xx
//!   logs people out for a server hiccup; treating a dead grant as transient
//!   retries forever and never prompts a re-login.

use anyhow::{bail, Result};
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use tokio::sync::{Mutex as AsyncMutex, OwnedMutexGuard};

use super::store::OAuthSession;
use super::token::TokenResponse;

/// Fold a refresh response into the stored session.
///
/// Pure, so the rotation rules are testable without a server — which matters,
/// because every one of them fails as a mysterious logout rather than as an
/// error anyone can trace.
pub fn apply_refresh(
    session: &OAuthSession,
    response: &TokenResponse,
    now: i64,
) -> Result<OAuthSession> {
    // A refresh must never be able to move a session to another account.
    if response.sub != session.sub {
        bail!(
            "refresh returned subject {:?}, expected {:?}; refusing to rebind the session",
            response.sub,
            session.sub
        );
    }

    Ok(OAuthSession {
        access_token: response.access_token.clone(),
        // Rotation is expected but not guaranteed. Storing an empty string when
        // the server omits it would destroy the session on the next refresh,
        // with nothing to indicate why.
        refresh_token: response
            .refresh_token
            .clone()
            .unwrap_or_else(|| session.refresh_token.clone()),
        token_type: response.token_type.clone(),
        // The GRANTED scope, which the server may have narrowed. Keeping the old
        // value would make later write failures unexplainable.
        granted_scope: response.granted_scope.clone(),
        expires_at: response.expires_in.map(|seconds| now + seconds),
        // Properties of the session, not of any one token response.
        sub: session.sub.clone(),
        issuer: session.issuer.clone(),
        aud: session.aud.clone(),
        dpop_key_jwk: session.dpop_key_jwk.clone(),
    })
}

/// Per-subject refresh locks.
///
/// Two concurrent refreshes of one session both present the same single-use
/// refresh token; the loser gets `invalid_grant`, and most authorization servers
/// revoke the winner's freshly issued tokens along with it. The symptom is
/// random, unreproducible logouts. The scheduler plus a single user request is
/// already enough concurrency to hit this.
///
/// **In-process only.** A second process — the invite bot, a rolling deploy —
/// is not covered, which is why the refresh path also re-reads the stored
/// session on `invalid_grant` before concluding the grant is dead.
#[derive(Default, Clone)]
pub struct RefreshLocks {
    locks: Arc<Mutex<HashMap<String, Arc<AsyncMutex<()>>>>>,
}

impl RefreshLocks {
    /// Acquire the lock for one subject. Unrelated subjects never contend.
    pub async fn lock(&self, sub: &str) -> OwnedMutexGuard<()> {
        let entry = {
            // Recover rather than panic: this gates every token refresh, so a
            // poisoned lock here would lock every user out until a restart.
            let mut locks = self.locks.lock().unwrap_or_else(|p| p.into_inner());
            Arc::clone(locks.entry(sub.to_string()).or_default())
        };
        entry.lock_owned().await
    }
}

/// Refuse a re-discovered issuer that is not the one the grant belongs to.
///
/// Discovery runs again from the network on both the callback and the refresh
/// path, and the token endpoint comes out of THAT document. Every discovery
/// check is internally consistent, so a hostile pair of documents satisfies all
/// of them; only this comparison notices that the pair describes a different
/// authorization server than the one that issued the grant.
///
/// A free function so it is reachable from a test. The refresh path's copy was
/// written inline, and deleting it — the single most serious defect found in
/// this branch, on the path that carries the REFRESH TOKEN — passed every test.
pub fn same_issuer(discovered: &str, expected: &str) -> Result<()> {
    if discovered != expected {
        anyhow::bail!(
            "the PDS now names a different authorization server ({discovered:?}) than this \
             grant was issued by ({expected:?}); refusing to send credentials to it"
        );
    }
    Ok(())
}

/// What a refresh needs beyond the session itself.
pub struct RefreshContext<'a> {
    pub token_endpoint: &'a str,
    pub client_id: &'a str,
    pub auth_method: super::client_auth::AuthMethod,
    /// The confidential client's signing key. Required for `private_key_jwt`,
    /// unused by the localhost dev client.
    pub client_key: Option<&'a super::keys::SigningKey>,
    /// From the same discovery as `token_endpoint`. Used only when a refresh
    /// finds its session was signed out while it ran: the tokens it just
    /// obtained belong to no one, and are revoked rather than left live.
    pub revocation_endpoint: Option<&'a str>,
}

/// Read a session, refreshing it first if it is close to expiry.
///
/// **Do not wrap this in a timeout.** Once the refresh request is in flight the
/// authorization server has consumed the single-use refresh token; abandoning
/// the future means the replacement is never stored and the session is dead.
/// The reference carries the same warning for the same reason.
pub async fn valid_session(
    pool: &sqlx::SqlitePool,
    codec: &super::crypto::Codec,
    http: &reqwest::Client,
    locks: &RefreshLocks,
    sub: &str,
    ctx: &RefreshContext<'_>,
    now: i64,
) -> Result<OAuthSession> {
    let session = super::store::get_session(pool, codec, sub)
        .await?
        .ok_or_else(|| anyhow::anyhow!("no session for {sub}"))?;
    if !super::token::is_stale(session.expires_at, now) {
        return Ok(session);
    }

    let _guard = locks.lock(sub).await;

    // Re-read AFTER acquiring the lock: whoever held it may have refreshed while
    // we waited, and presenting the token they just replaced would burn it.
    let (session, version) = super::store::get_session_versioned(pool, codec, sub)
        .await?
        .ok_or_else(|| anyhow::anyhow!("session for {sub} disappeared while waiting to refresh"))?;
    if !super::token::is_stale(session.expires_at, now) {
        return Ok(session);
    }

    // NOT timed here. `oauth_refresh` wraps this call AND the discovery that
    // precedes it, in `Repo::session` — a review found that timing only this
    // line missed every refresh that failed in discovery, which is where the two
    // likeliest failures live (an unreachable PDS, and the issuer-mismatch
    // check). See the span in `repo.rs`.
    refresh_locked(pool, codec, http, &session, &version, ctx, now).await
}

/// A refresh succeeded at the server, but the row it started from is no longer
/// on record. Two cases:
///
/// Either way the tokens this refresh just obtained are stored NOWHERE, and
/// are revoked (best-effort, bounded) rather than left live at the PDS:
///
/// * **Rewritten** by another writer: theirs is the session on record, so it
///   is returned and ours is not written over it. In practice that writer is
///   a RE-LOGIN — a new grant. It cannot be a concurrent refresh of the same
///   token: refresh tokens are single-use, and ours succeeded. So our tokens
///   belong to the old grant, which nothing references any more.
/// * **Gone**: it was signed out while we refreshed. We now hold a live token
///   set for a session that no longer exists, and the caller gets the same
///   "no session" error a signed-out user gets.
async fn lost_the_row(
    pool: &sqlx::SqlitePool,
    codec: &super::crypto::Codec,
    http: &reqwest::Client,
    ctx: &RefreshContext<'_>,
    obtained: &OAuthSession,
    now: i64,
) -> Result<OAuthSession> {
    let current = super::store::get_session(pool, codec, &obtained.sub).await?;
    let outcome = super::revoke::revoke_orphaned(
        pool,
        http,
        &super::revoke::RevokeContext {
            revocation_endpoint: ctx.revocation_endpoint,
            client_id: ctx.client_id,
            auth_method: ctx.auth_method,
            client_key: ctx.client_key,
            deadline: super::revoke::ORPHAN_REVOKE_DEADLINE,
        },
        obtained,
        now,
    )
    .await;
    if let super::revoke::Revocation::Failed(reason) = &outcome {
        tracing::warn!(
            sub = %obtained.sub,
            %reason,
            "a session changed or was signed out during its refresh: the refresh's \
             unstored new tokens could not be revoked"
        );
    }
    if let Some(current) = current {
        return Ok(current);
    }
    bail!(
        "no session for {}: it was signed out while being refreshed (the new tokens were \
         not stored{})",
        obtained.sub,
        if outcome == super::revoke::Revocation::Revoked {
            ", and were revoked"
        } else {
            ""
        }
    )
}

/// The refresh itself. Called with the subject's lock held; `version` is the
/// stored form of `session` as read under that lock.
async fn refresh_locked(
    pool: &sqlx::SqlitePool,
    codec: &super::crypto::Codec,
    http: &reqwest::Client,
    session: &OAuthSession,
    version: &super::store::SessionVersion,
    ctx: &RefreshContext<'_>,
    now: i64,
) -> Result<OAuthSession> {
    let key = super::keys::SigningKey::from_jwk_json(&session.dpop_key_jwk, "session-dpop")?;

    let assertion = match ctx.auth_method {
        super::client_auth::AuthMethod::PrivateKeyJwt => {
            let client_key = ctx
                .client_key
                .ok_or_else(|| anyhow::anyhow!("private_key_jwt refresh needs the client key"))?;
            Some(super::client_auth::client_assertion(
                client_key,
                ctx.client_id,
                &session.issuer,
                now,
            )?)
        }
        super::client_auth::AuthMethod::None => None,
    };

    let mut params = super::token::refresh_request_params(&session.refresh_token);
    params.extend(super::client_auth::credential_params(
        ctx.auth_method,
        ctx.client_id,
        assertion.as_deref(),
    )?);
    let borrowed: Vec<(&str, &str)> = params.iter().map(|(k, v)| (*k, v.as_str())).collect();

    let outcome = super::request::send_with_dpop(
        http,
        pool,
        &super::request::DpopRequest {
            endpoint: super::dpop::Endpoint::AuthorizationServer,
            url: ctx.token_endpoint,
            key: &key,
            access_token: None,
            body: super::request::DpopBody::Form(&borrowed),
            // A refresh is safe to repeat on a nonce challenge: unlike the code
            // exchange, a rejected attempt consumes nothing.
            retry: super::request::Retry::Allowed,
        },
    )
    .await?;

    if outcome.is_success() {
        let response = super::token::parse_token_response(&outcome.json()?)?;
        let updated = apply_refresh(session, &response, now)?;
        // **Conditional on the row we started from — never an upsert.** The
        // lock above is in-process; a sign-out (`/logout`, account deletion,
        // the operator's revoke-all in another process) does not take it. An
        // upsert landing after such a sign-out's delete RESURRECTED the
        // session with these fresh tokens, after it had been reported signed
        // out.
        if !super::store::update_session_if_unchanged(pool, codec, &updated, version).await? {
            return lost_the_row(pool, codec, http, ctx, &updated, now).await;
        }
        // Logged because a refresh is otherwise INVISIBLE: it rotates both
        // tokens and is the one path that can silently end a session, but it
        // happens inside an ordinary page load and the metrics record that call
        // no differently. Without this line, "everyone was logged out overnight"
        // has nothing to correlate against. No token material is logged — only
        // that it happened, and when the replacement expires.
        tracing::info!(
            sub = %updated.sub,
            expires_at = ?updated.expires_at,
            "refreshed the OAuth session"
        );
        return Ok(updated);
    }

    match super::token::classify_refresh_failure(outcome.status, &outcome.body) {
        super::token::RefreshFailure::Transient => {
            // Network, 5xx, anything unrecognised: leave the stored tokens
            // exactly as they are and fail THIS request. Deleting here would log
            // someone out for a server hiccup.
            bail!(
                "refresh for {} failed transiently (status {}); the session is left intact",
                session.sub,
                outcome.status
            )
        }
        super::token::RefreshFailure::SessionInvalid => {
            // Before concluding the grant is dead, check whether another PROCESS
            // refreshed it while we held only an in-process lock. If the stored
            // refresh token has changed, ours was simply stale and theirs is
            // live — use it rather than deleting a working session.
            if let Some(current) = super::store::get_session(pool, codec, &session.sub).await? {
                if current.refresh_token != session.refresh_token {
                    return Ok(current);
                }
            }
            super::store::delete_session(pool, &session.sub).await?;
            bail!(
                "refresh for {} was rejected as invalid_grant; the session has been \
                 removed and the user must log in again",
                session.sub
            )
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    const DID: &str = "did:plc:ewvi7nxzyoun6zhxrhs64oiz";
    const NOW: i64 = 1_700_000_000;

    fn session() -> OAuthSession {
        OAuthSession {
            sub: DID.into(),
            issuer: "https://auth.example.com".into(),
            aud: "https://pds.example.com".into(),
            dpop_key_jwk: r#"{"kty":"EC","d":"k"}"#.into(),
            access_token: "old-access".into(),
            refresh_token: "old-refresh".into(),
            token_type: "DPoP".into(),
            granted_scope: "atproto transition:generic".into(),
            expires_at: Some(NOW + 60),
        }
    }

    fn response() -> TokenResponse {
        TokenResponse {
            access_token: "new-access".into(),
            refresh_token: Some("new-refresh".into()),
            token_type: "DPoP".into(),
            granted_scope: "atproto transition:generic".into(),
            sub: DID.into(),
            expires_in: Some(3600),
        }
    }

    #[test]
    fn a_refresh_replaces_both_tokens_and_the_expiry() {
        let updated = apply_refresh(&session(), &response(), NOW).unwrap();
        assert_eq!(updated.access_token, "new-access");
        assert_eq!(updated.refresh_token, "new-refresh");
        assert_eq!(updated.expires_at, Some(NOW + 3600));
    }

    /// **Rotation is expected but not guaranteed.** A response omitting
    /// `refresh_token` means keep the one we have — storing an empty string
    /// would destroy the session on the next refresh with no way to tell why.
    #[test]
    fn an_omitted_refresh_token_keeps_the_existing_one() {
        let mut response = response();
        response.refresh_token = None;
        let updated = apply_refresh(&session(), &response, NOW).unwrap();
        assert_eq!(updated.refresh_token, "old-refresh");
        assert_eq!(
            updated.access_token, "new-access",
            "the access token still rotates"
        );
    }

    /// `expires_in` is optional, and absent means no proactive refresh rather
    /// than an invented deadline.
    #[test]
    fn an_omitted_expiry_clears_rather_than_invents_one() {
        let mut response = response();
        response.expires_in = None;
        assert_eq!(
            apply_refresh(&session(), &response, NOW)
                .unwrap()
                .expires_at,
            None
        );
    }

    /// **A refresh must not be able to move a session to another account.**
    /// The reference re-checks `sub` on the refresh response for this reason.
    #[test]
    fn a_refresh_for_a_different_subject_is_rejected() {
        let mut response = response();
        response.sub = "did:plc:aaaaaaaaaaaaaaaaaaaaaaaa".into();
        assert!(apply_refresh(&session(), &response, NOW).is_err());
    }

    /// A narrowed grant must be recorded, not silently kept at the old value —
    /// otherwise writes start failing with nothing to explain it.
    #[test]
    fn the_granted_scope_is_taken_from_the_response() {
        let mut response = response();
        response.granted_scope = "atproto".into();
        assert_eq!(
            apply_refresh(&session(), &response, NOW)
                .unwrap()
                .granted_scope,
            "atproto"
        );
    }

    /// The DPoP key and the PDS are properties of the session, not of any one
    /// token response; a refresh must leave them alone.
    #[test]
    fn a_refresh_preserves_the_session_key_and_audience() {
        let updated = apply_refresh(&session(), &response(), NOW).unwrap();
        assert_eq!(updated.dpop_key_jwk, session().dpop_key_jwk);
        assert_eq!(updated.aud, session().aud);
        assert_eq!(updated.issuer, session().issuer);
        assert_eq!(updated.sub, session().sub);
    }

    // ── the per-subject lock ─────────────────────────────────────────────────

    /// Two concurrent refreshes of the SAME subject must not overlap: both would
    /// present the same single-use refresh token, and the loser's
    /// `invalid_grant` typically revokes the winner's new tokens too.
    #[tokio::test]
    async fn the_same_subject_is_serialized() {
        let locks = RefreshLocks::default();
        let held = locks.lock(DID).await;

        let second = locks.lock(DID);
        tokio::pin!(second);
        assert!(
            futures_lite_poll_pending(&mut second),
            "a second holder acquired the lock while the first held it"
        );
        drop(held);
        // Once released, the waiter proceeds.
        let _ = second.await;
    }

    /// Different subjects must NOT block each other, or one slow refresh stalls
    /// every other account.
    #[tokio::test]
    async fn different_subjects_do_not_block_each_other() {
        let locks = RefreshLocks::default();
        let _a = locks.lock(DID).await;
        let b = locks.lock("did:plc:aaaaaaaaaaaaaaaaaaaaaaaa");
        tokio::pin!(b);
        assert!(
            !futures_lite_poll_pending(&mut b),
            "an unrelated subject was blocked"
        );
    }

    /// Poll a future once; true if it is still pending.
    fn futures_lite_poll_pending<F: std::future::Future>(fut: &mut std::pin::Pin<&mut F>) -> bool {
        use std::task::{Context, Poll, Waker};
        let mut cx = Context::from_waker(Waker::noop());
        matches!(fut.as_mut().poll(&mut cx), Poll::Pending)
    }

    /// **The re-discovered issuer must be the one the grant belongs to.**
    ///
    /// This was written inline on both the callback and the refresh path, and
    /// deleting the refresh copy — which sends the REFRESH TOKEN, the credential
    /// that mints every other one — passed all 575 tests. A hostile pair of
    /// documents satisfies every discovery check, because those checks only ask
    /// whether the documents agree with each other.
    #[test]
    fn a_re_discovered_issuer_must_match_the_grants_own() {
        same_issuer("https://pds.example.com", "https://pds.example.com")
            .expect("the same issuer must pass");

        let err = same_issuer("https://evil.example", "https://pds.example.com")
            .expect_err("a different authorization server must be refused");
        let rendered = format!("{err:#}");
        assert!(
            rendered.contains("evil.example") && rendered.contains("pds.example.com"),
            "the error must name both, or an operator cannot tell what moved: {rendered}"
        );
    }

    /// Exact comparison. Issuers are canonicalised by `validate_issuer_form`
    /// before they are ever stored, so a trailing slash or a case difference is
    /// a DIFFERENT issuer, not a spelling of the same one.
    #[test]
    fn the_issuer_comparison_is_exact() {
        assert!(same_issuer("https://pds.example.com/", "https://pds.example.com").is_err());
        assert!(same_issuer("https://PDS.example.com", "https://pds.example.com").is_err());
        assert!(same_issuer("", "https://pds.example.com").is_err());
    }

    // ── the refresh's write vs a concurrent sign-out ─────────────────────────

    const KEY: &str = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";

    /// An authorization server over real TLS whose `/token` rotates to
    /// `rotated-refresh` and whose `/revoke` accepts.
    async fn token_server() -> (String, std::sync::Arc<std::sync::Mutex<Vec<String>>>) {
        let (addr, log) = crate::net::spawn_tls(|_| {
            let mut r = std::collections::HashMap::new();
            r.insert(
                "/token".to_string(),
                vec![crate::net::TestResponse::json(
                    200,
                    serde_json::json!({
                        "access_token": "rotated-access",
                        "refresh_token": "rotated-refresh",
                        "token_type": "DPoP",
                        "scope": "atproto",
                        "sub": DID,
                        "expires_in": 3600,
                    })
                    .to_string(),
                )],
            );
            r.insert(
                "/revoke".to_string(),
                vec![crate::net::TestResponse::json(200, "{}")],
            );
            r
        })
        .await;
        crate::net::test_host_override("as-e2e.test", addr);
        (format!("https://as-e2e.test:{}", addr.port()), log)
    }

    /// An expired session stored under a real codec, and the version read back.
    async fn stored_stale() -> (
        sqlx::SqlitePool,
        crate::oauth::crypto::Codec,
        OAuthSession,
        crate::oauth::store::SessionVersion,
    ) {
        let pool = crate::store::init_url("sqlite::memory:").await.unwrap();
        let codec = crate::oauth::crypto::Codec::new(Some(KEY)).unwrap();
        let stale = OAuthSession {
            dpop_key_jwk: crate::oauth::keys::SigningKey::generate("session-dpop")
                .to_jwk_json()
                .unwrap(),
            expires_at: Some(NOW - 1),
            ..session()
        };
        crate::oauth::store::put_session(&pool, &codec, &stale)
            .await
            .unwrap();
        let (read, version) = crate::oauth::store::get_session_versioned(&pool, &codec, DID)
            .await
            .unwrap()
            .unwrap();
        (pool, codec, read, version)
    }

    fn ctx<'a>(token: &'a str, revoke: &'a str) -> RefreshContext<'a> {
        RefreshContext {
            token_endpoint: token,
            client_id: "http://localhost",
            auth_method: super::super::client_auth::AuthMethod::None,
            client_key: None,
            revocation_endpoint: Some(revoke),
        }
    }

    /// **A refresh must not resurrect a session signed out while it ran.**
    ///
    /// The refresh lock is in-process; a sign-out does not take it (and the
    /// operator's revoke-all runs in another process). The refresh's write was
    /// an upsert, so a sign-out that deleted the row while the token request
    /// was in flight was undone: the session came back holding the fresh
    /// tokens, after being reported signed out — and on Fly no later pass
    /// would ever see it. The fresh tokens belong to no one, so they are
    /// revoked instead.
    #[tokio::test]
    async fn a_refresh_does_not_resurrect_a_session_signed_out_mid_refresh() {
        let (base, log) = token_server().await;
        let (token, revoke) = (format!("{base}/token"), format!("{base}/revoke"));
        let (pool, codec, session, version) = stored_stale().await;

        // The sign-out lands while the refresh is in flight.
        crate::oauth::store::delete_session(&pool, DID)
            .await
            .unwrap();

        let result = refresh_locked(
            &pool,
            &codec,
            &reqwest::Client::new(),
            &session,
            &version,
            &ctx(&token, &revoke),
            NOW,
        )
        .await;

        assert!(
            crate::oauth::store::get_session(&pool, &codec, DID)
                .await
                .unwrap()
                .is_none(),
            "the refresh RESURRECTED a signed-out session"
        );
        let err = result.expect_err("a signed-out session must not be handed back");
        assert!(
            format!("{err:#}").contains("signed out while being refreshed"),
            "{err:#}"
        );
        let seen = log.lock().unwrap().join("\n---\n");
        assert!(
            seen.contains("POST /revoke") && seen.contains("token=rotated-refresh"),
            "the orphaned fresh token was left live:\n{seen}"
        );
    }

    /// **No lost update, and no orphaned grant:** a row another writer
    /// rewrote while this refresh ran (in practice a RE-LOGIN — a concurrent
    /// refresh of the same single-use token cannot also have succeeded) is
    /// kept, not overwritten, and theirs is what the caller gets. The tokens
    /// THIS refresh obtained belong to the old grant, are stored nowhere, and
    /// would stay live at the PDS — so they are revoked.
    #[tokio::test]
    async fn a_refresh_does_not_overwrite_a_concurrent_rotation() {
        let (base, log) = token_server().await;
        let (token, revoke) = (format!("{base}/token"), format!("{base}/revoke"));
        let (pool, codec, session, version) = stored_stale().await;

        let theirs = OAuthSession {
            refresh_token: "their-refresh".into(),
            access_token: "their-access".into(),
            expires_at: Some(NOW + 3600),
            ..session.clone()
        };
        crate::oauth::store::put_session(&pool, &codec, &theirs)
            .await
            .unwrap();

        let got = refresh_locked(
            &pool,
            &codec,
            &reqwest::Client::new(),
            &session,
            &version,
            &ctx(&token, &revoke),
            NOW,
        )
        .await
        .expect("a concurrent rotation is not an error");

        let stored = crate::oauth::store::get_session(&pool, &codec, DID)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(
            stored.refresh_token, "their-refresh",
            "the other writer's rotation was overwritten"
        );
        assert_eq!(got.refresh_token, "their-refresh");
        let revokes: Vec<String> = log
            .lock()
            .unwrap()
            .iter()
            .filter(|r| r.starts_with("POST /revoke"))
            .cloned()
            .collect();
        assert!(
            revokes.iter().any(|r| r.contains("token=rotated-refresh")),
            "the refresh's own fresh token was left live, stored nowhere:\n{revokes:#?}"
        );
        assert!(
            !revokes.iter().any(|r| r.contains("their-refresh")),
            "revoked the OTHER writer's live token:\n{revokes:#?}"
        );
    }

    /// The ordinary refresh still writes.
    #[tokio::test]
    async fn an_uncontested_refresh_stores_the_rotated_tokens() {
        let (base, _log) = token_server().await;
        let (token, revoke) = (format!("{base}/token"), format!("{base}/revoke"));
        let (pool, codec, session, version) = stored_stale().await;

        let got = refresh_locked(
            &pool,
            &codec,
            &reqwest::Client::new(),
            &session,
            &version,
            &ctx(&token, &revoke),
            NOW,
        )
        .await
        .expect("refresh");
        assert_eq!(got.refresh_token, "rotated-refresh");
        assert_eq!(
            crate::oauth::store::get_session(&pool, &codec, DID)
                .await
                .unwrap()
                .unwrap()
                .refresh_token,
            "rotated-refresh"
        );
    }
}