Module: Pgbus::FairShare

Defined in:
lib/pgbus/fair_share.rb

Overview

Fair share scheduling across tenants (issue #426).

At enqueue time the configured callable (config.fair_share) resolves a key — typically a tenant id — and an optional weight for the job. Both ride inside the job payload hash (same pattern as Concurrency::METADATA_KEY and Batch::METADATA_KEY), so they survive every path that re-sends a payload: blocked-execution promotion, DLQ retry, dashboard retry, bulk enqueue. On the read side Client#read_batch_fair interleaves across keys proportionally to weight (see Client::FairRead).

Events (issue #427) use the same keys and the same read primitive: the configured config.event_fair_share callable receives the Pgbus::Event at publish time and the key rides in the event envelope (a sibling of event_id / payload / published_at — never inside the user payload), so the same message->>'pgbus_fair_key' expression, index and read SQL serve both jobs and events, and the outbox (which stores the envelope) carries it.

Constant Summary collapse

METADATA_KEY = 
"pgbus_fair_key"
WEIGHT_KEY = 
"pgbus_fair_weight"
DEFAULT_WEIGHT = 
1

Class Method Summary collapse

Class Method Details

.enabled?(config = Pgbus.configuration) ⇒ Boolean

Returns:

  • (Boolean) 


26
27
28
 
# File 'lib/pgbus/fair_share.rb', line 26

def enabled?(config = Pgbus.configuration)
  !config.fair_share.nil?
end

.event_enabled?(config = Pgbus.configuration) ⇒ Boolean

Returns:

  • (Boolean) 


30
31
32
 
# File 'lib/pgbus/fair_share.rb', line 30

def event_enabled?(config = Pgbus.configuration)
  !config.event_fair_share.nil?
end

.extract_key(payload) ⇒ Object 



64
65
66
 
# File 'lib/pgbus/fair_share.rb', line 64

def extract_key(payload)
  payload[METADATA_KEY]
end

.extract_weight(payload) ⇒ Object 



68
69
70
 
# File 'lib/pgbus/fair_share.rb', line 68

def extract_weight(payload)
  payload[WEIGHT_KEY] || DEFAULT_WEIGHT
end

.inject_event_metadata(event, event_data, config = Pgbus.configuration) ⇒ Object

Event twin of inject_metadata: returns the event envelope (the hash Publisher.build_event_data produced) with fair-share metadata merged in, or the envelope itself (same object) when event fair share is off or the callable declines to key the event.



48
49
50
51
52
 
# File 'lib/pgbus/fair_share.rb', line 48

def inject_event_metadata(event, event_data, config = Pgbus.configuration)
  return event_data unless event_enabled?(config)

  tag(event_data, resolve_event(event, config))
end

.inject_metadata(active_job, payload_hash, config = Pgbus.configuration) ⇒ Object

Returns the payload with fair-share metadata merged in, or the payload itself (same object) when fair share is off or the job is unkeyed. Exceptions raised by the callable propagate — a key resolver that cannot run is a programmer error, not something to swallow at enqueue.



38
39
40
41
42
 
# File 'lib/pgbus/fair_share.rb', line 38

def inject_metadata(active_job, payload_hash, config = Pgbus.configuration)
  return payload_hash unless enabled?(config)

  tag(payload_hash, resolve(active_job, config))
end

.resolve(active_job, config = Pgbus.configuration) ⇒ Object

[key, weight] for the job, or nil when the callable declines to key it.



55
56
57
 
# File 'lib/pgbus/fair_share.rb', line 55

def resolve(active_job, config = Pgbus.configuration)
  normalize(config.fair_share.call(active_job))
end

.resolve_event(event, config = Pgbus.configuration) ⇒ Object

[key, weight] for the event, or nil when the callable declines.



60
61
62
 
# File 'lib/pgbus/fair_share.rb', line 60

def resolve_event(event, config = Pgbus.configuration)
  normalize(config.event_fair_share.call(event))
end