Class: Lucerna::Gates::Exposures

Inherits:
Object
  • Object
show all
Defined in:
lib/lucerna/gates/exposures.rb

Overview

Exposure delivery: recorded when the app reads a variant (Client#experiment), deduped, batched, and posted fire-and-forget to POST /sdk/v1/events. Nothing here may ever raise into a read or do network on one — record is a pure queue operation; delivery happens on the poller tick (or close). Failures report through on_error and the batch drops on the floor; the server dedupes idempotently per experiment+iteration+user, so anything we resend is harmless.

Constant Summary collapse

DEDUPE_TTL =

Local dedupe window (seconds) — the server dedupes forever; this just saves traffic.

600.0
FLUSH_AT =

Queue length at which record hints the caller to wake the poller.

20
MAX_REQUEST_BATCH =

The endpoint accepts at most 100 events per request.

100
MAX_SEEN =

Dedupe-map bound: past this, expired entries are pruned on record.

10_000

Instance Method Summary collapse

Constructor Details

#initialize(url:, headers:, transport:, timeout:, on_error:, dedupe_ttl: DEDUPE_TTL, clock: nil) ⇒ Exposures

Returns a new instance of Exposures.



27
28
29
30
31
32
33
34
35
36
37
38
 
# File 'lib/lucerna/gates/exposures.rb', line 27

def initialize(url:, headers:, transport:, timeout:, on_error:, dedupe_ttl: DEDUPE_TTL, clock: nil)
  @url = url
  @headers = headers.merge("Content-Type" => "application/json")
  @transport = transport
  @timeout = timeout
  @on_error = on_error
  @dedupe_ttl = dedupe_ttl
  @clock = clock || -> { Process.clock_gettime(Process::CLOCK_MONOTONIC) }
  @mutex = Mutex.new
  @seen = {} # dedupe key -> monotonic time the entry expires
  @pending = []
end

Instance Method Details

#clear ⇒ Object

Forked children inherit the parent's queue; they must not replay it.



62
63
64
 
# File 'lib/lucerna/gates/exposures.rb', line 62

def clear
  @mutex.synchronize { @pending.clear }
end

#flush ⇒ Object

Poller/close threads only. Sends everything queued in ≤100-event requests; a failed request reports and its batch drops.



68
69
70
71
72
73
74
75
 
# File 'lib/lucerna/gates/exposures.rb', line 68

def flush
  loop do
    batch = @mutex.synchronize { @pending.shift(MAX_REQUEST_BATCH) }
    break if batch.empty?

    deliver(batch)
  end
end

#pending_count ⇒ Object 



57
58
59
 
# File 'lib/lucerna/gates/exposures.rb', line 57

def pending_count
  @mutex.synchronize { @pending.size }
end

#record(event) ⇒ Object

Called from app threads on the read path: sync, never raises, no network. Returns true when the queue has reached FLUSH_AT and the caller should wake the delivery tick.



43
44
45
46
47
48
49
50
51
52
53
54
55
 
# File 'lib/lucerna/gates/exposures.rb', line 43

def record(event)
  now = @clock.call
  key = "#{event["experimentKey"]}\n#{event["iteration"]}\n#{event["userId"]}\n#{event["variantId"]}"
  @mutex.synchronize do
    expiry = @seen[key]
    return false if expiry && expiry > now

    prune(now) if @seen.size >= MAX_SEEN
    @seen[key] = now + @dedupe_ttl
    @pending << event
    @pending.size >= FLUSH_AT
  end
end