HTTP policy control (concurrency, etc)

CHANGELOG.md

@@ -23,6 +23,11 @@ Pre-1.0 note: while `pg_durable` is in major version `0`, minor releases may inc
 
 ### Added
 
+- **HTTP execution limits:** `pg_durable.max_http_concurrency` bounds ordinary
+  and multipart HTTP executions together across all origin databases, while
+  `pg_durable.max_activity_concurrency` controls total activity capacity.
+  Both default to two and require a server restart. Capacity-blocked HTTP work
+  stays durably queued without occupying execution or origin-connection slots.
 - **Multi-database installations (#401):** satellites share one control runtime
   while keeping local `df` APIs, metadata, grants, RLS, endpoints and credentials.
   SQL defaults to the submitting database unless explicitly targeted and retains

Cargo.toml

@@ -73,6 +73,10 @@ tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 [dev-dependencies]
 pgrx-tests = "=0.16.1"
 
+# Temporary: pin the admission hook until it is available in a released version.
+[patch.crates-io]
+duroxide = { git = "https://github.com/microsoft/duroxide", rev = "a4cc556ce01dc5beca48e9a4cca424b6877153de" }
+
 [profile.dev]
 panic = "unwind"
 

USER_GUIDE.md

@@ -3008,6 +3008,43 @@ for the full syntax and default domain list.
 pg_durable uses separate limits for control management, provider, origin metadata,
 SQL execution, and independent-start connections.
 
+### Activity and HTTP Concurrency
+
+Connection budgets are separate from execution capacity:
+
+| Setting | Default | Scope |
+|---------|---------|-------|
+| `pg_durable.max_activity_concurrency` | 2 | All concurrently executing activities, including SQL, HTTP, and metadata work |
+| `pg_durable.max_http_concurrency` | 2 | HTTP and multipart HTTP activities combined |
+
+Both settings accept integers from 1 to 1000 and require a PostgreSQL restart.
+The HTTP ceiling is shared across users, workflows, and databases served by the
+cluster's single runtime. It counts an HTTP execution from request preparation
+through response-body processing and optional sink storage, not idle pooled
+connections. It is not a requests-per-second limit.
+
+Requests waiting for HTTP capacity stay in the durable queue. They do not hold
+an execution slot or an origin-database connection, fail due to capacity
+contention, or consume their request timeout while waiting. Cancellation still
+applies. Actual request failures retain their usual behavior.
+
+For non-HTTP headroom, set the HTTP ceiling below total activity capacity:
+
+```ini
+pg_durable.max_activity_concurrency = 8
+pg_durable.max_http_concurrency = 4
+```
+
+Increasing activity capacity does not increase database connection budgets.
+If both settings are equal, active HTTP requests can occupy all activity slots;
+the unchanged defaults preserve the previous two-activity execution capacity.
+There is no strict fairness guarantee between workflows.
+
+Pending requests use the configured limits after restart; recorded results
+replay unchanged. A crash or cancellation cannot guarantee that a remote server
+stops processing an already-sent request. Limits bound local executions, not
+remote effects, and do not provide exactly-once HTTP delivery.
+
 ### Connection Architecture
 
 The main connection categories are:

docs/E2E_TESTING.md

@@ -89,6 +89,7 @@ restart with the phase's configuration, not a rebuild.
 | `48_http_allow_all.sql` | `http-allow-all` | Unrestricted mode admits plaintext HTTP, unlisted domains, and private destinations for both HTTP activities |
 | `69_http_allowed_domains.sql` | `http-custom-domains` | Custom allow-list enforcement, malformed startup rejection, protected mode settings, and restart-only policy changes |
 | `70_http_allowed_domains_empty.sql` | `http-empty-domains` | Empty restricted-mode allow-list denies all domains |
+| `90_http_concurrency.sql` | `http-concurrency` | Measures HTTP ceilings of one and two with four activity slots using slow response bodies; covers both HTTP functions, satellite origins, SQL headroom, queued cancellation, queue-independent request timeouts, and permit release after failure |
 
 ## Test Structure
 

docs/api-reference.md

@@ -893,6 +893,27 @@ These settings are configured via `ALTER SYSTEM SET` or `postgresql.conf`. See e
 
 ---
 
+### pg_durable.max_activity_concurrency / pg_durable.max_http_concurrency
+
+| Setting | Meaning | Default | Range |
+|---------|---------|---------|-------|
+| `pg_durable.max_activity_concurrency` | Concurrent activities in the shared runtime, including SQL, HTTP, and metadata | `2` | `1` to `1000` |
+| `pg_durable.max_http_concurrency` | Concurrent HTTP and multipart HTTP activities combined, across all origin databases | `2` | `1` to `1000` |
+
+Both are `POSTMASTER` settings and require a PostgreSQL restart. An HTTP slot is
+held through preparation, response-body handling, and optional sink storage.
+Requests awaiting capacity remain queued without occupying activity slots or
+origin connections; admission waiting does not consume the request timeout.
+Contention does not fail the request or count as an execution retry.
+
+HTTP concurrency cannot exceed total activity capacity. Set the HTTP ceiling
+lower than activity capacity to leave headroom for non-HTTP work. These settings
+do not raise connection budgets or impose request-rate limits. No extension SQL
+update is required for existing installations to use these settings with the
+new binary. See [Activity and HTTP Concurrency](../USER_GUIDE.md#activity-and-http-concurrency).
+
+---
+
 ### pg_durable.http_security
 
 Server-wide outbound HTTP policy for `df.http()` and `df.http_multipart()`.

docs/http-security.md

@@ -566,6 +566,20 @@ retrying an activity alone does not clear the error. Ordinary DNS, connection,
 and request-timeout failures happen after construction and are not cached this
 way, so they do not require a worker restart.
 
+### 6.4 HTTP execution capacity
+
+`pg_durable.max_http_concurrency` limits ordinary and multipart HTTP executions
+together across all origin databases. Capacity is reserved before routing,
+privilege checks, or credential lookup and retained through response processing
+and sink storage. Queued requests hold no HTTP or origin-connection reservation.
+Admission does not replace any execution-time authorization or destination check.
+
+The limit defaults to two and requires a server restart to change. It is
+separate from connection-pool size and from the total activity capacity set by
+`pg_durable.max_activity_concurrency`. See
+[Activity and HTTP Concurrency](../USER_GUIDE.md#activity-and-http-concurrency)
+for headroom, timeout, cancellation, and crash semantics.
+
 ---
 
 ## 7. Audit Logging

scripts/test-e2e-docker.sh

@@ -60,6 +60,7 @@ SKIP_TESTS=(
     # Require http-allow-all and the local runner's TLS mock and environment.
     "80_multi_database_http_origin"
     "85_multi_database_http_admission"
+    "90_http_concurrency"
 )
 
 # Defaults

scripts/test-e2e-local.sh

@@ -94,6 +94,7 @@ ALL_PHASES=(
     "http-empty-domains"
     "http-disabled"
     "http-allow-all"
+    "http-concurrency"
 )
 
 PGRX_HOME="$HOME/.pgrx"
@@ -189,6 +190,9 @@ phase_label() {
         http-allow-all)
             echo "HTTP unrestricted startup policy"
             ;;
+        http-concurrency)
+            echo "HTTP concurrency and nonblocking admission"
+            ;;
         *)
             echo "$1"
             ;;
@@ -239,6 +243,9 @@ phase_for_test() {
         48_http_allow_all|80_multi_database_http_origin|85_multi_database_http_admission)
             echo "http-allow-all"
             ;;
+        90_http_concurrency)
+            echo "http-concurrency"
+            ;;
         *)
             echo "standard"
             ;;
