Synchronizing requests between two Separate Rate Limited Endpoints

I am utilizing some third party APIs that each have their own rate limits. Endpoint 1 has a rate limit of 10/s and Endpoint2 has a rate limit of 20/s.

I need to process my data through endpoint 1 which will return an array of objects (between 2-3000 objects). I then need to take each object and send some data within to the 2nd endpoint while respecting that 2nd endpoint's rate limit.

I plan on sending the first endpoint's request in go routines via a batch of 10 at a time, ensuring that if all 10 requests finish in <1s, I do not exceed by sending more within that 1 second window.

Ultimately, I want to be able to limit how many concurrent responses are going out at a time per endpoint. Especially if I have to build in retries for failed requests due to server 500+ responses, etc.

For purpose of question I am using httpbin requests to simulate the scenario below:

package main
import (
"bytes"
"encoding/json"
"fmt"
"io"
"net/http"
"sync"
"time"
)
type HttpBinGetRequest struct {
url string
}
type HttpBinGetResponse struct {
Uuid       string `json:"uuid"`
StatusCode int
}
type HttpBinPostRequest struct {
url  string
uuid string // Item to post to API
}
type HttpBinPostResponse struct {
Data       string `json:"data"`
StatusCode int
}
func main() {
// Prepare GET requests for 500 requests
var requests []*HttpBinGetRequest
for i := 0; i < 500; i++ {
uri := "https://httpbin.org/uuid"
request := &HttpBinGetRequest{
url: uri,
}
requests = append(requests, request)
}
// Create semaphore and rate limit for the GET endpoint
getSemaphore := make(chan struct{}, 10)
getRate := make(chan struct{}, 10)
for i := 0; i < cap(getRate); i++ {
getRate <- struct{}{}
}
go func() {
// ticker corresponding to 1/10th of a second
ticker := time.NewTicker(100 * time.Millisecond)
defer ticker.Stop()
for range ticker.C {
_, ok := <-getRate
if !ok {
return
}
}
}()
// Send our GET requests to obtain a random UUID
var wg sync.WaitGroup
for _, request := range requests {
wg.Add(1)
// Go func to make request and receive the response
go func(r *HttpBinGetRequest) {
defer wg.Done()
// Check the rate limiter and block if it is empty
getRate <- struct{}{}
// Add a token to the semaphore
getSemaphore <- struct{}{}
// Remove token when function is complete
defer func() {
<-getSemaphore
}()
resp, _ := get(r)
fmt.Printf("%+v\n", resp)
}(request)
}
wg.Wait()
// I need to add code that obtains the response data from the above for loop
// then sends the UUID it to its own go routines for a POST request, following a similar pattern above
// To not violate the rate limit of the second endpoint which is 20 calls per second
// postSemaphore := make(chan struct{}, 20)
// postRate := make(chan struct{}, 20)
// for i := 0; i < cap(postRate); i++ {
// 	postRate <- struct{}{}
// }
}
func get(hbgr *HttpBinGetRequest) (*HttpBinGetResponse, error) {
httpResp := &HttpBinGetResponse{}
client := &http.Client{}
req, err := http.NewRequest("GET", hbgr.url, nil)
if err != nil {
fmt.Println("error making request")
return httpResp, err
}
req.Header = http.Header{
"accept": {"application/json"},
}
resp, err := client.Do(req)
if err != nil {
fmt.Println(err)
fmt.Println("error getting response")
return httpResp, err
}
// Read Response
body, err := io.ReadAll(resp.Body)
if err != nil {
fmt.Println("error reading response body")
return httpResp, err
}
json.Unmarshal(body, &httpResp)
httpResp.StatusCode = resp.StatusCode
return httpResp, nil
}
// Method to post data to httpbin
func post(hbr *HttpBinPostRequest) (*HttpBinPostResponse, error) {
httpResp := &HttpBinPostResponse{}
client := &http.Client{}
req, err := http.NewRequest("POST", hbr.url, bytes.NewBuffer([]byte(hbr.uuid)))
if err != nil {
fmt.Println("error making request")
return httpResp, err
}
req.Header = http.Header{
"accept": {"application/json"},
}
resp, err := client.Do(req)
if err != nil {
fmt.Println("error getting response")
return httpResp, err
}
if resp.StatusCode == 429 {
fmt.Println(resp.Header.Get("Retry-After"))
}
// Read Response
body, err := io.ReadAll(resp.Body)
if err != nil {
fmt.Println("error reading response body")
return httpResp, err
}
json.Unmarshal(body, &httpResp)
httpResp.StatusCode = resp.StatusCode
fmt.Printf("%+v", httpResp)
return httpResp, nil
}

This is a producer/consumer pattern. You can use a chan to connect them.

Regarding the rate limiter, I would use the package golang.org/x/time/rate.

Since we have decided to use a chan to connect the producer and the consumer, it's natural to send the failed tasks to the same chan so that the consumer can try it again.

I have encapsulated the logic into the type Scheduler[T]. See the demo below. Please note that the demo is written in a hurry to illustrate the idea only. It's not tested thoroughly.

package main

import (
	"context"
	"fmt"
	"io"
	"log"
	"math/rand"
	"net/http"
	"net/http/httptest"
	"sort"
	"sync"
	"time"

	"golang.org/x/time/rate"
)

type task[T any] struct {
	param       T
	failedCount int
}

