Skip to content

Commit 5657032

Browse files
nyannyachanyannyacha
authored
fix(sb_workers): don't propagate promise rejection on request body streaming failure (#374)
* fix(sb_workers): don't propagate promise rejection on request body streaming failure * stamp: oops * stamp: add an integration test
1 parent 853c9d8 commit 5657032

File tree

2 files changed

+89
-60
lines changed

2 files changed

+89
-60
lines changed

‎crates/base/tests/integration_tests.rs‎

Lines changed: 49 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -714,6 +714,7 @@ async fn test_file_upload_real_multipart_bytes() {
714714
test_oak_file_upload(
715715
Cow::Borrowed("./test_cases/main"),
716716
(9.98 * MB as f32) as usize, // < 10MB (in binary)
717+
None,
717718
|resp| async {
718719
let res = resp.unwrap();
719720

@@ -731,16 +732,21 @@ async fn test_file_upload_real_multipart_bytes() {
731732
#[tokio::test]
732733
#[serial]
733734
async fn test_file_upload_size_exceed() {
734-
test_oak_file_upload(Cow::Borrowed("./test_cases/main"), 10 * MB, |resp| async {
735-
let res = resp.unwrap();
735+
test_oak_file_upload(
736+
Cow::Borrowed("./test_cases/main"),
737+
10 * MB,
738+
None,
739+
|resp| async {
740+
let res = resp.unwrap();
736741

737-
assert_eq!(res.status().as_u16(), 500);
742+
assert_eq!(res.status().as_u16(), 500);
738743

739-
let res = res.text().await;
744+
let res = res.text().await;
740745

741-
assert!(res.is_ok());
742-
assert_eq!(res.unwrap(), "Error!");
743-
})
746+
assert!(res.is_ok());
747+
assert_eq!(res.unwrap(), "Error!");
748+
},
749+
)
744750
.await;
745751
}
746752

@@ -1166,6 +1172,7 @@ async fn req_failure_case_op_cancel_from_server_due_to_cpu_resource_limit() {
11661172
test_oak_file_upload(
11671173
Cow::Borrowed("./test_cases/main_small_cpu_time"),
11681174
48 * MB,
1175+
None,
11691176
|resp| async {
11701177
let res = resp.unwrap();
11711178

@@ -1183,8 +1190,40 @@ async fn req_failure_case_op_cancel_from_server_due_to_cpu_resource_limit() {
11831190
.await;
11841191
}
11851192

1186-
async fn test_oak_file_upload<F, R>(main_service: Cow<'static, str>, bytes: usize, resp_callback: F)
1187-
where
1193+
#[tokio::test]
1194+
#[serial]
1195+
async fn req_failure_case_op_cancel_from_server_due_to_cpu_resource_limit_2() {
1196+
test_oak_file_upload(
1197+
Cow::Borrowed("./test_cases/main_small_cpu_time"),
1198+
1024 * 64,
1199+
Some("image/png"),
1200+
|resp| async {
1201+
let res = resp.unwrap();
1202+
1203+
assert_eq!(res.status().as_u16(), 500);
1204+
1205+
let res = res.json::<ErrorResponsePayload>().await;
1206+
1207+
assert!(res.is_ok());
1208+
1209+
let msg = res.unwrap().msg;
1210+
1211+
assert!(!msg.starts_with("TypeError: request body receiver not connected"));
1212+
assert_eq!(
1213+
msg,
1214+
"WorkerRequestCancelled: request has been cancelled by supervisor"
1215+
);
1216+
},
1217+
)
1218+
.await;
1219+
}
1220+
1221+
async fn test_oak_file_upload<F, R>(
1222+
main_service: Cow<'static, str>,
1223+
bytes: usize,
1224+
mime: Option<&str>,
1225+
resp_callback: F,
1226+
) where
11881227
F: FnOnce(Result<Response, reqwest::Error>) -> R,
11891228
R: Future<Output = ()>,
11901229
{
@@ -1199,7 +1238,7 @@ where
11991238
"meow",
12001239
Part::bytes(vec![0u8; bytes])
12011240
.file_name("meow.bin")
1202-
.mime_str("application/octet-stream")
1241+
.mime_str(mime.unwrap_or("application/octet-stream"))
12031242
.unwrap(),
12041243
),
12051244
)

‎crates/sb_workers/user_workers.js‎

Lines changed: 40 additions & 50 deletions
Original file line numberDiff line numberDiff line change
@@ -1,16 +1,11 @@
11
import { primordials, core } from "ext:core/mod.js";
2-
import { readableStreamForRid, writableStreamForRid } from 'ext:deno_web/06_streams.js';
3-
import { getSupabaseTag } from 'ext:sb_core_main_js/js/http.js';
2+
import { readableStreamForRid, writableStreamForRid } from "ext:deno_web/06_streams.js";
3+
import { getSupabaseTag } from "ext:sb_core_main_js/js/http.js";
44

55
const ops = core.ops;
6-
const {
7-
InterruptedPrototype,
8-
} = core;
9-
const {
10-
TypeError,
11-
ObjectPrototypeIsPrototypeOf,
12-
StringPrototypeIncludes,
13-
} = primordials;
6+
7+
const { TypeError } = primordials;
8+
149
const {
1510
op_user_worker_fetch_send,
1611
op_user_worker_create,
@@ -33,11 +28,11 @@ class UserWorker {
3328
this.key = key;
3429
}
3530

36-
async fetch(req, opts = {}) {
37-
const tag = getSupabaseTag(req);
31+
async fetch(request, options = {}) {
32+
const tag = getSupabaseTag(request);
3833

39-
const { method, url, headers, body, bodyUsed } = req;
40-
const { signal } = opts;
34+
const { method, url, headers, body, bodyUsed } = request;
35+
const { signal } = options;
4136

4237
signal?.throwIfAborted();
4338

@@ -60,62 +55,56 @@ class UserWorker {
6055
);
6156

6257
// stream the request body
63-
let reqBodyPromise = null;
58+
let requestBodyPromise = null;
59+
6460
if (hasBody) {
6561
let writableStream = writableStreamForRid(requestBodyRid);
66-
reqBodyPromise = body.pipeTo(writableStream, { signal });
62+
requestBodyPromise = body.pipeTo(writableStream, { signal });
6763
}
6864

69-
const resPromise = op_user_worker_fetch_send(
65+
const responsePromise = op_user_worker_fetch_send(
7066
this.key,
7167
requestRid,
7268
requestBodyRid,
7369
tag.streamRid,
7470
tag.watcherRid
7571
);
7672

77-
let [sent, res] = await Promise.allSettled([reqBodyPromise, resPromise]);
78-
79-
if (sent.status === "rejected") {
80-
if (res.status === "fulfilled") {
81-
res = res.value;
82-
} else {
83-
if (
84-
ObjectPrototypeIsPrototypeOf(InterruptedPrototype, sent.reason) ||
85-
StringPrototypeIncludes(sent.reason.message, "operation canceled")
86-
) {
87-
throw res.reason;
88-
} else {
89-
throw sent.reason;
90-
}
91-
}
92-
} else if (res.status === "rejected") {
93-
throw res.reason;
94-
} else {
95-
res = res.value;
73+
const [requestBodyPromiseResult, responsePromiseResult] = await Promise.allSettled([
74+
requestBodyPromise,
75+
responsePromise
76+
]);
77+
78+
if (requestBodyPromiseResult.status === "rejected") {
79+
// console.warn(requestBodyPromiseResult.reason);
9680
}
9781

82+
if (responsePromiseResult.status === "rejected") {
83+
throw responsePromiseResult.reason;
84+
}
85+
86+
const result = responsePromiseResult.value;
9887
const response = {
99-
headers: res.headers,
100-
status: res.status,
101-
statusText: res.statusText,
88+
headers: result.headers,
89+
status: result.status,
90+
statusText: result.statusText,
10291
body: null,
10392
};
10493

10594
// TODO: add a test
106-
if (nullBodyStatus(res.status) || redirectStatus(res.status)) {
107-
core.close(res.bodyRid);
95+
if (nullBodyStatus(result.status) || redirectStatus(result.status)) {
96+
core.tryClose(result.bodyRid);
10897
} else {
109-
if (req.method === 'HEAD' || req.method === 'CONNECT') {
110-
response.body = null;
111-
core.close(res.bodyRid);
98+
if (request.method === "HEAD" || request.method === "CONNECT") {
99+
core.tryClose(result.bodyRid);
112100
} else {
113-
const bodyStream = readableStreamForRid(res.bodyRid);
101+
const stream = readableStreamForRid(result.bodyRid);
114102

115-
signal?.addEventListener('abort', () => {
116-
core.tryClose(res.bodyRid);
103+
signal?.addEventListener("abort", () => {
104+
core.tryClose(result.bodyRid);
117105
});
118-
response.body = bodyStream;
106+
107+
response.body = stream;
119108
}
120109
}
121110

@@ -148,8 +137,8 @@ class UserWorker {
148137

149138
const { servicePath, maybeEszip } = readyOptions;
150139

151-
if (!maybeEszip && (!servicePath || servicePath === '')) {
152-
throw new TypeError('service path must be defined');
140+
if (!maybeEszip && (!servicePath || servicePath === "")) {
141+
throw new TypeError("service path must be defined");
153142
}
154143

155144
const key = await op_user_worker_create(readyOptions);
@@ -159,4 +148,5 @@ class UserWorker {
159148
}
160149

161150
const SUPABASE_USER_WORKERS = UserWorker;
151+
162152
export { SUPABASE_USER_WORKERS };

0 commit comments

Comments
 (0)