Class: Featureflip::Events::EventProcessor

Inherits:
Object
  • Object
show all
Defined in:
lib/featureflip/events/event_processor.rb

Constant Summary collapse

DEFAULT_MAX_QUEUE_SIZE =

Upper bound on buffered events.

Only reachable once #flush starts putting batches back faster than they drain -- i.e. a sustained outage of the events endpoint. Past the bound the OLDEST events are shed, which caps memory and keeps the freshest analytics. It also means a long outage sheds the stale re-queued batches rather than starving new events, so the SDK degrades to the old drop-on-failure behaviour instead of hoarding data it cannot send.

10_000

Instance Method Summary collapse

Constructor Details

#initialize(http_client, flush_interval: 30, flush_batch_size: 100, max_queue_size: DEFAULT_MAX_QUEUE_SIZE, logger: nil) ⇒ EventProcessor

Returns a new instance of EventProcessor.



14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
 
# File 'lib/featureflip/events/event_processor.rb', line 14

def initialize(http_client, flush_interval: 30, flush_batch_size: 100,
               max_queue_size: DEFAULT_MAX_QUEUE_SIZE, logger: nil)
  @http_client = http_client
  @flush_interval = flush_interval
  # Clamped to at least 1: #flush drains @flush_batch_size events per pass and stops
  # when the queue is empty, so a non-positive size would shift nothing off a
  # non-empty queue and spin forever. Config#validate! already rejects such values,
  # but this class is constructed directly too.
  @flush_batch_size = flush_batch_size.to_i.positive? ? flush_batch_size.to_i : 1
  @max_queue_size = max_queue_size.to_i.positive? ? max_queue_size.to_i : DEFAULT_MAX_QUEUE_SIZE
  @logger = logger
  @queue = []
  @mutex = Mutex.new
  @stop_flag = false
  @stopped = false
  @thread = nil

  # Monotonic instant before which the batch-size trigger must not start another
  # flush, and a latch held while a size-triggered flush is in flight. Both exist to
  # stop a re-queued batch from turning every subsequent event into another request
  # — see #auto_flush.
  @next_auto_flush_at = 0.0
  @auto_flush_in_flight = false

  # Coalescing state for the drain loop. @auto_flush_in_flight above only ever
  # guarded the SIZE trigger; nothing stopped the background thread's interval
  # tick, an explicit Client#flush and a size-triggered flush from entering the
  # loop together. Two concurrent drains mean two request streams against the
  # endpoint the backoff gate exists to protect — and a success in one clears
  # the gate a failure in the other has just armed, re-opening the
  # one-request-per-event behaviour outright (#2477).
  #
  # Generation counters rather than a bare flag: a waiter has to be able to
  # tell "the drain I was waiting for has finished" from "a later drain is
  # running", or it would sleep through its own completion.
  @drain_in_flight = false
  @drain_started = 0
  @drain_finished = 0
  @drain_done = ConditionVariable.new
end

Instance Method Details

#flush ⇒ Object

Drains the queue a batch at a time, one request per batch.

This used to post the WHOLE queue in a single request, which was harmless while a failure emptied the queue: it never grew far past @flush_batch_size. Re-queuing failures (#2456) is what changed that — after a sustained outage the queue can sit at its 10,000-event bound, and posting all of that at once risks a body the server rejects outright. A 413 is non-retryable, so the entire backlog would be dropped by the very path added to preserve it. At most one drain runs at a time. A caller arriving while one is already going waits for it and returns — it does NOT start its own, and it does NOT return early, because a caller that asked for a flush is asking for its events to be sent. This matches the js/node SDKs, whose flush() has always returned the in-flight promise (#2477).



82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
 
# File 'lib/featureflip/events/event_processor.rb', line 82

def flush
  mine = @mutex.synchronize do
    if @drain_in_flight
      nil
    else
      @drain_in_flight = true
      @drain_started += 1
    end
  end

  if mine.nil?
    @mutex.synchronize do
      waiting_for = @drain_started
      @drain_done.wait(@mutex) while @drain_finished < waiting_for
    end
    return
  end

  begin
    drain
  ensure
    @mutex.synchronize do
      @drain_in_flight = false
      @drain_finished = mine
      @drain_done.broadcast
    end
  end
end

#queue_event(event) ⇒ Object 



55
56
57
58
59
60
61
62
63
64
65
66
67
 
# File 'lib/featureflip/events/event_processor.rb', line 55

def queue_event(event)
  dropped = 0
  @mutex.synchronize do
    # After #stop nothing will flush again, so buffering here would only leak.
    return if @stopped

    @queue << event
    dropped = trim_to_bound
  end

  warn_overflow(dropped)
  auto_flush
end

#start ⇒ Object 



127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
 
# File 'lib/featureflip/events/event_processor.rb', line 127

def start
  @stop_flag = false
  @thread = Thread.new do
    elapsed = 0
    until @stop_flag
      sleep(1)
      elapsed += 1
      next if @stop_flag

      if elapsed >= @flush_interval
        # The interval tick is the retry vehicle for a re-queued batch, so it is
        # deliberately NOT subject to the size trigger's backoff gate.
        elapsed = 0
        flush
      elsif auto_flush
        elapsed = 0
      end
    end
  end
end

#stop ⇒ Object 



148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
 
# File 'lib/featureflip/events/event_processor.rb', line 148

def stop
  @stop_flag = true
  @thread&.wakeup rescue nil
  @thread&.join(5)
  @thread = nil

  # Closed BEFORE the final flush so a failure there is dropped rather than
  # re-queued: nothing will flush again, and retrying until the queue drains would
  # hang shutdown for as long as the endpoint stayed down. One attempt, then let go.
  @mutex.synchronize { @stopped = true }
  # drain, not flush: shutdown must never be the call that gets coalesced
  # away. If the interval tick's drain happens to be in flight, flush would
  # wait for it and return, and anything queued after that loop's last look
  # would be discarded unsent. Two drains overlapping is safe here precisely
  # because @stopped is already set, so neither can re-queue and there is no
  # backoff left to disarm.
  drain
  @mutex.synchronize { @queue.clear }
end