[TT-12318] SSE Streaming is broken

[alekc]

Branch/Environment/Version

• 5.3.1, master, some earlier versions
• Environment: On-prem

Describe the bug
if upstream of the reverse proxy is a sse enabled/streaming server, tyk should be providing data back as soon as it receives it, whats happening instead is "tyk not providing any response until either it reaches the timeout or the connections is closed upstream". So basically buffering until it can.

Reproduction steps
Given following app

{
  "name": "Tyk Test API",
  "api_id": "3000",
  "org_id": "default",
  "definition": {
    "location": "",
    "key": ""
  },
  "use_keyless": true,
  "auth": {
    "auth_header_name": ""
  },
  "version_data": {
    "not_versioned": true,
    "versions": {
      "Default": {
        "name": "Default",
        "expires": "3000-01-02 15:04",
        "use_extended_paths": true,
        "extended_paths": {
          "ignored": [],
          "white_list": [],
          "black_list": []
        }
      }
    }
  },
  "proxy": {
    "listen_path": "/sse/",
    "target_url": "https://sse.dev/test",
    "strip_listen_path": true
  },
  "do_not_track": true
}

and this tyk conf

{
  "listen_address": "",
  "listen_port": 8080,
  "secret": "352d20ee67be67f6340b4c0605b044b7",
  "template_path": "./templates",
  "use_db_app_configs": false,
  "app_path": "./apps",
  "middleware_path": "./middleware",
  "log_level": "debug",
  "track_404_logs": true,
  "http_server_options": {
      "read_timeout": 0,
      "write_timeout": 0,
      "use_ssl": false,
      "enable_http2": true,
      "enable_strict_routes": false,
      "ssl_insecure_skip_verify": false,
      "enable_websockets": true,
      "certificates": null,
      "ssl_certificates": null,
      "server_name": "",
      "min_version": 0,
      "max_version": 0,
      "skip_client_ca_announcement": false,
      "flush_interval": 0,
      "skip_url_cleaning": false,
      "skip_target_path_escaping": false,
      "ssl_ciphers": null,
      "max_request_body_size": 0
  },
  "storage": {
    "type": "redis",
    "host": "10.19.80.14",
    "port": 6379,
    "username": "",
    "password": "",
    "database": 0,
    "optimisation_max_idle": 2000,
    "optimisation_max_active": 4000
  },
  "enable_analytics": false,
  "analytics_config": {
    "type": "",
    "ignored_ips": []
  },
  "dns_cache": {
    "enabled": false,
    "ttl": 3600,
    "check_interval": 60
  },
  "allow_master_keys": false,
  "policies": {
    "policy_source": "file"
  },
  "hash_keys": true,
  "hash_key_function": "murmur64",
  "suppress_redis_signal_reload": false,
  "force_global_session_lifetime": false,
  "max_idle_connections_per_host": 500
}

any calls to /sse/ will result in no response being received in realtime.

Additional context
Upon the investigation I found the issue (but I do not have enough familiarity with the project to attempt a proper resolution).

when request arrives at this point

tyk/gateway/reverse_proxy.go

Lines 1446 to 1465 in 0d1183c

	func (p *ReverseProxy) CopyResponse(dst io.Writer, src io.Reader, flushInterval time.Duration) {
	if flushInterval != 0 {
	if wf, ok := dst.(writeFlusher); ok {
	mlw := &maxLatencyWriter{
	dst: wf,
	latency: flushInterval,
	}
	defer mlw.stop()
	// set up initial timer so headers get flushed even if body writes are delayed
	mlw.flushPending = true
	mlw.t = time.AfterFunc(flushInterval, mlw.delayedFlush)
	dst = mlw
	}
	}
	p.copyBuffer(dst, src)
	}

the flow will enter correctly in the if flushInterval != 0 { block (due to being recognised here

tyk/gateway/reverse_proxy.go

Lines 1428 to 1436 in 0d1183c

	// overriding its value for a specific request/response.
	func (p *ReverseProxy) flushInterval(res *http.Response) time.Duration {
	resCT := res.Header.Get("Content-Type")
	// For Server-Sent Events responses, flush immediately.
	// The MIME type is defined in https://www.w3.org/TR/eventsource/#text-event-stream
	if resCT == "text/event-stream" {
	return -1 // negative means immediately
	}

) but the casting if wf, ok := dst.(writeFlusher); ok { will fail, and it won't enter inside.

The reason for this can be traced here:

tyk/gateway/reverse_proxy.go

Lines 1337 to 1341 in 0d1183c

	var bodyBuffer bytes.Buffer
	bodyBuffer2 := new(bytes.Buffer)
	p.CopyResponse(&bodyBuffer, res.Body, p.flushInterval(res))
	*bodyBuffer2 = bodyBuffer

If the cache is enabled, then the dst is going to be a bytes.Buffer and won't satisfy the required interface.

As a workaround I've changed

tyk/gateway/reverse_proxy.go

Lines 517 to 521 in 0d1183c

	if IsGrpcStreaming(req) {
	resp = p.WrappedServeHTTP(rw, req, false)
	} else {
	resp = p.WrappedServeHTTP(rw, req, true)
	}

into

switch {
	case IsGrpcStreaming(req), req.Header.Get("Enable-Stream") == "true":
		resp = p.WrappedServeHTTP(rw, req, false)
	default:
		resp = p.WrappedServeHTTP(rw, req, true)
	}

With that piece which disables the caching, the streaming works correctly.

The downside with that solution (besides usage of a specific header), is that if the TYK_GW_ANALYTICSCONFIG_ENABLEDETAILEDRECORDING is true, the tyk will panic out on

tyk/gateway/handler_success.go

Line 227 in 0d1183c

responseContent, err := io.ReadAll(responseCopy.Body)

because the responseCopy.Body will be nil (it has been already dumped earlier during the streaming).

[oluwaseyeayinla]

Thanks for reporting this. I can reproduce the behaviour and have opened an internal ticket to track this

[buger]

@alekc can you try to set flush interval to 1?

[alekc]

potentially yes, but would not that break the real time streaming? as in, it would be a workaround not a fix.

EDIT: It won't probably work because TYK will override the flush interval for sse

tyk/gateway/reverse_proxy.go

Lines 1434 to 1436 in 0d1183c

	if resCT == "text/event-stream" {
	return -1 // negative means immediately
	}