test: remove unnecessary sleeps causing flakiness

Instead we just wait for the last response in several places.

Author: steve-chavez

test/test_http_errors.py

@@ -77,14 +77,24 @@ def test_bad_utils(sess):
 def test_it_keeps_working_after_many_connection_refused(sess):
     """the worker doesn't crash on many failed responses with connection refused"""
 
-    res = sess.execute(text(
+    (request_id,) = sess.execute(text(
         f"""
-        select net.http_get('http://localhost:{wrong_port}') from generate_series(1,10);
+        select net.http_get('http://localhost:{wrong_port}') from generate_series(1,10) offset 9;
     """
-    ))
+    )).fetchone()
     sess.commit()
 
-    time.sleep(2)
+    # Collect the last response, waiting as needed
+    response = sess.execute(
+        text(
+            """
+        select * from net._http_collect_response(:request_id, async:=false);
+    """
+        ),
+        {"request_id": request_id},
+        ).fetchone()
+    assert response is not None
+    assert response[0] == "ERROR"
 
     (error_msg,count) = sess.execute(text(
     """
@@ -121,14 +131,24 @@ def test_it_keeps_working_after_many_connection_refused(sess):
 def test_it_keeps_working_after_server_returns_nothing(sess):
     """the worker doesn't crash on many failed responses with server returned nothing"""
 
-    sess.execute(text(
+    (request_id,) = sess.execute(text(
         """
-        select net.http_get('http://localhost:8080/pathological?disconnect=true') from generate_series(1,10);
+        select net.http_get('http://localhost:8080/pathological?disconnect=true') from generate_series(1,10) offset 9;
     """
-    ))
+    )).fetchone()
     sess.commit()
 
-    time.sleep(1.5)
+    # Collect the last response, waiting as needed
+    response = sess.execute(
+        text(
+            """
+        select * from net._http_collect_response(:request_id, async:=false);
+    """
+        ),
+        {"request_id": request_id},
+        ).fetchone()
+    assert response is not None
+    assert response[0] == "ERROR"
 
     (error_msg,count) = sess.execute(text(
     """
@@ -139,15 +159,25 @@ def test_it_keeps_working_after_server_returns_nothing(sess):
     assert error_msg == "Server returned nothing (no headers, no data)"
     assert count == 10
 
-    sess.execute(text(
+    (request_id,) = sess.execute(text(
         """
-        select net.http_get('http://localhost:8080/pathological?status=200') from generate_series(1,10);
+        select net.http_get('http://localhost:8080/pathological?status=200') from generate_series(1,10) offset 9;
     """
     )).fetchone()
 
     sess.commit()
 
-    time.sleep(1.5)
+    # Collect the last response, waiting as needed
+    response = sess.execute(
+        text(
+            """
+        select * from net._http_collect_response(:request_id, async:=false);
+    """
+        ),
+        {"request_id": request_id},
+        ).fetchone()
+    assert response is not None
+    assert response[0] == "SUCCESS"
 
     (status_code,count) = sess.execute(text(
     """

test/test_http_requests_deleted_after_ttl.py

@@ -6,7 +6,7 @@
 def test_http_responses_deleted_after_ttl(sess, autocommit_sess):
     """Check that http responses will be deleted when they reach their ttl, not immediately but when the worker wakes again"""
 
-    autocommit_sess.execute(text("alter system set pg_net.ttl to '4 seconds'"))
+    autocommit_sess.execute(text("alter system set pg_net.ttl to '1 second'"))
     autocommit_sess.execute(text("select net.worker_restart()"))
     autocommit_sess.execute(text("select net.wait_until_running()"))
 
@@ -34,13 +34,15 @@ def test_http_responses_deleted_after_ttl(sess, autocommit_sess):
     assert response[0] == "SUCCESS"
 
     # Sleep until after request should have been deleted
-    time.sleep(5)
+    time.sleep(1.1)
 
     # Wake the worker manually, under normal operation this will happen when new requests are received
     sess.execute(text("select net.wake()"))
 
     sess.commit() # commit so worker  wakes
 
+    time.sleep(0.1) # wait for deletion
+
     # Ensure the response is now empty
     (count,) = sess.execute(
         text(
@@ -60,16 +62,25 @@ def test_http_responses_deleted_after_ttl(sess, autocommit_sess):
 def test_http_responses_will_complete_deletion(sess, autocommit_sess):
     """Check that http responses will keep being deleted until completion despite no new requests coming"""
 
-    sess.execute(text(
+    (request_id,) = sess.execute(text(
         """
-        select net.http_get('http://localhost:8080/pathological?status=200') from generate_series(1,4);
+        select net.http_get('http://localhost:8080/pathological?status=200') from generate_series(1,4) offset 3;
     """
-    ))
+    )).fetchone()
 
     sess.commit()
 
-    # wait for reqs
-    time.sleep(0.1)
+    # Collect the last response, waiting as needed
+    response = sess.execute(
+        text(
+            """
+        select * from net._http_collect_response(:request_id, async:=false);
+    """
+        ),
+        {"request_id": request_id},
+        ).fetchone()
+    assert response is not None
+    assert response[0] == "SUCCESS"
 
     (count,) = sess.execute(
         text(
@@ -122,16 +133,25 @@ def test_http_responses_will_complete_deletion(sess, autocommit_sess):
 def test_http_responses_will_delete_despite_restart(sess, autocommit_sess):
     """Check that http responses will keep being despite no new requests coming" and despite restart"""
 
-    sess.execute(text(
+    (request_id,) = sess.execute(text(
         """
-        select net.http_get('http://localhost:8080/pathological?status=200') from generate_series(1,4);
+        select net.http_get('http://localhost:8080/pathological?status=200') from generate_series(1,4) offset 3;
     """
-    ))
+    )).fetchone()
 
     sess.commit()
 
-    # wait for reqs
-    time.sleep(0.1)
+    # Collect the last response, waiting as needed
+    response = sess.execute(
+        text(
+            """
+        select * from net._http_collect_response(:request_id, async:=false);
+    """
+        ),
+        {"request_id": request_id},
+        ).fetchone()
+    assert response is not None
+    assert response[0] == "SUCCESS"
 
     (count,) = sess.execute(
         text(

test/test_stat_statements.py

@@ -106,7 +106,6 @@ def test_wakes_at_commit_time(sess):
     # wait for reqs
     time.sleep(2)
 
-
     (commit_calls,) = sess.execute(text(
         """
         select coalesce(sum(calls), 0)
@@ -118,6 +117,7 @@ def test_wakes_at_commit_time(sess):
     )).fetchone()
 
     assert commit_calls == initial_calls + 4 # only 4 queries should be made for the above requests
+                                             # 2 queries at wake, 2 extra to check if there are more rows to be processed
 
     # if the new requests are rollbacked/aborted, then no new queries will be made by the bg worker
     sess.execute(text(