Evitando requisições duplicadas com singleflight 🇧🇷

Você tem algum endpoint que precisa processar muita coisa, consome dados de terceiros, lento, etc.... E para ajudar esse endpoint recebe muitas requisições simultâneas( algo que carrega na tua página inicial para todos users e tem o mesmo conteúdo)

Cada vez que aquele endpoint é chamado seus olhos se enchem de lágrimas, pois então isso vai mudar :) e vou te contar como.
Vamos usar o pacote singleflight. Nas palavras do pacote:

"Package singleflight provides a duplicate function call suppression mechanism."
"O pacote singleflight fornece um mecanismo de supressão de chamada de função duplicado."

A idéia do pacote é, você cria uma chave para identificar a requisição e quando houver outras requisições com a mesma chave ela vai aguardar a resposta que está em andamento de outra request. Quando a request retornar com o resultado ela compartilhará com as outras requests que estavam esperando pelo resultado, assim evitando múltiplas chamadas/processos pesados.

Chega de papo e vamos ver código, afinal é disso que gostamos :). Criei uma api para que possamos ver o pacote em ação, você pode ver o código no repositório

Criei um serviço http que consome dados vindos de uma api externa.

package main
import (
    "fmt"
    "io/ioutil"
    "log"
    "net/http"
    "os"
    "time"
)
func main() {
    http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
        fmt.Println("calling the endpoint")
        response, err := http.Get("https://jsonplaceholder.typicode.com/photos")
        if err != nil {
            fmt.Print(err.Error())
            os.Exit(1)
        }
        responseData, err := ioutil.ReadAll(response.Body)
        if err != nil {
            log.Fatal(err)
        }
        time.Sleep(2 * time.Second)
        w.Write(responseData)
    })
    http.ListenAndServe(":3000", nil)
}

Quando você acessar http://127.0.0.1:3000/ ele vai chamar pela api jsonplaceholder, para tornar mais interessante eu adicionei um sleep de 2 segundos para simular que o processo é mais lento.

Agora vamos usar o vegeta a idéia aqui é executar várias requests para ver o singleflight brilhar. Eu defino para executar 10 requests por segundo e a duração de 1 segundo.

echo "GET http://localhost:3000/" | vegeta attack -duration=1s -rate=10 | tee results.bin | vegeta report

Aqui você pode ver o resultado do Vegeta e o output do nosso serviço:

Como podemos ver todas requests chamaram a api externa.

Agora vamos ver o singleflight brilhar, usaremos as mesmas configurações do vegeta.

Neste código eu adicionei um novo endpoint /singleflight, na chamada da função requestGroup.Do() eu defini a chave como singleflight, agora a request vai verificar se há um processo em andamento, caso sim ele aguarda o resultado.
Adicionei um print no terminal para indicar quando a request aguarda pelo resultado e usa a resposta compartilhada.

// Como ele não faz parte da standard library você precisa adicionar ele no seu gopath,go mod,vendor,etc...
import "golang.org/x/sync/singleflight"
var requestGroup singleflight.Group
//para este endpoint funcionar você precisa importar o pacote singleflight e criar essa variável(eu sei global e tal, mas para este post é suficiente).
http.HandleFunc("/singleflight", func(w http.ResponseWriter, r *http.Request) {
        res, err, shared := requestGroup.Do("singleflight", func() (interface{}, error) {
            fmt.Println("calling the endpoint")
            response, err := http.Get("https://jsonplaceholder.typicode.com/photos")

            if err != nil {
                http.Error(w, err.Error(), http.StatusInternalServerError)
                return nil, err
            }
            responseData, err := ioutil.ReadAll(response.Body)
            if err != nil {
                log.Fatal(err)
            }
            time.Sleep(2 * time.Second)
            return string(responseData), err
        })
        if err != nil {
            http.Error(w, err.Error(), http.StatusInternalServerError)
            return
        }
        result := res.(string)
        fmt.Println("shared = ", shared)
        fmt.Fprintf(w, "%q", result)
    })

Vegeta novamente

echo "GET http://localhost:3000/" | vegeta attack -duration=1s -rate=10 | tee results.bin | vegeta report

Recomendo você executar o código+vegeta e ver isso executando na sua máquina.
No primeiro endpoint, você verá que os requests são executados e o log mostra as chamadas.
No segundo endpoint, você verá uma request, e de repente 10 true indicando que todas requests usaram a resposta compartilhada.

Isso é um recurso incrível, pense em endpoints que tem um processo pesado/lento ou por serviços externos que você paga por requisições, neste último caso além de ajudar o serviço evitando processamento também poderá poupar dinheiro com requests duplicadas.
Outro ponto é que estou usando em um serviço de http, mas poderia ser qualquer coisa, por ex. poderia ser uma consulta na base de dados , enfim cenários não faltam :).

Bom é isso o que gostaria de mostrar, espero que te ajude como me ajudou saber deste package lindão e deixo meu agradecimento ao Henrique por te me mostrado o pacote. Vale dizer que o pacote conta com uma opção para "esquecer" a chave e tbm uma opção que o resultado é retornado via channel. Você pode por ex. definir um timeout, depois de n tempo você pode por ex. cancelar a chave ou pelo channel.