From caa3bd5ee3f056657e7a964e93a07374a3654c5f Mon Sep 17 00:00:00 2001
From: Raminder Singh <romi_ssk@yahoo.co.in>
Date: Wed, 29 Jul 2026 20:32:22 +0530
Subject: [PATCH 1/3] tests: make tests less flaky and more deterministic

Instead of using sleeps in tests which assumed that certain operations will
complete within those timeouts, we now poll for events to ensure that tests pass
deterministically everywhere: locally and in ci. Before this change the tests
would pass locally but will fail in ci. Such flaky tests make it hard to make
changes in the code because the confidence in the tests went down and developers
could dismiss genuine failures due to alert fatigue.
---
 CONTRIBUTING.md                              |   9 +
 Makefile                                     |   2 +-
 test/common.py                               | 329 ++++++++
 test/conftest.py                             |   1 +
 test/test_engine.py                          |   3 +
 test/test_http_delete.py                     | 166 ++--
 test/test_http_errors.py                     | 118 ++-
 test/test_http_get_collect.py                | 144 +---
 test/test_http_headers.py                    |  22 +-
 test/test_http_malformed_headers.py          |  97 +--
 test/test_http_params.py                     |  51 +-
 test/test_http_post_collect.py               | 135 +---
 test/test_http_requests_deleted_after_ttl.py | 224 ++----
 test/test_http_timeout.py                    |  52 +-
 test/test_privileges.py                      |  64 +-
 test/test_stat_statements.py                 | 123 ++-
 test/test_user_db.py                         |  43 +-
 test/test_worker_behavior.py                 | 763 +++++++++----------
 18 files changed, 1123 insertions(+), 1223 deletions(-)
 create mode 100644 test/common.py

diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index ebf27d06..aa47893f 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -64,6 +64,15 @@ $ xpg -v 13 test
 
 This will spawn a local db and an nginx server for testing.
 
+To run a single test , set `PYTEST_ARGS` to `-k test_function_name`. For example:
+
+```bash
+$ nix develop
+$ PYTEST_ARGS="-k test_connect" xpg test
+```
+
+Will run the `test_connect` test only. `PYTEST_ARGS` is passed through to pytest, so you can pass other arguments to pytest as well.
+
 ### Debugging
 
 You can turn on logging level to see curl traces with
diff --git a/Makefile b/Makefile
index 240a5dc2..0d1b0e35 100644
--- a/Makefile
+++ b/Makefile
@@ -85,4 +85,4 @@ include $(PGXS)
 
 .PHONY: test
 test:
-	net-with-nginx python -m pytest -s -vv
+	net-with-nginx python -m pytest -s -vv $(PYTEST_ARGS)
diff --git a/test/common.py b/test/common.py
new file mode 100644
index 00000000..a0ee8741
--- /dev/null
+++ b/test/common.py
@@ -0,0 +1,329 @@
+import time
+from sqlalchemy import create_engine, text
+from sqlalchemy.orm import Session
+
+
+def http_request(sess, query):
+    """
+    Execute query and commit to wake up the background worker.
+
+    The query should return a single row with a single column
+    containing the request id of the request. Returns the request
+    id
+    """
+    request_id = sess.execute(query).scalar_one()
+    # Commit to wakeup background worker
+    sess.commit()
+    return request_id
+
+
+def http_requests(sess, query):
+    """
+    Execute query and commit to wake up the background worker.
+
+    The query usually contains multiple http requests. Returns
+    the reqeust id of the first request in the query
+    """
+    (request_id,) = sess.execute(query).first()
+    # Commit to wakeup background worker
+    sess.commit()
+    return request_id
+
+
+def collect_response_sync(sess, request_id):
+    """
+    Wait for request with request_id to complete and return its response.
+
+    Flattens net._http_collect_response's nested composite return type
+    (status, message, response(status_code, headers, body)) into a single
+    row, so callers get a name-addressable mapping (e.g. result["body"])
+    instead of indexing into a stringified nested tuple.
+    """
+    return sess.execute(
+        text(
+            """
+        select
+            status,
+            message,
+            (response).status_code,
+            (response).headers,
+            (response).body
+        from net._http_collect_response(:request_id, async:=false);
+    """
+        ),
+        {"request_id": request_id},
+    ).mappings().fetchone()
+
+
+def is_worker_up(autocommit_sess):
+    """
+    Returns a function that checks whether worker is up or not
+
+    The returned function captures autocommit_sess argument and
+    uses it to track worker status. 
+    """
+
+    def fetch():
+        (worker_is_up,) = autocommit_sess.execute(
+            text("select is_worker_up();")
+        ).fetchone()
+        return worker_is_up
+    return fetch
+
+
+def get_queue_length(autocommit_sess):
+    """
+    Returns a function that returns the queue length
+
+    The returned function captures autocommit_sess argument and
+    uses it to track queue length. 
+    """
+
+    def fetch():
+        (queue_length,) = autocommit_sess.execute(text("""
+            select count(*) from net.http_request_queue;
+        """)).fetchone()
+        return queue_length
+    return fetch
+
+
+def get_response_count(autocommit_sess):
+    """
+    Returns a function that returns the number of rows in net._http_response table
+
+    The returned function captures autocommit_sess argument and
+    uses it to run the sql query. 
+    """
+
+    def fetch():
+        (response_count,) = autocommit_sess.execute(text("""
+            select count(*) from net._http_response;
+        """)).fetchone()
+        return response_count
+    return fetch
+
+
+def get_worker_state(autocommit_sess):
+    """
+    Returns a function that returns the background worker state
+
+    The returned function captures autocommit_sess argument and
+    uses it to run the sql query. 
+    """
+
+    def fetch():
+        (state,) = autocommit_sess.execute(text("""
+            select state from pg_stat_activity where backend_type ilike '%pg_net%';
+        """)).fetchone()
+        return state
+    return fetch
+
+
+def is_extension_installed(autocommit_sess):
+    """
+    Returns a function that returns whether pg_net is installed or not
+
+    The returned function captures autocommit_sess argument and
+    uses it to run the sql query. 
+    """
+
+    def fetch():
+        (extension_installed,) = autocommit_sess.execute(text("""
+            select count(*) = 1 from pg_extension where extname = 'pg_net';
+        """)).fetchone()
+        return extension_installed
+    return fetch
+
+
+def try_connect(engine, tmp_sess):
+    """
+    Returns a function that return whether postgres can accept connections.
+    """
+
+    def fetch():
+        try:
+            engine = create_engine("postgresql:///postgres")
+            ac_engine = engine.execution_options(
+                isolation_level="AUTOCOMMIT")
+            tmp_sess = Session(ac_engine)
+            return tmp_sess.execute(text("select 1")).fetchone()
+        except Exception:
+            return None
+    return fetch
+
+
+def wait_for_worker_down(autocommit_sess):
+    """
+    Waits until worker goes down
+
+    Or throws an error if it doesn't within a timeout
+    """
+
+    wait_until(
+        is_worker_up(autocommit_sess),
+        lambda worker_is_up: not worker_is_up,
+        description="background worker to go down",
+    )
+
+
+def wait_for_worker_up(autocommit_sess):
+    """
+    Waits until worker comes up
+
+    Or throws an error if it doesn't within a timeout
+    """
+
+    wait_until(
+        is_worker_up(autocommit_sess),
+        lambda worker_is_up: worker_is_up,
+        description="background worker to come up",
+    )
+
+
+def wait_for_worker_state(autocommit_sess, expected_state):
+    """
+    Waits until worker state matches expected_state
+
+    Or throws an error if it doesn't within a timeout
+    """
+
+    wait_until(
+        get_worker_state(autocommit_sess),
+        lambda state: state == expected_state,
+        description=f"background worker state to become {expected_state}",
+    )
+
+
+def wait_for_queue_drain(autocommit_sess):
+    """
+    Waits until the request queue is empty
+
+    Or throws an error if it doesn't within a timeout
+    """
+
+    wait_until(
+        get_queue_length(autocommit_sess),
+        lambda queue_length: queue_length == 0,
+        description="queue to drain"
+    )
+
+
+def wait_for_response_count(autocommit_sess, expected_count):
+    """
+    Waits until number of rows in net._http_response match expected_count
+
+    Or throws an error if it doesn't within a timeout
+    """
+
+    wait_until(
+        get_response_count(autocommit_sess),
+        lambda response_count: response_count == expected_count,
+        description="all responses to arrive"
+    )
+
+
+def wait_for_any_response(autocommit_sess):
+    """
+    Waits for at least one row in in net._http_response
+
+    Or throws an error if it doesn't within a timeout
+    """
+
+    wait_until(
+        get_response_count(autocommit_sess),
+        lambda response_count: response_count > 0,
+        description="any response to arrive"
+    )
+
+
+def wait_for_extension_drop(autocommit_sess):
+    """
+    Waits for pg_net to be dropped
+
+    Or throws an error if it doesn't within a timeout
+    """
+
+    wait_until(
+        is_extension_installed(autocommit_sess),
+        lambda extension_installed: not extension_installed,
+        description="extension to be dropped"
+    )
+
+
+def wait_for_postgres_ready(engine, tmp_sess):
+    """
+    Waits for postgres to be ready to accept connections
+
+    Or throws an error if it doesn't within a timeout
+    """
+
+    wait_until(
+        try_connect(engine, tmp_sess),
+        lambda result: result is not None,
+        description="postgres to become ready"
+    )
+
+
+def wait_until(fetch, predicate, timeout=10, sleep_interval=0.1, description="condition"):
+    deadline = time.time() + timeout
+    result = None
+    while time.time() < deadline:
+        result = fetch()
+        if predicate(result):
+            return result
+        time.sleep(sleep_interval)
+    raise AssertionError(
+        f"Timed out after {timeout}s waiting for {description} (last value: {result!r})"
+    )
+
+
+def wakeup_worker(sess):
+    """
+    Wakes up the worker manually by calling net.wake() and committing
+    """
+
+    sess.execute(text("select net.wake()"))
+    sess.commit()  # commit so worker  wakes
+
+
+def restart_worker(sess):
+    """
+    Restarts the worker and waits for it to come back up
+
+    You'd think that the following implementation should
+    restart the worker and wait for it to come back up:
+
+    sess.execute(text("select net.worker_restart()"))
+    sess.execute(text("select net.wait_until_running()"))
+
+    But it has a race condition in which this function might
+    return before the worker has restarted. This happens because
+    net.worker_restart() returns immediately after setting a flag
+    to indicate to the core worker loop to restart. Then the
+    net.wait_until_running() function waits for the worker state to
+    become WS_RUNNING. But it can read the state from either the worker
+    before the restart or after. In the first case it returns before
+    the worker has restarted properly, and in the second case it
+    behaves correctly.
+
+    Instead we compare the pids of the workers before and after the
+    restart which guarantees that the worker has restarted. After the
+    restart we still run net.wait_until_running() for it to be
+    intialized properly.
+    """
+
+    def fetch_worker_pid():
+        return sess.execute(text("""
+            select pid from pg_stat_activity where backend_type ilike '%pg_net%';
+        """)).scalar()
+
+    old_pid = fetch_worker_pid()
+    sess.execute(text("select net.worker_restart()"))
+    wait_until(
+        fetch_worker_pid,
+        lambda pid: pid is not None and pid != old_pid,
+        description="background worker to restart with a new pid",
+    )
+    # the new worker's pg_stat_activity row appears slightly before it
+    # publishes WS_RUNNING, so also wait for it to be fully up
+    sess.execute(text("select net.wait_until_running()"))
diff --git a/test/conftest.py b/test/conftest.py
index 811cad2a..c38f9c19 100644
--- a/test/conftest.py
+++ b/test/conftest.py
@@ -3,6 +3,7 @@
 from sqlalchemy.orm import Session
 from sqlalchemy import text
 
+
 @pytest.fixture(scope="function")
 def engine():
     engine = create_engine("postgresql:///postgres")
diff --git a/test/test_engine.py b/test/test_engine.py
index 12a075e0..7bc80f10 100644
--- a/test/test_engine.py
+++ b/test/test_engine.py
@@ -1,5 +1,8 @@
 from sqlalchemy import text
 
+
 def test_connect(sess):
+    """Sanity test verifying connection to postgres works"""
+
     (x,) = sess.execute(text("select 1")).fetchone()
     assert x == 1
diff --git a/test/test_http_delete.py b/test/test_http_delete.py
index eeb6074d..3da3753b 100644
--- a/test/test_http_delete.py
+++ b/test/test_http_delete.py
@@ -1,24 +1,26 @@
+import json
 from sqlalchemy import text
+from common import collect_response_sync, http_request
+
 
 def test_http_delete_returns_id(sess):
-    """net.http_delete returns a bigint id"""
+    """Test net.http_delete returns an id"""
 
-    (request_id,) = sess.execute(text(
+    request_id = http_request(sess, text(
         """
         select net.http_get(
             url:='http://localhost:8080/delete'
         );
     """
-    )).fetchone()
+    ))
 
     assert request_id == 1
 
 
 def test_http_delete_collect_sync_success(sess):
-    """test net.http_delete works"""
+    """Test net.http_delete works"""
 
-    # Create a request
-    (request_id,) = sess.execute(text(
+    request_id = http_request(sess, text(
         """
         select net.http_delete(
             url:='http://localhost:8080/delete'
@@ -26,86 +28,59 @@ def test_http_delete_collect_sync_success(sess):
         ,   headers:= '{"X-Baz": "foo"}'
         );
     """
-    )).fetchone()
-
-    # Commit so background worker can start
-    sess.commit()
+    ))
 
-    # Collect the response, waiting as needed
-    response = sess.execute(
-        text(
-            """
-        select * from net._http_collect_response(:request_id, async:=false);
-    """
-        ),
-        {"request_id": request_id},
-        ).fetchone()
+    response = collect_response_sync(sess, request_id)
 
     assert response is not None