@@ -665,7 +672,7 @@ configure_phase() {
             set_conf_line "pg_durable.enable_superuser_instances" "on"
             set_conf_line "pg_durable.http_allowed_domains" "'example.com'"
             ;;
-        http-empty-domains|http-allow-all)
+        http-empty-domains|http-allow-all|http-concurrency)
             set_conf_line "shared_preload_libraries" "'pg_durable'"
             set_conf_line "pg_durable.worker_role" "'postgres'"
             set_conf_line "pg_durable.database" "'postgres'"
@@ -674,6 +681,10 @@ configure_phase() {
             if [ "$phase" = "http-allow-all" ]; then
                 set_conf_line "pg_durable.max_user_connections" "1"
                 set_conf_line "pg_durable.reconcile_interval" "0"
+            elif [ "$phase" = "http-concurrency" ]; then
+                set_conf_line "pg_durable.max_activity_concurrency" "4"
+                set_conf_line "pg_durable.max_http_concurrency" "1"
+                set_conf_line "pg_durable.reconcile_interval" "0"
             fi
             ;;
     esac
@@ -682,7 +693,7 @@ configure_phase() {
         http-disabled)
             set_conf_line "pg_durable.http_security" "'disabled'"
             ;;
-        http-allow-all)
+        http-allow-all|http-concurrency)
             set_conf_line "pg_durable.http_security" "'unrestricted'"
             ;;
     esac
