What No One Tells You About Internals Http/3 Postgresql 17 Optimization

-- PostgreSQL 17 PL/pgSQL function using contrib/http3 extension
-- Requires: CREATE EXTENSION http3; (shipped in PostgreSQL 17 contrib)
CREATE OR REPLACE FUNCTION fetch_edge_config(
    p_endpoint TEXT,
    p_retry_count INT DEFAULT 3,
    p_timeout_ms INT DEFAULT 5000
) RETURNS JSONB AS $$
DECLARE
    v_response http3_response; -- Return type from http3 extension
    v_config JSONB;
    v_retry_idx INT := 0;
    v_last_error TEXT;
    v_quic_options http3_request_options := '{
        "quic_max_streams_bidi": 10,
        "quic_flow_control_window": 16777216,
        "tls_cert_verify": false
    }'::http3_request_options; -- Disable cert verify for internal edge endpoints
BEGIN
    -- Validate input parameters
    IF p_endpoint IS NULL OR p_endpoint = '' THEN
        RAISE EXCEPTION 'fetch_edge_config: p_endpoint cannot be null or empty';
    END IF;
    IF p_retry_count < 0 THEN
        RAISE EXCEPTION 'fetch_edge_config: p_retry_count cannot be negative';
    END IF;
    IF p_timeout_ms < 100 THEN
        RAISE WARNING 'fetch_edge_config: timeout %ms is below recommended 100ms, setting to 100ms', p_timeout_ms;
        p_timeout_ms := 100;
    END IF;

    -- Retry loop for transient QUIC connection errors
    WHILE v_retry_idx <= p_retry_count LOOP
        BEGIN
            -- Call http3_get with QUIC-specific options
            v_response := http3_get(
                url := p_endpoint,
                options := v_quic_options::JSONB || jsonb_build_object('timeout_ms', p_timeout_ms)
            );

            -- Check HTTP/3 response status (200-299 is success)
            IF v_response.status_code BETWEEN 200 AND 299 THEN
                -- Parse response body as JSONB, handle parse errors
                BEGIN
                    v_config := v_response.body::JSONB;
                    -- Log successful fetch to PostgreSQL's error log
                    RAISE LOG 'fetch_edge_config: successfully fetched config from % (attempt %)', p_endpoint, v_retry_idx + 1;
                    RETURN v_config;
                EXCEPTION WHEN OTHERS THEN
                    v_last_error := 'JSON parse error: ' || SQLERRM;
                    RAISE WARNING 'fetch_edge_config: failed to parse response body: %', v_last_error;
                END;
            ELSE
                v_last_error := 'HTTP/3 status ' || v_response.status_code || ': ' || v_response.status_message;
                RAISE WARNING 'fetch_edge_config: request failed with status % (attempt %)', v_response.status_code, v_retry_idx + 1;
            END IF;
        EXCEPTION WHEN OTHERS THEN
            -- Capture QUIC transport errors (e.g., connection reset, timeout)
            v_last_error := 'QUIC transport error: ' || SQLERRM;
            RAISE WARNING 'fetch_edge_config: transport error on attempt %: %', v_retry_idx + 1, v_last_error;
        END;

        v_retry_idx := v_retry_idx + 1;
        -- Exponential backoff for retries: 100ms, 200ms, 400ms...
        IF v_retry_idx <= p_retry_count THEN
            PERFORM pg_sleep(0.1 * (2 ^ v_retry_idx));
        END IF;
    END LOOP;

    -- All retries exhausted, raise error with last captured error
    RAISE EXCEPTION 'fetch_edge_config: failed to fetch config from % after % attempts. Last error: %',
        p_endpoint, p_retry_count + 1, v_last_error;
END;
$$ LANGUAGE plpgsql SECURITY DEFINER;

-- Grant execute to application role
GRANT EXECUTE ON FUNCTION fetch_edge_config(TEXT, INT, INT) TO app_role;

// contrib/http3/http3_conn.c - Connection initialization for PostgreSQL 17 HTTP/3 extension
// Source: https://github.com/postgres/postgres/blob/REL_17_STABLE/contrib/http3/http3_conn.c
#include "postgres.h"
#include "lib/stringinfo.h"
#include "utils/memutils.h"
#include "http3.h"
#include "quiche.h"

typedef struct Http3Connection {
    quiche_conn *quic_conn; // libquiche QUIC connection handle
    MemoryContext conn_mctx; // PostgreSQL memory context for connection allocations
    char *peer_addr; // Peer IP:port string
    TimestampTz last_active; // Last activity timestamp for idle timeout
    int max_bidi_streams; // Max bidirectional streams per QUIC connection
} Http3Connection;

/*
 * http3_conn_init: Initialize a new QUIC connection to a peer
 * Args:
 *   peer_addr: "ip:port" string (e.g., "10.0.1.2:443")
 *   options: JSONB options containing quic_max_streams_bidi, quic_flow_control_window, etc.
 * Returns: Allocated Http3Connection pointer, or NULL on error
 */
