-
Notifications
You must be signed in to change notification settings - Fork 106
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[ASI-614] Bull queue to expire user session tokens (#1905)
* [ASI-614] Bull queue to expire user session tokens Prevent replay attacks by expiring user session tokens with a daily cron and Bull queue in content node. See also: [ASI-614][1] [1]: https://linear.app/audius/issue/ASI-614 * [ASI-614] Batch redis tx for latency, retry DB tx * Use Redis MULTI feature to batch handle transactions in SessionExpirationQueue * Implement retry for DB bulk session delete call in SessionManager before rollback See also: [ASI-614][1] [1]: https://linear.app/audius/issue/ASI-614 i * [ASI-614] Add test scaffolding See also: [ASI-614][1] [1]: https://linear.app/audius/issue/ASI-614 * [ASI-614] Test coverage for session expiration * add tests for SessionManager * add tests for DBManager See also: [ASI-614][1] [1]: https://linear.app/audius/issue/ASI-614 * [ASI-614] Clean up logs and documentation for session token expiration queue * Fix documentation for dbManager
- Loading branch information
Showing
8 changed files
with
382 additions
and
20 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,112 @@ | ||
| const Bull = require('bull') | ||
| const Sequelize = require('sequelize') | ||
| const sessionManager = require('../sessionManager') | ||
| const config = require('../config') | ||
| const { logger } = require('../logging') | ||
| const { SessionToken } = require('../models') | ||
|
|
||
| const RUN_INTERVAL = 60 * 1000 * 60 * 24 // daily run | ||
| const SESSION_EXPIRATION_AGE = 60 * 1000 * 60 * 24 * 14 // 2 weeks | ||
| const BATCH_SIZE = 100 | ||
| const PROCESS_NAMES = Object.freeze({ | ||
| expire_sessions: 'expire_sessions' | ||
| }) | ||
|
|
||
| /** | ||
| * A persistent cron-style queue that periodically deletes expired session tokens from Redis cache and the database. Runs on startup, deleting 100 sessions at a time, and then runs daily to clear sessions older than 14d. | ||
| * | ||
| */ | ||
| class SessionExpirationQueue { | ||
| constructor () { | ||
| this.sessionExpirationAge = SESSION_EXPIRATION_AGE | ||
| this.batchSize = BATCH_SIZE | ||
| this.runInterval = RUN_INTERVAL | ||
| this.queue = new Bull( | ||
| 'session-expiration-queue', | ||
| { | ||
| redis: { | ||
| port: config.get('redisPort'), | ||
| host: config.get('redisHost') | ||
| }, | ||
| defaultJobOptions: { | ||
| removeOnComplete: true, | ||
| removeOnFail: true | ||
| } | ||
| } | ||
| ) | ||
| this.logStatus = this.logStatus.bind(this) | ||
| this.expireSessions = this.expireSessions.bind(this) | ||
|
|
||
| // Clean up anything that might be still stuck in the queue on restart | ||
| this.queue.empty() | ||
|
|
||
| this.queue.process( | ||
| PROCESS_NAMES.expire_sessions, | ||
| /* concurrency */ 1, | ||
| async (job, done) => { | ||
| try { | ||
| this.logStatus('Starting') | ||
| let progress = 0 | ||
| const SESSION_EXPIRED_CONDITION = { | ||
| where: { | ||
| created_at: { | ||
| [Sequelize.Op.gt]: new Date(Date.now() - this.sessionExpirationAge) | ||
| } | ||
| } | ||
| } | ||
| const numExpiredSessions = await SessionToken.count(SESSION_EXPIRED_CONDITION) | ||
| this.logStatus(`${numExpiredSessions} expired sessions ready for deletion.`) | ||
|
|
||
| let sessionsToDelete = numExpiredSessions | ||
| while (sessionsToDelete > 0) { | ||
| await this.expireSessions(SESSION_EXPIRED_CONDITION) | ||
| progress += (this.batchSize / numExpiredSessions) * 100 | ||
| job.progress(progress) | ||
| sessionsToDelete -= this.batchSize | ||
| } | ||
| done(null, {}) | ||
| } catch (e) { | ||
| this.logStatus(`Error ${e}`) | ||
| done(e) | ||
| } | ||
| } | ||
| ) | ||
| } | ||
|
|
||
| async expireSessions (sessionExpiredCondition) { | ||
| const sessionsToDelete = await SessionToken.findAll(Object.assign(sessionExpiredCondition, { limit: this.batchSize })) | ||
| await sessionManager.deleteSessions(sessionsToDelete) | ||
| } | ||
|
|
||
| /** | ||
| * Logs a status message and includes current queue info | ||
| * @param {string} message | ||
| */ | ||
| async logStatus (message) { | ||
| const { waiting, active, completed, failed, delayed } = await this.queue.getJobCounts() | ||
| logger.info(`Session Expiration Queue: ${message} || active: ${active}, waiting: ${waiting}, failed ${failed}, delayed: ${delayed}, completed: ${completed} `) | ||
| } | ||
|
|
||
| /** | ||
| * Starts the session expiration queue on a daily cron. | ||
| */ | ||
| async start () { | ||
| try { | ||
| // Run the job immediately | ||
| await this.queue.add(PROCESS_NAMES.expire_sessions) | ||
|
|
||
| // Then enqueue the job to run on a regular interval | ||
| setInterval(async () => { | ||
| try { | ||
| await this.queue.add(PROCESS_NAMES.expire_sessions) | ||
| } catch (e) { | ||
| this.logStatus('Failed to enqueue!') | ||
| } | ||
| }, this.runInterval) | ||
| } catch (e) { | ||
| this.logStatus('Startup failed!') | ||
| } | ||
| } | ||
| } | ||
|
|
||
| module.exports = SessionExpirationQueue |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.