CVE-2024-23820: Missing Release of Memory after Effective Lifetime (CWE-401)
value: OpenFGA, an authorization/permission engine, is vulnerable to a denial of service attack in versions prior to 1.4.3. In some scenarios that depend on the model and tuples used, a call to `ListObjects` may not  release memory properly. So when a sufficiently high number of those calls are executed, the OpenFGA server can create an `out of memory` error and terminate. Version 1.4.3 contains a patch for this issue.
CWE description: The product does not sufficiently track and release allocated memory after it has been used, which slowly consumes remaining memory.
CVSS v3 base score: 6.5
Published: 2024-01-26T17:15Z
Repository: https://github.com/openfga/openfga
Language: Go
Files changed: pkg/server/commands/list_objects.go, pkg/server/commands/reverseexpand/reverse_expand.go, pkg/server/commands/reverseexpand/reverse_expand_test.go, pkg/server/server_test.go, pkg/server/test/list_objects.go, pkg/server/test/reverse_expand.go, pkg/testfixtures/storage/mysql.go, pkg/testfixtures/storage/postgres.go, tests/listobjects/listobjects_test.go
Commit date: 2024-01-25 15:10:51 -0700
Commit message: fix: handle ReverseExpand channel closure correctly (#1315)


Diff: diff --git a/pkg/server/commands/list_objects.go b/pkg/server/commands/list_objects.go
index 58a7c9e8..aea751da 100644
--- a/pkg/server/commands/list_objects.go
+++ b/pkg/server/commands/list_objects.go
@@ -146,10 +146,18 @@ type listObjectsRequest interface {
 	GetUser() string
 	GetContextualTuples() *openfgav1.ContextualTupleKeys
 	GetContext() *structpb.Struct
 }
 
+// evaluate fires of evaluation of the ListObjects query by delegating to
+// [[reverseexpand.ReverseExpand#Execute]] and resolving the results yielded
+// from it. If any results yielded by reverse expansion require further eval,
+// then these results get dispatched to Check to resolve the residual outcome.
+//
+// The resultsChan is **always** closed by evaluate when it is done with its work,
+// which is either when all results have been yielded, the deadline has been met,
+// or some other terminal error case has occurred.
 func (q *ListObjectsQuery) evaluate(
 	ctx context.Context,
 	req listObjectsRequest,
 	resultsChan chan<- ListObjectsResult,
 	maxResults uint32,
@@ -216,100 +224,114 @@ func (q *ListObjectsQuery) evaluate(
 		}
 
 		reverseExpandResultsChan := make(chan *reverseexpand.ReverseExpandResult, 1)
 		objectsFound := atomic.Uint32{}
 
-		reverseExpandQuery := reverseexpand.NewReverseExpandQuery(q.datastore, typesys,
+		reverseExpandQuery := reverseexpand.NewReverseExpandQuery(
+			q.datastore,
+			typesys,
 			reverseexpand.WithResolveNodeLimit(q.resolveNodeLimit),
 			reverseexpand.WithResolveNodeBreadthLimit(q.resolveNodeBreadthLimit),
 			reverseexpand.WithLogger(q.logger),
 		)
 
 		cancelCtx, cancel := context.WithCancel(ctx)
 
 		wg := sync.WaitGroup{}
 
+		errChan := make(chan error, 1)
+
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
 
-			reverseExpandQuery.Execute(cancelCtx, &reverseexpand.ReverseExpandRequest{
+			err := reverseExpandQuery.Execute(cancelCtx, &reverseexpand.ReverseExpandRequest{
 				StoreID:          req.GetStoreId(),
 				ObjectType:       targetObjectType,
 				Relation:         targetRelation,
 				User:             sourceUserRef,
 				ContextualTuples: req.GetContextualTuples().GetTupleKeys(),
 				Context:          req.GetContext(),
 			}, reverseExpandResultsChan, resolutionMetadata)
+			if err != nil {
+				errChan <- err
+			}
 		}()
 
 		checkResolver := graph.NewLocalChecker(
 			storagewrappers.NewCombinedTupleReader(q.datastore, req.GetContextualTuples().GetTupleKeys()),
 			q.checkOptions...,
 		)
 		defer checkResolver.Close()
 
 		concurrencyLimiterCh := make(chan struct{}, q.resolveNodeBreadthLimit)
 
-		for res := range reverseExpandResultsChan {
-			if res.Err != nil {
-				err := res.Err
-
-				if errors.Is(err, graph.ErrResolutionDepthExceeded) || errors.Is(err, graph.ErrCycleDetected) {
-					err = serverErrors.AuthorizationModelResolutionTooComplex
+	ConsumerReadLoop:
+		for {
+			select {
+			case <-ctx.Done():
+				break ConsumerReadLoop
+			case res, channelOpen := <-reverseExpandResultsChan:
+				if !channelOpen {
+					break ConsumerReadLoop
 				}
 
-				resultsChan <- ListObjectsResult{Err: err}
-				break
-			}
-
-			if !(maxResults == 0) && objectsFound.Load() >= maxResults {
-				break
-			}
+				if !(maxResults == 0) && objectsFound.Load() >= maxResults {
+					break ConsumerReadLoop
+				}
 
-			if res.ResultStatus == reverseexpand.NoFurtherEvalStatus {
-				noFurtherEvalRequiredCounter.Inc()
-				trySendObject(res.Object, &objectsFound, maxResults, resultsChan)
-				continue
-			}
+				if res.ResultStatus == reverseexpand.NoFurtherEvalStatus {
+					noFurtherEvalRequiredCounter.Inc()
+					trySendObject(res.Object, &objectsFound, maxResults, resultsChan)
+					continue
+				}
 
-			furtherEvalRequiredCounter.Inc()
-
-			wg.Add(1)
-			go func(res *reverseexpand.ReverseExpandResult) {
-				defer func() {
-					<-concurrencyLimiterCh
-					wg.Done()
-				}()
-
-				concurrencyLimiterCh <- struct{}{}
-
-				resp, err := checkResolver.ResolveCheck(ctx, &graph.ResolveCheckRequest{
-					StoreID:              req.GetStoreId(),
-					AuthorizationModelID: req.GetAuthorizationModelId(),
-					TupleKey:             tuple.NewTupleKey(res.Object, req.GetRelation(), req.GetUser()),
-					ContextualTuples:     req.GetContextualTuples().GetTupleKeys(),
-					Context:              req.GetContext(),
-					ResolutionMetadata: &graph.ResolutionMetadata{
-						Depth: q.resolveNodeLimit,
-					},
-				})
-				if err != nil {
-					if errors.Is(err, graph.ErrResolutionDepthExceeded) || errors.Is(err, graph.ErrCycleDetected) {
-						resultsChan <- ListObjectsResult{Err: serverErrors.AuthorizationModelResolutionTooComplex}
+				furtherEvalRequiredCounter.Inc()
+
+				wg.Add(1)
+				go func(res *reverseexpand.ReverseExpandResult) {
+					defer func() {
+						<-concurrencyLimiterCh
+						wg.Done()
+					}()
+
+					concurrencyLimiterCh <- struct{}{}
+
+					resp, err := checkResolver.ResolveCheck(ctx, &graph.ResolveCheckRequest{
+						StoreID:              req.GetStoreId(),
+						AuthorizationModelID: req.GetAuthorizationModelId(),
+						TupleKey:             tuple.NewTupleKey(res.Object, req.GetRelation(), req.GetUser()),
+						ContextualTuples:     req.GetContextualTuples().GetTupleKeys(),
+						Context:              req.GetContext(),
+						ResolutionMetadata: &graph.ResolutionMetadata{
+							Depth: q.resolveNodeLimit,
+						},
+					})
+					if err != nil {
+						if errors.Is(err, graph.ErrResolutionDepthExceeded) || errors.Is(err, graph.ErrCycleDetected) {
+							resultsChan <- ListObjectsResult{Err: serverErrors.AuthorizationModelResolutionTooComplex}
+							return
+						}
+
+						resultsChan <- ListObjectsResult{Err: err}
 						return
 					}
+					atomic.AddUint32(resolutionMetadata.QueryCount, resp.GetResolutionMetadata().DatastoreQueryCount)
 
-					resultsChan <- ListObjectsResult{Err: err}
-					return
-				}
-				atomic.AddUint32(resolutionMetadata.QueryCount, resp.GetResolutionMetadata().DatastoreQueryCount)
+					if resp.Allowed {
+						trySendObject(res.Object, &objectsFound, maxResults, resultsChan)
+					}
+				}(res)
 
-				if resp.Allowed {
-					trySendObject(res.Object, &objectsFound, maxResults, resultsChan)
+			case err := <-errChan:
+				if errors.Is(err, graph.ErrResolutionDepthExceeded) || errors.Is(err, graph.ErrCycleDetected) {
+					err = serverErrors.AuthorizationModelResolutionTooComplex
 				}
-			}(res)
+
+				resultsChan <- ListObjectsResult{Err: err}
+				break ConsumerReadLoop
+			}
 		}
 
 		cancel()
 		wg.Wait()
 		close(resultsChan)
@@ -356,52 +378,40 @@ func (q *ListObjectsQuery) Execute(
 	}
 
 	objects := make([]string, 0)
 
 	var errs *multierror.Error
-	for {
-		select {
-		case <-timeoutCtx.Done():
-			q.logger.WarnWithContext(
-				ctx, fmt.Sprintf("list objects timeout after %s", q.listObjectsDeadline.String()),
-			)
-			return &ListObjectsResponse{
-				Objects:            objects,
-				ResolutionMetadata: *resolutionMetadata,
-			}, nil
-
-		case result, channelOpen := <-resultsChan:
-			if result.Err != nil {
-				if errors.Is(result.Err, serverErrors.AuthorizationModelResolutionTooComplex) {
-					return nil, result.Err
-				}
 
-				if errors.Is(result.Err, condition.ErrEvaluationFailed) {
-					errs = multierror.Append(errs, result.Err)
-					continue
-				}
-
-				if errors.Is(result.Err, context.Canceled) || errors.Is(result.Err, context.DeadlineExceeded) {
-					continue
-				}
-
-				return nil, serverErrors.HandleError("", result.Err)
+	for result := range resultsChan {
+		if result.Err != nil {
+			if errors.Is(result.Err, serverErrors.AuthorizationModelResolutionTooComplex) {
+				return nil, result.Err
 			}
 
-			if !channelOpen {
-				if len(objects) < int(maxResults) && errs.ErrorOrNil() != nil {
-					return nil, errs
-				}
+			if errors.Is(result.Err, condition.ErrEvaluationFailed) {
+				errs = multierror.Append(errs, result.Err)
+				continue
+			}
 
-				return &ListObjectsResponse{
-					Objects:            objects,
-					ResolutionMetadata: *resolutionMetadata,
-				}, nil
+			if errors.Is(result.Err, context.Canceled) || errors.Is(result.Err, context.DeadlineExceeded) {
+				continue
 			}
-			objects = append(objects, result.ObjectID)
+
+			return nil, serverErrors.HandleError("", result.Err)
 		}
+
+		objects = append(objects, result.ObjectID)
 	}
+
+	if len(objects) < int(maxResults) && errs.ErrorOrNil() != nil {
+		return nil, errs
+	}
+
+	return &ListObjectsResponse{
+		Objects:            objects,
+		ResolutionMetadata: *resolutionMetadata,
+	}, nil
 }
 
 // ExecuteStreamed executes the ListObjectsQuery, returning a stream of object IDs.
 // It ignores the value of q.listObjectsMaxResults and returns all available results
 // until q.listObjectsDeadline is hit
@@ -422,39 +432,27 @@ func (q *ListObjectsQuery) ExecuteStreamed(ctx context.Context, req *openfgav1.S
 	err := q.evaluate(timeoutCtx, req, resultsChan, maxResults, resolutionMetadata)
 	if err != nil {
 		return nil, err
 	}
 
-	for {
-		select {
-		case <-timeoutCtx.Done():
-			q.logger.WarnWithContext(
-				ctx, fmt.Sprintf("list objects timeout after %s", q.listObjectsDeadline.String()),
-			)
-			return resolutionMetadata, nil
-
-		case result, channelOpen := <-resultsChan:
-			if !channelOpen {
-				// Channel closed! No more results.
-				return resolutionMetadata, nil
+	for result := range resultsChan {
+		if result.Err != nil {
+			if errors.Is(result.Err, serverErrors.AuthorizationModelResolutionTooComplex) {
+				return nil, result.Err
 			}
 
-			if result.Err != nil {
-				if errors.Is(result.Err, serverErrors.AuthorizationModelResolutionTooComplex) {
-					return nil, result.Err
-				}
-
-				if errors.Is(result.Err, condition.ErrEvaluationFailed) {
-					return nil, serverErrors.ValidationError(result.Err)
-				}
-
-				return nil, serverErrors.HandleError("", result.Err)
+			if errors.Is(result.Err, condition.ErrEvaluationFailed) {
+				return nil, serverErrors.ValidationError(result.Err)
 			}
 
-			if err := srv.Send(&openfgav1.StreamedListObjectsResponse{
-				Object: result.ObjectID,
-			}); err != nil {
-				return nil, serverErrors.NewInternalError("", err)
-			}
+			return nil, serverErrors.HandleError("", result.Err)
+		}
+
+		if err := srv.Send(&openfgav1.StreamedListObjectsResponse{
+			Object: result.ObjectID,
+		}); err != nil {
+			return nil, serverErrors.NewInternalError("", err)
 		}
 	}
+
+	return resolutionMetadata, nil
 }
diff --git a/pkg/server/commands/reverseexpand/reverse_expand.go b/pkg/server/commands/reverseexpand/reverse_expand.go
index 09dec136..c41914ca 100644
--- a/pkg/server/commands/reverseexpand/reverse_expand.go
+++ b/pkg/server/commands/reverseexpand/reverse_expand.go
@@ -159,11 +159,10 @@ const (
 	RequiresFurtherEvalStatus ConditionalResultStatus = iota
 	NoFurtherEvalStatus
 )
 
 type ReverseExpandResult struct {
-	Err          error
 	Object       string
 	ResultStatus ConditionalResultStatus
 }
 
 type ResolutionMetadata struct {
@@ -180,35 +179,35 @@ func WithLogger(logger logger.Logger) ReverseExpandQueryOption {
 	return func(d *ReverseExpandQuery) {
 		d.logger = logger
 	}
 }
 
-// Execute yields all the objects of the provided objectType that the given user has, possibly, a specific relation with
-// and sends those objects to resultChan. It MUST guarantee no duplicate objects sent.
+// Execute yields all the objects of the provided objectType that the
+// given user possibly has, a specific relation with and sends those
+// objects to resultChan. It MUST guarantee no duplicate objects sent.
 //
-// If an error is encountered before resolving all objects: the provided channel will NOT be closed and
-// - if the error is context cancellation or deadline: Execute may send the error through the channel
-// - otherwise: Execute will send the error through the channel
-// If no errors, Execute will yield all of the objects on the provided channel and then close the channel
-// to signal that it is done.
+// This function respects context timeouts and cancellations. If an
+// error is encountered (e.g. context timeout) before resolving all
+// objects, then the provided channel will NOT be closed, and it will
+// send the error through the channel.
+//
+// If no errors occur, then Execute will yield all of the objects on
+// the provided channel and then close the channel to signal that it
+// is done.
 func (c *ReverseExpandQuery) Execute(
 	ctx context.Context,
 	req *ReverseExpandRequest,
 	resultChan chan<- *ReverseExpandResult,
 	resolutionMetadata *ResolutionMetadata,
-) {
+) error {
 	err := c.execute(ctx, req, resultChan, false, resolutionMetadata)
 	if err != nil {
-		select {
-		case <-ctx.Done():
-			return
-		case resultChan <- &ReverseExpandResult{Err: err}:
-			return
-		}
+		return err
 	}
 
 	close(resultChan)
+	return nil
 }
 
 func (c *ReverseExpandQuery) execute(
 	ctx context.Context,
 	req *ReverseExpandRequest,
diff --git a/pkg/server/commands/reverseexpand/reverse_expand_test.go b/pkg/server/commands/reverseexpand/reverse_expand_test.go
index 797097bf..71c33e10 100644
--- a/pkg/server/commands/reverseexpand/reverse_expand_test.go
+++ b/pkg/server/commands/reverseexpand/reverse_expand_test.go
@@ -14,15 +14,82 @@ import (
 
 	"go.uber.org/goleak"
 
 	"github.com/openfga/openfga/internal/mocks"
 	"github.com/openfga/openfga/pkg/storage"
+	"github.com/openfga/openfga/pkg/storage/memory"
 	"github.com/openfga/openfga/pkg/testutils"
 	"github.com/openfga/openfga/pkg/tuple"
 	"github.com/openfga/openfga/pkg/typesystem"
 )
 
+func TestReverseExpandResultChannelClosed(t *testing.T) {
+	defer goleak.VerifyNone(t)
+
+	store := ulid.Make().String()
+
+	model := testutils.MustTransformDSLToProtoWithID(`model
+  schema 1.1
+type user
+type document
+  relations
+	define viewer: [user]`)
+
+	typeSystem := typesystem.New(model)
+	mockController := gomock.NewController(t)
+	defer mockController.Finish()
+
+	var tuples []*openfgav1.Tuple
+
+	mockDatastore := mocks.NewMockOpenFGADatastore(mockController)
+	mockDatastore.EXPECT().ReadStartingWithUser(gomock.Any(), store, gomock.Any()).
+		Times(1).
+		DoAndReturn(func(_ context.Context, _ string, _ storage.ReadStartingWithUserFilter) (storage.TupleIterator, error) {
+			iterator := storage.NewStaticTupleIterator(tuples)
+			return iterator, nil
+		})
+
+	ctx := context.Background()
+
+	resultChan := make(chan *ReverseExpandResult)
+	errChan := make(chan error, 1)
+
+	// process query in one goroutine, but it will be cancelled almost right away
+	go func() {
+		reverseExpandQuery := NewReverseExpandQuery(mockDatastore, typeSystem)
+		t.Logf("before execute reverse expand")
+		err := reverseExpandQuery.Execute(ctx, &ReverseExpandRequest{
+			StoreID:    store,
+			ObjectType: "document",
+			Relation:   "viewer",
+			User: &UserRefObject{
+				Object: &openfgav1.Object{
+					Type: "user",
+					Id:   "maria",
+				},
+			},
+			ContextualTuples: []*openfgav1.TupleKey{},
+		}, resultChan, NewResolutionMetadata())
+		t.Logf("after execute reverse expand")
+
+		if err != nil {
+			errChan <- err
+		}
+	}()
+
+	select {
+	case _, open := <-resultChan:
+		if open {
+			require.FailNow(t, "expected immediate closure of result channel")
+		}
+	case err := <-errChan:
+		require.FailNow(t, "unexpected error received on error channel :%v", err)
+	case <-time.After(30 * time.Millisecond):
+		require.FailNow(t, "unexpected timeout on channel receive, expected receive on error channel")
+	}
+}
+
 func TestReverseExpandRespectsContextCancellation(t *testing.T) {
 	defer goleak.VerifyNone(t)
 
 	store := ulid.Make().String()
 
@@ -53,18 +120,17 @@ type document
 			return iterator, nil
 		})
 	ctx, cancelFunc := context.WithCancel(context.Background())
 
 	resultChan := make(chan *ReverseExpandResult)
-
-	done := make(chan struct{})
+	errChan := make(chan error, 1)
 
 	// process query in one goroutine, but it will be cancelled almost right away
 	go func() {
 		reverseExpandQuery := NewReverseExpandQuery(mockDatastore, typeSystem)
 		t.Logf("before execute reverse expand")
-		reverseExpandQuery.Execute(ctx, &ReverseExpandRequest{
+		err := reverseExpandQuery.Execute(ctx, &ReverseExpandRequest{
 			StoreID:    store,
 			ObjectType: "document",
 			Relation:   "viewer",
 			User: &UserRefObject{
 				Object: &openfgav1.Object{
@@ -73,29 +139,30 @@ type document
 				},
 			},
 			ContextualTuples: []*openfgav1.TupleKey{},
 		}, resultChan, NewResolutionMetadata())
 		t.Logf("after execute reverse expand")
-		done <- struct{}{}
+
+		if err != nil {
+			errChan <- err
+		}
 	}()
 	go func() {
 		// simulate max_results=1
 		t.Logf("before receive one result")
 		res := <-resultChan
 		t.Logf("after receive one result")
 		cancelFunc()
 		t.Logf("after send cancellation")
 		require.NotNil(t, res.Object)
-		require.NoError(t, res.Err)
 	}()
 
 	select {
-	case <-done:
-		t.Log("OK!")
-		return
+	case err := <-errChan:
+		require.Error(t, err)
 	case <-time.After(30 * time.Millisecond):
-		require.FailNow(t, "timed out")
+		require.FailNow(t, "unexpected timeout on channel receive, expected receive on error channel")
 	}
 }
 
 func TestReverseExpandRespectsContextTimeout(t *testing.T) {
 	defer goleak.VerifyNone(t)
@@ -119,15 +186,15 @@ type document
 		MaxTimes(2) // we expect it to be 0 most of the time
 
 	timeoutCtx, cancel := context.WithTimeout(context.Background(), time.Nanosecond)
 	defer cancel()
 	resultChan := make(chan *ReverseExpandResult)
-	done := make(chan struct{})
+	errChan := make(chan error, 1)
 
 	go func() {
 		reverseExpandQuery := NewReverseExpandQuery(mockDatastore, typeSystem)
-		reverseExpandQuery.Execute(timeoutCtx, &ReverseExpandRequest{
+		err := reverseExpandQuery.Execute(timeoutCtx, &ReverseExpandRequest{
 			StoreID:    store,
 			ObjectType: "document",
 			Relation:   "viewer",
 			User: &UserRefObject{
 				Object: &openfgav1.Object{
@@ -135,22 +202,24 @@ type document
 					Id:   "maria",
 				},
 			},
 			ContextualTuples: []*openfgav1.TupleKey{},
 		}, resultChan, NewResolutionMetadata())
-		done <- struct{}{}
+
+		if err != nil {
+			errChan <- err
+		}
 	}()
 	select {
-	case res, open := <-resultChan:
-		if open {
-			require.Error(t, res.Err)
-		} else {
-			require.Nil(t, res)
+	case _, open := <-resultChan:
+		if !open {
+			require.FailNow(t, "unexpected closure of result channel")
 		}
-		<-done
-	case <-done:
-		// OK!
+	case err := <-errChan:
+		require.Error(t, err)
+	case <-time.After(1 * time.Second):
+		require.FailNow(t, "unexpected timeout encountered, expected other receive")
 	}
 }
 
 func TestReverseExpandErrorInTuples(t *testing.T) {
 	defer goleak.VerifyNone(t)
@@ -178,20 +247,21 @@ type document
 	mockDatastore.EXPECT().ReadStartingWithUser(gomock.Any(), store, gomock.Any()).
 		DoAndReturn(func(_ context.Context, _ string, _ storage.ReadStartingWithUserFilter) (storage.TupleIterator, error) {
 			iterator := mocks.NewErrorTupleIterator(tuples)
 			return iterator, nil
 		})
+
 	ctx, cancelFunc := context.WithCancel(context.Background())
+	defer cancelFunc()
 
 	resultChan := make(chan *ReverseExpandResult)
-
-	done := make(chan struct{})
+	errChan := make(chan error, 1)
 
 	// process query in one goroutine, but it will be cancelled almost right away
 	go func() {
 		reverseExpandQuery := NewReverseExpandQuery(mockDatastore, typeSystem)
-		reverseExpandQuery.Execute(ctx, &ReverseExpandRequest{
+		err := reverseExpandQuery.Execute(ctx, &ReverseExpandRequest{
 			StoreID:    store,
 			ObjectType: "document",
 			Relation:   "viewer",
 			User: &UserRefObject{
 				Object: &openfgav1.Object{
@@ -199,22 +269,86 @@ type document
 					Id:   "maria",
 				},
 			},
 			ContextualTuples: []*openfgav1.TupleKey{},
 		}, resultChan, NewResolutionMetadata())
-		done <- struct{}{}
+		if err != nil {
+			errChan <- err
+		}
 	}()
 
-	go func() {
-		<-resultChan
-		// We want to read resultChan twice because Next() will fail after first read
-		<-resultChan
-		cancelFunc()
-	}()
+ConsumerLoop:
+	for {
+		select {
+		case _, open := <-resultChan:
+			if !open {
+				require.FailNow(t, "unexpected closure of result channel")
+			}
+
+			cancelFunc()
+		case err := <-errChan:
+			require.Error(t, err)
+			break ConsumerLoop
+		case <-time.After(30 * time.Millisecond):
+			require.FailNow(t, "unexpected timeout waiting for channel receive, expected an error on the error channel")
+		}
+	}
+}
 
-	select {
-	case <-done:
-		return
-	case <-time.After(30 * time.Millisecond):
-		require.FailNow(t, "timed out")
+func TestReverseExpandSendsAllErrorsThroughChannel(t *testing.T) {
+	defer goleak.VerifyNone(t)
+
+	store := ulid.Make().String()
+
+	model := testutils.MustTransformDSLToProtoWithID(`model
+  schema 1.1
+type user
+type document
+  relations
+    define viewer: [user]`)
+
+	mockDatastore := mocks.NewMockSlowDataStorage(memory.New(), 1*time.Second)
+
+	for i := 0; i < 50; i++ {
+		t.Logf("iteration %d", i)
+		ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(1*time.Nanosecond))
+		t.Cleanup(func() {
+			cancel()
+		})
+
+		resultChan := make(chan *ReverseExpandResult)
+		errChan := make(chan error, 1)
+
+		go func() {
+			reverseExpandQuery := NewReverseExpandQuery(mockDatastore, typesystem.New(model))
+			t.Logf("before produce")
+			err := reverseExpandQuery.Execute(ctx, &ReverseExpandRequest{
+				StoreID:    store,
+				ObjectType: "document",
+				Relation:   "viewer",
+				User: &UserRefObject{
+					Object: &openfgav1.Object{
+						Type: "user",
+						Id:   "maria",
+					},
+				},
+				ContextualTuples: []*openfgav1.TupleKey{},
+			}, resultChan, NewResolutionMetadata())
+			t.Logf("after produce")
+
+			if err != nil {
+				errChan <- err
+			}
+		}()
+
+		select {
+		case _, channelOpen := <-resultChan:
+			if !channelOpen {
+				require.FailNow(t, "unexpected closure of result channel")
+			}
+		case err := <-errChan:
+			require.Error(t, err)
+		case <-time.After(3 * time.Second):
+			require.FailNow(t, "unexpected timeout waiting for channel receive, expected an error on the error channel")
+		}
 	}
 }
diff --git a/pkg/server/server_test.go b/pkg/server/server_test.go
index a4c6bf4f..5b3bd51c 100644
--- a/pkg/server/server_test.go
+++ b/pkg/server/server_test.go
@@ -14,10 +14,11 @@ import (
 
 	"github.com/oklog/ulid/v2"
 	openfgav1 "github.com/openfga/api/proto/openfga/v1"
 	parser "github.com/openfga/language/pkg/go/transformer"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/goleak"
 	"go.uber.org/mock/gomock"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 
@@ -94,21 +95,21 @@ func TestServerPanicIfEmptyRequestDurationDatastoreCountBuckets(t *testing.T) {
 
 func TestServerWithPostgresDatastore(t *testing.T) {
 	ds, stopFunc := MustBootstrapDatastore(t, "postgres")
 	defer func() {
 		stopFunc()
-		//goleak.VerifyNone(t)
+		goleak.VerifyNone(t)
 	}()
 
 	test.RunAllTests(t, ds)
 }
 
 func TestServerWithPostgresDatastoreAndExplicitCredentials(t *testing.T) {
 	testDatastore, stopFunc := storagefixtures.RunDatastoreTestContainer(t, "postgres")
 	defer func() {
 		stopFunc()
-		//goleak.VerifyNone(t)
+		goleak.VerifyNone(t)
 	}()
 
 	uri := testDatastore.GetConnectionURI(false)
 	ds, err := postgres.New(
 		uri,
@@ -125,31 +126,31 @@ func TestServerWithPostgresDatastoreAndExplicitCredentials(t *testing.T) {
 
 func TestServerWithMemoryDatastore(t *testing.T) {
 	ds, stopFunc := MustBootstrapDatastore(t, "memory")
 	defer func() {
 		stopFunc()
-		//goleak.VerifyNone(t)
+		goleak.VerifyNone(t)
 	}()
 
 	test.RunAllTests(t, ds)
 }
 
 func TestServerWithMySQLDatastore(t *testing.T) {
 	ds, stopFunc := MustBootstrapDatastore(t, "mysql")
 	defer func() {
 		stopFunc()
-		//goleak.VerifyNone(t)
+		goleak.VerifyNone(t)
 	}()
 
 	test.RunAllTests(t, ds)
 }
 
 func TestServerWithMySQLDatastoreAndExplicitCredentials(t *testing.T) {
 	testDatastore, stopFunc := storagefixtures.RunDatastoreTestContainer(t, "mysql")
 	defer func() {
 		stopFunc()
-		//goleak.VerifyNone(t)
+		goleak.VerifyNone(t)
 	}()
 
 	uri := testDatastore.GetConnectionURI(false)
 	ds, err := mysql.New(
 		uri,
diff --git a/pkg/server/test/list_objects.go b/pkg/server/test/list_objects.go
index 5612ae17..658dd5e5 100644
--- a/pkg/server/test/list_objects.go
+++ b/pkg/server/test/list_objects.go
@@ -518,15 +518,26 @@ condition condition1(x: int) {
 				}
 
 				done := make(chan struct{})
 				var streamedObjectIds []string
 				go func() {
-					for x := range server.channel {
-						streamedObjectIds = append(streamedObjectIds, x)
+					for {
+						select {
+						case objectID, open := <-server.channel:
+							if !open {
+								done <- struct{}{}
+								return
+							}
+
+							streamedObjectIds = append(streamedObjectIds, objectID)
+
+						// for tests whose deadline is sooner than the latency of the storage layer
+						case <-time.After(test.readTuplesDelay + 1*time.Second):
+							done <- struct{}{}
+							return
+						}
 					}
-
-					done <- struct{}{}
 				}()
 
 				_, err := listObjectsQuery.ExecuteStreamed(ctx, &openfgav1.StreamedListObjectsRequest{
 					StoreId:          storeID,
 					Type:             test.objectType,
diff --git a/pkg/server/test/reverse_expand.go b/pkg/server/test/reverse_expand.go
index b9346520..0a110e4e 100644
--- a/pkg/server/test/reverse_expand.go
+++ b/pkg/server/test/reverse_expand.go
@@ -1213,22 +1213,20 @@ type document
 		},
 	}
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			require := require.New(t)
-
 			ctx := context.Background()
 			store := ulid.Make().String()
 			test.request.StoreID = store
 
 			model := testutils.MustTransformDSLToProtoWithID(test.model)
 			err := ds.WriteAuthorizationModel(ctx, store, model)
-			require.NoError(err)
+			require.NoError(t, err)
 
 			err = ds.Write(ctx, store, nil, test.tuples)
-			require.NoError(err)
+			require.NoError(t, err)
 
 			var opts []reverseexpand.ReverseExpandQueryOption
 
 			if test.resolveNodeLimit != 0 {
 				opts = append(opts, reverseexpand.WithResolveNodeLimit(test.resolveNodeLimit))
@@ -1241,38 +1239,40 @@ type document
 			timeoutCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 			defer cancel()
 
 			resolutionMetadata := reverseexpand.NewResolutionMetadata()
 
+			reverseExpandErrCh := make(chan error, 1)
 			go func() {
-				reverseExpandQuery.Execute(timeoutCtx, test.request, resultChan, resolutionMetadata)
+				err := reverseExpandQuery.Execute(timeoutCtx, test.request, resultChan, resolutionMetadata)
+				if err != nil {
+					reverseExpandErrCh <- err
+					t.Logf("sent err %s", err)
+				}
 			}()
 
 			var results []*reverseexpand.ReverseExpandResult
-			reverseExpandErrCh := make(chan error)
-			go func() {
-				for result := range resultChan {
-					if result.Err != nil {
-						reverseExpandErrCh <- result.Err
+
+			for {
+				select {
+				case err := <-reverseExpandErrCh:
+					require.ErrorIs(t, err, test.expectedError)
+					return
+				case res, channelOpen := <-resultChan:
+					if !channelOpen {
+						t.Log("channel closed")
+						if test.expectedError == nil {
+							require.ElementsMatch(t, test.expectedResult, results)
+							require.Equal(t, test.expectedDSQueryCount, *resolutionMetadata.QueryCount)
+						} else {
+							require.FailNow(t, "expected an error, got none")
+						}
 						return
+					} else {
+						t.Logf("appending result %s", res.Object)
+						results = append(results, res)
 					}
-
-					results = append(results, result)
 				}
-
-				reverseExpandErrCh <- nil
-			}()
-
-			select {
-			case <-timeoutCtx.Done():
-				require.FailNow("timed out waiting for response")
-			case err := <-reverseExpandErrCh:
-				require.ErrorIs(err, test.expectedError)
-			}
-
-			if test.expectedError == nil {
-				require.ElementsMatch(test.expectedResult, results)
-				require.Equal(test.expectedDSQueryCount, *resolutionMetadata.QueryCount)
 			}
 		})
 	}
 }
diff --git a/pkg/testfixtures/storage/mysql.go b/pkg/testfixtures/storage/mysql.go
index 555717b9..9cbcf85f 100644
--- a/pkg/testfixtures/storage/mysql.go
+++ b/pkg/testfixtures/storage/mysql.go
@@ -1,10 +1,9 @@
 package storage
 
 import (
 	"context"
-	"database/sql"
 	"fmt"
 	"io"
 	"log"
 	"strings"
 	"testing"
@@ -138,20 +137,18 @@ func (m *mySQLTestContainer) RunMySQLTestContainer(t testing.TB) (DatastoreTestC
 	err = mysql.SetLogger(log.New(io.Discard, "", 0))
 	require.NoError(t, err)
 
 	goose.SetLogger(goose.NopLogger())
 
-	var db *sql.DB
+	db, err := goose.OpenDBWithDriver("mysql", uri)
+	require.NoError(t, err)
+	defer db.Close()
 
 	backoffPolicy := backoff.NewExponentialBackOff()
 	backoffPolicy.MaxElapsedTime = 2 * time.Minute
 	err = backoff.Retry(
 		func() error {
-			db, err = goose.OpenDBWithDriver("mysql", uri)
-			if err != nil {
-				return err
-			}
 			return db.Ping()
 		},
 		backoffPolicy,
 	)
 	if err != nil {
@@ -165,13 +162,10 @@ func (m *mySQLTestContainer) RunMySQLTestContainer(t testing.TB) (DatastoreTestC
 	require.NoError(t, err)
 	version, err := goose.GetDBVersion(db)
 	require.NoError(t, err)
 	mySQLTestContainer.version = version
 
-	err = db.Close()
-	require.NoError(t, err)
-
 	return mySQLTestContainer, stopContainer
 }
 
 // GetConnectionURI returns the mysql connection uri for the running mysql test container.
 func (m *mySQLTestContainer) GetConnectionURI(includeCredentials bool) string {
diff --git a/pkg/testfixtures/storage/postgres.go b/pkg/testfixtures/storage/postgres.go
index 7983f539..28db0a8d 100644
--- a/pkg/testfixtures/storage/postgres.go
+++ b/pkg/testfixtures/storage/postgres.go
@@ -135,10 +135,11 @@ func (p *postgresTestContainer) RunPostgresTestContainer(t testing.TB) (Datastor
 
 	goose.SetLogger(goose.NopLogger())
 
 	db, err := goose.OpenDBWithDriver("pgx", uri)
 	require.NoError(t, err)
+	defer db.Close()
 
 	backoffPolicy := backoff.NewExponentialBackOff()
 	backoffPolicy.MaxElapsedTime = 30 * time.Second
 	err = backoff.Retry(
 		func() error {
@@ -158,13 +159,10 @@ func (p *postgresTestContainer) RunPostgresTestContainer(t testing.TB) (Datastor
 
 	version, err := goose.GetDBVersion(db)
 	require.NoError(t, err)
 	pgTestContainer.version = version
 
-	err = db.Close()
-	require.NoError(t, err)
-
 	return pgTestContainer, stopContainer
 }
 
 // GetConnectionURI returns the postgres connection uri for the running postgres test container.
 func (p *postgresTestContainer) GetConnectionURI(includeCredentials bool) string {
diff --git a/tests/listobjects/listobjects_test.go b/tests/listobjects/listobjects_test.go
index e43fd0bd..a12d1602 100644
--- a/tests/listobjects/listobjects_test.go
+++ b/tests/listobjects/listobjects_test.go
@@ -3,10 +3,11 @@ package listobjects
 import (
 	"testing"
 
 	openfgav1 "github.com/openfga/api/proto/openfga/v1"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/goleak"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
 
 	"github.com/openfga/openfga/cmd/run"
 	"github.com/openfga/openfga/tests"
@@ -23,12 +24,11 @@ func TestListObjectsPostgres(t *testing.T) {
 func TestListObjectsMySQL(t *testing.T) {
 	testRunAll(t, "mysql")
 }
 
 func testRunAll(t *testing.T, engine string) {
-	// uncomment in https://github.com/openfga/openfga/pull/1315
-	// defer goleak.VerifyNone(t)
+	defer goleak.VerifyNone(t)
 	cfg := run.MustDefaultConfigWithRandomPorts()
 	cfg.Log.Level = "error"
 	cfg.Datastore.Engine = engine
 
 	cancel := tests.StartServer(t, cfg)

Vulnerable code: 		reverseExpandQuery := reverseexpand.NewReverseExpandQuery(q.datastore, typesys,
			reverseExpandQuery.Execute(cancelCtx, &reverseexpand.ReverseExpandRequest{
		for res := range reverseExpandResultsChan {
			if res.Err != nil {
				err := res.Err

				if errors.Is(err, graph.ErrResolutionDepthExceeded) || errors.Is(err, graph.ErrCycleDetected) {
					err = serverErrors.AuthorizationModelResolutionTooComplex
				resultsChan <- ListObjectsResult{Err: err}
				break
			}

			if !(maxResults == 0) && objectsFound.Load() >= maxResults {
				break
			}
			if res.ResultStatus == reverseexpand.NoFurtherEvalStatus {
				noFurtherEvalRequiredCounter.Inc()
				trySendObject(res.Object, &objectsFound, maxResults, resultsChan)
				continue
			}
			furtherEvalRequiredCounter.Inc()

			wg.Add(1)
			go func(res *reverseexpand.ReverseExpandResult) {
				defer func() {
					<-concurrencyLimiterCh
					wg.Done()
				}()

				concurrencyLimiterCh <- struct{}{}

				resp, err := checkResolver.ResolveCheck(ctx, &graph.ResolveCheckRequest{
					StoreID:              req.GetStoreId(),
					AuthorizationModelID: req.GetAuthorizationModelId(),
					TupleKey:             tuple.NewTupleKey(res.Object, req.GetRelation(), req.GetUser()),
					ContextualTuples:     req.GetContextualTuples().GetTupleKeys(),
					Context:              req.GetContext(),
					ResolutionMetadata: &graph.ResolutionMetadata{
						Depth: q.resolveNodeLimit,
					},
				})
				if err != nil {
					if errors.Is(err, graph.ErrResolutionDepthExceeded) || errors.Is(err, graph.ErrCycleDetected) {
						resultsChan <- ListObjectsResult{Err: serverErrors.AuthorizationModelResolutionTooComplex}
					resultsChan <- ListObjectsResult{Err: err}
					return
				}
				atomic.AddUint32(resolutionMetadata.QueryCount, resp.GetResolutionMetadata().DatastoreQueryCount)
				if resp.Allowed {
					trySendObject(res.Object, &objectsFound, maxResults, resultsChan)
			}(res)
	for {
		select {
		case <-timeoutCtx.Done():
			q.logger.WarnWithContext(
				ctx, fmt.Sprintf("list objects timeout after %s", q.listObjectsDeadline.String()),
			)
			return &ListObjectsResponse{
				Objects:            objects,
				ResolutionMetadata: *resolutionMetadata,
			}, nil

		case result, channelOpen := <-resultsChan:
			if result.Err != nil {
				if errors.Is(result.Err, serverErrors.AuthorizationModelResolutionTooComplex) {
					return nil, result.Err
				}
				if errors.Is(result.Err, condition.ErrEvaluationFailed) {
					errs = multierror.Append(errs, result.Err)
					continue
				}

				if errors.Is(result.Err, context.Canceled) || errors.Is(result.Err, context.DeadlineExceeded) {
					continue
				}

				return nil, serverErrors.HandleError("", result.Err)
			if !channelOpen {
				if len(objects) < int(maxResults) && errs.ErrorOrNil() != nil {
					return nil, errs
				}
				return &ListObjectsResponse{
					Objects:            objects,
					ResolutionMetadata: *resolutionMetadata,
				}, nil
			objects = append(objects, result.ObjectID)
	for {
		select {
		case <-timeoutCtx.Done():
			q.logger.WarnWithContext(
				ctx, fmt.Sprintf("list objects timeout after %s", q.listObjectsDeadline.String()),
			)
			return resolutionMetadata, nil

		case result, channelOpen := <-resultsChan:
			if !channelOpen {
				// Channel closed! No more results.
				return resolutionMetadata, nil
			if result.Err != nil {
				if errors.Is(result.Err, serverErrors.AuthorizationModelResolutionTooComplex) {
					return nil, result.Err
				}

				if errors.Is(result.Err, condition.ErrEvaluationFailed) {
					return nil, serverErrors.ValidationError(result.Err)
				}

				return nil, serverErrors.HandleError("", result.Err)
			if err := srv.Send(&openfgav1.StreamedListObjectsResponse{
				Object: result.ObjectID,
			}); err != nil {
				return nil, serverErrors.NewInternalError("", err)
			}
	Err          error
// Execute yields all the objects of the provided objectType that the given user has, possibly, a specific relation with
// and sends those objects to resultChan. It MUST guarantee no duplicate objects sent.
// If an error is encountered before resolving all objects: the provided channel will NOT be closed and
// - if the error is context cancellation or deadline: Execute may send the error through the channel
// - otherwise: Execute will send the error through the channel
// If no errors, Execute will yield all of the objects on the provided channel and then close the channel
// to signal that it is done.
) {
		select {
		case <-ctx.Done():
			return
		case resultChan <- &ReverseExpandResult{Err: err}:
			return
		}

	done := make(chan struct{})
		reverseExpandQuery.Execute(ctx, &ReverseExpandRequest{
		done <- struct{}{}
		require.NoError(t, res.Err)
	case <-done:
		t.Log("OK!")
		return
		require.FailNow(t, "timed out")
	done := make(chan struct{})
		reverseExpandQuery.Execute(timeoutCtx, &ReverseExpandRequest{
		done <- struct{}{}
	case res, open := <-resultChan:
		if open {
			require.Error(t, res.Err)
		} else {
			require.Nil(t, res)
		<-done
	case <-done:
		// OK!

	done := make(chan struct{})
		reverseExpandQuery.Execute(ctx, &ReverseExpandRequest{
		done <- struct{}{}
	go func() {
		<-resultChan
		// We want to read resultChan twice because Next() will fail after first read
		<-resultChan
		cancelFunc()
	}()
	select {
	case <-done:
		return
	case <-time.After(30 * time.Millisecond):
		require.FailNow(t, "timed out")
		//goleak.VerifyNone(t)
		//goleak.VerifyNone(t)
		//goleak.VerifyNone(t)
		//goleak.VerifyNone(t)
		//goleak.VerifyNone(t)
					for x := range server.channel {
						streamedObjectIds = append(streamedObjectIds, x)

					done <- struct{}{}
			require := require.New(t)

			require.NoError(err)
			require.NoError(err)
				reverseExpandQuery.Execute(timeoutCtx, test.request, resultChan, resolutionMetadata)
			reverseExpandErrCh := make(chan error)
			go func() {
				for result := range resultChan {
					if result.Err != nil {
						reverseExpandErrCh <- result.Err

					results = append(results, result)

				reverseExpandErrCh <- nil
			}()

			select {
			case <-timeoutCtx.Done():
				require.FailNow("timed out waiting for response")
			case err := <-reverseExpandErrCh:
				require.ErrorIs(err, test.expectedError)
			}

			if test.expectedError == nil {
				require.ElementsMatch(test.expectedResult, results)
				require.Equal(test.expectedDSQueryCount, *resolutionMetadata.QueryCount)
	"database/sql"
	var db *sql.DB
			db, err = goose.OpenDBWithDriver("mysql", uri)
			if err != nil {
				return err
			}
	err = db.Close()
	require.NoError(t, err)

	err = db.Close()
	require.NoError(t, err)

	// uncomment in https://github.com/openfga/openfga/pull/1315
	// defer goleak.VerifyNone(t)
Fixed code: // evaluate fires of evaluation of the ListObjects query by delegating to
// [[reverseexpand.ReverseExpand#Execute]] and resolving the results yielded
// from it. If any results yielded by reverse expansion require further eval,
// then these results get dispatched to Check to resolve the residual outcome.
//
// The resultsChan is **always** closed by evaluate when it is done with its work,
// which is either when all results have been yielded, the deadline has been met,
// or some other terminal error case has occurred.
		reverseExpandQuery := reverseexpand.NewReverseExpandQuery(
			q.datastore,
			typesys,
		errChan := make(chan error, 1)

			err := reverseExpandQuery.Execute(cancelCtx, &reverseexpand.ReverseExpandRequest{
			if err != nil {
				errChan <- err
			}
	ConsumerReadLoop:
		for {
			select {
			case <-ctx.Done():
				break ConsumerReadLoop
			case res, channelOpen := <-reverseExpandResultsChan:
				if !channelOpen {
					break ConsumerReadLoop
				if !(maxResults == 0) && objectsFound.Load() >= maxResults {
					break ConsumerReadLoop
				}
				if res.ResultStatus == reverseexpand.NoFurtherEvalStatus {
					noFurtherEvalRequiredCounter.Inc()
					trySendObject(res.Object, &objectsFound, maxResults, resultsChan)
					continue
				}
				furtherEvalRequiredCounter.Inc()

				wg.Add(1)
				go func(res *reverseexpand.ReverseExpandResult) {
					defer func() {
						<-concurrencyLimiterCh
						wg.Done()
					}()

					concurrencyLimiterCh <- struct{}{}

					resp, err := checkResolver.ResolveCheck(ctx, &graph.ResolveCheckRequest{
						StoreID:              req.GetStoreId(),
						AuthorizationModelID: req.GetAuthorizationModelId(),
						TupleKey:             tuple.NewTupleKey(res.Object, req.GetRelation(), req.GetUser()),
						ContextualTuples:     req.GetContextualTuples().GetTupleKeys(),
						Context:              req.GetContext(),
						ResolutionMetadata: &graph.ResolutionMetadata{
							Depth: q.resolveNodeLimit,
						},
					})
					if err != nil {
						if errors.Is(err, graph.ErrResolutionDepthExceeded) || errors.Is(err, graph.ErrCycleDetected) {
							resultsChan <- ListObjectsResult{Err: serverErrors.AuthorizationModelResolutionTooComplex}
							return
						}

						resultsChan <- ListObjectsResult{Err: err}
					atomic.AddUint32(resolutionMetadata.QueryCount, resp.GetResolutionMetadata().DatastoreQueryCount)
					if resp.Allowed {
						trySendObject(res.Object, &objectsFound, maxResults, resultsChan)
					}
				}(res)
			case err := <-errChan:
				if errors.Is(err, graph.ErrResolutionDepthExceeded) || errors.Is(err, graph.ErrCycleDetected) {
					err = serverErrors.AuthorizationModelResolutionTooComplex

				resultsChan <- ListObjectsResult{Err: err}
				break ConsumerReadLoop
			}
	for result := range resultsChan {
		if result.Err != nil {
			if errors.Is(result.Err, serverErrors.AuthorizationModelResolutionTooComplex) {
				return nil, result.Err
			if errors.Is(result.Err, condition.ErrEvaluationFailed) {
				errs = multierror.Append(errs, result.Err)
				continue
			}
			if errors.Is(result.Err, context.Canceled) || errors.Is(result.Err, context.DeadlineExceeded) {
				continue

			return nil, serverErrors.HandleError("", result.Err)

		objects = append(objects, result.ObjectID)

	if len(objects) < int(maxResults) && errs.ErrorOrNil() != nil {
		return nil, errs
	}

	return &ListObjectsResponse{
		Objects:            objects,
		ResolutionMetadata: *resolutionMetadata,
	}, nil
	for result := range resultsChan {
		if result.Err != nil {
			if errors.Is(result.Err, serverErrors.AuthorizationModelResolutionTooComplex) {
				return nil, result.Err
			if errors.Is(result.Err, condition.ErrEvaluationFailed) {
				return nil, serverErrors.ValidationError(result.Err)
			return nil, serverErrors.HandleError("", result.Err)
		}

		if err := srv.Send(&openfgav1.StreamedListObjectsResponse{
			Object: result.ObjectID,
		}); err != nil {
			return nil, serverErrors.NewInternalError("", err)

	return resolutionMetadata, nil
// Execute yields all the objects of the provided objectType that the
// given user possibly has, a specific relation with and sends those
// objects to resultChan. It MUST guarantee no duplicate objects sent.
// This function respects context timeouts and cancellations. If an
// error is encountered (e.g. context timeout) before resolving all
// objects, then the provided channel will NOT be closed, and it will
// send the error through the channel.
//
// If no errors occur, then Execute will yield all of the objects on
// the provided channel and then close the channel to signal that it
// is done.
) error {
		return err
	return nil
	"github.com/openfga/openfga/pkg/storage/memory"
func TestReverseExpandResultChannelClosed(t *testing.T) {
	defer goleak.VerifyNone(t)

	store := ulid.Make().String()

	model := testutils.MustTransformDSLToProtoWithID(`model
  schema 1.1
type user
type document
  relations
	define viewer: [user]`)

	typeSystem := typesystem.New(model)
	mockController := gomock.NewController(t)
	defer mockController.Finish()

	var tuples []*openfgav1.Tuple

	mockDatastore := mocks.NewMockOpenFGADatastore(mockController)
	mockDatastore.EXPECT().ReadStartingWithUser(gomock.Any(), store, gomock.Any()).
		Times(1).
		DoAndReturn(func(_ context.Context, _ string, _ storage.ReadStartingWithUserFilter) (storage.TupleIterator, error) {
			iterator := storage.NewStaticTupleIterator(tuples)
			return iterator, nil
		})

	ctx := context.Background()

	resultChan := make(chan *ReverseExpandResult)
	errChan := make(chan error, 1)

	// process query in one goroutine, but it will be cancelled almost right away
	go func() {
		reverseExpandQuery := NewReverseExpandQuery(mockDatastore, typeSystem)
		t.Logf("before execute reverse expand")
		err := reverseExpandQuery.Execute(ctx, &ReverseExpandRequest{
			StoreID:    store,
			ObjectType: "document",
			Relation:   "viewer",
			User: &UserRefObject{
				Object: &openfgav1.Object{
					Type: "user",
					Id:   "maria",
				},
			},
			ContextualTuples: []*openfgav1.TupleKey{},
		}, resultChan, NewResolutionMetadata())
		t.Logf("after execute reverse expand")

		if err != nil {
			errChan <- err
		}
	}()

	select {
	case _, open := <-resultChan:
		if open {
			require.FailNow(t, "expected immediate closure of result channel")
		}
	case err := <-errChan:
		require.FailNow(t, "unexpected error received on error channel :%v", err)
	case <-time.After(30 * time.Millisecond):
		require.FailNow(t, "unexpected timeout on channel receive, expected receive on error channel")
	}
}

	errChan := make(chan error, 1)
		err := reverseExpandQuery.Execute(ctx, &ReverseExpandRequest{

		if err != nil {
			errChan <- err
		}
	case err := <-errChan:
		require.Error(t, err)
		require.FailNow(t, "unexpected timeout on channel receive, expected receive on error channel")
	errChan := make(chan error, 1)
		err := reverseExpandQuery.Execute(timeoutCtx, &ReverseExpandRequest{

		if err != nil {
			errChan <- err
		}
	case _, open := <-resultChan:
		if !open {
			require.FailNow(t, "unexpected closure of result channel")
	case err := <-errChan:
		require.Error(t, err)
	case <-time.After(1 * time.Second):
		require.FailNow(t, "unexpected timeout encountered, expected other receive")

	defer cancelFunc()
	errChan := make(chan error, 1)
		err := reverseExpandQuery.Execute(ctx, &ReverseExpandRequest{
		if err != nil {
			errChan <- err
		}
ConsumerLoop:
	for {
		select {
		case _, open := <-resultChan:
			if !open {
				require.FailNow(t, "unexpected closure of result channel")
			}

			cancelFunc()
		case err := <-errChan:
			require.Error(t, err)
			break ConsumerLoop
		case <-time.After(30 * time.Millisecond):
			require.FailNow(t, "unexpected timeout waiting for channel receive, expected an error on the error channel")
		}
	}
}
func TestReverseExpandSendsAllErrorsThroughChannel(t *testing.T) {
	defer goleak.VerifyNone(t)

	store := ulid.Make().String()

	model := testutils.MustTransformDSLToProtoWithID(`model
  schema 1.1
type user
type document
  relations
    define viewer: [user]`)

	mockDatastore := mocks.NewMockSlowDataStorage(memory.New(), 1*time.Second)

	for i := 0; i < 50; i++ {
		t.Logf("iteration %d", i)
		ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(1*time.Nanosecond))
		t.Cleanup(func() {
			cancel()
		})

		resultChan := make(chan *ReverseExpandResult)
		errChan := make(chan error, 1)

		go func() {
			reverseExpandQuery := NewReverseExpandQuery(mockDatastore, typesystem.New(model))
			t.Logf("before produce")
			err := reverseExpandQuery.Execute(ctx, &ReverseExpandRequest{
				StoreID:    store,
				ObjectType: "document",
				Relation:   "viewer",
				User: &UserRefObject{
					Object: &openfgav1.Object{
						Type: "user",
						Id:   "maria",
					},
				},
				ContextualTuples: []*openfgav1.TupleKey{},
			}, resultChan, NewResolutionMetadata())
			t.Logf("after produce")

			if err != nil {
				errChan <- err
			}
		}()

		select {
		case _, channelOpen := <-resultChan:
			if !channelOpen {
				require.FailNow(t, "unexpected closure of result channel")
			}
		case err := <-errChan:
			require.Error(t, err)
		case <-time.After(3 * time.Second):
			require.FailNow(t, "unexpected timeout waiting for channel receive, expected an error on the error channel")
		}
	"go.uber.org/goleak"
		goleak.VerifyNone(t)
		goleak.VerifyNone(t)
		goleak.VerifyNone(t)
		goleak.VerifyNone(t)
		goleak.VerifyNone(t)
					for {
						select {
						case objectID, open := <-server.channel:
							if !open {
								done <- struct{}{}
								return
							}

							streamedObjectIds = append(streamedObjectIds, objectID)

						// for tests whose deadline is sooner than the latency of the storage layer
						case <-time.After(test.readTuplesDelay + 1*time.Second):
							done <- struct{}{}
							return
						}
			require.NoError(t, err)
			require.NoError(t, err)
			reverseExpandErrCh := make(chan error, 1)
				err := reverseExpandQuery.Execute(timeoutCtx, test.request, resultChan, resolutionMetadata)
				if err != nil {
					reverseExpandErrCh <- err
					t.Logf("sent err %s", err)
				}

			for {
				select {
				case err := <-reverseExpandErrCh:
					require.ErrorIs(t, err, test.expectedError)
					return
				case res, channelOpen := <-resultChan:
					if !channelOpen {
						t.Log("channel closed")
						if test.expectedError == nil {
							require.ElementsMatch(t, test.expectedResult, results)
							require.Equal(t, test.expectedDSQueryCount, *resolutionMetadata.QueryCount)
						} else {
							require.FailNow(t, "expected an error, got none")
						}
					} else {
						t.Logf("appending result %s", res.Object)
						results = append(results, res)
	db, err := goose.OpenDBWithDriver("mysql", uri)
	require.NoError(t, err)
	defer db.Close()
	defer db.Close()
	"go.uber.org/goleak"
	defer goleak.VerifyNone(t)