-    assert response[0] == "SUCCESS"
-    assert response[1] == "ok"
-    assert response[2] is not None
-    assert "X-Baz" in response[2]
-    assert "param-foo" in response[2]
+    assert response["status"] == "SUCCESS"
+    assert response["message"] == "ok"
+
+    # /delete endpoint returns params and headers in the response body
+    assert response["body"] is not None
+    assert "X-Baz" in response["body"]
+    assert "param-foo" in response["body"]
 
 
 def test_http_delete_positional_args(sess):
-    """test net.http_delete works with positional arguments. This to ensure backwards compat when a new parameter is added to the function."""
+    """
+    Test net.http_delete works with positional arguments.
+    This to ensure backwards compat when a new parameter is added to the function.
+    """
 
-    (request_id,) = sess.execute(text(
+    # Delete call with url only
+    request_id = http_request(sess, text(
         """
         select net.http_delete(
             'http://localhost:8080/delete'
         );
     """
-    )).fetchone()
-
-    # Commit so background worker can start
-    sess.commit()
+    ))
 
-    # Collect the response, waiting as needed
-    response = sess.execute(
-        text(
-            """
-        select * from net._http_collect_response(:request_id, async:=false);
-    """
-        ),
-        {"request_id": request_id},
-        ).fetchone()
+    response = collect_response_sync(sess, request_id)
 
     assert response is not None
-    assert response[0] == "SUCCESS"
-    assert response[1] == "ok"
+    assert response["status"] == "SUCCESS"
+    assert response["message"] == "ok"
 
-
-    (request_id,) = sess.execute(text(
+    # Delete call with url and params
+    request_id = http_request(sess, text(
         """
         select net.http_delete(
             'http://localhost:8080/delete',
             '{"param-foo": "bar"}'
         );
     """
-    )).fetchone()
-
-    # Commit so background worker can start
-    sess.commit()
+    ))
 
-    # Collect the response, waiting as needed
-    response = sess.execute(
-        text(
-            """
-        select * from net._http_collect_response(:request_id, async:=false);
-    """
-        ),
-        {"request_id": request_id},
-        ).fetchone()
+    response = collect_response_sync(sess, request_id)
 
     assert response is not None
-    assert response[0] == "SUCCESS"
-    assert response[1] == "ok"
-
+    assert response["status"] == "SUCCESS"
+    assert response["message"] == "ok"
 
-    (request_id,) = sess.execute(text(
+    # Delete call with url, params, and headers
+    request_id = http_request(sess, text(
         """
         select net.http_delete(
             'http://localhost:8080/delete',
@@ -113,27 +88,16 @@ def test_http_delete_positional_args(sess):
             '{"X-Baz": "foo"}'
         );
     """
-    )).fetchone()
+    ))
 
-    # Commit so background worker can start
-    sess.commit()
-
-    # Collect the response, waiting as needed
-    response = sess.execute(
-        text(
-            """
-        select * from net._http_collect_response(:request_id, async:=false);
-    """
-        ),
-        {"request_id": request_id},
-        ).fetchone()
+    response = collect_response_sync(sess, request_id)
 
     assert response is not None
-    assert response[0] == "SUCCESS"
-    assert response[1] == "ok"
-
+    assert response["status"] == "SUCCESS"
+    assert response["message"] == "ok"
 
-    (request_id,) = sess.execute(text(
+    # Delete call with url, params, headers, and timeout
+    request_id = http_request(sess, text(
         """
         select net.http_delete(
             'http://localhost:8080/delete',
@@ -142,53 +106,29 @@ def test_http_delete_positional_args(sess):
             5000
         );
     """
-    )).fetchone()
+    ))
 
-    # Commit so background worker can start
-    sess.commit()
-
-    # Collect the response, waiting as needed
-    response = sess.execute(
-        text(
-            """
-        select * from net._http_collect_response(:request_id, async:=false);
-    """
-        ),
-        {"request_id": request_id},
-        ).fetchone()
+    response = collect_response_sync(sess, request_id)
 
     assert response is not None
-    assert response[0] == "SUCCESS"
-    assert response[1] == "ok"
+    assert response["status"] == "SUCCESS"
+    assert response["message"] == "ok"
 
 
 def test_http_delete_with_body(sess):
-    """delete with request body works"""
+    """Test delete with request body works"""
 
-    # Create a request
-    (request_id,) = sess.execute(text(
+    request_id = http_request(sess, text(
         """
         select net.http_delete(
             url  :='http://localhost:8080/delete_w_body'
         ,   body := '{"key": "val"}'
         );
     """
-    )).fetchone()
-
-    # Commit so background worker can start
-    sess.commit()
-
-    # Collect the response, waiting as needed
-    (response_json,) = sess.execute(
-        text(
-            """
-        select
-            ((x.response).body)::jsonb body_json
-        from
-            net._http_collect_response(:request_id, async:=false) x;
-    """
-        ),
-        {"request_id": request_id},
-        ).fetchone()
+    ))
 
-    assert response_json["key"] == "val"
+    response = collect_response_sync(sess, request_id)
+
+    assert response is not None
+    assert response["body"] is not None
+    assert json.loads(response["body"])["key"] == "val"
diff --git a/test/test_http_errors.py b/test/test_http_errors.py
index 78cd4f3e..ec52b449 100644
--- a/test/test_http_errors.py
+++ b/test/test_http_errors.py
@@ -1,15 +1,15 @@
-import time
-
 import pytest
 from sqlalchemy import text
+from common import collect_response_sync, http_request, http_requests
 
 wrong_port = 6666
 
+
 def test_get_bad_url(sess):
-    """net.http_get returns a descriptive errors for bad urls"""
+    """Test net.http_get returns a descriptive errors for bad urls"""
 
     with pytest.raises(Exception) as execinfo:
