Merge branch 'nindurkar/rate-limit-ban-action' into 'master'

the same way it terminates `check`: matched, `:allow`, no Redis read, no

`info`.

### Clearing state

`Limiter#clear(identifier)` deletes this limiter's counters for one

identifier, along with any ban a rule carrying `ban_for` has written, and returns the

number of keys removed.

```ruby

RACK_LIMITER.clear({ ip: "1.2.3.4" }) # => 2

```

Counters otherwise only disappear when their window expires; this is the only

way to end one early. It exists for call sites where a later success should

wipe earlier failures, such as an authentication ban cleared by a valid

login.

Every rule in the limiter is cleared, matched or not, because a caller

clearing state after a success knows the identifier rather than which rules

happened to match on the way in. Each rule is deleted in its own call, since

rules do not share a Redis Cluster slot. Clearing an identifier with no state

is a no-op returning `0`. A Redis failure fails open like `check`, leaving the

rest of the state to expire on its own, and the return value counts whatever

was removed before the failure.

## Identifier

`Identifier` is a small value object wrapping a hash of caller attributes.

| `period`          | Window length in seconds. May be a callable resolved on every check.                                                                                                 |

| `action`          | What the rule does when it matches. One of `:limit`, `:log`, `:skip`. Default `:limit`. See [Actions](#actions).                                                     |

| `characteristics` | Array of identifier keys whose values are folded into the Redis counter key. Each unique combination gets its own counter.                                           |

| `count_distinct`  | Optional identifier key. When set, the rule counts distinct values seen for that key within the window, backed by a Redis SET rather than a counter. Must not overlap `characteristics`. |

| `ban_for`         | Optional ban duration, whole seconds, minimum 1. Changes the accounting; `action` still decides who is blocked. Rejected on `:skip`. May be a callable resolved on every check.     |

Making `limit` or `period` callable is the supported pattern for

runtime-tunable thresholds (e.g. feature flags or database-backed settings):

    Match -->|yes| Skip{rule.action<br/>== :skip?}

    Skip -->|"yes (no Redis op)"| SkipEmit[Emit rule_evaluations_total<br/>action=skip, result=skip]

    SkipEmit --> SkipReturn([Return matched=true<br/>action=:allow])

    Skip -->|no| Eval["INCR Redis counter<br/>(see Redis sequence below)"]

    Skip -->|no| Ban{"ban_for rule with<br/>a ban in force?"}

    Ban -->|"yes (no increment)"| Build

    Ban -->|no| Eval["INCR Redis counter<br/>(see Redis sequence below)"]

    Eval --> Build[Build Evaluation<br/>resolve limit/period]

    Build --> Add[Add evaluation to Result]

    Add --> Emit[Emit rule_evaluations_total<br/>+ limit/period gauges]

    Emit --> Act{"result.block?<br/>(:limit rule over limit)"}

    Emit --> Act{"result.block?<br/>(:limit rule over its limit,<br/>or its ban in force)"}

    Act -->|yes| Return([Return Result<br/>action=:block])

    Act -->|"no (:log, or :limit under limit)"| Iter

    Act -->|"no (:log, or under limit)"| Iter

    Iter -->|no more rules| Return2([Return Result reporting<br/>most-constraining evaluation,<br/>or matched=false if none])

    Return --> Check[Emit checks_total<br/>action, matched, error]

    SkipReturn --> Check

The rule's `action` describes what the rule does; the result's `action`

describes the outcome — what the caller should do — and is only ever `:allow`

or `:block`. The Redis counter is always incremented when a rule matches,

except for `:skip` rules, which never touch Redis. The `result` label on

or `:block`. A matching rule increments its counter, with two exceptions:

`:skip` rules never touch Redis, and a rule carrying `ban_for` whose ban is

already in force short-circuits before incrementing. The `result` label on

`rule_evaluations_total` records what each evaluation decided:

| rule action | what it does                                  | exceeded? | result action | `result` label | terminating?  |

| `:log`      | count against the limit (observability only)  | yes       | `:allow`      | `log`          | no — continue |

| `:skip`     | don't count (bypass)                          | n/a       | `:allow`      | `skip`         | yes — stop    |

A rule may also carry `ban_for`, which changes the accounting rather than the

action. On crossing the limit it writes a ban that outlives the counting

window, and it stops counting while that ban holds:

| rule action | with `ban_for`                                | exceeded? | result action | `result` label | terminating?  |

|-------------|-----------------------------------------------|-----------|---------------|----------------|---------------|

| `:limit`    | count, or skip counting while banned          | yes       | `:block`      | `banned`       | yes — stop    |

| `:log`      | same accounting, records what it would do     | yes       | `:allow`      | `banned`       | no — continue |

- `:limit` — when exceeded, `Result#action` is `:block` and evaluation

  terminates. Caller should reject the request (e.g. with HTTP 429). When under

  the limit, evaluation continues to the next rule.

  inert and the result carries no `info` (`to_response_headers` is `{}`).

  The match is still observable via `rule_evaluations_total{action="skip"}`.

  Use this for bypasses.

- `ban_for` — enforcement that outlives the window, on either counting action.

  On crossing the limit the rule writes a second key that keeps blocking for

  `ban_for` seconds, long after the counter window has expired. While that ban

  is in force the rule reports `exceeded? == true` and does not count, so a

  caller cannot extend its own ban, and `reset_at` reports when the ban lifts

  rather than when the window rolls. Use this where repeated failures should

  cost more than a window, such as authentication attempts. State can be

  discarded early with [`#clear`](#clearing-state).

  With `action: :limit` the ban blocks. With `action: :log` the rule does all

  of the same accounting, including writing the ban and suppressing counting

  while it holds, and only skips the blocking, so a shadow measures what

  enforcement would have produced rather than approximating it.

  A `ban_for` shorter than `period` is legal but surprising: the ban can lapse

  while the counter is still over the limit, so `peek` reports not-exceeded

  while the next `check` re-bans immediately. Bans are normally longer than the

  window they are counted in.

  The duration is whole seconds and must be at least 1, since the ban is

  written with `SET EX`. A callable is checked again each time it resolves, and

  a value under a second raises rather than reaching Redis. Being an error, it

  fails open like any other, so a rule whose `ban_for` cannot resolve stops

  blocking entirely: watch `errors_total` after changing one.

  Take care combining `ban_for` with `cost:`. One expensive call can cross the

  limit on its own, and with a ban attached that costs the caller the whole ban

  rather than a single rejection. Checking with `peek` first has the same edge,

  since the call it deliberately lets through is enough to start a ban. Bans

  suit counting discrete failures, such as bad credentials, better than they

  suit variable-cost resource limits.

### Redis keys

Each matched check writes a key shaped:

```

labkit:rl:<limiter_name>:<rule_name>:<char>:<value>[:<char>:<value>...]

labkit:rl:{<limiter_name>:<rule_name>:<char>:<value>[:<char>:<value>...]}

```

The braces are a Redis [hash tag](https://redis.io/docs/latest/operate/oss_and_stack/reference/cluster-spec/#hash-tags):

on a Redis Cluster only the braced part decides the slot, so every key for one

rule and identifier lands on the same shard. `BAN_SCRIPT` touches two keys in

one call, and a cluster rejects a script whose keys span slots.

Characteristic values longer than 200 bytes are replaced with a SHA-256

hexdigest to bound key length. Missing or empty characteristic values are

encoded as `_unknown_`. The TTL is set on the first write of each window and

window starting at the first request, not a sliding window.

A check is a single `EVALSHA` of `INCR_SCRIPT` (or `SADD_SCRIPT` for

`count_distinct` rules). Doing the whole read-modify-write inside Lua means

there is no window between the increment and the `EXPIRE` in which a key

could be left without a TTL.

`count_distinct` rules, or `BAN_SCRIPT` for rules carrying `ban_for`). Doing the whole

read-modify-write inside Lua means there is no window between the increment

and the `EXPIRE` in which a key could be left without a TTL.

A rule carrying `ban_for` writes a second key alongside its counter, sharing the

same hash tag and carrying its own `ban_for` TTL:

```

labkit:rl:{<limiter_name>:<rule_name>:<char>:<value>[:<char>:<value>...]}:ban

```

The suffix sits outside the tag, so it cannot change the slot and cannot

collide with a characteristic value, and both keys are found by the same

`labkit:rl:{<limiter>:<rule>*` glob. The ban key outlives its counter by

design: that is the whole point of `ban_for`.

`BAN_SCRIPT` checks the ban before incrementing, so the ban check,

the increment and the ban write are one atomic operation and a concurrent

check cannot miss the threshold crossing.

```mermaid

sequenceDiagram

| metric                                              | type    | labels                                       | meaning                                                              |

|-----------------------------------------------------|---------|----------------------------------------------|----------------------------------------------------------------------|

| `gitlab_labkit_rate_limiter_checks_total`           | counter | `rate_limiter`, `action`, `matched`, `error` | Exactly one increment per `check` call, including fail-open. `action` is what the caller should do (`"allow"` or `"block"`); `matched` and `error` are `"true"`/`"false"`. |

| `gitlab_labkit_rate_limiter_rule_evaluations_total` | counter | `rate_limiter`, `rule`, `action`, `result`   | One increment per evaluated rule (plus one per matched `:skip` rule). `action` is the configured rule action (`"limit"`, `"log"`, `"skip"`); `result` is what the evaluation decided (`"allow"`, `"block"`, `"log"`, `"skip"` — see the Actions table). |

| `gitlab_labkit_rate_limiter_rule_evaluations_total` | counter | `rate_limiter`, `rule`, `action`, `result`   | One increment per evaluated rule (plus one per matched `:skip` rule). `action` is the configured rule action (`"limit"`, `"log"`, `"skip"`); `result` is what the evaluation decided (`"allow"`, `"block"`, `"log"`, `"skip"`, `"banned"` — see the Actions table). |

| `gitlab_labkit_rate_limiter_calls_total`            | counter | `rate_limiter`, `rule`, `action`             | **Deprecated** — superseded by `checks_total` + `rule_evaluations_total`. Historical per-rule counter: one increment per counted rule (plus one per matched `:skip` rule), `rule="unmatched", action="allow"` when no rule matched. Emitted unchanged during the transition. |

| `gitlab_labkit_rate_limiter_errors_total`           | counter | `rate_limiter`                               | **Deprecated** — use `checks_total{error="true"}`. Fail-open events (any `StandardError` in the labkit path); still the only error metric for `peek`. |

| `gitlab_labkit_rate_limiter_limit`                  | gauge   | `rate_limiter`, `rule`                       | Resolved limit at the last check (useful when `limit:` is callable). |

    # @api private

    class Evaluator

      REDIS_KEY_PREFIX = "labkit:rl"

      BAN_KEY_SUFFIX = ":ban"

      CHAR_VALUE_MAX_LENGTH = 200

      MISSING_VALUE_SENTINEL = "_unknown_"

        return {count, ttl_after}

      LUA

      # Atomic ban check + INCR + conditional ban write, for rules carrying

      # ban_for.

      #

      # An active ban short-circuits before the increment, so a banned caller

      # cannot extend its own window, matching Rack::Attack::Allow2Ban. This

      # applies to :log rules too: a shadow that kept counting through its own

      # ban would report more than enforcement would produce. The

      # counter and the ban must be read and written together or a concurrent

      # check can miss the threshold crossing.

      #

      # Returns {count, counter TTL, ban TTL}, where a ban TTL below zero means

      # not banned.

      BAN_SCRIPT = Labkit::Redis::Script.new(<<~LUA)

        local rule_counter_key = KEYS[1]

        local ban_key = KEYS[2]

        local ttl = ARGV[1]

        local cost = tonumber(ARGV[2])

        local limit = tonumber(ARGV[3])

        local ban_for = tonumber(ARGV[4])

        -- TTL answers both questions at once: >= 0 means the ban key exists

        -- and tells us how long is left, so no separate EXISTS is needed.

        local ban_ttl = redis.call('TTL', ban_key)

        if ban_ttl >= 0 then

          local existing = redis.call('GET', rule_counter_key)

          return {existing or '0', redis.call('TTL', rule_counter_key), ban_ttl}

        end

        local count = redis.call('INCRBYFLOAT', rule_counter_key, cost)

        local ttl_after = redis.call('TTL', rule_counter_key)

        if ttl_after < 0 then

          redis.call('EXPIRE', rule_counter_key, ttl)

          ttl_after = tonumber(ttl)

        end

        if tonumber(count) > limit then

          redis.call('SET', ban_key, '1', 'EX', ban_for)

          return {count, ttl_after, ban_for}

        end

        return {count, ttl_after, -1}

      LUA

      def initialize(name:, rules:, redis:, logger:)

        @name   = name

        @rules  = rules

        Result.error

      end

      # Deletes this limiter's counters, and any bans, for one identifier.

      #

      # Every rule is cleared, matched or not: a caller clearing state after a

      # success knows the identifier, not which rules happened to match on the

      # way in. Rules whose keys do not exist are a no-op.

      #

      # Fails open like {#check}: an unreachable Redis leaves the state to

      # expire on its own rather than raising into the caller.

      def clear(identifier)

        # Counted as we go, so a connection lost mid-loop still reports the keys

        # that did go rather than claiming nothing was cleared.

        removed = 0

        key_groups = @rules.filter_map do |rule|

          next if rule.action == :skip

          [build_redis_key(rule, identifier), build_redis_key(rule, identifier, BAN_KEY_SUFFIX)]

        end

        return 0 if key_groups.empty?

        # One DEL per rule. The hash tag holds the rule name, so a rule's own

        # keys share a slot but two rules do not, and Redis Cluster rejects a

        # DEL spanning slots.

        @redis.with do |conn|

          key_groups.each { |group| removed += conn.del(*group) }

        end

        removed

      rescue StandardError => e

        report_error_metrics

        log_error(e, identifier, nil)

        removed

      end

      private

      # Every rule that matches is evaluated and counted; matching does not

      #   rejected and later rules cannot change that. Rules declared after it

      #   are neither counted nor evaluated, so a blocked request debits every

      #   rule up to and including the one that blocked, and none after it.

      #   A :limit rule carrying ban_for counts as over its limit for as long

      #   as its ban holds, so it terminates for the whole ban. The same rule

      #   with action :log does identical accounting and terminates nothing,

      #   which is what makes it a faithful shadow.

      #

      # Every other matched rule - :log, and :limit while under its limit - is

      # counted and collected into the returned Result, which reports the

      end

      def evaluate_rule(rule, identifier, cost, rule_context)

        return evaluate_ban_rule(rule, identifier, cost, rule_context) if rule.ban_for

        redis_key = build_redis_key(rule, identifier)

        resolved_limit = Integer(resolve_value(rule.limit, rule_context))

        resolved_period = Integer(resolve_value(rule.period, rule_context))

      end

      def peek_rule(rule, identifier, rule_context)

        return peek_ban_rule(rule, identifier, rule_context) if rule.ban_for

        redis_key = build_redis_key(rule, identifier)

        resolved_limit = Integer(resolve_value(rule.limit, rule_context))

        resolved_period = Integer(resolve_value(rule.period, rule_context))

        build_evaluation(rule, resolved_limit, resolved_period, count, ttl)

      end

      def evaluate_ban_rule(rule, identifier, cost, rule_context)

        resolved_limit = Integer(resolve_value(rule.limit, rule_context))

        resolved_period = Integer(resolve_value(rule.period, rule_context))

        resolved_ban_for = resolve_ban_for(rule, rule_context)

        count, ttl, ban_ttl = ban_incr_with_ttl(

          build_redis_key(rule, identifier),

          build_redis_key(rule, identifier, BAN_KEY_SUFFIX),

          resolved_period, cost, resolved_limit, resolved_ban_for

        )

        build_ban_evaluation(rule, resolved_limit, resolved_period, count, ttl, ban_ttl)

      end

      # A callable ban_for is only checked at construction, so it can still hand

      # back something Redis would reject on the SET that writes the ban. Failing

      # here names ban_for rather than surfacing a line number from the script.

      def resolve_ban_for(rule, rule_context)

        raw = resolve_value(rule.ban_for, rule_context)

        seconds = Integer(raw, exception: false)

        raise ArgumentError, "ban_for resolved to #{raw.inspect}, need at least 1 second" if seconds.nil? || seconds < 1

        seconds

      end

      def peek_ban_rule(rule, identifier, rule_context)

        resolved_limit = Integer(resolve_value(rule.limit, rule_context))

        resolved_period = Integer(resolve_value(rule.period, rule_context))

        count, ttl, ban_ttl = read_ban_with_ttl(

          build_redis_key(rule, identifier),

          build_redis_key(rule, identifier, BAN_KEY_SUFFIX)

        )

        build_ban_evaluation(rule, resolved_limit, resolved_period, count, ttl, ban_ttl)

      end

      # A ban blocks whether it was written by this call or an earlier one, so

      # +exceeded+ tracks the ban rather than the count: once the counter window

      # has expired the count can sit below the limit while the ban still holds.

      # reset_at reports when the caller may retry, which is the ban expiry.

      def build_ban_evaluation(rule, resolved_limit, resolved_period, count, ttl, ban_ttl)

        banned = ban_ttl >= 0

        window_remaining = ttl >= 0 ? ttl : resolved_period

        info = Result::Info.new(

          resolved_limit: resolved_limit, resolved_period: resolved_period,

          count: count,

          # A ban outlives its counter, so count reads 0 once the window has

          # gone. Nothing is remaining while the ban still blocks.

          remaining: banned ? 0 : [resolved_limit - count, 0].max,

          reset_at: Time.now.utc + (banned ? ban_ttl : window_remaining)

        )

        Result::Evaluation.new(rule: rule, exceeded: banned, info: info)

      end

      def build_evaluation(rule, resolved_limit, resolved_period, count, ttl)

        info = Result::Info.new(

          resolved_limit: resolved_limit, resolved_period: resolved_period,

        Result::Evaluation.new(rule: rule, exceeded: count > resolved_limit, info: info)

      end

      def build_redis_key(rule, identifier)

        key = "#{REDIS_KEY_PREFIX}:#{@name}:#{rule.name}"

      # The limiter, rule and characteristics sit inside a Redis hash tag, so a

      # counter and its ban always hash to the same cluster slot. BAN_SCRIPT

      # touches both in one call, and Redis Cluster rejects a script whose keys

      # span slots. +suffix+ is appended outside the tag, so it cannot change

      # the slot and cannot collide with a characteristic value.

      def build_redis_key(rule, identifier, suffix = nil)

        key = "#{REDIS_KEY_PREFIX}:{#{@name}:#{rule.name}"

        rule.characteristics.each do |char|

          value = resolve_char_value(char, identifier)

          key += ":#{char}:#{encode_char_value(value)}"

          key << ":#{char}:#{encode_char_value(value)}"

        end

        key << "}"

        key << suffix if suffix

        key

      end

        end

      end

      # Atomic ban check + increment + conditional ban write. See BAN_SCRIPT.

      # The returned ban TTL is negative when no ban is in force.

      def ban_incr_with_ttl(counter_key, ban_key, period, cost, limit, ban_for)

        @redis.with do |conn|

          raw_count, ttl, ban_ttl = BAN_SCRIPT.eval(

            conn, keys: [counter_key, ban_key], argv: [period, cost, limit, ban_for]

          )

          [Float(raw_count), Integer(ttl), Integer(ban_ttl)]

        end

      end

      # Pipelined read of both keys for #peek. No EXPIRE and no SET: peeking a

      # banned identifier must not extend either the window or the ban.

      def read_ban_with_ttl(counter_key, ban_key)

        @redis.with do |conn|

          raw_count, ttl, ban_ttl = conn.pipelined do |pipe|

            pipe.get(counter_key)

            pipe.ttl(counter_key)

            pipe.ttl(ban_key)

          end

          [raw_count.nil? ? 0.0 : Float(raw_count), Integer(ttl), Integer(ban_ttl)]

        end

      end

      # Pipelined GET + TTL. No EXPIRE: peek must not extend the window.

      # A missing key (GET => nil, TTL => -2) is reported as count=0; the

      # build_evaluation fallback then derives reset_at from the rule period

      # An exceeded :log rule reports "log" rather than the "allow" the caller

      # sees, so shadow rules over their limit stay visible.

      # A ban is not a distinct action, so without this a blocking ban would be

      # indistinguishable from an ordinary block. On a ban_for rule exceeded?

      # means the ban is in force, whether this call wrote it or an earlier one

      # did, which is exactly what is worth counting separately.

      def evaluation_result(evaluation)

        return "allow" unless evaluation.exceeded?

        return "banned" if evaluation.rule.ban_for

        evaluation.rule.action == :limit ? "block" : "log"

      end

      #   the rule definition and the call site colocated.

      # @return [Result]

      def check(identifier, cost: 1, rule_context: nil)

        id = identifier.is_a?(Identifier) ? identifier : Identifier.new(identifier)

        id = to_identifier(identifier)

        @evaluator.check(id, cost: cost, rule_context: rule_context)

      end

      # @param rule_context [Hash, nil] see {#check}

      # @return [Result]

      def peek(identifier, rule_context: nil)

        id = identifier.is_a?(Identifier) ? identifier : Identifier.new(identifier)

        id = to_identifier(identifier)

        @evaluator.peek(id, rule_context: rule_context)

      end

      # Discards this limiter's state for one identifier: every rule's counter,

      # and any ban written by a rule carrying ban_for.

      #

      # For call sites where a later success should wipe earlier failures, such

      # as an authentication ban cleared by a valid login. Counters otherwise

      # only expire with their window; this is the only way to end one early.

      #

      # Scoped to this limiter and identifier, not to a single rule: a caller

      # clearing after a success knows who succeeded, not which rules matched.

      #

      # @param identifier [Identifier, Hash] caller attributes

      # @return [Integer] number of Redis keys removed; on a Redis error, those

      #   removed before it failed

      def clear(identifier)

        @evaluator.clear(to_identifier(identifier))

      end

      private

      def to_identifier(identifier)

        identifier.is_a?(Identifier) ? identifier : Identifier.new(identifier)

      end

      def validate_name!(name)

        raise ArgumentError, "name must be a non-empty String" unless name.is_a?(String) && !name.empty?

        return name if NAME_PATTERN.match?(name)

    # exceeded - whether the post-increment count exceeded the resolved limit

    # info     - Result::Info with the per-window counters

    #

    # Evaluations order by constraint: a blocking evaluation (:limit rule over

    # its limit) ranks strictly first - it decides the request no matter what

    # any other rule reports - then exceeded ones, then fewest remaining.

    # Evaluations order by constraint: a blocking evaluation (a :limit rule over

    # its limit, or one whose ban is still in force) ranks strictly first - it

    # decides the request no matter what any other rule reports - then exceeded

    # ones, then fewest remaining.

    # remaining floors at 0, so an exceeded :log rule ties with one sitting

    # exactly on its limit; ranking exceeded ahead keeps a breach from being

    # hidden by a rule that merely reached its limit.