fix: minor improvements (xact callback leak, pg_net.ttl validation)

README.md

@@ -134,7 +134,7 @@ create extension pg_net;
 The extension creates the following configurable variables:
 
 1. **pg_net.batch_size** _(default: 200)_: An integer that limits the max number of rows that the extension will process from _`net.http_request_queue`_ during each read
-2. **pg_net.ttl** _(default: 6 hours)_: An interval that defines the max time a row in the _`net.http_response`_ will live before being deleted. Note that this won't happen exactly after the TTL has passed. The worker will perform this deletion while its processing requests.
+2. **pg_net.ttl** _(default: 6 hours)_: An interval that defines the max time a row in the _`net.http_response`_ will live before being deleted. Any interval that PostgreSQL accepts can be used (e.g. `'1 minute 10 seconds'`); invalid or negative values are rejected when the setting is changed. Note that the deletion won't happen exactly after the TTL has passed. The worker will perform this deletion while its processing requests.
 3. **pg_net.database_name** _(default: 'postgres')_: A string that defines which database the extension is applied to
 4. **pg_net.username** _(default: NULL)_: A string that defines which user will the background worker be connected with. If not set (`NULL`), it will assume the bootstrap user.
 

src/worker.c

@@ -36,6 +36,7 @@ static const int    curl_handle_event_timeout_ms = 1000;
 static const int    net_worker_restart_time_sec  = 1;
 static const long   no_timeout                   = -1L;
 static bool         wake_commit_cb_active        = false;
+static bool         wake_commit_cb_registered    = false;
 static bool         worker_should_restart        = false;
 static const size_t total_extension_tables       = 2;
 
