Class: DiscordStore::Transport::Bucket

Inherits:
Object
  • Object
show all
Defined in:
lib/discord_store/transport/bucket.rb

Overview

One Discord rate-limit bucket.

Discord does not publish its per-route limits; it reports them, per response, in the X-RateLimit-* headers, and groups routes into opaque buckets identified by X-RateLimit-Bucket. Several routes can share one bucket, so the limiter learns the mapping at runtime rather than assuming a table of constants that will be wrong by next quarter.

The limit that dominates this library is five messages per five seconds per channel. That is the number that caps a naive Discord-backed store, and the only real answer to it is to write to more than one channel — which is why ChannelShard exists.

Constant Summary collapse

UNKNOWN = 
nil

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(hash_key: nil, clock: -> { Process.clock_gettime(Process::CLOCK_MONOTONIC) }) ⇒ Bucket

Returns a new instance of Bucket.



22
23
24
25
26
27
28
29
30
 
# File 'lib/discord_store/transport/bucket.rb', line 22

def initialize(hash_key: nil, clock: -> { Process.clock_gettime(Process::CLOCK_MONOTONIC) })
  @hash_key = hash_key
  @clock = clock
  @limit = UNKNOWN
  @remaining = UNKNOWN
  @reset_at = nil
  @mutex = Mutex.new
  @condition = ConditionVariable.new
end

Instance Attribute Details

#hash_key ⇒ Object

Returns the value of attribute hash_key.



20
21
22
 
# File 'lib/discord_store/transport/bucket.rb', line 20

def hash_key
  @hash_key
end

Instance Method Details

#acquire(timeout: nil) ⇒ void

This method returns an undefined value.

Blocks until this bucket has room, then claims a slot optimistically. The claim is provisional: #update reconciles it against whatever the server actually reports, and the server's number always wins.

Parameters:

  • timeout (Numeric, nil) (defaults to: nil)

Raises:



39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
 
# File 'lib/discord_store/transport/bucket.rb', line 39

def acquire(timeout: nil)
  deadline = timeout && (@clock.call + timeout)

  @mutex.synchronize do
    loop do
      expire_window

      # Nothing learned yet: let one request through so there is something
      # to learn from.
      if @remaining == UNKNOWN
        @remaining = UNKNOWN
        return
      end

      if @remaining.positive?
        @remaining -= 1
        return
      end

      wait = [@reset_at.to_f - @clock.call, 0.0].max

      if deadline
        remaining_time = deadline - @clock.call
        if remaining_time <= 0
          raise QuotaTimeoutError,
                "waited #{timeout}s for rate-limit bucket #{@hash_key || "(unlearned)"} to reset"
        end

        wait = [wait, remaining_time].min
      end

      # A zero wait with no deadline would spin; give the scheduler a tick.
      @condition.wait(@mutex, wait.positive? ? wait : 0.01)
    end
  end
end

#penalize(seconds) ⇒ void

This method returns an undefined value.

Closes the bucket for seconds after a 429.

Parameters:

  • seconds (Numeric) 


111
112
113
114
115
116
 
# File 'lib/discord_store/transport/bucket.rb', line 111

def penalize(seconds)
  @mutex.synchronize do
    @remaining = 0
    @reset_at = @clock.call + seconds.to_f
  end
end

#state ⇒ Hash

Returns a snapshot, for instrumentation and tests.

Returns:

  • (Hash) —

    a snapshot, for instrumentation and tests



119
120
121
122
123
124
125
126
127
128
 
# File 'lib/discord_store/transport/bucket.rb', line 119

def state
  @mutex.synchronize do
    {
      hash_key: @hash_key,
      limit: @limit,
      remaining: @remaining,
      reset_in: @reset_at ? [@reset_at - @clock.call, 0.0].max : nil
    }
  end
end

#update(limit: nil, remaining: nil, reset_after: nil) ⇒ void

This method returns an undefined value.

Reconciles local state with the headers on a response.

Parameters:

  • limit (Integer, nil) (defaults to: nil)
  • remaining (Integer, nil) (defaults to: nil)
  • reset_after (Float, nil) (defaults to: nil) —

    seconds until the window resets



82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
 
# File 'lib/discord_store/transport/bucket.rb', line 82

def update(limit: nil, remaining: nil, reset_after: nil)
  @mutex.synchronize do
    @limit = Integer(limit) if limit
    @reset_at = @clock.call + Float(reset_after) if reset_after

    if remaining
      reported = Integer(remaining)
      # Requests we have already let through but whose responses have not
      # come back yet are not reflected in the server's count, so take the
      # pessimistic view.
      @remaining = @remaining == UNKNOWN ? reported : [@remaining, reported].min
    end

    @condition.broadcast
  end
end