multistore_cf_workers/
backend.rs

1//! Backend client for the Cloudflare Workers runtime.
2//!
3//! Contains [`WorkerBackend`], which implements [`ProxyBackend`] by forwarding
4//! requests through the Workers Fetch API and reading responses as
5//! `web_sys::Response` streams.
6
7use crate::body::JsBody;
8use crate::fetch_connector::FetchConnector;
9use crate::headers::WsHeaders;
10use crate::response::headermap_from_js;
11use bytes::Bytes;
12use http::HeaderMap;
13use multistore::backend::ForwardResponse;
14use multistore::backend::{build_signer, create_builder, ProxyBackend, RawResponse, StoreBuilder};
15use multistore::error::ProxyError;
16use multistore::route_handler::ForwardRequest;
17use multistore::types::BucketConfig;
18
19use object_store::list::PaginatedListStore;
20use object_store::signer::Signer;
21use object_store::RetryConfig;
22use std::sync::Arc;
23use worker::Fetch;
24
25/// Backend for the Cloudflare Workers runtime.
26///
27/// Uses `FetchConnector` for `object_store` HTTP requests and `web_sys::fetch`
28/// for raw multipart operations.
29#[derive(Clone)]
30pub struct WorkerBackend;
31
32/// The byte length to wrap a streamed PUT body in, or `None` to forward the raw
33/// stream. Returns `None` for aws-chunked bodies (S3 sizes those from
34/// `x-amz-decoded-content-length`) and for bodies with no usable
35/// `Content-Length`.
36fn fixed_body_length(headers: &HeaderMap) -> Option<u64> {
37    let aws_chunked = headers
38        .get(http::header::CONTENT_ENCODING)
39        .and_then(|v| v.to_str().ok())
40        .map(|v| v.contains("aws-chunked"))
41        .unwrap_or(false);
42    if aws_chunked {
43        return None;
44    }
45    headers
46        .get(http::header::CONTENT_LENGTH)
47        .and_then(|v| v.to_str().ok())
48        .and_then(|v| v.parse::<u64>().ok())
49}
50
51/// Build a Cloudflare `FixedLengthStream` of the given length (parts can exceed
52/// `u32::MAX`, so fall back to the BigInt constructor).
53fn new_fixed_length_stream(
54    len: u64,
55) -> Result<worker::worker_sys::FixedLengthStream, wasm_bindgen::JsValue> {
56    if len <= u32::MAX as u64 {
57        worker::worker_sys::FixedLengthStream::new(len as u32)
58    } else {
59        worker::worker_sys::FixedLengthStream::new_big_int(js_sys::BigInt::from(len))
60    }
61}
62
63impl ProxyBackend for WorkerBackend {
64    type ResponseBody = web_sys::Response;
65    type Body = JsBody;
66
67    async fn forward(
68        &self,
69        request: ForwardRequest,
70        body: JsBody,
71    ) -> Result<ForwardResponse<Self::ResponseBody>, ProxyError> {
72        let js_body = body;
73
74        // Build web_sys::RequestInit.
75        let init = web_sys::RequestInit::new();
76        init.set_method(request.method.as_str());
77        init.set_headers(&WsHeaders::from(&request.headers).into_inner().into());
78
79        // Bypass Cloudflare's subrequest cache for reads where leaving it on
80        // would break the request (HEAD rewritten to GET on cacheable-extension
81        // URLs, or Range poisoning the full-object cache). See
82        // `ForwardRequest::should_bypass_cache`.
83        if request.should_bypass_cache() {
84            init.set_cache(web_sys::RequestCache::NoStore);
85        }
86
87        // For PUT: stream the body through without buffering it in WASM memory.
88        // A bare `ReadableStream` body makes the Workers runtime send the
89        // subrequest with
90        // `Transfer-Encoding: chunked` and *drop* `Content-Length` — which S3
91        // rejects for a non-aws-chunked payload (it can't size the object/part),
92        // leaving the subrequest hung until the whole body streams through.
93        // Wrapping the stream in a `FixedLengthStream` makes the runtime emit a
94        // real `Content-Length`. aws-chunked bodies are sized by S3 from
95        // `x-amz-decoded-content-length` and keep their chunk framing, so those
96        // pass through raw.
97        if request.method == http::Method::PUT {
98            if let Some(stream) = js_body.stream() {
99                match fixed_body_length(&request.headers) {
100                    Some(len) => {
101                        let fls = new_fixed_length_stream(len).map_err(|e| {
102                            ProxyError::Internal(format!("FixedLengthStream init failed: {e:?}"))
103                        })?;
104                        let transform: &web_sys::TransformStream = fls.as_ref();
105                        // The outbound fetch consuming `readable` pulls the body
106                        // through the transform; the pipe is driven by that
107                        // backpressure, so it streams rather than buffers. The
108                        // returned promise is intentionally dropped: a pipe
109                        // failure (e.g. the client sending fewer/more bytes than
110                        // Content-Length, which errors the FixedLengthStream)
111                        // also errors `readable`, so the awaited outbound fetch
112                        // fails and the error surfaces there.
113                        let _ = stream.pipe_to(&transform.writable());
114                        init.set_body(&transform.readable());
115                    }
116                    None => init.set_body(stream),
117                }
118            }
119        }
120
121        // Build the outgoing request.
122        let ws_request = web_sys::Request::new_with_str_and_init(request.url.as_str(), &init)
123            .map_err(|e| ProxyError::Internal(format!("failed to create request: {:?}", e)))?;
124
125        // Fetch via the worker crate's Fetch API.
126        let worker_req: worker::Request = ws_request.into();
127        let worker_resp = worker::Fetch::Request(worker_req)
128            .send()
129            .await
130            .map_err(|e| ProxyError::BackendError(format!("fetch failed: {}", e)))?;
131
132        // Convert to web_sys::Response to access the body stream.
133        let backend_ws: web_sys::Response = worker_resp.into();
134        let status = backend_ws.status();
135
136        let headers = headermap_from_js(&backend_ws.headers());
137        let content_length = headers
138            .get(http::header::CONTENT_LENGTH)
139            .and_then(|v| v.to_str().ok())
140            .and_then(|v| v.parse::<u64>().ok());
141
142        Ok(ForwardResponse {
143            status,
144            headers,
145            body: backend_ws,
146            content_length,
147        })
148    }
149
150    fn create_paginated_store(
151        &self,
152        config: &BucketConfig,
153    ) -> Result<Box<dyn PaginatedListStore>, ProxyError> {
154        // Disable retries: object_store's retry logic uses `tokio::time::sleep`
155        // which panics on WASM (`std::time::Instant::now` is unsupported).
156        // See: https://github.com/apache/arrow-rs-object-store/issues/624
157        let no_retry = RetryConfig {
158            max_retries: 0,
159            ..Default::default()
160        };
161        let builder = match create_builder(config)? {
162            StoreBuilder::S3(s) => {
163                StoreBuilder::S3(s.with_http_connector(FetchConnector).with_retry(no_retry))
164            }
165            #[cfg(feature = "azure")]
166            StoreBuilder::Azure(a) => {
167                StoreBuilder::Azure(a.with_http_connector(FetchConnector).with_retry(no_retry))
168            }
169            #[cfg(feature = "gcp")]
170            StoreBuilder::Gcs(g) => {
171                StoreBuilder::Gcs(g.with_http_connector(FetchConnector).with_retry(no_retry))
172            }
173        };
174        builder.build()
175    }
176
177    fn create_signer(&self, config: &BucketConfig) -> Result<Arc<dyn Signer>, ProxyError> {
178        build_signer(config)
179    }
180
181    async fn send_raw(
182        &self,
183        method: http::Method,
184        url: String,
185        headers: HeaderMap,
186        body: Bytes,
187    ) -> Result<RawResponse, ProxyError> {
188        tracing::debug!(
189            method = %method,
190            url = %url,
191            "worker: sending raw backend request via Fetch API"
192        );
193
194        // Build web_sys::RequestInit
195        let init = web_sys::RequestInit::new();
196        init.set_method(method.as_str());
197        init.set_headers(&WsHeaders::from(&headers).into_inner().into());
198
199        // Set body for methods that carry one
200        if !body.is_empty() {
201            let uint8 = js_sys::Uint8Array::from(body.as_ref());
202            init.set_body(&uint8.into());
203        }
204
205        let ws_request = web_sys::Request::new_with_str_and_init(&url, &init)
206            .map_err(|e| ProxyError::BackendError(format!("failed to create request: {:?}", e)))?;
207
208        // Fetch via worker
209        let worker_req: worker::Request = ws_request.into();
210        let worker_resp = Fetch::Request(worker_req)
211            .send()
212            .await
213            .map_err(|e| ProxyError::BackendError(format!("fetch failed: {}", e)))?;
214
215        let status = worker_resp.status_code();
216
217        // Convert to `web_sys::Response` and read the headers BEFORE consuming
218        // the body. The `worker::Response → web_sys::Response` conversion panics
219        // once the body has been read, so reading bytes first (and converting
220        // after) blew up `send_raw` on every multipart/batch-delete response.
221        // `forward()` relies on the same before-body ordering.
222        let ws_response: web_sys::Response = worker_resp.into();
223        let resp_headers = headermap_from_js(&ws_response.headers());
224
225        // Read the (small) response body via `arrayBuffer()`.
226        let buf = wasm_bindgen_futures::JsFuture::from(
227            ws_response
228                .array_buffer()
229                .map_err(|e| ProxyError::Internal(format!("arrayBuffer() failed: {:?}", e)))?,
230        )
231        .await
232        .map_err(|e| ProxyError::Internal(format!("failed to read response: {:?}", e)))?;
233        let resp_bytes = js_sys::Uint8Array::new(&buf).to_vec();
234
235        Ok(RawResponse {
236            status,
237            headers: resp_headers,
238            body: Bytes::from(resp_bytes),
239        })
240    }
241}
multistore_cf_workers/backend.rs

multistore_cf_workers/
backend.rs
