feat(httputil): add HTTP request throttling for memory-constrained devices

README.md

@@ -21,6 +21,7 @@ Bluetooth is in progress, along with more processors.
 - TCP and UDP Berkeley sockets API
 - Raw Ethernet frame send/receive
 - MQTT client support (uses [`natiu-mqtt`](https://github.com/soypat/natiu-mqtt))
+- HTTP concurrency throttling for memory-constrained devices (`httputil` package)
 - QEMU simulation target for ESP32-C3
 
 ## How to use
@@ -38,6 +39,7 @@ import (
 
 	"tinygo.org/x/drivers/netdev"
 	nl "tinygo.org/x/drivers/netlink"
+	"tinygo.org/x/espradio/httputil"
 	link "tinygo.org/x/espradio/netlink"
 )
 
@@ -69,7 +71,9 @@ func main() {
 	h, _ := link.Addr()
 	host := h.String()
 	println("HTTP server listening on http://" + host + port)
-	err = http.ListenAndServe(host+port, nil)
+
+	// Throttle to 2 concurrent requests to avoid OOM on ESP32.
+	err = http.ListenAndServe(host+port, httputil.ThrottleHandler(2, nil))
 	for err != nil {
 		println("error:", err.Error())
 		time.Sleep(5 * time.Second)
@@ -361,6 +365,22 @@ AP: rems RSSI -79
 
 Starts the ESP32 radio.
 
+## httputil
+
+The `httputil` package provides HTTP utilities for memory-constrained devices. The main utility is `ThrottleHandler`, which limits concurrent HTTP request processing to a fixed number of slots. When all slots are in use, excess requests are immediately rejected with `503 Service Unavailable` instead of being queued — this prevents unbounded goroutine/memory growth on ESP32-class hardware.
+
+```go
+import "tinygo.org/x/espradio/httputil"
+
+mux := http.NewServeMux()
+mux.HandleFunc("/", handler)
+
+// Allow at most 2 concurrent requests; excess get 503.
+err := http.ListenAndServe(":80", httputil.ThrottleHandler(2, mux))
+```
+
+All responses include `Connection: close` to discourage HTTP/1.1 keep-alive pipelining which consumes memory for idle connections.
+
 ## Architecture
 
 ```mermaid

examples/apwebserver/apwebserver.go

@@ -16,6 +16,7 @@ import (
 
 	"tinygo.org/x/drivers/netdev"
 	"tinygo.org/x/espradio"
+	"tinygo.org/x/espradio/httputil"
 	link "tinygo.org/x/espradio/netlink"
 )
 
@@ -65,7 +66,7 @@ func main() {
 	http.Handle("/on", logRequest(LED_ON))
 
 	println("HTTP server listening on http://" + apIP + port)
-	err = http.ListenAndServe(apIP+port, nil)
+	err = http.ListenAndServe(apIP+port, httputil.ThrottleHandler(lnk.TCPPoolSize(), nil))
 	if err != nil {
 		failure("http.ListenAndServe: " + err.Error())
 	}

examples/hello/main.go

@@ -7,6 +7,7 @@ import (
 
 	"tinygo.org/x/drivers/netdev"
 	nl "tinygo.org/x/drivers/netlink"
+	"tinygo.org/x/espradio/httputil"
 	link "tinygo.org/x/espradio/netlink"
 )
 
@@ -38,7 +39,7 @@ func main() {
 	h, _ := link.Addr()
 	host := h.String()
 	println("HTTP server listening on http://" + host + port)
-	err = http.ListenAndServe(host+port, nil)
+	err = http.ListenAndServe(host+port, httputil.ThrottleHandler(link.TCPPoolSize(), nil))
 	if err != nil {
 		failure("HTTP server error: " + err.Error())
 	}

examples/webserver/webserver.go

@@ -13,6 +13,7 @@ import (
 
 	"tinygo.org/x/drivers/netdev"
 	nl "tinygo.org/x/drivers/netlink"
+	"tinygo.org/x/espradio/httputil"
 	link "tinygo.org/x/espradio/netlink"
 )
 
@@ -54,7 +55,7 @@ func main() {
 	h, _ := link.Addr()
 	host := h.String()
 	println("HTTP server listening on http://" + host + port)
-	err = http.ListenAndServe(host+port, nil)
+	err = http.ListenAndServe(host+port, httputil.ThrottleHandler(link.TCPPoolSize(), nil))
 	if err != nil {
 		failure("http.ListenAndServe: " + err.Error())
 	}

httputil/httputil.go

@@ -0,0 +1,48 @@
+// Package httputil provides HTTP utilities for memory-constrained espradio devices.
+//
+// The primary utility is [ThrottleHandler], a concurrency limiter designed for
+// ESP32-class microcontrollers where RAM is scarce and unbounded goroutine
+// creation from incoming HTTP requests can exhaust the heap. It uses a
+// non-blocking semaphore so excess requests are shed immediately (503) rather
+// than queued, keeping memory usage deterministic.
+//
+// Typical usage on an ESP32 with ~200 KB usable RAM:
+//
+//	mux := http.NewServeMux()
+//	mux.HandleFunc("/", handler)
+//	err := http.ListenAndServe(":80", httputil.ThrottleHandler(2, mux))
+package httputil
+
+import (
+	"net/http"
+)
+
+// ThrottleHandler returns an http.Handler that limits concurrent request
+// processing to maxConn. When all slots are occupied, excess requests receive
+// 503 Service Unavailable immediately with minimal memory allocation.
+//
+// All responses include a "Connection: close" header to discourage HTTP/1.1
+// keep-alive pipelining on memory-constrained devices.
+//
+// If handler is nil, [http.DefaultServeMux] is used. If maxConn <= 0 it is
+// clamped to 1.
+func ThrottleHandler(maxConn int, handler http.Handler) http.Handler {
+	if handler == nil {
+		handler = http.DefaultServeMux
+	}
+	if maxConn <= 0 {
+		maxConn = 1
+	}
+	sem := make(chan struct{}, maxConn)
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		select {
+		case sem <- struct{}{}:
+			defer func() { <-sem }()
+			w.Header().Set("Connection", "close")
+			handler.ServeHTTP(w, r)
+		default:
+			w.Header().Set("Connection", "close")
+			http.Error(w, "busy", http.StatusServiceUnavailable)
+		}
+	})
+}

httputil/httputil_test.go

@@ -0,0 +1,163 @@
+package httputil
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"sync"
+	"sync/atomic"
+	"testing"
+)
+
+func TestThrottleHandler_LimitsConcurrency(t *testing.T) {
+	const maxConn = 2
+	var active int64
+	var maxSeen int64
+
+	gate := make(chan struct{})
+	inner := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		cur := atomic.AddInt64(&active, 1)
+		defer atomic.AddInt64(&active, -1)
+		// track peak concurrency
+		for {
+			old := atomic.LoadInt64(&maxSeen)
+			if cur <= old || atomic.CompareAndSwapInt64(&maxSeen, old, cur) {
+				break
+			}
+		}
+		<-gate // block until test releases
+		w.WriteHeader(http.StatusOK)
+	})
+
+	h := ThrottleHandler(maxConn, inner)
+
+	// Launch maxConn requests that will block in handler.
+	var wg sync.WaitGroup
+	for i := 0; i < maxConn; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			req := httptest.NewRequest(http.MethodGet, "/", nil)
+			rec := httptest.NewRecorder()
+			h.ServeHTTP(rec, req)
+		}()
+	}
+
+	// Wait until both slots occupied.
+	for atomic.LoadInt64(&active) < int64(maxConn) {
+		// spin-wait; in practice converges instantly
+	}
+
+	// Extra request should get 503 (non-blocking).
+	req := httptest.NewRequest(http.MethodGet, "/", nil)
+	rec := httptest.NewRecorder()
+	h.ServeHTTP(rec, req)
+	if rec.Code != http.StatusServiceUnavailable {
+		t.Errorf("expected 503, got %d", rec.Code)
+	}
+
+	// Release blocked handlers.
+	close(gate)
+	wg.Wait()
+
+	if atomic.LoadInt64(&maxSeen) > int64(maxConn) {
+		t.Errorf("peak concurrency %d exceeded maxConn %d", maxSeen, maxConn)
+	}
+}
+
+func TestThrottleHandler_SetsConnectionClose(t *testing.T) {
+	inner := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusOK)
+	})
+	h := ThrottleHandler(1, inner)
+
+	req := httptest.NewRequest(http.MethodGet, "/", nil)
+	rec := httptest.NewRecorder()
+	h.ServeHTTP(rec, req)
+
+	if got := rec.Header().Get("Connection"); got != "close" {
+		t.Errorf("Connection header = %q, want %q", got, "close")
+	}
+}
+
+func TestThrottleHandler_503SetsConnectionClose(t *testing.T) {
+	gate := make(chan struct{})
+	inner := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		<-gate
+	})
+	h := ThrottleHandler(1, inner)
+
+	// Occupy the single slot.
+	go func() {
+		req := httptest.NewRequest(http.MethodGet, "/", nil)
+		rec := httptest.NewRecorder()
+		h.ServeHTTP(rec, req)
+	}()
+
+	// Wait for slot to be taken (handler blocks on gate).
+	// Send another request synchronously — it should get 503.
+	// We need the first goroutine to actually enter the handler.
+	// Use a small sync mechanism:
+	occupied := make(chan struct{})
+	innerWithSignal := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		close(occupied)
+		<-gate
+	})
+	h2 := ThrottleHandler(1, innerWithSignal)
+
+	go func() {
+		req := httptest.NewRequest(http.MethodGet, "/first", nil)
+		rec := httptest.NewRecorder()
+		h2.ServeHTTP(rec, req)
+	}()
+	<-occupied
+
+	req := httptest.NewRequest(http.MethodGet, "/second", nil)
+	rec := httptest.NewRecorder()
+	h2.ServeHTTP(rec, req)
+
+	if rec.Code != http.StatusServiceUnavailable {
+		t.Fatalf("expected 503, got %d", rec.Code)
+	}
+	if got := rec.Header().Get("Connection"); got != "close" {
+		t.Errorf("503 response Connection header = %q, want %q", got, "close")
+	}
+	close(gate)
+}
+
+func TestThrottleHandler_NilHandlerUsesDefaultMux(t *testing.T) {
+	// Register a handler on DefaultServeMux for this test.
+	pattern := "/httputil-test-nil-handler"
+	http.HandleFunc(pattern, func(w http.ResponseWriter, r *http.Request) {
+		w.Write([]byte("ok"))
+	})
+
+	h := ThrottleHandler(1, nil)
+	req := httptest.NewRequest(http.MethodGet, pattern, nil)
+	rec := httptest.NewRecorder()
+	h.ServeHTTP(rec, req)
+
+	if rec.Code != http.StatusOK {
+		t.Errorf("expected 200, got %d", rec.Code)
+	}
+}
+
+func TestThrottleHandler_MaxConnClampedToOne(t *testing.T) {
+	var called int64
+	inner := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		atomic.AddInt64(&called, 1)
+		w.WriteHeader(http.StatusOK)
+	})
+
+	// maxConn=0 should be treated as 1.
+	h := ThrottleHandler(0, inner)
+	req := httptest.NewRequest(http.MethodGet, "/", nil)
+	rec := httptest.NewRecorder()
+	h.ServeHTTP(rec, req)
+
+	if rec.Code != http.StatusOK {
+		t.Errorf("expected 200, got %d", rec.Code)
+	}
+	if atomic.LoadInt64(&called) != 1 {
+		t.Error("handler was not called")
+	}
+}

netlink/ap.go

@@ -145,17 +145,19 @@ func (n *Esplink) NetConnectAP(params APConnectParams) error {
 		println("Esplink NetConnectAP: stack started, addr:", params.StaticAddr.String())
 	}
 	n.stackloop.Do(func() {
+		poolCfg := xnet.TCPPoolConfig{
+			PoolSize:           2,
+			QueueSize:          4,
+			TxBufSize:          4096,
+			RxBufSize:          1024,
+			EstablishedTimeout: 10 * time.Second,
+			ClosingTimeout:     5 * time.Second,
+			NewBackoff:         func() lneto.BackoffStrategy { return pollBackoff },
+		}
 		gostack := n.netstack.LnetoStack().StackGo(pollBackoff, xnet.StackGoConfig{
-			ListenerPoolConfig: xnet.TCPPoolConfig{
-				PoolSize:           2,
-				QueueSize:          4,
-				TxBufSize:          4096,
-				RxBufSize:          1024,
-				EstablishedTimeout: 10 * time.Second,
-				ClosingTimeout:     5 * time.Second,
-				NewBackoff:         func() lneto.BackoffStrategy { return pollBackoff },
-			},
+			ListenerPoolConfig: poolCfg,
 		})
+		n.tcpPoolSize = int(poolCfg.PoolSize)
 		n.berkeley = *xnet.NewBerkeleyStack(gostack.Socket)
 		go handleStack(espstack)
 	})

netlink/netlink.go

@@ -26,14 +26,22 @@ type Esplink struct {
 	params   *nl.ConnectParams
 	notifyCb func(nl.Event)
 
-	netstack  *espradio.Stack
-	berkeley  xnet.StackBerkeley
-	stackloop sync.Once
+	netstack    *espradio.Stack
+	berkeley    xnet.StackBerkeley
+	stackloop   sync.Once
+	tcpPoolSize int
 
 	// ArenaPoolSize overrides the default arena pool size (bytes). Zero uses target default.
 	ArenaPoolSize int
 }
 
+// TCPPoolSize returns the maximum number of concurrent TCP connections
+// supported by the listener pool. Use this value with espradio.ThrottleHandler
+// to limit in-flight HTTP requests and prevent out-of-memory crashes.
+func (n *Esplink) TCPPoolSize() int {
+	return n.tcpPoolSize
+}
+
 func (n *Esplink) rstack() xnet.StackRetrying {
 	return n.netstack.LnetoStack().StackRetrying(pollBackoff)
 }
@@ -122,18 +130,19 @@ func (n *Esplink) NetConnect(params *nl.ConnectParams) error {
 	}
 	n.stackloop.Do(func() {
 		// Start stack goroutine once.
+		poolCfg := xnet.TCPPoolConfig{
+			PoolSize:           2,
+			QueueSize:          4,
+			TxBufSize:          4096,
+			RxBufSize:          1024,
+			EstablishedTimeout: 2 * time.Second,
+			ClosingTimeout:     2 * time.Second,
+			NewBackoff:         func() lneto.BackoffStrategy { return pollBackoff },
+		}
 		gostack := n.netstack.LnetoStack().StackGo(pollBackoff, xnet.StackGoConfig{
-			ListenerPoolConfig: xnet.TCPPoolConfig{
-				PoolSize:           2,
-				QueueSize:          4,
-				TxBufSize:          4096,
-				RxBufSize:          1024,
-				EstablishedTimeout: 2 * time.Second,
-				ClosingTimeout:     2 * time.Second,
-				NewBackoff:         func() lneto.BackoffStrategy { return pollBackoff },
-			},
+			ListenerPoolConfig: poolCfg,
 		})
-
+		n.tcpPoolSize = int(poolCfg.PoolSize)
 		n.berkeley = *xnet.NewBerkeleyStack(gostack.Socket)
 		go handleStack(espstack)
 	})