feat(rate_limit): add :skip rule action

mutate Redis. It is useful when one code path should account for the request

(`check`) and another should gate a side-effect on whether the caller is

already over-limit. `peek` skips `:log` rules — their state is unobservable

without incrementing.

without incrementing. A matched `:skip` rule terminates `peek` the same way

it terminates `check`: matched, `:allow`, no Redis read, no `info`.

## Identifier

| `match`           | Hash of identifier key/value predicates that must **all** be satisfied for the rule to apply. Empty hash matches anything. See [Matchers](#matchers).                |

| `limit`           | Integer request threshold per `period`. May be a callable resolved on every check.                                                                                   |

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

| `action`          | What the result reports when the limit is exceeded. One of `:block`, `:log`, `:allow`. Default `:block`. See [Actions](#actions).                                    |

| `action`          | What the result reports when the limit is exceeded. One of `:block`, `:log`, `:allow`, `:skip`. Default `:block`. See [Actions](#actions).                           |

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

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

    Start([check identifier]) --> Iter{Next rule?}

    Iter -->|yes| Match{rule.match<br/>all satisfied?}

    Match -->|no| Iter

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

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

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

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

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

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

    Build --> Emit[Emit calls_total + limit/period gauges]

    Emit --> Act{rule.action}

### Actions

The rule's `action` controls how the `Result` reports an over-limit hit. The

counter is always incremented when a rule matches, regardless of `action`:

counter is always incremented when a rule matches, except for `:skip` rules,

which never touch Redis:

- `:block` — when exceeded, `Result#action` is `:block`. Caller should reject

  the request (e.g. with HTTP 429). When under the limit, action is `:allow`.

  `:block`). Useful for "always allow this caller even if they're over the

  limit" cases while still observing them via metrics. Evaluation terminates

  on the first match.

- `:skip` — bypass. A matching rule terminates evaluation with

  `Result#action` `:allow` **without any Redis operation**: nothing is

  counted, so `limit`, `period`, `characteristics`, and `count_distinct` are

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

  The match is still observable via `calls_total{action="skip"}`. Use this

  for bypasses that don't need a counter; use `:allow` only when you want

  the bypassed traffic counted.

### Redis keys

| metric                                          | type    | labels                              | meaning                                                              |

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

| `gitlab_labkit_rate_limiter_calls_total`        | counter | `rate_limiter`, `rule`, `action`    | One increment per terminating decision; also incremented per matched `:log` rule. `action` is one of `"allow"`, `"block"`, `"log"`. `rule="unmatched", action="allow"` when no rule terminated. |

| `gitlab_labkit_rate_limiter_calls_total`        | counter | `rate_limiter`, `rule`, `action`    | One increment per terminating decision; also incremented per matched `:log` rule. `action` is one of `"allow"`, `"block"`, `"log"`, `"skip"`. `rule="unmatched", action="allow"` when no rule terminated. |

| `gitlab_labkit_rate_limiter_errors_total`       | counter | `rate_limiter`                      | Fail-open events (any `StandardError` in the labkit path).            |

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

| `gitlab_labkit_rate_limiter_period_seconds`     | gauge   | `rate_limiter`, `rule`              | Resolved period at the last check.                                   |

      # :log rules are non-terminating: they emit metrics and continue,

      # so a shadow :log rule cannot disable a following :block rule.

      #

      # :skip rules terminate on match without touching Redis: no counter is

      # incremented, so the branch sits before the count_distinct check

      # (identifier completeness is irrelevant to a rule that builds no key).

      # calls_total still increments so the bypass stays observable.

      #

      # SET-mode rules (rule.count_distinct set) that match but whose identifier

      # is missing the count_distinct key fail open + log + bump errors_total, and

      # the loop continues to the next rule (the rule is treated as not applicable

        @rules.each do |rule|

          next unless rule_matches?(rule, identifier)

          if rule.action == :skip

            report_skipped_metrics(rule)

            return skip_result(rule)

          end

          if rule.count_distinct && missing_count_distinct_value?(rule, identifier)

            log_missing_count_distinct(rule, identifier)

            report_error_metrics

          next if rule.action == :log

          next unless rule_matches?(rule, identifier)

          return skip_result(rule) if rule.action == :skip

          return peek_rule(rule, identifier, rule_context)

        end

        rule.match.all? { |key, matcher| matcher.match?(identifier[key]) }

      end

      # A matched :skip rule allows without evaluating: no counter exists, so

      # info is nil (to_response_headers is {} for skip results).

      def skip_result(rule)

        Result.new(matched: true, action: :allow, rule: rule)

      end

      def missing_count_distinct_value?(rule, identifier)

        value = identifier[rule.count_distinct]

        value.nil? || value.to_s.empty?

        )

      end

      # calls_total carries action="skip" (the rule action, not the :allow the

      # caller sees) so bypass traffic stays distinguishable from counted

      # allows. No limit/period gauges: a skip rule has no limit to report.

      def report_skipped_metrics(rule)

        Metrics.calls_total.increment(

          rate_limiter: @name,

          rule: rule.name,

          action: "skip"

        )

      end

      def report_unmatched_metrics

        Metrics.calls_total.increment(

          rate_limiter: @name,

    #             :block = rule matched, exceeded, rule configured to block

    #             :log   = rule matched, exceeded, rule configured to log only

    #             :allow = rule matched but count within limit, rule configured to allow,

    #                      rule configured to skip (bypass, nothing counted),

    #                      no rule matched, or error (fail-open)

    #             The rule's configured action is available via rule.action.

    # rule      - the matched Rule object (nil when matched? is false)

    # error?    - true if Redis was unavailable; result fails open (exceeded? is false)

    # info      - Result::Info with per-window counters; nil when matched? is false or error?

    # info      - Result::Info with per-window counters; nil when matched? is false,

    #             error?, or the matched rule is :skip (no counter exists)

    Result = Data.define(:matched, :exceeded, :action, :rule, :error, :info) do

      def initialize(matched:, action: nil, exceeded: false, rule: nil, error: false, info: nil)

        super

module Labkit

  module RateLimit

    KNOWN_ACTIONS = %i[block log allow].freeze

    KNOWN_ACTIONS = %i[block log allow skip].freeze

    RULE_NAME_PATTERN = /\A[a-z0-9_]+\z/

    RULE_NAME_MAX_LENGTH = 64

    # limit           - request threshold; may be a callable (resolved per check)

    # period          - window in seconds; may be a callable (resolved per check)

    # action          - :block (enforce), :log (count and log only, do not block,

    #                   evaluation continues to subsequent rules), or :allow

    #                   evaluation continues to subsequent rules), :allow

    #                   (count but always permit; terminates evaluation on match

    #                   regardless of whether the limit was exceeded)

    #                   regardless of whether the limit was exceeded), or :skip

    #                   (bypass: permit and terminate evaluation on match without

    #                   counting; performs no Redis operation, so limit, period,

    #                   characteristics, and count_distinct are inert)

    # characteristics - identifier keys used to build the compound Redis counter key

    # count_distinct  - optional Symbol naming an identifier key. When set, the rule

    #                   counts the number of distinct values seen for that key within

    end

  end

  describe "Multi-rule evaluation with :log/:block/:allow", :with_metrics_config do

  describe "Multi-rule evaluation with :log/:block/:allow/:skip", :with_metrics_config do

    let(:metrics) { Labkit::RateLimit::Metrics }

    it "with a single :log rule that exceeds, increments the counter and returns the fall-through Result" do

      expect(result.info.resolved_limit).to eq(5)

      expect(raw_redis.keys("labkit:rl:rack_request:block_rule_l:*")).to be_empty

    end

    it "with a :skip rule whose match: is satisfied, terminates without counting and never evaluates :block" do

      skip_r = make_rule(name: "skip_rule_o", action: :skip, match: { bypass: true }, limit: 5, period: 60)

      block_r = make_rule(name: "block_rule_o", action: :block, limit: 1, characteristics: [:user])

      bypass_id = Labkit::RateLimit::Identifier.new(bypass: true, user: 42)

      result = evaluator(rules: [skip_r, block_r]).check(bypass_id)

      expect(result.matched?).to be(true)

      expect(result.action).to eq(:allow)

      expect(result.rule).to eq(skip_r)

      expect(result.exceeded?).to be(false)

      expect(result.info).to be_nil

      expect(result.to_response_headers).to eq({})

      expect(raw_redis.keys("labkit:rl:*")).to be_empty

    end

    it "with a matched :skip rule, performs no Redis operation at all and reports action=skip" do

      # A verifying double with no stubbed methods: any Redis call raises,

      # which the evaluator rescues into a fail-open matched:false result,

      # failing the matched?/rule expectations below.

      untouchable_redis = instance_double(PooledRedis)

      skip_r = make_rule(name: "skip_rule_p", action: :skip, match: { bypass: true }, limit: 5, period: 60)

      ev = described_class.new(name: "rack_request", rules: [skip_r], redis: untouchable_redis, logger: null_logger)

      result = ev.check(Labkit::RateLimit::Identifier.new(bypass: true))

      expect(result.matched?).to be(true)

      expect(result.rule).to eq(skip_r)

      expect(metrics.calls_total.get(rate_limiter: "rack_request", rule: "skip_rule_p", action: "skip")).to eq(1.0)

    end

    it "with a :skip rule whose match: is NOT satisfied, skips it and evaluates the :block" do

      skip_r = make_rule(name: "skip_rule_q", action: :skip, match: { bypass: true }, limit: 5, period: 60)

      block_r = make_rule(name: "block_rule_q", action: :block, limit: 1, characteristics: [:user])

      result = evaluator(rules: [skip_r, block_r]).check(identifier, cost: 2)

      expect(result.action).to eq(:block)

      expect(result.rule).to eq(block_r)

    end

    it "with a :log rule followed by a same-match :skip rule, counts the :log and terminates on the :skip" do

      log_r = make_rule(name: "log_rule_r", action: :log, match: { path: "/x" }, limit: 1, characteristics: [:user])

      skip_r = make_rule(name: "skip_rule_r", action: :skip, match: { path: "/x" }, limit: 1, period: 60)

      block_r = make_rule(name: "block_rule_r", action: :block, limit: 1, characteristics: [:user])

      id = Labkit::RateLimit::Identifier.new(user: 42, path: "/x")

      result = evaluator(rules: [log_r, skip_r, block_r]).check(id, cost: 2)

      expect(result.action).to eq(:allow)

      expect(result.rule).to eq(skip_r)

      expect(stored_count("labkit:rl:rack_request:log_rule_r:user:42")).to eq(2.0)

      expect(raw_redis.exists?("labkit:rl:rack_request:block_rule_r:user:42")).to be(false)

    end

    it "with a :skip rule carrying count_distinct, terminates even when the identifier lacks the count_distinct key" do

      skip_r = make_rule(name: "skip_rule_t", action: :skip, match: { bypass: true },

        limit: 5, period: 60, count_distinct: :project)

      bypass_id = Labkit::RateLimit::Identifier.new(bypass: true, user: 42)

      result = evaluator(rules: [skip_r]).check(bypass_id)

      expect(result.matched?).to be(true)

      expect(result.rule).to eq(skip_r)

      expect(metrics.errors_total.get(rate_limiter: "rack_request")).to eq(0.0)

    end

    it "peek with a :skip rule whose match: is satisfied, terminates without reading Redis" do

      untouchable_redis = instance_double(PooledRedis)

      skip_r = make_rule(name: "skip_rule_s", action: :skip, match: { bypass: true }, limit: 5, period: 60)

      block_r = make_rule(name: "block_rule_s", action: :block, limit: 1, characteristics: [:user])

      ev = described_class.new(name: "rack_request", rules: [skip_r, block_r], redis: untouchable_redis, logger: null_logger)

      result = ev.peek(Labkit::RateLimit::Identifier.new(bypass: true, user: 42))

      expect(result.matched?).to be(true)

      expect(result.action).to eq(:allow)

      expect(result.rule).to eq(skip_r)

      expect(result.info).to be_nil

    end

  end

  describe "#check with count_distinct (SET-mode)" do