beacon/light/request: simple server event rate limit

beacon/light/request/scheduler_test.go

@@ -28,18 +28,12 @@ func TestEventFilter(t *testing.T) {
 	})
 	// let module1 send a request
 	srv.canRequest = 1
-	module1.reqc = testRequest
+	module1.sendReq = testRequest
 	s.Trigger()
-	// first triggered round sends the request, no events yet
+	// in first triggered round module1 sends the request, no events yet
 	s.testWaitCh <- struct{}{}
 	module1.expProcess(t, nil)
 	module2.expProcess(t, nil)
-	// next round triggered by EvRequest; only module1 should receive it
-	s.testWaitCh <- struct{}{}
-	module1.expProcess(t, []Event{
-		Event{Type: EvRequest, Server: srv, Data: RequestResponse{ID: 1, Request: testRequest}},
-	})
-	module2.expProcess(t, nil)
 	// server emits EvTimeout; only module1 should receive it
 	srv.eventCb(Event{Type: EvTimeout, Data: RequestResponse{ID: 1, Request: testRequest}})
 	s.testWaitCh <- struct{}{}
@@ -80,32 +74,32 @@ func (s *testServer) subscribe(eventCb func(Event)) {
 	s.eventCb = eventCb
 }
 
-func (s *testServer) canRequestNow() (bool, float32) {
+func (s *testServer) canRequestNow() bool {
-	return s.canRequest > 0, 0
+	return s.canRequest > 0
 }
 
 func (s *testServer) sendRequest(req Request) ID {
 	s.canRequest--
 	s.lastID++
-	s.eventCb(Event{Type: EvRequest, Data: RequestResponse{ID: s.lastID, Request: req}})
 	return s.lastID
 }
 
-func (s *testServer) Fail(string)  {}
+func (s *testServer) fail(string)  {}
 func (s *testServer) unsubscribe() {}
 
 type testModule struct {
 	name      string
 	processed [][]Event
-	reqc      Request // request candidate
+	sendReq   Request
 }
 
-func (m *testModule) Process(events []Event) {
+func (m *testModule) Process(requester Requester, events []Event) {
 	m.processed = append(m.processed, events)
-}
+	if m.sendReq != nil {
-
+		if cs := requester.CanSendTo(); len(cs) > 0 {
-func (m *testModule) MakeRequest(Server) (Request, float32) {
+			requester.Send(cs[0], m.sendReq)
-	return m.reqc, 0
+		}
+	}
 }
 
 func (m *testModule) expProcess(t *testing.T, expEvents []Event) {

beacon/light/request/server.go

@@ -49,6 +49,8 @@ const (
 	defaultParallelLimit = 3                      // parallelLimit initial value
 	minFailureDelay      = time.Millisecond * 100 // minimum disable time in case of request failure
 	maxFailureDelay      = time.Minute            // maximum disable time in case of request failure
+	maxServerEventBuffer = 5                      // server event allowance buffer limit
+	maxServerEventRate   = time.Second            // server event allowance buffer recharge rate
 )
 
 // requestServer can send requests in a non-blocking way and feed back events
@@ -226,13 +228,11 @@ func (s *serverWithTimeout) unsubscribe() {
 
 // serverWithLimits wraps serverWithTimeout and implements server. It limits the
 // number of parallel in-flight requests and prevents sending new requests when a
-// pending one has already timed out. It also implements a failure delay mechanism
+// pending one has already timed out. Server events are also rate limited.
-// that adds an exponentially growing delay each time a request fails (wrong answer
+// It also implements a failure delay mechanism that adds an exponentially growing
-// or hard timeout). This makes the syncing mechanism less brittle as temporary
+// delay each time a request fails (wrong answer or hard timeout). This makes the
-// failures of the server might happen sometimes, but still avoids hammering a
+// syncing mechanism less brittle as temporary failures of the server might happen
-// non-functional server with requests.
+// sometimes, but still avoids hammering a non-functional server with requests.
-//
-// TODO protect against excessive server events
 type serverWithLimits struct {
 	serverWithTimeout
 	lock                       sync.Mutex
@@ -245,12 +245,15 @@ type serverWithLimits struct {
 	delayCounter               int
 	failureDelayEnd            mclock.AbsTime
 	failureDelay               float64
+	serverEventBuffer          int
+	eventBufferUpdated         mclock.AbsTime
 }
 
 // init initializes serverWithLimits
 func (s *serverWithLimits) init() {
 	s.softTimeouts = make(map[ID]struct{})
 	s.parallelLimit = defaultParallelLimit
+	s.serverEventBuffer = maxServerEventBuffer
 }
 
 // subscribe subscribes to events which include parent (serverWithTimeout) events
@@ -267,6 +270,7 @@ func (s *serverWithLimits) subscribe(eventCallback func(event Event)) {
 func (s *serverWithLimits) eventCallback(event Event) {
 	s.lock.Lock()
 	var sendCanRequestAgain bool
+	passEvent := true
 	switch event.Type {
 	case EvTimeout:
 		id := event.Data.(RequestResponse).ID
@@ -295,10 +299,31 @@ func (s *serverWithLimits) eventCallback(event Event) {
 		if event.Type == EvFail {
 			s.failLocked("failed request")
 		}
+	default:
+		// server event; check rate limit
+		if s.serverEventBuffer < maxServerEventBuffer {
+			now := s.clock.Now()
+			sinceUpdate := time.Duration(now - s.eventBufferUpdated)
+			if sinceUpdate >= maxServerEventRate*time.Duration(maxServerEventBuffer-s.serverEventBuffer) {
+				s.serverEventBuffer = maxServerEventBuffer
+				s.eventBufferUpdated = now
+			} else {
+				addBuffer := int(sinceUpdate / maxServerEventRate)
+				s.serverEventBuffer += addBuffer
+				s.eventBufferUpdated += mclock.AbsTime(maxServerEventRate * time.Duration(addBuffer))
+			}
+		}
+		if s.serverEventBuffer > 0 {
+			s.serverEventBuffer--
+		} else {
+			passEvent = false
+		}
 	}
 	childEventCb := s.childEventCb
 	s.lock.Unlock()
-	childEventCb(event)
+	if passEvent {
+		childEventCb(event)
+	}
 	if sendCanRequestAgain {
 		childEventCb(Event{Type: EvCanRequestAgain})
 	}

beacon/light/request/server_test.go

@@ -36,7 +36,6 @@ func TestServerEvents(t *testing.T) {
 	expEvent(testEventType)
 	// send request, soft timeout, then valid response
 	srv.sendRequest(testRequest)
-	expEvent(EvRequest)
 	clock.WaitForTimers(1)
 	clock.Run(softRequestTimeout)
 	expEvent(EvTimeout)
@@ -44,7 +43,6 @@ func TestServerEvents(t *testing.T) {
 	expEvent(EvResponse)
 	// send request, hard timeout (response after hard timeout should be ignored)
 	srv.sendRequest(testRequest)
-	expEvent(EvRequest)
 	clock.WaitForTimers(1)
 	clock.Run(softRequestTimeout)
 	expEvent(EvTimeout)
@@ -63,7 +61,7 @@ func TestServerParallel(t *testing.T) {
 	expSend := func(expSent int) {
 		var sent int
 		for sent <= expSent {
-			if ok, _ := srv.canRequestNow(); !ok {
+			if !srv.canRequestNow() {
 				break
 			}
 			sent++
@@ -94,7 +92,7 @@ func TestServerFail(t *testing.T) {
 	srv := NewServer(rs, clock)
 	srv.subscribe(func(event Event) {})
 	expCanRequest := func(expCanRequest bool) {
-		if canRequest, _ := srv.canRequestNow(); canRequest != expCanRequest {
+		if canRequest := srv.canRequestNow(); canRequest != expCanRequest {
 			t.Errorf("Wrong result for canRequestNow (expected %v, got %v)", expCanRequest, canRequest)
 		}
 	}
@@ -108,7 +106,7 @@ func TestServerFail(t *testing.T) {
 	rs.eventCb(Event{Type: EvResponse, Data: RequestResponse{ID: 1, Request: testRequest, Response: testResponse}})
 	expCanRequest(true)
 	// explicit server.Fail
-	srv.Fail("")
+	srv.fail("")
 	clock.WaitForTimers(1)
 	expCanRequest(false) // cannot request for a while after a failure
 	clock.Run(minFailureDelay)
@@ -125,6 +123,32 @@ func TestServerFail(t *testing.T) {
 	srv.unsubscribe()
 }
 
+func TestServerEventRateLimit(t *testing.T) {
+	rs := &testRequestServer{}
+	clock := &mclock.Simulated{}
+	srv := NewServer(rs, clock)
+	var eventCount int
+	srv.subscribe(func(event Event) {
+		if !event.IsRequestEvent() {
+			eventCount++
+		}
+	})
+	expEvents := func(send, expAllowed int) {
+		eventCount = 0
+		for sent := 0; sent < send; sent++ {
+			rs.eventCb(Event{Type: testEventType})
+		}
+		if eventCount != expAllowed {
+			t.Errorf("Wrong number of server events passing rate limitation (sent %d, expected %d, got %d)", send, expAllowed, eventCount)
+		}
+	}
+	expEvents(maxServerEventBuffer+5, maxServerEventBuffer)
+	clock.Run(maxServerEventRate)
+	expEvents(5, 1)
+	clock.Run(maxServerEventRate * maxServerEventBuffer * 2)
+	expEvents(maxServerEventBuffer+5, maxServerEventBuffer)
+}
+
 type testRequestServer struct {
 	eventCb func(Event)
 }