OpenFGA DoS

@@ -148,6 +148,14 @@ type listObjectsRequest interface {
 	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,
@@ -218,7 +226,9 @@ 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),
@@ -228,18 +238,23 @@ func (q *ListObjectsQuery) evaluate(
 
 		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(
@@ -250,64 +265,71 @@ func (q *ListObjectsQuery) evaluate(
 
 		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()
@@ -358,48 +380,36 @@ 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.
@@ -424,37 +434,25 @@ func (q *ListObjectsQuery) ExecuteStreamed(ctx context.Context, req *openfgav1.S
 		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
 }

@@ -161,7 +161,6 @@ const (
 )
 
 type ReverseExpandResult struct {
-	Err          error
 	Object       string
 	ResultStatus ConditionalResultStatus
 }
@@ -182,31 +181,31 @@ func WithLogger(logger logger.Logger) ReverseExpandQueryOption {
 	}
 }
 
-// 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(

@@ -16,11 +16,78 @@ import (
 
 	"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)
 
@@ -55,14 +122,13 @@ type document
 	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",
@@ -75,7 +141,10 @@ 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
@@ -85,15 +154,13 @@ type document
 		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")
 	}
 }
 
@@ -121,11 +188,11 @@ type document
 	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",
@@ -137,18 +204,20 @@ type document
 			},
 			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")
 	}
 }
 
@@ -180,16 +249,17 @@ type document
 			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",
@@ -201,20 +271,84 @@ type document
 			},
 			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")
+		}
 	}
 }

@@ -16,6 +16,7 @@ import (
 	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"
@@ -96,7 +97,7 @@ func TestServerWithPostgresDatastore(t *testing.T) {
 	ds, stopFunc := MustBootstrapDatastore(t, "postgres")
 	defer func() {
 		stopFunc()
-		//goleak.VerifyNone(t)
+		goleak.VerifyNone(t)
 	}()
 
 	test.RunAllTests(t, ds)
@@ -106,7 +107,7 @@ func TestServerWithPostgresDatastoreAndExplicitCredentials(t *testing.T) {
 	testDatastore, stopFunc := storagefixtures.RunDatastoreTestContainer(t, "postgres")
 	defer func() {
 		stopFunc()
-		//goleak.VerifyNone(t)
+		goleak.VerifyNone(t)
 	}()
 
 	uri := testDatastore.GetConnectionURI(false)
@@ -127,7 +128,7 @@ func TestServerWithMemoryDatastore(t *testing.T) {
 	ds, stopFunc := MustBootstrapDatastore(t, "memory")
 	defer func() {
 		stopFunc()
-		//goleak.VerifyNone(t)
+		goleak.VerifyNone(t)
 	}()
 
 	test.RunAllTests(t, ds)
@@ -137,7 +138,7 @@ func TestServerWithMySQLDatastore(t *testing.T) {
 	ds, stopFunc := MustBootstrapDatastore(t, "mysql")
 	defer func() {
 		stopFunc()
-		//goleak.VerifyNone(t)
+		goleak.VerifyNone(t)
 	}()
 
 	test.RunAllTests(t, ds)
@@ -147,7 +148,7 @@ func TestServerWithMySQLDatastoreAndExplicitCredentials(t *testing.T) {
 	testDatastore, stopFunc := storagefixtures.RunDatastoreTestContainer(t, "mysql")
 	defer func() {
 		stopFunc()
-		//goleak.VerifyNone(t)
+		goleak.VerifyNone(t)
 	}()
 
 	uri := testDatastore.GetConnectionURI(false)

@@ -520,11 +520,22 @@ 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{

@@ -1215,18 +1215,16 @@ 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
 
@@ -1243,35 +1241,37 @@ type document
 
 			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)
 			}
 		})
 	}

@@ -2,7 +2,6 @@ package storage
 
 import (
 	"context"
-	"database/sql"
 	"fmt"
 	"io"
 	"log"
@@ -140,16 +139,14 @@ func (m *mySQLTestContainer) RunMySQLTestContainer(t testing.TB) (DatastoreTestC
 
 	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,
@@ -167,9 +164,6 @@ func (m *mySQLTestContainer) RunMySQLTestContainer(t testing.TB) (DatastoreTestC
 	require.NoError(t, err)
 	mySQLTestContainer.version = version
 
-	err = db.Close()
-	require.NoError(t, err)
-
 	return mySQLTestContainer, stopContainer
 }
 

@@ -137,6 +137,7 @@ func (p *postgresTestContainer) RunPostgresTestContainer(t testing.TB) (Datastor
 
 	db, err := goose.OpenDBWithDriver("pgx", uri)
 	require.NoError(t, err)
+	defer db.Close()
 
 	backoffPolicy := backoff.NewExponentialBackOff()
 	backoffPolicy.MaxElapsedTime = 30 * time.Second
@@ -160,9 +161,6 @@ func (p *postgresTestContainer) RunPostgresTestContainer(t testing.TB) (Datastor
 	require.NoError(t, err)
 	pgTestContainer.version = version
 
-	err = db.Close()
-	require.NoError(t, err)
-
 	return pgTestContainer, stopContainer
 }
 

@@ -5,6 +5,7 @@ import (
 
 	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"
 
@@ -25,8 +26,7 @@ func TestListObjectsMySQL(t *testing.T) {
 }
 
 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