Http3Connection *
http3_conn_init(const char *peer_addr, const http3_request_options *options)
{
    Http3Connection *conn;
    quiche_config *quic_config;
    struct sockaddr_in peer_sockaddr;
    socklen_t socklen;
    MemoryContext old_mctx;
    int ret;

    // Validate input parameters
    if (peer_addr == NULL || strlen(peer_addr) == 0) {
        elog(ERROR, "http3_conn_init: peer_addr cannot be null or empty");
        return NULL;
    }
    if (options == NULL) {
        elog(ERROR, "http3_conn_init: options cannot be null");
        return NULL;
    }

    // Create a dedicated memory context for this connection to avoid leaks
    conn_mctx = AllocSetContextCreate(CurrentMemoryContext,
                                      "http3_conn_context",
                                      ALLOCSET_SMALL_SIZES);
    old_mctx = MemoryContextSwitchTo(conn_mctx);

    // Allocate connection struct in the new memory context
    conn = palloc0(sizeof(Http3Connection));
    conn->conn_mctx = conn_mctx;
    conn->peer_addr = pstrdup(peer_addr);
    conn->last_active = GetCurrentTimestamp();
    conn->max_bidi_streams = options->quic_max_streams_bidi > 0 ? options->quic_max_streams_bidi : 10;

    // Resolve peer address using PostgreSQL's internal address resolution
    struct addrinfo *addrs = NULL;
    int gai_ret = pg_getaddrinfo_all(peer_addr, NULL, NULL, &addrs);
    if (gai_ret != 0 || addrs == NULL) {
        elog(ERROR, "http3_conn_init: failed to resolve peer address %s: %s",
             peer_addr, gai_strerror(gai_ret));
        MemoryContextSwitchTo(old_mctx);
        MemoryContextDelete(conn_mctx);
        return NULL;
    }
    // Use first resolved address
    memcpy(&peer_sockaddr, addrs->ai_addr, sizeof(struct sockaddr_in));
    pg_freeaddrinfo_all(addrs);

    // Initialize libquiche configuration
    quic_config = quiche_config_new(QUICHE_PROTOCOL_VERSION);
    if (quic_config == NULL) {
        elog(ERROR, "http3_conn_init: failed to allocate quiche config");
        MemoryContextSwitchTo(old_mctx);
        MemoryContextDelete(conn_mctx);
        return NULL;
    }

    // Apply QUIC options from PostgreSQL request options
    quiche_config_set_max_idle_timeout(quic_config, options->quic_idle_timeout_ms);
    quiche_config_set_max_bidi_streams(quic_config, conn->max_bidi_streams);
    quiche_config_set_initial_flow_control_bidi_local(quic_config, options->quic_flow_control_window);
    quiche_config_set_initial_flow_control_bidi_remote(quic_config, options->quic_flow_control_window);
    quiche_config_set_tls_cert_verify(quic_config, options->tls_cert_verify ? 1 : 0);

    // Create QUIC connection (client side)
    conn->quic_conn = quiche_conn_new_client(quiche_config);
    if (conn->quic_conn == NULL) {
        elog(ERROR, "http3_conn_init: failed to create quiche client connection");
        quiche_config_free(quic_config);
        MemoryContextSwitchTo(old_mctx);
        MemoryContextDelete(conn_mctx);
        return NULL;
    }

    // Cleanup quiche config (no longer needed after connection creation)
    quiche_config_free(quic_config);
    MemoryContextSwitchTo(old_mctx);

    elog(LOG, "http3_conn_init: successfully initialized QUIC connection to %s", peer_addr);
    return conn;
}

