fixed window & small fixes

Author: stefanhts

lib/db.js

@@ -14,6 +14,8 @@ const TAKE_ELEVATED_LUA = fs.readFileSync(`${__dirname}/take_elevated.lua`, 'utf
 const TAKE_EXPONENTIAL_LUA = fs.readFileSync(`${__dirname}/take_exponential.lua`, 'utf8');
 const PUT_LUA = fs.readFileSync(`${__dirname}/put.lua`, 'utf8');
 
+const BACKOFF_FACTOR_DEFAULT = 2;
+const MULTIPLE_UNIT_DEFAULT = 1000;
 const DEFAULT_KEEPALIVE = 10000; // Milliseconds
 
 class LimitDBRedis extends EventEmitter {
@@ -270,31 +272,37 @@ class LimitDBRedis extends EventEmitter {
 
   takeExponential(params, callback) {
     this._doTake(params, callback, (key, bucketKeyConfig, count) => {
+      const useFixedWindow = isFixedWindowEnabled(bucketKeyConfig.fixed_window, params.fixed_window);
       this.redis.takeExponential(key,
         bucketKeyConfig.ms_per_interval || 0,
         bucketKeyConfig.size,
         count,
         Math.ceil(bucketKeyConfig.ttl || this.globalTTL),
         bucketKeyConfig.drip_interval || 0,
-        bucketKeyConfig.backoff_factor || 2,
-        bucketKeyConfig.multiple_unit || 1000,
+        bucketKeyConfig.backoff_factor || BACKOFF_FACTOR_DEFAULT,
+        bucketKeyConfig.multiple_unit || MULTIPLE_UNIT_DEFAULT,
+        useFixedWindow ? bucketKeyConfig.interval : 0,
         (err, results) => {
           if (err) {
             return callback(err);
           }
           const remaining = parseInt(results[0], 10);
           const conformant = parseInt(results[1], 10) ? true : false;
+          const current_timestamp_ms = parseInt(results[2], 10);
           const reset = parseInt(results[3], 10);
           const backoff_factor = parseInt(results[4], 10);
           const backoff_time = parseInt(results[5], 10);
+          const next_token_ms = parseInt(results[6], 10);
           const res = {
             conformant,
             remaining,
             reset: Math.ceil(reset / 1000),
             limit: bucketKeyConfig.size,
             delayed: false,
             backoff_factor,
-            backoff_time
+            backoff_time,
+            delta_reset_ms: Math.max(reset - current_timestamp_ms, 0),
+            delta_backoff_time: next_token_ms - current_timestamp_ms,
           };
           if (bucketKeyConfig.skip_n_calls > 0) {
             this.callCounts.set(key, { res, count: 0 });

lib/take_exponential.lua

@@ -6,6 +6,7 @@ local ttl                  = tonumber(ARGV[4])
 local drip_interval        = tonumber(ARGV[5])
 local backoff_factor       = tonumber(ARGV[6])
 local mult_unit            = tonumber(ARGV[7])
+local fixed_window         = tonumber(ARGV[8])
 
 local current_time         = redis.call('TIME')
 local current_timestamp_ms = current_time[1] * 1000 + current_time[2] / 1000
@@ -18,20 +19,29 @@ end
 
 -- calculate the time of next available token
 local last_token_ms = current[1] or 0
-local step = 0
+local remaining_tokens = 0
 if current[2] then
-    step = tonumber(current[2])
+    remaining_tokens = tonumber(current[2])
 else
-    step = bucket_size
+    remaining_tokens = bucket_size
 end
 
-local backoff_step = bucket_size - step
+local backoff_step = bucket_size - remaining_tokens
 local backoff_time = math.ceil(backoff_factor ^ backoff_step) * mult_unit
 local next_token_ms = last_token_ms + backoff_time
 local is_passed_wait_time = current_timestamp_ms >= next_token_ms
 
 if current[1] and tokens_per_ms then
     -- drip bucket
+
+    if fixed_window > 0 then
+        -- fixed window for granting new tokens
+        local interval_correction = (current_timestamp_ms - last_token_ms) % fixed_window
+        current_timestamp_ms = current_timestamp_ms - interval_correction
+    end
+
+    is_passed_wait_time = current_timestamp_ms >= next_token_ms
+
     if not is_passed_wait_time then
         new_content = tonumber(current[2])
         last_token_ms = current[1]
@@ -63,8 +73,10 @@ redis.call('HMSET', KEYS[1],
 redis.call('EXPIRE', KEYS[1], ttl)
 
 local reset_ms = 0
-if drip_interval > 0 then
+if fixed_window > 0 then
+    reset_ms = current_timestamp_ms + fixed_window
+elseif drip_interval > 0 then
     reset_ms = math.ceil(current_timestamp_ms + (bucket_size - new_content) * drip_interval)
 end
 
-return { new_content, enough_tokens, current_timestamp_ms, reset_ms, backoff_factor, backoff_time, backoff_step }
+return { new_content, enough_tokens, current_timestamp_ms, reset_ms, backoff_factor, backoff_time, next_token_ms }

lib/utils.js

@@ -11,6 +11,11 @@ const INTERVAL_TO_MS = {
 
 const INTERVAL_SHORTCUTS = Object.keys(INTERVAL_TO_MS);
 
+const EXPONENTIAL_BACKOFF_DEFAULTS = {
+  backoff_factor: 0,
+  multiple_unit: 1
+};
+
 const ERL_QUOTA_INTERVAL_PER_CALENDAR_MONTH = 'quota_per_calendar_month';
 const ERL_QUOTA_INTERVALS = {
   [ERL_QUOTA_INTERVAL_PER_CALENDAR_MONTH]: () => endOfMonthTimestamp()
@@ -41,8 +46,8 @@ function normalizeTemporals(params) {
   }
 
   if(type.exponential_backoff) {
-    type.backoff_factor= type.exponential_backoff.backoff_factor || 0;
-    type.multiple_unit = type.exponential_backoff.multiple_unit || 1;
+    type.backoff_factor= type.exponential_backoff.backoff_factor || EXPONENTIAL_BACKOFF_DEFAULTS.backoff_factor;
+    type.multiple_unit = type.exponential_backoff.multiple_unit || EXPONENTIAL_BACKOFF_DEFAULTS.multiple_unit;
     delete type.exponential_backoff;
   }
 

test/db.tests.js

@@ -162,7 +162,6 @@ module.exports.tests = (clientCreator) => {
 
     describe('#configurateBucketKey', () => {
       it('should add new bucket to existing configuration', () => {
-        console.log("this one works....")
         db.configurateBucket('test', { size: 5 });
         assert.containsAllKeys(db.buckets, ['ip', 'test']);
       });

test/utils.tests.js

@@ -55,9 +55,7 @@ describe('utils', () => {
       expect(rest).to.deep.equal({
        backoff_factor: 5,
          multiple_unit: 6
-      })
-
-
+      });
     });
 
     it('should return normalized bucket without ERL', () => {