type Scheduler[T any] struct {
	name     string
	limit    int
	maxTries int
	wg       sync.WaitGroup
	tasks    chan task[T]
	action   func(param T) error
}

// NewScheduler creates a scheduler that runs the action with the specified rate limit.
// It will retry the action if the action returns a non-nil error.
func NewScheduler[T any](name string, limit, maxTries, chanSize int, action func(param T) error) *Scheduler[T] {
	return &Scheduler[T]{
		name:     name,
		limit:    limit,
		maxTries: maxTries,
		tasks:    make(chan task[T], chanSize),
		action:   action,
	}
}

func (s *Scheduler[T]) AddTask(param T) {
	s.wg.Add(1)
	s.tasks <- task[T]{param: param}
}

func (s *Scheduler[T]) retryLater(t task[T]) {
	s.wg.Add(1)
	s.tasks <- t
}

func (s *Scheduler[T]) Run() {
	lim := rate.NewLimiter(rate.Limit(s.limit), 1)
	for t := range s.tasks {
		t := t
		if err := lim.Wait(context.Background()); err != nil {
			log.Fatalf("wait: %s", err)
			return
		}
		go func() {
			defer s.wg.Done()
			err := s.action(t.param)
			if err != nil {
				log.Printf("task %s, param %v failed: %v", s.name, t.param, err)
				t.failedCount++

				if t.failedCount == s.maxTries {
					log.Printf("task %s, param %v failed with %d tries", s.name, t.param, s.maxTries)
					return
				}

				s.retryLater(t)
			}
		}()
	}
}

func (s *Scheduler[T]) Wait() {
	s.wg.Wait()
	close(s.tasks)
}

func main() {
	s := &server{}
	ts := httptest.NewServer(s)
	defer ts.Close()

	schedulerPost := NewScheduler("post", 20, 3, 1, func(param string) error {
		return post(fmt.Sprintf("%s/%s", ts.URL, param))
	})

	go schedulerPost.Run()

	schedulerGet := NewScheduler("get", 10, 3, 1, func(param int) error {
		id, err := get(fmt.Sprintf("%s/%d", ts.URL, param))
		if err != nil {
			return err
		}

		schedulerPost.AddTask(id)
		return nil
	})

	go schedulerGet.Run()

	for i := 0; i < 100; i++ {
		schedulerGet.AddTask(i)
	}

	schedulerGet.Wait()
	schedulerPost.Wait()

	s.printStats()
}

func get(url string) (string, error) {
	resp, err := http.Get(url)
	if err != nil {
		return "", err
	}
	defer resp.Body.Close()

	if resp.StatusCode != 200 {
		return "", fmt.Errorf("unexpected status code: %d", resp.StatusCode)
	}

	body, err := io.ReadAll(resp.Body)
	if err != nil {
		return "", err
	}

	return string(body), nil
}

func post(url string) error {
	resp, err := http.Post(url, "", nil)
	if err != nil {
		return err
	}
	defer resp.Body.Close()

	if resp.StatusCode != 200 {
		return fmt.Errorf("unexpected status code: %d", resp.StatusCode)
	}

	return nil
}

type server struct {
	gMu  sync.Mutex
	gets []int64

	pMu   sync.Mutex
	posts []int64
}

func (s *server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	log.Printf("%s: %s", r.Method, r.URL.Path)

	// collect request stats.
	if r.Method == http.MethodGet {
		s.gMu.Lock()
		s.gets = append(s.gets, time.Now().UnixMilli())
		s.gMu.Unlock()
	} else {
		s.pMu.Lock()
		s.posts = append(s.posts, time.Now().UnixMilli())
		s.pMu.Unlock()
	}

	n := rand.Intn(1000)
	// simulate latency.
	time.Sleep(time.Duration(n) * time.Millisecond)

	// simulate errors.
	if n%10 == 0 {
		w.WriteHeader(http.StatusInternalServerError)
		return
	}

	if r.Method == http.MethodGet {
		fmt.Fprintf(w, "%s", r.URL.Path[1:])
		return
	}
}

func (s *server) printStats() {
	log.Printf("GETS (total: %d):\n", len(s.gets))
	printStats(s.gets)
	log.Printf("POSTS (total: %d):\n", len(s.posts))
	printStats(s.posts)
}

func printStats(ts []int64) {
	sort.Slice(ts, func(i, j int) bool {
		return ts[i] < ts[j]
	})

	count := 0
	to := ts[0] + 1000
	for i := 0; i < len(ts); i++ {
		if ts[i] < to {
			count++
		} else {
			fmt.Printf("  %d: %d\n", to, count)
			i-- // push back the current item
			count = 0
			to += 1000
		}
	}
	if count > 0 {
		fmt.Printf("  %d: %d\n", to, count)
	}
}

The output looks like this:

...
2023/03/25 21:03:30 GETS (total: 112):
1679749398998: 10
1679749399998: 10
1679749400998: 10
1679749401998: 10
1679749402998: 10
1679749403998: 10
1679749404998: 10
1679749405998: 10
1679749406998: 10
1679749407998: 10
1679749408998: 10
1679749409998: 2
2023/03/25 21:03:30 POSTS (total: 111):
1679749399079: 8
1679749400079: 8
1679749401079: 12
1679749402079: 8
1679749403079: 10
1679749404079: 9
1679749405079: 9
1679749406079: 8
1679749407079: 14
1679749408079: 12
1679749409079: 9
1679749410079: 4