fix: respect consistency preference in read and listObjects (open…

…fga#2113)

CHANGELOG.md

@@ -18,6 +18,7 @@ Try to keep listed changes to a concise bulleted list of simple explanations of
 ### Fixed
 - Improve `Check` performance in the case that the query involves types that cannot be reached from the source. Enable via experimental flag `enable-check-optimizations`. [#2104](https://github.com/openfga/openfga/pull/2104)
 - Fix regression introduced in #2091: error message for invalid Writes. [#2110](https://github.com/openfga/openfga/pull/2110)
+- Ensure `/read` and `/list-objects` respect the received `Consistency` values [#2113](https://github.com/openfga/openfga/pull/2113)
 
 ## [1.8.0] - 2024-11-08
 [Full changelog](https://github.com/openfga/openfga/compare/v1.7.0...v1.8.0)

pkg/server/commands/expand_test.go

@@ -4,13 +4,18 @@ import (
 	"context"
 	"testing"
 
+	"github.com/oklog/ulid/v2"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/mock/gomock"
 
 	openfgav1 "github.com/openfga/api/proto/openfga/v1"
 
+	"github.com/openfga/openfga/internal/mocks"
 	serverErrors "github.com/openfga/openfga/pkg/server/errors"
+	"github.com/openfga/openfga/pkg/storage"
 	"github.com/openfga/openfga/pkg/storage/memory"
 	storagetest "github.com/openfga/openfga/pkg/storage/test"
+	"github.com/openfga/openfga/pkg/testutils"
 	"github.com/openfga/openfga/pkg/tuple"
 	"github.com/openfga/openfga/pkg/typesystem"
 )
@@ -147,4 +152,59 @@ func TestExpand(t *testing.T) {
 		finalUsers := root.GetLeaf().GetUsers().GetUsers()
 		require.ElementsMatch(t, finalUsers, []string{"user:bob", "user:alice"})
 	})
+
+	t.Run("respects_consistency_preference", func(t *testing.T) {
+		mockController := gomock.NewController(t)
+		defer mockController.Finish()
+
+		ctx := context.Background()
+
+		// arrange: write model
+		model := testutils.MustTransformDSLToProtoWithID(modelStr)
+
+		typesys, err := typesystem.NewAndValidate(ctx, model)
+		require.NoError(t, err)
+
+		ctx = typesystem.ContextWithTypesystem(ctx, typesys)
+
+		mockDatastore := mocks.NewMockOpenFGADatastore(mockController)
+		query := NewExpandQuery(mockDatastore)
+
+		// No Consistency Specified in this request
+		expandRequest := &openfgav1.ExpandRequest{
+			StoreId: ulid.Make().String(),
+			TupleKey: &openfgav1.ExpandRequestTupleKey{
+				Object: "document:1", Relation: "viewer",
+			},
+		}
+		unspecified := storage.ReadOptions{
+			Consistency: storage.ConsistencyOptions{
+				Preference: openfgav1.ConsistencyPreference_UNSPECIFIED,
+			},
+		}
+
+		// expect to receive UNSPECIFIED
+		mockDatastore.EXPECT().Read(gomock.Any(), gomock.Any(), gomock.Any(), unspecified).Times(1)
+		_, err = query.Execute(ctx, expandRequest)
+		require.NoError(t, err)
+
+		// Now run it again with HIGHER_CONSISTENCY
+		expandRequest.Consistency = openfgav1.ConsistencyPreference_HIGHER_CONSISTENCY
+		higherConsistency := storage.ReadOptions{
+			Consistency: storage.ConsistencyOptions{
+				Preference: openfgav1.ConsistencyPreference_HIGHER_CONSISTENCY,
+			},
+		}
+
+		// expect to receive HIGHER_CONSISTENCY
+		mockDatastore.EXPECT().Read(gomock.Any(), gomock.Any(), gomock.Any(), higherConsistency).Times(1)
+		_, err = query.Execute(ctx, &openfgav1.ExpandRequest{
+			StoreId: ulid.Make().String(),
+			TupleKey: &openfgav1.ExpandRequestTupleKey{
+				Object: "document:1", Relation: "viewer",
+			},
+			Consistency: openfgav1.ConsistencyPreference_HIGHER_CONSISTENCY,
+		})
+		require.NoError(t, err)
+	})
 }

pkg/server/commands/listusers/list_users_rpc_test.go

@@ -3955,3 +3955,68 @@ func TestListUsers_CorrectContext(t *testing.T) {
 		require.ErrorContains(t, err, "typesystem missing in context")
 	})
 }
