🗳️ Add synchronization logic for upstream-async tests (#252)

* 🗳️ Add synchronization logic for upstream-async tests

This allows us to observe that the implementation of `poll()` doesn't block, unlike its counterpart
`wait()`.

* explicit rustup update step?

* that wasn't the ticket

.github/workflows/test.yml

@@ -17,7 +17,7 @@ jobs:
       with:
         submodules: true
     - name: Install Rust
-      run: rustup update stable && rustup default stable
+      run: rustup update --no-self-update stable && rustup default stable
       shell: bash
     - name: Add wasm32-wasi Rust target
       run: rustup target add wasm32-wasi
@@ -56,7 +56,7 @@ jobs:
       with:
         submodules: true
     - name: Install Rust
-      run: rustup update stable && rustup default stable
+      run: rustup update --no-self-update stable && rustup default stable
       shell: bash
     - name: Add wasm32-wasi Rust target
       run: rustup target add wasm32-wasi

cli/tests/integration/upstream_async.rs

@@ -1,28 +1,45 @@
+use std::sync::Arc;
+
+use tokio::sync::Semaphore;
+
 use {
     crate::common::{Test, TestResult},
     hyper::{Response, StatusCode},
 };
 
 #[tokio::test(flavor = "multi_thread")]
 async fn upstream_async_methods() -> TestResult {
-    // Set up the test harness
+    // Set up two backends that share a semaphore that starts with zero permits. `backend1` must
+    // take a semaphore permit and then "forget" it before returning its response. `backend2` adds a
+    // permit to the semaphore and promptly returns. This relationship allows the test fixtures to
+    // examine the behavior of the various pending request operators beyond just whether they
+    // eventually return the expected response.
+    let sema_backend1 = Arc::new(Semaphore::new(0));
+    let sema_backend2 = sema_backend1.clone();
     let test = Test::using_fixture("upstream-async.wasm")
-        // Set up the backends, which just return responses with an identifying header
         .backend("backend1", "http://127.0.0.1:9000/", None)
-        .host(9000, |_| {
-            Response::builder()
-                .header("Backend-1-Response", "")
-                .status(StatusCode::OK)
-                .body(vec![])
-                .unwrap()
+        .async_host(9000, move |_| {
+            let sema_backend1 = sema_backend1.clone();
+            Box::new(async move {
+                sema_backend1.acquire().await.unwrap().forget();
+                Response::builder()
+                    .header("Backend-1-Response", "")
+                    .status(StatusCode::OK)
+                    .body(hyper::Body::empty())
+                    .unwrap()
+            })
         })
         .backend("backend2", "http://127.0.0.1:9001/", None)
-        .host(9001, |_| {
-            Response::builder()
-                .header("Backend-2-Response", "")
-                .status(StatusCode::OK)
-                .body(vec![])
-                .unwrap()
+        .async_host(9001, move |_| {
+            let sema_backend2 = sema_backend2.clone();
+            Box::new(async move {
+                sema_backend2.add_permits(1);
+                Response::builder()
+                    .header("Backend-2-Response", "")
+                    .status(StatusCode::OK)
+                    .body(hyper::Body::empty())
+                    .unwrap()
+            })
         });
 
     // The meat of the test is on the guest side; we just check that we made it through successfully

test-fixtures/src/bin/upstream-async.rs

@@ -68,10 +68,23 @@ fn test_wait() {
     tracker.assert_complete();
 }
 
-/// Run the test using the `poll` API, polling both responses in a loop and processing them as they become ready.
 fn test_poll() {
+    let req1 = Request::get("http://www.example1.com/")
+        .send_async("backend1")
+        .unwrap();
+
+    // req1 should not be ready until a request is sent to backend2
+    let PollResult::Pending(req1) = req1.poll() else {
+        panic!("req1 finished too soon")
+    };
+
+    // sending req2 should unblock req1, and req2 itself should return immediately.
+    let req2 = Request::get("http://www.example2.com/")
+        .send_async("backend2")
+        .unwrap();
+
+    // avoid races by resolving the responses to both requests in a loop
     let mut tracker = ResponseTracker::new();
-    let (req1, req2) = send_async_reqs();
     let mut reqs = vec![req1, req2];
 
     while !reqs.is_empty() {