login: wait for a session's end from before each check

A stream opened with a token ends once refused_now refuses it. In a
swarm without a registry, a session ended on this node is kept in memory
(AppState::ended_sessions), and the streams waiting on session_ended are
woken (notify_waiters). credentials_end now makes the Notified it waits
on before each refused_now, the first one included, and not after: a
session ended between a check and the wait is seen at once rather than
at the next change to protocol data.

1 files changed, 6 insertions, 2 deletions

diff --git a/src/api/login.rs b/src/api/login.rs
index 694ad0a..1c8aa44 100644
--- a/src/api/login.rs
+++ b/src/api/login.rs
@@ -428,6 +428,10 @@ async fn credentials_end(
         Credentials::Token(claims) => claims.user.clone(),
         Credentials::User(user) => user.clone(),
     };
+    // Made before each check, so a session ended on this node while it runs is seen
+    // (`notify_waiters` wakes a `Notified` as soon as it is made).
+    let session_ended = state.session_ended.notified();
+    tokio::pin!(session_ended);
     if let Some(refused) = refused_now(&state, &who).await {
         debug!(user = %user, reason = %refused, "Ended a stream with what let it in");
         return;
@@ -446,10 +450,9 @@ async fn credentials_end(
     };
     tokio::pin!(expiry);
     loop {
-        let session_ended = state.session_ended.notified();
         tokio::select! {
             () = &mut expiry => return,
-            () = session_ended => {}
+            () = &mut session_ended => {}
             change = changes.recv() => match change {
                 // Protocol data, the root (a switch) among it.
                 Ok(set) if !set.paths.iter().any(|path| {
@@ -459,6 +462,7 @@ async fn credentials_end(
                 Err(tokio::sync::broadcast::error::RecvError::Closed) => return,
             },
         }
+        session_ended.set(state.session_ended.notified());
         if let Some(refused) = refused_now(&state, &who).await {
             debug!(user = %user, reason = %refused, "Ended a stream with what let it in");
             return;