+
+func TestListUsersRespectsConsistency(t *testing.T) {
+	t.Cleanup(func() {
+		goleak.VerifyNone(t)
+	})
+
+	modelStr := `
+		model
+			schema 1.1
+		type user
+		type document
+			relations
+				define viewer: [user]`
+
+	mockController := gomock.NewController(t)
+	defer mockController.Finish()
+
+	mockDatastore := mocks.NewMockOpenFGADatastore(mockController)
+
+	ctx := context.Background()
+
+	// arrange: write model
+	model := testutils.MustTransformDSLToProtoWithID(modelStr)
+
+	typesys, err := typesystem.NewAndValidate(ctx, model)
+	require.NoError(t, err)
+
+	query := NewListUsersQuery(mockDatastore)
+	ctx = typesystem.ContextWithTypesystem(ctx, typesys)
+
+	t.Run("uses_passed_consistency_preference", func(t *testing.T) {
+		higherConsistency := storage.ReadOptions{
+			Consistency: storage.ConsistencyOptions{
+				Preference: openfgav1.ConsistencyPreference_HIGHER_CONSISTENCY,
+			},
+		}
+		mockDatastore.EXPECT().Read(gomock.Any(), gomock.Any(), gomock.Any(), higherConsistency).Times(1)
+
+		_, err = query.ListUsers(ctx, &openfgav1.ListUsersRequest{
+			Consistency: openfgav1.ConsistencyPreference_HIGHER_CONSISTENCY,
+			UserFilters: []*openfgav1.UserTypeFilter{{Type: "user"}},
+			Object:      &openfgav1.Object{Type: "document", Id: "1"},
+			Relation:    "viewer",
+			StoreId:     ulid.Make().String(),
+		})
+		require.NoError(t, err)
+	})
+
+	t.Run("unspecified_consistency_if_user_didnt_specify", func(t *testing.T) {
+		unspecified := storage.ReadOptions{
+			Consistency: storage.ConsistencyOptions{
+				Preference: openfgav1.ConsistencyPreference_UNSPECIFIED,
+			},
+		}
+		mockDatastore.EXPECT().Read(gomock.Any(), gomock.Any(), gomock.Any(), unspecified).Times(1)
+
+		_, err = query.ListUsers(ctx, &openfgav1.ListUsersRequest{
+			UserFilters: []*openfgav1.UserTypeFilter{{Type: "user"}},
+			Object:      &openfgav1.Object{Type: "document", Id: "1"},
+			Relation:    "viewer",
+			StoreId:     ulid.Make().String(),
+		})
+		require.NoError(t, err)
+	})
+}

pkg/server/commands/read.go

@@ -90,8 +90,10 @@ func (q *ReadQuery) Execute(ctx context.Context, req *openfgav1.ReadRequest) (*o
 	}
 
 	opts := storage.ReadPageOptions{
-		Pagination: storage.NewPaginationOptions(req.GetPageSize().GetValue(), string(decodedContToken)),
+		Pagination:  storage.NewPaginationOptions(req.GetPageSize().GetValue(), string(decodedContToken)),
+		Consistency: storage.ConsistencyOptions{Preference: req.GetConsistency()},
 	}
+
 	tuples, contUlid, err := q.datastore.ReadPage(ctx, store, tupleUtils.ConvertReadRequestTupleKeyToTupleKey(tk), opts)
 	if err != nil {
 		return nil, serverErrors.HandleError("", err)

pkg/server/commands/read_test.go

@@ -91,6 +91,7 @@ func TestReadCommand(t *testing.T) {
 				PageSize: storage.DefaultPageSize,
 				From:     "",
 			},
+			Consistency: storage.ConsistencyOptions{Preference: openfgav1.ConsistencyPreference_HIGHER_CONSISTENCY},
 		}
 		mockDatastore.EXPECT().ReadPage(gomock.Any(), storeID, tupleKey, opts).Times(1)
 