-        res = sess.execute(text(
+        sess.execute(text(
             f"""
             select net.http_get('localhost:{wrong_port}');
         """
@@ -19,7 +19,7 @@ def test_get_bad_url(sess):
 
 
 def test_http_get_rejects_relative_url(sess):
-    """net.http_get with a correct error when given a relative url"""
+    """Test net.http_get with a correct error when given a relative url"""
 
     with pytest.raises(Exception) as execinfo:
         sess.execute(text(
@@ -32,10 +32,10 @@ def test_http_get_rejects_relative_url(sess):
 
 
 def test_bad_post(sess):
-    """net.http_post with an empty url + body returns an error"""
+    """Test net.http_post with an empty url + body returns an error"""
 
     with pytest.raises(Exception) as execinfo:
-        res = sess.execute(text(
+        sess.execute(text(
             """
             select net.http_post(null, '{"hello": "world"}');
         """
@@ -44,7 +44,7 @@ def test_bad_post(sess):
 
 
 def test_bad_get(sess):
-    """net.http_get with an empty url + body returns an error"""
+    """Test net.http_get with an empty url + body returns an error"""
 
     with pytest.raises(Exception) as execinfo:
         res = sess.execute(text(
@@ -56,10 +56,10 @@ def test_bad_get(sess):
 
 
 def test_bad_delete(sess):
-    """net.http_delete with an empty url + body returns an error"""
+    """Test net.http_delete with an empty url + body returns an error"""
 
     with pytest.raises(Exception) as execinfo:
-        res = sess.execute(text(
+        sess.execute(text(
             """
             select net.http_delete(null);
         """
@@ -68,7 +68,7 @@ def test_bad_delete(sess):
 
 
 def test_bad_utils(sess):
-    """util functions of pg_net return null"""
+    """Test util functions of pg_net return null"""
 
     res = sess.execute(text(
         """
@@ -88,29 +88,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"""
+    """
+    Test the worker doesn't crash on many failed responses
+    with connection refused
+    """
 
-    (request_id,) = sess.execute(text(
+    request_id = http_requests(sess, text(
         f"""
         select net.http_get('http://localhost:{wrong_port}') from generate_series(1,10) offset 9;
     """
-    )).fetchone()
-    sess.commit()
+    ))
+
+    response = collect_response_sync(sess, request_id)
 
-    # 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"
+    assert response["status"] == "ERROR"
 
-    (error_msg,count) = sess.execute(text(
-    """
+    (error_msg, count) = sess.execute(text(
+        """
         select error_msg, count(*) from net._http_response where status_code is null group by error_msg;
     """
     )).fetchone()
@@ -120,51 +115,38 @@ def test_it_keeps_working_after_many_connection_refused(sess):
     assert error_msg in expected
     assert count == 10
 
-    (request_id,) = sess.execute(text(
+    request_id = http_request(sess, text(
         """
         select net.http_get('http://localhost:8080/pathological?status=200');
     """
-    )).fetchone()
+    ))
 
-    sess.commit()
+    response = collect_response_sync(sess, request_id)
 
-    response = sess.execute(
-        text(
-            """
-        select * from net._http_collect_response(:request_id, async:=false);
-    """
-        ),
-        {"request_id": request_id},
-    ).fetchone()
+    assert response["status"] == "SUCCESS"
+    assert response["message"] == "ok"
+    assert response["status_code"] == 200
 
-    assert response[0] == "SUCCESS"
-    assert response[1] == "ok"
-    assert response[2].startswith("(200")
 
 def test_it_keeps_working_after_server_returns_nothing(sess):
-    """the worker doesn't crash on many failed responses with server returned nothing"""
+    """
+    Test the worker doesn't crash on many failed responses
+    with server returned nothing
+    """
 
-    (request_id,) = sess.execute(text(
+    request_id = http_requests(sess, text(
         """
         select net.http_get('http://localhost:8080/pathological?disconnect=true') from generate_series(1,10) offset 9;
     """
-    )).fetchone()
-    sess.commit()
+    ))
+
+    response = collect_response_sync(sess, request_id)
 
-    # 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"
+    assert response["status"] == "ERROR"
 
-    (error_msg,count) = sess.execute(text(
-    """
+    (error_msg, count) = sess.execute(text(
+        """
         select error_msg, count(*) from net._http_response where status_code is null group by error_msg;
     """
     )).fetchone()
@@ -172,28 +154,18 @@ def test_it_keeps_working_after_server_returns_nothing(sess):
     assert error_msg == "Server returned nothing (no headers, no data)"
     assert count == 10
 
-    (request_id,) = sess.execute(text(
+    request_id = http_requests(sess, text(
         """
         select net.http_get('http://localhost:8080/pathological?status=200') from generate_series(1,10) offset 9;
     """
-    )).fetchone()
+    ))
 
-    sess.commit()
+    response = collect_response_sync(sess, request_id)
 
-    # 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"
+    assert response["status"] == "SUCCESS"
 
-    (status_code,count) = sess.execute(text(
-    """
+    (status_code, count) = sess.execute(text(
+        """
         select status_code, count(*) from net._http_response where status_code = 200 group by status_code;
     """
     )).fetchone()
diff --git a/test/test_http_get_collect.py b/test/test_http_get_collect.py
index cafa6ce1..0cf64ffb 100644
--- a/test/test_http_get_collect.py
+++ b/test/test_http_get_collect.py
@@ -1,27 +1,29 @@
 from sqlalchemy import text
 import time
 
+from common import collect_response_sync, http_request
+
 
 def test_http_get_returns_id(sess):
-    """net.http_get returns a bigint id"""
+    """Test net.http_get returns an id"""
 
-    (request_id,) = sess.execute(text(
+    request_id = http_request(sess, text(
         """
         select net.http_get('http://localhost:8080');
     """
-    )).fetchone()
+    ))
 
     assert request_id == 1
 
 
 def test_http_get_works_with_ip(sess):
-    """net.http_get returns a bigint id when using an IP with port"""
+    """Test net.http_get returns an id when using an IP with port"""
 
-    (request_id,) = sess.execute(text(
+    request_id = http_request(sess, text(
         """
         select net.http_get('http://127.0.0.1:8080');
     """
-    )).fetchone()
+    ))
 
     assert request_id == 1
 
@@ -29,60 +31,19 @@ def test_http_get_works_with_ip(sess):
 def test_http_get_collect_sync_success(sess):
     """Collect a response, waiting if it has not completed yet"""
 
-    # Create a request
-    (request_id,) = sess.execute(text(
+    request_id = http_request(sess, text(
         """
         select net.http_get('http://localhost:8080');
     """
-    )).fetchone()
-
-    # Commit so background worker can start
-    sess.commit()
+    ))
 
-    # Collect the response, waiting as needed
-    response = sess.execute(text(
-            """
-        select * from net._http_collect_response(:request_id, async:=false);
-    """
-        ),
-        {"request_id": request_id},
-    ).fetchone()
+    response = collect_response_sync(sess, request_id)
 
     assert response is not None
-    assert response[0] == "SUCCESS"
-    assert response[1] == "ok"
-    assert response[2] is not None
-    # psycopg2 does not deserialize nested composites
-    assert response[2].startswith("(200")
-
-
-# def test_http_get_collect_async_pending(sess):
-#     """Collect a response async before completed"""
-
-#     # Create a request
-#     (request_id,) = sess.execute(
-#         """
-#         select net.http_get('https://news.ycombinator.com');
-#     """
-#     ).fetchone()
-
-#     # Commit so background worker can start
-#     sess.commit()
-
-#     # Collect the response, waiting as needed
-#     response = sess.execute(
-#         text(
-#             """
-#         select * from net._http_collect_response(:request_id, async:=true);
-#     """
-#         ),
-#         {"request_id": request_id},
-#     ).fetchone()
-
-#     assert response is not None
-#     assert response[0] == "PENDING"
-#     assert "pending" in response[1]
-#     assert response[2] is None
+    assert response["status"] == "SUCCESS"
+    assert response["message"] == "ok"
+    assert response["body"] is not None
+    assert response["status_code"] == 200
 
 
 def test_http_collect_response_async_does_not_exist(sess):
@@ -99,114 +60,83 @@ def test_http_collect_response_async_does_not_exist(sess):
     assert "not found" in response[1]
     assert response[2] is None
 
+
 def test_http_get_responses_have_different_created_times(sess):
     """Ensure the rows in net._http_response have different created times"""
 
-    sess.execute(text(
+    http_request(sess, text(
         """
         select net.http_get('http://localhost:8080/echo-method')
     """
     ))
-    sess.commit()
 
     time.sleep(1)
 
-    sess.execute(text(
+    http_request(sess, text(
         """
         select net.http_get('http://localhost:8080/echo-method')
     """
     ))
-    sess.commit()
 
     time.sleep(1)
 
-    sess.execute(text(
+    http_request(sess, text(
         """
         select net.http_get('http://localhost:8080/echo-method')
     """
     ))
-    sess.commit()
 
     time.sleep(1)
 
     count = sess.execute(text(
-            """
+        """
         select count(distinct created) from net._http_response;
     """
     )).scalar()
 
     assert count == 3
 
+
 def test_http_get_collect_with_redirect(sess):
-    """Follows a redirect and collects a response"""
+    """Test pg_net follows a redirect and collects a response"""
 
-    # Create a request
-    (request_id,) = sess.execute(text(
+    request_id = http_request(sess, text(
         """
         select net.http_get('http://localhost:8080/redirect_me');
     """
-    )).fetchone()
-
-    # Commit so background worker can start
-    sess.commit()
+    ))
 
-    # Collect the response, waiting as needed
-    response = sess.execute(text(
-            """
-        select * from net._http_collect_response(:request_id, async:=false);
-    """
-        ),
-        {"request_id": request_id},
-    ).fetchone()
+    response = collect_response_sync(sess, request_id)
 
     assert response is not None
-    assert "I got redirected" in response[2]
+    assert response["body"] == "I got redirected\n"
+
 
 def test_http_get_ipv6(sess):
-    """Can resolve an ipv6 only server"""
+    """Test pg_net can resolve an ipv6 only server"""
 
-    # Create a request
-    (request_id,) = sess.execute(text(
+    request_id = http_request(sess, text(
         """
         select net.http_get('http://localhost:8888/');
     """
-    )).fetchone()
+    ))
 
-    # Commit so background worker can start
-    sess.commit()
-
-    # Collect the response, waiting as needed
-    response = sess.execute(text(
-            """
-        select * from net._http_collect_response(:request_id, async:=false);
-    """
-        ),
-        {"request_id": request_id},
-    ).fetchone()
+    response = collect_response_sync(sess, request_id)
 
     assert response is not None
-    assert "Hello ipv6 only" in response[2]
+    assert response["body"] == "Hello ipv6 only\n"
 
 
 def test_http_get_null_headers(sess):
-    """net.http_get can have null headers"""
+    """Test net.http_get can have null headers"""
 
-    (request_id,) = sess.execute(text(
+    request_id = http_request(sess, text(
         """
         select net.http_get('http://localhost:8080', null::jsonb, null::jsonb, 100);
     """
-    )).fetchone()
-
-    sess.commit()
+    ))
 
-    response = sess.execute(
-        text(
-            """
-        select * from net._http_collect_response(:request_id, async:=false);
-    """
-        ),
-        {"request_id": request_id},
-    ).fetchone()
+    response = collect_response_sync(sess, request_id)
 
     assert response is not None
-    assert "Hello world" in response[2]
+    assert response["body"] == "Hello world\n"
diff --git a/test/test_http_headers.py b/test/test_http_headers.py
index e01624d4..85ade949 100644
--- a/test/test_http_headers.py
+++ b/test/test_http_headers.py
@@ -1,31 +1,19 @@
 from sqlalchemy import text
+from common import collect_response_sync, http_request
 
 
 def test_http_headers_set(sess):
     """Check that headers are being set"""
-    # Create a request
-    (request_id,) = sess.execute(text(
+    request_id = http_request(sess, text(
         """
         select net.http_get(
             url:='http://localhost:8080/headers',
             headers:='{"pytest-header": "pytest-header", "accept": "application/json"}'
         );
     """
-    )).fetchone()
+    ))
 
-    # Commit so background worker can start
-    sess.commit()
+    response = collect_response_sync(sess, request_id)
 
-    # Collect the response, waiting as needed
-    response = sess.execute(
-        text(
-            """
-        select * from net._http_collect_response(:request_id, async:=false);
-    """
-        ),
-        {"request_id": request_id},
-    ).fetchone()
-    print(response)
     assert response is not None
-    assert response[0] == "SUCCESS"
-    assert "pytest-header" in response[2]
+    assert "pytest-header" in response["body"]
diff --git a/test/test_http_malformed_headers.py b/test/test_http_malformed_headers.py
index 9e22dfbc..8497b678 100644
--- a/test/test_http_malformed_headers.py
+++ b/test/test_http_malformed_headers.py
@@ -1,112 +1,83 @@
 from sqlalchemy import text
+from common import collect_response_sync, http_request
+
 
 def test_http_header_missing_value(sess):
     """Check that a `MissingValue: ` header is processed correctly"""
 
-    (request_id,) = sess.execute(text(
+    request_id = http_request(sess, text(
         """
         select net.http_get(
             url:='http://localhost:8080/pathological?malformed-header=missing-value'
         );
     """
-    )).fetchone()
+    ))
 
-    # Commit so background worker can start
-    sess.commit()
+    response = collect_response_sync(sess, request_id)
 
-    # Collect the 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"
-    assert "MissingValue" in response[2]
+    assert response["status"] == "SUCCESS"
+    assert "MissingValue" in response["headers"]
 
 
 def test_http_header_injection(sess):
-    """Check that a `HeaderInjection Injected-Header: This header contains an injection` header fails without crashing"""
+    """
+    Check that a `HeaderInjection Injected-Header: This header
+    contains an injection` header fails without crashing
+    """
 
-    (request_id,) = sess.execute(text(
+    request_id = http_request(sess, text(
         """
         select net.http_get(
             url:='http://localhost:8080/pathological?malformed-header=header-injection'
         );
     """
-    )).fetchone()
+    ))
 
-    # Commit so background worker can start
-    sess.commit()
+    response = collect_response_sync(sess, request_id)
 
-    # Collect the 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"
-    assert "Weird server reply" in response[1]
+    assert response["status"] == "ERROR"
+    assert "Weird server reply" in response["message"]
 
 
 def test_http_header_spaces(sess):
-    """Check that a `Spaces In Header Name: This header name contains spaces` header is processed correctly"""
+    """
+    Check that a `Spaces In Header Name: This header name contains spaces`
+    header is processed correctly
+    """
 
-    (request_id,) = sess.execute(text(
+    request_id = http_request(sess, text(
         """
         select net.http_get(
             url:='http://localhost:8080/pathological?malformed-header=spaces-in-header-name'
         );
     """
-    )).fetchone()
+    ))
 
-    # Commit so background worker can start
-    sess.commit()
+    response = collect_response_sync(sess, request_id)
 
-    # Collect the 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"
-    assert "Spaces In Header Name" in response[2]
+    assert response["status"] == "SUCCESS"
+    assert "Spaces In Header Name" in response["headers"]
 
 
 def test_http_header_non_printable_chars(sess):
-    """Check that a `NonPrintableChars: NonPrintableChars\\u0001\\u0002` header is processed correctly"""
+    """
+    Check that a `NonPrintableChars: NonPrintableChars\\u0001\\u0002`
+    header is processed correctly
+    """
 
-    (request_id,) = sess.execute(text(
+    request_id = http_request(sess, text(
         """
         select net.http_get(
             url:='http://localhost:8080/pathological?malformed-header=non-printable-chars'
         );
     """
-    )).fetchone()
+    ))
 
-    # Commit so background worker can start
-    sess.commit()
+    response = collect_response_sync(sess, request_id)
 
-    # Collect the 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"
-    assert r"NonPrintableChars\\u0001\\u0002" in response[2]
+    assert response["status"] == "SUCCESS"
+    assert response["headers"]["NonPrintableChars"] == "NonPrintableChars\x01\x02"
diff --git a/test/test_http_params.py b/test/test_http_params.py
index 624cd269..53615fc1 100644
--- a/test/test_http_params.py
+++ b/test/test_http_params.py
@@ -1,63 +1,38 @@
 from sqlalchemy import text
+from common import collect_response_sync, http_request
 
 
 def test_http_get_url_params_set(sess):
-    """Check that params are being set on GET
-    """
-    # Create a request
-    (request_id,) = sess.execute(text(
+    """Check that params are being set on GET"""
+    request_id = http_request(sess, text(
         """
         select net.http_get(
             url:='http://localhost:8080/anything',
             params:='{"hello": "world"}'::jsonb
         );
     """
-    )).fetchone()
+    ))
 
-    # Commit so background worker can start
-    sess.commit()
+    response = collect_response_sync(sess, request_id)
 
-    # Collect the response, waiting as needed
-    response = sess.execute(
-        text(
-            """
-        select * from net._http_collect_response(:request_id, async:=false);
-    """
-        ),
-        {"request_id": request_id},
-    ).fetchone()
-    print(response)
     assert response is not None
-    assert response[0] == "SUCCESS"
-    assert "?hello=world" in response[2]
+    assert response["status"] == "SUCCESS"
+    assert "?hello=world" in response["body"]
 
 
 def test_http_post_url_params_set(sess):
-    """Check that params are being set on POST
-    """
-    # Create a request
-    (request_id,) = sess.execute(text(
+    """Check that params are being set on POST"""
+    request_id = http_request(sess, text(
         """
         select net.http_post(
             url:='http://localhost:8080/anything',
             params:='{"hello": "world"}'::jsonb
         );
     """
-    )).fetchone()
+    ))
 
-    # Commit so background worker can start
-    sess.commit()
+    response = collect_response_sync(sess, request_id)
 
-    # Collect the response, waiting as needed
-    response = sess.execute(
-        text(
-            """
-        select * from net._http_collect_response(:request_id, async:=false);
-    """
-        ),
-        {"request_id": request_id},
-    ).fetchone()
-    print(response)
     assert response is not None
-    assert response[0] == "SUCCESS"
-    assert "?hello=world" in response[2]
+    assert response["status"] == "SUCCESS"
+    assert "?hello=world" in response["body"]
diff --git a/test/test_http_post_collect.py b/test/test_http_post_collect.py
index 7687792f..004f930a 100644
--- a/test/test_http_post_collect.py
+++ b/test/test_http_post_collect.py
@@ -1,32 +1,34 @@
+import json
 from sqlalchemy import text
+from common import collect_response_sync, http_request
 
 
 def test_http_post_returns_id(sess):
-    """net.http_post returns a bigint id"""
+    """Test net.http_post returns an id"""
 
-    (request_id,) = sess.execute(text(
+    request_id = http_request(sess, text(
         """
         select net.http_post(
             url:='http://localhost:8080/post',
             body:='{}'::jsonb
         );
     """
-    )).fetchone()
+    ))
 
     assert request_id == 1
 
 
 def test_http_post_special_chars_body(sess):
-    """net.http_post returns a bigint id"""
+    """Test net.http_post returns an id"""
 
-    (request_id,) = sess.execute(text(
+    request_id = http_request(sess, text(
         """
         select net.http_post(
             url:='http://localhost:8080/post',
             body:=json_build_object('foo', 'ba"r')::jsonb
         );
     """
-    )).fetchone()
+    ))
 
     assert request_id == 1
 
@@ -34,71 +36,26 @@ def test_http_post_special_chars_body(sess):
 def test_http_post_collect_sync_success(sess):
     """Collect a response, waiting if it has not completed yet"""
 
-    # Create a request
-    (request_id,) = sess.execute(text(
+    request_id = http_request(sess, text(
         """
         select net.http_post(
             url:='http://localhost:8080/post'
         );
     """
-    )).fetchone()
-
-    # Commit so background worker can start
-    sess.commit()
+    ))
 
-    # Collect the response, waiting as needed
-    response = sess.execute(
-        text(
-            """
-        select * from net._http_collect_response(:request_id, async:=false);
-    """
-        ),
-        {"request_id": request_id},
-    ).fetchone()
+    response = collect_response_sync(sess, request_id)
 
     assert response is not None
-    assert response[0] == "SUCCESS"
-    assert response[1] == "ok"
-    assert response[2] is not None
-
-
-# def test_http_post_collect_async_pending(sess):
-#     """Collect a response async before completed"""
-
-#     # Create a request
-#     (request_id,) = sess.execute(
-#         """
-#         select net.http_post(
-#             url:='http://localhost:8080/post',
-#             body:='{}'::jsonb
-#         );
-#     """
-#     ).fetchone()
-
-#     # Commit so background worker can start
-#     sess.commit()
-
-#     # Collect the response, waiting as needed
-#     response = sess.execute(
-#         text(
-#             """
-#         select * from net._http_collect_response(:request_id, async:=true);
-#     """
-#         ),
-#         {"request_id": request_id},
-#     ).fetchone()
-
-#     assert response is not None
-#     assert response[0] == "PENDING"
-#     assert "pending" in response[1]
-#     assert response[2] is None
+    assert response["status"] == "SUCCESS"
+    assert response["message"] == "ok"
+    assert response["body"] is not None
 
 
 def test_http_post_collect_non_empty_body(sess):
     """Collect a response async before completed"""
 
-    # Create a request
-    (request_id,) = sess.execute(text(
+    request_id = http_request(sess, text(
         """
         select net.http_post(
             url:='http://localhost:8080/post',
@@ -106,40 +63,16 @@ def test_http_post_collect_non_empty_body(sess):
             headers:='{"Content-Type": "application/json", "accept": "application/json"}'::jsonb
         );
     """
-    )).fetchone()
+    ))
 
-    # Commit so background worker can start
-    sess.commit()
+    response = collect_response_sync(sess, request_id)
 
-    # Collect the 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"
-    assert "ok" in response[1]
-    assert "hello" in response[2]
-    assert "world" in response[2]
-
-    # Make sure response is json
-    (response_json,) = sess.execute(
-        text(
-            """
-        select
-            ((x.response).body)::jsonb body_json
-        from
-            net._http_collect_response(:request_id, async:=false) x;
-    """
-        ),
-        {"request_id": request_id},
-    ).fetchone()
-
-    assert response_json["hello"] == "world"
+    assert response["status"] == "SUCCESS"
+    assert response["message"] == "ok"
+    assert response["body"] is not None
+    # Assert that response is json
+    assert json.loads(response["body"])["hello"] == "world"
 
 
 def test_http_post_wrong_header_exception(sess):
@@ -166,7 +99,6 @@ def test_http_post_wrong_header_exception(sess):
 def test_http_post_no_content_type_coerce(sess):
     """Confirm that a missing content type coerces to application/json"""
 
-    # Create a request
     request_id, = sess.execute(text(
         """
         select net.http_post(
@@ -176,7 +108,6 @@ def test_http_post_no_content_type_coerce(sess):
     """
     )).fetchone()
 
-
     headers, = sess.execute(text(
         """
         select
@@ -193,28 +124,18 @@ def test_http_post_no_content_type_coerce(sess):
 
 
 def test_http_post_empty_body(sess):
-    """net.http_post can post a null body"""
+    """Test net.http_post can post a null body"""
 
-    (request_id,) = sess.execute(text(
+    request_id = http_request(sess, text(
         """
         select net.http_post(
             url:='http://localhost:8080/echo-method',
             body:=null
         );
     """
-    )).fetchone()
+    ))
 
-    sess.commit()
+    response = collect_response_sync(sess, request_id)
 
-    (body) = sess.execute(
-        text(
-            """
-        select
-            (x.response).body as body
-        from net._http_collect_response(:request_id, async:=false) x;
-    """
-        ),
-        {"request_id": request_id},
-    ).fetchone()
-
-    assert 'POST' in str(body)
+    assert response is not None
+    assert response["body"] == "POST\n"
diff --git a/test/test_http_requests_deleted_after_ttl.py b/test/test_http_requests_deleted_after_ttl.py
index 071d7957..33b6d838 100644
--- a/test/test_http_requests_deleted_after_ttl.py
+++ b/test/test_http_requests_deleted_after_ttl.py
@@ -1,187 +1,127 @@
 import time
-
-import pytest
 from sqlalchemy import text
+from common import collect_response_sync, http_request, http_requests, restart_worker
+from common import wait_for_response_count, wakeup_worker
 
-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 '1 second'"))
-    autocommit_sess.execute(text("select net.worker_restart()"))
-    autocommit_sess.execute(text("select net.wait_until_running()"))
 
-    # Create a request
-    (request_id,) = sess.execute(text(
-        """
-        select net.http_get(
-            'http://localhost:8080/anything'
-        );
+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
     """
-    )).fetchone()
 
-    # Commit so background worker can start
-    sess.commit()
+    try:
+        autocommit_sess.execute(
+            text("alter system set pg_net.ttl to '1 second'"))
+        restart_worker(autocommit_sess)
 
-    # Confirm that the request was retrievable
-    response = sess.execute(
-        text(
+        request_id = http_request(sess, text(
             """
-        select * from net._http_collect_response(:request_id, async:=false);
-    """
-        ),
-        {"request_id": request_id},
-    ).fetchone()
-    assert response[0] == "SUCCESS"
+            select net.http_get(
+                'http://localhost:8080/anything'
+            );
+        """
+        ))
 
-    # Sleep until after request should have been deleted
-    time.sleep(1.1)
+        response = collect_response_sync(sess, request_id)
 
-    # Wake the worker manually, under normal operation this will happen when new requests are received
-    sess.execute(text("select net.wake()"))
+        assert response is not None
+        assert response["status"] == "SUCCESS"
 
-    sess.commit() # commit so worker  wakes
+        # Sleep a little more than ttl so that the request expires
+        time.sleep(1.1)
 
-    time.sleep(0.1) # wait for deletion
+        wakeup_worker(sess)
 
-    # Ensure the response is now empty
-    (count,) = sess.execute(
-        text(
-            """
-        select count(*) from net._http_response where id = :request_id;
-    """
-        ),
-        {"request_id": request_id},
-    ).fetchone()
-    assert count == 0
+        # Check that the worker deleted the expired response
+        wait_for_response_count(autocommit_sess, 0)
 
-    autocommit_sess.execute(text("alter system reset pg_net.ttl"))
-    autocommit_sess.execute(text("select net.worker_restart()"))
-    autocommit_sess.execute(text("select net.wait_until_running()"))
+    finally:
+        autocommit_sess.execute(text("alter system reset pg_net.ttl"))
+        restart_worker(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"""
+    """
+    Check that http responses will keep being deleted
+    until completion despite no new requests coming
+    """
 
-    (request_id,) = sess.execute(text(
+    request_id = http_requests(sess, text(
         """
         select net.http_get('http://localhost:8080/pathological?status=200') from generate_series(1,4) offset 3;
     """
-    )).fetchone()
+    ))
 
-    sess.commit()
+    response = collect_response_sync(sess, request_id)
 
-    # 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"
+    assert response["status"] == "SUCCESS"
 
-    (count,) = sess.execute(
-        text(
-            """
-        select count(*) from net._http_response
-    """
-        )
-    ).fetchone()
-    assert count == 4
+    wait_for_response_count(autocommit_sess, 4)
 
-    autocommit_sess.execute(text("alter system set pg_net.ttl to '1 second';"))
-    autocommit_sess.execute(text("alter system set pg_net.batch_size to 2;"))
-    autocommit_sess.execute(text("select pg_reload_conf();"))
+    try:
+        autocommit_sess.execute(
+            text("alter system set pg_net.ttl to '1 second';"))
+        autocommit_sess.execute(
+            text("alter system set pg_net.batch_size to 2;"))
+        autocommit_sess.execute(text("select pg_reload_conf();"))
 
-    # wait for ttl
-    time.sleep(1)
+        # Wait for ttl so that when we wakeup the worker it has
+        # some expired responses to delete
+        time.sleep(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
+        wakeup_worker(sess)
 
-    time.sleep(0.1)
+        # In one inner loop, the worker will delete batch size
+        # worth of responses
+        wait_for_response_count(autocommit_sess, 2)
 
-    (count,) = sess.execute(
-        text(
-            """
-        select count(*) from net._http_response
-    """
-        )
-    ).fetchone()
-    assert count == 2
-
-    # wait for another batch
-    time.sleep(1.1)
-
-    (count,) = sess.execute(
-        text(
-            """
-        select count(*) from net._http_response
-    """
-        )
-    ).fetchone()
-    assert count == 0
+        # But it will keep going as long as it had deleted
+        # some responses. So after a wait of 1 second it
+        # will delete another batch before going back to sleep
+        wait_for_response_count(autocommit_sess, 0)
 
-    autocommit_sess.execute(text("alter system reset pg_net.ttl"))
-    autocommit_sess.execute(text("alter system reset pg_net.batch_size"))
-    autocommit_sess.execute(text("select pg_reload_conf();"))
+    finally:
+        autocommit_sess.execute(text("alter system reset pg_net.ttl"))
+        autocommit_sess.execute(text("alter system reset pg_net.batch_size"))
+        autocommit_sess.execute(text("select pg_reload_conf();"))
 
 
 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"""
+    """
+    Check that http responses will keep being deleted despite no
+    new requests coming and despite worker restart
+    """
 
-    (request_id,) = sess.execute(text(
+    request_id = http_requests(sess, text(
         """
         select net.http_get('http://localhost:8080/pathological?status=200') from generate_series(1,4) offset 3;
     """
-    )).fetchone()
+    ))
 
-    sess.commit()
+    response = collect_response_sync(sess, request_id)
 
-    # 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"
+    assert response["status"] == "SUCCESS"
 
-    (count,) = sess.execute(
-        text(
-            """
-        select count(*) from net._http_response
-    """
-        )
-    ).fetchone()
-    assert count == 4
+    wait_for_response_count(autocommit_sess, 4)
 
-    # restart
-    autocommit_sess.execute(text("alter system set pg_net.ttl to '1 second';"))
-    autocommit_sess.execute(text("alter system set pg_net.batch_size to 2;"))
-    autocommit_sess.execute(text("select net.worker_restart()"))
-    autocommit_sess.execute(text("select net.wait_until_running()"))
+    try:
+        # Restart the worker
+        autocommit_sess.execute(
+            text("alter system set pg_net.ttl to '1 second';"))
+        autocommit_sess.execute(
+            text("alter system set pg_net.batch_size to 2;"))
+        restart_worker(autocommit_sess)
 
-    # wait for ttl
-    time.sleep(1.1)
+        # Wait for ttl so that the requests expire
+        time.sleep(1.1)
 
-    (count,) = sess.execute(
-        text(
-            """
-        select count(*) from net._http_response
-    """
-        )
-    ).fetchone()
-    assert count == 0
-
-    # reset
-    autocommit_sess.execute(text("alter system reset pg_net.ttl"))
-    autocommit_sess.execute(text("alter system reset pg_net.batch_size"))
-    autocommit_sess.execute(text("select net.worker_restart()"))
-    autocommit_sess.execute(text("select net.wait_until_running()"))
+        wait_for_response_count(autocommit_sess, 0)
+
+    finally:
+        # reset
+        autocommit_sess.execute(text("alter system reset pg_net.ttl"))
+        autocommit_sess.execute(text("alter system reset pg_net.batch_size"))
+        restart_worker(autocommit_sess)
diff --git a/test/test_http_timeout.py b/test/test_http_timeout.py
index bdf08ec8..aee6a282 100644
--- a/test/test_http_timeout.py
+++ b/test/test_http_timeout.py
@@ -1,24 +1,22 @@
 import time
-
-import pytest
 import re
 from sqlalchemy import text
+from common import http_request
+
 
 def test_http_get_timeout_reached(sess):
-    """net.http_get with timeout errs on a slow reply"""
+    """Test net.http_get with timeout errs on a slow reply"""
 
-    (request_id,) = sess.execute(text(
+    request_id = http_request(sess, text(
         """
         select net.http_get(url := 'http://localhost:8080/pathological?status=200&delay=6');
     """
-    )).fetchone()
-
-    sess.commit()
+    ))
 
     # wait for timeout
     time.sleep(7)
 
-    (content_type, content, response,timed_out) = sess.execute(
+    (content_type, content, response, timed_out) = sess.execute(
         text(
             """
         select content_type, content, error_msg, timed_out from net._http_response where id = :request_id;
@@ -34,7 +32,7 @@ def test_http_get_timeout_reached(sess):
 
 
 def test_http_detailed_timeout(sess):
-    """the timeout shows a detailed error msg"""
+    """Test the timeout shows a detailed error msg"""
 
     pattern = r"""
         Total\stime:\s*                     # Match 'Total time:' with optional spaces
@@ -58,18 +56,16 @@ def test_http_detailed_timeout(sess):
     # select net.http_get(url := 'http://localhost:8080/pathological', timeout_milliseconds := 1000);
 
     # Timeout at the HTTP step
-    (request_id,) = sess.execute(text(
+    request_id = http_request(sess, text(
         """
         select net.http_get(url := 'http://localhost:8080/pathological?delay=1', timeout_milliseconds := 1000)
     """
-    )).fetchone()
-
-    sess.commit()
+    ))
 
     # wait for timeout
     time.sleep(2.1)
 
-    (content_type, content, response,timed_out) = sess.execute(
+    (content_type, content, response, timed_out) = sess.execute(
         text(
             """
         select content_type, content, error_msg, timed_out from net._http_response where id = :request_id;
@@ -80,10 +76,10 @@ def test_http_detailed_timeout(sess):
 
     match = regex.search(response)
 
-    total_time   = float(match.group('A'))
-    dns_time     = float(match.group('B'))
+    total_time = float(match.group('A'))
+    dns_time = float(match.group('B'))
     tcp_ssl_time = float(match.group('C'))
-    http_time    = float(match.group('D'))
+    http_time = float(match.group('D'))
 
     assert content_type == None
     assert content == None
@@ -93,16 +89,18 @@ def test_http_detailed_timeout(sess):
     assert tcp_ssl_time > 0
     assert http_time > 0
 
+
 def test_http_get_succeed_with_gt_timeout(sess):
-    """net.http_get with timeout succeeds when the timeout is greater than the slow reply response time"""
+    """
+    Test net.http_get with timeout succeeds when the timeout
+    is greater than the slow reply response time
+    """
 
-    (request_id,) = sess.execute(text(
+    request_id = http_request(sess, text(
         """
         select net.http_get(url := 'http://localhost:8080?status=200&delay=3', timeout_milliseconds := 3500);
     """
-    )).fetchone()
-
-    sess.commit()
+    ))
 
     time.sleep(4.5)
 
@@ -117,11 +115,15 @@ def test_http_get_succeed_with_gt_timeout(sess):
 
     assert status_code == 200
 
+
 def test_many_slow_mixed_with_fast(sess):
-    """many fast responses finish despite being mixed with slow responses, the fast responses will wait the timeout duration"""
+    """
+    Test many fast responses finish despite being mixed with slow responses,
+    the fast responses will wait the timeout duration
+    """
 
     sess.execute(text(
-    """
+        """
       select
         net.http_get(url := 'http://localhost:8080/pathological?status=200')
       , net.http_get(url := 'http://localhost:8080/pathological?status=200&delay=2', timeout_milliseconds := 1000)
@@ -137,7 +139,7 @@ def test_many_slow_mixed_with_fast(sess):
     time.sleep(3)
 
     (request_successes, request_timeouts) = sess.execute(text(
-    """
+        """
       select
         count(*) filter (where error_msg is null and status_code = 200) as request_successes,
         count(*) filter (where error_msg is not null and error_msg like 'Timeout of 1000 ms reached%') as request_timeouts
diff --git a/test/test_privileges.py b/test/test_privileges.py
index aca3b4dd..1c8511a1 100644
--- a/test/test_privileges.py
+++ b/test/test_privileges.py
@@ -1,5 +1,6 @@
-import pytest
 from sqlalchemy import text
+from common import collect_response_sync, http_request
+
 
 def test_net_on_postgres_role(sess):
     """Check that the postgres role can use the net schema by default"""
@@ -7,28 +8,18 @@ def test_net_on_postgres_role(sess):
     role = sess.execute(text("select current_user;")).fetchone()
     assert role[0] == "postgres"
 
-    # Create a request
-    (request_id,) = sess.execute(text(
+    request_id = http_request(sess, text(
         """
         select net.http_get(
             'http://localhost:8080/anything'
         );
     """
-    )).fetchone()
+    ))
 
-    # Commit so background worker can start
-    sess.commit()
+    response = collect_response_sync(sess, request_id)
 
-    # Confirm that the request was retrievable
-    response = sess.execute(
-        text(
-            """
-        select * from net._http_collect_response(:request_id, async:=false);
-    """
-        ),
-        {"request_id": request_id},
-    ).fetchone()
-    assert response[0] == "SUCCESS"
+    assert response is not None
+    assert response["status"] == "SUCCESS"
 
 
 def test_net_on_pre_existing_role(sess):
@@ -37,7 +28,6 @@ def test_net_on_pre_existing_role(sess):
     role = sess.execute(text("select current_user;")).fetchone()
     assert role[0] == "postgres"
 
-    # Create a request
     (request_id, current_user) = sess.execute(text(
         """
         set local role to pre_existing;
@@ -49,21 +39,16 @@ def test_net_on_pre_existing_role(sess):
     assert request_id == 1
     assert current_user == 'pre_existing'
 
-    # Commit so background worker can start
+    # Commit to wakeup background worker
     sess.commit()
 
     # Confirm that the request was retrievable
-    response = sess.execute(
-        text(
-            """
-        set local role to pre_existing;
-        select *, current_user from net._http_collect_response(:request_id, async:=false);
-    """
-        ),
-        {"request_id": request_id},
-    ).fetchone()
-    assert response[0] == "SUCCESS"
-    assert response[3] == 'pre_existing' # current-user
+    sess.execute(text("set local role to pre_existing;"))
+    response = collect_response_sync(sess, request_id)
+    current_user = sess.execute(text("select current_user;")).scalar()
+    assert response["status"] == "SUCCESS"
+    assert current_user == 'pre_existing'
+
 
 def test_net_on_new_role(sess):
     """Check that a newly created role can use the net schema"""
@@ -75,7 +60,6 @@ def test_net_on_new_role(sess):
         create role another;
     """))
 
-    # Create a request
     (request_id, current_user) = sess.execute(text(
         """
         set local role to another;
@@ -87,23 +71,17 @@ def test_net_on_new_role(sess):
     assert request_id == 1
     assert current_user == 'another'
 
-    # Commit so background worker can start
+    # Commit to wakeup background worker
     sess.commit()
 
     # Confirm that the request was retrievable
-    response = sess.execute(
-        text(
-            """
-        set local role to another;
-        select *, current_user from net._http_collect_response(:request_id, async:=false);
-    """
-        ),
-        {"request_id": request_id},
-    ).fetchone()
-    assert response[0] == "SUCCESS"
-    assert response[3] == 'another' # current-user
+    sess.execute(text("set local role to another;"))
+    response = collect_response_sync(sess, request_id)
+    current_user = sess.execute(text("select current_user;")).scalar()
+    assert response["status"] == "SUCCESS"
+    assert current_user == 'another'
 
-    ## can use the net.worker_restart function
+    # can use the net.worker_restart function
     (res, current_user) = sess.execute(
         text(
             """
diff --git a/test/test_stat_statements.py b/test/test_stat_statements.py
index 6117c8c9..6b6b4e4d 100644
--- a/test/test_stat_statements.py
+++ b/test/test_stat_statements.py
@@ -1,11 +1,10 @@
 import time
-
 import pytest
 from sqlalchemy import text
+from common import http_requests
 
-def test_query_stat_statements(sess):
-    """Check that the background worker doesn't execute queries when no new requests arrive"""
 
+def skip_test_if_pg_is_old(sess):
     (pg_version,) = sess.execute(text(
         """
         select current_setting('server_version_num');
@@ -13,8 +12,11 @@ def test_query_stat_statements(sess):
     )).fetchone()
 
     if int(pg_version) < 140000:
-        pytest.skip("Skipping fixture on pg version < 14. The query_id column on pg_stat_statements is only available on >= 14")
+        pytest.skip(
+            "Skipping fixture on pg version < 14. The query_id column on pg_stat_statements is only available on >= 14")
+
 
+def create_pg_stat_statement(sess):
     sess.execute(text(
         """
         create extension pg_stat_statements;
@@ -23,22 +25,20 @@ def test_query_stat_statements(sess):
 
     sess.commit()
 
-    time.sleep(1)
 
-    (old_calls,) = sess.execute(text(
+def drop_pg_stat_statement(sess):
+    sess.execute(text(
         """
-        select coalesce(sum(calls), 0)
-        from pg_stat_statements
-        where
-            query ilike '%DELETE FROM net._http_response r %' or
-            query ilike '%DELETE FROM net.http_request_queue%';
+        select pg_stat_statements_reset();
+        drop extension pg_stat_statements;
     """
-    )).fetchone()
+    ))
+
+    sess.commit()
 
-    # sleep for some time to see if new queries arrive
-    time.sleep(3)
 
-    (new_calls,) = sess.execute(text(
+def get_worker_query_count(sess):
+    (count,) = sess.execute(text(
         """
         select coalesce(sum(calls), 0)
         from pg_stat_statements
@@ -48,76 +48,62 @@ def test_query_stat_statements(sess):
     """
     )).fetchone()
 
-    assert new_calls == old_calls
+    return count
 
-    sess.execute(text(
-        """
-        select pg_stat_statements_reset();
-        drop extension pg_stat_statements;
+
+def test_query_stat_statements(sess):
+    """
+    Check that the background worker doesn't execute
+    queries when no new requests arrive
     """
-    ))
 
-    sess.commit()
+    skip_test_if_pg_is_old(sess)
 
+    create_pg_stat_statement(sess)
 
-def test_wakes_at_commit_time(sess):
-    """Check that the background worker only does one wake at commit time, avoiding unnecessary wakes and work"""
+    old_calls = get_worker_query_count(sess)
 
-    (pg_version,) = sess.execute(text(
-        """
-        select current_setting('server_version_num');
-    """
-    )).fetchone()
+    # sleep for some time to see if new queries arrive
+    time.sleep(3)
 
-    if int(pg_version) < 140000:
-        pytest.skip("Skipping fixture on pg version < 14. The query_id column on pg_stat_statements is only available on >= 14")
+    new_calls = get_worker_query_count(sess)
 
-    sess.execute(text(
-        """
-        create extension pg_stat_statements;
+    assert new_calls == old_calls
+
+    drop_pg_stat_statement(sess)
+
+
+def test_wakes_at_commit_time(sess):
+    """
+    Check that the background worker only does one wake at
+    commit time, avoiding unnecessary wakes and work
     """
-    ))
 
-    sess.commit()
+    skip_test_if_pg_is_old(sess)
+
+    create_pg_stat_statement(sess)
 
     # wait for initial queries
     time.sleep(1)
 
-    (initial_calls,) = sess.execute(text(
-        """
-        select coalesce(sum(calls), 0)
-        from pg_stat_statements
-        where
-            query ilike '%DELETE FROM net._http_response r %' or
-            query ilike '%DELETE FROM net.http_request_queue%';
-    """
-    )).fetchone()
+    initial_calls = get_worker_query_count(sess)
 
     assert initial_calls >= 0
 
-    sess.execute(text(
+    http_requests(sess, text(
         """
         select net.http_get('http://localhost:8080/pathological?status=200') from generate_series(1,100);
     """
     ))
 
-    sess.commit()
-
     # wait for reqs
     time.sleep(2)
 
-    (commit_calls,) = sess.execute(text(
-        """
-        select coalesce(sum(calls), 0)
-        from pg_stat_statements
-        where
-            query ilike '%DELETE FROM net._http_response r %' or
-            query ilike '%DELETE FROM net.http_request_queue%';
-    """
-    )).fetchone()
+    commit_calls = get_worker_query_count(sess)
 
-    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
+    # 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
+    assert commit_calls == initial_calls + 4
 
     # if the new requests are rollbacked/aborted, then no new queries will be made by the bg worker
     sess.execute(text(
@@ -131,23 +117,8 @@ def test_wakes_at_commit_time(sess):
     # wait for requests
     time.sleep(2)
 
-    (rollback_calls,) = sess.execute(text(
-        """
-        select coalesce(sum(calls), 0)
-        from pg_stat_statements
-        where
-            query ilike '%DELETE FROM net._http_response r %' or
-            query ilike '%DELETE FROM net.http_request_queue%';
-    """
-    )).fetchone()
+    rollback_calls = get_worker_query_count(sess)
 
     assert rollback_calls == commit_calls
 
-    sess.execute(text(
-        """
-        select pg_stat_statements_reset();
-        drop extension pg_stat_statements;
-    """
-    ))
-
-    sess.commit()
+    drop_pg_stat_statement(sess)
diff --git a/test/test_user_db.py b/test/test_user_db.py
index faf06c3a..421cac21 100644
--- a/test/test_user_db.py
+++ b/test/test_user_db.py
@@ -1,30 +1,33 @@
-import time
-
-import pytest
 from sqlalchemy import text
+from common import restart_worker
+
 
 def test_net_with_different_username_dbname(sess, autocommit_sess):
     """Check that a pre existing role can use the net schema"""
 
-    autocommit_sess.execute(text("alter system set pg_net.username to 'pre_existing'"))
-    autocommit_sess.execute(text("alter system set pg_net.database_name to 'pre_existing'"))
-    autocommit_sess.execute(text("select net.worker_restart()"))
-    autocommit_sess.execute(text("select net.wait_until_running()"))
+    try:
+        autocommit_sess.execute(
+            text("alter system set pg_net.username to 'pre_existing'"))
+        autocommit_sess.execute(
+            text("alter system set pg_net.database_name to 'pre_existing'"))
+        restart_worker(autocommit_sess)
 
-    (username,datname) = sess.execute(
-        text(
-            """
-        select usename, datname from pg_stat_activity where backend_type ilike '%pg_net%';
-    """
-        )
-    ).fetchone()
-    assert username == 'pre_existing'
-    assert datname == 'pre_existing'
+        (username, datname) = sess.execute(
+            text(
+                """
+            select usename, datname from pg_stat_activity where backend_type ilike '%pg_net%';
+        """
+            )
+        ).fetchone()
+        assert username == 'pre_existing'
+        assert datname == 'pre_existing'
+
+    finally:
+        autocommit_sess.execute(text("alter system reset pg_net.username"))
+        autocommit_sess.execute(
+            text("alter system reset pg_net.database_name"))
+        restart_worker(autocommit_sess)
 
-    autocommit_sess.execute(text("alter system reset pg_net.username"))
-    autocommit_sess.execute(text("alter system reset pg_net.database_name"))
-    autocommit_sess.execute(text("select net.worker_restart()"))
-    autocommit_sess.execute(text("select net.wait_until_running()"))
 
 def test_net_appname(sess):
     """Check that pg_stat_activity has appname set"""
diff --git a/test/test_worker_behavior.py b/test/test_worker_behavior.py
index 93bcab5c..c2f8e3f9 100644
--- a/test/test_worker_behavior.py
+++ b/test/test_worker_behavior.py
@@ -1,14 +1,17 @@
-from sqlalchemy import create_engine
+from sqlalchemy import create_engine, text
 from sqlalchemy.orm import Session
-from sqlalchemy import text
-import sqlalchemy as sa
-import pytest
 import time
 import subprocess
 import os
+from common import http_request, http_requests, restart_worker, wait_for_any_response
+from common import wait_for_extension_drop, wait_for_postgres_ready
+from common import wait_for_queue_drain, wait_for_response_count
+from common import wait_for_worker_down, wait_for_worker_state
+from common import wait_for_worker_up, wait_until, wakeup_worker
+
 
 def test_worker_will_not_block_drop_database(autocommit_sess):
-    """the worker will not block a session doing drop database"""
+    """Check that the worker will not block a session doing drop database"""
 
     autocommit_sess.execute(text("create database foo;"))
     autocommit_sess.execute(text("drop database foo;"))
@@ -51,169 +54,153 @@ def test_success_when_worker_is_up(sess):
     assert result == ''
 
 
-def test_worker_will_process_queue_when_up(sess):
-    """when pg background worker is down and requests arrive, it will process them once it wakes up"""
+def test_worker_will_process_queue_when_up(sess, autocommit_sess):
+    """
+    Check that when pg background worker is down and requests arrive,
+    it will process them once it wakes up
+    """
 
-    # check worker up
-    (up,) = sess.execute(text("""
+    # Assert that background worker is worker up
+    (worker_is_up,) = sess.execute(text("""
         select is_worker_up();
     """)).fetchone()
-    assert up is not None
-    assert up == True
+    assert worker_is_up
 
-    # restart it
-    (restarted,) = sess.execute(text("""
+    # kill background worker
+    (killed,) = sess.execute(text("""
         select public.kill_worker();
     """)).fetchone()
-    assert restarted is not None
-    assert restarted == True
+    assert killed is not None
+    assert killed == True
 
-    time.sleep(0.1)
+    # Wait for background worker to go down
+    wait_for_worker_down(autocommit_sess)
 
-    # check worker down
-    (up,) = sess.execute(text("""
-        select is_worker_up();
-    """)).fetchone()
-    assert up is not None
-    assert up == False
-
-    sess.execute(text(
+    # Make a request while the worker is down
+    http_requests(sess, text(
         """
         select net.http_get('http://localhost:8080/pathological?status=200') from generate_series(1,10);
     """
-    )).fetchone()
-
-    sess.commit()
+    ))
 
-    # check requests where enqueued
+    # Check that requests were enqueued
     (count,) = sess.execute(text(
-    """
+        """
         select count(*) from net.http_request_queue;
     """
     )).fetchone()
-
     assert count == 10
 
-    # check worker is still down
-    (up,) = sess.execute(text("""
+    # Check that worker is still down. Note that this is a bit racy
+    # as there is no guarantee that worker hasn't come back up at this
+    # time, but in practice this rarely fails because worker takes 2
+    # seconds before coming back up, which is more than enough time
+    # to reach here.
+    (worker_is_up,) = sess.execute(text("""
         select is_worker_up();
     """)).fetchone()
-    assert up is not None
-    assert up == False
+    assert not worker_is_up
 
     sess.commit()
 
-    # wait until up
-    time.sleep(2.1)
+    # Wait for background worker to come back up
+    # It's critical to use autocommit_sess to see a new snapshot
+    # on each retry in wait_until, otherwise it might keep reading
+    # stale data and fail with a timeout.
+    wait_for_worker_up(autocommit_sess)
 
-    # check worker up
-    (up,) = sess.execute(text("""
-        select is_worker_up();
-    """)).fetchone()
-    assert up is not None
-    assert up == True
+    # Wait for request queue to drain
+    wait_for_queue_drain(autocommit_sess)
 
-    # wait until new requests are done
-    time.sleep(1.1)
+    # Wait until all responses have arrived
+    wait_for_response_count(autocommit_sess, 10)
 
-    (count,) = sess.execute(text(
-    """
-        select count(*) from net.http_request_queue;
-    """
-    )).fetchone()
-
-    assert count == 0
 
-    (status_code,count) = sess.execute(text(
+def test_can_delete_rows_while_processing_queue(sess, autocommit_sess):
     """
-        select status_code, count(*) from net._http_response group by status_code;
+    Check that a user can delete the queue rows while the worker is
+    processing them
     """
-    )).fetchone()
-
-    assert status_code == 200
-    assert count == 10
-
 
-def test_can_delete_rows_while_processing_queue(sess, autocommit_sess):
-    """user can delete the queue rows while the worker is processing them"""
+    try:
+        autocommit_sess.execute(
+            text("alter system set pg_net.batch_size to '1';"))
+        restart_worker(autocommit_sess)
 
-    autocommit_sess.execute(text("alter system set pg_net.batch_size to '1';"))
-    autocommit_sess.execute(text("select net.worker_restart();"))
-    autocommit_sess.execute(text("select net.wait_until_running();"))
-
-    sess.execute(text(
+        http_requests(sess, 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);
-    """
-    ))
+        ))
 
-    sess.commit()
-
-    # leave time for some processing
-    time.sleep(0.1)
+        # Wait until responses have started arriving
+        wait_for_any_response(autocommit_sess)
 
-    (count,) = sess.execute(text(
+        (count,) = sess.execute(text(
+            """
+            with deleted as (delete from net.http_request_queue returning *) select count(*) from deleted;
         """
-        WITH deleted AS (DELETE FROM net.http_request_queue RETURNING *) SELECT count(*) FROM deleted;
-    """
-    )).fetchone()
-    assert count > 1
-
-    sess.commit()
+        )).fetchone()
+        assert count > 1
 
-    autocommit_sess.execute(text("alter system reset pg_net.batch_size"))
-    autocommit_sess.execute(text("select net.worker_restart()"))
-    autocommit_sess.execute(text("select net.wait_until_running()"))
+        sess.commit()
+    finally:
+        autocommit_sess.execute(text("alter system reset pg_net.batch_size"))
+        restart_worker(autocommit_sess)
 
 
 def test_truncate_wait_while_processing_queue(sess, autocommit_sess):
-    """a truncate will not wait until the worker is done processing all requests"""
+    """
+    Check that a truncate will not wait until the worker
+    is done processing all requests
+    """
 
-    # ensure the worker will be processing the queue 1 by 1 (slowly) so it doesn't clear the whole
-    # net.http_request_queue in one go
-    autocommit_sess.execute(text("alter system set pg_net.batch_size to '1';"))
-    autocommit_sess.execute(text("select net.worker_restart();"))
-    autocommit_sess.execute(text("select net.wait_until_running();"))
+    try:
+        # ensure the worker will be processing the queue 1 by 1 (slowly) so it doesn't clear the whole
+        # net.http_request_queue in one go
+        autocommit_sess.execute(
+            text("alter system set pg_net.batch_size to '1';"))
+        restart_worker(autocommit_sess)
 
-    sess.execute(text(
+        http_requests(sess, 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);
-    """
-    ))
-    sess.commit()
+        ))
 
-    # truncate succeeds fast, despite the worker still processing the queue 1 by 1
-    sess.execute(text(
+        # truncate succeeds fast, despite the worker still processing the queue 1 by 1
+        sess.execute(text(
+            """
+            truncate net.http_request_queue;
         """
-        truncate net.http_request_queue;
-    """
-    ))
+        ))
 
-    # now the queue will be empty
-    (count,) = sess.execute(text(
+        # now the queue will be empty
+        (count,) = sess.execute(text(
+            """
+            select count(*) from net.http_request_queue;
         """
-        select count(*) from net.http_request_queue;
-    """
-    )).fetchone()
-    assert count == 0
-
-    autocommit_sess.execute(text("alter system reset pg_net.batch_size"))
-    autocommit_sess.execute(text("select net.worker_restart()"))
-    autocommit_sess.execute(text("select net.wait_until_running()"))
-
+        )).fetchone()
+        assert count == 0
+    finally:
+        autocommit_sess.execute(text("alter system reset pg_net.batch_size"))
+        restart_worker(autocommit_sess)
 
-def test_no_failure_on_drop_extension(sess):
-    """while waiting for a slow request, a drop extension should wait and not crash the worker"""
 
-    (request_id,) = sess.execute(text("""
-        select net.http_get(url := 'http://localhost:8080/pathological?status=200&delay=2');
-    """)).fetchone()
-    assert request_id == 1
+def test_no_failure_on_drop_extension(sess, autocommit_sess):
+    """
+    Check that while waiting for a slow request, a drop extension should
+    wait and not crash the worker
+    """
 
-    sess.commit()
+    http_requests(sess, text(
+        """
+        select net.http_get('http://localhost:8080/pathological?status=200&delay=2') from generate_series(1,10);
+    """
+    ))
 
-    # wait until processing
-    time.sleep(1)
+    # Wait until responses have started arriving
+    wait_for_any_response(autocommit_sess)
 
     sess.execute(text("""
         drop extension pg_net cascade;
@@ -221,9 +208,10 @@ def test_no_failure_on_drop_extension(sess):
 
     sess.commit()
 
-    # wait until request is finished
-    time.sleep(3)
+    # wait until the extension is fully gone
+    wait_for_extension_drop(autocommit_sess)
 
+    # The background worker should not have crash even after dropping the extension
     (up,) = sess.execute(text("""
         select is_worker_up();
     """)).fetchone()
@@ -232,94 +220,80 @@ def test_no_failure_on_drop_extension(sess):
 
 
 def test_worker_will_keep_processing_queue_when_restarted(sess, autocommit_sess):
-    """when the background worker is restarted while working, it will pick up the remaining requests"""
-
-    autocommit_sess.execute(text("alter system set pg_net.batch_size to '1';"))
-    autocommit_sess.execute(text("select net.worker_restart();"))
-    autocommit_sess.execute(text("select net.wait_until_running();"))
-
-    sess.execute(text(
-        """
-        select net.http_get('http://localhost:8080/pathological?status=200') from generate_series(1,5);
     """
-    ))
-
-    sess.commit()
-
-    # one restart will likely keep the worker awake since the wake signal could still be on, so do two restarts
-    # to ensure the wake signal is cleared
-    sess.execute(text(
-        """
-        select net.worker_restart();
-        select net.wait_until_running();
+    Check that when the background worker is restarted while working,
+    it will pick up the remaining requests
     """
-    ))
 
-    time.sleep(0.1)
+    try:
+        autocommit_sess.execute(
+            text("alter system set pg_net.batch_size to '1';"))
+        restart_worker(autocommit_sess)
 
-    sess.execute(text(
+        http_requests(sess, text(
+            """
+            select net.http_get('http://localhost:8080/pathological?status=200&delay=1') from generate_series(1,5);
         """
-        select net.worker_restart();
-        select net.wait_until_running();
-    """
-    ))
-
-    time.sleep(0.1)
-
-    (status_code,count) = sess.execute(text(
-    """
-        select status_code, count(*) from net._http_response group by status_code;
-    """
-    )).fetchone()
-
-    # at most 2 requests should have finished by now because of the low batch_size
-    assert count <= 2
-    assert count > 0 # at least 1 request should be finished
-    assert status_code == 200
-
-    # if we sleep for 4 seconds the whole 5 requests should be finished
-    time.sleep(4)
-
-    (status_code,count) = sess.execute(text(
-    """
-        select status_code, count(*) from net._http_response group by status_code;
-    """
-    )).fetchone()
-
-    assert status_code == 200
-    assert count == 5
-
-    autocommit_sess.execute(text("alter system reset pg_net.batch_size"))
-    autocommit_sess.execute(text("select net.worker_restart()"))
-    autocommit_sess.execute(text("select net.wait_until_running()"))
-
-
-def test_new_requests_get_attended_asap(sess):
-    """new requests get attended as soon as possible"""
-
-    sess.execute(text(
+        ))
+
+        # Wait until responses have started arriving
+        (processed,) = wait_until(
+            fetch=lambda: autocommit_sess.execute(text("""
+                select count(*) from net._http_response;
+            """)).fetchone(),
+            predicate=lambda result: result[0] > 0,
+            timeout=5,
+            sleep_interval=0.1,
+            description="responses to arrive before first restart",
+        )
+
+        restart_worker(autocommit_sess)
+
+        # Check that more requests are processed after a restart
+        wait_until(
+            fetch=lambda: autocommit_sess.execute(text("""
+                select count(*) from net._http_response;
+            """)).fetchone(),
+            predicate=lambda result: result[0] >= processed,
+            timeout=5,
+            sleep_interval=0.1,
+            description="responses to arrive after first restart",
+        )
+
+        restart_worker(autocommit_sess)
+
+        # And now wait until all responses have arrived
+        wait_for_response_count(autocommit_sess, 5)
+
+    finally:
+        autocommit_sess.execute(text("alter system reset pg_net.batch_size"))
+        restart_worker(autocommit_sess)
+
+
+def test_new_requests_get_attended_without_explicit_wakeup(sess, autocommit_sess):
+    """Check that new requests get attended without an explicit wakeup"""
+
+    http_requests(sess, text(
         """
         select net.http_get('http://localhost:8080/pathological?status=200') from generate_series(1,10);
     """
     ))
 
-    sess.commit()
+    # wait until all responses have arrived
+    wait_for_response_count(autocommit_sess, 10)
 
-    # less than a second
-    time.sleep(0.1)
 
-    (status_code,count) = sess.execute(text(
+def test_direct_inserts_no_requests(sess, autocommit_sess):
     """
-        select status_code, count(*) from net._http_response group by status_code;
+    Check that direct insertions to the net.http_request_queue doesn't
+    trigger new requests
     """
-    )).fetchone()
-
-    assert status_code == 200
-    assert count == 10
-
 
-def test_direct_inserts_no_requests(sess):
-    """direct insertions to the net.http_request_queue doesn't trigger new requests"""
+    # Make sure the worker has already settled into its idle wait before we
+    # insert. If a prior test left it mid-batch, its trailing WORKER_WAIT_ONE_SECOND
+    # recheck (worker.c) can pick up this test's direct insert on its own,
+    # with no net.wake() involved, and make this test flake.
+    wait_for_worker_state(autocommit_sess, 'idle')
 
     sess.execute(text(
         """
@@ -335,117 +309,99 @@ def test_direct_inserts_no_requests(sess):
 
     sess.commit()
 
-    # wait for req
-    time.sleep(0.1)
+    # Even waiting for 5 seconds doesn't process the request
+    time.sleep(5)
 
-    # no response
+    # No response still
     (count,) = sess.execute(text(
-    """
+        """
         select count(*) from net._http_response;
     """
     )).fetchone()
 
     assert count == 0
 
-    # req still in queue
+    # Reqest is still in queue
     (count,) = sess.execute(text(
-    """
+        """
         select count(*) from net.http_request_queue;
     """
     )).fetchone()
 
     assert count == 1
 
-    # an explicit wake will make it serve requests though
+    # An explicit wake will make it serve requests though
+    wakeup_worker(sess)
 
-    sess.execute(text(
-    """
-        select net.wake();
-    """
-    ))
+    # wait until the response has arrived
+    wait_for_response_count(autocommit_sess, 1)
 
-    sess.commit()
 
-    # wait for req
-    time.sleep(0.1)
-
-    (status_code, count) = sess.execute(text(
+def test_processing_survives_postmaster_crash(autocommit_sess):
     """
-        select status_code, count(*) from net._http_response group by status_code;
+    Check that the queue will continue processing even when a postmaster
+    crash or restart happens
     """
-    )).fetchone()
-
-    assert status_code == 200
-    assert count == 1
-
-
-def test_processing_survives_postmaster_crash():
-    """the queue will continue processing even when a postmaster crash or restart happens"""
 
     engine = create_engine("postgresql:///postgres")
     ac_engine = engine.execution_options(isolation_level="AUTOCOMMIT")
     tmp_sess = Session(ac_engine)
 
-    tmp_sess.execute(text("create extension if not exists pg_net;"))
+    try:
+        tmp_sess.execute(text("create extension if not exists pg_net;"))
 
-    tmp_sess.execute(text("alter system set pg_net.batch_size to '5';"))
-    tmp_sess.execute(text("select net.worker_restart();"))
-    tmp_sess.execute(text("select net.wait_until_running();"))
+        tmp_sess.execute(text("alter system set pg_net.batch_size to '5';"))
+        restart_worker(tmp_sess)
 
-    tmp_sess.execute(text(
+        tmp_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);
-    """
-    )).fetchone()
+        )).fetchone()
 
-    (count,) = tmp_sess.execute(text(
-    """
-        select count(*) from net.http_request_queue;
-    """
-    )).fetchone()
-    assert count == 10
+        (count,) = tmp_sess.execute(text(
+            """
+            select count(*) from net.http_request_queue;
+        """
+        )).fetchone()
+        assert count == 10
 
-    engine.dispose()
+        engine.dispose()
 
-    pgdata_env = os.getenv('PGDATA')
-    subprocess.run(["pg_ctl", "restart", "-D", pgdata_env])
+        pgdata_env = os.getenv('PGDATA')
+        subprocess.run(["pg_ctl", "restart", "-D", pgdata_env])
 
-    # give it some time to finish restart
-    time.sleep(1)
+        # wait for postmaster to finish restarting and accept connections
+        wait_for_postgres_ready(engine, tmp_sess)
 
-    engine = create_engine("postgresql:///postgres")
-    ac_engine = engine.execution_options(isolation_level="AUTOCOMMIT")
-    tmp_sess = Session(ac_engine)
+        # Recreate engine and session after restart
+        engine = create_engine("postgresql:///postgres")
+        ac_engine = engine.execution_options(isolation_level="AUTOCOMMIT")
+        tmp_sess = Session(ac_engine)
 
-    # give it enough time to finish processing the queue
-    time.sleep(1)
+        # wait until the queue has finished processing
+        wait_for_queue_drain(autocommit_sess)
 
-    (count,) = tmp_sess.execute(text(
-    """
-        select count(*) from net.http_request_queue;
-    """
-    )).fetchone()
-    assert count == 0
-
-    (status_code,count) = tmp_sess.execute(text(
-    """
-        select status_code, count(*) from net._http_response group by status_code;
-    """
-    )).fetchone()
+        (status_code, count) = tmp_sess.execute(text(
+            """
+            select status_code, count(*) from net._http_response group by status_code;
+        """
+        )).fetchone()
 
-    assert status_code == 200
-    assert count == 10
+        assert status_code == 200
+        assert count == 10
 
-    tmp_sess.execute(text("alter system reset pg_net.batch_size"))
-    tmp_sess.execute(text("select net.worker_restart()"))
-    tmp_sess.execute(text("select net.wait_until_running()"))
+    finally:
+        tmp_sess.execute(text("alter system reset pg_net.batch_size"))
+        restart_worker(autocommit_sess)
 
-    engine.dispose()
+        engine.dispose()
 
 
 def test_worker_writes_increment_pgstat_counters(sess, autocommit_sess):
-    """the worker's INSERTs into net._http_response must be reflected in
-    pg_stat_user_tables. Without this, autovacuum/autoanalyze can never be
+    """
+    Check that the worker's INSERTs into net._http_response must be reflected
+    in pg_stat_user_tables. Without this, autovacuum/autoanalyze can never be
     scheduled and the table silently bloats.
     """
 
@@ -459,22 +415,23 @@ def test_worker_writes_increment_pgstat_counters(sess, autocommit_sess):
     ))
 
     # Drive a batch of requests through the worker.
-    sess.execute(text("""
+    http_requests(sess, text("""
         select net.http_get('http://localhost:8080/pathological?status=200')
         from generate_series(1,30);
     """))
-    sess.commit()
 
     # Wait until the worker has actually drained the queue and written all
     # responses to net._http_response. Don't assume "30 rows" - the worker
     # may pick up the queue in chunks depending on wake() coalescing.
-    for _ in range(20):
-        time.sleep(0.5)
-        (queue_count,) = sess.execute(text(
+    wait_until(
+        fetch=lambda: sess.execute(text(
             "select count(*) from net.http_request_queue;"
-        )).fetchone()
-        if queue_count == 0:
-            break
+        )).fetchone(),
+        predicate=lambda result: result[0] == 0,
+        timeout=10,
+        sleep_interval=0.5,
+        description="worker to drain the request queue",
+    )
 
     # Confirm the worker actually wrote rows, otherwise the pgstat assertion
     # below would be meaningless.
@@ -488,17 +445,16 @@ def test_worker_writes_increment_pgstat_counters(sess, autocommit_sess):
     # worker's first flush attempt is normally a hit (last_flush is far in
     # the past after a long idle), but we allow generous slack here so an
     # off-by-a-tick scheduling doesn't flake the suite.
-    deadline = time.time() + 30.0
-    resp_ins = 0
-    resp_mod = 0
-    while time.time() < deadline:
-        (resp_ins, resp_mod) = autocommit_sess.execute(text("""
+    (resp_ins, resp_mod) = wait_until(
+        fetch=lambda: autocommit_sess.execute(text("""
             select n_tup_ins, n_mod_since_analyze
             from pg_stat_user_tables where relname='_http_response';
-        """)).fetchone()
-        if resp_ins > 0:
-            break
-        time.sleep(0.5)
+        """)).fetchone(),
+        predicate=lambda result: result[0] > 0,
+        timeout=30,
+        sleep_interval=0.5,
+        description="net._http_response pgstat counters to reflect worker INSERTs",
+    )
 
     assert resp_ins > 0, (
         f"net._http_response.n_tup_ins is still 0 after 30s. "
@@ -513,85 +469,98 @@ def test_worker_writes_increment_pgstat_counters(sess, autocommit_sess):
 
 
 def test_worker_writes_trigger_autoanalyze_on_http_response(sess, autocommit_sess):
-    """autoanalyze on net._http_response must fire after the worker writes
-    enough rows. Without working pgstat counters, autovacuum/autoanalyze
+    """
+    Check that autoanalyze on net._http_response must fire after the worker
+    writes enough rows. Without working pgstat counters, autovacuum/autoanalyze
     never get scheduled and the table bloats - this is the primary symptom
     seen on production (slow expiry DELETEs from a bloated index).
     """
 
-    # Make sure the worker is fully up before we start.
-    autocommit_sess.execute(text("select net.wait_until_running();"))
-
-    # Make autovacuum eager *before* generating traffic so the launcher is
-    # already running on a 1s naptime by the time stats threshold is crossed.
-    # autovacuum_naptime is PGC_SIGHUP (reloadable). Give the reload a moment
-    # to propagate to the launcher.
-    autocommit_sess.execute(text("alter system set autovacuum_naptime = '1s';"))
-    autocommit_sess.execute(text("select pg_reload_conf();"))
-    time.sleep(1)
-
-    # Per-table: trip the autoanalyze threshold after a handful of rows.
-    # Reloptions take effect immediately; no reload required.
-    autocommit_sess.execute(text("""
-        alter table net._http_response set (
-            autovacuum_analyze_threshold = 10,
-            autovacuum_analyze_scale_factor = 0,
-            autovacuum_vacuum_threshold = 10,
-            autovacuum_vacuum_scale_factor = 0
-        );
-    """))
-
-    autocommit_sess.execute(text(
-        "select pg_stat_reset_single_table_counters('net._http_response'::regclass);"
-    ))
-
-    # Drive 30 inserts through the worker. 30 is well above the threshold (10).
-    sess.execute(text("""
-        select net.http_get('http://localhost:8080/pathological?status=200')
-        from generate_series(1,30);
-    """))
-    sess.commit()
-
-    # 30s budget covers worst-case worker pgstat flush (PGSTAT_MIN_INTERVAL
-    # = 1s slack) + worst-case launcher cycle (autovacuum_max_workers=3,
-    # 3 databases at 1s naptime each ~= 3s/cycle, with 2-3 cycle slack) +
-    # autoanalyze worker spawn + ANALYZE on a tiny table (sub-second).
-    # Real wall time on a clean rig is typically ~2-5s; the slack is to
-    # absorb test-rig load and not flake.
-    deadline = time.time() + 30.0
-    autoanalyze_count = 0
-    while time.time() < deadline:
-        (autoanalyze_count,) = autocommit_sess.execute(text("""
-            select autoanalyze_count
-            from pg_stat_user_tables where relname='_http_response';
-        """)).fetchone()
-        if autoanalyze_count > 0:
-            break
-        time.sleep(0.5)
-
-    assert autoanalyze_count > 0, (
-        "autoanalyze never fired on net._http_response within 30s. "
-        "Worker writes are not making pgstat threshold visible to the "
-        "autovacuum launcher - the customer-facing symptom (silent bloat) "
-        "would manifest in production."
-    )
-
-    # Cleanup: restore defaults so we don't bleed into other tests.
-    autocommit_sess.execute(text("""
-        alter table net._http_response reset (
-            autovacuum_analyze_threshold,
-            autovacuum_analyze_scale_factor,
-            autovacuum_vacuum_threshold,
-            autovacuum_vacuum_scale_factor
-        );
-    """))
-    autocommit_sess.execute(text("alter system reset autovacuum_naptime;"))
-    autocommit_sess.execute(text("select pg_reload_conf();"))
+    try:
+        # Make sure the worker is fully up before we start.
+        autocommit_sess.execute(text("select net.wait_until_running();"))
+
+        # Make autovacuum eager *before* generating traffic so the launcher is
+        # already running on a 1s naptime by the time stats threshold is crossed.
+        # autovacuum_naptime is PGC_SIGHUP (reloadable). Give the reload a moment
+        # to propagate to the launcher.
+        autocommit_sess.execute(
+            text("alter system set autovacuum_naptime = '1s';"))
+        autocommit_sess.execute(text("select pg_reload_conf();"))
+
+        wait_until(
+            fetch=lambda: autocommit_sess.execute(text(
+                "select current_setting('autovacuum_naptime');"
+            )).fetchone(),
+            predicate=lambda result: result[0] == '1s',
+            timeout=5,
+            sleep_interval=0.1,
+            description="autovacuum_naptime reload to propagate",
+        )
+
+        # Per-table: trip the autoanalyze threshold after a handful of rows.
+        # Reloptions take effect immediately; no reload required.
+        autocommit_sess.execute(text("""
+            alter table net._http_response set (
+                autovacuum_analyze_threshold = 10,
+                autovacuum_analyze_scale_factor = 0,
+                autovacuum_vacuum_threshold = 10,
+                autovacuum_vacuum_scale_factor = 0
+            );
+        """))
+
+        autocommit_sess.execute(text(
+            "select pg_stat_reset_single_table_counters('net._http_response'::regclass);"
+        ))
+
+        # Drive 30 inserts through the worker. 30 is well above the threshold (10).
+        http_requests(sess, text("""
+            select net.http_get('http://localhost:8080/pathological?status=200')
+            from generate_series(1,30);
+        """))
+
+        # 30s budget covers worst-case worker pgstat flush (PGSTAT_MIN_INTERVAL
+        # = 1s slack) + worst-case launcher cycle (autovacuum_max_workers=3,
+        # 3 databases at 1s naptime each ~= 3s/cycle, with 2-3 cycle slack) +
+        # autoanalyze worker spawn + ANALYZE on a tiny table (sub-second).
+        # Real wall time on a clean rig is typically ~2-5s; the slack is to
+        # absorb test-rig load and not flake.
+        (autoanalyze_count,) = wait_until(
+            fetch=lambda: autocommit_sess.execute(text("""
+                select autoanalyze_count
+                from pg_stat_user_tables where relname='_http_response';
+            """)).fetchone(),
+            predicate=lambda result: result[0] > 0,
+            timeout=30,
+            sleep_interval=0.5,
+            description="autoanalyze to fire on net._http_response",
+        )
+
+        assert autoanalyze_count > 0, (
+            "autoanalyze never fired on net._http_response within 30s. "
+            "Worker writes are not making pgstat threshold visible to the "
+            "autovacuum launcher - the customer-facing symptom (silent bloat) "
+            "would manifest in production."
+        )
+
+    finally:
+        # Cleanup: restore defaults so we don't bleed into other tests.
+        autocommit_sess.execute(text("""
+            alter table net._http_response reset (
+                autovacuum_analyze_threshold,
+                autovacuum_analyze_scale_factor,
+                autovacuum_vacuum_threshold,
+                autovacuum_vacuum_scale_factor
+            );
+        """))
+        autocommit_sess.execute(text("alter system reset autovacuum_naptime;"))
+        autocommit_sess.execute(text("select pg_reload_conf();"))
 
 
 def test_worker_reports_activity_in_pg_stat_activity(sess, autocommit_sess):
-    """the pg_net worker must call pgstat_report_activity() so its row in
-    pg_stat_activity has a valid state column.
+    """
+    Check that the pg_net worker must call pgstat_report_activity() so
+    its row in pg_stat_activity has a valid state column.
     """
 
     autocommit_sess.execute(text("select net.wait_until_running();"))
@@ -599,39 +568,21 @@ def test_worker_reports_activity_in_pg_stat_activity(sess, autocommit_sess):
     # Wait for the worker to drain any leftover work from previous tests
     # and settle into idle. Polling makes this robust regardless of what
     # ran before.
-    deadline = time.time() + 5.0
-    state = None
-    while time.time() < deadline:
-        (state,) = autocommit_sess.execute(text(
-            "select state from pg_stat_activity where backend_type ilike '%pg_net%';"
-        )).fetchone()
-        if state == 'idle':
-            break
-        time.sleep(0.1)
-    assert state == 'idle', (
-        f"pg_net worker state expected 'idle' at rest, got {state!r}. "
-        "Without pgstat_report_activity(STATE_IDLE, ...) the state column "
-        "stays NULL."
-    )
+    wait_for_worker_state(autocommit_sess, 'idle')
 
     # Fire a slow request so the worker stays active long enough to observe.
-    sess.execute(text("""
+    http_requests(sess, text("""
         select net.http_get('http://localhost:8080/pathological?status=200&delay=2');
     """))
-    sess.commit()
 
     # Poll for 'active' for up to 5s. The slow request keeps the worker
     # busy for ~2s, so we have a wide observation window.
-    deadline = time.time() + 5.0
     saw_active = False
-    while time.time() < deadline:
-        (state,) = autocommit_sess.execute(text(
-            "select state from pg_stat_activity where backend_type ilike '%pg_net%';"
-        )).fetchone()
-        if state == 'active':
-            saw_active = True
-            break
-        time.sleep(0.1)
+    try:
+        wait_for_worker_state(autocommit_sess, 'active')
+        saw_active = True
+    except AssertionError:
+        pass
 
     assert saw_active, (
         "pg_net worker state was never observed as 'active' during a slow "
@@ -640,11 +591,12 @@ def test_worker_reports_activity_in_pg_stat_activity(sess, autocommit_sess):
     )
 
 
-
 def test_worker_idles_when_net_schema_exists_without_extension(sess, autocommit_sess):
-    """when a schema named "net" exists but the pg_net tables don't (e.g. another
-    extension installed into a schema named "net"), the worker should treat the
-    extension as not installed instead of crash looping"""
+    """
+    Check that when a schema named "net" exists but the pg_net tables don't
+    (e.g. another extension installed into a schema named "net"), the worker
+    should treat the extension as not installed instead of crash looping
+    """
 
     sess.execute(text("drop extension pg_net cascade;"))
     sess.execute(text("create schema net;"))
@@ -654,26 +606,41 @@ def test_worker_idles_when_net_schema_exists_without_extension(sess, autocommit_
     autocommit_sess.execute(text("select kill_worker();"))
 
     # wait for the worker to come back up (bgw_restart_time is 1 second)
-    pid = None
-    deadline = time.time() + 5.0
-    while time.time() < deadline:
-        row = autocommit_sess.execute(text(
+    (pid,) = wait_until(
+        fetch=lambda: autocommit_sess.execute(text(
             "select pid from pg_stat_activity where backend_type ilike '%pg_net%';"
-        )).fetchone()
-        if row:
-            pid = row[0]
-            break
-        time.sleep(0.1)
-    assert pid is not None, "pg_net worker did not come back up after restart"
+        )).fetchone(),
+        predicate=lambda result: result,
+        timeout=5,
+        sleep_interval=0.1,
+        description="pg_net worker to to be back up",
+    )
 
-    # wait several restart cycles; a crash loop would respawn the worker with a new pid
-    time.sleep(3)
+    assert pid is not None, "pg_net worker did not come back up after restart"
 
-    row = autocommit_sess.execute(text(
-        "select pid from pg_stat_activity where backend_type ilike '%pg_net%';"
-    )).fetchone()
-    assert row is not None, "pg_net worker is down, it crashed after seeing the net schema"
-    assert row[0] == pid, "pg_net worker restarted, it's crash looping on the net schema"
+    # Watch for several seconds; a crash loop would respawn the worker with a
+    # new pid. Poll for the bad condition (crash or restart) so we fail as
+    # soon as it happens instead of only checking once at the end of a blind
+    # sleep; a timeout here means the worker stayed up with the same pid.
+    changed = False
+    last_row = None
+    try:
+        last_row = wait_until(
+            fetch=lambda: autocommit_sess.execute(text(
+                "select pid from pg_stat_activity where backend_type ilike '%pg_net%';"
+            )).fetchone(),
+            predicate=lambda result: result is None or result[0] != pid,
+            timeout=3,
+            sleep_interval=0.1,
+            description="pg_net worker to remain stable with the same pid",
+        )
+        changed = True
+    except AssertionError:
+        pass
+
+    if changed:
+        assert last_row is not None, "pg_net worker is down, it crashed after seeing the net schema"
+        assert last_row[0] == pid, "pg_net worker restarted, it's crash looping on the net schema"
 
     sess.execute(text("drop schema net;"))
     sess.commit()

From 565d42f4558faf344a7f03f801cb54eb06ad4396 Mon Sep 17 00:00:00 2001
From: Raminder Singh <romi_ssk@yahoo.co.in>
Date: Sat, 1 Aug 2026 11:43:16 +0530
Subject: [PATCH 2/3] ci: improve CI timing

We no longer download all PG versions to reduce CI time. xpg 2.5.0 added support
for downloading only one PG version. We bumped to xpg 2.5.0 to use this feature
and updated commands in CI to use this feature.

We also avoid running two workflows on push to a PR branch. The on: [push,
pull_request] ran two duplicate workflows for each push to a PR branch.  This
was wasteful. Now we only run one workflow for each push to a PR branch. We also
run a workflow once commits land on the master branch, e.g.  after a PR is
merged.
---
 .github/workflows/main.yml | 35 ++++++++++++++++++++---------------
 flake.lock                 |  8 ++++----
 flake.nix                  |  2 +-
 shell.nix                  |  4 +++-
 4 files changed, 28 insertions(+), 21 deletions(-)

diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml
index cf11eccc..e692fa7e 100644
--- a/.github/workflows/main.yml
+++ b/.github/workflows/main.yml
@@ -1,14 +1,16 @@
 name: CI
 
-on: [push, pull_request]
+on:
+  pull_request:
+  push:
+    branches: [master]
 
 jobs:
-
   test:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        pg-version: ['12', '13', '14', '15', '16', '17', '18', '19']
+        pg-version: ["12", "13", "14", "15", "16", "17", "18", "19"]
 
     steps:
       - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
@@ -23,16 +25,16 @@ jobs:
           authToken: ${{ secrets.CACHIX_AUTH_TOKEN }}
 
       - name: Build
-        run: nix-shell --run "xpg -v ${{ matrix.pg-version }} build"
+        run: nix-shell --argstr pgVersion "${{ matrix.pg-version }}" --arg cassert false --run "xpg -v ${{ matrix.pg-version }} build"
 
       - name: Run tests
-        run: nix-shell --run "xpg -v ${{ matrix.pg-version }} test"
+        run: nix-shell --argstr pgVersion "${{ matrix.pg-version }}" --arg cassert false --run "xpg -v ${{ matrix.pg-version }} test"
 
   test-on-macos:
     runs-on: macos-15
     strategy:
       matrix:
-        pg-version: ['17']
+        pg-version: ["17"]
     steps:
       - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4.3.1
 
@@ -48,16 +50,21 @@ jobs:
           authToken: ${{ secrets.CACHIX_AUTH_TOKEN }}
 
       - name: Build
-        run: nix-shell --run "xpg -v ${{ matrix.pg-version }} build"
+        run: nix-shell --argstr pgVersion "${{ matrix.pg-version }}" --arg cassert false --run "xpg -v ${{ matrix.pg-version }} build"
 
       - name: Run tests
-        run: nix-shell --run "xpg -v ${{ matrix.pg-version }} test"
+        run: nix-shell --argstr pgVersion "${{ matrix.pg-version }}" --arg cassert false --run "xpg -v ${{ matrix.pg-version }} test"
 
   loadtest:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        params: [ {reqs: 10000, batch: 200}, {reqs: 20000, batch: 400}, {reqs: 40000, batch: 800} ]
+        params:
+          [
+            { reqs: 10000, batch: 200 },
+            { reqs: 20000, batch: 400 },
+            { reqs: 40000, batch: 800 },
+          ]
     steps:
       - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
 
@@ -71,19 +78,18 @@ jobs:
           authToken: ${{ secrets.CACHIX_AUTH_TOKEN }}
 
       - name: Build
-        run: nix-shell --run "xpg build"
+        run: nix-shell --argstr pgVersion "17" --arg cassert false --run "xpg build"
 
       - name: Run load test
         run: |
-          nix-shell --run "net-loadtest ${{ matrix.params.reqs }} ${{ matrix.params.batch }}" >> "$GITHUB_STEP_SUMMARY"
+          nix-shell --argstr pgVersion "17" --arg cassert false --run "net-loadtest ${{ matrix.params.reqs }} ${{ matrix.params.batch }}" >> "$GITHUB_STEP_SUMMARY"
 
   coverage:
-
     runs-on: ubuntu-latest
 
     strategy:
       matrix:
-        pg-version: ['17']
+        pg-version: ["17"]
 
     steps:
       - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
@@ -98,7 +104,7 @@ jobs:
           authToken: ${{ secrets.CACHIX_AUTH_TOKEN }}
 
       - name: Coverage
-        run: nix-shell --run "xpg -v ${{ matrix.pg-version }} coverage"
+        run: nix-shell --argstr pgVersion "${{ matrix.pg-version }}" --arg cassert false --run "xpg -v ${{ matrix.pg-version }} coverage"
 
       - name: Send coverage to Coveralls
         uses: coverallsapp/github-action@648a8eb78e6d50909eff900e4ec85cab4524a45b # v2.3.6
@@ -107,7 +113,6 @@ jobs:
           files: ./build-${{ matrix.pg-version }}/coverage.info
 
   style:
-
     runs-on: ubuntu-latest
 
     steps:
diff --git a/flake.lock b/flake.lock
index fa6f807a..86062f8c 100644
--- a/flake.lock
+++ b/flake.lock
@@ -43,16 +43,16 @@
         "nixpkgs": "nixpkgs_2"
       },
       "locked": {
-        "lastModified": 1782266378,
-        "narHash": "sha256-moRjdGvymGEyREU4++qpNzDuUN+nVC69CUYjhuY4Yzs=",
+        "lastModified": 1785529904,
+        "narHash": "sha256-DDIeJCHRtmvB7yocWTo1MISIGf0DsUmyLw6ISuEV4e0=",
         "owner": "steve-chavez",
         "repo": "xpg",
-        "rev": "d81725666a193c644fe0db96a988ece66edf460c",
+        "rev": "e7d47eb58f19168b5facc6b78a64f6cc7f04bbcb",
         "type": "github"
       },
       "original": {
         "owner": "steve-chavez",
-        "ref": "v2.4.0",
+        "ref": "v2.5.0",
         "repo": "xpg",
         "type": "github"
       }
diff --git a/flake.nix b/flake.nix
index a906059f..f12922de 100644
--- a/flake.nix
+++ b/flake.nix
@@ -5,7 +5,7 @@
     # 2025-11-13
     nixpkgs.url = "github:NixOS/nixpkgs/91c9a64ce2a84e648d0cf9671274bb9c2fb9ba60";
     xpg = {
-      url = "github:steve-chavez/xpg/v2.4.0";
+      url = "github:steve-chavez/xpg/v2.5.0";
     };
   };
 
diff --git a/shell.nix b/shell.nix
index eecd3d51..ee568e9b 100644
--- a/shell.nix
+++ b/shell.nix
@@ -14,6 +14,8 @@ in
     inherit (xpgLock) owner repo rev;
     sha256 = xpgLock.narHash;
   })
+, pgVersion ? null
+, cassert ? true
 }:
 let
   nginxCustom = pkgs.callPackage ./nix/nginxCustom.nix {};
@@ -36,7 +38,7 @@ in
 pkgs.mkShell {
   buildInputs =
     [
-      xpgPkgs.xpg
+      (if pgVersion == null then xpgPkgs.xpg else xpgPkgs.xpg.forVersions { versions = [ pgVersion ]; inherit cassert; })
       pythonDeps
       nginxCustom.nginxScript
       pkgs.curlWithGnuTls

From fcb0717f3aa174da25b70e1118a9a450742c8e66 Mon Sep 17 00:00:00 2001
From: Raminder Singh <romi_ssk@yahoo.co.in>
Date: Fri, 31 Jul 2026 20:54:03 +0530
Subject: [PATCH 3/3] docs: update docs on how to run all or only some tests

---
 CONTRIBUTING.md | 10 +++++++++-
 1 file changed, 9 insertions(+), 1 deletion(-)

diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index aa47893f..773b84c1 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -71,7 +71,15 @@ $ nix develop
 $ PYTEST_ARGS="-k test_connect" xpg test
 ```
 
-Will run the `test_connect` test only. `PYTEST_ARGS` is passed through to pytest, so you can pass other arguments to pytest as well.
+Will run the `test_connect` test only. `PYTEST_ARGS` is passed through to pytest, so you can pass other arguments to pytest as well. in fact you can pass any args via `PYTEST_ARGS`. You can e.g. run all tests in a file:
+
+```bash
+$ nix develop
+$ PYTEST_ARGS="PYTEST_ARGS="test/test_user_db.py" xpg test
+```
+
+Will run tests in `test/test_user_db.py`, or:
+
 
 ### Debugging