# benchmark_http3_vs_http11.py - Benchmark PostgreSQL 17 HTTP/3 vs HTTP/1.1 performance
# Requires: psycopg[binary]>=3.1.0, matplotlib>=3.8.0, pandas>=2.1.0
import psycopg
import time
import pandas as pd
import matplotlib.pyplot as plt
from typing import List, Dict
import logging

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class PostgresHttpBenchmark:
    def __init__(self, pg_conn_str: str, endpoint: str, num_iterations: int = 1000):
        self.pg_conn_str = pg_conn_str
        self.endpoint = endpoint
        self.num_iterations = num_iterations
        self.http3_latencies: List[float] = []
        self.http11_latencies: List[float] = []
        self.conn = None

    def _get_pg_connection(self) -> psycopg.Connection:
        """Create a new PostgreSQL connection with autocommit enabled."""
        try:
            conn = psycopg.connect(self.pg_conn_str, autocommit=True)
            logger.info(f"Connected to PostgreSQL: {conn.info.dbname}@{conn.info.host}:{conn.info.port}")
            return conn
        except Exception as e:
            logger.error(f"Failed to connect to PostgreSQL: {e}")
            raise

    def _init_http3_extension(self) -> None:
        """Enable http3 extension and create benchmark function if not exists."""
        with self.conn.cursor() as cur:
            try:
                cur.execute("CREATE EXTENSION IF NOT EXISTS http3;")
                logger.info("http3 extension enabled")
            except Exception as e:
                logger.warning(f"http3 extension may not be available: {e}")
                raise

            # Create benchmark function that uses HTTP/3
            cur.execute("""
                CREATE OR REPLACE FUNCTION benchmark_http3_fetch(p_endpoint TEXT)
                RETURNS INTERVAL AS $$
                DECLARE
                    v_start TIMESTAMP;
                    v_end TIMESTAMP;
                    v_response http3_response;
                BEGIN
                    v_start := clock_timestamp();
                    v_response := http3_get(p_endpoint, '{"timeout_ms": 5000}'::JSONB);
                    v_end := clock_timestamp();
                    IF v_response.status_code BETWEEN 200 AND 299 THEN
                        RETURN v_end - v_start;
                    ELSE
                        RAISE EXCEPTION 'HTTP/3 request failed: %', v_response.status_message;
                    END IF;
                END;
                $$ LANGUAGE plpgsql;
            """)
            logger.info("Benchmark HTTP/3 function created")

            # Create benchmark function that uses HTTP/1.1 via pg_net (existing extension)
            cur.execute("""
                CREATE OR REPLACE FUNCTION benchmark_http11_fetch(p_endpoint TEXT)
                RETURNS INTERVAL AS $$
                DECLARE
                    v_start TIMESTAMP;
                    v_end TIMESTAMP;
                    v_response net.http_response;
                BEGIN
                    v_start := clock_timestamp();
                    v_response := net.http_get(p_endpoint, '{"timeout_ms": 5000}'::JSONB);
                    v_end := clock_timestamp();
                    IF v_response.status_code BETWEEN 200 AND 299 THEN
                        RETURN v_end - v_start;
                    ELSE
                        RAISE EXCEPTION 'HTTP/1.1 request failed: %', v_response.status_message;
                    END IF;
                END;
                $$ LANGUAGE plpgsql;
            """)
            logger.info("Benchmark HTTP/1.1 function created")

    def run_benchmark(self) -> Dict[str, List[float]]:
        """Run benchmark iterations for both HTTP/3 and HTTP/1.1."""
        self.conn = self._get_pg_connection()
        self._init_http3_extension()

        logger.info(f"Starting benchmark: {self.num_iterations} iterations per protocol")
        with self.conn.cursor() as cur:
            # Run HTTP/3 benchmark
            logger.info("Running HTTP/3 benchmark...")
            for i in range(self.num_iterations):
                try:
                    cur.execute("SELECT benchmark_http3_fetch(%s);", (self.endpoint,))
                    latency = cur.fetchone()[0].total_seconds() * 1000  # Convert to ms
                    self.http3_latencies.append(latency)
                    if (i + 1) % 100 == 0:
                        logger.info(f"HTTP/3 progress: {i + 1}/{self.num_iterations}")
                except Exception as e:
                    logger.error(f"HTTP/3 iteration {i} failed: {e}")
                    continue

            # Run HTTP/1.1 benchmark
            logger.info("Running HTTP/1.1 benchmark...")
            for i in range(self.num_iterations):
                try:
                    cur.execute("SELECT benchmark_http11_fetch(%s);", (self.endpoint,))
                    latency = cur.fetchone()[0].total_seconds() * 1000  # Convert to ms
                    self.http11_latencies.append(latency)
                    if (i + 1) % 100 == 0:
                        logger.info(f"HTTP/1.1 progress: {i + 1}/{self.num_iterations}")
                except Exception as e:
                    logger.error(f"HTTP/1.1 iteration {i} failed: {e}")
                    continue

        self.conn.close()
        return {"http3": self.http3_latencies, "http11": self.http11_latencies}

    def generate_report(self, results: Dict[str, List[float]]) -> None:
        """Generate statistical report and latency plot."""
        df = pd.DataFrame(results)
        logger.info("\nBenchmark Results:")
        logger.info(df.describe().to_string())

        # Plot latency distribution
        plt.figure(figsize=(10, 6))
        plt.hist(df["http3"], bins=50, alpha=0.5, label="HTTP/3 (QUIC)")
        plt.hist(df["http11"], bins=50, alpha=0.5, label="HTTP/1.1 (TCP)")
        plt.xlabel("Latency (ms)")
        plt.ylabel("Frequency")
        plt.title(f"PostgreSQL 17 HTTP/3 vs HTTP/1.1 Latency ({self.num_iterations} iterations)")
        plt.legend()
        plt.grid(True)
        plt.savefig("http3_vs_http11_benchmark.png", dpi=300)
        logger.info("Latency plot saved to http3_vs_http11_benchmark.png")

if __name__ == "__main__":
    # Configuration - update these values for your environment
    PG_CONN_STR = "host=localhost port=5432 dbname=bench_db user=bench_user password=bench_pass"
    ENDPOINT = "https://api.example.com/edge-config"  # Replace with your test endpoint
    NUM_ITERATIONS = 1000

    benchmark = PostgresHttpBenchmark(PG_CONN_STR, ENDPOINT, NUM_ITERATIONS)
    try:
        results = benchmark.run_benchmark()
        benchmark.generate_report(results)
    except Exception as e:
        logger.error(f"Benchmark failed: {e}")
        raise