@@ -100,6 +101,7 @@ func TestReadCommand(t *testing.T) {
 			TupleKey:          &openfgav1.ReadRequestTupleKey{Object: "document:1", Relation: "reader", User: "user:maria"},
 			PageSize:          nil,
 			ContinuationToken: "",
+			Consistency:       openfgav1.ConsistencyPreference_HIGHER_CONSISTENCY,
 		})
 		require.NoError(t, err)
 	})

pkg/server/commands/reverseexpand/reverse_expand.go

@@ -339,6 +339,7 @@ LoopOnEdges:
 			ContextualTuples: req.ContextualTuples,
 			Context:          req.Context,
 			edge:             innerLoopEdge,
+			Consistency:      req.Consistency,
 		}
 		switch innerLoopEdge.Type {
 		case graph.DirectEdge:
@@ -566,6 +567,7 @@ LoopOnIterator:
 				ContextualTuples: req.ContextualTuples,
 				Context:          req.Context,
 				edge:             req.edge,
+				Consistency:      req.Consistency,
 			}, resultChan, intersectionOrExclusionInPreviousEdges, resolutionMetadata)
 		})
 	}

pkg/server/commands/reverseexpand/reverse_expand_test.go

@@ -753,3 +753,79 @@ func TestReverseExpandDispatchCount(t *testing.T) {
 		})
 	}
 }
+
+func TestReverseExpandHonorsConsistency(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, err := typesystem.New(model)
+	require.NoError(t, err)
+	mockController := gomock.NewController(t)
+	defer mockController.Finish()
+
+	ctx := context.Background()
+	mockDatastore := mocks.NewMockOpenFGADatastore(mockController)
+
+	// run once with no consistency specified
+	unspecifiedConsistency := storage.ReadStartingWithUserOptions{
+		Consistency: storage.ConsistencyOptions{Preference: openfgav1.ConsistencyPreference_UNSPECIFIED},
+	}
+	mockDatastore.EXPECT().
+		ReadStartingWithUser(gomock.Any(), store, gomock.Any(), unspecifiedConsistency).
+		Times(1)
+
+	request := &ReverseExpandRequest{
+		StoreID:    store,
+		ObjectType: "document",
+		Relation:   "viewer",
+		User: &UserRefObject{
+			Object: &openfgav1.Object{
+				Type: "user",
+				Id:   "maria",
+			},
+		},
+		ContextualTuples: []*openfgav1.TupleKey{},
+	}
+
+	reverseExpandQuery := NewReverseExpandQuery(mockDatastore, typeSystem)
+	resultChan := make(chan *ReverseExpandResult)
+
+	err = reverseExpandQuery.Execute(ctx, request, resultChan, NewResolutionMetadata())
+	require.NoError(t, err)
+
+	// now do it again with specified consistency
+	highConsistency := storage.ReadStartingWithUserOptions{
+		Consistency: storage.ConsistencyOptions{Preference: openfgav1.ConsistencyPreference_HIGHER_CONSISTENCY},
+	}
+
+	request.Consistency = openfgav1.ConsistencyPreference_HIGHER_CONSISTENCY
+	mockDatastore.EXPECT().
+		ReadStartingWithUser(gomock.Any(), store, gomock.Any(), highConsistency).
+		Times(1)
+
+	resultChanTwo := make(chan *ReverseExpandResult)
+	err = reverseExpandQuery.Execute(ctx, request, resultChanTwo, NewResolutionMetadata())
+	require.NoError(t, err)
+
+	// Make sure we didn't leave channels open
+	select {
+	case _, open := <-resultChan:
+		if open {
+			require.FailNow(t, "results channels should be closed")
+		}
+	case _, open := <-resultChanTwo:
+		if open {
+			require.FailNow(t, "results channels should be closed")
+		}
+	}
+}