From 74f06f23fbd4a359c57ad5c56d4664b925e80514 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marko=20Mili=C4=87?= Date: Wed, 28 Oct 2026 07:43:56 +0100 Subject: [PATCH] perf: check and count the consumer quota in one Redis script RateLimitingUtil.underCallLimits used to spend up to 19 Redis round trips per authenticated request on the counters alone: TTL and GET for every limited period, six INCRs when all passed, and another TTL and GET to fill the X-Rate-Limit-Reset header on a 429. The check and the count were also separate operations, so concurrent requests one below a limit could all read the same count, all pass and all increment (hazard X in CONCURRENCY_HAZARDS.md). Redis.quotaCheckAndCount runs one Lua script that checks every period's limit, shortest first, and only when all pass counts the call in every period, returning each counter's count and TTL. A refusal returns the exhausted period with its count and TTL, so the 429 headers need no further read. The counters now cost one round trip per request and the gate is atomic: at most `limit` calls pass per period. Behaviour is unchanged otherwise: same 429 messages and precedence, same X-Rate-Limit-* values, every period still counted whether limited or not, a limit of 0 still refuses before counting, use_consumer_limits=false still skips Redis, and a Redis outage still fails open. The dead underConsumerLimits, incrementConsumerCounters and ttl helpers are removed; getCounterState stays for the call-counter endpoints and incrementCounter for the other limiters. ConcurrentConsumerQuotaScriptTest pins the script's contract and drives 20 concurrent callers against a limit of 5. The Glossary and the hazards table describe the new flow. --- .../src/main/scala/code/api/cache/Redis.scala | 88 ++++++ .../main/scala/code/api/util/Glossary.scala | 4 +- .../code/api/util/RateLimitingUtil.scala | 265 ++++++------------ .../code/concurrency/CONCURRENCY_HAZARDS.md | 2 +- .../ConcurrentConsumerQuotaScriptTest.scala | 173 ++++++++++++ .../ConcurrentRateLimiterRaceTest.scala | 9 +- 6 files changed, 361 insertions(+), 180 deletions(-) create mode 100644 obp-api/src/test/scala/code/concurrency/ConcurrentConsumerQuotaScriptTest.scala diff --git a/obp-api/src/main/scala/code/api/cache/Redis.scala b/obp-api/src/main/scala/code/api/cache/Redis.scala index cd0495b635..8cba91faa3 100644 --- a/obp-api/src/main/scala/code/api/cache/Redis.scala +++ b/obp-api/src/main/scala/code/api/cache/Redis.scala @@ -302,6 +302,94 @@ object Redis extends MdcLoggable { finally jedis.close() } + /** Outcome of [[quotaCheckAndCount]]. */ + sealed trait QuotaDecision + /** A period refused the call: `periodIndex` into the lists passed in (0-based), the counter's + * current count and its remaining TTL in seconds (negative when the key has no expiry or does + * not exist, as Redis TTL reports it). Nothing was counted. */ + final case class QuotaRefused(periodIndex: Int, count: Long, ttlSeconds: Long) extends QuotaDecision + /** Every period passed and the call was counted in all of them: (count after this call, + * remaining TTL in seconds) per key, in the order passed in. */ + final case class QuotaCounted(countsAndTtls: List[(Long, Long)]) extends QuotaDecision + + /** + * The consumer quota in one script: check every period's limit against its counter and, only + * when all pass, count the call in every period. + * + * KEYS[i] counter key of period i, shortest period first (precedence of the 429) + * ARGV[i] limit of period i: -1 unlimited (not checked, still counted), 0 blocked + * (refused without reading the counter's value), > 0 max calls in the period + * ARGV[n + i] length of period i in seconds, the TTL a new counter gets + * + * Returns {0, i, count, ttl} when period i refuses: count is what its counter holds now (0 when + * the key is absent or has no expiry) and ttl is the key's TTL as Redis reports it. Returns + * {1, c1, ttl1, ..., cn, ttln} after counting: the post-increment count and TTL of each key. + * A key found without an expiry is treated as expired on the check and recreated on the count, + * as incrementWithTtlScript does. Everything runs atomically server-side, so the check and + * the count of one call cannot interleave with another call's: at most `limit` calls pass. + */ + private val quotaCheckAndCountScript = + """local n = #KEYS + |for i = 1, n do + | local limit = tonumber(ARGV[i]) + | if limit >= 0 then + | local ttl = redis.call('TTL', KEYS[i]) + | local count = 0 + | if ttl > 0 then count = tonumber(redis.call('GET', KEYS[i]) or '0') end + | if limit == 0 or count + 1 > limit then + | return {0, i, count, ttl} + | end + | end + |end + |local out = {1} + |for i = 1, n do + | local c = redis.call('INCR', KEYS[i]) + | if c == 1 then + | redis.call('EXPIRE', KEYS[i], ARGV[n + i]) + | elseif redis.call('TTL', KEYS[i]) < 0 then + | redis.call('SET', KEYS[i], 1, 'EX', ARGV[n + i]) + | c = 1 + | end + | out[#out + 1] = c + | out[#out + 1] = redis.call('TTL', KEYS[i]) + |end + |return out""".stripMargin + + /** + * Check and count one call against several period counters in a single atomic round trip + * (see quotaCheckAndCountScript). `keys`, `limits` and `periodSeconds` are parallel lists, + * shortest period first. Throws when Redis is unreachable; the caller decides whether to fail + * open. + */ + def quotaCheckAndCount(keys: List[String], limits: List[Long], periodSeconds: List[Int]): QuotaDecision = { + require(keys.nonEmpty && keys.size == limits.size && keys.size == periodSeconds.size, + s"quotaCheckAndCount needs one limit and one period per key: ${keys.size} keys, ${limits.size} limits, ${periodSeconds.size} periods") + import scala.jdk.CollectionConverters._ + val startNanos = System.nanoTime() + var succeeded = false + try { + val reply: List[Long] = withJedis { jedis => + jedis.eval( + quotaCheckAndCountScript, + keys.asJava, + (limits.map(_.toString) ++ periodSeconds.map(_.toString)).asJava + ).asInstanceOf[java.util.List[java.lang.Long]].asScala.map(_.longValue()).toList + } + val decision = reply match { + case 0L :: index :: count :: ttl :: Nil => QuotaRefused(index.toInt - 1, count, ttl) + case 1L :: countsAndTtls if countsAndTtls.size == keys.size * 2 => + QuotaCounted(countsAndTtls.grouped(2).collect { case List(count, ttl) => (count, ttl) }.toList) + case other => throw new RuntimeException(s"quotaCheckAndCount: unexpected reply $other for ${keys.size} keys") + } + succeeded = true + decision + } finally { + code.telemetry.Telemetry.timer("obp.api.redis.commands", + "command" -> "EVAL_QUOTA_CHECK_AND_COUNT", "result" -> (if (succeeded) "success" else "error")) + .record(System.nanoTime() - startNanos, java.util.concurrent.TimeUnit.NANOSECONDS) + } + } + /** * Atomic `SET key value EX ttlSeconds NX` (Jedis 2.9.0 five-arg overload). Sets the key with a TTL * only if it does not already exist, in a single command. Returns true iff this call set the key. diff --git a/obp-api/src/main/scala/code/api/util/Glossary.scala b/obp-api/src/main/scala/code/api/util/Glossary.scala index 8c59b4f8a7..0cde9f60f7 100644 --- a/obp-api/src/main/scala/code/api/util/Glossary.scala +++ b/obp-api/src/main/scala/code/api/util/Glossary.scala @@ -606,8 +606,8 @@ object Glossary extends MdcLoggable { |1. **Rate Limit Records**: Stored in the `RateLimiting` table with date ranges (from_date, to_date) |2. **Multiple Records**: A consumer can have multiple active rate limit records that overlap |3. **Aggregation**: When multiple records are active, per period: `-1` values are ignored and the rest (`0` or positive) are summed; a sum of `0` blocks the period; nothing to sum (all `-1`) means unlimited - |4. **Enforcement**: On every API request, the system checks Redis counters against the aggregated limits - |5. **Counting**: Every served request is counted in the Redis counter of every period, whether or not that period has a limit, so the call-counter endpoints show a Consumer's activity even when nothing limits it. A blocked period (sum `0`) serves nothing, so nothing is counted under it. + |4. **Enforcement**: On every API request, one Redis script checks the counter of every period against the aggregated limits, shortest period first, and refuses with 429 at the first period that is full + |5. **Counting**: In the same script, once every period has passed, the request is counted in the Redis counter of every period, whether or not that period has a limit, so the call-counter endpoints show a Consumer's activity even when nothing limits it. A blocked period (sum `0`) serves nothing, so nothing is counted under it. Because the check and the count run atomically in one round trip, concurrent requests cannot pass a check together and overshoot a limit. | |### Time Periods | diff --git a/obp-api/src/main/scala/code/api/util/RateLimitingUtil.scala b/obp-api/src/main/scala/code/api/util/RateLimitingUtil.scala index 52b64c2afb..0c5c1e3dfb 100644 --- a/obp-api/src/main/scala/code/api/util/RateLimitingUtil.scala +++ b/obp-api/src/main/scala/code/api/util/RateLimitingUtil.scala @@ -228,40 +228,6 @@ object RateLimitingUtil extends MdcLoggable { RateLimitCounterState(calls, normalizedTtl, status) } private def createUniqueKey(consumerKey: String, period: LimitCallPeriod) = CALL_COUNTER_PREFIX + consumerKey + "_" + RateLimitingPeriod.toString(period) - private def underConsumerLimits(consumerKey: String, period: LimitCallPeriod, limit: Long): Boolean = { - - if (useConsumerLimits) { - (limit) match { - case l if l > 0 => // Limit is set, check against Redis counter - val state = getCounterState(consumerKey, period) - state.status match { - case "ACTIVE" => - // Counter is active, check if we're under limit - // +1 means we count the current call as well. We increment later i.e after successful call. - state.calls.getOrElse(0L) + 1 <= limit - case "NO_COUNTER" | "EXPIRED" => - // No counter or expired -> allow (first call or period expired) - true - case "REDIS_UNAVAILABLE" => - // Redis unavailable -> fail open (allow request) - logger.warn(s"Redis unavailable when checking rate limit for consumer $consumerKey, period $period - allowing request") - true - case _ => - // Unknown status -> fail open (allow request) - logger.warn(s"Unknown status '${state.status}' when checking rate limit for consumer $consumerKey, period $period - allowing request") - true - } - case 0 => - // A limit of 0 means blocked: refuse every call for this period, without touching Redis - false - case _ => - // A negative limit (-1) means unlimited for this period - true - } - } else { - true // Rate Limiting disabled implies successful result - } - } /** Pure Redis INCR with create-if-missing for a fully-formed key. * No gates, no key formatting — call sites pass the final key and supply their own enable flags. @@ -282,41 +248,65 @@ object RateLimitingUtil extends MdcLoggable { } } + /** The periods of the consumer quota, shortest first: the order the limits are checked in, so the + * 429 names the shortest exhausted period, and the order the X-Rate-Limit-* headers prefer. */ + private val consumerPeriods: List[LimitCallPeriod] = List(PER_SECOND, PER_MINUTE, PER_HOUR, PER_DAY, PER_WEEK, PER_MONTH) + + private def limitOf(c: CallLimit, period: LimitCallPeriod): Long = period match { + case PER_SECOND => c.per_second + case PER_MINUTE => c.per_minute + case PER_HOUR => c.per_hour + case PER_DAY => c.per_day + case PER_WEEK => c.per_week + case PER_MONTH => c.per_month + case PER_YEAR => -1 + } + + /** What one call did to the quota counters. */ + private[util] sealed trait QuotaOutcome + /** `period` refused the call (its limit is 0, or its counter is full). Nothing was counted. + * `resetSeconds` is the counter's remaining TTL, 0 when it has none. */ + private[util] final case class Refused(period: LimitCallPeriod, limit: Long, resetSeconds: Long) extends QuotaOutcome + /** Every period passed and the call was counted in all of them, unlimited ones included: + * (period, limit, (ttl, count after this call)) in [[consumerPeriods]] order. */ + private[util] final case class Counted(perPeriod: List[(LimitCallPeriod, Long, (Long, Long))]) extends QuotaOutcome + /** - * Count the call in one period's counter, after the request passed every limit check. - * - * Counted whatever the period's aggregated limit is, -1 (unlimited) included. Enforcement never - * reads the counter of an unlimited period - underConsumerLimits decides -1 and 0 without touching - * Redis and only compares the counter when the limit is positive - so a counter that is always - * kept changes nothing about what is refused, and gives visibility of the consumer's activity - * through the call-counter endpoints even when no limit applies. A period whose aggregated limit - * is 0 is refused before counting, so nothing is ever served or counted under it. + * Check the call against every period's limit and, when all pass, count it in every period, in + * ONE Redis round trip (Redis.quotaCheckAndCount, a Lua script). The script runs atomically, so + * two concurrent calls cannot both read a counter one below its limit and both pass: at most + * `limit` calls are served per period. * - * Only `use_consumer_limits=false` skips counting, so an instance without Redis is left alone. + * Counted whatever a period's limit is, -1 (unlimited) included: a limit of -1 is not checked, so + * a counter that is always kept changes nothing about what is refused, and gives visibility of + * the consumer's activity through the call-counter endpoints even when no limit applies. A limit + * of 0 refuses before counting, so nothing is ever served or counted under it. * * @param consumerKey The consumer ID (or the client IP for anonymous access) - * @param period The time period (PER_SECOND, PER_MINUTE, ...) - * @return (TTL in seconds, current counter value); (-1, -1) when not counted or Redis is unreachable + * @param periodsWithLimits the periods to check and count, shortest first, each with its limit + * @return None when nothing was checked or counted: `use_consumer_limits=false` (an instance + * without Redis is left alone) or Redis unreachable (fail open: the call is allowed and + * no X-Rate-Limit-* headers are set) */ - private def incrementConsumerCounters(consumerKey: String, period: LimitCallPeriod): (Long, Long) = - if (useConsumerLimits) incrementCounter(createUniqueKey(consumerKey, period), period) else (-1, -1) - - /** - * Get remaining TTL (time to live) for a rate limit counter. - * Used to populate X-Rate-Limit-Reset header when rate limit is exceeded. - * - * NOTE: This function could be further optimized by eliminating it entirely. - * We already call getCounterState() in underConsumerLimits(), so we could - * cache/reuse that TTL value instead of making another Redis call here. - * - * @param consumerKey The consumer ID or IP address - * @param period The time period - * @return Seconds until counter resets, or 0 if no counter exists - */ - private def ttl(consumerKey: String, period: LimitCallPeriod): Long = { - val state = getCounterState(consumerKey, period) - state.ttl.getOrElse(0L) - } + private def checkAndCount(consumerKey: String, periodsWithLimits: List[(LimitCallPeriod, Long)]): Option[QuotaOutcome] = + if (!useConsumerLimits) None + else try { + Redis.quotaCheckAndCount( + keys = periodsWithLimits.map { case (period, _) => createUniqueKey(consumerKey, period) }, + limits = periodsWithLimits.map(_._2), + periodSeconds = periodsWithLimits.map { case (period, _) => RateLimitingPeriod.toSeconds(period).toInt } + ) match { + case Redis.QuotaRefused(index, _, ttl) => + val (period, limit) = periodsWithLimits(index) + Some(Refused(period, limit, Math.max(0L, ttl))) // TTL -1/-2 (no expiry / no key) reads as 0, as getCounterState does + case Redis.QuotaCounted(countsAndTtls) => + Some(Counted(periodsWithLimits.zip(countsAndTtls).map { case ((period, limit), (count, ttl)) => (period, limit, (ttl, count)) })) + } + } catch { + case e: Throwable => + logger.error(s"Redis unavailable when checking the rate limit for $consumerKey - allowing request", e) + None + } @@ -359,7 +349,10 @@ object RateLimitingUtil extends MdcLoggable { * - Keys format: {consumer_id}_{PERIOD} (e.g., "consumer123_PER_MINUTE") * - Values: current call count within the time window * - TTL: automatically expires keys when time period ends - * - Atomic operations ensure thread-safe counter increments + * - One Lua script per request (Redis.quotaCheckAndCount) checks every period and, when all + * pass, counts the call in every period: one round trip, atomic, so concurrent calls cannot + * pass a check together and overshoot the limit. A refusal comes back with the counter's TTL, + * so the 429 headers need no further read. * * RATE LIMIT HEADERS: * - Sets X-Rate-Limit-Limit: maximum allowed requests for the period @@ -385,132 +378,58 @@ object RateLimitingUtil extends MdcLoggable { if (limit == 0) TooManyRequests + s" Anonymous access is blocked: the anonymous rate limit ${RateLimitingPeriod.humanReadable(period)} is 0." else TooManyRequests + s" We only allow $limit requests ${RateLimitingPeriod.humanReadable(period)} for anonymous access." - // Helper function to set rate limit headers in successful responses - def setXRateLimits(c: CallLimit, z: (Long, Long), period: LimitCallPeriod): Option[CallContext] = { - val limit = period match { - case PER_SECOND => c.per_second - case PER_MINUTE => c.per_minute - case PER_HOUR => c.per_hour - case PER_DAY => c.per_day - case PER_WEEK => c.per_week - case PER_MONTH => c.per_month - case PER_YEAR => -1 - } - userAndCallContext._2.map(_.copy(xRateLimitLimit = limit)) - .map(_.copy(xRateLimitReset = z._1)) - .map(_.copy(xRateLimitRemaining = Math.max(0, limit - z._2))) - } - // Helper function to set rate limit headers for anonymous access - def setXRateLimitsAnonymous(id: String, z: (Long, Long), period: LimitCallPeriod): Option[CallContext] = { - val limit = period match { - case PER_HOUR => perHourLimitAnonymous - case _ => -1 - } + // The X-Rate-Limit-* headers of a served call: the limit, the counter's TTL and what is left of the limit + def setXRateLimits(limit: Long, ttlAndCount: (Long, Long)): Option[CallContext] = userAndCallContext._2.map(_.copy(xRateLimitLimit = limit)) - .map(_.copy(xRateLimitReset = z._1)) - .map(_.copy(xRateLimitRemaining = Math.max(0, limit - z._2))) - } + .map(_.copy(xRateLimitReset = ttlAndCount._1)) + .map(_.copy(xRateLimitRemaining = Math.max(0, limit - ttlAndCount._2))) - // Helper function to create rate limit exceeded response with remaining TTL for authorized users - def exceededRateLimit(c: CallLimit, period: LimitCallPeriod): Option[CallContextLight] = { - val remain = ttl(c.consumer_id, period) - val limit = period match { - case PER_SECOND => c.per_second - case PER_MINUTE => c.per_minute - case PER_HOUR => c.per_hour - case PER_DAY => c.per_day - case PER_WEEK => c.per_week - case PER_MONTH => c.per_month - case PER_YEAR => -1 - } - userAndCallContext._2.map(_.copy(xRateLimitLimit = limit)) - .map(_.copy(xRateLimitReset = remain)) + // The X-Rate-Limit-* headers of a refused call: the exhausted limit, its reset and nothing remaining + def exceededRateLimit(refused: Refused): Option[CallContextLight] = + userAndCallContext._2.map(_.copy(xRateLimitLimit = refused.limit)) + .map(_.copy(xRateLimitReset = refused.resetSeconds)) .map(_.copy(xRateLimitRemaining = 0)).map(_.toLight) - } - // Helper function to create rate limit exceeded response for anonymous users - def exceededRateLimitAnonymous(id: String, period: LimitCallPeriod): Option[CallContextLight] = { - val remain = ttl(id, period) - val limit = period match { - case PER_HOUR => perHourLimitAnonymous - case _ => -1 - } - userAndCallContext._2.map(_.copy(xRateLimitLimit = limit)) - .map(_.copy(xRateLimitReset = remain)) - .map(_.copy(xRateLimitRemaining = 0)).map(_.toLight) - } + def refuse(message: String, refused: Refused): (Box[User], Option[CallContext]) = + (fullBoxOrException(Empty ~> APIFailureNewStyle(message, 429, exceededRateLimit(refused))), userAndCallContext._2) // Main logic: check if we have a CallContext and determine access type userAndCallContext._2 match { case Some(cc) => cc.rateLimiting match { case Some(rl) => // AUTHORIZED ACCESS - consumer has valid credentials and rate limits - // Create rate limiting key for Redis storage using consumer_id - val rateLimitingKey = rl.consumer_id - // Check if current request would exceed any of the 6 rate limits - val checkLimits = List( - underConsumerLimits(rateLimitingKey, PER_SECOND, rl.per_second), - underConsumerLimits(rateLimitingKey, PER_MINUTE, rl.per_minute), - underConsumerLimits(rateLimitingKey, PER_HOUR, rl.per_hour), - underConsumerLimits(rateLimitingKey, PER_DAY, rl.per_day), - underConsumerLimits(rateLimitingKey, PER_WEEK, rl.per_week), - underConsumerLimits(rateLimitingKey, PER_MONTH, rl.per_month) - ) - // Return 429 error for first exceeded limit (shorter periods take precedence) - checkLimits match { - case x1 :: x2 :: x3 :: x4 :: x5 :: x6 :: Nil if x1 == false => - (fullBoxOrException(Empty ~> APIFailureNewStyle(composeMsgAuthorizedAccess(PER_SECOND, rl.per_second, rl.consumer_id), 429, exceededRateLimit(rl, PER_SECOND))), userAndCallContext._2) - case x1 :: x2 :: x3 :: x4 :: x5 :: x6 :: Nil if x2 == false => - (fullBoxOrException(Empty ~> APIFailureNewStyle(composeMsgAuthorizedAccess(PER_MINUTE, rl.per_minute, rl.consumer_id), 429, exceededRateLimit(rl, PER_MINUTE))), userAndCallContext._2) - case x1 :: x2 :: x3 :: x4 :: x5 :: x6 :: Nil if x3 == false => - (fullBoxOrException(Empty ~> APIFailureNewStyle(composeMsgAuthorizedAccess(PER_HOUR, rl.per_hour, rl.consumer_id), 429, exceededRateLimit(rl, PER_HOUR))), userAndCallContext._2) - case x1 :: x2 :: x3 :: x4 :: x5 :: x6 :: Nil if x4 == false => - (fullBoxOrException(Empty ~> APIFailureNewStyle(composeMsgAuthorizedAccess(PER_DAY, rl.per_day, rl.consumer_id), 429, exceededRateLimit(rl, PER_DAY))), userAndCallContext._2) - case x1 :: x2 :: x3 :: x4 :: x5 :: x6 :: Nil if x5 == false => - (fullBoxOrException(Empty ~> APIFailureNewStyle(composeMsgAuthorizedAccess(PER_WEEK, rl.per_week, rl.consumer_id), 429, exceededRateLimit(rl, PER_WEEK))), userAndCallContext._2) - case x1 :: x2 :: x3 :: x4 :: x5 :: x6 :: Nil if x6 == false => - (fullBoxOrException(Empty ~> APIFailureNewStyle(composeMsgAuthorizedAccess(PER_MONTH, rl.per_month, rl.consumer_id), 429, exceededRateLimit(rl, PER_MONTH))), userAndCallContext._2) - case _ => - // All limits passed - count the call in every period, then set the X-Rate-Limit-* - // headers from the shortest period that HAS a limit. The counter alone cannot pick - // that period: every period is counted, unlimited ones included, so a live - // per-second counter says nothing about whether a per-second limit exists. With no - // limited period the CallContext keeps its defaults and the headers read -1. - val counted: List[(LimitCallPeriod, Long, (Long, Long))] = List( - (PER_SECOND, rl.per_second, incrementConsumerCounters(rateLimitingKey, PER_SECOND)), - (PER_MINUTE, rl.per_minute, incrementConsumerCounters(rateLimitingKey, PER_MINUTE)), - (PER_HOUR, rl.per_hour, incrementConsumerCounters(rateLimitingKey, PER_HOUR)), - (PER_DAY, rl.per_day, incrementConsumerCounters(rateLimitingKey, PER_DAY)), - (PER_WEEK, rl.per_week, incrementConsumerCounters(rateLimitingKey, PER_WEEK)), - (PER_MONTH, rl.per_month, incrementConsumerCounters(rateLimitingKey, PER_MONTH)) - ) - counted.collectFirst { case (period, limit, ttlAndCount) if limit > 0 && ttlAndCount._1 > 0 => (period, ttlAndCount) } match { - case Some((period, ttlAndCount)) => (userAndCallContext._1, setXRateLimits(rl, ttlAndCount, period)) - case None => (userAndCallContext._1, userAndCallContext._2) + // One round trip checks the 6 limits, shortest period first, and counts the call when all pass + checkAndCount(rl.consumer_id, consumerPeriods.map(period => (period, limitOf(rl, period)))) match { + case Some(refused: Refused) => + // 429 for the first exceeded limit (shorter periods take precedence) + refuse(composeMsgAuthorizedAccess(refused.period, refused.limit, rl.consumer_id), refused) + case Some(Counted(perPeriod)) => + // All limits passed and every period counted. Set the X-Rate-Limit-* headers from the + // shortest period that HAS a limit. The counter alone cannot pick that period: every + // period is counted, unlimited ones included, so a live per-second counter says nothing + // about whether a per-second limit exists. With no limited period the CallContext keeps + // its defaults and the headers read -1. + perPeriod.collectFirst { case (_, limit, ttlAndCount) if limit > 0 && ttlAndCount._1 > 0 => (limit, ttlAndCount) } match { + case Some((limit, ttlAndCount)) => (userAndCallContext._1, setXRateLimits(limit, ttlAndCount)) + case None => (userAndCallContext._1, userAndCallContext._2) } + case None => // limits off, or Redis unreachable: served, not counted, no headers + (userAndCallContext._1, userAndCallContext._2) } case None => // ANONYMOUS ACCESS - no consumer credentials, use IP-based limiting // Use client IP address as rate limiting key for anonymous access val consumerId = cc.ipAddress // Anonymous access only has per-hour limits to prevent abuse - val checkLimits = List( - underConsumerLimits(consumerId, PER_HOUR, perHourLimitAnonymous) - ) - checkLimits match { - case x1 :: Nil if !x1 => + val limit = perHourLimitAnonymous.toLong + checkAndCount(consumerId, List((PER_HOUR, limit))) match { + case Some(refused: Refused) => // Return 429 error if anonymous hourly limit exceeded - (fullBoxOrException(Empty ~> APIFailureNewStyle(composeMsgAnonymousAccess(PER_HOUR, perHourLimitAnonymous), 429, exceededRateLimitAnonymous(consumerId, PER_HOUR))), userAndCallContext._2) + refuse(composeMsgAnonymousAccess(PER_HOUR, limit), refused) + case Some(Counted((_, _, ttlAndCount) :: _)) if ttlAndCount._1 > 0 => + // Limit not exceeded - counted, set headers + (userAndCallContext._1, setXRateLimits(limit, ttlAndCount)) case _ => - // Limit not exceeded - increment counter and set headers - val incrementCounters = List ( - incrementConsumerCounters(consumerId, PER_HOUR) - ) - incrementCounters match { - case x1 :: Nil if x1._1 > 0 => - (userAndCallContext._1, setXRateLimitsAnonymous(consumerId, x1, PER_HOUR)) - case _ => - (userAndCallContext._1, userAndCallContext._2) - } + (userAndCallContext._1, userAndCallContext._2) } } case _ => // No CallContext available - pass through without rate limiting diff --git a/obp-api/src/test/scala/code/concurrency/CONCURRENCY_HAZARDS.md b/obp-api/src/test/scala/code/concurrency/CONCURRENCY_HAZARDS.md index 7e87bc2f38..dba94336bb 100644 --- a/obp-api/src/test/scala/code/concurrency/CONCURRENCY_HAZARDS.md +++ b/obp-api/src/test/scala/code/concurrency/CONCURRENCY_HAZARDS.md @@ -202,7 +202,7 @@ The remaining entries (Q, T, V, X, Y) are intentionally untested. | Q | `revokeAccess` vs `grant` check-then-act | — | Same `AccountAccess` check-then-act family as **R**; the window is narrow → non-deterministic barrier test would be flaky (false-green). The class is proven by **R**. | | T | `createTransactionRequestBulk` per-leg balance | — | Verdict: unconfirmed intra-request self-race. `saveTransaction` mutates the passed object's `accountBalance` field — sequential legs may see the updated value, not a stale one. Writing a possibly-false test was rejected. | | V | Berlin Group `usesSoFarTodayCounter` lost-increment | — | Same counter lost-update class as H/K; requires fully-signed recurring BG consent + TPP headers — disproportionate setup for a class already proven. | -| X | Consumer rate-limit `underConsumerLimits` TOCTOU | — | Real and high-impact (limit bypass), but active-limit lookup is cached ~1 hour → HTTP-layer timing unreliable → would be flaky. | +| X | Consumer rate-limit `underConsumerLimits` TOCTOU | `ConcurrentConsumerQuotaScriptTest` Q6 | **Fixed.** The GET-then-INCR gate is gone: `RateLimitingUtil.underCallLimits` now checks every period and counts the call in one Lua script (`Redis.quotaCheckAndCount`), so the check and the count cannot interleave. Q6 drives the script directly (20 callers, limit 5, exactly 5 pass), sidestepping the cached active-limit lookup that made an HTTP-layer test flaky. | | Y | `AuthRateLimiter` cold-start SET-vs-INCR collision | — | Same rate-limit class as X; runs in shadow mode by default. Same flakiness concern. | | Z | `MappedAgentProvider.updateAgentStatus` | — | Re-audited as **not a hazard**: sets both fields and `saveMe()`s the whole row — normal last-writer-wins PUT semantics, not field tearing. | diff --git a/obp-api/src/test/scala/code/concurrency/ConcurrentConsumerQuotaScriptTest.scala b/obp-api/src/test/scala/code/concurrency/ConcurrentConsumerQuotaScriptTest.scala new file mode 100644 index 0000000000..e2abe50ec7 --- /dev/null +++ b/obp-api/src/test/scala/code/concurrency/ConcurrentConsumerQuotaScriptTest.scala @@ -0,0 +1,173 @@ +/** +Open Bank Project - API +Copyright (C) 2011-2026, TESOBE GmbH. + +This program is free software: you can redistribute it and/or modify +it under the terms of the GNU Affero General Public License as published by +the Free Software Foundation, either version 3 of the License, or +(at your option) any later version. + +This program is distributed in the hope that it will be useful, +but WITHOUT ANY WARRANTY; without even the implied warranty of +MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +GNU Affero General Public License for more details. + +You should have received a copy of the GNU Affero General Public License +along with this program. If not, see . + +Email: contact@tesobe.com +TESOBE GmbH. +Osloer Strasse 16/17 +Berlin 13359, Germany + +This product includes software developed at +TESOBE (http://www.tesobe.com/) + */ +package code.concurrency + +import code.api.JedisMethod +import code.api.cache.Redis +import code.api.cache.Redis.{QuotaCounted, QuotaDecision, QuotaRefused} +import code.setup.RedisTestTarget + +import java.util.UUID +import java.util.concurrent.atomic.AtomicInteger + +/** + * Redis.quotaCheckAndCount: the one Lua script behind the consumer quota + * (RateLimitingUtil.underCallLimits). It checks every period's limit and, only when all pass, + * counts the call in every period, atomically. These scenarios pin the contract the Scala side + * relies on and close hazard X of CONCURRENCY_HAZARDS.md (the former GET-then-INCR gate). + * + * Needs a reachable Redis, like ConcurrentRateLimiterRaceTest. + */ +class ConcurrentConsumerQuotaScriptTest extends ConcurrentRaceSetup { + + private def redisUp: Boolean = Redis.isRedisReady + + private def freshKeys(n: Int): List[String] = { + val base = "__conc_quota_" + UUID.randomUUID.toString.take(8) + "_" + (1 to n).map(i => base + i).toList + } + + private def count(key: String): Long = Redis.use(JedisMethod.GET, key).map(_.toLong).getOrElse(0L) + private def ttl(key: String): Long = Redis.use(JedisMethod.TTL, key).map(_.toLong).getOrElse(-2L) + + private def call(keys: List[String], limits: List[Long], seconds: List[Int]): QuotaDecision = + Redis.quotaCheckAndCount(keys, limits, seconds) + + feature("Redis.quotaCheckAndCount checks every period and counts the call in one atomic script") { + + scenario("Q1: a call under every limit is counted in every period, unlimited periods included", ConcurrencyRace) { + RedisTestTarget.requireReachable(redisUp, "Q1") + Given("three periods: unlimited, 100 per 60s, 1000 per 3600s, counters absent") + val keys = freshKeys(3) + When("one call is checked and counted") + val decision = call(keys, List(-1L, 100L, 1000L), List(1, 60, 3600)) + Then("it is counted, and every key holds 1 with its period as TTL") + decision match { + case QuotaCounted(countsAndTtls) => + countsAndTtls.map(_._1) should equal(List(1L, 1L, 1L)) + countsAndTtls.map(_._2).zip(List(1, 60, 3600)).foreach { case (t, p) => t should (be > 0L and be <= p.toLong) } + case other => fail(s"expected QuotaCounted, got $other") + } + keys.map(count) should equal(List(1L, 1L, 1L)) + And("a second call makes them 2") + call(keys, List(-1L, 100L, 1000L), List(1, 60, 3600)) match { + case QuotaCounted(countsAndTtls) => countsAndTtls.map(_._1) should equal(List(2L, 2L, 2L)) + case other => fail(s"expected QuotaCounted, got $other") + } + } + + scenario("Q2: the shortest exhausted period refuses, with its count and TTL, and nothing is counted", ConcurrencyRace) { + RedisTestTarget.requireReachable(redisUp, "Q2") + Given("two periods limited to 2 per 60s and 2 per 3600s, each counter already at 2") + val keys = freshKeys(2) + call(keys, List(2L, 2L), List(60, 3600)) + call(keys, List(2L, 2L), List(60, 3600)) + When("a third call is checked") + val decision = call(keys, List(2L, 2L), List(60, 3600)) + Then("the first (shortest) period refuses, reporting a count of 2 and a live TTL") + decision match { + case QuotaRefused(index, c, t) => + index should equal(0) + c should equal(2L) + t should (be > 0L and be <= 60L) + case other => fail(s"expected QuotaRefused, got $other") + } + And("neither counter moved: a refused call costs nothing") + keys.map(count) should equal(List(2L, 2L)) + } + + scenario("Q3: a later period at its limit refuses even though earlier periods have room", ConcurrencyRace) { + RedisTestTarget.requireReachable(redisUp, "Q3") + Given("per-minute 100 and per-hour 1, with one call already counted") + val keys = freshKeys(2) + call(keys, List(100L, 1L), List(60, 3600)) + When("a second call is checked") + val decision = call(keys, List(100L, 1L), List(60, 3600)) + Then("the per-hour period (index 1) refuses and the per-minute counter stays at 1") + decision match { + case QuotaRefused(index, c, _) => + index should equal(1) + c should equal(1L) + case other => fail(s"expected QuotaRefused, got $other") + } + keys.map(count) should equal(List(1L, 1L)) + } + + scenario("Q4: a limit of 0 refuses before reading the value, and reports a TTL of -2 for an absent key", ConcurrencyRace) { + RedisTestTarget.requireReachable(redisUp, "Q4") + Given("an unlimited per-second period and a blocked per-minute period, no counters") + val keys = freshKeys(2) + When("a call is checked") + val decision = call(keys, List(-1L, 0L), List(1, 60)) + Then("the blocked period refuses with count 0 and the TTL Redis reports for a missing key") + decision should equal(QuotaRefused(1, 0L, -2L)) + And("the unlimited period was not counted either") + keys.map(count) should equal(List(0L, 0L)) + } + + scenario("Q5: a counter without an expiry is treated as expired on the check and recreated on the count", ConcurrencyRace) { + RedisTestTarget.requireReachable(redisUp, "Q5") + Given("a counter stuck at 50 with no TTL under a limit of 10") + val keys = freshKeys(1) + Redis.use(JedisMethod.SET, keys.head, value = Some("50")) + ttl(keys.head) should equal(-1L) + When("a call is checked and counted") + val decision = call(keys, List(10L), List(60)) + Then("it is served (the stuck key counts as 0) and the key is recreated at 1 with a TTL") + decision match { + case QuotaCounted((c, t) :: Nil) => + c should equal(1L) + t should (be > 0L and be <= 60L) + case other => fail(s"expected QuotaCounted, got $other") + } + count(keys.head) should equal(1L) + } + + scenario("Q6: at most `limit` of many concurrent callers pass the gate (hazard X closed)", ConcurrencyRace) { + RedisTestTarget.requireReachable(redisUp, "Q6") + Given("a per-minute limit of 5, an unlimited per-hour period, and 20 concurrent callers") + val keys = freshKeys(2) + val limit = 5L + val n = 20 + val passed = new AtomicInteger(0) + When(s"$n threads check-and-count through the script at the same time") + // Before this script the gate was GET (check) then INCR (count) as two round trips, so many + // callers read the same low count, all passed `count + 1 <= limit`, and all incremented. + val results = runConcurrentWithBarrier(n) { _ => + call(keys, List(limit, -1L), List(60, 3600)) match { + case QuotaCounted(_) => passed.incrementAndGet(); true + case QuotaRefused(_, _, _) => false + } + } + Then(s"exactly $limit callers were served and the counters hold exactly $limit") + results.count(_.isFailure) should equal(0) + withClue(s"passed=${passed.get()} limit=$limit: the check and the count run in one atomic script, so no interleaving can let more than `limit` through - ") { + passed.get().toLong should equal(limit) + } + keys.map(count) should equal(List(limit, limit)) + } + } +} diff --git a/obp-api/src/test/scala/code/concurrency/ConcurrentRateLimiterRaceTest.scala b/obp-api/src/test/scala/code/concurrency/ConcurrentRateLimiterRaceTest.scala index f6e58dcc3e..e19076f099 100644 --- a/obp-api/src/test/scala/code/concurrency/ConcurrentRateLimiterRaceTest.scala +++ b/obp-api/src/test/scala/code/concurrency/ConcurrentRateLimiterRaceTest.scala @@ -85,10 +85,11 @@ class ConcurrentRateLimiterRaceTest extends ConcurrentRaceSetup { val passedCount = passed.get() withClue( s"passedCount=$passedCount limit=$limit (results.size=${results.size}): " + - s"RateLimitingUtil checks the counter (GET) and increments it (INCR) as two separate Redis " + - s"round-trips. Under concurrency, many callers read the same low count, all pass `count+1 <= limit`, " + - s"then all increment — far more than `limit` requests slip through (rate-limit bypass). " + - s"Fix: a single atomic Lua `INCR + compare` so the gate and the increment cannot interleave — " + s"the gate and the increment must be one atomic Redis operation. RateLimitingUtil used to check " + + s"the counter (GET) and increment it (INCR) as two separate round-trips, so many callers read the " + + s"same low count, all passed `count+1 <= limit`, then all incremented (rate-limit bypass). " + + s"The consumer quota now runs check and count in one script (Redis.quotaCheckAndCount, see " + + s"ConcurrentConsumerQuotaScriptTest Q6); this primitive must stay atomic too — " ) { passedCount.toLong should be <= limit }