git.stevedylan.dev

chore: added inbound/outbound concurrency limiters cf00c988

Steve Simkins · 2026-07-07 02:27 4 file(s) · +279 −13

app.go +21 −0

	return mux
}

// limitInFlight caps concurrent in-flight requests. Under a traffic spike,
// excess requests get an immediate 503 rather than piling up goroutines and
// outbound connections that would exhaust file descriptors. Static asset
// requests are cheap (served from memory) and bypass the limiter.
func limitInFlight(next http.Handler, max int) http.Handler {
	sem := make(chan struct{}, max)
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if strings.HasPrefix(r.URL.Path, "/static/") {
			next.ServeHTTP(w, r)
			return
		}
		select {
		case sem <- struct{}{}:
			defer func() { <-sem }()
			next.ServeHTTP(w, r)
		default:
			http.Error(w, "server busy", http.StatusServiceUnavailable)
		}
	})
}

// embeddedHandler serves files from an embed.FS under the given URL prefix.
func embeddedHandler(fs embed.FS, prefix string) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {

feeds.go +47 −12


const appUserAgent = "feeds/0.1 (+https://github.com/stevedylandev/feeds)"

// fetchSem bounds total concurrent outbound HTTP requests across every handler.
// A single page can reference maxFeedURLs feeds (each a separate fetch), and
// that fan-out times concurrent viral traffic would otherwise open unbounded
// sockets and goroutines. Sized well above renderSem because fetches are
// IO-bound; excess fetches queue until a slot frees or their context expires.
var fetchSem = make(chan struct{}, 64)

// acquireFetch takes an outbound-fetch slot, honoring context cancellation so a
// request that gives up doesn't keep waiting for a slot it no longer needs.
func acquireFetch(ctx context.Context) error {
	select {
	case fetchSem <- struct{}{}:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}

func releaseFetch() { <-fetchSem }

func buildHTTPClient() *http.Client {
	return &http.Client{Timeout: 15 * time.Second}
	return &http.Client{Timeout: 10 * time.Second}
}

func newRequest(ctx context.Context, method, rawURL string) (*http.Request, error) {

	if lastModified != "" {
		req.Header.Set("If-Modified-Since", lastModified)
	}
	if err := acquireFetch(ctx); err != nil {
		return nil, err
	}
	defer releaseFetch()
	resp, err := client.Do(req)
	if err != nil {
		return nil, fmt.Errorf("fetch failed: %w", err)

	if err != nil {
		return ""
	}
	if err := acquireFetch(ctx); err != nil {
		return ""
	}
	defer releaseFetch()
	resp, err := client.Do(req)
	if err == nil {
		defer resp.Body.Close()

		}
	}
	for _, page := range scanPages {
		req, err := newRequest(ctx, http.MethodGet, page)
		if err != nil {
			continue
		}
		resp, err := client.Do(req)
		if err != nil {
			continue
		}
		body, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
		_ = resp.Body.Close()
		if resp.StatusCode < 200 || resp.StatusCode >= 300 {
		body, status := func() ([]byte, int) {
			req, err := newRequest(ctx, http.MethodGet, page)
			if err != nil {
				return nil, 0
			}
			if err := acquireFetch(ctx); err != nil {
				return nil, 0
			}
			defer releaseFetch()
			resp, err := client.Do(req)
			if err != nil {
				return nil, 0
			}
			defer resp.Body.Close()
			b, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
			return b, resp.StatusCode
		}()
		if status < 200 || status >= 300 {
			continue
		}
		for _, href := range findAlternateFeedLinks(string(body)) {

feeds_test.go (added) +201 −0

1	+	package main
2	+
3	+	import (
4	+	"context"
5	+	"fmt"
6	+	"net/http"
7	+	"net/http/httptest"
8	+	"net/url"
9	+	"strings"
10	+	"sync/atomic"
11	+	"testing"
12	+	"time"
13	+	)
14	+
15	+	// --- inbound limiter -------------------------------------------------------
16	+
17	+	func TestLimitInFlightRejectsOverflow(t *testing.T) {
18	+	entered := make(chan struct{})
19	+	release := make(chan struct{})
20	+	defer close(release) // unblock the parked handler when the test ends
21	+	h := limitInFlight(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
22	+	entered <- struct{}{}
23	+	<-release
24	+	}), 1)
25	+
26	+	// Occupy the single slot with a request parked inside the handler.
27	+	go h.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest(http.MethodGet, "/", nil))
28	+	<-entered
29	+
30	+	// A second request must be shed immediately rather than queue.
31	+	rec := httptest.NewRecorder()
32	+	h.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/", nil))
33	+	if rec.Code != http.StatusServiceUnavailable {
34	+	t.Fatalf("overflow: want 503, got %d", rec.Code)
35	+	}
36	+	}
37	+
38	+	func TestLimitInFlightStaticBypass(t *testing.T) {
39	+	// Capacity 0: no non-static request can ever get a slot.
40	+	h := limitInFlight(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
41	+	w.WriteHeader(http.StatusOK)
42	+	}), 0)
43	+
44	+	rec := httptest.NewRecorder()
45	+	h.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/static/app.css", nil))
46	+	if rec.Code != http.StatusOK {
47	+	t.Fatalf("static bypass: want 200, got %d", rec.Code)
48	+	}
49	+
50	+	rec = httptest.NewRecorder()
51	+	h.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/", nil))
52	+	if rec.Code != http.StatusServiceUnavailable {
53	+	t.Fatalf("non-static: want 503, got %d", rec.Code)
54	+	}
55	+	}
56	+
57	+	// --- outbound fetch limiter ------------------------------------------------
58	+
59	+	func TestAcquireFetchRespectsContext(t *testing.T) {
60	+	saved := fetchSem
61	+	fetchSem = make(chan struct{}, 1)
62	+	defer func() { fetchSem = saved }()
63	+
64	+	if err := acquireFetch(context.Background()); err != nil {
65	+	t.Fatalf("first acquire: %v", err)
66	+	}
67	+	// Slot is full; a cancelled context must bail instead of blocking forever.
68	+	ctx, cancel := context.WithCancel(context.Background())
69	+	cancel()
70	+	if err := acquireFetch(ctx); err == nil {
71	+	t.Fatal("want error acquiring under cancelled ctx, got nil")
72	+	}
73	+	releaseFetch()
74	+	if err := acquireFetch(context.Background()); err != nil {
75	+	t.Fatalf("acquire after release: %v", err)
76	+	}
77	+	releaseFetch()
78	+	}
79	+
80	+	// --- request parsing -------------------------------------------------------
81	+
82	+	func TestFeedURLsFromRequestCap(t *testing.T) {
83	+	parts := make([]string, 0, 30)
84	+	for i := 0; i < 30; i++ {
85	+	parts = append(parts, fmt.Sprintf("https://e%d.com/feed", i))
86	+	}
87	+	q := url.QueryEscape(strings.Join(parts, ","))
88	+	req := httptest.NewRequest(http.MethodGet, "/?url="+q, nil)
89	+	if got := len(feedURLsFromRequest(req)); got != maxFeedURLs {
90	+	t.Fatalf("cap: want %d urls, got %d", maxFeedURLs, got)
91	+	}
92	+	}
93	+
94	+	func TestFeedURLsFromRequestFallsBackToUrls(t *testing.T) {
95	+	req := httptest.NewRequest(http.MethodGet, "/?urls=https://a.com/feed,https://b.com/feed", nil)
96	+	if got := len(feedURLsFromRequest(req)); got != 2 {
97	+	t.Fatalf("urls param: want 2, got %d", got)
98	+	}
99	+	}
100	+
101	+	func TestOGCacheKeyOrderIndependent(t *testing.T) {
102	+	a := ogCacheKey([]string{"https://b.com", "https://a.com"})
103	+	b := ogCacheKey([]string{"https://a.com", "https://b.com"})
104	+	if a != b {
105	+	t.Fatalf("cache key not order-independent: %q vs %q", a, b)
106	+	}
107	+	}
108	+
109	+	// --- feed cache ------------------------------------------------------------
110	+
111	+	func TestFeedCacheEvictsLRU(t *testing.T) {
112	+	c := newFeedCache(time.Minute, 1)
113	+	c.store("a", &FetchResult{Title: "a"})
114	+	c.store("b", &FetchResult{Title: "b"})
115	+	if _, ok := c.lookup("a"); ok {
116	+	t.Fatal("a should have been evicted")
117	+	}
118	+	if _, ok := c.lookup("b"); !ok {
119	+	t.Fatal("b should still be cached")
120	+	}
121	+	}
122	+
123	+	func TestFeedCacheExpires(t *testing.T) {
124	+	c := newFeedCache(time.Millisecond, 8)
125	+	c.store("a", &FetchResult{Title: "a"})
126	+	time.Sleep(5 * time.Millisecond)
127	+	if _, ok := c.lookup("a"); ok {
128	+	t.Fatal("entry should have expired")
129	+	}
130	+	}
131	+
132	+	// --- image cache -----------------------------------------------------------
133	+
134	+	func TestImageCacheRendersOnce(t *testing.T) {
135	+	c := newImageCache(time.Minute, 4)
136	+	var calls int32
137	+	render := func() ([]byte, error) {
138	+	atomic.AddInt32(&calls, 1)
139	+	return []byte("png"), nil
140	+	}
141	+	for i := 0; i < 3; i++ {
142	+	if _, err := c.getOrRender("k", render); err != nil {
143	+	t.Fatalf("getOrRender: %v", err)
144	+	}
145	+	}
146	+	if calls != 1 {
147	+	t.Fatalf("render calls: want 1, got %d", calls)
148	+	}
149	+	}
150	+
151	+	// --- fetchFeed -------------------------------------------------------------
152	+
153	+	const sampleRSS = `<?xml version="1.0"?>
154	+	<rss version="2.0"><channel>
155	+	<title>Example</title><link>https://example.com</link>
156	+	<item><title>Hello</title><link>https://example.com/hello</link></item>
157	+	</channel></rss>`
158	+
159	+	func TestFetchFeedParses(t *testing.T) {
160	+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
161	+	w.Header().Set("ETag", `"v1"`)
162	+	fmt.Fprint(w, sampleRSS)
163	+	}))
164	+	defer srv.Close()
165	+
166	+	res, err := fetchFeed(context.Background(), srv.URL, "", "")
167	+	if err != nil {
168	+	t.Fatalf("fetchFeed: %v", err)
169	+	}
170	+	if res.Title != "Example" {
171	+	t.Fatalf("title: want Example, got %q", res.Title)
172	+	}
173	+	if len(res.Entries) != 1 \|\| res.Entries[0].Link != "https://example.com/hello" {
174	+	t.Fatalf("entries: %+v", res.Entries)
175	+	}
176	+	if res.ETag != `"v1"` {
177	+	t.Fatalf("etag: want v1, got %q", res.ETag)
178	+	}
179	+	}
180	+
181	+	func TestFetchFeedNotModified(t *testing.T) {
182	+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
183	+	if r.Header.Get("If-None-Match") == `"v1"` {
184	+	w.WriteHeader(http.StatusNotModified)
185	+	return
186	+	}
187	+	fmt.Fprint(w, sampleRSS)
188	+	}))
189	+	defer srv.Close()
190	+
191	+	res, err := fetchFeed(context.Background(), srv.URL, `"v1"`, "")
192	+	if err != nil {
193	+	t.Fatalf("fetchFeed 304: %v", err)
194	+	}
195	+	if res.Status != http.StatusNotModified {
196	+	t.Fatalf("status: want 304, got %d", res.Status)
197	+	}
198	+	if res.ETag != `"v1"` {
199	+	t.Fatalf("etag carried forward: want v1, got %q", res.ETag)
200	+	}
201	+	}

