diff --git a/backend/app/rest/api/rest.go b/backend/app/rest/api/rest.go index 0d029ea2..a56d180f 100644 --- a/backend/app/rest/api/rest.go +++ b/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{ diff --git a/backend/app/rest/api/rest_public.go b/backend/app/rest/api/rest_public.go index 7be58849..388886d4 100644 --- a/backend/app/rest/api/rest_public.go +++ b/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 diff --git a/backend/app/rest/api/rest_public_test.go b/backend/app/rest/api/rest_public_test.go index 51504333..2cd4ed22 100644 --- a/backend/app/rest/api/rest_public_test.go +++ b/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"}}`)