From 637e5d1e1c35c090a8e8e9de4fdb8e24004d5e94 Mon Sep 17 00:00:00 2001 From: Martin Donadieu Date: Fri, 17 Jul 2026 02:06:02 +0200 Subject: [PATCH] feat: add net.http_request callback interface (on_success/on_error) Implements the callback-based interface discussed in #62 on top of the existing queue, as an addition to the current API: net.http_request() hands the response to an on_success SQL command ($1 = status_code, $2 = headers, $3 = body) or, for failures without a response, to an on_error SQL command ($1 = error message, $2 = timed_out). Requests made this way never touch net._http_response, so callers that don't read responses (webhook-style notifications, queue wake-ups) stop paying for the response row insert plus its later TTL delete. With no on_success callback the response is discarded without even buffering the body in memory (fire-and-forget). Callbacks are executed by the background worker inside a subtransaction, as the role that enqueued the request, with SET ROLE/SET SESSION AUTHORIZATION blocked, so a callback can't do anything the enqueuing role couldn't. A failing callback is rolled back and reported as a WARNING without aborting the batch. Retry policies stay in user hands: callbacks can re-enqueue via net.http_request() again. The existing http_get/http_post/http_delete functions and their table-based behavior are completely unchanged. The worker tolerates a binary newer than the installed extension version (shared library replaced before ALTER EXTENSION pg_net UPDATE runs): it checks for the new queue columns and falls back to the legacy query when they don't exist. Co-Authored-By: Claude Fable 5 --- Makefile | 2 +- README.md | 69 ++++ sql/pg_net--0.20.4--0.21.0.sql | 60 ++++ sql/pg_net.sql | 66 +++- src/core.c | 198 ++++++++++- src/core.h | 10 +- src/pg_prelude.h | 1 + src/worker.c | 2 +- test/test_http_request_callbacks.py | 515 ++++++++++++++++++++++++++++ 9 files changed, 902 insertions(+), 21 deletions(-) create mode 100644 sql/pg_net--0.20.4--0.21.0.sql create mode 100644 test/test_http_request_callbacks.py diff --git a/Makefile b/Makefile index 240a5dc2..2496cd93 100644 --- a/Makefile +++ b/Makefile @@ -33,7 +33,7 @@ else endif EXTENSION = pg_net -EXTVERSION = 0.20.4 +EXTVERSION = 0.21.0 DATA = $(wildcard sql/*--*.sql) diff --git a/README.md b/README.md index dc01a0e6..9cf7583b 100644 --- a/README.md +++ b/README.md @@ -386,6 +386,75 @@ FROM selected_row --- +## Callback requests (net.http_request) + +### net.http_request function signature + +```sql +net.http_request( + -- http method (GET, POST or DELETE) + method net.http_method, + -- url for the request + url text, + -- key/value pairs to be url encoded and appended to the `url` + params jsonb default '{}'::jsonb, + -- key/values to be included in request headers + headers jsonb default '{}'::jsonb, + -- optional body of the request + body jsonb default null, + -- the maximum number of milliseconds the request may take before being cancelled + timeout_milliseconds int default 5000, + -- SQL command executed when an HTTP response arrives (any status code). + -- Parameters: $1 = status_code int, $2 = headers jsonb, $3 = body text. + -- When null, the response is discarded (fire-and-forget). + on_success text default null, + -- SQL command executed when the request fails without an HTTP response + -- (timeout, connection error). Parameters: $1 = error message text, + -- $2 = timed_out bool. When null, the failure is only logged. + on_error text default null +) + returns void + + volatile + parallel unsafe + language plpgsql +``` + +Unlike `http_get`/`http_post`/`http_delete`, requests made with `net.http_request` never touch the `net._http_response` table: the response is handed to the `on_success` callback, or discarded when no callback is given. This avoids the write-then-expire churn on `net._http_response` for callers that never read responses (see [#62](https://github.com/supabase/pg_net/issues/62)). + +- Every HTTP response — including 4xx/5xx — goes through `on_success`, with the status code as `$1`. Only failures without a response (timeouts, connection errors) go through `on_error`. +- Callbacks are executed by the background worker **as the role that enqueued the request**, with `SET ROLE`/`SET SESSION AUTHORIZATION` blocked, so they can't do anything the enqueuing role couldn't. +- A callback that raises an error is rolled back and reported as a `WARNING` in the database logs; the batch keeps processing and the request is considered handled. +- Retry policies stay in user hands: an `on_error` (or `on_success` checking `$1`) callback can re-enqueue with `net.http_request(...)` again. + +### Examples: + +#### Fire-and-forget + +```sql +SELECT net.http_request( + 'POST', + 'https://postman-echo.com/post', + headers := '{"Content-Type": "application/json"}'::JSONB, + body := '{"key": "value"}'::JSONB +); +``` + +No response row is stored, and the response body is not even buffered in memory. + +#### Processing the response + +```sql +CREATE TABLE webhook_results(status int, body jsonb); + +SELECT net.http_request( + 'GET', + 'https://postman-echo.com/get', + on_success := 'insert into webhook_results values ($1, $3::jsonb)', + on_error := $$select pg_notify('webhook_failures', $1)$$ +); +``` + # Practical Examples ## Syncing data with an external data source using triggers diff --git a/sql/pg_net--0.20.4--0.21.0.sql b/sql/pg_net--0.20.4--0.21.0.sql new file mode 100644 index 00000000..e17a5784 --- /dev/null +++ b/sql/pg_net--0.20.4--0.21.0.sql @@ -0,0 +1,60 @@ +alter table net.http_request_queue + add column use_callbacks bool not null default false, + add column on_success text, + add column on_error text, + add column calling_role text; + +-- Interface to make an async request handled by callbacks instead of the +-- net._http_response table. See https://github.com/supabase/pg_net/issues/62 +-- API: Public +create or replace function net.http_request( + -- http method (GET, POST or DELETE) + method net.http_method, + -- url for the request + url text, + -- key/value pairs to be url encoded and appended to the `url` + params jsonb default '{}'::jsonb, + -- key/values to be included in request headers + headers jsonb default '{}'::jsonb, + -- optional body of the request + body jsonb default null, + -- the maximum number of milliseconds the request may take before being cancelled + timeout_milliseconds int default 5000, + -- SQL command executed by the background worker when an HTTP response + -- arrives (any status code). Parameters: $1 = status_code int, + -- $2 = headers jsonb, $3 = body text. When null, the response is + -- discarded without being stored anywhere (fire-and-forget). + on_success text default null, + -- SQL command executed by the background worker when the request fails + -- without an HTTP response (timeout, connection error). Parameters: + -- $1 = error message text, $2 = timed_out bool. When null, the failure + -- is only reported in the database logs. + on_error text default null +) + returns void + language plpgsql +as $$ +declare + params_array text[]; +begin + select coalesce(array_agg(net._urlencode_string(key) || '=' || net._urlencode_string(value)), '{}') + into params_array + from jsonb_each_text(params); + + -- Add to the request queue + insert into net.http_request_queue(method, url, headers, body, timeout_milliseconds, use_callbacks, on_success, on_error, calling_role) + values ( + method, + net._encode_url_with_params_array(url, params_array), + headers, + convert_to(body::text, 'UTF8'), + timeout_milliseconds, + true, + on_success, + on_error, + current_user + ); + + perform net.wake(); +end +$$; diff --git a/sql/pg_net.sql b/sql/pg_net.sql index b0acd86a..bffa0d19 100644 --- a/sql/pg_net.sql +++ b/sql/pg_net.sql @@ -15,7 +15,16 @@ create unlogged table net.http_request_queue( url text not null, headers jsonb, body bytea, - timeout_milliseconds int not null + timeout_milliseconds int not null, + -- when true, the response is handed to the callbacks below and never + -- stored in net._http_response + use_callbacks bool not null default false, + -- SQL command executed when an HTTP response arrives (any status code) + on_success text, + -- SQL command executed when the request fails without an HTTP response + on_error text, + -- role the callbacks are executed as + calling_role text ); create or replace function net.check_worker_is_up() returns void as $$ @@ -261,6 +270,61 @@ begin end $$; +-- Interface to make an async request handled by callbacks instead of the +-- net._http_response table. See https://github.com/supabase/pg_net/issues/62 +-- API: Public +create or replace function net.http_request( + -- http method (GET, POST or DELETE) + method net.http_method, + -- url for the request + url text, + -- key/value pairs to be url encoded and appended to the `url` + params jsonb default '{}'::jsonb, + -- key/values to be included in request headers + headers jsonb default '{}'::jsonb, + -- optional body of the request + body jsonb default null, + -- the maximum number of milliseconds the request may take before being cancelled + timeout_milliseconds int default 5000, + -- SQL command executed by the background worker when an HTTP response + -- arrives (any status code). Parameters: $1 = status_code int, + -- $2 = headers jsonb, $3 = body text. When null, the response is + -- discarded without being stored anywhere (fire-and-forget). + on_success text default null, + -- SQL command executed by the background worker when the request fails + -- without an HTTP response (timeout, connection error). Parameters: + -- $1 = error message text, $2 = timed_out bool. When null, the failure + -- is only reported in the database logs. + on_error text default null +) + returns void + language plpgsql +as $$ +declare + params_array text[]; +begin + select coalesce(array_agg(net._urlencode_string(key) || '=' || net._urlencode_string(value)), '{}') + into params_array + from jsonb_each_text(params); + + -- Add to the request queue + insert into net.http_request_queue(method, url, headers, body, timeout_milliseconds, use_callbacks, on_success, on_error, calling_role) + values ( + method, + net._encode_url_with_params_array(url, params_array), + headers, + convert_to(body::text, 'UTF8'), + timeout_milliseconds, + true, + on_success, + on_error, + current_user + ); + + perform net.wake(); +end +$$; + -- Lifecycle states of a request (all protocols) -- API: Public create type net.request_status as enum ('PENDING', 'SUCCESS', 'ERROR'); diff --git a/src/core.c b/src/core.c index 918acf7f..936e6d04 100644 --- a/src/core.c +++ b/src/core.c @@ -11,9 +11,10 @@ #include "errors.h" #include "event.h" -static SPIPlanPtr del_response_plan = NULL; -static SPIPlanPtr del_return_queue_plan = NULL; -static SPIPlanPtr ins_response_plan = NULL; +static SPIPlanPtr del_response_plan = NULL; +static SPIPlanPtr del_return_queue_plan = NULL; +static SPIPlanPtr del_return_queue_plan_legacy = NULL; +static SPIPlanPtr ins_response_plan = NULL; static size_t body_cb(void *contents, size_t size, size_t nmemb, void *userp) { CurlHandle *handle = (CurlHandle *)userp; @@ -22,6 +23,12 @@ static size_t body_cb(void *contents, size_t size, size_t nmemb, void *userp) { return realsize; } +// discards the response body without buffering it, used when store_response is false +static size_t discard_cb(__attribute__((unused)) void *contents, size_t size, size_t nmemb, + __attribute__((unused)) void *userp) { + return size * nmemb; +} + static struct curl_slist *pg_text_array_to_slist(ArrayType *array, struct curl_slist *headers) { ArrayIterator iterator; Datum value; @@ -45,8 +52,19 @@ static struct curl_slist *pg_text_array_to_slist(ArrayType *array, struct curl_s } void init_curl_handle(CurlHandle *handle, RequestQueueRow row) { - handle->id = row.id; - handle->body = makeStringInfo(); + handle->id = row.id; + handle->use_callbacks = row.use_callbacks; + handle->on_success = + !row.onSuccessBin.isnull ? TextDatumGetCString(row.onSuccessBin.value) : NULL; + handle->on_error = !row.onErrorBin.isnull ? TextDatumGetCString(row.onErrorBin.value) : NULL; + handle->calling_role = + !row.callingRoleBin.isnull ? TextDatumGetCString(row.callingRoleBin.value) : NULL; + + // the body is only needed when it ends up in net._http_response or in the + // on_success callback, otherwise don't buffer it + bool body_needed = !handle->use_callbacks || handle->on_success != NULL; + handle->body = body_needed ? makeStringInfo() : NULL; + handle->ez_handle = curl_easy_init(); handle->timeout_milliseconds = row.timeout_milliseconds; @@ -96,8 +114,12 @@ void init_curl_handle(CurlHandle *handle, RequestQueueRow row) { } } - EREPORT_CURL_SETOPT(handle->ez_handle, CURLOPT_WRITEFUNCTION, body_cb); - EREPORT_CURL_SETOPT(handle->ez_handle, CURLOPT_WRITEDATA, handle); + if (handle->body) { + EREPORT_CURL_SETOPT(handle->ez_handle, CURLOPT_WRITEFUNCTION, body_cb); + EREPORT_CURL_SETOPT(handle->ez_handle, CURLOPT_WRITEDATA, handle); + } else { + EREPORT_CURL_SETOPT(handle->ez_handle, CURLOPT_WRITEFUNCTION, discard_cb); + } EREPORT_CURL_SETOPT(handle->ez_handle, CURLOPT_HEADER, 0L); EREPORT_CURL_SETOPT(handle->ez_handle, CURLOPT_URL, handle->url); EREPORT_CURL_SETOPT(handle->ez_handle, CURLOPT_HTTPHEADER, handle->request_headers); @@ -157,9 +179,20 @@ uint64 delete_expired_responses(char *ttl, int batch_size) { return affected_rows; } -uint64 consume_request_queue(const int batch_size) { - if (del_return_queue_plan == NULL) { - SPIPlanPtr tmp = SPI_prepare("\ +uint64 consume_request_queue(const int batch_size, Oid queue_oid) { + /* + * The callback columns only exist from extension version 0.21.0 onwards. + * The loaded binary can be newer than the installed extension version (the + * shared library is replaced on upgrade while `ALTER EXTENSION pg_net + * UPDATE` might not have been executed yet), so fall back to a query that + * doesn't reference the columns when they don't exist. + */ + bool has_callbacks = get_attnum(queue_oid, "use_callbacks") != InvalidAttrNumber; + + SPIPlanPtr *plan = has_callbacks ? &del_return_queue_plan : &del_return_queue_plan_legacy; + + if (*plan == NULL) { + const char *query = has_callbacks ? "\ WITH\ rows AS (\ SELECT id\ @@ -169,18 +202,29 @@ uint64 consume_request_queue(const int batch_size) { )\ DELETE FROM net.http_request_queue q\ USING rows WHERE q.id = rows.id\ - RETURNING q.id, q.method, q.url, timeout_milliseconds, array(select key || ': ' || value from jsonb_each_text(q.headers)), q.body", - 1, (Oid[]){INT4OID}); + RETURNING q.id, q.method, q.url, timeout_milliseconds, array(select key || ': ' || value from jsonb_each_text(q.headers)), q.body, q.use_callbacks, q.on_success, q.on_error, q.calling_role" + : "\ + WITH\ + rows AS (\ + SELECT id\ + FROM net.http_request_queue\ + ORDER BY id\ + LIMIT $1\ + )\ + DELETE FROM net.http_request_queue q\ + USING rows WHERE q.id = rows.id\ + RETURNING q.id, q.method, q.url, timeout_milliseconds, array(select key || ': ' || value from jsonb_each_text(q.headers)), q.body, false, null::text, null::text, null::text"; + + SPIPlanPtr tmp = SPI_prepare(query, 1, (Oid[]){INT4OID}); if (tmp == NULL) ereport(ERROR, errmsg("SPI_prepare failed: %s", SPI_result_code_string(SPI_result))); - del_return_queue_plan = SPI_saveplan(tmp); - if (del_return_queue_plan == NULL) ereport(ERROR, errmsg("SPI_saveplan failed")); + *plan = SPI_saveplan(tmp); + if (*plan == NULL) ereport(ERROR, errmsg("SPI_saveplan failed")); } - int ret_code = - SPI_execute_plan(del_return_queue_plan, (Datum[]){Int32GetDatum(batch_size)}, NULL, false, 0); + int ret_code = SPI_execute_plan(*plan, (Datum[]){Int32GetDatum(batch_size)}, NULL, false, 0); if (ret_code != SPI_OK_DELETE_RETURNING) ereport(ERROR, @@ -213,7 +257,21 @@ RequestQueueRow get_request_queue_row(HeapTuple spi_tupval, TupleDesc spi_tupdes NullableDatum bodyBin = {.value = SPI_getbinval(spi_tupval, spi_tupdesc, 6, &tupIsNull), .isnull = tupIsNull}; - return (RequestQueueRow){id, method, url, timeout_milliseconds, headersBin, bodyBin}; + bool use_callbacks = DatumGetBool(SPI_getbinval(spi_tupval, spi_tupdesc, 7, &tupIsNull)); + EREPORT_NULL_ATTR(tupIsNull, use_callbacks); + + NullableDatum onSuccessBin = {.value = SPI_getbinval(spi_tupval, spi_tupdesc, 8, &tupIsNull), + .isnull = tupIsNull}; + + NullableDatum onErrorBin = {.value = SPI_getbinval(spi_tupval, spi_tupdesc, 9, &tupIsNull), + .isnull = tupIsNull}; + + NullableDatum callingRoleBin = {.value = SPI_getbinval(spi_tupval, spi_tupdesc, 10, &tupIsNull), + .isnull = tupIsNull}; + + return (RequestQueueRow){id, method, url, timeout_milliseconds, + headersBin, bodyBin, use_callbacks, onSuccessBin, + onErrorBin, callingRoleBin}; } static Jsonb *jsonb_headers_from_curl_handle(CURL *ez_handle) { @@ -234,7 +292,110 @@ static Jsonb *jsonb_headers_from_curl_handle(CURL *ez_handle) { return PG_JSONB_OBJECT_FINISH(headers); } +/* + * Execute a callback command with SPI inside a subtransaction, as the role + * that enqueued the request. A failing callback is reported as a WARNING and + * doesn't abort the batch: the request is considered processed. + */ +static void exec_callback(CurlHandle *handle, const char *label, const char *command, int nargs, + Oid *argtypes, Datum *values, const char *nulls) { + Oid roleid = handle->calling_role ? get_role_oid(handle->calling_role, true) : InvalidOid; + + if (!OidIsValid(roleid)) { + ereport(WARNING, errmsg("pg_net: skipping %s callback of request id " INT64_FORMAT + ": role \"%s\" does not exist", + label, handle->id, handle->calling_role ? handle->calling_role : "")); + return; + } + + Oid saved_userid; + int saved_sec_context; + GetUserIdAndSecContext(&saved_userid, &saved_sec_context); + + MemoryContext oldcontext = CurrentMemoryContext; + ResourceOwner oldowner = CurrentResourceOwner; + + BeginInternalSubTransaction(NULL); + + // run the callback as the enqueuing role, and prevent it from escalating + // via SET ROLE/SET SESSION AUTHORIZATION + SetUserIdAndSecContext(roleid, saved_sec_context | SECURITY_LOCAL_USERID_CHANGE | + SECURITY_RESTRICTED_OPERATION); + + PG_TRY(); + { + int rc = SPI_execute_with_args(command, nargs, argtypes, values, nulls, false, 0); + if (rc < 0) + ereport(ERROR, errmsg("SPI_execute_with_args failed: %s", SPI_result_code_string(rc))); + + SetUserIdAndSecContext(saved_userid, saved_sec_context); + ReleaseCurrentSubTransaction(); + MemoryContextSwitchTo(oldcontext); + CurrentResourceOwner = oldowner; + } + PG_CATCH(); + { + SetUserIdAndSecContext(saved_userid, saved_sec_context); + + MemoryContextSwitchTo(oldcontext); + ErrorData *edata = CopyErrorData(); + FlushErrorState(); + + RollbackAndReleaseCurrentSubTransaction(); + MemoryContextSwitchTo(oldcontext); + CurrentResourceOwner = oldowner; + + ereport(WARNING, errmsg("pg_net: %s callback of request id " INT64_FORMAT " failed: %s", label, + handle->id, edata->message)); + FreeErrorData(edata); + } + PG_END_TRY(); +} + +static void exec_request_callbacks(CurlHandle *handle, CURLcode curl_return_code) { + if (curl_return_code == CURLE_OK) { + if (handle->on_success == NULL) return; // fire-and-forget, discard the response + + long res_http_status_code = 0; + EREPORT_CURL_GETINFO(handle->ez_handle, CURLINFO_RESPONSE_CODE, &res_http_status_code); + + Jsonb *jsonb_headers = jsonb_headers_from_curl_handle(handle->ez_handle); + + bool body_is_empty = !handle->body || handle->body->data[0] == '\0'; + + Datum values[3] = {Int32GetDatum((int32)res_http_status_code), JsonbPGetDatum(jsonb_headers), + body_is_empty ? (Datum)0 : CStringGetTextDatum(handle->body->data)}; + + exec_callback(handle, "on_success", handle->on_success, 3, (Oid[]){INT4OID, JSONBOID, TEXTOID}, + values, body_is_empty ? " n" : " "); + } else { + bool timed_out = curl_return_code == CURLE_OPERATION_TIMEDOUT; + + curl_timeout_msg timeout_msg = {.msg = ""}; + if (timed_out) + timeout_msg = detailed_timeout_strerror(handle->ez_handle, handle->timeout_milliseconds); + + const char *error_msg = timed_out ? timeout_msg.msg : curl_easy_strerror(curl_return_code); + + if (handle->on_error == NULL) { + ereport(LOG, errmsg("pg_net: request id " INT64_FORMAT " failed: %s", handle->id, error_msg)); + return; + } + + Datum values[2] = {CStringGetTextDatum(error_msg), BoolGetDatum(timed_out)}; + + exec_callback(handle, "on_error", handle->on_error, 2, (Oid[]){TEXTOID, BOOLOID}, values, " "); + } +} + void insert_response(CurlHandle *handle, CURLcode curl_return_code) { + // requests made through net.http_request() are handled by callbacks and + // never touch the net._http_response table + if (handle->use_callbacks) { + exec_request_callbacks(handle, curl_return_code); + return; + } + enum { nparams = 7 }; // using an enum because const size_t nparams doesn't compile Datum vals[nparams]; char nulls[nparams]; @@ -317,6 +478,9 @@ void pfree_handle(CurlHandle *handle) { pfree(handle->url); pfree(handle->method); if (handle->req_body) pfree(handle->req_body); + if (handle->on_success) pfree(handle->on_success); + if (handle->on_error) pfree(handle->on_error); + if (handle->calling_role) pfree(handle->calling_role); if (handle->body) destroyStringInfo(handle->body); diff --git a/src/core.h b/src/core.h index dff2f671..20e276b5 100644 --- a/src/core.h +++ b/src/core.h @@ -26,6 +26,10 @@ typedef struct { int32 timeout_milliseconds; NullableDatum headersBin; NullableDatum bodyBin; + bool use_callbacks; + NullableDatum onSuccessBin; + NullableDatum onErrorBin; + NullableDatum callingRoleBin; } RequestQueueRow; // The curl easy handle plus additional data, this acts for both the request and @@ -38,12 +42,16 @@ typedef struct { char *url; char *req_body; char *method; + bool use_callbacks; + char *on_success; + char *on_error; + char *calling_role; CURL *ez_handle; } CurlHandle; uint64 delete_expired_responses(char *ttl, int batch_size); -uint64 consume_request_queue(const int batch_size); +uint64 consume_request_queue(const int batch_size, Oid queue_oid); RequestQueueRow get_request_queue_row(HeapTuple spi_tupval, TupleDesc spi_tupdesc); diff --git a/src/pg_prelude.h b/src/pg_prelude.h index addc4d19..f2e00c31 100644 --- a/src/pg_prelude.h +++ b/src/pg_prelude.h @@ -43,6 +43,7 @@ #include #include #include +#include #include #include diff --git a/src/worker.c b/src/worker.c index 98862ffd..93d07eb6 100644 --- a/src/worker.c +++ b/src/worker.c @@ -317,7 +317,7 @@ void pg_net_worker(__attribute__((unused)) Datum main_arg) { elog(DEBUG1, "Deleted " UINT64_FORMAT " expired rows", expired_responses); - requests_consumed = consume_request_queue(guc_batch_size); + requests_consumed = consume_request_queue(guc_batch_size, ext_table_oids[0]); elog(DEBUG1, "Consumed " UINT64_FORMAT " request rows", requests_consumed); diff --git a/test/test_http_request_callbacks.py b/test/test_http_request_callbacks.py new file mode 100644 index 00000000..537ff701 --- /dev/null +++ b/test/test_http_request_callbacks.py @@ -0,0 +1,515 @@ +import time + +from sqlalchemy import text + + +def wait_until_queue_empty(sess, timeout=10): + """Poll until the request queue is drained. + + The worker consumes the queue, runs the callbacks and commits in the same + transaction, so once the queue is visibly empty the callbacks of the + consumed batch are committed too. + """ + deadline = time.time() + timeout + while time.time() < deadline: + (count,) = sess.execute(text( + """ + select count(*) from net.http_request_queue; + """ + )).fetchone() + sess.commit() + if count == 0: + return + time.sleep(0.1) + raise TimeoutError("request queue did not drain") + + +def test_http_request_fire_and_forget(sess): + """without callbacks the response is discarded and nothing is stored""" + + sess.execute(text( + """ + select net.http_request('GET', 'http://localhost:8080/pathological?status=200') + from generate_series(1, 50); + """ + )) + sess.commit() + + wait_until_queue_empty(sess) + + (count,) = sess.execute(text( + """ + select count(*) from net._http_response; + """ + )).fetchone() + + assert count == 0 + + (up,) = sess.execute(text( + """ + select is_worker_up(); + """ + )).fetchone() + + assert up == True + + +def test_http_request_on_success(sess): + """on_success receives $1 = status_code, $2 = headers, $3 = body""" + + sess.execute(text( + """ + drop table if exists public.cb_results; + create table public.cb_results(status int, headers jsonb, body text); + """ + )) + sess.commit() + + # the root path returns a non-empty body ("Hello world!") + sess.execute(text( + """ + select net.http_request( + 'GET', + 'http://localhost:8080/', + on_success := 'insert into public.cb_results values ($1, $2, $3)' + ); + """ + )) + sess.commit() + + wait_until_queue_empty(sess) + + row = sess.execute(text( + """ + select status, headers is not null, body from public.cb_results; + """ + )).fetchone() + + assert row is not None + assert row[0] == 200 + assert row[1] == True + assert row[2] is not None + assert "Hello world" in row[2] + + sess.execute(text("drop table public.cb_results;")) + sess.commit() + + # nothing was stored in the response table + (count,) = sess.execute(text( + """ + select count(*) from net._http_response; + """ + )).fetchone() + + assert count == 0 + + +def test_http_request_on_success_empty_body(sess): + """an empty response body reaches on_success as null""" + + sess.execute(text( + """ + drop table if exists public.cb_empty; + create table public.cb_empty(status int, body text); + """ + )) + sess.commit() + + # the pathological endpoint returns an empty body + sess.execute(text( + """ + select net.http_request( + 'GET', + 'http://localhost:8080/pathological?status=200', + on_success := 'insert into public.cb_empty values ($1, $3)' + ); + """ + )) + sess.commit() + + wait_until_queue_empty(sess) + + row = sess.execute(text( + """ + select status, body is null from public.cb_empty; + """ + )).fetchone() + + assert row is not None + assert row[0] == 200 + assert row[1] == True + + sess.execute(text("drop table public.cb_empty;")) + sess.commit() + + +def test_http_request_on_success_gets_non_2xx_status(sess): + """HTTP errors still go through on_success, with the status code""" + + sess.execute(text( + """ + drop table if exists public.cb_statuses; + create table public.cb_statuses(status int); + """ + )) + sess.commit() + + sess.execute(text( + """ + select net.http_request( + 'GET', + 'http://localhost:8080/pathological?status=500', + on_success := 'insert into public.cb_statuses values ($1)' + ); + """ + )) + sess.commit() + + wait_until_queue_empty(sess) + + (status,) = sess.execute(text( + """ + select status from public.cb_statuses; + """ + )).fetchone() + + assert status == 500 + + sess.execute(text("drop table public.cb_statuses;")) + sess.commit() + + +def test_http_request_post_with_body(sess): + """POST requests work through the callback interface""" + + sess.execute(text( + """ + drop table if exists public.cb_post; + create table public.cb_post(status int); + """ + )) + sess.commit() + + sess.execute(text( + """ + select net.http_request( + 'POST', + 'http://localhost:8080/pathological?status=200', + headers := '{"Content-Type": "application/json"}'::jsonb, + body := '{"hello": "world"}'::jsonb, + on_success := 'insert into public.cb_post values ($1)' + ); + """ + )) + sess.commit() + + wait_until_queue_empty(sess) + + (status,) = sess.execute(text( + """ + select status from public.cb_post; + """ + )).fetchone() + + assert status == 200 + + sess.execute(text("drop table public.cb_post;")) + sess.commit() + + +def test_http_request_on_error(sess): + """on_error receives $1 = error message, $2 = timed_out""" + + sess.execute(text( + """ + drop table if exists public.cb_errors; + create table public.cb_errors(error_msg text, timed_out bool); + """ + )) + sess.commit() + + # port 1 is closed, the request fails with a connection error + sess.execute(text( + """ + select net.http_request( + 'GET', + 'http://localhost:1', + on_error := 'insert into public.cb_errors values ($1, $2)' + ); + """ + )) + sess.commit() + + wait_until_queue_empty(sess) + + row = sess.execute(text( + """ + select error_msg, timed_out from public.cb_errors; + """ + )).fetchone() + + assert row is not None + assert len(row[0]) > 0 + assert row[1] == False + + sess.execute(text("drop table public.cb_errors;")) + sess.commit() + + (up,) = sess.execute(text( + """ + select is_worker_up(); + """ + )).fetchone() + + assert up == True + + +def test_http_request_failing_callback_does_not_kill_worker(sess): + """a callback that raises doesn't abort the batch or crash the worker""" + + sess.execute(text( + """ + select net.http_request( + 'GET', + 'http://localhost:8080/pathological?status=200', + on_success := 'select 1/0' + ); + """ + )) + sess.commit() + + wait_until_queue_empty(sess) + + (up,) = sess.execute(text( + """ + select is_worker_up(); + """ + )).fetchone() + + assert up == True + + # the worker keeps processing new requests afterwards + (request_id,) = sess.execute(text( + """ + select net.http_get('http://localhost:8080/pathological?status=200'); + """ + )).fetchone() + sess.commit() + + wait_until_queue_empty(sess) + + (count,) = sess.execute( + text( + """ + select count(*) from net._http_response where id = :request_id; + """ + ), + {"request_id": request_id}, + ).fetchone() + + assert count == 1 + + +def test_http_request_callback_runs_as_calling_role(sess): + """callbacks are executed as the role that enqueued the request""" + + sess.execute(text( + """ + drop table if exists public.cb_ident; + create table public.cb_ident(who text); + drop role if exists cb_limited; + create role cb_limited; + grant insert on public.cb_ident to cb_limited; + """ + )) + sess.commit() + + sess.execute(text( + """ + set role cb_limited; + select net.http_request( + 'GET', + 'http://localhost:8080/pathological?status=200', + on_success := 'insert into public.cb_ident select current_user' + ); + reset role; + """ + )) + sess.commit() + + wait_until_queue_empty(sess) + + (who,) = sess.execute(text( + """ + select who from public.cb_ident; + """ + )).fetchone() + + assert who == "cb_limited" + + sess.execute(text( + """ + drop table public.cb_ident; + drop role cb_limited; + """ + )) + sess.commit() + + +def test_http_request_callback_cannot_bypass_privileges(sess): + """a callback enqueued by a low-privilege role can't write where the role can't""" + + sess.execute(text( + """ + drop table if exists public.cb_secured; + create table public.cb_secured(status int); + drop role if exists cb_limited2; + create role cb_limited2; + -- note: no grant on cb_secured + """ + )) + sess.commit() + + sess.execute(text( + """ + set role cb_limited2; + select net.http_request( + 'GET', + 'http://localhost:8080/pathological?status=200', + on_success := 'insert into public.cb_secured values ($1)' + ); + reset role; + """ + )) + sess.commit() + + wait_until_queue_empty(sess) + + # the callback failed with permission denied, so no row was written + (count,) = sess.execute(text( + """ + select count(*) from public.cb_secured; + """ + )).fetchone() + + assert count == 0 + + (up,) = sess.execute(text( + """ + select is_worker_up(); + """ + )).fetchone() + + assert up == True + + sess.execute(text( + """ + drop table public.cb_secured; + drop role cb_limited2; + """ + )) + sess.commit() + + +def test_legacy_functions_still_store_responses(sess): + """net.http_get keeps the existing table-based behavior""" + + (request_id,) = sess.execute(text( + """ + select net.http_get('http://localhost:8080/pathological?status=200'); + """ + )).fetchone() + sess.commit() + + wait_until_queue_empty(sess) + + (count,) = sess.execute( + text( + """ + select count(*) from net._http_response where id = :request_id; + """ + ), + {"request_id": request_id}, + ).fetchone() + + assert count == 1 + + +def test_callback_columns_missing_fallback(sess): + """the worker keeps processing requests when the callback columns don't exist + + This simulates a binary that is newer than the installed extension version: + the shared library is replaced on upgrade while `ALTER EXTENSION pg_net + UPDATE` might not have been executed yet. + """ + + sess.execute(text( + """ + alter table net.http_request_queue + drop column use_callbacks, + drop column on_success, + drop column on_error, + drop column calling_role; + """ + )) + sess.commit() + + # emulate the 0.20.4 net.http_get function, which doesn't reference the + # callback columns (the 0.21.0 function can't run without the columns) + (request_id,) = sess.execute(text( + """ + insert into net.http_request_queue(method, url, headers, timeout_milliseconds) + values ('GET', 'http://localhost:8080/pathological?status=200', '{}', 5000) + returning id; + """ + )).fetchone() + + sess.execute(text( + """ + select net.wake(); + """ + )) + sess.commit() + + wait_until_queue_empty(sess) + + # without the columns, the legacy behavior applies: the response is stored + (count,) = sess.execute( + text( + """ + select count(*) from net._http_response where id = :request_id; + """ + ), + {"request_id": request_id}, + ).fetchone() + + assert count == 1 + + # restore the columns, as `alter extension pg_net update` would + sess.execute(text( + """ + alter table net.http_request_queue + add column use_callbacks bool not null default false, + add column on_success text, + add column on_error text, + add column calling_role text; + """ + )) + sess.commit() + + sess.execute(text( + """ + select net.http_request('GET', 'http://localhost:8080/pathological?status=200'); + """ + )) + sess.commit() + + wait_until_queue_empty(sess) + + (up,) = sess.execute(text( + """ + select is_worker_up(); + """ + )).fetchone() + + assert up == True