Class: Cosmo::Job::Limit

Inherits:
Object
  • Object
show all
Defined in:
lib/cosmo/job/limit.rb,
sig/cosmo/job/limit.rbs

Overview

Distributed concurrency limiter backed by NATS Key-Value with per-message TTL.

Each unit of concurrency is a numbered KV slot:

"{concurrency_key}/0", "{concurrency_key}/1", ..., "{concurrency_key}/{limit-1}"

Acquiring a slot is a single atomic set (CAS with last-revision=0). Only one worker can win a given slot; losers try the next number. When a job finishes, the slot is erased; if the worker crashes, NATS expires it automatically via the per-message Nats-TTL header. Both paths leave the slot equally empty -- no tombstone, no delete marker.

Constant Summary collapse

BUCKET =

Returns:

  • (::String) 
"cosmo_jobs_limits"

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Limit

Returns a new instance of Limit.



22
23
24
 
# File 'lib/cosmo/job/limit.rb', line 22

def initialize
  @kv = API::KV.new(BUCKET, allow_msg_ttl: true)
end

Class Method Details

.instance ⇒ Limit

Returns:



18
19
20
 
# File 'lib/cosmo/job/limit.rb', line 18

def self.instance
  @instance ||= new
end

Instance Method Details

#acquire(key, jid:, limit:, duration:) ⇒ String?

Try to acquire one of the numbered slots for key.

Parameters:

  • key (String) —

    concurrency key

  • jid (String) —

    stored as the slot value for observability

  • limit (Integer) —

    number of slots (0 … limit-1)

  • duration (Integer) —

    seconds before the slot is auto-expired by NATS

  • jid: (::String)
  • limit: (::Integer)
  • duration: (::Integer)

Returns:

  • (String, nil) —

    the acquired slot key, or nil when all slots are taken



33
34
35
36
37
38
39
40
41
42
 
# File 'lib/cosmo/job/limit.rb', line 33

def acquire(key, jid:, limit:, duration:)
  0.upto(limit - 1) do |i|
    slot = "#{key}/#{i}"
    @kv.set(slot, jid, ttl: duration)
    return slot
  rescue NATS::KeyValue::KeyWrongLastSequenceError
    next # slot is live, try the next one
  end
  nil # all slots occupied
end

#release(slot) ⇒ void

This method returns an undefined value.

Release a previously acquired slot. Erases the slot entirely (no tombstone left behind) so a released slot looks identical to one reclaimed by Nats-TTL expiry -- callers never have to special-case a delete marker.

Parameters:

  • slot (::String) 


48
49
50
51
52
 
# File 'lib/cosmo/job/limit.rb', line 48

def release(slot)
  @kv.erase(slot)
rescue NATS::Error
  # best effort — slot TTL will reclaim it if erase fails
end