limit max number of active streams

backend/app/rest/api/rest.go

@@ -338,6 +338,7 @@ func (s *Rest) controllerGroups() (public, private, admin, rss) {
 		webRoot:          s.WebRoot,
 		streamTimeOut:    s.StreamTimeOut,
 		streamRefresh:    s.StreamRefresh,
+		maxActiveStreams: 500,
 	}
 
 	privGrp := private{

backend/app/rest/api/rest_public.go

@@ -10,6 +10,7 @@ import (
 	"path"
 	"strconv"
 	"strings"
+	"sync/atomic"
 	"time"
 
 	"github.com/go-chi/chi"
@@ -34,6 +35,9 @@ type public struct {
 	webRoot          string
 	streamTimeOut    time.Duration
 	streamRefresh    time.Duration
+	maxActiveStreams int32
+
+	activeStreamsCount int32
 }
 
 type pubStore interface {
@@ -178,6 +182,14 @@ func (s *public) infoStreamCtrl(w http.ResponseWriter, r *http.Request) {
 		return data, upd, nil
 	}
 
+	count := atomic.AddInt32(&s.activeStreamsCount, 1)
+	defer atomic.AddInt32(&s.activeStreamsCount, -1)
+	if count > s.maxActiveStreams {
+		rest.SendErrorJSON(w, r, http.StatusTooManyRequests, errors.New("too many streams"),
+			"can't open new stream", rest.ErrActionRejected)
+		return
+	}
+
 	updCh := s.eventsCh(r.Context(), info)
 
 	for {
@@ -275,6 +287,14 @@ func (s *public) lastCommentsStreamCtrl(w http.ResponseWriter, r *http.Request)
 
 	updCh := s.eventsCh(r.Context(), info)
 
+	count := atomic.AddInt32(&s.activeStreamsCount, 1)
+	defer atomic.AddInt32(&s.activeStreamsCount, -1)
+	if count > s.maxActiveStreams {
+		rest.SendErrorJSON(w, r, http.StatusTooManyRequests, errors.New("too many streams"),
+			"can't open new stream", rest.ErrActionRejected)
+		return
+	}
+
 	for {
 		select {
 		case <-r.Context().Done(): // request closed by remote client

backend/app/rest/api/rest_public_test.go

@@ -8,6 +8,7 @@ import (
 	"net/http"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -528,13 +529,13 @@ func TestRest_Info(t *testing.T) {
 
 func TestRest_InfoStream(t *testing.T) {
 	ts, srv, teardown := startupT(t)
+	defer teardown()
 	srv.pubRest.readOnlyAge = 10000000 // make sure we don't hit read-only
 	srv.pubRest.streamRefresh = 1 * time.Millisecond
 	srv.pubRest.streamTimeOut = 300 * time.Millisecond
 
 	postComment(t, ts.URL)
 
-	defer teardown()
 	wg := sync.WaitGroup{}
 	wg.Add(1)
 	go func() {
@@ -555,6 +556,32 @@ func TestRest_InfoStream(t *testing.T) {
 	assert.True(t, strings.Contains(recs[9], `"count":10`), recs[9])
 }
 
+func TestRest_InfoStreamTooMany(t *testing.T) {
+	ts, srv, teardown := startupT(t)
+	defer teardown()
+	srv.pubRest.readOnlyAge = 10000000 // make sure we don't hit read-only
+	srv.pubRest.streamRefresh = 1 * time.Millisecond
+	srv.pubRest.streamTimeOut = 300 * time.Millisecond
+	srv.pubRest.maxActiveStreams = 10
+
+	postComment(t, ts.URL)
+
+	var errsCount int32
+	wg := sync.WaitGroup{}
+	wg.Add(20)
+	for i := 0; i < 20; i++ {
+		go func() {
+			_, code := get(t, ts.URL+"/api/v1/stream/info?site=radio-t&url=https://radio-t.com/blah1")
+			if code == 429 {
+				atomic.AddInt32(&errsCount, 1)
+			}
+			wg.Done()
+		}()
+	}
+	wg.Wait()
+	assert.Equal(t, int32(10), atomic.LoadInt32(&errsCount), "10 streams rejected")
+}
+
 func TestRest_InfoStreamTimeout(t *testing.T) {
 	ts, srv, teardown := startupT(t)
 	defer teardown()
@@ -712,6 +739,32 @@ func TestRest_LastCommentsStreamCancel(t *testing.T) {
 	assert.True(t, strings.Contains(recs[0], `test 123`), recs[0])
 }
 
+func TestRest_LastCommentsStreamTooMany(t *testing.T) {
+	ts, srv, teardown := startupT(t)
+	defer teardown()
+	srv.pubRest.readOnlyAge = 10000000 // make sure we don't hit read-only
+	srv.pubRest.streamRefresh = 1 * time.Millisecond
+	srv.pubRest.streamTimeOut = 300 * time.Millisecond
+	srv.pubRest.maxActiveStreams = 10
+
+	postComment(t, ts.URL)
+
+	var errsCount int32
+	wg := sync.WaitGroup{}
+	wg.Add(20)
+	for i := 0; i < 20; i++ {
+		go func() {
+			_, code := get(t, ts.URL+"/api/v1/stream/last?site=radio-t")
+			if code == 429 {
+				atomic.AddInt32(&errsCount, 1)
+			}
+			wg.Done()
+		}()
+	}
+	wg.Wait()
+	assert.Equal(t, int32(10), atomic.LoadInt32(&errsCount), "10 streams rejected")
+}
+
 func postComment(t *testing.T, url string) {
 	resp, e := post(t, url+"/api/v1/comment",
 		`{"text": "test 123", "locator":{"url": "https://radio-t.com/blah1", "site": "radio-t"}}`)