Class: Marketing::AdSpend::RunState

Inherits:
Object
  • Object
show all
Defined in:
app/services/marketing/ad_spend/run_state.rb

Overview

All of ad-spend syncing's cross-process run state, in one place.

Three distinct facts live here, and they are deliberately together because
every worker in the pipeline reads some combination of them — the backfill
chain, the per-provider jobs, and the batch finalizer. Splitting them left
the day marker with two owners, with the finalizer reaching through
send(:mark_completed) into a worker's private API.

  • completed — a day finished with EVERY provider succeeding. Data
    presence can't encode this: all-zero rows aren't written, so a genuinely
    zero-spend day looks identical to one that never ran, and a
    partially-completed day still wrote rows for whoever succeeded.
  • in flight — a day's batch has been enqueued but hasn't finished. The
    advisory lock cannot cover this: it is released the moment the batch is
    enqueued, long before the providers finish, so without this a second
    chain would find no completion marker and enqueue a duplicate batch.
  • cooldown — a provider is rate-limiting us. Pausing only the job that
    tripped it is not enough; every worker has to stand down or the quota
    that is already exhausted keeps being spent.

Everything is TTL'd, so a worker dying mid-day can't wedge a date forever.

Constant Summary collapse

COMPLETED_TTL =

Long enough to outlive any realistic backfill, short enough to self-clean.

120.days
IN_FLIGHT_TTL =

Longer than the Amazon poller's two-hour terminal age, with headroom for
request creation and the finalizer, but shorter than self-heal's six-hour
sweep cadence. A shorter claim can expire while the owning batch is healthy;
a longer claim can survive the next sweep and delay recovery another cycle.

5.hours
COOLDOWN_WINDOW =

How long to stand down after a provider rate-limits us.

20.minutes
COOLDOWN_KEY =

Redis key for the cohort-wide throttle gate.

'ad_spend_backfill:throttled_until'
LOCK_NAME =

Advisory lock serializing day dispatch. Lives here rather than on either
worker because BOTH dispatchers take it — the backfill chain and the
self-heal sweep — and a lock owned by one of them is a lock the other
couples to.

'ad_spend_backfill'
LOCK_TIMEOUT =

Short: the work under the lock is a Redis read and an enqueue, never the
provider fetch itself.

5

Class Method Summary collapse

Class Method Details

.clear_cooldown! ⇒ void

This method returns an undefined value.



118
119
120
 
# File 'app/services/marketing/ad_spend/run_state.rb', line 118

def clear_cooldown!
  del(COOLDOWN_KEY)
end

.clear_in_flight!(date, scope: nil) ⇒ void

This method returns an undefined value.

Parameters:

  • date (Date)
  • scope (String, nil) (defaults to: nil) 


92
93
94
 
# File 'app/services/marketing/ad_spend/run_state.rb', line 92

def clear_in_flight!(date, scope: nil)
  del(in_flight_key(date, scope))
end

.completed?(date, scope: nil) ⇒ Boolean

Returns whether the day completed with every provider succeeding.

Parameters:

  • date (Date)
  • scope (String, nil) (defaults to: nil) —

    provider scope; nil is the full-provider run

Returns:

  • (Boolean) —

    whether the day completed with every provider succeeding



64
65
66
 
# File 'app/services/marketing/ad_spend/run_state.rb', line 64

def completed?(date, scope: nil)
  exists?(completed_key(date, scope))
end

.cooldown_active? ⇒ Boolean

Returns:

  • (Boolean) 


113
114
115
 
# File 'app/services/marketing/ad_spend/run_state.rb', line 113

def cooldown_active?
  cooldown_remaining.positive?
end

.cooldown_remaining ⇒ Integer

Returns seconds left on the gate, 0 when clear.

Returns:

  • (Integer) —

    seconds left on the gate, 0 when clear



107
108
109
110
 
# File 'app/services/marketing/ad_spend/run_state.rb', line 107

def cooldown_remaining
  until_epoch = Sidekiq.redis { |redis| redis.call('GET', COOLDOWN_KEY) }.to_i
  [until_epoch - Time.current.to_i, 0].max
end

.in_flight?(date, scope: nil) ⇒ Boolean

Parameters:

  • date (Date)
  • scope (String, nil) (defaults to: nil)

Returns:

  • (Boolean) 


78
79
80
 
# File 'app/services/marketing/ad_spend/run_state.rb', line 78

def in_flight?(date, scope: nil)
  exists?(in_flight_key(date, scope))
end

.mark_completed!(date, scope: nil) ⇒ void

This method returns an undefined value.

Parameters:

  • date (Date)
  • scope (String, nil) (defaults to: nil) 


71
72
73
 
# File 'app/services/marketing/ad_spend/run_state.rb', line 71

def mark_completed!(date, scope: nil)
  set(completed_key(date, scope), COMPLETED_TTL)
end

.mark_in_flight!(date, scope: nil) ⇒ void

This method returns an undefined value.

Parameters:

  • date (Date)
  • scope (String, nil) (defaults to: nil) 


85
86
87
 
# File 'app/services/marketing/ad_spend/run_state.rb', line 85

def mark_in_flight!(date, scope: nil)
  set(in_flight_key(date, scope), IN_FLIGHT_TTL)
end

.open_cooldown!(window: COOLDOWN_WINDOW) ⇒ void

This method returns an undefined value.

Opens the cohort-wide throttle gate.

Parameters:

  • window (ActiveSupport::Duration) (defaults to: COOLDOWN_WINDOW) 


100
101
102
103
104
 
# File 'app/services/marketing/ad_spend/run_state.rb', line 100

def open_cooldown!(window: COOLDOWN_WINDOW)
  Sidekiq.redis do |redis|
    redis.call('SET', COOLDOWN_KEY, (Time.current + window).to_i, 'EX', window.to_i)
  end
end

.with_day_lock ⇒ Object

Serializes "is this day taken?" with acting on the answer.

The check is only meaningful while nobody else can act on it, and the
claim it guards has to be written before the lock drops — so whatever
runs in here must call mark_in_flight! synchronously. Enqueueing a job
that will claim the day later leaves the same race the lock is for: the
next dispatcher acquires, sees no claim, and dispatches a duplicate.

Returns:

  • (Object) —

    the block's value, or false when the lock was busy



55
56
57
58
59
 
# File 'app/services/marketing/ad_spend/run_state.rb', line 55

def with_day_lock(&)
  # Any AR class exposes this; the lock's identity is LOCK_NAME, not the
  # class. Source is what the backfill chain has always used.
  Source.with_advisory_lock(LOCK_NAME, timeout_seconds: LOCK_TIMEOUT, &)
end