[TT-14253] Unload Streams Properly (#7033)

### **User description**
PR for https://tyktech.atlassian.net/browse/TT-14253

This story is related to https://tyktech.atlassian.net/browse/TT-14252

I cannot find a reliable way to implement an integration test for this
particular case. The test suite and Tyk Streams require a refactor to
test this case in a programatic way.

It's not possible to test it via tyk-analytics integration tests because
the effect of calling `Unload` method is hidden under layers of
abstraction. We can only understand the stream is unloaded from GW logs
if the debug mode is enabled.

It's possible to reproduce TT-14253 and TT-14252 with the following
streams config:

```yaml
input:
  label: ""
  http_client:
    url: "" # No default (required)
    verb: GET
    headers: {}
    rate_limit: "" # No default (optional)
    timeout: 5s
    payload: "" # No default (optional)
    stream:
      enabled: false
      reconnect: true
      scanner:
        lines: {}
    auto_replay_nacks: true
output:
  label: ""
  http_client:
    url: "" # No default (required)
    verb: POST
    headers: {}
    rate_limit: "" # No default (optional)
    timeout: 5s
    max_in_flight: 64
    batching:
      count: 0
      byte_size: 0
      period: ""
      check: ""
```


___

### **PR Type**
Enhancement


___

### **Description**
- Properly unload Tyk Streams when an API is removed

- Add Unload method to middleware interface and wrappers

- Ensure all middleware implement Unload, even if empty

- Improve stream reset logic and logging in Streams middleware


___



### **Changes walkthrough** 📝
<table><thead><tr><th></th><th align="left">Relevant
files</th></tr></thead><tbody><tr><td><strong>Enhancement</strong></td><td><table>
<tr>
  <td>
    <details>
<summary><strong>middleware.go</strong><dd><code>Refactor and enhance
stream unloading logic</code>&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; </dd></summary>
<hr>

ee/middleware/streams/middleware.go

<li>Refactored stream reset logic into a helper method<br> <li> Enhanced
Unload to reset all cached and default streams<br> <li> Improved logging
for stream reset errors<br> <li> Clarified Unload's purpose in comments


</details>


  </td>
<td><a
href="https://github.com/TykTechnologies/tyk/pull/7033/files#diff-0ce428c0f09dca65e3df6e72d01fee63b6f237785e41e6ecf0ce34a8b65c74a5">+19/-7</a>&nbsp;
&nbsp; </td>

</tr>

<tr>
  <td>
    <details>
<summary><strong>middleware.go</strong><dd><code>Add Unload method to
UpstreamBasicAuth middleware</code>&nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
&nbsp; &nbsp; &nbsp; </dd></summary>
<hr>

ee/middleware/upstreambasicauth/middleware.go

<li>Added Unload method with no-op implementation<br> <li> Ensured
middleware conforms to new interface


</details>


  </td>
<td><a
href="https://github.com/TykTechnologies/tyk/pull/7033/files#diff-4738eae6b1fa23f58598d492a9fa4fa6682545bf00bf122040180425a5e9dc60">+4/-0</a>&nbsp;
&nbsp; &nbsp; </td>

</tr>

<tr>
  <td>
    <details>
<summary><strong>middleware.go</strong><dd><code>Add Unload method to
UpstreamOAuth middleware</code>&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; </dd></summary>
<hr>

ee/middleware/upstreamoauth/middleware.go

<li>Added Unload method with no-op implementation<br> <li> Ensured
middleware conforms to new interface


</details>


  </td>
<td><a
href="https://github.com/TykTechnologies/tyk/pull/7033/files#diff-1347b256f3728407a2697f1e824391eaa329162e644741d8321f25c7a0630363">+4/-0</a>&nbsp;
&nbsp; &nbsp; </td>

</tr>

<tr>
  <td>
    <details>
<summary><strong>api_loader.go</strong><dd><code>Detect and prepare
removed APIs for unloading</code>&nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; </dd></summary>
<hr>

gateway/api_loader.go

<li>Identified removed API specs for unloading<br> <li> Prepared
specsToUnload list for proper cleanup


</details>


  </td>
<td><a
href="https://github.com/TykTechnologies/tyk/pull/7033/files#diff-cdf0b7f176c9d18e1a314b78ddefc2cb3a94b3de66f1f360174692c915734c68">+7/-0</a>&nbsp;
&nbsp; &nbsp; </td>

</tr>

<tr>
  <td>
    <details>
<summary><strong>middleware_wrap.go</strong><dd><code>Add Unload to
middleware wrapper and fix receiver names</code>&nbsp; &nbsp;
</dd></summary>
<hr>

gateway/middleware_wrap.go

<li>Implemented Unload method in wrapMiddleware<br> <li> Fixed receiver
naming for consistency<br> <li> Ensured interface compliance


</details>


  </td>
<td><a
href="https://github.com/TykTechnologies/tyk/pull/7033/files#diff-1da43bd02220acad12bce8d8c5600b4acfee3c40f90c53825802747004c9fb0a">+7/-3</a>&nbsp;
&nbsp; &nbsp; </td>

</tr>

<tr>
  <td>
    <details>
<summary><strong>interfaces.go</strong><dd><code>Extend Middleware
interface with Unload method</code>&nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; </dd></summary>
<hr>

internal/model/interfaces.go

<li>Added Unload method to Middleware interface<br> <li> Updated
interface documentation accordingly


</details>


  </td>
<td><a
href="https://github.com/TykTechnologies/tyk/pull/7033/files#diff-43ba6dd4a8d193850dea32e8af5c361470cd62bfa390c580a39f7142a56bd391">+1/-0</a>&nbsp;
&nbsp; &nbsp; </td>

</tr>
</table></td></tr></tr></tbody></table>

___

> <details> <summary> Need help?</summary><li>Type <code>/help how to
...</code> in the comments thread for any questions about PR-Agent
usage.</li><li>Check out the <a
href="https://qodo-merge-docs.qodo.ai/usage-guide/">documentation</a>
for more information.</li></details>

Author: buraksezer

ee/middleware/streams/middleware.go

@@ -263,30 +263,42 @@ func (s *Middleware) ProcessRequest(w http.ResponseWriter, r *http.Request, _ in
 	return nil, middleware.StatusRespond
 }
 
-// Unload closes and remove active streams.
+func (s *Middleware) resetStream(streamValue any) {
+	if stream, ok := streamValue.(*Stream); ok {
+		if err := stream.Reset(); err != nil {
+			s.Logger().WithError(err).Error("Failed to reset stream")
+		}
+	}
+}
+
+// Unload closes and remove active streams. This method is called when the API is removed.
 func (s *Middleware) Unload() {
 	s.Logger().Debugf("Unloading streaming middleware %s", s.Spec.Name)
 
 	totalStreams := 0
 	s.cancel()
 
+	// Reset cached streams and stop the underlying Bento instances
 	s.StreamManagerCache.Range(func(_, value interface{}) bool {
 		manager, ok := value.(*Manager)
 		if !ok {
 			return true
 		}
 		manager.streams.Range(func(_, streamValue interface{}) bool {
 			totalStreams++
-			if stream, ok := streamValue.(*Stream); ok {
-				if err := stream.Reset(); err != nil {
-					s.Logger().WithError(err).Error("Failed to reset stream")
-				}
-			}
-			return true
+			s.resetStream(streamValue)
+			return true // continue iterating
 		})
 		return true
 	})
 
+	// Finally, reset the default manager and stop the underlying Bento instance
+	s.defaultManager.streams.Range(func(_, streamValue interface{}) bool {
+		totalStreams++
+		s.resetStream(streamValue)
+		return true // continue iterating
+	})
+
 	GlobalStreamCounter.Add(-int64(totalStreams))
 	s.StreamManagerCache = sync.Map{}
 	s.Logger().Info("All streams successfully removed")

ee/middleware/upstreambasicauth/middleware.go

@@ -19,6 +19,10 @@ type Middleware struct {
 	base BaseMiddleware
 }
 
+func (m *Middleware) Unload() {
+	// nothing to do here
+}
+
 // Middleware implements model.Middleware.
 var _ model.Middleware = &Middleware{}
 

ee/middleware/upstreamoauth/middleware.go

@@ -23,6 +23,10 @@ type Middleware struct {
 	passwordStorageHandler          Storage
 }
 
+func (m *Middleware) Unload() {
+	// nothing to do here
+}
+
 // Middleware implements model.Middleware.
 var _ model.Middleware = &Middleware{}
 

gateway/api_loader.go

@@ -1070,6 +1070,13 @@ func (gw *Gateway) loadApps(specs []*APISpec) {
 		}
 	}
 
+	// Find the removed specs to unload them
+	for apiID, curSpec := range gw.apisByID {
+		if _, ok := tmpSpecRegister[apiID]; !ok {
+			specsToUnload = append(specsToUnload, curSpec)
+		}
+	}
+
 	gw.apisByID = tmpSpecRegister
 	gw.apisHandlesByID = tmpSpecHandles
 

gateway/middleware_wrap.go

@@ -13,7 +13,7 @@ type wrapMiddleware struct {
 	mw model.Middleware
 }
 
-var _ TykMiddleware = &wrapMiddleware{}
+var _ TykMiddleware = (*wrapMiddleware)(nil)
 
 // WrapMiddleware returns a new TykMiddleware with the provided base middleware,
 // and the smaller model.Middleware interface. It allows to implement model.Middleware,
@@ -41,8 +41,8 @@ func (w *wrapMiddleware) Name() string {
 	return w.mw.Name()
 }
 
-func (s *wrapMiddleware) Logger() *logrus.Entry {
-	return s.mw.Logger()
+func (w *wrapMiddleware) Logger() *logrus.Entry {
+	return w.mw.Logger()
 }
 
 func (w *wrapMiddleware) EnabledForSpec() bool {
@@ -52,3 +52,7 @@ func (w *wrapMiddleware) EnabledForSpec() bool {
 func (w *wrapMiddleware) ProcessRequest(rw http.ResponseWriter, r *http.Request, data interface{}) (error, int) {
 	return w.mw.ProcessRequest(rw, r, data)
 }
+
+func (w *wrapMiddleware) Unload() {
+	w.mw.Unload()
+}

internal/graphengine/engine_v2.go

@@ -239,24 +239,23 @@ func (e *EngineV2) HandleReverseProxy(params ReverseProxyParams) (res *http.Resp
 		return nil, false, err
 	}
 
-	gqlRequest := e.ctxRetrieveRequestFunc(params.OutRequest)
-	// Cleanup method, frees allocated resources, if they are eligible for freeing up.
-	// Currently, it only frees up the allocated resources of a GraphQL query that
-	// has a cached query plan.
-	//
-	// graphql-go-tools uses the parsed query (ast.Document in graphql-go-tools codebase)
-	// in the planner and caches the plans. If a plan has been cached, we can reset the created
-	// ast.Document struct and put it back to the pool for later use. By this way, we can reduce the GC
-	// pressure and number of allocations per GraphQL query.
-	// See TT-9864 for the details.
-	defer gqlRequest.Cleanup()
-
 	switch reverseProxyType {
 	case ReverseProxyTypeIntrospection:
 		return e.gqlTools.handleIntrospection(e.Schema)
 	case ReverseProxyTypeWebsocketUpgrade:
 		return e.handoverWebSocketConnectionToGraphQLExecutionEngine(&params)
 	case ReverseProxyTypeGraphEngine:
+		gqlRequest := e.ctxRetrieveRequestFunc(params.OutRequest)
+		// Cleanup method frees allocated resources if they are eligible for freeing up.
+		// Currently, it only frees up the allocated resources of a GraphQL query that
+		// has a cached query plan.
+		//
+		// graphql-go-tools uses the parsed query (ast.Document in graphql-go-tools codebase)
+		// in the planner and caches the plans. If a plan has been cached, we can reset the created
+		// ast.Document struct and put it back in the pool for later use. In this way, we can reduce the GC
+		// pressure and number of allocations per GraphQL query.
+		// See TT-9864 for the details.
+		defer gqlRequest.Cleanup()
 		return e.handoverRequestToGraphQLExecutionEngine(gqlRequest, params.OutRequest)
 	case ReverseProxyTypePreFlight:
 		if e.ApiDefinition.GraphQL.ExecutionMode == apidef.GraphQLExecutionModeProxyOnly {

internal/model/interfaces.go

@@ -27,6 +27,7 @@ type Middleware interface {
 	Logger() *logrus.Entry
 	ProcessRequest(w http.ResponseWriter, r *http.Request, conf interface{}) (error, int) // Handles request
 	EnabledForSpec() bool
+	Unload()
 }
 
 // LoggerProvider returns a new *logrus.Entry for the request.