Repository: 99designs/gqlgen
Pull request #3466 (2025-01-02 00:52:42)

Add `KeepAlivePingInterval` to SSE transport
## Proposal
Add optional `KeepAlivePingInterval` parameter to the SSE transport.

```go
srv.AddTransport(transport.SSE{
	KeepAlivePingInterval: 10 * time.Second,
})
```

SSE (Server-Sent Events) connections are designed to keep a connection open between the client and server to send real-time updates from the server to the client. While SSE connections are intended to be long-lived and persistent, they can be affected by idle timeout settings if there are long periods of inactivity.

## How Idle Timeout Affects SSE:
### Idle Timeout :
Load balancers, proxies, or firewalls often have idle timeout settings that specify the maximum duration a connection can remain open without data being sent across it. If the idle timeout is exceeded without any data being transmitted, the connection may be closed.

### SSE Behavior :
In a properly functioning SSE setup, the server is typically expected to send periodic heartbeat messages or comments (using a colon : in SSE) to ensure that the connection stays alive and doesn't hit the idle timeout. This ensures that some traffic occurs to reset the idle timeout counter.

I have:
 - [x] Added tests covering the bug / feature (see [testing](https://github.com/99designs/gqlgen/blob/master/TESTING.md))
 - [x] Updated any relevant documentation (see [docs](https://github.com/99designs/gqlgen/tree/master/docs/content))


StevenACoffman: @UnAfraid @oxisto @o8x @jmic @giulio-opal Do you have information or thoughts on this?

I'm unclear if Keep-Alive Ping should differ depending on HTTP/2 SSE vs HTTP/1.1 SSE? I don't know if Go's http2 automatically persists streaming connections within the data frames, or if we still need to do some sort of Ping.

I see a note that the Keep-Alive header is deprecated for HTTP2: 
https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Keep-Alive

oxisto: > @UnAfraid @oxisto @o8x @jmic @giulio-opal Do you have information or thoughts on this?
> 
> I'm unclear if Keep-Alive Ping should differ depending on HTTP/2 SSE vs HTTP/1.1 SSE? I don't know if Go's http2 automatically persists streaming connections within the data frames, or if we still need to do some sort of Ping.
> 
> I see a note that the Keep-Alive header is deprecated for HTTP2: https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Keep-Alive
> 
> AFAICT, this Keep-Alive Ping is a good idea for HTTP/1.1, but prohibited for HTTP/2, so this should at least have a comment to that effect (and pass the existing tests).
> 
> I also [read here](https://github.com/r3labs/sse/issues/101#issuecomment-962464451):
> 
> > In HTTP/2, the concept of persistent connections is built into the protocol itself.
> > HTTP/2 allows multiple streams of data to be multiplexed over a single connection, inherently supporting what "Keep-Alive" aims to achieve in HTTP/1.1.
> > Therefore, connection-specific headers like "Connection" and "Keep-Alive" are not only unnecessary in HTTP/2 but are also prohibited.
> > Browsers like Chrome and Firefox ignore these headers in HTTP/2 responses, and Safari may not load responses that contain them.
> > Given this, you do not need to turn off "Keep-Alive" in your Apache configuration when using HTTP/2, as it will be ignored by the protocol.

If I read the issues surrounding http2 correctly, there is now a `PingTimeout` in `http2.Server` that can be configured to send such pings: https://pkg.go.dev/golang.org/x/net/http2#Server. See also https://github.com/golang/go/issues/67812. This should be in x/net v0.33.0 and this will eventually end up in std. Not sure if this is already targeted for Go 1.24.
jlaprise: Thanks for looking into this request. Since submitting this PR we found a solution, and a bit more info that hopefully will help anyone else having a similar issue.

SSE over an end-to-end HTTP/2 connection doesn't require a ping to keep the connection open.

The issue we had is related to SSE over HTTP/1. We didn't have any issues in our dev/test environments but once deployed to prod (AWS) we saw SSE connections get disconnected after 1 minute due to the default [ALB](https://aws.amazon.com/elasticloadbalancing/application-load-balancer/) idle connection timeout. We had ALB configured for HTTP/2 but the routed target group was configured for HTTP/1. ALB will accept HTTP 1 or 2 and translate the request to whichever protocol the load balancer's target group is configured to accept. 

We tried changing the target group protocol to HTTP/2, but then clients that sent HTTP/1 requests received response code [464 Incompatible protocol](https://http.dev/464).

The solution we ended at was to create two separate target groups for HTTP/1 and HTTP/2 pointing at the same backend, then configure a rule on the load balancer to forward to the target group matching the request's HTTP version.

Sending a ping like this PR suggested does keep the connection open when there's a network hop only compatible with HTTP/1. However, I'm unsure if there are caveats to doing that.
StevenACoffman: Ok, this sounds like it's beneficial for HTTP/1.1 environments, and people can just opt-out if they don't need it. Can you fix the test to pass (and it would be nice if you rebased but not required)?

I appreciate you contributing it, as well as circling back to provide more context.

Patch:
diff --git a/docs/content/recipes/subscriptions.md b/docs/content/recipes/subscriptions.md
index 028ba744..9d2f42fc 100644
--- a/docs/content/recipes/subscriptions.md
+++ b/docs/content/recipes/subscriptions.md
@@ -252,6 +252,21 @@ srv.AddTransport(transport.GET{})
 srv.AddTransport(transport.POST{})
 ```
 
+Optionally add `KeepAlivePingInterval` to send a periodic heartbeat over the SSE transport.
+```go
+srv.AddTransport(transport.SSE{
+	// Load balancers, proxies, or firewalls often have idle timeout
+	// settings that specify the maximum duration a connection can
+	// remain open without data being sent across it. If the idle
+	// timeout is exceeded without any data being transmitted, the
+	// connection may be closed when connecting SSE over HTTP/1.
+	//
+	// End-to-end HTTP/2 connections do not require a ping interval
+	// to keep the connection open.
+	KeepAlivePingInterval: 10 * time.Second,
+})
+```
+
 The GraphQL playground does not support SSE yet. You can try out the subscription via curl:
 
 ```bash
diff --git a/graphql/handler/transport/sse.go b/graphql/handler/transport/sse.go
index 1d59fdff..b9d08e2d 100644
--- a/graphql/handler/transport/sse.go
+++ b/graphql/handler/transport/sse.go
@@ -1,6 +1,7 @@
 package transport
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"io"
@@ -8,13 +9,26 @@ import (
 	"mime"
 	"net/http"
 	"strings"
+	"sync"
+	"time"
 
 	"github.com/vektah/gqlparser/v2/gqlerror"
 
 	"github.com/99designs/gqlgen/graphql"
 )
 
-type SSE struct{}
+type (
+	SSE struct {
+		KeepAlivePingInterval time.Duration
+	}
+
+	sseConnection struct {
+		ctx             context.Context
+		mu              sync.Mutex
+		f               http.Flusher
+		keepAliveTicker *time.Ticker
+	}
+)
 
 var _ graphql.Transport = SSE{}
 
@@ -36,7 +50,13 @@ func (t SSE) Do(w http.ResponseWriter, r *http.Request, exec graphql.GraphExecut
 		SendErrorf(w, http.StatusInternalServerError, "streaming unsupported")
 		return
 	}
-	defer flusher.Flush()
+
+	c := &sseConnection{
+		ctx: ctx,
+		f:   flusher,
+	}
+
+	defer c.flush()
 
 	w.Header().Set("Cache-Control", "no-cache")
 	w.Header().Set("Connection", "keep-alive")
@@ -75,10 +95,19 @@ func (t SSE) Do(w http.ResponseWriter, r *http.Request, exec graphql.GraphExecut
 
 	rc, opErr := exec.CreateOperationContext(ctx, params)
 	ctx = graphql.WithOperationContext(ctx, rc)
+	c.ctx = ctx
 
 	w.Header().Set("Content-Type", "text/event-stream")
 	fmt.Fprint(w, ":\n\n")
-	flusher.Flush()
+	c.flush()
+
+	if t.KeepAlivePingInterval > 0 {
+		c.mu.Lock()
+		c.keepAliveTicker = time.NewTicker(t.KeepAlivePingInterval)
+		c.mu.Unlock()
+
+		go c.keepAlive(w)
+	}
 
 	if opErr != nil {
 		resp := exec.DispatchError(ctx, opErr)
@@ -91,13 +120,42 @@ func (t SSE) Do(w http.ResponseWriter, r *http.Request, exec graphql.GraphExecut
 				break
 			}
 			writeJsonWithSSE(w, response)
-			flusher.Flush()
+			c.flush()
+
+			c.resetTicker(t.KeepAlivePingInterval)
 		}
 	}
 
 	fmt.Fprint(w, "event: complete\n\n")
 }
 
+func (c *sseConnection) resetTicker(interval time.Duration) {
+	if interval != 0 {
+		c.mu.Lock()
+		c.keepAliveTicker.Reset(interval)
+		c.mu.Unlock()
+	}
+}
+
+func (c *sseConnection) keepAlive(w io.Writer) {
+	for {
+		select {
+		case <-c.ctx.Done():
+			c.keepAliveTicker.Stop()
+			return
+		case <-c.keepAliveTicker.C:
+			fmt.Fprintf(w, ": ping\n\n")
+			c.flush()
+		}
+	}
+}
+
+func (c *sseConnection) flush() {
+	c.mu.Lock()
+	c.f.Flush()
+	c.mu.Unlock()
+}
+
 func writeJsonWithSSE(w io.Writer, response *graphql.Response) {
 	b, err := json.Marshal(response)
 	if err != nil {


Test patch:
diff --git a/graphql/handler/transport/sse_test.go b/graphql/handler/transport/sse_test.go
index 1b8c8e5e..c85c497e 100644
--- a/graphql/handler/transport/sse_test.go
+++ b/graphql/handler/transport/sse_test.go
@@ -8,15 +8,17 @@ import (
 	"strings"
 	"sync"
 	"testing"
-
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
+	"time"
 
 	"github.com/99designs/gqlgen/graphql/handler/testserver"
 	"github.com/99designs/gqlgen/graphql/handler/transport"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestSSE(t *testing.T) {
+	pingInterval := time.Second * 1
+
 	initialize := func() *testserver.TestServer {
 		h := testserver.New()
 		h.AddTransport(transport.SSE{})
@@ -28,6 +30,14 @@ func TestSSE(t *testing.T) {
 		return h, httptest.NewServer(h)
 	}
 
+	initializeKeepAliveWithServer := func() (*testserver.TestServer, *httptest.Server) {
+		h := testserver.New()
+		h.AddTransport(transport.SSE{
+			KeepAlivePingInterval: pingInterval,
+		})
+		return h, httptest.NewServer(h)
+	}
+
 	createHTTPTestRequest := func(query string) *http.Request {
 		req := httptest.NewRequest(http.MethodPost, "/graphql", strings.NewReader(query))
 		req.Header.Set("Accept", "text/event-stream")
@@ -147,4 +157,50 @@ func TestSSE(t *testing.T) {
 
 		wg.Wait()
 	})
+
+	t.Run("subscribe with keep alive", func(t *testing.T) {
+		handler, srv := initializeKeepAliveWithServer()
+		defer srv.Close()
+
+		var wg sync.WaitGroup
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			// Wait for ping interval to trigger
+			time.Sleep(pingInterval + time.Millisecond*100)
+		}()
+
+		client := &http.Client{}
+		req := createHTTPRequest(srv.URL, `{"query":"subscription { name }"}`)
+		res, err := client.Do(req)
+		require.NoError(t, err, "Request threw error -> %s", err)
+		defer func() {
+			require.NoError(t, res.Body.Close())
+		}()
+
+		assert.Equal(t, 200, res.StatusCode, "Request return wrong status -> %d", res.Status)
+		assert.Equal(t, "keep-alive", res.Header.Get("Connection"))
+		assert.Equal(t, "text/event-stream", res.Header.Get("Content-Type"))
+
+		br := bufio.NewReader(res.Body)
+
+		assert.Equal(t, ":\n", readLine(br))
+		assert.Equal(t, "\n", readLine(br))
+		assert.Equal(t, ": ping\n", readLine(br))
+		assert.Equal(t, "\n", readLine(br))
+
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			handler.SendCompleteSubscriptionMessage()
+		}()
+
+		assert.Equal(t, "event: complete\n", readLine(br))
+		assert.Equal(t, "\n", readLine(br))
+
+		_, err = br.ReadByte()
+		assert.Equal(t, err, io.EOF)
+
+		wg.Wait()
+	})
 }