Class: Angarium::Delivery

Inherits:
ApplicationRecord show all
Defined in:
app/models/angarium/delivery.rb

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.reap_stalled(older_than: Angarium.config.delivering_timeout) ⇒ Object

Recover deliveries stranded in "delivering": a worker set the state to "delivering" (in #deliver!) but died (crash, deploy, OOM) before recording the attempt or rescheduling, so the job's pending? guard never re-runs it. Anything still "delivering" whose last attempt started before older_than.ago is presumed abandoned and reset to "pending" + re-enqueued. Returns the number requeued. Keep older_than well above a single attempt's worst-case duration (open_timeout + http_timeout) so a live-but-slow worker isn't reaped; a redelivery is at-least-once-safe regardless.



26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
 
# File 'app/models/angarium/delivery.rb', line 26

def self.reap_stalled(older_than: Angarium.config.delivering_timeout)
  return 0 unless older_than

  ids = where(state: "delivering").where(last_attempt_at: ..older_than.ago).pluck(:id)
  return 0 if ids.empty?

  # Reset each id with a state-scoped compare-and-swap, not a blanket
  # `where(id: ids)`: a delivery can leave `delivering` (succeed, be marked
  # gone, etc.) between the pluck above and this reset. Re-asserting
  # `state: "delivering"` means we never drag a completed delivery back to
  # pending, and — since only the reaper that actually flipped the row
  # enqueues — two reapers racing the same snapshot can't double-enqueue.
  requeued = 0
  ids.each do |id|
    changed = where(id: id, state: "delivering")
      .update_all(state: "pending", next_attempt_at: Time.current, updated_at: Time.current)
    next if changed.zero?

    DeliverJob.perform_later(id)
    requeued += 1
  end
  requeued
end

Instance Method Details

#deliver!(client: Client.new, force: forced) ⇒ Object

Performs one attempt. Records a DeliveryAttempt, then transitions to succeeded, blocked (SSRF), schedules a retry, or exhausts. Returns the attempt. force defaults to the persisted forced flag so a reaped or redelivered forced attempt keeps bypassing the guard; pass it explicitly only to override in tests.



55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
 
# File 'app/models/angarium/delivery.rb', line 55

def deliver!(client: Client.new, force: forced)
  payload = {delivery_id: id, endpoint_id: endpoint_id, event: event.name, force: force}
  ActiveSupport::Notifications.instrument("deliver.angarium", payload) do
    # An endpoint's status can change after a delivery is queued: auto-disable
    # partway through a retry cycle, an operator pause!, or a 410 from a sibling
    # delivery. The dispatch-time `enabled` filter only gates delivery creation,
    # not queued retries, so re-check here before attempting. `force: true`
    # (a manual ping!/redeliver!) overrides the guard for this one attempt, so
    # you can test an endpoint before re-enabling it; any retry it schedules
    # follows the normal status rules again.
    unless force
      if endpoint.paused?
        payload[:outcome] = :held
        return hold_for_pause!
      end
      unless endpoint.enabled?
        payload[:outcome] = :canceled
        return cancel!(reason: endpoint.status)
      end
    end

    # Atomically claim this delivery (pending -> delivering) so only one
    # worker ever attempts it. Without the compare-and-swap, two jobs for the
    # same id (a reaper requeue racing a stale enqueue, an adapter retry, etc.)
    # would both pass the guard above, both POST, and both read the same stale
    # attempt_count. If the CAS changes no row, another worker owns it: bail.
    return nil unless claim_for_attempt!
    payload[:attempt] = attempt_count

    # Re-resolve at delivery time (rather than trusting the save-time check)
    # to catch DNS rebinding: a host that now resolves to a private/disallowed
    # IP is blocked even if it was fine when the endpoint was saved. Any
    # disallowed resolved address is a terminal block.
    addresses = AddressPolicy.resolve(destination_host)

    if addresses.any? { |ip| !AddressPolicy.ip_allowed?(ip, endpoint) }
      payload[:outcome] = :blocked
      payload[:error] = "blocked: destination address not permitted"
      attempt = delivery_attempts.create!(error: payload[:error])
      update!(state: "blocked", next_attempt_at: nil)
      endpoint.record_delivery_failure!
      return attempt
    end

    # Fail closed: our resolver is the single source of truth. If we can't
    # resolve the host, do NOT let HTTPX resolve it unvalidated; record a
    # retryable failure. A transient DNS blip is retried; a persistently
    # unresolvable host eventually exhausts.
    if addresses.empty?
      payload[:outcome] = :unresolvable
      payload[:error] = "unresolvable host: #{destination_host}"
      attempt = delivery_attempts.create!(error: payload[:error])
      handle_failure!
      return attempt
    end

    body = request_body
    ts = Time.now.to_i
    webhook_id = id.to_s
    signature = Signature.sign(payload: body, id: webhook_id, timestamp: ts, secret: endpoint.active_signing_secrets)
    headers = (endpoint.custom_headers || {}).merge(
      "webhook-id" => webhook_id,
      "webhook-timestamp" => ts.to_s,
      "webhook-signature" => signature
    )
    result = client.post(
      endpoint.url,
      body: body,
      headers: headers,
      # Pin the connection to exactly the IP(s) we just validated, so HTTPX
      # can't re-resolve and connect somewhere else after our check (the
      # rebinding window). TLS SNI/cert verification still uses the URL's
      # host. `addresses` is guaranteed non-empty here (the fail-closed
      # branch above returned early otherwise), so the connection always pins.
      addresses: addresses.map(&:to_s)
    )

    attempt = delivery_attempts.create!(
      response_code: result.code,
      response_body: result.body,
      error: result.error,
      duration: result.duration
    )
    payload[:code] = result.code
    payload[:http_duration] = result.duration
    payload[:error] = result.error

    # Status handling follows the Standard Webhooks receiver-etiquette guidance:
    #   2xx           -> success
    #   410 Gone      -> the receiver wants no more webhooks: disable + stop (terminal)
    #   everything else (3xx, 429, 5xx, ...) -> retryable failure. 429/502/504
    #     ("throttle" codes) are retried with backoff and honor Retry-After.
    if result.success?
      payload[:outcome] = :delivered
      succeed!
    elsif result.code == 410
      payload[:outcome] = :gone
      handle_gone!
    else
      payload[:outcome] = :failed
      handle_failure!(retry_after: retry_after_seconds(result.headers))
    end
    attempt
  end
end

#redeliver!(force: false) ⇒ Object

Reset the retry cycle and re-enqueue immediately. Keeps prior DeliveryAttempt history. force: true sends even if the endpoint is no longer enabled (for the re-enqueued attempt); it is persisted so the reaper honors it too. Returns self.



165
166
167
168
169
 
# File 'app/models/angarium/delivery.rb', line 165

def redeliver!(force: false)
  update!(state: "pending", next_attempt_at: nil, attempt_count: 0, forced: force)
  DeliverJob.perform_later(id)
  self
end