Sessions list pagination (#4079)

* Session list pagination refactor

* address pr comments, add explanatory comments, clean up sql queries, move tests into proper files

* fix build error

* fix test error

internal/daemon/controller/common/scopeids/scope_ids_test.go

@@ -51,7 +51,7 @@ func TestListingScopeIds(t *testing.T) {
 	sessionsRepoFn := func(opt ...session.Option) (*session.Repository, error) {
 		return session.NewRepository(ctx, rw, rw, kms, opt...)
 	}
-	sess, err := sessions.NewService(ctx, sessionsRepoFn, iamRepoFn)
+	sess, err := sessions.NewService(ctx, sessionsRepoFn, iamRepoFn, 1000)
 	require.NoError(t, err)
 
 	tcs := []struct {

internal/daemon/controller/handler.go

@@ -252,7 +252,7 @@ func (c *Controller) registerGrpcServices(s *grpc.Server) error {
 		services.RegisterRoleServiceServer(s, rs)
 	}
 	if _, ok := currentServices[services.SessionService_ServiceDesc.ServiceName]; !ok {
-		ss, err := sessions.NewService(c.baseContext, c.SessionRepoFn, c.IamRepoFn)
+		ss, err := sessions.NewService(c.baseContext, c.SessionRepoFn, c.IamRepoFn, c.conf.RawConfig.Controller.MaxPageSize)
 		if err != nil {
 			return fmt.Errorf("failed to create session handler service: %w", err)
 		}

internal/daemon/controller/handlers/sessions/session_list_benchmarks_test.go

@@ -109,7 +109,7 @@ func BenchmarkSessionList(b *testing.B) {
 				return serversRepo, nil
 			}
 
-			s, err := sessions.NewService(ctx, sessRepoFn, iamRepoFn)
+			s, err := sessions.NewService(ctx, sessRepoFn, iamRepoFn, 1000)
 			require.NoError(b, err)
 
 			var users []*userWithToken

internal/daemon/controller/handlers/sessions/session_service.go

@@ -14,6 +14,8 @@ import (
 	"github.com/hashicorp/boundary/internal/daemon/controller/handlers"
 	"github.com/hashicorp/boundary/internal/errors"
 	pbs "github.com/hashicorp/boundary/internal/gen/controller/api/services"
+	"github.com/hashicorp/boundary/internal/listtoken"
+	"github.com/hashicorp/boundary/internal/pagination"
 	"github.com/hashicorp/boundary/internal/perms"
 	"github.com/hashicorp/boundary/internal/requests"
 	"github.com/hashicorp/boundary/internal/session"
@@ -53,14 +55,15 @@ func init() {
 type Service struct {
 	pbs.UnsafeSessionServiceServer
 
-	repoFn    session.RepositoryFactory
-	iamRepoFn common.IamRepoFactory
+	repoFn      session.RepositoryFactory
+	iamRepoFn   common.IamRepoFactory
+	maxPageSize uint
 }
 
 var _ pbs.SessionServiceServer = (*Service)(nil)
 
 // NewService returns a session service which handles session related requests to boundary.
-func NewService(ctx context.Context, repoFn session.RepositoryFactory, iamRepoFn common.IamRepoFactory) (Service, error) {
+func NewService(ctx context.Context, repoFn session.RepositoryFactory, iamRepoFn common.IamRepoFactory, maxPageSize uint) (Service, error) {
 	const op = "sessions.NewService"
 	if repoFn == nil {
 		return Service{}, errors.New(ctx, errors.InvalidParameter, op, "missing session repository")
@@ -68,7 +71,10 @@ func NewService(ctx context.Context, repoFn session.RepositoryFactory, iamRepoFn
 	if iamRepoFn == nil {
 		return Service{}, errors.New(ctx, errors.InvalidParameter, op, "missing iam repository")
 	}
-	return Service{repoFn: repoFn, iamRepoFn: iamRepoFn}, nil
+	if maxPageSize == 0 {
+		maxPageSize = uint(globals.DefaultMaxPageSize)
+	}
+	return Service{repoFn: repoFn, iamRepoFn: iamRepoFn, maxPageSize: maxPageSize}, nil
 }
 
 // GetSessions implements the interface pbs.SessionServiceServer.
@@ -130,7 +136,7 @@ func (s Service) ListSessions(ctx context.Context, req *pbs.ListSessionsRequest)
 	const op = "session.(Service).ListSessions"
 
 	if err := validateListRequest(ctx, req); err != nil {
-		return nil, err
+		return nil, errors.Wrap(ctx, err, op)
 	}
 
 	authResults := s.authResult(ctx, req.GetScopeId(), action.List, false)
@@ -143,13 +149,12 @@ func (s Service) ListSessions(ctx context.Context, req *pbs.ListSessionsRequest)
 			req.GetRecursive() &&
 			authResults.AuthenticationFinished {
 		} else {
-			return nil, authResults.Error
+			return nil, errors.Wrap(ctx, authResults.Error, op)
 		}
 	}
 
 	var scopeIds map[string]*scopes.ScopeInfo
 	var err error
-
 	if !req.GetRecursive() {
 		scopeIds = map[string]*scopes.ScopeInfo{authResults.Scope.Id: authResults.Scope}
 	} else {
@@ -159,6 +164,42 @@ func (s Service) ListSessions(ctx context.Context, req *pbs.ListSessionsRequest)
 		}
 	}
 
+	pageSize := int(s.maxPageSize)
+	// Use the requested page size only if it is smaller than
+	// the configured max.
+	if req.GetPageSize() != 0 && uint(req.GetPageSize()) < s.maxPageSize {
+		pageSize = int(req.GetPageSize())
+	}
+	var filterItemFn func(ctx context.Context, item *session.Session) (bool, error)
+	switch {
+	case req.GetFilter() != "":
+		// Only use a filter if we need to
+		filter, err := handlers.NewFilter(ctx, req.GetFilter())
+		if err != nil {
+			return nil, err
+		}
+		filterItemFn = func(ctx context.Context, item *session.Session) (bool, error) {
+			outputOpts, ok := newOutputOpts(ctx, item, scopeIds, authResults)
+			if !ok {
+				return false, nil
+			}
+			pbItem, err := toProto(ctx, item, outputOpts...)
+			if err != nil {
+				return false, err
+			}
+			return filter.Match(pbItem), nil
+		}
+	default:
+		filterItemFn = func(ctx context.Context, item *session.Session) (bool, error) {
+			return true, nil
+		}
+	}
+
+	grantsHash, err := authResults.GrantsHash(ctx)
+	if err != nil {
+		return nil, errors.Wrap(ctx, err, op)
+	}
+
 	listPerms := authResults.ACL().ListPermissions(scopeIds, resource.Session, IdActions, authResults.UserId)
 
 	repo, err := s.repoFn(session.WithPermissions(&perms.UserPermissions{
@@ -169,51 +210,76 @@ func (s Service) ListSessions(ctx context.Context, req *pbs.ListSessionsRequest)
 		return nil, errors.Wrap(ctx, err, op)
 	}
 
-	sesList, err := repo.ListSessions(ctx, session.WithTerminated(req.GetIncludeTerminated()), session.WithLimit(-1))
-	if err != nil {
-		return nil, err
-	}
-	if len(sesList) == 0 {
-		return &pbs.ListSessionsResponse{}, nil
-	}
-
-	filter, err := handlers.NewFilter(ctx, req.GetFilter())
-	if err != nil {
-		return nil, err
-	}
-	finalItems := make([]*pb.Session, 0, len(sesList))
-	res := perms.Resource{
-		Type: resource.Session,
-	}
-	for _, item := range sesList {
-		res.Id = item.GetPublicId()
-		res.ScopeId = item.GetProjectId()
-		authorizedActions := authResults.FetchActionSetForId(ctx, item.GetPublicId(), IdActions, auth.WithResource(&res)).Strings()
-		if len(authorizedActions) == 0 {
-			continue
-		}
-
-		outputFields := authResults.FetchOutputFields(res, action.List).SelfOrDefaults(authResults.UserId)
-		outputOpts := make([]handlers.Option, 0, 3)
-		outputOpts = append(outputOpts, handlers.WithOutputFields(outputFields))
-		if outputFields.Has(globals.ScopeField) {
-			outputOpts = append(outputOpts, handlers.WithScope(scopeIds[item.ProjectId]))
-		}
-		if outputFields.Has(globals.AuthorizedActionsField) {
-			outputOpts = append(outputOpts, handlers.WithAuthorizedActions(authorizedActions))
-		}
-
-		item, err := toProto(ctx, item, outputOpts...)
+	var listResp *pagination.ListResponse[*session.Session]
+	var sortBy string
+	includeTerminated := req.GetIncludeTerminated()
+	if req.GetListToken() == "" {
+		sortBy = "created_time"
+		listResp, err = session.List(ctx, grantsHash, pageSize, filterItemFn, repo, includeTerminated)
 		if err != nil {
 			return nil, err
 		}
-
-		if filter.Match(item) {
-			finalItems = append(finalItems, item)
+	} else {
+		listToken, err := handlers.ParseListToken(ctx, req.GetListToken(), resource.Session, grantsHash)
+		if err != nil {
+			return nil, err
+		}
+		switch st := listToken.Subtype.(type) {
+		case *listtoken.PaginationToken:
+			sortBy = "created_time"
+			listResp, err = session.ListPage(ctx, grantsHash, pageSize, filterItemFn, listToken, repo, includeTerminated)
+			if err != nil {
+				return nil, err
+			}
+		case *listtoken.StartRefreshToken:
+			sortBy = "updated_time"
+			listResp, err = session.ListRefresh(ctx, grantsHash, pageSize, filterItemFn, listToken, repo, includeTerminated)
+			if err != nil {
+				return nil, err
+			}
+		case *listtoken.RefreshToken:
+			sortBy = "updated_time"
+			listResp, err = session.ListRefreshPage(ctx, grantsHash, pageSize, filterItemFn, listToken, repo, includeTerminated)
+			if err != nil {
+				return nil, err
+			}
+		default:
+			return nil, handlers.ApiErrorWithCodeAndMessage(codes.InvalidArgument, "unexpected list token subtype: %T", st)
 		}
 	}
 
-	return &pbs.ListSessionsResponse{Items: finalItems}, nil
+	finalItems := make([]*pb.Session, 0, len(listResp.Items))
+	for _, item := range listResp.Items {
+		outputOpts, ok := newOutputOpts(ctx, item, scopeIds, authResults)
+		if !ok {
+			continue
+		}
+		item, err := toProto(ctx, item, outputOpts...)
+		if err != nil {
+			return nil, errors.Wrap(ctx, err, op)
+		}
+		finalItems = append(finalItems, item)
+	}
+	respType := "delta"
+	if listResp.CompleteListing {
+		respType = "complete"
+	}
+	resp := &pbs.ListSessionsResponse{
+		Items:        finalItems,
+		EstItemCount: uint32(listResp.EstimatedItemCount),
+		RemovedIds:   listResp.DeletedIds,
+		ResponseType: respType,
+		SortBy:       sortBy,
+		SortDir:      "desc",
+	}
+
+	if listResp.ListToken != nil {
+		resp.ListToken, err = handlers.MarshalListToken(ctx, listResp.ListToken, pbs.ResourceType_RESOURCE_TYPE_SESSION)
+		if err != nil {
+			return nil, err
+		}
+	}
+	return resp, nil
 }
 
 // CancelSession implements the interface pbs.SessionServiceServer.
@@ -506,3 +572,26 @@ func validateCancelRequest(req *pbs.CancelSessionRequest) error {
 	}
 	return nil
 }
+
+func newOutputOpts(ctx context.Context, item *session.Session, scopeIds map[string]*scopes.ScopeInfo, authResults auth.VerifyResults) ([]handlers.Option, bool) {
+	res := perms.Resource{
+		Type:    resource.Session,
+		Id:      item.GetPublicId(),
+		ScopeId: item.GetProjectId(),
+	}
+	authorizedActions := authResults.FetchActionSetForId(ctx, item.GetPublicId(), IdActions, auth.WithResource(&res)).Strings()
+	if len(authorizedActions) == 0 {
+		return nil, false
+	}
+
+	outputFields := authResults.FetchOutputFields(res, action.List).SelfOrDefaults(authResults.UserId)
+	outputOpts := make([]handlers.Option, 0, 3)
+	outputOpts = append(outputOpts, handlers.WithOutputFields(outputFields))
+	if outputFields.Has(globals.ScopeField) {
+		outputOpts = append(outputOpts, handlers.WithScope(scopeIds[item.ProjectId]))
+	}
+	if outputFields.Has(globals.AuthorizedActionsField) {
+		outputOpts = append(outputOpts, handlers.WithAuthorizedActions(authorizedActions))
+	}
+	return outputOpts, true
+}

internal/daemon/controller/handlers/sessions/session_service_test.go

@@ -34,6 +34,7 @@ import (
 	pb "github.com/hashicorp/boundary/sdk/pbs/controller/api/resources/sessions"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"golang.org/x/exp/slices"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/protobuf/testing/protocmp"
 )
@@ -139,7 +140,7 @@ func TestGetSession(t *testing.T) {
 		t.Run(tc.name, func(t *testing.T) {
 			assert, require := assert.New(t), require.New(t)
 
-			s, err := sessions.NewService(ctx, sessRepoFn, iamRepoFn)
+			s, err := sessions.NewService(ctx, sessRepoFn, iamRepoFn, 1000)
 			require.NoError(err, "Couldn't create new session service.")
 
 			requestInfo := authpb.RequestInfo{
@@ -233,7 +234,7 @@ func TestList_Self(t *testing.T) {
 		Endpoint:    "tcp://127.0.0.1:22",
 	})
 
-	s, err := sessions.NewService(ctx, sessRepoFn, iamRepoFn)
+	s, err := sessions.NewService(ctx, sessRepoFn, iamRepoFn, 1000)
 	require.NoError(t, err, "Couldn't create new session service.")
 
 	cases := []struct {
@@ -283,6 +284,9 @@ func TestList(t *testing.T) {
 	kms := kms.TestKms(t, conn, wrap)
 	ctx := context.Background()
 
+	sqlDB, err := conn.SqlDB(ctx)
+	require.NoError(t, err)
+
 	iamRepo := iam.TestRepo(t, conn, wrap)
 
 	rw := db.New(conn)
@@ -456,6 +460,16 @@ func TestList(t *testing.T) {
 		wantIncludeTerminatedSessions = append(wantIncludeTerminatedSessions, expected)
 	}
 
+	// Run analyze to update postgres estimates
+	_, err = sqlDB.ExecContext(ctx, "analyze")
+	require.NoError(t, err)
+
+	// Reverse slices since response is ordered by created_time descending (newest first)
+	slices.Reverse(wantSession)
+	slices.Reverse(wantOtherSession)
+	slices.Reverse(wantAllSessions)
+	slices.Reverse(wantIncludeTerminatedSessions)
+
 	cases := []struct {
 		name          string
 		req           *pbs.ListSessionsRequest
@@ -465,38 +479,92 @@ func TestList(t *testing.T) {
 		err           error
 	}{
 		{
-			name:          "List Many Sessions",
-			req:           &pbs.ListSessionsRequest{ScopeId: pWithSessions.GetPublicId()},
-			res:           &pbs.ListSessionsResponse{Items: wantSession},
-			otherRes:      &pbs.ListSessionsResponse{Items: []*pb.Session{}},
+			name: "List Many Sessions",
+			req:  &pbs.ListSessionsRequest{ScopeId: pWithSessions.GetPublicId()},
+			res: &pbs.ListSessionsResponse{
+				Items:        wantSession,
+				ResponseType: "complete",
+				SortBy:       "created_time",
+				SortDir:      "desc",
+				EstItemCount: 10,
+			},
+			otherRes: &pbs.ListSessionsResponse{
+				Items:        []*pb.Session{},
+				ResponseType: "complete",
+				SortBy:       "created_time",
+				SortDir:      "desc",
+			},
 			allSessionRes: &pbs.ListSessionsResponse{Items: wantSession},
 		},
 		{
-			name:          "List Many Include Terminated",
-			req:           &pbs.ListSessionsRequest{ScopeId: pWithSessions.GetPublicId(), IncludeTerminated: true},
-			res:           &pbs.ListSessionsResponse{Items: wantIncludeTerminatedSessions},
-			otherRes:      &pbs.ListSessionsResponse{Items: []*pb.Session{}},
+			name: "List Many Include Terminated",
+			req:  &pbs.ListSessionsRequest{ScopeId: pWithSessions.GetPublicId(), IncludeTerminated: true},
+			res: &pbs.ListSessionsResponse{
+				Items:        wantIncludeTerminatedSessions,
+				ResponseType: "complete",
+				SortBy:       "created_time",
+				SortDir:      "desc",
+				EstItemCount: 11,
+			},
+			otherRes: &pbs.ListSessionsResponse{
+				Items:        []*pb.Session{},
+				ResponseType: "complete",
+				SortBy:       "created_time",
+				SortDir:      "desc",
+			},
 			allSessionRes: &pbs.ListSessionsResponse{Items: wantIncludeTerminatedSessions},
 		},
 		{
-			name:          "List No Sessions",
-			req:           &pbs.ListSessionsRequest{ScopeId: pNoSessions.GetPublicId()},
-			res:           &pbs.ListSessionsResponse{},
-			otherRes:      &pbs.ListSessionsResponse{Items: []*pb.Session{}},
+			name: "List No Sessions",
+			req:  &pbs.ListSessionsRequest{ScopeId: pNoSessions.GetPublicId()},
+			res: &pbs.ListSessionsResponse{
+				ResponseType: "complete",
+				SortBy:       "created_time",
+				SortDir:      "desc",
+			},
+			otherRes: &pbs.ListSessionsResponse{
+				Items:        []*pb.Session{},
+				ResponseType: "complete",
+				SortBy:       "created_time",
+				SortDir:      "desc",
+			},
 			allSessionRes: &pbs.ListSessionsResponse{},
 		},
 		{
-			name:          "List Sessions Recursively",
-			req:           &pbs.ListSessionsRequest{ScopeId: scope.Global.String(), Recursive: true},
-			res:           &pbs.ListSessionsResponse{Items: wantSession},
-			otherRes:      &pbs.ListSessionsResponse{Items: wantOtherSession},
+			name: "List Sessions Recursively",
+			req:  &pbs.ListSessionsRequest{ScopeId: scope.Global.String(), Recursive: true},
+			res: &pbs.ListSessionsResponse{
+				Items:        wantSession,
+				ResponseType: "complete",
+				SortBy:       "created_time",
+				SortDir:      "desc",
+				EstItemCount: 10,
+			},
+			otherRes: &pbs.ListSessionsResponse{
+				Items:        wantOtherSession,
+				ResponseType: "complete",
+				SortBy:       "created_time",
+				SortDir:      "desc",
+				EstItemCount: 10,
+			},
 			allSessionRes: &pbs.ListSessionsResponse{Items: wantAllSessions},
 		},
 		{
-			name:          "Filter To Single Sessions",
-			req:           &pbs.ListSessionsRequest{ScopeId: pWithSessions.GetPublicId(), Filter: fmt.Sprintf(`"/item/id"==%q`, wantSession[4].Id)},
-			res:           &pbs.ListSessionsResponse{Items: wantSession[4:5]},
-			otherRes:      &pbs.ListSessionsResponse{Items: []*pb.Session{}},
+			name: "Filter To Single Sessions",
+			req:  &pbs.ListSessionsRequest{ScopeId: pWithSessions.GetPublicId(), Filter: fmt.Sprintf(`"/item/id"==%q`, wantSession[4].Id)},
+			res: &pbs.ListSessionsResponse{
+				Items:        wantSession[4:5],
+				ResponseType: "complete",
+				SortBy:       "created_time",
+				SortDir:      "desc",
+				EstItemCount: 1,
+			},
+			otherRes: &pbs.ListSessionsResponse{
+				Items:        []*pb.Session{},
+				ResponseType: "complete",
+				SortBy:       "created_time",
+				SortDir:      "desc",
+			},
 			allSessionRes: &pbs.ListSessionsResponse{Items: wantSession[4:5]},
 		},
 		{
@@ -505,17 +573,55 @@ func TestList(t *testing.T) {
 				ScopeId: scope.Global.String(), Recursive: true,
 				Filter: fmt.Sprintf(`"/item/scope/id" matches "^%s"`, pWithSessions.GetPublicId()[:8]),
 			},
-			res:           &pbs.ListSessionsResponse{Items: wantSession},
-			otherRes:      &pbs.ListSessionsResponse{Items: []*pb.Session{}},
+			res: &pbs.ListSessionsResponse{
+				Items:        wantSession,
+				ResponseType: "complete",
+				SortBy:       "created_time",
+				SortDir:      "desc",
+				EstItemCount: 10,
+			},
+			otherRes: &pbs.ListSessionsResponse{
+				Items:        []*pb.Session{},
+				ResponseType: "complete",
+				SortBy:       "created_time",
+				SortDir:      "desc",
+			},
 			allSessionRes: &pbs.ListSessionsResponse{Items: wantSession},
 		},
 		{
-			name:          "Filter To Nothing",
-			req:           &pbs.ListSessionsRequest{ScopeId: pWithSessions.GetPublicId(), Filter: `"/item/id" == ""`},
-			res:           &pbs.ListSessionsResponse{},
-			otherRes:      &pbs.ListSessionsResponse{Items: []*pb.Session{}},
+			name: "Filter To Nothing",
+			req:  &pbs.ListSessionsRequest{ScopeId: pWithSessions.GetPublicId(), Filter: `"/item/id" == ""`},
+			res: &pbs.ListSessionsResponse{
+				ResponseType: "complete",
+				SortBy:       "created_time",
+				SortDir:      "desc",
+			},
+			otherRes: &pbs.ListSessionsResponse{
+				Items:        []*pb.Session{},
+				ResponseType: "complete",
+				SortBy:       "created_time",
+				SortDir:      "desc",
+			},
 			allSessionRes: &pbs.ListSessionsResponse{},
 		},
+		{
+			name: "Paginate listing",
+			req:  &pbs.ListSessionsRequest{ScopeId: pWithSessions.GetPublicId(), Recursive: true, PageSize: 2},
+			res: &pbs.ListSessionsResponse{
+				Items:        wantSession[:2],
+				ResponseType: "delta",
+				SortBy:       "created_time",
+				SortDir:      "desc",
+				EstItemCount: 21,
+			},
+			otherRes: &pbs.ListSessionsResponse{
+				Items:        []*pb.Session{},
+				ResponseType: "complete",
+				SortBy:       "created_time",
+				SortDir:      "desc",
+			},
+			allSessionRes: &pbs.ListSessionsResponse{Items: wantSession[:2]},
+		},
 		{
 			name:     "Filter Bad Format",
 			req:      &pbs.ListSessionsRequest{ScopeId: pWithSessions.GetPublicId(), Filter: `//badformat/`},
@@ -526,7 +632,7 @@ func TestList(t *testing.T) {
 	for _, tc := range cases {
 		t.Run(tc.name, func(t *testing.T) {
 			require, assert := require.New(t), assert.New(t)
-			s, err := sessions.NewService(ctx, sessRepoFn, iamRepoFn)
+			s, err := sessions.NewService(ctx, sessRepoFn, iamRepoFn, 1000)
 			require.NoError(err, "Couldn't create new session service.")
 
 			// Test without anon user
@@ -545,7 +651,7 @@ func TestList(t *testing.T) {
 			}
 			require.NoError(gErr)
 			if tc.res != nil {
-				require.Equal(len(tc.res.GetItems()), len(got.GetItems()), "Didn't get expected number of sessions: %v", got.GetItems())
+				require.Equal(len(tc.res.GetItems()), len(got.GetItems()), "Didn't get expected number of sessions: %v", len(got.GetItems()))
 				for i, wantSess := range tc.res.GetItems() {
 					assert.True(got.GetItems()[i].GetExpirationTime().AsTime().Sub(wantSess.GetExpirationTime().AsTime()) < 10*time.Millisecond)
 					assert.Equal(0, len(wantSess.GetConnections())) // no connections on list
@@ -605,6 +711,338 @@ func TestList(t *testing.T) {
 	}
 }
 
+func TestListPagination(t *testing.T) {
+	// Set database read timeout to avoid duplicates in response
+	oldReadTimeout := globals.RefreshReadLookbackDuration
+	globals.RefreshReadLookbackDuration = 0
+	t.Cleanup(func() {
+		globals.RefreshReadLookbackDuration = oldReadTimeout
+	})
+	conn, _ := db.TestSetup(t, "postgres")
+	wrap := db.TestWrapper(t)
+	kms := kms.TestKms(t, conn, wrap)
+	ctx := context.Background()
+
+	sqlDB, err := conn.SqlDB(ctx)
+	require.NoError(t, err)
+
+	iamRepo := iam.TestRepo(t, conn, wrap)
+
+	rw := db.New(conn)
+	sessRepo, err := session.NewRepository(ctx, rw, rw, kms)
+	require.NoError(t, err)
+
+	iamRepoFn := func() (*iam.Repository, error) {
+		return iamRepo, nil
+	}
+	sessRepoFn := func(opt ...session.Option) (*session.Repository, error) {
+		return session.NewRepository(ctx, rw, rw, kms, opt...)
+	}
+	tokenRepoFn := func() (*authtoken.Repository, error) {
+		return authtoken.NewRepository(ctx, rw, rw, kms)
+	}
+	serversRepoFn := func() (*server.Repository, error) {
+		return server.NewRepository(ctx, rw, rw, kms)
+	}
+
+	o, pWithSessions := iam.TestScopes(t, iamRepo)
+
+	at := authtoken.TestAuthToken(t, conn, kms, o.GetPublicId())
+	uId := at.GetIamUserId()
+
+	hc := static.TestCatalogs(t, conn, pWithSessions.GetPublicId(), 1)[0]
+	hs := static.TestSets(t, conn, hc.GetPublicId(), 1)[0]
+	h := static.TestHosts(t, conn, hc.GetPublicId(), 1)[0]
+	static.TestSetMembers(t, conn, hs.GetPublicId(), []*static.Host{h})
+	tar := tcp.TestTarget(ctx, t, conn, pWithSessions.GetPublicId(), "test", target.WithHostSources([]string{hs.GetPublicId()}))
+
+	var allSessions []*pb.Session
+	for i := 0; i < 10; i++ {
+		sess := session.TestSession(t, conn, wrap, session.ComposedOf{
+			UserId:      uId,
+			HostId:      h.GetPublicId(),
+			TargetId:    tar.GetPublicId(),
+			HostSetId:   hs.GetPublicId(),
+			AuthTokenId: at.GetPublicId(),
+			ProjectId:   pWithSessions.GetPublicId(),
+			Endpoint:    "tcp://127.0.0.1:22",
+		})
+
+		session.TestConnection(t, conn, sess.PublicId, "127.0.0.1", 22, "127.0.0.2", 23, "127.0.0.1")
+
+		status, states := convertStates(sess.States)
+
+		firstOrgSession := &pb.Session{
+			Id:                sess.GetPublicId(),
+			ScopeId:           pWithSessions.GetPublicId(),
+			AuthTokenId:       at.GetPublicId(),
+			UserId:            at.GetIamUserId(),
+			TargetId:          sess.TargetId,
+			Endpoint:          sess.Endpoint,
+			HostSetId:         sess.HostSetId,
+			HostId:            sess.HostId,
+			Version:           sess.Version,
+			UpdatedTime:       sess.UpdateTime.GetTimestamp(),
+			CreatedTime:       sess.CreateTime.GetTimestamp(),
+			ExpirationTime:    sess.ExpirationTime.GetTimestamp(),
+			Scope:             &scopes.ScopeInfo{Id: pWithSessions.GetPublicId(), Type: scope.Project.String(), ParentScopeId: o.GetPublicId()},
+			Status:            status,
+			States:            states,
+			Certificate:       sess.Certificate,
+			Type:              tcp.Subtype.String(),
+			AuthorizedActions: testAuthorizedActions,
+			Connections:       []*pb.Connection{}, // connections should not be returned for list
+		}
+		allSessions = append(allSessions, firstOrgSession)
+	}
+	// Reverse slices since response is ordered by created_time descending (newest first)
+	slices.Reverse(allSessions)
+
+	// Run analyze to update postgres estimates
+	_, err = sqlDB.ExecContext(ctx, "analyze")
+	require.NoError(t, err)
+
+	requestInfo := authpb.RequestInfo{
+		TokenFormat: uint32(auth.AuthTokenTypeBearer),
+		PublicId:    at.GetPublicId(),
+		Token:       at.GetToken(),
+	}
+	requestContext := context.WithValue(context.Background(), requests.ContextRequestInformationKey, &requests.RequestContext{})
+	ctx = auth.NewVerifierContext(requestContext, iamRepoFn, tokenRepoFn, serversRepoFn, kms, &requestInfo)
+
+	s, err := sessions.NewService(ctx, sessRepoFn, iamRepoFn, 1000)
+	require.NoError(t, err, "Couldn't create new session service.")
+
+	// Start paginating, recursively
+	req := &pbs.ListSessionsRequest{
+		ScopeId:   "global",
+		Recursive: true,
+		Filter:    "",
+		ListToken: "",
+		PageSize:  2,
+	}
+	got, err := s.ListSessions(ctx, req)
+	require.NoError(t, err)
+	require.Len(t, got.GetItems(), 2)
+	// Compare without comparing the list token
+	assert.Empty(t,
+		cmp.Diff(
+			got,
+			&pbs.ListSessionsResponse{
+				Items:        allSessions[0:2],
+				ResponseType: "delta",
+				SortBy:       "created_time",
+				SortDir:      "desc",
+				RemovedIds:   nil,
+				EstItemCount: 10,
+			},
+			protocmp.Transform(),
+			protocmp.IgnoreFields(&pbs.ListSessionsResponse{}, "list_token"),
+		),
+	)
+
+	// Request second page
+	req.ListToken = got.ListToken
+	got, err = s.ListSessions(ctx, req)
+	require.NoError(t, err)
+	require.Len(t, got.GetItems(), 2)
+	// Compare without comparing the list token
+	assert.Empty(t,
+		cmp.Diff(
+			got,
+			&pbs.ListSessionsResponse{
+				Items:        allSessions[2:4],
+				ResponseType: "delta",
+				SortBy:       "created_time",
+				SortDir:      "desc",
+				RemovedIds:   nil,
+				EstItemCount: 10,
+			},
+			protocmp.Transform(),
+			protocmp.IgnoreFields(&pbs.ListSessionsResponse{}, "list_token"),
+		),
+	)
+
+	// Request rest of results
+	req.ListToken = got.ListToken
+	req.PageSize = 10
+	got, err = s.ListSessions(ctx, req)
+	require.NoError(t, err)
+	require.Len(t, got.GetItems(), 6)
+	// Compare without comparing the list token
+	assert.Empty(t,
+		cmp.Diff(
+			got,
+			&pbs.ListSessionsResponse{
+				Items:        allSessions[4:],
+				ResponseType: "complete",
+				SortBy:       "created_time",
+				SortDir:      "desc",
+				RemovedIds:   nil,
+				EstItemCount: 10,
+			},
+			protocmp.Transform(),
+			protocmp.IgnoreFields(&pbs.ListSessionsResponse{}, "list_token"),
+		),
+	)
+
+	// Create another session
+	sess := session.TestSession(t, conn, wrap, session.ComposedOf{
+		UserId:      uId,
+		HostId:      h.GetPublicId(),
+		TargetId:    tar.GetPublicId(),
+		HostSetId:   hs.GetPublicId(),
+		AuthTokenId: at.GetPublicId(),
+		ProjectId:   pWithSessions.GetPublicId(),
+		Endpoint:    "tcp://127.0.0.1:22",
+	})
+
+	session.TestConnection(t, conn, sess.PublicId, "127.0.0.1", 22, "127.0.0.2", 23, "127.0.0.1")
+
+	status, states := convertStates(sess.States)
+
+	newSession := &pb.Session{
+		Id:                sess.GetPublicId(),
+		ScopeId:           pWithSessions.GetPublicId(),
+		AuthTokenId:       at.GetPublicId(),
+		UserId:            at.GetIamUserId(),
+		TargetId:          sess.TargetId,
+		Endpoint:          sess.Endpoint,
+		HostSetId:         sess.HostSetId,
+		HostId:            sess.HostId,
+		Version:           sess.Version,
+		UpdatedTime:       sess.UpdateTime.GetTimestamp(),
+		CreatedTime:       sess.CreateTime.GetTimestamp(),
+		ExpirationTime:    sess.ExpirationTime.GetTimestamp(),
+		Scope:             &scopes.ScopeInfo{Id: pWithSessions.GetPublicId(), Type: scope.Project.String(), ParentScopeId: o.GetPublicId()},
+		Status:            status,
+		States:            states,
+		Certificate:       sess.Certificate,
+		Type:              tcp.Subtype.String(),
+		AuthorizedActions: testAuthorizedActions,
+		Connections:       []*pb.Connection{}, // connections should not be returned for list
+	}
+	// Add to the front since it's most recently updated
+	allSessions = append([]*pb.Session{newSession}, allSessions...)
+
+	// Delete one of the other sessions
+	_, err = sessRepo.DeleteSession(ctx, allSessions[len(allSessions)-1].Id)
+	require.NoError(t, err)
+	deletedSession := allSessions[len(allSessions)-1]
+	allSessions = allSessions[:len(allSessions)-1]
+
+	// cancel session in lieu of updating it
+	canceledSession, err := sessRepo.CancelSession(ctx, allSessions[1].Id, allSessions[1].Version)
+	require.NoError(t, err)
+	canceledStatus, canceledStates := convertStates(canceledSession.States)
+	allSessions[1].Status = canceledStatus
+	allSessions[1].States = canceledStates
+	allSessions[1].UpdatedTime = canceledSession.UpdateTime.GetTimestamp()
+	allSessions[1].Version = 2
+	allSessions = append(
+		[]*pb.Session{allSessions[1]},
+		append(
+			[]*pb.Session{allSessions[0]},
+			allSessions[2:]...,
+		)...,
+	)
+
+	// Run analyze to update postgres estimates
+	_, err = sqlDB.ExecContext(ctx, "analyze")
+	require.NoError(t, err)
+
+	// Request updated results
+	req.ListToken = got.ListToken
+	req.PageSize = 1
+	got, err = s.ListSessions(ctx, req)
+	require.NoError(t, err)
+	require.Len(t, got.GetItems(), 1)
+	// Compare without comparing the list token
+	assert.Empty(t,
+		cmp.Diff(
+			got,
+			&pbs.ListSessionsResponse{
+				Items:        []*pb.Session{allSessions[0]},
+				ResponseType: "delta",
+				SortBy:       "updated_time",
+				SortDir:      "desc",
+				// Should contain the deleted session
+				RemovedIds:   []string{deletedSession.Id},
+				EstItemCount: 10,
+			},
+			protocmp.Transform(),
+			protocmp.IgnoreFields(&pbs.ListSessionsResponse{}, "list_token"),
+		),
+	)
+
+	// Get next page
+	req.ListToken = got.ListToken
+	got, err = s.ListSessions(ctx, req)
+	require.NoError(t, err)
+	require.Len(t, got.GetItems(), 1)
+	// Compare without comparing the list token
+	assert.Empty(t,
+		cmp.Diff(
+			got,
+			&pbs.ListSessionsResponse{
+				Items:        []*pb.Session{allSessions[1]},
+				ResponseType: "complete",
+				SortBy:       "updated_time",
+				SortDir:      "desc",
+				RemovedIds:   nil,
+				EstItemCount: 10,
+			},
+			protocmp.Transform(),
+			protocmp.IgnoreFields(&pbs.ListSessionsResponse{}, "list_token"),
+		),
+	)
+
+	// Request new page with filter requiring looping
+	// to fill the page.
+	req.ListToken = ""
+	req.PageSize = 1
+	req.Filter = fmt.Sprintf(`"/item/id"==%q or "/item/id"==%q`, allSessions[len(allSessions)-2].Id, allSessions[len(allSessions)-1].Id)
+	got, err = s.ListSessions(ctx, req)
+	require.NoError(t, err)
+	require.Len(t, got.GetItems(), 1)
+	assert.Empty(t,
+		cmp.Diff(
+			got,
+			&pbs.ListSessionsResponse{
+				Items:        []*pb.Session{allSessions[len(allSessions)-2]},
+				ResponseType: "delta",
+				SortBy:       "created_time",
+				SortDir:      "desc",
+				// Should be empty again
+				RemovedIds:   nil,
+				EstItemCount: 10,
+			},
+			protocmp.Transform(),
+			protocmp.IgnoreFields(&pbs.ListSessionsResponse{}, "list_token"),
+		),
+	)
+	req.ListToken = got.ListToken
+	// Get the second page
+	got, err = s.ListSessions(ctx, req)
+	require.NoError(t, err)
+	require.Len(t, got.GetItems(), 1)
+	assert.Empty(t,
+		cmp.Diff(
+			got,
+			&pbs.ListSessionsResponse{
+				Items:        []*pb.Session{allSessions[len(allSessions)-1]},
+				ResponseType: "complete",
+				SortBy:       "created_time",
+				SortDir:      "desc",
+				RemovedIds:   nil,
+				EstItemCount: 10,
+			},
+			protocmp.Transform(),
+			protocmp.IgnoreFields(&pbs.ListSessionsResponse{}, "list_token"),
+		),
+	)
+}
+
 func convertStates(in []*session.State) (string, []*pb.SessionState) {
 	out := make([]*pb.SessionState, 0, len(in))
 	for _, s := range in {
@@ -727,7 +1165,7 @@ func TestCancel(t *testing.T) {
 		t.Run(tc.name, func(t *testing.T) {
 			assert, require := assert.New(t), require.New(t)
 
-			s, err := sessions.NewService(ctx, sessRepoFn, iamRepoFn)
+			s, err := sessions.NewService(ctx, sessRepoFn, iamRepoFn, 1000)
 			require.NoError(err, "Couldn't create new session service.")
 
 			tc.req.Version = version