main.go +10 −1


	addr := getenv("HOST", "0.0.0.0") + ":" + getenv("PORT", "3000")
	logger.Info("feeds server running", "addr", addr)
	if err := http.ListenAndServe(addr, app.routes()); err != nil {
	srv := &http.Server{
		Addr:              addr,
		Handler:           limitInFlight(app.routes(), 300),
		ReadHeaderTimeout: 5 * time.Second,
		ReadTimeout:       10 * time.Second,
		WriteTimeout:      30 * time.Second,
		IdleTimeout:       60 * time.Second,
		MaxHeaderBytes:    1 << 16,
	}
	if err := srv.ListenAndServe(); err != nil {
		log.Fatal(err)
	}
}

57	57		return mux
58	58		}
59	59
	60	+	// limitInFlight caps concurrent in-flight requests. Under a traffic spike,
	61	+	// excess requests get an immediate 503 rather than piling up goroutines and
	62	+	// outbound connections that would exhaust file descriptors. Static asset
	63	+	// requests are cheap (served from memory) and bypass the limiter.
	64	+	func limitInFlight(next http.Handler, max int) http.Handler {
	65	+	sem := make(chan struct{}, max)
	66	+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
	67	+	if strings.HasPrefix(r.URL.Path, "/static/") {
	68	+	next.ServeHTTP(w, r)
	69	+	return
	70	+	}
	71	+	select {
	72	+	case sem <- struct{}{}:
	73	+	defer func() { <-sem }()
	74	+	next.ServeHTTP(w, r)
	75	+	default:
	76	+	http.Error(w, "server busy", http.StatusServiceUnavailable)
	77	+	}
	78	+	})
	79	+	}
	80	+
