Module: Wurk::StreamConcurrencyGuard

Extended by:
ActiveSupport::Concern
Included in:
ApiController
Defined in:
app/controllers/concerns/wurk/stream_concurrency_guard.rb

Overview

Per-process cap on concurrent SSE streams. ActionController::Live pins one Puma thread for every open /api/stream; without a bound, a burst of stale browser tabs could hold every thread and starve the JSON API. Past the cap we 503 with Retry-After — the SPA's EventSource reconnects (and its polling fallback honors Retry-After) once a slot frees. Per-process is the right scope: it's this process's own thread pool we're protecting.

Slots are held as thread references rather than tallied in a counter so the cap can heal itself. A stream whose thread is killed mid-flight never reaches the ensure below (Puma hard-reaps worker threads past force_shutdown_after, and a thread killed inside an uninterruptible read can skip its ensure), which a counter would record as a slot held by nobody — ten of those and /api/stream 503s for the life of the process. A dead holder is instead evicted by the next acquire.

Constant Summary collapse

MAX_CONCURRENT_STREAMS = 
10
RETRY_AFTER_SECONDS = 
3

Class Method Summary collapse

Class Method Details

.acquire ⇒ Object

Reserve a stream slot for the calling thread; false when the cap is already reached by threads that are still alive.



30
31
32
33
34
35
36
37
38
 
# File 'app/controllers/concerns/wurk/stream_concurrency_guard.rb', line 30

def acquire
  @lock.synchronize do
    @holders.keep_if(&:alive?)
    return false if @holders.size >= MAX_CONCURRENT_STREAMS

    @holders << Thread.current
    true
  end
end

.release ⇒ Object

Drops one slot held by the calling thread. Acquire and release always bracket a single block on one thread (#with_stream_slot), so a call from a thread holding nothing is a no-op rather than a slot taken away from whoever is actually streaming.



44
45
46
47
48
49
 
# File 'app/controllers/concerns/wurk/stream_concurrency_guard.rb', line 44

def release
  @lock.synchronize do
    index = @holders.rindex(Thread.current)
    @holders.delete_at(index) if index
  end
end