axum_session 0.19.0

📝 Session management layer for axum that supports HTTP and Rest.
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

use crate::{DatabasePool, SessionError, SessionStore};
use chrono::Utc;
#[cfg(feature = "key-store")]
use fastbloom_rs::Deletable;
use std::fmt::Debug;
use tokio::time::sleep;

pub async fn runner<T>(
    session_store: SessionStore<T>,
    kill_as_duplicate: bool,
) -> Result<(), SessionError>
where
    T: DatabasePool + Clone + Debug + Sync + Send + 'static,
{
    if kill_as_duplicate {
        tracing::trace!(
            "Killed Session Manager Duplicate on thread ID: {:?}.",
            std::thread::current().id()
        );
        return Ok(());
    } else {
        tracing::trace!("Session Manager is starting loop");
    }

    loop {
        let (last_sweep, last_database_sweep) = {
            let timers = session_store.timers.read().await;
            (timers.last_expiry_sweep, timers.last_database_expiry_sweep)
        };

        let current_time = Utc::now();

        if last_sweep <= current_time && !session_store.config.memory.memory_lifespan.is_zero() {
            tracing::trace!("Session Memory Cleaning Started");

            // Only unload these from filter if the Client is None as this means no database.
            // Otherwise only unload from the filter if removed from the Database.
            #[cfg(feature = "key-store")]
            if session_store.is_persistent()
                && session_store.auto_handles_expiry()
                && session_store.config.memory.use_bloom_filters
            {
                let mut filter = session_store.filter.write().await;
                session_store
                    .inner
                    .iter()
                    .filter(|r| r.autoremove < current_time)
                    .for_each(|r| filter.remove(r.key().as_bytes()));
            }

            // We do this to ensure that if Types that never got updated due to the newer timers that on memory unload
            // we ensure they get synced before unloading them from memory but only if the database/cookie has not expired.
            if session_store.is_persistent() {
                for session in session_store
                    .inner
                    .iter()
                    .filter(|r| r.autoremove < current_time)
                {
                    if !session.expired() {
                        if let Err(err) = session_store.store_session(&session).await {
                            tracing::debug!(
                                "Session Failed to save to Database with error: {}",
                                err
                            );
                            continue;
                        } else {
                            tracing::debug!(
                                "Session id {}: was saved to the database.",
                                session.id
                            );
                        }
                    }
                }
            }

            session_store
                .inner
                .retain(|_k, v| v.autoremove > current_time);

            session_store.timers.write().await.last_expiry_sweep =
                Utc::now() + session_store.config.memory.purge_update;

            tracing::trace!("Session Memory Cleaning Finished");
        }

        // Throttle by database lifespan - e.g. sweep every 6 hours
        if last_database_sweep <= current_time && session_store.is_persistent() {
            tracing::trace!("Session Database Cleaning Started");

            //Remove any old keys that expired and Remove them from our loaded filter.
            #[cfg(feature = "key-store")]
            let expired = match session_store.cleanup().await {
                Ok(v) => v,
                Err(err) => {
                    tracing::error!("Session Database Cleaning Failed",);
                    continue;
                }
            };

            #[cfg(not(feature = "key-store"))]
            if let Err(err) = session_store.cleanup().await {
                tracing::error!("Session Database Cleaning Failed with error: {}", err);
                continue;
            }

            #[cfg(feature = "key-store")]
            if !session_store.auto_handles_expiry() {
                let mut filter = session_store.filter.write().await;
                expired.iter().for_each(|id| filter.remove(id.as_bytes()));
            }

            session_store
                .timers
                .write()
                .await
                .last_database_expiry_sweep =
                Utc::now() + session_store.config.database.purge_database_update;

            tracing::trace!("Session Database Cleaning Finished");
        }

        tracing::trace!("Session Manager is Sleeping");
        sleep(
            session_store
                .config
                .thread_sleep_duration
                .to_std()
                .unwrap_or_default(),
        )
        .await;
    }
}