60	81		// embeddedHandler serves files from an embed.FS under the given URL prefix.
61	82		func embeddedHandler(fs embed.FS, prefix string) http.HandlerFunc {
62	83		return func(w http.ResponseWriter, r *http.Request) {

44	44
45	45		const appUserAgent = "feeds/0.1 (+https://github.com/stevedylandev/feeds)"
46	46
	47	+	// fetchSem bounds total concurrent outbound HTTP requests across every handler.
	48	+	// A single page can reference maxFeedURLs feeds (each a separate fetch), and
	49	+	// that fan-out times concurrent viral traffic would otherwise open unbounded
	50	+	// sockets and goroutines. Sized well above renderSem because fetches are
	51	+	// IO-bound; excess fetches queue until a slot frees or their context expires.
	52	+	var fetchSem = make(chan struct{}, 64)
	53	+
	54	+	// acquireFetch takes an outbound-fetch slot, honoring context cancellation so a
	55	+	// request that gives up doesn't keep waiting for a slot it no longer needs.
	56	+	func acquireFetch(ctx context.Context) error {
	57	+	select {
	58	+	case fetchSem <- struct{}{}:
	59	+	return nil
	60	+	case <-ctx.Done():
	61	+	return ctx.Err()
	62	+	}
	63	+	}
	64	+
	65	+	func releaseFetch() { <-fetchSem }
	66	+
47	67		func buildHTTPClient() *http.Client {
48		-	return &http.Client{Timeout: 15 * time.Second}
	68	+	return &http.Client{Timeout: 10 * time.Second}
49	69		}
50	70
51	71		func newRequest(ctx context.Context, method, rawURL string) (*http.Request, error) {

69	89		if lastModified != "" {
70	90		req.Header.Set("If-Modified-Since", lastModified)
71	91		}
	92	+	if err := acquireFetch(ctx); err != nil {
	93	+	return nil, err
	94	+	}
	95	+	defer releaseFetch()
72	96		resp, err := client.Do(req)
73	97		if err != nil {
74	98		return nil, fmt.Errorf("fetch failed: %w", err)

236	260		if err != nil {
237	261		return ""
238	262		}
	263	+	if err := acquireFetch(ctx); err != nil {
	264	+	return ""
	265	+	}
	266	+	defer releaseFetch()
239	267		resp, err := client.Do(req)
240	268		if err == nil {
241	269		defer resp.Body.Close()

304	332		}
305	333		}
306	334		for _, page := range scanPages {
307		-	req, err := newRequest(ctx, http.MethodGet, page)
308		-	if err != nil {
309		-	continue
310		-	}
311		-	resp, err := client.Do(req)
312		-	if err != nil {
313		-	continue
314		-	}
315		-	body, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
316		-	_ = resp.Body.Close()
317		-	if resp.StatusCode < 200 \|\| resp.StatusCode >= 300 {
	335	+	body, status := func() ([]byte, int) {
	336	+	req, err := newRequest(ctx, http.MethodGet, page)
	337	+	if err != nil {
	338	+	return nil, 0
	339	+	}
	340	+	if err := acquireFetch(ctx); err != nil {
	341	+	return nil, 0
	342	+	}
	343	+	defer releaseFetch()
	344	+	resp, err := client.Do(req)
	345	+	if err != nil {
	346	+	return nil, 0
	347	+	}
	348	+	defer resp.Body.Close()
	349	+	b, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
	350	+	return b, resp.StatusCode
	351	+	}()
	352	+	if status < 200 \|\| status >= 300 {
318	353		continue
319	354		}
320	355		for _, href := range findAlternateFeedLinks(string(body)) {

30	30
31	31		addr := getenv("HOST", "0.0.0.0") + ":" + getenv("PORT", "3000")
32	32		logger.Info("feeds server running", "addr", addr)
33		-	if err := http.ListenAndServe(addr, app.routes()); err != nil {
	33	+	srv := &http.Server{
	34	+	Addr: addr,
	35	+	Handler: limitInFlight(app.routes(), 300),
	36	+	ReadHeaderTimeout: 5 * time.Second,
	37	+	ReadTimeout: 10 * time.Second,
	38	+	WriteTimeout: 30 * time.Second,
	39	+	IdleTimeout: 60 * time.Second,
	40	+	MaxHeaderBytes: 1 << 16,
	41	+	}
	42	+	if err := srv.ListenAndServe(); err != nil {
34	43		log.Fatal(err)
35	44		}
36	45		}