Repository: openfga/openfga

Issue:
Getting deadline_exceeded response and running out of db conns
### Checklist

- [x] I have looked into the [README](https://github.com/openfga/openfga/blob/main/README.md) and have not found a suitable solution or answer.
- [x] I have looked into the [documentation](https://openfga.dev/docs) and have not found a suitable solution or answer.
- [x] I have searched the [issues](https://github.com/openfga/openfga/issues) and have not found a suitable solution or answer.
- [x] I have upgraded to the [latest version](https://github.com/openfga/openfga/releases/latest) of OpenFGA and the issue still persists.
- [x] I have searched the [OpenFGA Community](https://openfga.dev/community) and have not found a suitable solution or answer.
- [x] I agree to the terms within the [OpenFGA Code of Conduct](https://github.com/openfga/.github/blob/main/CODE_OF_CONDUCT.md).

### Description

We've run into a problem where all fga requests fail with deadline_exceeded error response, including even the health check!

We've done some experimentation and we've found the folllowing:

- When this happens the database connections are consumed up to the max-pool size via OPENFGA_DATASTORE_MAX_OPEN_CONNS (in our case, the default 30)
- The db conns never shrink back to the min 10 (OPENFGA_DATASTORE_MAX_IDLE_CONNS) even when idle
- Versions up to 1.8.6 work correctly
- Version 1.8.7 and higher all fail basically immediately within a small number of check requests - 30-40
- On versions up to 1.8.6 the db connections are all cleared/freed up and go back to the min (10) as expected after the requests
- Once it reaches this state the service is broken and no further requests are handled.  Only a restart will fix it.

The actual response we get for every API request after this happens is:

```
{"code":"deadline_exceeded","message":"Request Deadline Exceeded"}
```

We don't have many tuples - only ~35k and the model is relatively simple.  I don't believe it is model or data related as everything works unchanged up to version 1.8.6 without issue.

Not sure where to go from here as the 1.8.7 changelogs don't really give any clues and the git diffs aren't too enlightening for me as I'm not familiar with the code base.

We've tried enabling additional logging with '--log-level=debug' but that doesn't seem to provide any data or any additional data at all.

Any suggestions or pointers to resolving this issue would be appreciated.  For now we're sticking with 1.8.6. as it works.



### OpenFGA version

1.8.7->1.8.11

### How are you running OpenFGA?

In Docker

### What datastore are you using?

Postgres

### OpenFGA Flags

OPENFGA_LOG_FORMAT=json OPENFGA_LOG_LEVEL=debug

### Logs

_No response_

Hints from issue comments:
@adriantam: Can you share what is the configuration on your system?  More specifically, the number of DB connections as well as whether any optimization flag is enabled?@frisket: Certainly.  It's a fairly straightforward test system, nothing special:

- Debian 12.8 (uname output: `Linux apptest 6.1.0-27-amd64 #1 SMP PREEMPT_DYNAMIC Debian 6.1.115-1 (2024-11-01) x86_64 GNU/Linux`)
- Postgres 15.9
- max_connections=500

Note that it's openfga running out of DB connections, not the database.  The database continues to run just fine and respond normally when openfga gets stuck.

Configuration wise, it's basically stock/defaults except for the ones noted to enable json logging and debug logs.

It's quite reproduceable as well - it's happened on 3 separate test instances already with similarly basic configuration.

Here's the docker compose we're using:

```
  openfga:
    depends_on:
      - openfga-migrate
    image: openfga/openfga:v1.8.11
    extra_hosts:
      - "host.docker.internal:host-gateway"
    command: run
    ports:
      - "48080:8080"
      - "48081:8081"
      - "13000:3000"
    environment:
      - OPENFGA_DATASTORE_ENGINE=postgres
      - OPENFGA_DATASTORE_URI=postgres://openfga:openfga@${DB_ADDRESS}:5432/openfga
      - OPENFGA_LOG_FORMAT=json
      - OPENFGA_PLAYGROUND_PORT=13000
      - OPENFGA_LOG_LEVEL=debug

```

Our next steps will be to try playing with the model/tuples and see if that makes any difference.


@adriantam: We will investigate into it.  In the meantime, perhaps you can update the [OPENFGA_DATASTORE_MAX_OPEN_CONNS](https://github.com/openfga/openfga/blob/5fba50f2570d0f755a7cc23654ba4ab876e09b84/.config-schema.json#L237) to match the SQL connections.@frisket: Haven't tried bumping the max conns to 500 yet, but I don't think that will change anything except take down the DB :-)

If it's useful, I've attached a thread dump after it gets stuck.  I enabled the profiler and got the dump from this url: `http://localhost:3001/debug/pprof/goroutine?debug=2`

Note that to reduce the thread/routine count I updated the max-conns to 5 and max-idle-conns to 1.

[fga.dump.gz](https://github.com/user-attachments/files/20068943/fga.dump.gz)

If there are other profiler commands you want me to try out, let me know and I can provide the output.
@adriantam: Thank you @frisket . We have identified the issue and are working on the fix for it.  We will let you know when the fix is ready.@nqvuong1998: Hi @adriantam , In my case (Lakekeeper), after updating OpenFGA to v1.8.12 the deadline_exceeded error still appears. How about @frisket ?@frisket: Just to follow-up - I can confirm we're no longer having any issues with openfga hanging after extensive testing.  @aaguiarz: @nqvuong1998 did you try with 1.8.16? @miketq: @aaguiarz do you get this error on 1.8.16? currently on the latest version and getting this error@the-vincent-y: confirm the problem existing on `1.8.16`. after randomly switching back to old version like `v1.8.12-arm64`, no more `{"code":"deadline_exceeded","message":"Request Deadline Exceeded"}`

And the problem seems the exact same as before, even check health will return deadline_exceeded@jigish: this issue seems to still exist in `1.9.0` -- I can easily trigger it by performing a few batch checks in quick succession. we're running 3 replicas with:

```
maxOpenConns: 250
maxIdleConns: 100
```@aaguiarz: @jigish does it happen in 1.8.6?

@Tom-Bury: Hi all. In our case, we have found similar issues across all versions (tested some between 1.8.5 up to latest 1.9.0). Small store, no load yet since we're not yet live and it only takes a few batch checks (proper ones, not `ClientBatchCheck`) to get into `deadline_exceeded` issues. I've tried to tweak the datastore connection connection configs iso defaults, but could not get any better results that way. Our setup allows us to run the whole stack with similar data locally, using local OpenFGA server with the in memory datastore and that all works like a charm. So I don't think it's our model or store size that is the issue.

I still want to dig a bit into Postgres logs to see whether there's any slow queries or connection issues. But in the short term we have found that enabling OpenFGA caching seems to do the trick, with the caveat of lower consistency ofc. (`OPENFGA_CHECK_ITERATOR_CACHE_ENABLED`, `OPENFGA_CHECK_QUERY_CACHE_ENABLED`, `OPENFGA_CACHE_CONTROLLER_ENABLED` & `OPENFGA_LIST_OBJECTS_ITERATOR_CACHE_ENABLED`)@nqvuong1998: Hi @Tom-Bury @jigish  OpenFGA is now at version 1.10.3. Have you tried using this latest version?@nqvuong1998: Hi @aaguiarz , The error still occurred in version 1.9.0, while it worked well with version 1.8.6.@n14zio: @nqvuong1998 if on k8s, what are cpu/memory specs?@lars-planvera: Still occurring with 1.10.3@aaguiarz: Hi @lars-planvera 

The `deadline_exceeded` error happens when the server time outs, and it can be caused by several issues (configuration, DB Capacity, OpenFGA model/data distribution/queries).

We can help you troubleshoot it but we need some help. We are drafting some notes here about how to report these issues to the team https://github.com/openfga/openfga.dev/pull/1127/files@Minibrams: Was also able to fix this by downgrading from v1.11 to v1.8.6, but unfortunately that version does not support ignoring conflicts on write/delete, which I used quite a bit for idempotency.

Important to note that enabling caching as @Tom-Bury did not do anything for me.

In case it's relevant to anyone, here's the setup I'm using to replicate it.
Here's one of the models I'm using - quite simple, and only 750 tuples, but 20-30 simultaneous requests and the OpenFGA container freezes up. Reproducible on a 10 core Ubuntu box with plenty of memory, so it seems unlikely that it's a hardware limitation. On v1.8.6 it runs without issue.

The database is PostgreSQL. Both running as Docker containers.

The first requests I see freezing are check requests for `(user:<id>, self_ref, user:<id>)`, but that might be a red herring.

Let me know if I can do anything to help work out what the problem is.

Other than this hiccup, OpenFGA has been a breeze to use from the beginning, GREAT work by everyone on the team! 

```
model
  schema 1.1

type user
  relations
    define self_ref: [user]
    define root_scope: [control_plane]
    define workspace_link: [workspace]
    define can_view: self_ref or participant from workspace_link or root_admin from root_scope
    define can_modify: self_ref or root_admin from root_scope
    define can_remove: self_ref or root_admin from root_scope

type control_plane
  relations
    define root_admin: [user]

type workspace
  relations
    define root_scope: [control_plane]
    define maintainer: [user] or root_admin from root_scope
    define collaborator: [user]
    define delegate: [user]
    define observer: [user]
    define asset_link: [asset]

    define participant: maintainer or collaborator or delegate or observer

    define can_view: participant or can_view from asset_link or root_admin from root_scope
    define can_modify: maintainer or root_admin from root_scope
    define can_remove: root_admin from root_scope
    define asset_admin: maintainer or collaborator or delegate or root_admin from root_scope
    define membership_admin: maintainer or root_admin from root_scope

type asset
  relations
    define root_scope: [control_plane]
    define workspace_owner: [workspace]
    define can_modify: [user] or asset_admin from workspace_owner or root_admin from root_scope
    define can_view: [user] or can_modify or can_submit or participant from workspace_owner
    define can_submit: [user] or observer from workspace_owner
```@aaguiarz: Thanks @Minibrams can you share the tuples and 
 a script that reproduces it?@mericano1: I can see this issue was closed but no idea what the solution is ? I am running into a similar issue when calling list-objects..@aaguiarz: Hi @mericano1 

There can be many reasons why this error can happen, and some are based on the model complexity/tuple distribution.

What is the specific issue you see? We'd need more data to be able to help you. This PR describes the data we'd need to help https://github.com/openfga/openfga.dev/pull/1127/files

Thanks

Solution patch:
diff --git a/.config-schema.json b/.config-schema.json
index 1c53b7d4..944330b8 100644
--- a/.config-schema.json
+++ b/.config-schema.json
@@ -71,7 +71,7 @@
         "resolveNodeBreadthLimit": {
             "description": "Defines how many nodes on a given level can be evaluated concurrently in a Check resolution tree.",
             "type": "integer",
-            "default": 100,
+            "default": 10,
             "x-env-variable": "OPENFGA_RESOLVE_NODE_BREADTH_LIMIT"
         },
         "listObjectsDeadline": {
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 531f2880..a156c052 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,9 +7,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 Try to keep listed changes to a concise bulleted list of simple explanations of changes. Aim for the amount of information needed so that readers can understand where they would look in the codebase to investigate the changes' implementation, or where they would look in the documentation to understand how to make use of the change in practice - better yet, link directly to the docs and provide detailed information there. Only elaborate if doing so is required to avoid breaking changes or experimental features from ruining someone's day.
 
 ## [Unreleased]
+### Changed
+- `DefaultResolveNodeBreadthLimit` changed from 100 to 10 in order to reduce connection contention. [#2425](https://github.com/openfga/openfga/pull/2425)
+- PostgreSQL and MySQL based iterators will load tuples only when needed (lazy loading). [#2425](https://github.com/openfga/openfga/pull/2425)
+
 ### Fixed
 - Replace hardcoded Prometheus datasource UID (`PBFA97CFB590B2093`) with `${DS_PROMETHEUS}` in `telemetry/grafana/dashboards/openfga.json`. This allows the Grafana dashboard to correctly reference the dynamic Prometheus datasource, resolving issues with improper binding. [#2287](https://github.com/openfga/openfga/issues/2287)
 - Handle case where iterator is stopped more than once for `current_iterator_cache_count`. [#2409](https://github.com/openfga/openfga/pull/2409)
+- Fix deadlock when number of SQL datastore connections is less than Resolve Max Breadth. [#2425](https://github.com/openfga/openfga/pull/2425)
 
 ## [1.8.11] - 2025-04-29
 [Full changelog](https://github.com/openfga/openfga/compare/v1.8.10...v1.8.11)
diff --git a/internal/graph/check_fast_path.go b/internal/graph/check_fast_path.go
index 09232c35..74554a5f 100644
--- a/internal/graph/check_fast_path.go
+++ b/internal/graph/check_fast_path.go
@@ -15,7 +15,7 @@ import (
 
 	"github.com/openfga/openfga/internal/checkutil"
 	"github.com/openfga/openfga/internal/concurrency"
-	"github.com/openfga/openfga/internal/graph/iterator"
+	"github.com/openfga/openfga/internal/iterator"
 	"github.com/openfga/openfga/internal/validation"
 	"github.com/openfga/openfga/pkg/storage"
 	"github.com/openfga/openfga/pkg/tuple"
@@ -426,14 +426,13 @@ func fastPathRewrite(
 // Right channel is the result set of the Read of ObjectID/Relation that yields the User's ObjectID.
 // Left channel is the result set of ReadStartingWithUser of User/Relation that yields Object's ObjectID.
 // From the perspective of the model, the left hand side of a TTU is the computed relationship being expanded.
-func (c *LocalChecker) resolveFastPath(ctx context.Context, leftChans []chan *iterator.Msg, iter storage.TupleMapper) (*ResolveCheckResponse, error) {
+func (c *LocalChecker) resolveFastPath(ctx context.Context, leftChans *iterator.FanIn, iter storage.TupleMapper) (*ResolveCheckResponse, error) {
 	ctx, span := tracer.Start(ctx, "resolveFastPath", trace.WithAttributes(
-		attribute.Int("sources", len(leftChans)),
 		attribute.Bool("allowed", false),
 	))
 	defer span.End()
 	cancellableCtx, cancel := context.WithCancel(ctx)
-	leftChan := fanInIteratorChannels(cancellableCtx, leftChans, c.concurrencyLimit)
+	leftChan := leftChans.Out()
 	rightChan := streamedLookupUsersetFromIterator(cancellableCtx, iter)
 	rightOpen := true
 	leftOpen := true
@@ -441,10 +440,7 @@ func (c *LocalChecker) resolveFastPath(ctx context.Context, leftChans []chan *it
 	defer func() {
 		cancel()
 		iter.Stop()
-		if !leftOpen {
-			return
-		}
-		go drainIteratorChannel(leftChan)
+		leftChans.Close()
 	}()
 
 	res := &ResolveCheckResponse{
@@ -528,16 +524,15 @@ ConsumerLoop:
 	return res, lastErr
 }
 
-func constructLeftChannels(
+func produceLeftChannels(
 	ctx context.Context,
+	leftChans *iterator.FanIn,
 	req *ResolveCheckRequest,
 	relationReferences []*openfgav1.RelationReference,
 	relationFunc checkutil.V2RelationFunc,
-	concurrencyLimit int,
-) ([]chan *iterator.Msg, error) {
+) {
 	typesys, _ := typesystem.TypesystemFromContext(ctx)
-
-	leftChans := make([]chan *iterator.Msg, 0, len(relationReferences))
+	defer leftChans.Done()
 	for _, parentType := range relationReferences {
 		relation := relationFunc(parentType)
 		rel, err := typesys.GetRelation(parentType.GetType(), relation)
@@ -553,15 +548,14 @@ func constructLeftChannels(
 		}
 		leftChan, err := fastPathRewrite(ctx, r, rel.GetRewrite())
 		if err != nil {
-			// if the resolver already started it needs to be drained
-			if len(leftChans) > 0 {
-				go drainIteratorChannel(fanInIteratorChannels(ctx, leftChans, concurrencyLimit))
-			}
-			return nil, err
+			errCh := make(chan *iterator.Msg, 1)
+			errCh <- &iterator.Msg{Err: err}
+			close(errCh)
+			leftChans.Add(errCh)
+			continue
 		}
-		leftChans = append(leftChans, leftChan)
+		leftChans.Add(leftChan)
 	}
-	return leftChans, nil
 }
 
 func (c *LocalChecker) checkUsersetFastPathV2(ctx context.Context, req *ResolveCheckRequest, iter storage.TupleKeyIterator) (*ResolveCheckResponse, error) {
@@ -571,20 +565,10 @@ func (c *LocalChecker) checkUsersetFastPathV2(ctx context.Context, req *ResolveC
 	typesys, _ := typesystem.TypesystemFromContext(ctx)
 	objectType := tuple.GetType(req.GetTupleKey().GetObject())
 
-	cancellableCtx, cancel := context.WithCancel(ctx)
-	defer cancel()
 	directlyRelatedUsersetTypes, _ := typesys.DirectlyRelatedUsersets(objectType, req.GetTupleKey().GetRelation())
 
-	leftChans, err := constructLeftChannels(cancellableCtx, req, directlyRelatedUsersetTypes, checkutil.BuildUsersetV2RelationFunc(), c.concurrencyLimit)
-	if err != nil {
-		return nil, err
-	}
-
-	if len(leftChans) == 0 {
-		return &ResolveCheckResponse{
-			Allowed: false,
-		}, nil
-	}
+	leftChans := iterator.NewFanIn(ctx, c.concurrencyLimit)
+	go produceLeftChannels(ctx, leftChans, req, directlyRelatedUsersetTypes, checkutil.BuildUsersetV2RelationFunc())
 
 	return c.resolveFastPath(ctx, leftChans, storage.WrapIterator(storage.UsersetKind, iter))
 }
@@ -602,68 +586,12 @@ func (c *LocalChecker) checkTTUFastPathV2(ctx context.Context, req *ResolveCheck
 		return nil, err
 	}
 
-	cancellableCtx, cancel := context.WithCancel(ctx)
-	defer cancel()
-
-	leftChans, err := constructLeftChannels(cancellableCtx, req, possibleParents, checkutil.BuildTTUV2RelationFunc(computedRelation), c.concurrencyLimit)
-	if err != nil {
-		return nil, err
-	}
-
-	if len(leftChans) == 0 {
-		return &ResolveCheckResponse{
-			Allowed: false,
-		}, nil
-	}
+	leftChans := iterator.NewFanIn(ctx, c.concurrencyLimit)
+	go produceLeftChannels(ctx, leftChans, req, possibleParents, checkutil.BuildTTUV2RelationFunc(computedRelation))
 
 	return c.resolveFastPath(ctx, leftChans, storage.WrapIterator(storage.TTUKind, iter))
 }
 
-// NOTE: Can we make this generic and move it to concurrency pkg?
-func fanInIteratorChannels(ctx context.Context, chans []chan *iterator.Msg, concurrencyLimit int) chan *iterator.Msg {
-	limit := len(chans)
-
-	out := make(chan *iterator.Msg, limit)
-
-	if limit == 0 {
-		close(out)
-		return out
-	}
-
-	// It is ok if the limit is < concurrencyLimit because that
-	// is the pool's max limit.
-	pool := concurrency.NewPool(ctx, concurrencyLimit)
-
-	for _, c := range chans {
-		pool.Go(func(ctx context.Context) error {
-			for v := range c {
-				if !concurrency.TrySendThroughChannel(ctx, v, out) {
-					if v.Iter != nil {
-						v.Iter.Stop()
-					}
-				}
-			}
-			return nil
-		})
-	}
-
-	go func() {
-		// NOTE: the consumer of this channel will block waiting for it to close
-		_ = pool.Wait()
-		close(out)
-	}()
-
-	return out
-}
-
-func drainIteratorChannel(c chan *iterator.Msg) {
-	for msg := range c {
-		if msg.Iter != nil {
-			msg.Iter.Stop()
-		}
-	}
-}
-
 // Note that visited does not necessary means that there are cycles.  For the following model,
 // type user
 // type group
@@ -692,37 +620,29 @@ func (c *LocalChecker) breadthFirstRecursiveMatch(ctx context.Context, req *Reso
 	relation := req.GetTupleKey().GetRelation()
 	user := req.GetTupleKey().GetUser()
 
-	chans := make([]chan *iterator.Msg, 0, currentUsersetLevel.Size())
-	for _, usersetInterface := range currentUsersetLevel.Values() {
-		userset := usersetInterface.(string)
-		_, visited := visitedUserset.LoadOrStore(userset, struct{}{})
-		if visited {
-			continue
+	leftChans := iterator.NewFanIn(ctx, c.concurrencyLimit)
+	defer leftChans.Close()
+	// allow both producer and consumers to run concurrently
+	go func(req *ResolveCheckRequest) {
+		defer leftChans.Done()
+		for _, usersetInterface := range currentUsersetLevel.Values() {
+			userset := usersetInterface.(string)
+			_, visited := visitedUserset.LoadOrStore(userset, struct{}{})
+			if visited {
+				continue
+			}
+			newReq := req.clone()
+			newReq.TupleKey = tuple.NewTupleKey(userset, relation, user)
+			mapper, err := buildRecursiveMapper(ctx, newReq, mapping)
+			c := make(chan *iterator.Msg, 1)
+			c <- &iterator.Msg{Iter: mapper, Err: err}
+			close(c)
+			leftChans.Add(c)
 		}
-		newReq := req.clone()
-		newReq.TupleKey = tuple.NewTupleKey(userset, relation, user)
-		mapper, err := buildRecursiveMapper(ctx, newReq, mapping)
+	}(req.clone())
 
-		if err != nil {
-			concurrency.TrySendThroughChannel(ctx, checkOutcome{err: err}, checkOutcomeChan)
-			// TODO: TBD if we hard exit here, if yes, the channel needs to be closed
-			continue
-		}
-		c := make(chan *iterator.Msg, 1)
-		if !concurrency.TrySendThroughChannel(ctx, &iterator.Msg{Iter: mapper}, c) {
-			mapper.Stop() // will not be received to be cleaned up
-		}
-		close(c)
-		chans = append(chans, c)
-	}
-	leftChan := fanInIteratorChannels(ctx, chans, c.concurrencyLimit)
-	leftOpen := true
-	defer func() {
-		if leftOpen {
-			go drainIteratorChannel(leftChan)
-		}
-	}()
 	nextUsersetLevel := hashset.New()
+	out := leftChans.Out()
 
 ConsumerLoop:
 	for {
@@ -730,10 +650,9 @@ ConsumerLoop:
 		case <-ctx.Done():
 			close(checkOutcomeChan)
 			return
-		case msg, ok := <-leftChan:
+		case msg, ok := <-out:
 			if !ok {
 				// nothing was found in this level, break to proceed into the next one
-				leftOpen = false
 				break ConsumerLoop
 			}
 			for {
@@ -749,6 +668,7 @@ ConsumerLoop:
 				}
 
 				if usersetFromUser.Contains(t) {
+					msg.Iter.Stop()
 					concurrency.TrySendThroughChannel(ctx, checkOutcome{resp: &ResolveCheckResponse{Allowed: true}}, checkOutcomeChan)
 					close(checkOutcomeChan)
 					return
diff --git a/internal/graph/object_providers.go b/internal/graph/object_providers.go
index 73f1dd4e..b1b5e101 100644
--- a/internal/graph/object_providers.go
+++ b/internal/graph/object_providers.go
@@ -9,7 +9,7 @@ import (
 
 	"github.com/openfga/openfga/internal/checkutil"
 	"github.com/openfga/openfga/internal/concurrency"
-	"github.com/openfga/openfga/internal/graph/iterator"
+	"github.com/openfga/openfga/internal/iterator"
 	"github.com/openfga/openfga/pkg/storage"
 	"github.com/openfga/openfga/pkg/tuple"
 	"github.com/openfga/openfga/pkg/typesystem"
@@ -92,17 +92,15 @@ func (c *recursiveTTUObjectProvider) Begin(ctx context.Context, req *ResolveChec
 		return nil, err
 	}
 
-	leftChannels, err := constructLeftChannels(ctx, req, possibleParents, checkutil.BuildTTUV2RelationFunc(c.computedRelation), c.concurrencyLimit)
-	if err != nil {
-		return nil, err
-	}
-	outChannel := make(chan usersetMessage, len(leftChannels))
-	leftChannel := fanInIteratorChannels(ctx, leftChannels, c.concurrencyLimit)
+	leftChans := iterator.NewFanIn(ctx, c.concurrencyLimit)
+	go produceLeftChannels(ctx, leftChans, req, possibleParents, checkutil.BuildTTUV2RelationFunc(c.computedRelation))
+
+	outChannel := make(chan usersetMessage, c.concurrencyLimit)
 	poolCtx, cancel := context.WithCancel(ctx)
 	c.cancel = cancel
 
 	c.pool = concurrency.NewPool(poolCtx, 1)
-	c.pool.Go(iteratorToUserset(leftChannel, outChannel))
+	c.pool.Go(iteratorToUserset(leftChans, outChannel))
 
 	return outChannel, nil
 }
@@ -132,36 +130,30 @@ func (c *recursiveUsersetObjectProvider) End() {
 func (c *recursiveUsersetObjectProvider) Begin(ctx context.Context, req *ResolveCheckRequest) (<-chan usersetMessage, error) {
 	objectType := tuple.GetType(req.GetTupleKey().GetObject())
 	reference := []*openfgav1.RelationReference{{Type: objectType, RelationOrWildcard: &openfgav1.RelationReference_Relation{Relation: req.GetTupleKey().GetRelation()}}}
-	leftChans, err := constructLeftChannels(ctx, req, reference, checkutil.BuildUsersetV2RelationFunc(), c.concurrencyLimit)
-	if err != nil {
-		return nil, err
-	}
-	outChannel := make(chan usersetMessage, len(leftChans))
-	leftChannel := fanInIteratorChannels(ctx, leftChans, c.concurrencyLimit)
+
+	leftChans := iterator.NewFanIn(ctx, c.concurrencyLimit)
+	go produceLeftChannels(ctx, leftChans, req, reference, checkutil.BuildUsersetV2RelationFunc())
+
+	outChannel := make(chan usersetMessage, c.concurrencyLimit)
 	poolCtx, cancel := context.WithCancel(ctx)
 	c.cancel = cancel
 	c.pool = concurrency.NewPool(poolCtx, 1)
-	c.pool.Go(iteratorToUserset(leftChannel, outChannel))
+	c.pool.Go(iteratorToUserset(leftChans, outChannel))
 
 	return outChannel, nil
 }
 
-func iteratorToUserset(src chan *iterator.Msg, dst chan usersetMessage) func(ctx context.Context) error {
+func iteratorToUserset(src *iterator.FanIn, dst chan usersetMessage) func(ctx context.Context) error {
 	return func(ctx context.Context) error {
-		leftOpen := true
 		defer func() {
+			src.Close()
 			close(dst)
-			if !leftOpen {
-				return
-			}
-			go drainIteratorChannel(src)
 		}()
 		select {
 		case <-ctx.Done():
 			return ctx.Err()
-		case msg, ok := <-src:
+		case msg, ok := <-src.Out():
 			if !ok {
-				leftOpen = false
 				return nil
 			}
 			if msg.Err != nil {
diff --git a/internal/iterator/fan_in.go b/internal/iterator/fan_in.go
new file mode 100644
index 00000000..214b1bf3
--- /dev/null
+++ b/internal/iterator/fan_in.go
@@ -0,0 +1,129 @@
+package iterator
+
+import (
+	"context"
+	"sync"
+
+	"github.com/sourcegraph/conc/pool"
+
+	"github.com/openfga/openfga/internal/concurrency"
+)
+
+type FanIn struct {
+	ctx       context.Context
+	cancel    context.CancelFunc
+	count     int
+	out       chan *Msg
+	addCh     chan (chan *Msg)
+	mu        sync.Mutex
+	accepting bool
+	pool      *pool.ContextPool
+	drained   chan bool
+}
+
+func NewFanIn(ctx context.Context, limit int) *FanIn {
+	ctx, cancel := context.WithCancel(ctx)
+
+	f := &FanIn{
+		ctx:       ctx,
+		cancel:    cancel,
+		count:     0,
+		out:       make(chan *Msg, limit),
+		addCh:     make(chan (chan *Msg), limit),
+		accepting: true,
+		mu:        sync.Mutex{},
+		pool:      concurrency.NewPool(ctx, limit),
+		drained:   make(chan bool, 1),
+	}
+
+	go f.run()
+	return f
+}
+
+func (f *FanIn) run() {
+	defer func() {
+		f.Done()
+		_ = f.pool.Wait()
+		close(f.out)
+		for ch := range f.addCh {
+			drainOnExit(ch)
+		}
+		f.drained <- true
+		close(f.drained)
+	}()
+	for {
+		select {
+		case <-f.ctx.Done():
+			return
+		case ch, ok := <-f.addCh:
+			if !ok {
+				return
+			}
+			f.pool.Go(func(ctx context.Context) error {
+				defer drainOnExit(ch)
+				for {
+					select {
+					case <-ctx.Done():
+						return nil
+					case v, ok := <-ch:
+						if !ok {
+							return nil
+						}
+						if !concurrency.TrySendThroughChannel(ctx, v, f.out) {
+							if v.Iter != nil {
+								v.Iter.Stop()
+							}
+						}
+					}
+				}
+			})
+		}
+	}
+}
+
+func drainOnExit(ch chan *Msg) {
+	for msg := range ch {
+		if msg.Iter != nil {
+			msg.Iter.Stop()
+		}
+	}
+}
+
+func (f *FanIn) Add(ch chan *Msg) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.count++
+	if !f.accepting {
+		drainOnExit(ch)
+		return
+	}
+	if !concurrency.TrySendThroughChannel(f.ctx, ch, f.addCh) {
+		drainOnExit(ch)
+		return
+	}
+}
+
+func (f *FanIn) Out() chan *Msg {
+	return f.out
+}
+
+func (f *FanIn) Count() int {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return f.count
+}
+
+func (f *FanIn) Done() {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	if f.accepting {
+		f.accepting = false
+		close(f.addCh)
+	}
+}
+
+func (f *FanIn) Close() {
+	// Done gets called internally
+	f.cancel()
+	drainOnExit(f.out)
+}
diff --git a/internal/graph/iterator/iterator_stream.go b/internal/iterator/stream.go
similarity index 100%
rename from internal/graph/iterator/iterator_stream.go
rename to internal/iterator/stream.go
diff --git a/internal/mocks/mock_cache.go b/internal/mocks/mock_cache.go
index b79f00ff..b151ef00 100644
--- a/internal/mocks/mock_cache.go
+++ b/internal/mocks/mock_cache.go
@@ -16,6 +16,44 @@ import (
 	gomock "go.uber.org/mock/gomock"
 )
 
+// MockCacheItem is a mock of CacheItem interface.
+type MockCacheItem struct {
+	ctrl     *gomock.Controller
+	recorder *MockCacheItemMockRecorder
+	isgomock struct{}
+}
+
+// MockCacheItemMockRecorder is the mock recorder for MockCacheItem.
+type MockCacheItemMockRecorder struct {
+	mock *MockCacheItem
+}
+
+// NewMockCacheItem creates a new mock instance.
+func NewMockCacheItem(ctrl *gomock.Controller) *MockCacheItem {
+	mock := &MockCacheItem{ctrl: ctrl}
+	mock.recorder = &MockCacheItemMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use.
+func (m *MockCacheItem) EXPECT() *MockCacheItemMockRecorder {
+	return m.recorder
+}
+
+// CacheEntityType mocks base method.
+func (m *MockCacheItem) CacheEntityType() string {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "CacheEntityType")
+	ret0, _ := ret[0].(string)
+	return ret0
+}
+
+// CacheEntityType indicates an expected call of CacheEntityType.
+func (mr *MockCacheItemMockRecorder) CacheEntityType() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CacheEntityType", reflect.TypeOf((*MockCacheItem)(nil).CacheEntityType))
+}
+
 // MockInMemoryCache is a mock of InMemoryCache interface.
 type MockInMemoryCache[T any] struct {
 	ctrl     *gomock.Controller
diff --git a/pkg/server/config/config.go b/pkg/server/config/config.go
index fb16a84e..6f9ff6f5 100644
--- a/pkg/server/config/config.go
+++ b/pkg/server/config/config.go
@@ -22,7 +22,7 @@ const (
 	DefaultMaxAuthorizationModelCacheSize   = 100000
 	DefaultChangelogHorizonOffset           = 0
 	DefaultResolveNodeLimit                 = 25
-	DefaultResolveNodeBreadthLimit          = 100
+	DefaultResolveNodeBreadthLimit          = 10
 	DefaultUsersetBatchSize                 = 1000
 	DefaultListObjectsDeadline              = 3 * time.Second
 	DefaultListObjectsMaxResults            = 1000
diff --git a/pkg/storage/mysql/mysql.go b/pkg/storage/mysql/mysql.go
index 986a0fcd..0d117fa0 100644
--- a/pkg/storage/mysql/mysql.go
+++ b/pkg/storage/mysql/mysql.go
@@ -116,7 +116,7 @@ func NewWithDB(db *sql.DB, cfg *sqlcommon.Config) (*Datastore, error) {
 	}
 
 	stbl := sq.StatementBuilder.RunWith(db)
-	dbInfo := sqlcommon.NewDBInfo(db, stbl, HandleSQLError, "mysql")
+	dbInfo := sqlcommon.NewDBInfo(db, stbl, sqlcommon.HandleSQLError, "mysql")
 
 	return &Datastore{
 		stbl:                   stbl,
@@ -161,11 +161,11 @@ func (s *Datastore) ReadPage(ctx context.Context, store string, tupleKey *openfg
 	}
 	defer iter.Stop()
 
-	return iter.ToArray(options.Pagination)
+	return iter.ToArray(ctx, options.Pagination)
 }
 
 func (s *Datastore) read(ctx context.Context, store string, tupleKey *openfgav1.TupleKey, options *storage.ReadPageOptions) (*sqlcommon.SQLTupleIterator, error) {
-	ctx, span := startTrace(ctx, "read")
+	_, span := startTrace(ctx, "read")
 	defer span.End()
 
 	sb := s.stbl.
@@ -201,12 +201,7 @@ func (s *Datastore) read(ctx context.Context, store string, tupleKey *openfgav1.
 		sb = sb.Limit(uint64(options.Pagination.PageSize + 1)) // + 1 is used to determine whether to return a continuation token.
 	}
 
-	rows, err := sb.QueryContext(ctx)
-	if err != nil {
-		return nil, HandleSQLError(err)
-	}
-
-	return sqlcommon.NewSQLTupleIterator(rows), nil
+	return sqlcommon.NewSQLTupleIterator(sb), nil
 }
 
 // Write see [storage.RelationshipTupleWriter].Write.
@@ -259,7 +254,7 @@ func (s *Datastore) ReadUserTuple(ctx context.Context, store string, tupleKey *o
 			&conditionContext,
 		)
 	if err != nil {
-		return nil, HandleSQLError(err)
+		return nil, sqlcommon.HandleSQLError(err)
 	}
 
 	if conditionName.String != "" {
@@ -284,7 +279,7 @@ func (s *Datastore) ReadUsersetTuples(
 	filter storage.ReadUsersetTuplesFilter,
 	_ storage.ReadUsersetTuplesOptions,
 ) (storage.TupleIterator, error) {
-	ctx, span := startTrace(ctx, "ReadUsersetTuples")
+	_, span := startTrace(ctx, "ReadUsersetTuples")
 	defer span.End()
 
 	sb := s.stbl.
@@ -323,12 +318,8 @@ func (s *Datastore) ReadUsersetTuples(
 		}
 		sb = sb.Where(orConditions)
 	}
-	rows, err := sb.QueryContext(ctx)
-	if err != nil {
-		return nil, HandleSQLError(err)
-	}
 
-	return sqlcommon.NewSQLTupleIterator(rows), nil
+	return sqlcommon.NewSQLTupleIterator(sb), nil
 }
 
 // ReadStartingWithUser see [storage.RelationshipTupleReader].ReadStartingWithUser.
@@ -338,7 +329,7 @@ func (s *Datastore) ReadStartingWithUser(
 	filter storage.ReadStartingWithUserFilter,
 	_ storage.ReadStartingWithUserOptions,
 ) (storage.TupleIterator, error) {
-	ctx, span := startTrace(ctx, "ReadStartingWithUser")
+	_, span := startTrace(ctx, "ReadStartingWithUser")
 	defer span.End()
 
 	var targetUsersArg []string
@@ -368,12 +359,7 @@ func (s *Datastore) ReadStartingWithUser(
 		builder = builder.Where(sq.Eq{"object_id": filter.ObjectIDs.Values()})
 	}
 
-	rows, err := builder.QueryContext(ctx)
-	if err != nil {
-		return nil, HandleSQLError(err)
-	}
-
-	return sqlcommon.NewSQLTupleIterator(rows), nil
+	return sqlcommon.NewSQLTupleIterator(builder), nil
 }
 
 // MaxTuplesPerWrite see [storage.RelationshipTupleWriter].MaxTuplesPerWrite.
@@ -411,7 +397,7 @@ func (s *Datastore) ReadAuthorizationModels(ctx context.Context, store string, o
 
 	rows, err := sb.QueryContext(ctx)
 	if err != nil {
-		return nil, "", HandleSQLError(err)
+		return nil, "", sqlcommon.HandleSQLError(err)
 	}
 	defer rows.Close()
 
@@ -421,14 +407,14 @@ func (s *Datastore) ReadAuthorizationModels(ctx context.Context, store string, o
 	for rows.Next() {
 		err = rows.Scan(&modelID)
 		if err != nil {
-			return nil, "", HandleSQLError(err)
+			return nil, "", sqlcommon.HandleSQLError(err)
 		}
 
 		modelIDs = append(modelIDs, modelID)
 	}
 
 	if err := rows.Err(); err != nil {
-		return nil, "", HandleSQLError(err)
+		return nil, "", sqlcommon.HandleSQLError(err)
 	}
 
 	var token string
@@ -484,7 +470,7 @@ func (s *Datastore) CreateStore(ctx context.Context, store *openfgav1.Store) (*o
 
 	txn, err := s.db.BeginTx(ctx, &sql.TxOptions{})
 	if err != nil {
-		return nil, HandleSQLError(err)
+		return nil, sqlcommon.HandleSQLError(err)
 	}
 	defer func() {
 		_ = txn.Rollback()
@@ -497,7 +483,7 @@ func (s *Datastore) CreateStore(ctx context.Context, store *openfgav1.Store) (*o
 		RunWith(txn).
 		ExecContext(ctx)
 	if err != nil {
-		return nil, HandleSQLError(err)
+		return nil, sqlcommon.HandleSQLError(err)
 	}
 
 	err = s.stbl.
@@ -508,12 +494,12 @@ func (s *Datastore) CreateStore(ctx context.Context, store *openfgav1.Store) (*o
 		QueryRowContext(ctx).
 		Scan(&id, &name, &createdAt, &updatedAt)
 	if err != nil {
-		return nil, HandleSQLError(err)
+		return nil, sqlcommon.HandleSQLError(err)
 	}
 
 	err = txn.Commit()
 	if err != nil {
-		return nil, HandleSQLError(err)
+		return nil, sqlcommon.HandleSQLError(err)
 	}
 
 	return &openfgav1.Store{
@@ -545,7 +531,7 @@ func (s *Datastore) GetStore(ctx context.Context, id string) (*openfgav1.Store,
 		if errors.Is(err, sql.ErrNoRows) {
 			return nil, storage.ErrNotFound
 		}
-		return nil, HandleSQLError(err)
+		return nil, sqlcommon.HandleSQLError(err)
 	}
 
 	return &openfgav1.Store{
@@ -589,7 +575,7 @@ func (s *Datastore) ListStores(ctx context.Context, options storage.ListStoresOp
 
 	rows, err := sb.QueryContext(ctx)
 	if err != nil {
-		return nil, "", HandleSQLError(err)
+		return nil, "", sqlcommon.HandleSQLError(err)
 	}
 	defer rows.Close()
 
@@ -600,7 +586,7 @@ func (s *Datastore) ListStores(ctx context.Context, options storage.ListStoresOp
 		var createdAt, updatedAt time.Time
 		err := rows.Scan(&id, &name, &createdAt, &updatedAt)
 		if err != nil {
-			return nil, "", HandleSQLError(err)
+			return nil, "", sqlcommon.HandleSQLError(err)
 		}
 
 		stores = append(stores, &openfgav1.Store{
@@ -612,7 +598,7 @@ func (s *Datastore) ListStores(ctx context.Context, options storage.ListStoresOp
 	}
 
 	if err := rows.Err(); err != nil {
-		return nil, "", HandleSQLError(err)
+		return nil, "", sqlcommon.HandleSQLError(err)
 	}
 
 	if len(stores) > options.Pagination.PageSize {
@@ -633,7 +619,7 @@ func (s *Datastore) DeleteStore(ctx context.Context, id string) error {
 		Where(sq.Eq{"id": id}).
 		ExecContext(ctx)
 	if err != nil {
-		return HandleSQLError(err)
+		return sqlcommon.HandleSQLError(err)
 	}
 
 	return nil
@@ -656,7 +642,7 @@ func (s *Datastore) WriteAssertions(ctx context.Context, store, modelID string,
 		Suffix("ON DUPLICATE KEY UPDATE assertions = ?", marshalledAssertions).
 		ExecContext(ctx)
 	if err != nil {
-		return HandleSQLError(err)
+		return sqlcommon.HandleSQLError(err)
 	}
 
 	return nil
@@ -681,7 +667,7 @@ func (s *Datastore) ReadAssertions(ctx context.Context, store, modelID string) (
 		if errors.Is(err, sql.ErrNoRows) {
 			return []*openfgav1.Assertion{}, nil
 		}
-		return nil, HandleSQLError(err)
+		return nil, sqlcommon.HandleSQLError(err)
 	}
 
 	var assertions openfgav1.Assertions
@@ -730,7 +716,7 @@ func (s *Datastore) ReadChanges(ctx context.Context, store string, filter storag
 
 	rows, err := sb.QueryContext(ctx)
 	if err != nil {
-		return nil, "", HandleSQLError(err)
+		return nil, "", sqlcommon.HandleSQLError(err)
 	}
 	defer rows.Close()
 
@@ -755,7 +741,7 @@ func (s *Datastore) ReadChanges(ctx context.Context, store string, filter storag
 			&insertedAt,
 		)
 		if err != nil {
-			return nil, "", HandleSQLError(err)
+			return nil, "", sqlcommon.HandleSQLError(err)
 		}
 
 		var conditionContextStruct structpb.Struct
@@ -793,22 +779,3 @@ func (s *Datastore) ReadChanges(ctx context.Context, store string, filter storag
 func (s *Datastore) IsReady(ctx context.Context) (storage.ReadinessStatus, error) {
 	return sqlcommon.IsReady(ctx, s.db)
 }
-
-// HandleSQLError processes an SQL error and converts it into a more
-// specific error type based on the nature of the SQL error.
-func HandleSQLError(err error, args ...interface{}) error {
-	if errors.Is(err, sql.ErrNoRows) {
-		return storage.ErrNotFound
-	}
-
-	if me, ok := err.(*mysql.MySQLError); ok && me.Number == 1062 {
-		if len(args) > 0 {
-			if tk, ok := args[0].(*openfgav1.TupleKey); ok {
-				return storage.InvalidWriteInputError(tk, openfgav1.TupleOperation_TUPLE_OPERATION_WRITE)
-			}
-		}
-		return storage.ErrCollision
-	}
-
-	return fmt.Errorf("sql error: %w", err)
-}
diff --git a/pkg/storage/postgres/postgres.go b/pkg/storage/postgres/postgres.go
index 9e8635ac..268a288a 100644
--- a/pkg/storage/postgres/postgres.go
+++ b/pkg/storage/postgres/postgres.go
@@ -130,7 +130,7 @@ func NewWithDB(db *sql.DB, cfg *sqlcommon.Config) (*Datastore, error) {
 	}
 
 	stbl := sq.StatementBuilder.PlaceholderFormat(sq.Dollar).RunWith(db)
-	dbInfo := sqlcommon.NewDBInfo(db, stbl, HandleSQLError, "postgres")
+	dbInfo := sqlcommon.NewDBInfo(db, stbl, sqlcommon.HandleSQLError, "postgres")
 
 	return &Datastore{
 		stbl:                   stbl,
@@ -175,11 +175,11 @@ func (s *Datastore) ReadPage(ctx context.Context, store string, tupleKey *openfg
 	}
 	defer iter.Stop()
 
-	return iter.ToArray(options.Pagination)
+	return iter.ToArray(ctx, options.Pagination)
 }
 
 func (s *Datastore) read(ctx context.Context, store string, tupleKey *openfgav1.TupleKey, options *storage.ReadPageOptions) (*sqlcommon.SQLTupleIterator, error) {
-	ctx, span := startTrace(ctx, "read")
+	_, span := startTrace(ctx, "read")
 	defer span.End()
 
 	sb := s.stbl.
@@ -215,12 +215,7 @@ func (s *Datastore) read(ctx context.Context, store string, tupleKey *openfgav1.
 		sb = sb.Limit(uint64(options.Pagination.PageSize + 1)) // + 1 is used to determine whether to return a continuation token.
 	}
 
-	rows, err := sb.QueryContext(ctx)
-	if err != nil {
-		return nil, HandleSQLError(err)
-	}
-
-	return sqlcommon.NewSQLTupleIterator(rows), nil
+	return sqlcommon.NewSQLTupleIterator(sb), nil
 }
 
 // Write see [storage.RelationshipTupleWriter].Write.
@@ -273,7 +268,7 @@ func (s *Datastore) ReadUserTuple(ctx context.Context, store string, tupleKey *o
 			&conditionContext,
 		)
 	if err != nil {
-		return nil, HandleSQLError(err)
+		return nil, sqlcommon.HandleSQLError(err)
 	}
 
 	if conditionName.String != "" {
@@ -298,7 +293,7 @@ func (s *Datastore) ReadUsersetTuples(
 	filter storage.ReadUsersetTuplesFilter,
 	_ storage.ReadUsersetTuplesOptions,
 ) (storage.TupleIterator, error) {
-	ctx, span := startTrace(ctx, "ReadUsersetTuples")
+	_, span := startTrace(ctx, "ReadUsersetTuples")
 	defer span.End()
 
 	sb := s.stbl.
@@ -337,12 +332,8 @@ func (s *Datastore) ReadUsersetTuples(
 		}
 		sb = sb.Where(orConditions)
 	}
-	rows, err := sb.QueryContext(ctx)
-	if err != nil {
-		return nil, HandleSQLError(err)
-	}
 
-	return sqlcommon.NewSQLTupleIterator(rows), nil
+	return sqlcommon.NewSQLTupleIterator(sb), nil
 }
 
 // ReadStartingWithUser see [storage.RelationshipTupleReader].ReadStartingWithUser.
@@ -352,7 +343,7 @@ func (s *Datastore) ReadStartingWithUser(
 	filter storage.ReadStartingWithUserFilter,
 	_ storage.ReadStartingWithUserOptions,
 ) (storage.TupleIterator, error) {
-	ctx, span := startTrace(ctx, "ReadStartingWithUser")
+	_, span := startTrace(ctx, "ReadStartingWithUser")
 	defer span.End()
 
 	var targetUsersArg []string
@@ -382,12 +373,7 @@ func (s *Datastore) ReadStartingWithUser(
 		builder = builder.Where(sq.Eq{"object_id": filter.ObjectIDs.Values()})
 	}
 
-	rows, err := builder.QueryContext(ctx)
-	if err != nil {
-		return nil, HandleSQLError(err)
-	}
-
-	return sqlcommon.NewSQLTupleIterator(rows), nil
+	return sqlcommon.NewSQLTupleIterator(builder), nil
 }
 
 // MaxTuplesPerWrite see [storage.RelationshipTupleWriter].MaxTuplesPerWrite.
@@ -424,7 +410,7 @@ func (s *Datastore) ReadAuthorizationModels(ctx context.Context, store string, o
 
 	rows, err := sb.QueryContext(ctx)
 	if err != nil {
-		return nil, "", HandleSQLError(err)
+		return nil, "", sqlcommon.HandleSQLError(err)
 	}
 	defer rows.Close()
 
@@ -434,14 +420,14 @@ func (s *Datastore) ReadAuthorizationModels(ctx context.Context, store string, o
 	for rows.Next() {
 		err = rows.Scan(&modelID)
 		if err != nil {
-			return nil, "", HandleSQLError(err)
+			return nil, "", sqlcommon.HandleSQLError(err)
 		}
 
 		modelIDs = append(modelIDs, modelID)
 	}
 
 	if err := rows.Err(); err != nil {
-		return nil, "", HandleSQLError(err)
+		return nil, "", sqlcommon.HandleSQLError(err)
 	}
 
 	var token string
@@ -503,7 +489,7 @@ func (s *Datastore) CreateStore(ctx context.Context, store *openfgav1.Store) (*o
 		QueryRowContext(ctx).
 		Scan(&id, &name, &createdAt, &updatedAt)
 	if err != nil {
-		return nil, HandleSQLError(err)
+		return nil, sqlcommon.HandleSQLError(err)
 	}
 
 	return &openfgav1.Store{
@@ -535,7 +521,7 @@ func (s *Datastore) GetStore(ctx context.Context, id string) (*openfgav1.Store,
 		if errors.Is(err, sql.ErrNoRows) {
 			return nil, storage.ErrNotFound
 		}
-		return nil, HandleSQLError(err)
+		return nil, sqlcommon.HandleSQLError(err)
 	}
 
 	return &openfgav1.Store{
@@ -579,7 +565,7 @@ func (s *Datastore) ListStores(ctx context.Context, options storage.ListStoresOp
 
 	rows, err := sb.QueryContext(ctx)
 	if err != nil {
-		return nil, "", HandleSQLError(err)
+		return nil, "", sqlcommon.HandleSQLError(err)
 	}
 	defer rows.Close()
 
@@ -590,7 +576,7 @@ func (s *Datastore) ListStores(ctx context.Context, options storage.ListStoresOp
 		var createdAt, updatedAt time.Time
 		err := rows.Scan(&id, &name, &createdAt, &updatedAt)
 		if err != nil {
-			return nil, "", HandleSQLError(err)
+			return nil, "", sqlcommon.HandleSQLError(err)
 		}
 
 		stores = append(stores, &openfgav1.Store{
@@ -602,7 +588,7 @@ func (s *Datastore) ListStores(ctx context.Context, options storage.ListStoresOp
 	}
 
 	if err := rows.Err(); err != nil {
-		return nil, "", HandleSQLError(err)
+		return nil, "", sqlcommon.HandleSQLError(err)
 	}
 
 	if len(stores) > options.Pagination.PageSize {
@@ -623,7 +609,7 @@ func (s *Datastore) DeleteStore(ctx context.Context, id string) error {
 		Where(sq.Eq{"id": id}).
 		ExecContext(ctx)
 	if err != nil {
-		return HandleSQLError(err)
+		return sqlcommon.HandleSQLError(err)
 	}
 
 	return nil
@@ -646,7 +632,7 @@ func (s *Datastore) WriteAssertions(ctx context.Context, store, modelID string,
 		Suffix("ON CONFLICT (store, authorization_model_id) DO UPDATE SET assertions = ?", marshalledAssertions).
 		ExecContext(ctx)
 	if err != nil {
-		return HandleSQLError(err)
+		return sqlcommon.HandleSQLError(err)
 	}
 
 	return nil
@@ -671,7 +657,7 @@ func (s *Datastore) ReadAssertions(ctx context.Context, store, modelID string) (
 		if errors.Is(err, sql.ErrNoRows) {
 			return []*openfgav1.Assertion{}, nil
 		}
-		return nil, HandleSQLError(err)
+		return nil, sqlcommon.HandleSQLError(err)
 	}
 
 	var assertions openfgav1.Assertions
@@ -720,7 +706,7 @@ func (s *Datastore) ReadChanges(ctx context.Context, store string, filter storag
 
 	rows, err := sb.QueryContext(ctx)
 	if err != nil {
-		return nil, "", HandleSQLError(err)
+		return nil, "", sqlcommon.HandleSQLError(err)
 	}
 	defer rows.Close()
 
@@ -745,7 +731,7 @@ func (s *Datastore) ReadChanges(ctx context.Context, store string, filter storag
 			&insertedAt,
 		)
 		if err != nil {
-			return nil, "", HandleSQLError(err)
+			return nil, "", sqlcommon.HandleSQLError(err)
 		}
 
 		var conditionContextStruct structpb.Struct
@@ -783,22 +769,3 @@ func (s *Datastore) ReadChanges(ctx context.Context, store string, filter storag
 func (s *Datastore) IsReady(ctx context.Context) (storage.ReadinessStatus, error) {
 	return sqlcommon.IsReady(ctx, s.db)
 }
-
-// HandleSQLError processes an SQL error and converts it into a more
-// specific error type based on the nature of the SQL error.
-func HandleSQLError(err error, args ...interface{}) error {
-	if errors.Is(err, sql.ErrNoRows) {
-		return storage.ErrNotFound
-	}
-
-	if strings.Contains(err.Error(), "duplicate key value") {
-		if len(args) > 0 {
-			if tk, ok := args[0].(*openfgav1.TupleKey); ok {
-				return storage.InvalidWriteInputError(tk, openfgav1.TupleOperation_TUPLE_OPERATION_WRITE)
-			}
-		}
-		return storage.ErrCollision
-	}
-
-	return fmt.Errorf("sql error: %w", err)
-}
diff --git a/pkg/storage/sqlcommon/sqlcommon.go b/pkg/storage/sqlcommon/sqlcommon.go
index 1fda79f7..e6aa9b8c 100644
--- a/pkg/storage/sqlcommon/sqlcommon.go
+++ b/pkg/storage/sqlcommon/sqlcommon.go
@@ -5,13 +5,18 @@ import (
 	"database/sql"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"strconv"
+	"strings"
 	"sync"
 	"time"
 
 	sq "github.com/Masterminds/squirrel"
+	"github.com/go-sql-driver/mysql"
 	"github.com/oklog/ulid/v2"
 	"github.com/pressly/goose/v3"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/trace"
 	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/structpb"
 
@@ -24,6 +29,8 @@ import (
 	tupleUtils "github.com/openfga/openfga/pkg/tuple"
 )
 
+var tracer = otel.Tracer("pkg/storage/sqlcommon")
+
 // Config defines the configuration parameters
 // for setting up and managing a sql connection.
 type Config struct {
@@ -188,6 +195,7 @@ func (s *SQLContinuationTokenSerializer) Deserialize(continuationToken string) (
 // interface for iterating over tuples fetched from a SQL database.
 type SQLTupleIterator struct {
 	rows *sql.Rows // GUARDED_BY(mu)
+	sb   sq.SelectBuilder
 
 	// firstRow is used as a temporary storage place if head is called.
 	// If firstRow is nil and Head is called, rows.Next() will return the first item and advance
@@ -201,17 +209,36 @@ type SQLTupleIterator struct {
 var _ storage.TupleIterator = (*SQLTupleIterator)(nil)
 
 // NewSQLTupleIterator returns a SQL tuple iterator.
-func NewSQLTupleIterator(rows *sql.Rows) *SQLTupleIterator {
+func NewSQLTupleIterator(sb sq.SelectBuilder) *SQLTupleIterator {
 	return &SQLTupleIterator{
-		rows:     rows,
+		sb:       sb,
+		rows:     nil,
 		firstRow: nil,
 		mu:       sync.Mutex{},
 	}
 }
 
-func (t *SQLTupleIterator) next() (*storage.TupleRecord, error) {
+func (t *SQLTupleIterator) fetchBuffer(ctx context.Context) error {
+	ctx, span := tracer.Start(ctx, "sqlcommon.fetchBuffer", trace.WithAttributes())
+	defer span.End()
+	rows, err := t.sb.QueryContext(ctx)
+	if err != nil {
+		return HandleSQLError(err)
+	}
+	t.rows = rows
+	return nil
+}
+
+func (t *SQLTupleIterator) next(ctx context.Context) (*storage.TupleRecord, error) {
 	t.mu.Lock()
 
+	if t.rows == nil {
+		if err := t.fetchBuffer(ctx); err != nil {
+			t.mu.Unlock()
+			return nil, err
+		}
+	}
+
 	if t.firstRow != nil {
 		// If head was called previously, we don't need to scan / next
 		// again as the data is already there and the internal iterator would be advanced via `t.rows.Next()`.
@@ -269,10 +296,16 @@ func (t *SQLTupleIterator) next() (*storage.TupleRecord, error) {
 	return &record, nil
 }
 
-func (t *SQLTupleIterator) head() (*storage.TupleRecord, error) {
+func (t *SQLTupleIterator) head(ctx context.Context) (*storage.TupleRecord, error) {
 	t.mu.Lock()
 	defer t.mu.Unlock()
 
+	if t.rows == nil {
+		if err := t.fetchBuffer(ctx); err != nil {
+			return nil, err
+		}
+	}
+
 	if t.firstRow != nil {
 		// If head was called previously, we don't need to scan / next
 		// again as the data is already there and the internal iterator would be advanced via `t.rows.Next()`.
@@ -328,12 +361,12 @@ func (t *SQLTupleIterator) head() (*storage.TupleRecord, error) {
 
 // ToArray converts the tupleIterator to an []*openfgav1.Tuple and a possibly empty continuation token.
 // If the continuation token exists it is the ulid of the last element of the returned array.
-func (t *SQLTupleIterator) ToArray(
+func (t *SQLTupleIterator) ToArray(ctx context.Context,
 	opts storage.PaginationOptions,
 ) ([]*openfgav1.Tuple, string, error) {
 	var res []*openfgav1.Tuple
 	for i := 0; i < opts.PageSize; i++ {
-		tupleRecord, err := t.next()
+		tupleRecord, err := t.next(ctx)
 		if err != nil {
 			if errors.Is(err, storage.ErrIteratorDone) {
 				return res, "", nil
@@ -346,7 +379,7 @@ func (t *SQLTupleIterator) ToArray(
 	// Check if we are at the end of the iterator.
 	// If we are then we do not need to return a continuation token.
 	// This is why we have LIMIT+1 in the query.
-	tupleRecord, err := t.next()
+	tupleRecord, err := t.next(ctx)
 	if err != nil {
 		if errors.Is(err, storage.ErrIteratorDone) {
 			return res, "", nil
@@ -363,7 +396,7 @@ func (t *SQLTupleIterator) Next(ctx context.Context) (*openfgav1.Tuple, error) {
 		return nil, ctx.Err()
 	}
 
-	record, err := t.next()
+	record, err := t.next(ctx)
 	if err != nil {
 		return nil, err
 	}
@@ -377,7 +410,7 @@ func (t *SQLTupleIterator) Head(ctx context.Context) (*openfgav1.Tuple, error) {
 		return nil, ctx.Err()
 	}
 
-	record, err := t.head()
+	record, err := t.head(ctx)
 	if err != nil {
 		return nil, err
 	}
@@ -387,7 +420,11 @@ func (t *SQLTupleIterator) Head(ctx context.Context) (*openfgav1.Tuple, error) {
 
 // Stop terminates iteration.
 func (t *SQLTupleIterator) Stop() {
-	t.rows.Close()
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	if t.rows != nil {
+		_ = t.rows.Close()
+	}
 }
 
 // DBInfo encapsulates DB information for use in common method.
@@ -728,3 +765,33 @@ func AddFromUlid(sb sq.SelectBuilder, fromUlid string, sortDescending bool) sq.S
 	}
 	return sb.Where(sq.Gt{"ulid": fromUlid})
 }
+
+// HandleSQLError processes an SQL error and converts it into a more
+// specific error type based on the nature of the SQL error.
+func HandleSQLError(err error, args ...interface{}) error {
+	if errors.Is(err, sql.ErrNoRows) {
+		return storage.ErrNotFound
+	}
+
+	// handle mysql
+	if me, ok := err.(*mysql.MySQLError); ok && me.Number == 1062 {
+		if len(args) > 0 {
+			if tk, ok := args[0].(*openfgav1.TupleKey); ok {
+				return storage.InvalidWriteInputError(tk, openfgav1.TupleOperation_TUPLE_OPERATION_WRITE)
+			}
+		}
+		return storage.ErrCollision
+	}
+
+	// handle psql
+	if strings.Contains(err.Error(), "duplicate key value") {
+		if len(args) > 0 {
+			if tk, ok := args[0].(*openfgav1.TupleKey); ok {
+				return storage.InvalidWriteInputError(tk, openfgav1.TupleOperation_TUPLE_OPERATION_WRITE)
+			}
+		}
+		return storage.ErrCollision
+	}
+
+	return fmt.Errorf("sql error: %w", err)
+}



Test patch:
diff --git a/internal/graph/check_fast_path_test.go b/internal/graph/check_fast_path_test.go
index 3111db32..a51298c5 100644
--- a/internal/graph/check_fast_path_test.go
+++ b/internal/graph/check_fast_path_test.go
@@ -18,7 +18,7 @@ import (
 	parser "github.com/openfga/language/pkg/go/transformer"
 
 	"github.com/openfga/openfga/internal/concurrency"
-	"github.com/openfga/openfga/internal/graph/iterator"
+	"github.com/openfga/openfga/internal/iterator"
 	"github.com/openfga/openfga/internal/mocks"
 	"github.com/openfga/openfga/pkg/server/config"
 	"github.com/openfga/openfga/pkg/storage"
@@ -1331,17 +1331,6 @@ func TestBreadthFirstRecursiveMatch(t *testing.T) {
 			},
 			expected: true,
 		},
-		{
-			name:                 "duplicates_match_with_recursion",
-			currentLevelUsersets: hashset.New("group:1", "group:2", "group:3"),
-			usersetFromUser:      hashset.New("group:4"),
-			readMocks: [][]*openfgav1.Tuple{
-				{{Key: tuple.NewTupleKey("group:1", "parent", "group:3")}},
-				{{Key: tuple.NewTupleKey("group:2", "parent", "group:1")}},
-				{{Key: tuple.NewTupleKey("group:3", "parent", "group:4")}},
-			},
-			expected: true,
-		},
 		{
 			name:                 "no_duplicates_no_match_counts",
 			currentLevelUsersets: hashset.New("group:1", "group:2", "group:3"),
@@ -1370,7 +1359,7 @@ func TestBreadthFirstRecursiveMatch(t *testing.T) {
 
 			mockDatastore := mocks.NewMockRelationshipTupleReader(ctrl)
 			for _, mock := range tt.readMocks {
-				mockDatastore.EXPECT().Read(gomock.Any(), storeID, gomock.Any(), gomock.Any()).Times(1).Return(storage.NewStaticTupleIterator(mock), nil)
+				mockDatastore.EXPECT().Read(gomock.Any(), storeID, gomock.Any(), gomock.Any()).MaxTimes(1).Return(storage.NewStaticTupleIterator(mock), nil)
 			}
 
 			model := parser.MustTransformDSLToProto(`
@@ -1405,7 +1394,10 @@ func TestBreadthFirstRecursiveMatch(t *testing.T) {
 
 			result := false
 			for outcome := range checkOutcomeChan {
-				result = outcome.resp.Allowed
+				if outcome.resp.Allowed {
+					result = true
+					break
+				}
 			}
 			require.Equal(t, tt.expected, result)
 		})
@@ -1419,14 +1411,16 @@ func TestBreadthFirstRecursiveMatch(t *testing.T) {
 		mockDatastore := mocks.NewMockRelationshipTupleReader(ctrl)
 		ctx := context.Background()
 		ctx, cancel := context.WithCancel(ctx)
-		// Stop is called either sync (when context is already cancelled before sending iterator through channel
-		// or async when calling drainIteratorChannel
+		// Stop is called under race conditions thus is not guaranteed these observers may see a call to it
 		iter1 := mocks.NewMockIterator[*openfgav1.Tuple](ctrl)
 		iter1.EXPECT().Stop().MaxTimes(1)
+		iter1.EXPECT().Next(gomock.Any()).MaxTimes(1).Return(nil, storage.ErrIteratorDone)
 		iter2 := mocks.NewMockIterator[*openfgav1.Tuple](ctrl)
 		iter2.EXPECT().Stop().MaxTimes(1)
+		iter2.EXPECT().Next(gomock.Any()).MaxTimes(1).Return(nil, storage.ErrIteratorDone)
 		iter3 := mocks.NewMockIterator[*openfgav1.Tuple](ctrl)
 		iter3.EXPECT().Stop().MaxTimes(1)
+		iter3.EXPECT().Next(gomock.Any()).MaxTimes(1).Return(nil, storage.ErrIteratorDone)
 		// currentUsersetLevel.Values() doesn't return results in order, thus there is no guarantee that `Times` will be consistent as it can return err due to context being cancelled
 		mockDatastore.EXPECT().Read(gomock.Any(), storeID, gomock.Any(), gomock.Any()).MaxTimes(1).Return(iter1, nil)
 		mockDatastore.EXPECT().Read(gomock.Any(), storeID, gomock.Any(), gomock.Any()).MaxTimes(1).DoAndReturn(func(ctx context.Context, store string, tupleKey *openfgav1.TupleKey, options storage.ReadOptions) (storage.TupleIterator, error) {
@@ -1648,7 +1642,7 @@ func TestRecursiveTTUFastPath(t *testing.T) {
 
 func TestRecursiveTTUFastPathV2(t *testing.T) {
 	t.Cleanup(func() {
-		goleak.VerifyNone(t)
+		goleak.VerifyNone(t, goleak.IgnoreTopFunction("github.com/openfga/openfga/internal/iterator.drainOnExit(...)"))
 	})
 
 	model := parser.MustTransformDSLToProto(`
@@ -2121,7 +2115,7 @@ type group
 
 func TestRecursiveUsersetFastPath(t *testing.T) {
 	t.Cleanup(func() {
-		goleak.VerifyNone(t)
+		goleak.VerifyNone(t, goleak.IgnoreTopFunction("github.com/openfga/openfga/internal/iterator.drainOnExit(...)"))
 	})
 	tests := []struct {
 		name                            string
diff --git a/internal/graph/check_test.go b/internal/graph/check_test.go
index c5a0be8b..ad70cc30 100644
--- a/internal/graph/check_test.go
+++ b/internal/graph/check_test.go
@@ -2688,6 +2688,7 @@ func TestConsumeUsersets(t *testing.T) {
 			expectedResolveCheckResponse: nil,
 			errorExpected:                fmt.Errorf("mock_error"),
 		},
+		// TODO: This test is flaky needs further investigation
 		{
 			name: "iterator_error_first_batch_but_success_second",
 			tuples: []dsResults{
diff --git a/internal/graph/object_providers_test.go b/internal/graph/object_providers_test.go
index 2e37a5a5..7612976a 100644
--- a/internal/graph/object_providers_test.go
+++ b/internal/graph/object_providers_test.go
@@ -120,7 +120,7 @@ func TestRecursiveObjectProvider(t *testing.T) {
 
 func TestRecursiveTTUObjectProvider(t *testing.T) {
 	t.Cleanup(func() {
-		goleak.VerifyNone(t)
+		goleak.VerifyNone(t, goleak.IgnoreTopFunction("github.com/openfga/openfga/internal/iterator.drainOnExit(...)"))
 	})
 
 	storeID := ulid.Make().String()
@@ -234,8 +234,10 @@ func TestRecursiveTTUObjectProvider(t *testing.T) {
 				t.Cleanup(c.End)
 
 				ctx := setRequestContext(context.Background(), ts, mockDatastore, nil)
-				_, err = c.Begin(ctx, req)
-				require.ErrorIs(t, err, mockError)
+				out, err := c.Begin(ctx, req)
+				require.NoError(t, err)
+				msg := <-out
+				require.ErrorIs(t, msg.err, mockError)
 			})
 
 			t.Run("when_iterator_errors", func(t *testing.T) {
@@ -296,7 +298,7 @@ func TestRecursiveTTUObjectProvider(t *testing.T) {
 
 func TestRecursiveUsersetObjectProvider(t *testing.T) {
 	t.Cleanup(func() {
-		goleak.VerifyNone(t)
+		goleak.VerifyNone(t, goleak.IgnoreTopFunction("github.com/openfga/openfga/internal/iterator.drainOnExit(...)"))
 	})
 
 	storeID := ulid.Make().String()
@@ -379,17 +381,20 @@ func TestRecursiveUsersetObjectProvider(t *testing.T) {
 			})
 
 			t.Run("when_fastPathRewrite_errors", func(t *testing.T) {
+				mockError := fmt.Errorf("error")
 				mockDatastore.EXPECT().
 					ReadStartingWithUser(gomock.Any(), storeID, gomock.Any(), gomock.Any()).
 					Times(1).
-					Return(nil, fmt.Errorf("error"))
+					Return(nil, mockError)
 
 				c := newRecursiveUsersetObjectProvider(ts, concurrencyLimit)
 				t.Cleanup(c.End)
 
 				ctx := setRequestContext(context.Background(), ts, mockDatastore, nil)
-				_, err = c.Begin(ctx, req)
-				require.Error(t, err)
+				out, err := c.Begin(ctx, req)
+				require.NoError(t, err)
+				msg := <-out
+				require.ErrorIs(t, mockError, msg.err)
 			})
 
 			t.Run("when_iterator_errors", func(t *testing.T) {
diff --git a/internal/iterator/fan_in_test.go b/internal/iterator/fan_in_test.go
new file mode 100644
index 00000000..2f23fdb0
--- /dev/null
+++ b/internal/iterator/fan_in_test.go
@@ -0,0 +1,71 @@
+package iterator
+
+import (
+	"context"
+	"strconv"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"go.uber.org/goleak"
+	"go.uber.org/mock/gomock"
+
+	"github.com/openfga/openfga/internal/mocks"
+	"github.com/openfga/openfga/pkg/storage"
+)
+
+func makeIterChan(ctrl *gomock.Controller, id string, next bool) chan *Msg {
+	iterChan := make(chan *Msg, 1)
+	iter := mocks.NewMockIterator[string](ctrl)
+	if next {
+		iter.EXPECT().Next(gomock.Any()).MaxTimes(1).Return(id, nil)
+		iter.EXPECT().Next(gomock.Any()).MaxTimes(1).Return("", storage.ErrIteratorDone)
+	}
+	iter.EXPECT().Stop().MinTimes(1)
+	iterChan <- &Msg{Iter: iter}
+	close(iterChan)
+	return iterChan
+}
+
+func TestFanIn(t *testing.T) {
+	t.Cleanup(func() {
+		goleak.VerifyNone(t)
+	})
+	ctx := context.Background()
+	ctrl := gomock.NewController(t)
+	t.Cleanup(ctrl.Finish)
+	fanin := NewFanIn(ctx, 2)
+
+	fanin.Add(makeIterChan(ctrl, "1", true))
+	fanin.Add(makeIterChan(ctrl, "2", true))
+	fanin.Add(makeIterChan(ctrl, "3", true))
+	fanin.Add(makeIterChan(ctrl, "4", true))
+	fanin.Add(makeIterChan(ctrl, "5", true))
+	fanin.Add(makeIterChan(ctrl, "6", true))
+	fanin.Add(makeIterChan(ctrl, "7", true))
+	fanin.Done()
+	fanin.Add(makeIterChan(ctrl, "8", false))
+	fanin.Add(makeIterChan(ctrl, "9", false))
+
+	require.Equal(t, 9, fanin.Count())
+
+	out := fanin.Out()
+	iterations := 0
+	for msg := range out {
+		id, err := msg.Iter.Next(ctx)
+		require.NoError(t, err)
+		i, err := strconv.Atoi(id)
+		require.NoError(t, err)
+		require.Positive(t, i)
+		require.LessOrEqual(t, i, 7)
+		_, err = msg.Iter.Next(ctx)
+		require.Equal(t, storage.ErrIteratorDone, err)
+		msg.Iter.Stop()
+		iterations++
+		if iterations > 2 {
+			break
+		}
+	}
+	fanin.Close()
+	drained := <-fanin.drained
+	require.True(t, drained)
+}
diff --git a/internal/graph/iterator/iterator_stream_test.go b/internal/iterator/stream_test.go
similarity index 100%
rename from internal/graph/iterator/iterator_stream_test.go
rename to internal/iterator/stream_test.go
diff --git a/pkg/server/server_test.go b/pkg/server/server_test.go
index 0987618d..92ee7cee 100644
--- a/pkg/server/server_test.go
+++ b/pkg/server/server_test.go
@@ -844,7 +844,7 @@ func BenchmarkOpenFGAServer(b *testing.B) {
 		testDatastore := storagefixtures.RunDatastoreTestContainer(b, "postgres")
 
 		uri := testDatastore.GetConnectionURI(true)
-		ds, err := postgres.New(uri, sqlcommon.NewConfig())
+		ds, err := postgres.New(uri, sqlcommon.NewConfig(sqlcommon.WithMaxOpenConns(10)))
 		require.NoError(b, err)
 		b.Cleanup(ds.Close)
 		test.RunAllBenchmarks(b, ds)
@@ -860,7 +860,7 @@ func BenchmarkOpenFGAServer(b *testing.B) {
 		testDatastore := storagefixtures.RunDatastoreTestContainer(b, "mysql")
 
 		uri := testDatastore.GetConnectionURI(true)
-		ds, err := mysql.New(uri, sqlcommon.NewConfig())
+		ds, err := mysql.New(uri, sqlcommon.NewConfig(sqlcommon.WithMaxOpenConns(10)))
 		require.NoError(b, err)
 		b.Cleanup(ds.Close)
 		test.RunAllBenchmarks(b, ds)
diff --git a/pkg/server/test/list_objects.go b/pkg/server/test/list_objects.go
index 8e75b251..f681c6cf 100644
--- a/pkg/server/test/list_objects.go
+++ b/pkg/server/test/list_objects.go
@@ -507,9 +507,7 @@ func TestListObjects(t *testing.T, ds storage.OpenFGADatastore) {
 				opts = append(opts, commands.WithListObjectsDeadline(test.listObjectsDeadline))
 			}
 
-			localCheckOpts := []graph.LocalCheckerOption{
-				graph.WithResolveNodeBreadthLimit(100),
-			}
+			var localCheckOpts []graph.LocalCheckerOption
 			cacheOpts := []graph.CachedCheckResolverOpt{
 				graph.WithCacheTTL(10 * time.Second),
 			}
diff --git a/pkg/storage/mysql/mysql_test.go b/pkg/storage/mysql/mysql_test.go
index 58da3592..62fd8f95 100644
--- a/pkg/storage/mysql/mysql_test.go
+++ b/pkg/storage/mysql/mysql_test.go
@@ -78,7 +78,7 @@ func TestReadEnsureNoOrder(t *testing.T) {
 			thirdTuple := tuple.NewTupleKey("doc:object_id_3", "relation", "user:user_3")
 
 			err = sqlcommon.Write(ctx,
-				sqlcommon.NewDBInfo(ds.db, ds.stbl, HandleSQLError, "mysql"),
+				sqlcommon.NewDBInfo(ds.db, ds.stbl, sqlcommon.HandleSQLError, "mysql"),
 				store,
 				[]*openfgav1.TupleKeyWithoutCondition{},
 				[]*openfgav1.TupleKey{firstTuple},
@@ -87,7 +87,7 @@ func TestReadEnsureNoOrder(t *testing.T) {
 
 			// Tweak time so that ULID is smaller.
 			err = sqlcommon.Write(ctx,
-				sqlcommon.NewDBInfo(ds.db, ds.stbl, HandleSQLError, "mysql"),
+				sqlcommon.NewDBInfo(ds.db, ds.stbl, sqlcommon.HandleSQLError, "mysql"),
 				store,
 				[]*openfgav1.TupleKeyWithoutCondition{},
 				[]*openfgav1.TupleKey{secondTuple},
@@ -96,7 +96,7 @@ func TestReadEnsureNoOrder(t *testing.T) {
 
 			// Tweak time so that ULID is smaller.
 			err = sqlcommon.Write(ctx,
-				sqlcommon.NewDBInfo(ds.db, ds.stbl, HandleSQLError, "mysql"),
+				sqlcommon.NewDBInfo(ds.db, ds.stbl, sqlcommon.HandleSQLError, "mysql"),
 				store,
 				[]*openfgav1.TupleKeyWithoutCondition{},
 				[]*openfgav1.TupleKey{thirdTuple},
@@ -180,7 +180,7 @@ func TestCtxCancel(t *testing.T) {
 			thirdTuple := tuple.NewTupleKey("doc:object_id_3", "relation", "user:user_3")
 
 			err = sqlcommon.Write(ctx,
-				sqlcommon.NewDBInfo(ds.db, ds.stbl, HandleSQLError, "mysql"),
+				sqlcommon.NewDBInfo(ds.db, ds.stbl, sqlcommon.HandleSQLError, "mysql"),
 				store,
 				[]*openfgav1.TupleKeyWithoutCondition{},
 				[]*openfgav1.TupleKey{firstTuple},
@@ -189,7 +189,7 @@ func TestCtxCancel(t *testing.T) {
 
 			// Tweak time so that ULID is smaller.
 			err = sqlcommon.Write(ctx,
-				sqlcommon.NewDBInfo(ds.db, ds.stbl, HandleSQLError, "mysql"),
+				sqlcommon.NewDBInfo(ds.db, ds.stbl, sqlcommon.HandleSQLError, "mysql"),
 				store,
 				[]*openfgav1.TupleKeyWithoutCondition{},
 				[]*openfgav1.TupleKey{secondTuple},
@@ -198,7 +198,7 @@ func TestCtxCancel(t *testing.T) {
 
 			// Tweak time so that ULID is smaller.
 			err = sqlcommon.Write(ctx,
-				sqlcommon.NewDBInfo(ds.db, ds.stbl, HandleSQLError, "mysql"),
+				sqlcommon.NewDBInfo(ds.db, ds.stbl, sqlcommon.HandleSQLError, "mysql"),
 				store,
 				[]*openfgav1.TupleKeyWithoutCondition{},
 				[]*openfgav1.TupleKey{thirdTuple},
@@ -241,7 +241,7 @@ func TestReadPageEnsureOrder(t *testing.T) {
 	secondTuple := tuple.NewTupleKey("doc:object_id_2", "relation", "user:user_2")
 
 	err = sqlcommon.Write(ctx,
-		sqlcommon.NewDBInfo(ds.db, ds.stbl, HandleSQLError, "mysql"),
+		sqlcommon.NewDBInfo(ds.db, ds.stbl, sqlcommon.HandleSQLError, "mysql"),
 		store,
 		[]*openfgav1.TupleKeyWithoutCondition{},
 		[]*openfgav1.TupleKey{firstTuple},
@@ -250,7 +250,7 @@ func TestReadPageEnsureOrder(t *testing.T) {
 
 	// Tweak time so that ULID is smaller.
 	err = sqlcommon.Write(ctx,
-		sqlcommon.NewDBInfo(ds.db, ds.stbl, HandleSQLError, "mysql"),
+		sqlcommon.NewDBInfo(ds.db, ds.stbl, sqlcommon.HandleSQLError, "mysql"),
 		store,
 		[]*openfgav1.TupleKeyWithoutCondition{},
 		[]*openfgav1.TupleKey{secondTuple},
@@ -546,7 +546,7 @@ func TestHandleSQLError(t *testing.T) {
 			Number:  1062,
 			Message: "Duplicate entry '' for key ''",
 		}
-		err := HandleSQLError(duplicateKeyError, &openfgav1.TupleKey{
+		err := sqlcommon.HandleSQLError(duplicateKeyError, &openfgav1.TupleKey{
 			Object:   "object",
 			Relation: "relation",
 			User:     "user",
@@ -559,13 +559,13 @@ func TestHandleSQLError(t *testing.T) {
 			Number:  1062,
 			Message: "Duplicate entry '' for key ''",
 		}
-		err := HandleSQLError(duplicateKeyError)
+		err := sqlcommon.HandleSQLError(duplicateKeyError)
 
 		require.ErrorIs(t, err, storage.ErrCollision)
 	})
 
 	t.Run("sql.ErrNoRows_is_converted_to_storage.ErrNotFound_error", func(t *testing.T) {
-		err := HandleSQLError(sql.ErrNoRows)
+		err := sqlcommon.HandleSQLError(sql.ErrNoRows)
 		require.ErrorIs(t, err, storage.ErrNotFound)
 	})
 }
diff --git a/pkg/storage/postgres/postgres_test.go b/pkg/storage/postgres/postgres_test.go
index 888280a8..bf271182 100644
--- a/pkg/storage/postgres/postgres_test.go
+++ b/pkg/storage/postgres/postgres_test.go
@@ -76,7 +76,7 @@ func TestReadEnsureNoOrder(t *testing.T) {
 			thirdTuple := tuple.NewTupleKey("doc:object_id_3", "relation", "user:user_3")
 
 			err = sqlcommon.Write(ctx,
-				sqlcommon.NewDBInfo(ds.db, ds.stbl, HandleSQLError, "postgres"),
+				sqlcommon.NewDBInfo(ds.db, ds.stbl, sqlcommon.HandleSQLError, "postgres"),
 				store,
 				[]*openfgav1.TupleKeyWithoutCondition{},
 				[]*openfgav1.TupleKey{firstTuple},
@@ -85,7 +85,7 @@ func TestReadEnsureNoOrder(t *testing.T) {
 
 			// Tweak time so that ULID is smaller.
 			err = sqlcommon.Write(ctx,
-				sqlcommon.NewDBInfo(ds.db, ds.stbl, HandleSQLError, "postgres"),
+				sqlcommon.NewDBInfo(ds.db, ds.stbl, sqlcommon.HandleSQLError, "postgres"),
 				store,
 				[]*openfgav1.TupleKeyWithoutCondition{},
 				[]*openfgav1.TupleKey{secondTuple},
@@ -93,7 +93,7 @@ func TestReadEnsureNoOrder(t *testing.T) {
 			require.NoError(t, err)
 
 			err = sqlcommon.Write(ctx,
-				sqlcommon.NewDBInfo(ds.db, ds.stbl, HandleSQLError, "postgres"),
+				sqlcommon.NewDBInfo(ds.db, ds.stbl, sqlcommon.HandleSQLError, "postgres"),
 				store,
 				[]*openfgav1.TupleKeyWithoutCondition{},
 				[]*openfgav1.TupleKey{thirdTuple},
@@ -177,7 +177,7 @@ func TestCtxCancel(t *testing.T) {
 			thirdTuple := tuple.NewTupleKey("doc:object_id_3", "relation", "user:user_3")
 
 			err = sqlcommon.Write(ctx,
-				sqlcommon.NewDBInfo(ds.db, ds.stbl, HandleSQLError, "postgres"),
+				sqlcommon.NewDBInfo(ds.db, ds.stbl, sqlcommon.HandleSQLError, "postgres"),
 				store,
 				[]*openfgav1.TupleKeyWithoutCondition{},
 				[]*openfgav1.TupleKey{firstTuple},
@@ -186,7 +186,7 @@ func TestCtxCancel(t *testing.T) {
 
 			// Tweak time so that ULID is smaller.
 			err = sqlcommon.Write(ctx,
-				sqlcommon.NewDBInfo(ds.db, ds.stbl, HandleSQLError, "postgres"),
+				sqlcommon.NewDBInfo(ds.db, ds.stbl, sqlcommon.HandleSQLError, "postgres"),
 				store,
 				[]*openfgav1.TupleKeyWithoutCondition{},
 				[]*openfgav1.TupleKey{secondTuple},
@@ -194,7 +194,7 @@ func TestCtxCancel(t *testing.T) {
 			require.NoError(t, err)
 
 			err = sqlcommon.Write(ctx,
-				sqlcommon.NewDBInfo(ds.db, ds.stbl, HandleSQLError, "postgres"),
+				sqlcommon.NewDBInfo(ds.db, ds.stbl, sqlcommon.HandleSQLError, "postgres"),
 				store,
 				[]*openfgav1.TupleKeyWithoutCondition{},
 				[]*openfgav1.TupleKey{thirdTuple},
@@ -236,7 +236,7 @@ func TestReadPageEnsureOrder(t *testing.T) {
 	secondTuple := tuple.NewTupleKey("doc:object_id_2", "relation", "user:user_2")
 
 	err = sqlcommon.Write(ctx,
-		sqlcommon.NewDBInfo(ds.db, ds.stbl, HandleSQLError, "postgres"),
+		sqlcommon.NewDBInfo(ds.db, ds.stbl, sqlcommon.HandleSQLError, "postgres"),
 		store,
 		[]*openfgav1.TupleKeyWithoutCondition{},
 		[]*openfgav1.TupleKey{firstTuple},
@@ -245,7 +245,7 @@ func TestReadPageEnsureOrder(t *testing.T) {
 
 	// Tweak time so that ULID is smaller.
 	err = sqlcommon.Write(ctx,
-		sqlcommon.NewDBInfo(ds.db, ds.stbl, HandleSQLError, "postgres"),
+		sqlcommon.NewDBInfo(ds.db, ds.stbl, sqlcommon.HandleSQLError, "postgres"),
 		store,
 		[]*openfgav1.TupleKeyWithoutCondition{},
 		[]*openfgav1.TupleKey{secondTuple},
@@ -538,7 +538,7 @@ func TestMarshalledAssertions(t *testing.T) {
 
 func TestHandleSQLError(t *testing.T) {
 	t.Run("duplicate_key_value_error_with_tuple_key_wraps_ErrInvalidWriteInput", func(t *testing.T) {
-		err := HandleSQLError(errors.New("duplicate key value"), &openfgav1.TupleKey{
+		err := sqlcommon.HandleSQLError(errors.New("duplicate key value"), &openfgav1.TupleKey{
 			Object:   "object",
 			Relation: "relation",
 			User:     "user",
@@ -548,12 +548,12 @@ func TestHandleSQLError(t *testing.T) {
 
 	t.Run("duplicate_key_value_error_without_tuple_key_returns_collision", func(t *testing.T) {
 		duplicateKeyError := errors.New("duplicate key value")
-		err := HandleSQLError(duplicateKeyError)
+		err := sqlcommon.HandleSQLError(duplicateKeyError)
 		require.ErrorIs(t, err, storage.ErrCollision)
 	})
 
 	t.Run("sql.ErrNoRows_is_converted_to_storage.ErrNotFound_error", func(t *testing.T) {
-		err := HandleSQLError(sql.ErrNoRows)
+		err := sqlcommon.HandleSQLError(sql.ErrNoRows)
 		require.ErrorIs(t, err, storage.ErrNotFound)
 	})
 }