@@ -762,7 +773,7 @@ prepare_phase() {
         reconcile|maintenance-backlog)
             wait_for_worker_ready
             ;;
-        http-custom-domains|http-empty-domains|http-disabled|http-allow-all)
+        http-custom-domains|http-empty-domains|http-disabled|http-allow-all|http-concurrency)
             ensure_e2e_role
             wait_for_worker_ready
             ;;
@@ -935,7 +946,7 @@ run_http_origin_test() (
         stop_server
         kill "$mock_pid" 2>/dev/null || true
         wait "$mock_pid" 2>/dev/null || true
-        rm -f -- "$work_dir/port" "$work_dir/server.log" "$work_dir/key.pem" "$work_dir/cert.pem" "$work_dir/cert.log" "$work_dir/requests.json" "$work_dir/requests.tmp"
+        rm -f -- "$work_dir/port" "$work_dir/server.log" "$work_dir/key.pem" "$work_dir/cert.pem" "$work_dir/cert.log" "$work_dir/requests.json" "$work_dir/requests.tmp" "$work_dir/capacity-release"
         rmdir -- "$work_dir"
         if [ -n "$previous_ssl_cert_file" ]; then
             export SSL_CERT_FILE="$previous_ssl_cert_file"
@@ -953,6 +964,7 @@ run_http_origin_test() (
             PGDURABLE_TEST_HTTP_PORT=$(cat "$work_dir/port")
             export PGDURABLE_TEST_SERVER_LOG="$LOG_FILE"
             export PGDURABLE_TEST_HTTP_COUNTS="$work_dir/requests.json"
+            export PGDURABLE_TEST_HTTP_RELEASE="$work_dir/capacity-release"
             export SSL_CERT_FILE="$work_dir/cert.pem"
             restart_server
             wait_for_worker_ready
@@ -977,6 +989,16 @@ run_test_file() {
 
     test_name=$(basename "$test_file" .sql)
 
+    if [ "$test_name" = "90_http_concurrency" ] && [ -z "${PGDURABLE_TEST_HTTP_PORT:-}" ]; then
+        local http_limit
+        for http_limit in 1 2; do
+            echo "  Testing HTTP ceiling $http_limit with four activity slots"
+            set_conf_line "pg_durable.max_http_concurrency" "$http_limit"
+            run_http_origin_test "$test_file" || return $?
+        done
+        return 0
+    fi
+
     if [[ "$test_name" = "80_multi_database_http_origin" || "$test_name" = "85_multi_database_http_admission" ]] && [ -z "${PGDURABLE_TEST_HTTP_PORT:-}" ]; then
         run_http_origin_test "$test_file"
         return $?

src/lib.rs

@@ -32,6 +32,8 @@ pub static MAX_MANAGEMENT_CONNECTIONS: GucSetting<i32> = GucSetting::<i32>::new(
 pub static MAX_ORIGIN_CONNECTIONS: GucSetting<i32> = GucSetting::<i32>::new(12);
 pub static MAX_DUROXIDE_CONNECTIONS: GucSetting<i32> = GucSetting::<i32>::new(10);
 pub static MAX_USER_CONNECTIONS: GucSetting<i32> = GucSetting::<i32>::new(10);
+pub static MAX_ACTIVITY_CONCURRENCY: GucSetting<i32> = GucSetting::<i32>::new(2);
+pub static MAX_HTTP_CONCURRENCY: GucSetting<i32> = GucSetting::<i32>::new(2);
 pub static MAX_NEW_TRANSACTION_STARTS: GucSetting<i32> = GucSetting::<i32>::new(2);
 pub static EXECUTION_ACQUIRE_TIMEOUT: GucSetting<i32> = GucSetting::<i32>::new(30);
 pub static NEW_TRANSACTION_START_TIMEOUT: GucSetting<i32> = GucSetting::<i32>::new(5);
@@ -229,6 +231,28 @@ pub extern "C-unwind" fn _PG_init() {
         GucFlags::default(),
     );
 
+    GucRegistry::define_int_guc(
+        c"pg_durable.max_activity_concurrency",
+        c"Maximum number of concurrent activities in the shared background worker runtime",
+        c"Includes SQL, HTTP, and metadata activities. Requires a server restart.",
+        &MAX_ACTIVITY_CONCURRENCY,
+        1,
+        1000,
+        GucContext::Postmaster,
+        GucFlags::default(),
+    );
+
+    GucRegistry::define_int_guc(
+        c"pg_durable.max_http_concurrency",
+        c"Maximum number of concurrent HTTP and multipart HTTP activities across all databases",
+        c"Includes request preparation, response handling, and sink storage. Requires a server restart.",
+        &MAX_HTTP_CONCURRENCY,
+        1,
+        1000,
+        GucContext::Postmaster,
+        GucFlags::default(),
+    );
+
     GucRegistry::define_int_guc(
         c"pg_durable.max_new_transaction_starts",
         c"Maximum concurrent transaction_mode => 'new' df.start() loopback launch sessions per database",
@@ -3164,6 +3188,23 @@ mod tests {
         );
     }
 
+    #[pg_test]
+    fn test_activity_concurrency_gucs() {
+        assert_eq!(crate::MAX_ACTIVITY_CONCURRENCY.get(), 2);
+        assert_eq!(crate::MAX_HTTP_CONCURRENCY.get(), 2);
+        assert_eq!(
+            Spi::get_one::<i64>(
+                "SELECT count(*) FROM pg_settings
+                 WHERE name IN ('pg_durable.max_activity_concurrency',
+                                'pg_durable.max_http_concurrency')
+                   AND context = 'postmaster'
+                   AND min_val = '1' AND max_val = '1000' AND boot_val = '2'"
+            )
+            .unwrap(),
+            Some(2)
+        );
+    }
+
     #[pg_test]
     fn test_host_guc_boot_default_is_unset() {
         let boot_val = Spi::get_one::<String>(

src/registry.rs

@@ -5,24 +5,41 @@
 
 use std::sync::Arc;
 
-use duroxide::{runtime::registry::ActivityRegistry, ActivityContext, OrchestrationRegistry};
+use duroxide::{
+    runtime::{registry::ActivityRegistry, ActivityAdmission},
+    ActivityContext, OrchestrationRegistry,
+};
 use sqlx::PgPool;
 use tokio::sync::Semaphore;
 
 use crate::activities;
 use crate::orchestrations;
 use crate::ssrf::HttpPolicy;
 
+fn admit_http(semaphore: &Arc<Semaphore>) -> Result<ActivityAdmission, String> {
+    match semaphore.clone().try_acquire_owned() {
+        Ok(permit) => Ok(ActivityAdmission::admit(permit)),
+        Err(tokio::sync::TryAcquireError::NoPermits) => Ok(ActivityAdmission::Defer(
+            std::time::Duration::from_millis(100),
+        )),
+        Err(tokio::sync::TryAcquireError::Closed) => {
+            Err("HTTP admission semaphore closed unexpectedly".to_string())
+        }
+    }
+}
+
 /// Create the activity registry with all registered activities
 pub fn create_activity_registry(
     pool: Arc<PgPool>,
     semaphore: Arc<Semaphore>,
     http_policy: Arc<HttpPolicy>,
+    http_capacity: Arc<Semaphore>,
 ) -> ActivityRegistry {
     let sql_semaphore = semaphore.clone();
     let http_semaphore = semaphore.clone();
     let multipart_semaphore = semaphore;
     let multipart_policy = http_policy.clone();
+    let multipart_capacity = http_capacity.clone();
     let router = Arc::new(crate::origin::Router::new(pool));
     let sql_pool = router.clone();
     let graph_pool = router.clone();
@@ -143,8 +160,9 @@ pub fn create_activity_registry(
                 }
             },
         )
-        .register(
+        .register_with_admission(
             activities::execute_http::NAME,
+            move |_ctx, _input| admit_http(&http_capacity),
             move |ctx: ActivityContext, config_json: String| {
                 let pool = http_pool.clone();
                 let semaphore = http_semaphore.clone();
@@ -164,8 +182,9 @@ pub fn create_activity_registry(
                 }
             },
         )
-        .register(
+        .register_with_admission(
             activities::execute_multipart::NAME,
+            move |_ctx, _input| admit_http(&multipart_capacity),
             move |ctx: ActivityContext, config_json: String| {
                 let pool = multipart_pool.clone();
                 let semaphore = multipart_semaphore.clone();