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/CONTRIBUTING.md b/CONTRIBUTING.md
index ebf27d06..773b84c1 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -64,6 +64,23 @@ $ 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. 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
 
 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/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
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()