Class: Angarium::Delivery

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

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Instance Attribute Details

#force_send ⇒ Object

Transient: set on a delivery built for a manual, forced send (e.g. endpoint.ping!(force: true)) so its first attempt bypasses the endpoint status guard. Not persisted; only the enqueued job carries it forward.



19
20
21
 
# File 'app/models/angarium/delivery.rb', line 19

def force_send
  @force_send
end

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.



31
32
33
34
35
36
37
38
39
40
 
# File 'app/models/angarium/delivery.rb', line 31

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?

  where(id: ids).update_all(state: "pending", next_attempt_at: Time.current, updated_at: Time.current)
  ids.each { |id| DeliverJob.perform_later(id) }
  ids.size
end

Instance Method Details

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

Performs one attempt. Records a DeliveryAttempt, then transitions to succeeded, blocked (SSRF), schedules a retry, or exhausts. Returns the attempt.



44
45
46
47
48
49
50
51
52
53
54
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
 
# File 'app/models/angarium/delivery.rb', line 44

def deliver!(client: Client.new, force: false)
  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

    update!(state: "delivering", attempt_count: attempt_count + 1, last_attempt_at: Time.current)
    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). Returns self.



148
149
150
151
152
 
# File 'app/models/angarium/delivery.rb', line 148

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