1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
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
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
use std::sync::Arc;
use std::time::Duration;
use async_trait::async_trait;
use dataflow_rs::engine::functions::AsyncFunctionHandler;
use dataflow_rs::engine::functions::HttpCallConfig;
use dataflow_rs::engine::task_context::TaskContext;
use dataflow_rs::engine::task_outcome::TaskOutcome;
use super::http_common::{self, build_url};
use super::schema::{FieldKind, FieldSchema};
use crate::connector::ConnectorRegistry;
/// This handler's name in metrics, profiles and error messages (F48).
const NAME: &str = "http_call";
/// Executes HTTP requests against named connectors with retry support.
pub struct HttpCallHandler {
pub registry: Arc<ConnectorRegistry>,
pub client: reqwest::Client,
}
#[async_trait]
impl AsyncFunctionHandler for HttpCallHandler {
type Input = HttpCallConfig;
async fn execute(
&self,
ctx: &mut TaskContext<'_>,
input: &HttpCallConfig,
) -> dataflow_rs::Result<TaskOutcome> {
// F40: read the channel before the body borrows `ctx` mutably.
let channel = super::extract_channel(ctx.message()).to_string();
super::connector_helpers::guarded_handler(
NAME,
&self.registry,
&input.connector,
&channel,
async move {
let connector_config =
super::connector_helpers::resolve_connector(&self.registry, &input.connector)
.await?;
let http_config = super::connector_helpers::require_http_connector(
connector_config.as_ref(),
&input.connector,
)?;
// F22e: the connector's method allow-list, empty by default.
// A connector pointed at a read-only upstream can be `["GET"]`
// and no workflow can POST through it. Checked ahead of the
// path logic, which is the only message-dependent step here:
// a refusal that no property of the message can change should
// not be reported after one that can (F58).
let method = super::to_reqwest_method(&input.method);
super::connector_helpers::require_method_allowed(
&http_config.operations,
method.as_str(),
&input.connector,
)?;
// The format axes are values-as-data on dataflow-rs's config;
// this parse is the value table that interprets them. Workflow
// validation checks the same table at authoring time, so this
// refusal only fires for definitions that bypassed it — and,
// like the method check above, it is message-independent, so
// it is reported before anything the message can change (F58).
let body_format = http_common::BodyFormat::parse(input.body_format.as_deref())
.map_err(dataflow_rs::engine::error::DataflowError::Validation)?;
let response_format =
http_common::ResponseFormat::parse(input.response_format.as_deref())
.map_err(dataflow_rs::engine::error::DataflowError::Validation)?;
// `resolve_path` / `resolve_body` are dataflow-rs's own
// sanctioned read of the (static, logic) pairs: they apply the
// static fallback, coerce a non-string path to compact JSON,
// and evaluate on the worker's pooled arena.
let path = input.resolve_path(ctx)?;
let url = build_url(&http_config.url, path.as_deref());
let body = input.resolve_body(ctx)?;
let timeout = Duration::from_millis(input.timeout_ms);
// F8: retrying a non-idempotent method resends the side effect.
// A POST that times out is indistinguishable from one the server
// applied, so re-sending it means double charges and double
// orders — out of the box, since max_retries defaults to 3.
// Idempotent methods retry as before; others need an explicit
// per-connector opt-in (the workflow can carry its own
// idempotency key in headers).
let retryable_method =
input.method.is_idempotent() || http_config.retry_non_idempotent;
let retry_config = &http_config.retry;
let max_retries = if retryable_method {
retry_config.max_retries
} else {
0
};
let policy = super::RetryPolicy {
max_retries,
retry_delay_ms: retry_config.retry_delay_ms,
// F47: the ceiling is the sum of the attempts' own budgets —
// `timeout_ms × (max_retries + 1)`, backoff included, since
// the deadline is measured from the first attempt. On
// shipped defaults (30 s × 4) that is 120 s, so the comment
// that used to sit here — "cannot outlive the channel
// deadline it sits under" — described something this does
// not do. What it does do is make the loop *bounded*:
// attempts plus backoff were previously unbounded, and a
// 60 s channel timeout could sit under a ~127 s retry loop
// still burning a connection after the caller gave up. The
// channel deadline itself is enforced upstream, by the
// timeout `run_for_channel` wraps the whole engine call in.
// Non-retryable methods get `max_retries = 0`, so their
// budget is exactly one `timeout_ms`.
deadline: Some(timeout.saturating_mul(max_retries.saturating_add(1))),
};
// #268: resolve the effective auth once for the whole retry
// loop — static variants pass through; a managed-OAuth2
// connector acquires (or reuses) its access token here.
let auth = crate::connector::oauth::effective_auth(
self.registry.oauth(),
&input.connector,
http_config,
)
.await
.map_err(http_common::oauth_error_to_dataflow)?;
// F6: the breaker is applied by `guarded_handler` above, the
// same shell every other egress path now uses. This branch used
// to carry its own copy — the only one in the codebase.
let result = super::retry_with_policy(policy, "HTTP call", || {
http_common::execute_request(
&self.client,
http_config,
http_common::RequestSpec {
method: &method,
url: &url,
task_headers: Some(&input.headers),
body: body.as_ref(),
body_format,
response_format,
timeout,
auth: auth.as_deref(),
},
)
})
.await;
let response_body = match result {
Ok(body) => body,
Err(e) => {
// #268: a 401 on a managed-OAuth2 connector means the
// cached access token was revoked IdP-side; drop it so
// the next call refetches instead of failing again for
// a full refresh margin.
if matches!(
&e,
dataflow_rs::engine::error::DataflowError::Http { status: 401, .. }
) && matches!(
http_config.auth,
Some(crate::connector::AuthConfig::OAuth2(_))
) {
self.registry.oauth().invalidate(&input.connector).await;
}
return Err(e);
}
};
if let Some(ref response_path) = input.response_path {
ctx.set_json(response_path, &response_body);
}
Ok(TaskOutcome::Success)
},
)
.await
}
}
// -- Input schema (F53) --
//
// The table describing this handler's `function.input` lives next to the
// handler it describes. It used to sit in `schema.rs` with the other nine,
// which is how every schema/handler divergence in the 1.0 audit happened:
// a field was added, renamed or made conditional here and the table saying
// so was in a different file.
pub(super) const HTTP_CALL_FIELDS: &[FieldSchema] = &[
FieldSchema {
name: "connector",
description: "Name of the HTTP connector to call.",
kind: FieldKind::String,
required: true,
resolvable: false,
alias: None,
},
FieldSchema {
name: "method",
description: "HTTP method (GET, POST, PUT, DELETE, PATCH). Defaults to GET.",
kind: FieldKind::String,
required: false,
resolvable: false,
alias: None,
},
FieldSchema {
name: "path",
description: "Static path appended to the connector's base URL.",
kind: FieldKind::String,
required: false,
resolvable: false,
alias: None,
},
FieldSchema {
name: "path_logic",
description: "JSONLogic expression evaluated to derive the request path.",
kind: FieldKind::Any,
required: false,
resolvable: false,
alias: None,
},
FieldSchema {
name: "headers",
description: "Additional request headers.",
kind: FieldKind::Object,
required: false,
resolvable: false,
alias: None,
},
FieldSchema {
name: "body",
description: "Static request body (any JSON value).",
kind: FieldKind::Any,
required: false,
resolvable: false,
alias: None,
},
FieldSchema {
name: "body_logic",
description: "JSONLogic expression evaluated to derive the request body.",
kind: FieldKind::Any,
required: false,
resolvable: false,
alias: None,
},
FieldSchema {
name: "body_format",
description: "How the body becomes request bytes: 'json' (default), 'form' \
(URL-encoded key/value pairs), or 'text' (string sent verbatim). \
Sets the content-type unless a header names one explicitly.",
kind: FieldKind::String,
required: false,
resolvable: false,
alias: None,
},
FieldSchema {
name: "output",
description: "Dotted path where the response body is written. Omit to discard it. (Was `response_path` before 1.0; still accepted, but not alongside `output`.)",
kind: FieldKind::String,
required: false,
resolvable: false,
// A real serde alias on dataflow-rs's `HttpCallConfig` since 3.1 —
// Orion used to rewrite the key in the storage repository instead.
alias: Some("response_path"),
},
FieldSchema {
name: "response_format",
description: "How the response bytes are captured at `output`: 'json' \
(default, parsed) or 'text' (a plain string).",
kind: FieldKind::String,
required: false,
resolvable: false,
alias: None,
},
FieldSchema {
name: "timeout_ms",
description: "Request timeout in milliseconds. Defaults to 30000.",
kind: FieldKind::Number,
required: false,
resolvable: false,
alias: None,
},
];