Wait for CLOSED delivery after unblocking overload control

tests/test_stream_service.c

@@ -416,12 +416,22 @@ static void test_pressure(void) {
         test_clock_run_for(ua, 2000);
     }
     info = source_load(); CHECK(info.released_count == 2 && info.released_rate >= info.required_rate && selects_seen);
+    queue_set_callback(queue, callback, callback_arg); queue_resume_callback(queue);
+    deadline = get_time_tb() + 100000;
+    for (;;) {
+        unsigned received = 0;
+        for (unsigned i = 6; i < 8; i++) {
+            struct stream_receive_info state; CHECK(stream_receive_get_info(receives[i], &state) == STREAM_OK);
+            if (state.state == STREAM_RECEIVE_CLOSED) received++;
+        }
+        if (received == 2 || get_time_tb() >= deadline) break;
+        test_clock_poll(ua, 100);
+    }
     for (unsigned i = 0; i < 8; i++) {
         struct stream_receive_info receive; CHECK(stream_receive_get_info(receives[i], &receive) == STREAM_OK);
         CHECK(i >= 6 ? receive.state == STREAM_RECEIVE_CLOSED && receive.reason == STREAM_REASON_OVERLOAD
                      : receive.state == STREAM_RECEIVE_ACTIVE);
     }
-    queue_set_callback(queue, callback, callback_arg); queue_resume_callback(queue);
     test_clock_run_for(ua, 10000); CHECK(!source_load().load.queued.bytes && conn->links_up);
     unsigned delivered = pressure_data;
     CHECK(stream_publish_write(publisher, pressure_payload, sizeof(pressure_payload), &accepted) == STREAM_OK);