flow/poll

Canonical reference for flow/poll: bounded polling loops for external async work using deterministic orchestration helpers.

Quick Answer

Use flow/poll when you need to repeatedly check external job status. Always bound it with :timeout or :max-attempts, and define success with :return-on.

flow/poll bounds Breyta's status-check loop. It does not cancel the remote
actor, cap the actor's emitted events, or enforce provider billing. Provider
inputs such as maxPosts, limit, or maxItems are request parameters whose
semantics vary by provider; do not treat them as a Breyta-side cost ceiling.
When the remote provider supports cancellation, add a separate cancellation
request for timeout/error cleanup.

When To Use

Prefer :wait when an external actor pushes a callback/signal (human approvals, webhook completions).

Config Shape

flow/poll requires a literal config map.

(flow/poll
  {:interval "10s"
   :timeout "20m" ;; or :max-attempts
   :backoff {:type :exponential :factor 2 :max "2m"}
   :abort-on {:status #{400 401 403} :error? true}
   :return-on (fn [result] (= :completed (:status result)))
   :id :order-status-poll-sleep}
  (flow/step :http :fetch-job-status
    {:connection :jobs-api
     :method :get
     :path (str "/jobs/" job-id)}))

Important fields:

High-Level Implementation Model

flow/poll is expanded before execution into ordinary orchestration primitives:

Practical implication: you get a reusable polling primitive built from the same deterministic constructs you use directly.

Advanced Pattern: Poll + Persist + Compact Output

'(let [input (flow/input)
       started (flow/step :http :start-export
                 {:connection :reports-api
                  :method :post
                  :path "/exports"
                  :json {:account-id (:account-id input)}})
       job-id (:job-id started)
       final-status ^{:label "Wait for export completion"}
                    (flow/poll
                      {:interval "5s"
                       :timeout "15m"
                       :backoff {:type :exponential :factor 1.5 :max "60s"}
                       :abort-on {:status #{400 404 500} :error? true}
                       :return-on (fn [r] (= :completed (:status r)))
                       :id :export-status-sleep}
                      (flow/step :http :get-export-status
                        {:connection :reports-api
                         :method :get
                         :path (str "/exports/" job-id)}))
       artifact (flow/step :function :persist-export-status
                  {:input final-status
                   :code '(fn [input] input)
                   :persist {:type :blob}})]
   {:job-id job-id
    :status (:status final-status)
    :status-ref (:uri artifact)})

This keeps orchestration clear and avoids large inline payloads while polling.

External Actor Cost Guardrails

For a third-party actor or scraper, bound three separate things:

1. the number of queries or partitions sent to the provider;
2. the provider's requested item limit and any remote run timeout; and
3. Breyta's polling attempts and elapsed time.

Start with a small probe and inspect the provider run's actual event/item count
before increasing target counts. A provider item limit can be a page size,
per-query limit, or best-effort hint rather than a hard cap on work performed.

(let [input (flow/input)
      max-query-count 25
      max-post-count 25
      queries (vec (take max-query-count (or (:queries input) [])))
      max-posts (min max-post-count
                     (max 1 (or (:max-posts input) max-post-count)))
      started-response (flow/step :http :start-search-actor
                         {:connection :search-provider
                          :method :post
                          :response-as :json
                          :path "/actor-runs"
                          :json {:queries queries
                                 :maxPosts max-posts}})
      started (:body started-response)
      run-id (:run-id started)
      terminal-result (try
                        (flow/poll
                          {:interval "15s"
                           :timeout "10m"
                           :max-attempts 40
                           :backoff {:type :exponential :factor 1.5 :max "60s"}
                           :abort-on {:status #{400 401 403 404 429 500 502 503}
                                       :error? true}
                           :return-on (fn [result]
                                        (#{"SUCCEEDED" "FAILED" "CANCELLED" "TIMED-OUT"}
                                         (:status result)))
                           :id :search-actor-status-sleep}
                          (let [status-response (flow/step :http :get-search-actor-status
                                                  {:connection :search-provider
                                                   :method :get
                                                   :response-as :json
                                                   :path (str "/actor-runs/" run-id)})]
                            (:body status-response)))
                        (catch Exception poll-error
                          ;; Replace this provider-specific path/method with the
                          ;; cancellation API for the actor service.
                          (flow/step :http :cancel-search-actor
                                     {:connection :search-provider
                                      :method :post
                                      :path (str "/actor-runs/" run-id "/cancel")
                                      :retry {:max-attempts 1}})
                          (throw poll-error)))]
  (if (= "SUCCEEDED" (:status terminal-result))
    terminal-result
    (throw (ex-info "Search actor did not succeed"
                    {:status (:status terminal-result)
                     :run-id run-id
                     :result terminal-result}))))

The :return-on predicate includes every terminal status exposed by the
provider, so the poll stops as soon as the actor finishes. The surrounding
branch must then treat failure statuses such as FAILED, CANCELLED, and
TIMED-OUT as failures (or route them to explicit cleanup); otherwise a
terminal provider failure can be recorded as a successful Breyta run.

The example limits Breyta's wait, not the remote run. flow/poll throws on a
timeout or abort, so cleanup belongs in a try/catch around the poll; code
placed only after the poll would never run on those paths. If the poll throws
while the actor is still active, call the provider's cancellation endpoint
(when available), record the remote run id, and inspect its terminal state
before retrying. The :cancel-search-actor path and method above are
provider-specific placeholders; make that request idempotent and use one
attempt. If no cancellation API exists, use a smaller probe and provider-side
quota/timeout controls; do not assume that a timed-out poll made the remote
work stop.

For high-volume search flows, define target-count as a requested output goal
unless the flow contract guarantees an exact count. Keep these quantities
distinct in the run result:

• requested target count;
• provider queries and per-platform candidate pools;
• candidates evaluated and accepted leads.

Discovery filters, duplicate removal, provider availability, and per-platform
caps can make the accepted count lower than the request. Preserve explicit
user-supplied ICP/role keywords; if the flow infers search terms, report the
effective terms and allow an explicit override so high-signal role terms are
not silently discarded.

Using Implicit Flow Context In Custom Patterns

flow/poll body runs inside your flow orchestration context, so it can safely use:

Patterns that work well:

Validation Rules You Should Expect

See platform caps in Limits And Recovery.

Related

• Orchestration Constructs
• Limits And Recovery
• Patterns (Do/Do Not)
• Waits, Signals, And Timeouts