@@ -123,11 +124,18 @@ static void wake_at_commit(XactEvent event, __attribute__((unused)) void *arg) {
 
 PG_FUNCTION_INFO_V1(wake);
 Datum wake(__attribute__((unused)) PG_FUNCTION_ARGS) {
-  if (!wake_commit_cb_active) { // register only one callback per transaction
+  // RegisterXactCallback appends a new entry to a backend-wide list on every call and never
+  // deduplicates, so it must be called at most once per backend. Otherwise every transaction that
+  // calls wake() leaks one entry in TopMemoryContext and CallXactCallbacks gets slower on every
+  // transaction of this backend for the rest of its life. `wake_commit_cb_active` is the
+  // per-transaction gate that decides whether the callback does anything at commit.
+  if (!wake_commit_cb_registered) {
     RegisterXactCallback(wake_at_commit, NULL);
-    wake_commit_cb_active = true;
+    wake_commit_cb_registered = true;
   }
 
+  wake_commit_cb_active = true;
+
   PG_RETURN_VOID();
 }
 
@@ -428,6 +436,47 @@ void pg_net_worker(__attribute__((unused)) Datum main_arg) {
   proc_exit(EXIT_FAILURE);
 }
 
+// GUC check hook for pg_net.ttl. The value is parsed with interval_in so that an invalid value is
+// rejected at SET/ALTER SYSTEM/config-reload time instead of making the worker fail when it tries
+// to delete expired responses. Negative intervals are rejected too since they'd expire every
+// response immediately.
+static bool check_ttl(char **newval, __attribute__((unused)) void **extra,
+                      __attribute__((unused)) GucSource source) {
+  if (*newval == NULL) return true;
+
+  MemoryContext ccxt      = CurrentMemoryContext;
+  Datum         ttl_datum = (Datum)0;
+  bool          parsed    = false;
+
+  PG_TRY();
+  {
+    ttl_datum = DirectFunctionCall3(interval_in, CStringGetDatum(*newval),
+                                    ObjectIdGetDatum(InvalidOid), Int32GetDatum(-1));
+    parsed    = true;
+  }
+  PG_CATCH();
+  {
+    MemoryContextSwitchTo(ccxt);
+    ErrorData *edata = CopyErrorData();
+    FlushErrorState();
+    GUC_check_errdetail("%s", edata->message);
+    FreeErrorData(edata);
+  }
+  PG_END_TRY();
+
+  if (!parsed) return false;
+
+  Datum zero = DirectFunctionCall3(interval_in, CStringGetDatum("0"), ObjectIdGetDatum(InvalidOid),
+                                   Int32GetDatum(-1));
+
+  if (DatumGetInt32(DirectFunctionCall2(interval_cmp, ttl_datum, zero)) < 0) {
+    GUC_check_errdetail("\"%s\" is a negative interval", *newval);
+    return false;
+  }
+
+  return true;
+}
+
 static Size net_memsize(void) {
   return MAXALIGN(sizeof(WorkerState));
 }
@@ -494,8 +543,8 @@ void _PG_init(void) {
   shmem_startup_hook      = net_shmem_startup;
 
   DefineCustomStringVariable("pg_net.ttl", "time to live for request/response rows",
-                             "should be a valid interval type", &guc_ttl, "6 hours", PGC_SIGHUP, 0,
-                             NULL, NULL, NULL);
+                             "should be a valid, non-negative interval", &guc_ttl, "6 hours",
+                             PGC_SIGHUP, 0, check_ttl, NULL, NULL);
 
   DefineCustomIntVariable(
       "pg_net.batch_size", "number of requests executed in one iteration of the background worker",

test/test_ttl_config.py

@@ -0,0 +1,91 @@
+import time
+import pytest
+from sqlalchemy import text
+from sqlalchemy.exc import DBAPIError
+from common import collect_response_sync, http_request, restart_worker
+from common import wait_for_response_count, wakeup_worker
+
+
+def reset_ttl(autocommit_sess):
+    autocommit_sess.execute(text("alter system reset pg_net.ttl"))
+    autocommit_sess.execute(text("select pg_reload_conf()"))
+
+
+def test_invalid_ttl_is_rejected(autocommit_sess):
+    """
+    An invalid interval must be rejected at ALTER SYSTEM time
+    instead of making the worker fail when it tries to delete
+    expired responses (issue #268)
+    """
+    for bad in ["1 blah", "yesterday", "abc", "1 minute 10 potatoes"]:
+        with pytest.raises(DBAPIError) as excinfo:
+            autocommit_sess.execute(text(f"alter system set pg_net.ttl to '{bad}'"))
+        msg = str(excinfo.value)
+        assert 'invalid value for parameter "pg_net.ttl"' in msg
+        assert "invalid input syntax for type interval" in msg
+
+    # the setting must be untouched
+    (ttl,) = autocommit_sess.execute(text("show pg_net.ttl")).fetchone()
+    assert ttl == "6 hours"
+
+
+def test_negative_ttl_is_rejected(autocommit_sess):
+    """
+    A negative ttl would expire every response immediately, reject it
+    """
+    for bad in ["-1 hour", "-10 seconds", "1 hour ago"]:
+        with pytest.raises(DBAPIError) as excinfo:
+            autocommit_sess.execute(text(f"alter system set pg_net.ttl to '{bad}'"))
+        msg = str(excinfo.value)
+        assert 'invalid value for parameter "pg_net.ttl"' in msg
+        assert "negative interval" in msg
+
+
+def test_valid_ttl_formats_are_accepted(autocommit_sess):
+    """
+    Any interval that postgres itself accepts must be accepted,
+    including compound values (issue #269)
+    """
+    try:
+        for good in [
+            "1 minute 10 seconds",
+            "90 seconds",
+            "1.5 hours",
+            "00:01:10",
+            "1 day 2 hours 3 minutes",
+            "500 milliseconds",
+            "0",
+        ]:
+            autocommit_sess.execute(text(f"alter system set pg_net.ttl to '{good}'"))
+            autocommit_sess.execute(text("select pg_reload_conf()"))
+    finally:
+        reset_ttl(autocommit_sess)
+
+
+def test_compound_ttl_is_honored_by_worker(sess, autocommit_sess):
+    """
+    A compound interval like '1 second 500 milliseconds' must be
+    applied by the worker as-is, i.e. a response must be expired
+    once that time has passed and the worker wakes (issue #269)
+    """
+    try:
+        autocommit_sess.execute(
+            text("alter system set pg_net.ttl to '1 second 500 milliseconds'"))
+        restart_worker(autocommit_sess)
+
+        request_id = http_request(sess, text(
+            "select net.http_get('http://localhost:8080/anything')"))
+
+        response = collect_response_sync(sess, request_id)
+        assert response is not None
+        assert response["status"] == "SUCCESS"
+
+        # Sleep past the ttl so the response is expired, then wake the worker
+        time.sleep(1.6)
+        wakeup_worker(sess)
+
+        wait_for_response_count(autocommit_sess, 0)
+
+    finally:
+        reset_ttl(autocommit_sess)
+        restart_worker(autocommit_sess)

test/test_xact_callback.py

@@ -0,0 +1,52 @@
+import pytest
+from sqlalchemy import text
+
+
+def test_wake_does_not_leak_xact_callbacks(sess, autocommit_sess):
+    """
+    net.wake() registers a transaction callback so the worker is woken at
+    commit time. RegisterXactCallback appends a new entry on every call and
+    never deduplicates, so it must be registered at most once per backend.
+    Otherwise every transaction that calls net.http_* leaks an entry in
+    TopMemoryContext for the lifetime of the connection.
+
+    Measure TopMemoryContext of this backend before and after thousands of
+    single-statement transactions that call net.wake(): it must stay flat.
+    """
+    (version_num,) = autocommit_sess.execute(
+        text("show server_version_num")).fetchone()
+    if int(version_num) < 140000:
+        pytest.skip("pg_backend_memory_contexts requires PostgreSQL 14+")
+
+    def top_memory_context_used_bytes():
+        return autocommit_sess.execute(text("""
+            select used_bytes
+            from pg_backend_memory_contexts
+            where name = 'TopMemoryContext'
+        """)).scalar_one()
+
+    iterations = 5000
+
+    # warm up so one-off allocations (first callback registration, catalog
+    # lookups, etc.) don't count towards the measurement
+    for _ in range(50):
+        autocommit_sess.execute(text("select net.wake()"))
+
+    before = top_memory_context_used_bytes()
+
+    # each execute is its own transaction under autocommit, so each one
+    # runs wake() and then the commit callback
+    for _ in range(iterations):
+        autocommit_sess.execute(text("select net.wake()"))
+
+    after = top_memory_context_used_bytes()
+    growth = after - before
+    print(f"TopMemoryContext growth over {iterations} net.wake() transactions: {growth} bytes")
+
+    # a leaked callback entry costs 32 bytes (24-byte struct in a 32-byte
+    # AllocSet chunk), so the buggy behavior grows by iterations * 32 bytes
+    # (160 KB for 5000 iterations). Allow for some unrelated noise.
+    assert growth < 32 * 1024, (
+        f"TopMemoryContext grew by {growth} bytes over {iterations} "
+        f"transactions calling net.wake(), expected it to stay flat"
+    )