diff --git a/lib/api/apiUtils/rateLimit/refillJob.js b/lib/api/apiUtils/rateLimit/refillJob.js index 3303d3c30a..3bc7b65e01 100644 --- a/lib/api/apiUtils/rateLimit/refillJob.js +++ b/lib/api/apiUtils/rateLimit/refillJob.js @@ -3,14 +3,10 @@ const { getAllTokenBuckets, cleanupTokenBuckets } = require('./tokenBucket'); let refillTimer = null; // Refill interval in milliseconds (how often to check and refill buckets) -const REFILL_INTERVAL_MS = process.env.REFILL_INTERVAL_MS - ? parseInt(process.env.REFILL_INTERVAL_MS, 10) - : 100; +const REFILL_INTERVAL_MS = process.env.REFILL_INTERVAL_MS ? parseInt(process.env.REFILL_INTERVAL_MS, 10) : 100; // Cleanup interval for expired buckets (every 10 seconds) -const CLEANUP_INTERVAL_MS = process.env.CLEANUP_INTERVAL_MS - ? parseInt(process.env.CLEANUP_INTERVAL_MS, 10) - : 10000; +const CLEANUP_INTERVAL_MS = process.env.CLEANUP_INTERVAL_MS ? parseInt(process.env.CLEANUP_INTERVAL_MS, 10) : 10000; let cleanupCounter = 0; @@ -38,19 +34,22 @@ async function refillTokenBuckets(logger) { checked++; // Trigger async refill if needed (non-blocking) - const promise = bucket.refillIfNeeded().then(bucketRefilled => { - // Check if refill actually happened - if (bucketRefilled) { - refilled++; - } - }).catch(err => { - logger.error('error refilling token bucket', { - bucketName, - method: 'rateLimit.refillTokenBuckets', - error: err.message, - stack: err.stack, + const promise = bucket + .refillIfNeeded(logger) + .then(bucketRefilled => { + // Check if refill actually happened + if (bucketRefilled) { + refilled++; + } + }) + .catch(err => { + logger.error('error refilling token bucket', { + bucketName, + method: 'rateLimit.refillTokenBuckets', + error: err.message, + stack: err.stack, + }); }); - }); refillPromises.push(promise); } diff --git a/lib/api/apiUtils/rateLimit/tokenBucket.js b/lib/api/apiUtils/rateLimit/tokenBucket.js index cdff9b969d..4490cf3ab9 100644 --- a/lib/api/apiUtils/rateLimit/tokenBucket.js +++ b/lib/api/apiUtils/rateLimit/tokenBucket.js @@ -12,7 +12,7 @@ const util = require('util'); -const { instance: redisClient } = require('./client'); +const rateLimitClient = require('./client'); const { config } = require('../../../Config'); const { calculateInterval } = require('./gcra'); @@ -23,12 +23,11 @@ const tokenBuckets = new Map(); * Per-resourceClass+resourceID+measure token bucket for a single worker */ class WorkerTokenBucket { - constructor(resourceClass, resourceId, measure, limitConfig, log) { + constructor(resourceClass, resourceId, measure, limitConfig) { this.resourceClass = resourceClass; this.resourceId = resourceId; this.measure = measure; this.limitConfig = limitConfig; - this.log = log; this.bufferSize = config.rateLimiting?.tokenBucketBufferSize; // Max tokens to hold this.refillThreshold = config.rateLimiting?.tokenBucketRefillThreshold; // Trigger refill when below this @@ -43,7 +42,8 @@ class WorkerTokenBucket { } updateLimit(updatedConfig) { - if (this.limitConfig.limit !== updatedConfig.limit || + if ( + this.limitConfig.limit !== updatedConfig.limit || this.limitConfig.burstCapacity !== updatedConfig.burstCapacity ) { const oldConfig = this.limitConfig; @@ -73,9 +73,13 @@ class WorkerTokenBucket { * Check if refill is needed and trigger async refill * Called by background job every 100ms * + * The logger is deliberately taken per call and never stored on the + * bucket: a retained request logger buffers entries forever. + * + * @param {object} log - Logger instance, supplied by the caller * @returns {Promise} */ - async refillIfNeeded() { + async refillIfNeeded(log) { // Already refilling, skip if (this.refillInProgress) { return false; @@ -99,6 +103,9 @@ class WorkerTokenBucket { // Calculate GCRA parameters let granted = requested; + // Read the instance at call time rather than destructuring it at + // module load, so tests can substitute a fake client. + const redisClient = rateLimitClient.instance; if (redisClient.isReady()) { // Request tokens from Redis (atomic GCRA enforcement) granted = await util.promisify(redisClient.grantTokens.bind(redisClient))( @@ -113,14 +120,11 @@ class WorkerTokenBucket { // Connection to redis has failed in some way. // Client will be reconnecting in the background. // We grant the requested amount of tokens anyway to avoid degrading service availability. - this.log.warn( - 'rate limit redis client not connected. granting tokens anyway to avoid service degradation', - { - resourceClass: this.resourceClass, - resourceId: this.resourceId, - measure: this.measure, - }, - ); + log.warn('rate limit redis client not connected. granting tokens anyway to avoid service degradation', { + resourceClass: this.resourceClass, + resourceId: this.resourceId, + measure: this.measure, + }); } // Add granted tokens to buffer @@ -129,7 +133,7 @@ class WorkerTokenBucket { this.lastRefillTime = Date.now(); const duration = this.lastRefillTime - startTime; - this.log.debug('Token refill completed', { + log.debug('Token refill completed', { resourceClass: this.resourceClass, resourceId: this.resourceId, measure: this.measure, @@ -141,7 +145,7 @@ class WorkerTokenBucket { // Warn if refill took too long or granted too few if (duration > 100) { - this.log.warn('Slow token refill detected', { + log.warn('Slow token refill detected', { resourceClass: this.resourceClass, resourceId: this.resourceId, measure: this.measure, @@ -150,7 +154,7 @@ class WorkerTokenBucket { } if (granted === 0 && requested > 0) { - this.log.trace('Token refill denied - quota exhausted', { + log.trace('Token refill denied - quota exhausted', { resourceClass: this.resourceClass, resourceId: this.resourceId, measure: this.measure, @@ -162,7 +166,7 @@ class WorkerTokenBucket { return true; } catch (err) { - this.log.error('Token refill failed', { + log.error('Token refill failed', { resourceClass: this.resourceClass, resourceId: this.resourceId, measure: this.measure, @@ -190,7 +194,7 @@ function getTokenBucket(resourceClass, resourceId, measure, limitConfig, log) { const cacheKey = `${resourceClass}:${resourceId}:${measure}`; let bucket = tokenBuckets.get(cacheKey); if (!bucket) { - bucket = new WorkerTokenBucket(resourceClass, resourceId, measure, limitConfig, log); + bucket = new WorkerTokenBucket(resourceClass, resourceId, measure, limitConfig); tokenBuckets.set(cacheKey, bucket); log.debug('Created token bucket', { @@ -225,7 +229,7 @@ function getAllTokenBuckets() { * Clean up expired token buckets * Called periodically by cleanup job * - * @param {number} maxIdleMs - Remove buckets idle for more than this duration + * @param {number} maxIdleMs - Remove buckets unused by any request for more than this duration * @returns {number} Number of buckets removed */ function cleanupTokenBuckets(maxIdleMs = 60000) { @@ -234,7 +238,7 @@ function cleanupTokenBuckets(maxIdleMs = 60000) { for (const [key, bucket] of tokenBuckets.entries()) { const idleTime = now - bucket.lastRefillTime; - if (idleTime > maxIdleMs && bucket.tokens === 0) { + if (idleTime > maxIdleMs) { toRemove.push(key); } } diff --git a/package.json b/package.json index e330afd66a..6a5ab20892 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@zenko/cloudserver", - "version": "9.3.13", + "version": "9.3.13-1", "description": "Zenko CloudServer, an open-source Node.js implementation of a server handling the Amazon S3 protocol", "main": "index.js", "engines": { diff --git a/tests/unit/api/apiUtils/rateLimit/tokenBucket.js b/tests/unit/api/apiUtils/rateLimit/tokenBucket.js index 7959c01adc..97f8120502 100644 --- a/tests/unit/api/apiUtils/rateLimit/tokenBucket.js +++ b/tests/unit/api/apiUtils/rateLimit/tokenBucket.js @@ -2,6 +2,7 @@ const assert = require('assert'); const sinon = require('sinon'); const tokenBucket = require('../../../../../lib/api/apiUtils/rateLimit/tokenBucket'); +const rateLimitClient = require('../../../../../lib/api/apiUtils/rateLimit/client'); const { config } = require('../../../../../lib/Config'); describe('WorkerTokenBucket', () => { @@ -145,7 +146,12 @@ describe('WorkerTokenBucket', () => { describe('updateLimit', () => { it('should update limitConfig and interval when limit changes', () => { const bucket = new tokenBucket.WorkerTokenBucket( - 'bucket', 'test-bucket', 'rps', { limit: 100, burstCapacity: 1000 }, mockLog); + 'bucket', + 'test-bucket', + 'rps', + { limit: 100, burstCapacity: 1000 }, + mockLog, + ); const oldInterval = bucket.interval; const result = bucket.updateLimit({ limit: 200, burstCapacity: 1000 }); @@ -158,7 +164,12 @@ describe('WorkerTokenBucket', () => { it('should update limitConfig when burstCapacity changes', () => { const bucket = new tokenBucket.WorkerTokenBucket( - 'bucket', 'test-bucket', 'rps', { limit: 100, burstCapacity: 1000 }, mockLog); + 'bucket', + 'test-bucket', + 'rps', + { limit: 100, burstCapacity: 1000 }, + mockLog, + ); const result = bucket.updateLimit({ limit: 100, burstCapacity: 2000 }); @@ -168,7 +179,12 @@ describe('WorkerTokenBucket', () => { it('should return updated: false when config is unchanged', () => { const bucket = new tokenBucket.WorkerTokenBucket( - 'bucket', 'test-bucket', 'rps', { limit: 100, burstCapacity: 1000 }, mockLog); + 'bucket', + 'test-bucket', + 'rps', + { limit: 100, burstCapacity: 1000 }, + mockLog, + ); const result = bucket.updateLimit({ limit: 100, burstCapacity: 1000 }); @@ -178,23 +194,21 @@ describe('WorkerTokenBucket', () => { describe('refillIfNeeded', () => { it('should skip refill when above threshold', async () => { - const bucket = new tokenBucket.WorkerTokenBucket( - 'bucket', 'test-bucket', 'rps', { limit: 100 }, mockLog); + const bucket = new tokenBucket.WorkerTokenBucket('bucket', 'test-bucket', 'rps', { limit: 100 }, mockLog); bucket.tokens = 30; // Above threshold of 20 - await bucket.refillIfNeeded(); + await bucket.refillIfNeeded(mockLog); // refillInProgress was never set (no refill attempted) assert.ok(!bucket.refillInProgress); }); it('should skip refill when already in progress', async () => { - const bucket = new tokenBucket.WorkerTokenBucket( - 'bucket', 'test-bucket', 'rps', { limit: 100 }, mockLog); + const bucket = new tokenBucket.WorkerTokenBucket('bucket', 'test-bucket', 'rps', { limit: 100 }, mockLog); bucket.tokens = 10; // Below threshold bucket.refillInProgress = true; - await bucket.refillIfNeeded(); + await bucket.refillIfNeeded(mockLog); // Still true — function returned early without clearing it assert.strictEqual(bucket.refillInProgress, true); @@ -202,15 +216,128 @@ describe('WorkerTokenBucket', () => { it('should trigger refill when below threshold', async () => { const bucket = new tokenBucket.WorkerTokenBucket( - 'bucket', 'test-bucket', 'rps', { limit: 100, burstCapacity: 1000 }, mockLog); + 'bucket', + 'test-bucket', + 'rps', + { limit: 100, burstCapacity: 1000 }, + mockLog, + ); bucket.tokens = 10; // Below threshold of 20 - await bucket.refillIfNeeded(); + await bucket.refillIfNeeded(mockLog); // refillInProgress is cleared in finally block regardless of outcome assert.strictEqual(bucket.refillInProgress, false); }); }); + + describe('refillIfNeeded against a fake Redis client', () => { + // In the unit environment rateLimitClient.instance is undefined + // (rate limiting is disabled at Config load), so every test above + // exercises only the error path. A fake instance reaches the grant, + // denial, disconnected and slow paths. + let savedRedisClientInstance; + + beforeEach(() => { + savedRedisClientInstance = rateLimitClient.instance; + }); + + afterEach(() => { + rateLimitClient.instance = savedRedisClientInstance; + }); + + function makeBucketBelowThreshold() { + const bucket = new tokenBucket.WorkerTokenBucket( + 'bucket', + 'test-bucket', + 'rps', + { limit: 100, burstCapacity: 1000 }, + mockLog, + ); + bucket.tokens = 10; // Below refill threshold of 20 + return bucket; + } + + it('should add granted tokens to the buffer and log the refill', async () => { + rateLimitClient.instance = { + isReady: () => true, + grantTokens: (resourceClass, resourceId, measure, requested, interval, burstCapacity, callback) => + callback(null, requested), + }; + const bucket = makeBucketBelowThreshold(); + + const refilled = await bucket.refillIfNeeded(mockLog); + + assert.strictEqual(refilled, true); + assert.strictEqual(bucket.tokens, bucket.bufferSize); + assert.ok(mockLog.debug.calledWithMatch('Token refill completed')); + }); + + it('should trace a denial when Redis grants zero tokens', async () => { + rateLimitClient.instance = { + isReady: () => true, + grantTokens: (resourceClass, resourceId, measure, requested, interval, burstCapacity, callback) => + callback(null, 0), + }; + const bucket = makeBucketBelowThreshold(); + + const refilled = await bucket.refillIfNeeded(mockLog); + + assert.strictEqual(refilled, false); + assert.strictEqual(bucket.tokens, 10); + assert.ok(mockLog.trace.calledWithMatch('Token refill denied - quota exhausted')); + }); + + it('should grant the requested tokens and warn when Redis is not connected', async () => { + rateLimitClient.instance = { + isReady: () => false, + grantTokens: () => { + throw new Error('must not be called while disconnected'); + }, + }; + const bucket = makeBucketBelowThreshold(); + + const refilled = await bucket.refillIfNeeded(mockLog); + + // fail-open: full requested amount granted locally + assert.strictEqual(refilled, true); + assert.strictEqual(bucket.tokens, bucket.bufferSize); + assert.ok( + mockLog.warn.calledWithMatch( + 'rate limit redis client not connected. granting tokens anyway to avoid service degradation', + ), + ); + }); + + it('should warn when a refill takes longer than 100ms', async () => { + rateLimitClient.instance = { + isReady: () => true, + grantTokens: (resourceClass, resourceId, measure, requested, interval, burstCapacity, callback) => + setTimeout(() => callback(null, requested), 110), + }; + const bucket = makeBucketBelowThreshold(); + + const refilled = await bucket.refillIfNeeded(mockLog); + + assert.strictEqual(refilled, true); + assert.ok(mockLog.warn.calledWithMatch('Slow token refill detected')); + }); + + it('should log and return false when the grant throws', async () => { + rateLimitClient.instance = { + isReady: () => true, + grantTokens: (resourceClass, resourceId, measure, requested, interval, burstCapacity, callback) => + callback(new Error('redis exploded')), + }; + const bucket = makeBucketBelowThreshold(); + + const refilled = await bucket.refillIfNeeded(mockLog); + + assert.strictEqual(refilled, false); + assert.strictEqual(bucket.refillInProgress, false); + assert.ok(mockLog.error.calledWithMatch('Token refill failed')); + }); + }); }); describe('Token bucket management functions', () => { @@ -291,11 +418,21 @@ describe('Token bucket management functions', () => { it('should update limitConfig when limit changes', () => { const bucket1 = tokenBucket.getTokenBucket( - 'bucket', 'test-bucket', 'rps', { limit: 100, source: 'bucket' }, mockLog); + 'bucket', + 'test-bucket', + 'rps', + { limit: 100, source: 'bucket' }, + mockLog, + ); assert.strictEqual(bucket1.limitConfig.limit, 100); const bucket2 = tokenBucket.getTokenBucket( - 'bucket', 'test-bucket', 'rps', { limit: 200, source: 'bucket' }, mockLog); + 'bucket', + 'test-bucket', + 'rps', + { limit: 200, source: 'bucket' }, + mockLog, + ); assert.strictEqual(bucket1, bucket2); assert.strictEqual(bucket2.limitConfig.limit, 200); @@ -352,7 +489,7 @@ describe('Token bucket management functions', () => { }); describe('cleanupTokenBuckets', () => { - it('should remove idle buckets with no tokens', () => { + it('should remove buckets unused for longer than maxIdleMs', () => { const bucket1 = tokenBucket.getTokenBucket('bucket', 'bucket-1', 'rps', { limit: 100 }, mockLog); const bucket2 = tokenBucket.getTokenBucket('bucket', 'bucket-2', 'rps', { limit: 200 }, mockLog); @@ -370,16 +507,16 @@ describe('Token bucket management functions', () => { assert(!tokenBucket.getAllTokenBuckets().has('bucket:bucket-1:rps')); }); - it('should not remove idle buckets with tokens', () => { + it('should remove unused buckets whatever the remaining token count', () => { const bucket = tokenBucket.getTokenBucket('bucket', 'test-bucket', 'rps', { limit: 100 }, mockLog); bucket.lastRefillTime = Date.now() - 120000; - bucket.tokens = 10; + bucket.tokens = bucket.bufferSize; const removed = tokenBucket.cleanupTokenBuckets(60000); - assert.strictEqual(removed, 0); - assert.strictEqual(tokenBucket.getAllTokenBuckets().size, 1); + assert.strictEqual(removed, 1); + assert.strictEqual(tokenBucket.getAllTokenBuckets().size, 0); }); it('should not remove recently active buckets', () => { diff --git a/tests/unit/api/apiUtils/rateLimit/tokenBucketRetention.js b/tests/unit/api/apiUtils/rateLimit/tokenBucketRetention.js new file mode 100644 index 0000000000..eff3f07810 --- /dev/null +++ b/tests/unit/api/apiUtils/rateLimit/tokenBucketRetention.js @@ -0,0 +1,111 @@ +const assert = require('assert'); +const sinon = require('sinon'); + +const tokenBucket = require('../../../../../lib/api/apiUtils/rateLimit/tokenBucket'); +const { config } = require('../../../../../lib/Config'); + +function makeLog() { + return { + trace: sinon.stub(), + debug: sinon.stub(), + info: sinon.stub(), + warn: sinon.stub(), + error: sinon.stub(), + }; +} + +function logCallCount(log) { + return log.trace.callCount + log.debug.callCount + log.info.callCount + log.warn.callCount + log.error.callCount; +} + +/** + * Regression tests: token buckets used to retain the request logger that + * created them, and the refill job logged through it forever - werelogs + * buffers those entries until an error-level write, so memory grew per + * resource until process restart. + */ +describe('rate limit token bucket retention', () => { + let sandbox; + + beforeEach(() => { + sandbox = sinon.createSandbox(); + sandbox.stub(config, 'rateLimiting').value({ + enabled: true, + nodes: 1, + tokenBucketBufferSize: 50, + tokenBucketRefillThreshold: 20, + }); + tokenBucket.getAllTokenBuckets().clear(); + }); + + afterEach(() => { + sandbox.restore(); + tokenBucket.getAllTokenBuckets().clear(); + }); + + it('should not retain the request logger that created the bucket', () => { + const requestLog = makeLog(); + + const bucket = tokenBucket.getTokenBucket( + 'account', + 'acct-retention-1', + 'rps', + { limit: 60, burstCapacity: 1000 }, + requestLog, + ); + + const retained = Object.keys(bucket).filter(key => bucket[key] === requestLog); + + assert.deepStrictEqual( + retained, + [], + `token bucket must not hold the per-request logger (held on: ${retained.join(', ')})`, + ); + }); + + it('should log refill activity to the caller-supplied logger, not the creating request logger', async () => { + const requestLog = makeLog(); + const jobLog = makeLog(); + + const bucket = tokenBucket.getTokenBucket( + 'account', + 'acct-retention-2', + 'rps', + { limit: 60, burstCapacity: 1000 }, + requestLog, + ); + + // ignore the "Created token bucket" line, which is legitimately + // request-scoped and dies with the request + requestLog.debug.resetHistory(); + + bucket.tokens = 0; // below refill threshold, forces a refill + await bucket.refillIfNeeded(jobLog); + + assert.strictEqual( + logCallCount(requestLog), + 0, + 'refill must not log through the request logger that created the bucket', + ); + assert.ok(logCallCount(jobLog) > 0, 'refill must log through the logger supplied by the refill job'); + }); + + it('should evict buckets idle longer than maxIdleMs even when they still hold tokens', () => { + const requestLog = makeLog(); + + const bucket = tokenBucket.getTokenBucket( + 'account', + 'acct-retention-3', + 'rps', + { limit: 60, burstCapacity: 1000 }, + requestLog, + ); + + bucket.lastRefillTime = Date.now() - 120000; + + const removed = tokenBucket.cleanupTokenBuckets(60000); + + assert.strictEqual(removed, 1, 'idle bucket must be evicted even with a full token buffer'); + assert.strictEqual(tokenBucket.getAllTokenBuckets().size, 0); + }); +});