Repository: bacalhau-project/bacalhau (language: python)

Issue:
Support Pagination for fetching History from JobStore



Discussion / hints:

Fix patch:
diff --git a/pkg/jobstore/boltdb/store.go b/pkg/jobstore/boltdb/store.go
index f1cd3845d7..ca87dd6b99 100644
--- a/pkg/jobstore/boltdb/store.go
+++ b/pkg/jobstore/boltdb/store.go
@@ -407,7 +407,6 @@ func (b *BoltJobStore) getJobs(tx *bolt.Tx, query jobstore.JobQuery) (*jobstore.
 		Limit:  query.Limit,
 	}
 
-	// If we don't have 'limit' jobs, then there definitely aren't any more
 	if more {
 		response.NextOffset = query.Offset + query.Limit
 	}
@@ -415,7 +414,7 @@ func (b *BoltJobStore) getJobs(tx *bolt.Tx, query jobstore.JobQuery) (*jobstore.
 	return response, nil
 }
 
-// getJobsWithinLimit returns the initial set of jobs to be considered for GetJobs response.
+// getJobsInitialSet returns the initial set of jobs to be considered for GetJobs response.
 // It either returns all jobs, or jobs for a specific client if specified in the query.
 func (b *BoltJobStore) getJobsInitialSet(tx *bolt.Tx, query jobstore.JobQuery) (map[string]struct{}, error) {
 	jobSet := make(map[string]struct{})
@@ -629,28 +628,29 @@ func createInProgressIndexKey(job *models.Job) string {
 // GetJobHistory returns the job (and execution) history for the provided options
 func (b *BoltJobStore) GetJobHistory(ctx context.Context,
 	jobID string,
-	options jobstore.JobHistoryFilterOptions,
-) ([]models.JobHistory, error) {
-	var history []models.JobHistory
+	query jobstore.JobHistoryQuery,
+) (*jobstore.JobHistoryQueryResponse, error) {
+	var response *jobstore.JobHistoryQueryResponse
 	err := b.view(ctx, func(tx *bolt.Tx) (err error) {
-		history, err = b.getJobHistory(tx, jobID, options)
+		response, err = b.getJobHistory(tx, jobID, query)
 		return
 	})
 
-	return history, err
+	return response, err
 }
 
+//nolint:gocyclo,funlen
 func (b *BoltJobStore) getJobHistory(tx *bolt.Tx, jobID string,
-	options jobstore.JobHistoryFilterOptions,
-) ([]models.JobHistory, error) {
+	query jobstore.JobHistoryQuery,
+) (*jobstore.JobHistoryQueryResponse, error) {
 	var history []models.JobHistory
 
 	jobID, err := b.reifyJobID(tx, jobID)
 	if err != nil {
-		return history, err
+		return nil, err
 	}
 
-	if !options.ExcludeJobLevel {
+	if !query.ExcludeJobLevel {
 		if bkt, err := NewBucketPath(BucketJobs, jobID, BucketJobHistory).Get(tx, false); err != nil {
 			return nil, err
 		} else {
@@ -671,7 +671,7 @@ func (b *BoltJobStore) getJobHistory(tx *bolt.Tx, jobID string,
 		}
 	}
 
-	if !options.ExcludeExecutionLevel {
+	if !query.ExcludeExecutionLevel {
 		// 	// Get the executions for this JobID
 		if bkt, err := NewBucketPath(BucketJobs, jobID, BucketExecutionHistory).Get(tx, false); err != nil {
 			return nil, err
@@ -696,15 +696,15 @@ func (b *BoltJobStore) getJobHistory(tx *bolt.Tx, jobID string,
 	// Filter out anything before the specified Since time, and anything that doesn't match the
 	// specified ExecutionID or NodeID
 	history = lo.Filter(history, func(event models.JobHistory, index int) bool {
-		if options.ExecutionID != "" && !strings.HasPrefix(event.ExecutionID, options.ExecutionID) {
+		if query.ExecutionID != "" && !strings.HasPrefix(event.ExecutionID, query.ExecutionID) {
 			return false
 		}
 
-		if options.NodeID != "" && !strings.HasPrefix(event.NodeID, options.NodeID) {
+		if query.NodeID != "" && !strings.HasPrefix(event.NodeID, query.NodeID) {
 			return false
 		}
 
-		if event.Time.Unix() < options.Since {
+		if event.Time.Unix() < query.Since {
 			return false
 		}
 		return true
@@ -712,7 +712,46 @@ func (b *BoltJobStore) getJobHistory(tx *bolt.Tx, jobID string,
 
 	sort.Slice(history, func(i, j int) bool { return history[i].Time.UTC().Before(history[j].Time.UTC()) })
 
-	return history, nil
+	offset := uint32(0)
+	if query.NextToken != "" {
+		token, err := models.NewPagingTokenFromString(query.NextToken)
+		if err != nil {
+			return nil, err
+		}
+		offset = token.Offset
+	}
+
+	if offset >= uint32(len(history)) {
+		return &jobstore.JobHistoryQueryResponse{}, nil
+	}
+
+	historyFiltered := history[offset:]
+	if query.Limit == 0 {
+		return &jobstore.JobHistoryQueryResponse{
+			JobHistory: historyFiltered,
+		}, nil
+	}
+
+	limit := math.Min(uint32(len(historyFiltered)), query.Limit)
+	fileteredLength := uint32(len(historyFiltered))
+	historyFiltered = historyFiltered[:limit]
+
+	response := &jobstore.JobHistoryQueryResponse{
+		JobHistory: historyFiltered,
+		Offset:     offset,
+	}
+
+	if fileteredLength >= query.Limit {
+		response.NextToken = models.NewPagingToken(&models.PagingTokenParams{
+			Offset: offset + query.Limit,
+		}).String()
+	} else {
+		response.NextToken = models.NewPagingToken(&models.PagingTokenParams{
+			Offset: fileteredLength,
+		}).String()
+	}
+
+	return response, nil
 }
 
 // CreateJob creates a new record of a job in the data store
diff --git a/pkg/jobstore/mocks.go b/pkg/jobstore/mocks.go
index 3570d1b842..1d0f57eee5 100644
--- a/pkg/jobstore/mocks.go
+++ b/pkg/jobstore/mocks.go
@@ -304,10 +304,10 @@ func (mr *MockStoreMockRecorder) GetJob(ctx, id interface{}) *gomock.Call {
 }
 
 // GetJobHistory mocks base method.
-func (m *MockStore) GetJobHistory(ctx context.Context, jobID string, options JobHistoryFilterOptions) ([]models.JobHistory, error) {
+func (m *MockStore) GetJobHistory(ctx context.Context, jobID string, options JobHistoryQuery) (*JobHistoryQueryResponse, error) {
 	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "GetJobHistory", ctx, jobID, options)
-	ret0, _ := ret[0].([]models.JobHistory)
+	ret0, _ := ret[0].(*JobHistoryQueryResponse)
 	ret1, _ := ret[1].(error)
 	return ret0, ret1
 }
diff --git a/pkg/jobstore/types.go b/pkg/jobstore/types.go
index b1f5753ac4..0ef55fab79 100644
--- a/pkg/jobstore/types.go
+++ b/pkg/jobstore/types.go
@@ -30,6 +30,22 @@ type JobQueryResponse struct {
 	NextOffset uint32 // Offset + Limit of the next page of results, 0 means no more results
 }
 
+type JobHistoryQuery struct {
+	Since                 int64  `json:"since"`
+	Limit                 uint32 `json:"limit"`
+	ExcludeExecutionLevel bool   `json:"exclude_execution_level"`
+	ExcludeJobLevel       bool   `json:"exclude_job_level"`
+	ExecutionID           string `json:"execution_id"`
+	NodeID                string `json:"node_id"`
+	NextToken             string `json:"next_token"`
+}
+
+type JobHistoryQueryResponse struct {
+	JobHistory []models.JobHistory
+	Offset     uint32
+	NextToken  string
+}
+
 // TxContext is a transactional context that can be used to commit or rollback
 type TxContext interface {
 	context.Context
@@ -73,7 +89,7 @@ type Store interface {
 	// GetJobHistory retrieves the history for the specified job.  The
 	// history returned is filtered by the contents of the provided
 	// [JobHistoryFilterOptions].
-	GetJobHistory(ctx context.Context, jobID string, options JobHistoryFilterOptions) ([]models.JobHistory, error)
+	GetJobHistory(ctx context.Context, jobID string, options JobHistoryQuery) (*JobHistoryQueryResponse, error)
 
 	// CreateJob will create a new job and persist it in the store.
 	CreateJob(ctx context.Context, j models.Job, event models.Event) error
@@ -181,14 +197,6 @@ func (condition UpdateExecutionCondition) Validate(execution models.Execution) e
 	return nil
 }
 
-type JobHistoryFilterOptions struct {
-	Since                 int64  `json:"since"`
-	ExcludeExecutionLevel bool   `json:"exclude_execution_level"`
-	ExcludeJobLevel       bool   `json:"exclude_job_level"`
-	ExecutionID           string `json:"execution_id"`
-	NodeID                string `json:"node_id"`
-}
-
 type GetExecutionsOptions struct {
 	JobID      string `json:"job_id"`
 	IncludeJob bool   `json:"include_job"`
diff --git a/pkg/publicapi/endpoint/orchestrator/job.go b/pkg/publicapi/endpoint/orchestrator/job.go
index 2fe906f391..0f60c02cdd 100644
--- a/pkg/publicapi/endpoint/orchestrator/job.go
+++ b/pkg/publicapi/endpoint/orchestrator/job.go
@@ -92,7 +92,8 @@ func (e *Endpoint) getJob(c echo.Context) error { //nolint: gocyclo
 			if response.History != nil {
 				continue
 			}
-			history, err := e.store.GetJobHistory(ctx, jobID, jobstore.JobHistoryFilterOptions{})
+			jobHistoryQueryResponse, err := e.store.GetJobHistory(ctx, jobID, jobstore.JobHistoryQuery{})
+			history := jobHistoryQueryResponse.JobHistory
 			if err != nil {
 				return err
 			}
@@ -272,6 +273,7 @@ func (e *Endpoint) stopJob(c echo.Context) error {
 //	@Param			event_type		query		string	false	"Only return history of this event type"
 //	@Param			execution_id	query		string	false	"Only return history of this execution ID"
 //	@Param			node_id			query		string	false	"Only return history of this node ID"
+//	@Param			next_token		query		string	false	"Token to get the next page of the jobs"
 //	@Success		200				{object}	apimodels.ListJobHistoryResponse
 //	@Failure		400				{object}	string
 //	@Failure		500				{object}	string
@@ -287,22 +289,30 @@ func (e *Endpoint) jobHistory(c echo.Context) error {
 		return err
 	}
 
-	options := jobstore.JobHistoryFilterOptions{
+	options := jobstore.JobHistoryQuery{
 		Since:                 args.Since,
 		ExcludeExecutionLevel: args.EventType == "job",
 		ExcludeJobLevel:       args.EventType == "execution",
 		ExecutionID:           args.ExecutionID,
 		NodeID:                args.NodeID,
+		Limit:                 args.Limit,
+		NextToken:             args.NextToken,
 	}
-	history, err := e.store.GetJobHistory(ctx, jobID, options)
+
+	jobHistoryQueryResponse, err := e.store.GetJobHistory(ctx, jobID, options)
 	if err != nil {
 		return err
 	}
+
 	res := &apimodels.ListJobHistoryResponse{
-		Items: make([]*models.JobHistory, len(history)),
+		Items: make([]*models.JobHistory, len(jobHistoryQueryResponse.JobHistory)),
+		BaseListResponse: apimodels.BaseListResponse{
+			NextToken: jobHistoryQueryResponse.NextToken,
+		},
 	}
-	for i := range history {
-		res.Items[i] = &history[i]
+
+	for i := range jobHistoryQueryResponse.JobHistory {
+		res.Items[i] = &jobHistoryQueryResponse.JobHistory[i]
 	}
 
 	return c.JSON(http.StatusOK, res)


Test patch:
diff --git a/pkg/jobstore/boltdb/store_test.go b/pkg/jobstore/boltdb/store_test.go
index 6607b3f3f8..3972fbb63d 100644
--- a/pkg/jobstore/boltdb/store_test.go
+++ b/pkg/jobstore/boltdb/store_test.go
@@ -192,54 +192,94 @@ func (s *BoltJobstoreTestSuite) TearDownTest() {
 }
 
 func (s *BoltJobstoreTestSuite) TestUnfilteredJobHistory() {
-	history, err := s.store.GetJobHistory(s.ctx, "110", jobstore.JobHistoryFilterOptions{})
+	jobHistoryQueryResponse, err := s.store.GetJobHistory(s.ctx, "110", jobstore.JobHistoryQuery{})
 	s.Require().NoError(err, "failed to get job history")
-	s.Require().Equal(8, len(history))
+	s.Require().Equal(8, len(jobHistoryQueryResponse.JobHistory))
 
-	history, err = s.store.GetJobHistory(s.ctx, "11", jobstore.JobHistoryFilterOptions{})
+	jobHistoryQueryResponse, err = s.store.GetJobHistory(s.ctx, "11", jobstore.JobHistoryQuery{})
 	s.Require().NoError(err)
-	s.NotEmpty(history)
-	s.Require().Equal("110", history[0].JobID)
+	s.NotEmpty(jobHistoryQueryResponse)
+	s.Require().Equal("110", jobHistoryQueryResponse.JobHistory[0].JobID)
 
-	history, err = s.store.GetJobHistory(s.ctx, "1", jobstore.JobHistoryFilterOptions{})
+	jobHistoryQueryResponse, err = s.store.GetJobHistory(s.ctx, "1", jobstore.JobHistoryQuery{})
 	s.Require().Error(err)
 	s.Require().IsType(err, &bacerrors.MultipleJobsFound{})
-	s.Require().Nil(history)
+	s.Require().Nil(jobHistoryQueryResponse)
 }
 
 func (s *BoltJobstoreTestSuite) TestJobHistoryOrdering() {
-	history, err := s.store.GetJobHistory(s.ctx, "110", jobstore.JobHistoryFilterOptions{})
+	jobHistoryQueryResponse, err := s.store.GetJobHistory(s.ctx, "110", jobstore.JobHistoryQuery{})
 	require.NoError(s.T(), err, "failed to get job history")
 
 	// There are 6 history entries that we created directly, and 2 created by
 	// CreateJob and CreateExecution
-	require.Equal(s.T(), 8, len(history))
+	require.Equal(s.T(), 8, len(jobHistoryQueryResponse.JobHistory))
 
 	// Make sure they come back in order
-	values := make([]int64, len(history))
-	for i, h := range history {
+	values := make([]int64, len(jobHistoryQueryResponse.JobHistory))
+	for i, h := range jobHistoryQueryResponse.JobHistory {
 		values[i] = h.Time.Unix()
 	}
 
 	require.Equal(s.T(), []int64{1, 2, 3, 4, 5, 6, 7, 8}, values)
 }
 
+func (s *BoltJobstoreTestSuite) TestJobHistoryPagination() {
+	jobHistoryQueryResponse, err := s.store.GetJobHistory(s.ctx, "110", jobstore.JobHistoryQuery{
+		Limit: 2,
+	})
+	require.NoError(s.T(), err, "Failed to get job history")
+
+	require.NotEmpty(s.T(), jobHistoryQueryResponse.NextToken, "Next Token should be populated")
+	nextToken := jobHistoryQueryResponse.NextToken
+	require.Len(s.T(), jobHistoryQueryResponse.JobHistory, 2)
+
+	jobHistoryQueryResponse, err = s.store.GetJobHistory(s.ctx, "110", jobstore.JobHistoryQuery{
+		Limit:     1,
+		NextToken: nextToken,
+	})
+	require.NoError(s.T(), err, "Failed to get job history")
+
+	require.NotEmpty(s.T(), jobHistoryQueryResponse.NextToken, "Next Token should be populated")
+	nextToken = jobHistoryQueryResponse.NextToken
+	require.Len(s.T(), jobHistoryQueryResponse.JobHistory, 1)
+
+	jobHistoryQueryResponse, err = s.store.GetJobHistory(s.ctx, "110", jobstore.JobHistoryQuery{
+		Limit:     5,
+		NextToken: nextToken,
+	})
+	require.NoError(s.T(), err, "Failed to get job history")
+
+	require.NotEmpty(s.T(), jobHistoryQueryResponse.NextToken, "Next Token should be populated")
+	nextToken = jobHistoryQueryResponse.NextToken
+	require.Len(s.T(), jobHistoryQueryResponse.JobHistory, 5)
+
+	jobHistoryQueryResponse, err = s.store.GetJobHistory(s.ctx, "110", jobstore.JobHistoryQuery{
+		Limit:     5,
+		NextToken: nextToken,
+	})
+	require.NoError(s.T(), err, "Failed to get job history")
+	require.Empty(s.T(), jobHistoryQueryResponse.NextToken)
+	require.Len(s.T(), jobHistoryQueryResponse.JobHistory, 0)
+
+}
+
 func (s *BoltJobstoreTestSuite) TestTimeFilteredJobHistory() {
-	options := jobstore.JobHistoryFilterOptions{
+	options := jobstore.JobHistoryQuery{
 		Since: 5,
 	}
 
-	history, err := s.store.GetJobHistory(s.ctx, "110", options)
+	jobHistoryQueryResponse, err := s.store.GetJobHistory(s.ctx, "110", options)
 	require.NoError(s.T(), err, "failed to get job history")
-	require.Equal(s.T(), 4, len(history))
+	require.Equal(s.T(), 4, len(jobHistoryQueryResponse.JobHistory))
 }
 
 func (s *BoltJobstoreTestSuite) TestExecutionFilteredJobHistory() {
-	allHistories, err := s.store.GetJobHistory(s.ctx, "110", jobstore.JobHistoryFilterOptions{})
+	jobHistoryQueryResponse, err := s.store.GetJobHistory(s.ctx, "110", jobstore.JobHistoryQuery{})
 	require.NoError(s.T(), err)
 
 	var executionID string
-	for _, h := range allHistories {
+	for _, h := range jobHistoryQueryResponse.JobHistory {
 		if h.ExecutionID != "" {
 			executionID = h.ExecutionID
 			break
@@ -247,24 +287,24 @@ func (s *BoltJobstoreTestSuite) TestExecutionFilteredJobHistory() {
 	}
 	require.NotEmpty(s.T(), executionID, "failed to find execution ID")
 
-	options := jobstore.JobHistoryFilterOptions{
+	options := jobstore.JobHistoryQuery{
 		ExecutionID: executionID,
 	}
 
-	history, err := s.store.GetJobHistory(s.ctx, "110", options)
+	jobHistoryQueryResponse, err = s.store.GetJobHistory(s.ctx, "110", options)
 	require.NoError(s.T(), err, "failed to get job history")
 
-	for _, h := range history {
+	for _, h := range jobHistoryQueryResponse.JobHistory {
 		require.Equal(s.T(), executionID, h.ExecutionID)
 	}
 }
 
 func (s *BoltJobstoreTestSuite) TestNodeFilteredJobHistory() {
-	allHistories, err := s.store.GetJobHistory(s.ctx, "110", jobstore.JobHistoryFilterOptions{})
+	jobHistoryQueryResponse, err := s.store.GetJobHistory(s.ctx, "110", jobstore.JobHistoryQuery{})
 	require.NoError(s.T(), err)
 
 	var nodeID string
-	for _, h := range allHistories {
+	for _, h := range jobHistoryQueryResponse.JobHistory {
 		if h.NodeID != "" {
 			nodeID = h.NodeID
 			break
@@ -272,32 +312,32 @@ func (s *BoltJobstoreTestSuite) TestNodeFilteredJobHistory() {
 	}
 	require.NotEmpty(s.T(), nodeID, "failed to find node ID")
 
-	options := jobstore.JobHistoryFilterOptions{
+	options := jobstore.JobHistoryQuery{
 		NodeID: nodeID,
 	}
 
-	history, err := s.store.GetJobHistory(s.ctx, "110", options)
+	jobHistoryQueryResponse, err = s.store.GetJobHistory(s.ctx, "110", options)
 	require.NoError(s.T(), err, "failed to get job history")
 
-	for _, h := range history {
+	for _, h := range jobHistoryQueryResponse.JobHistory {
 		require.Equal(s.T(), nodeID, h.NodeID)
 	}
 }
 
 func (s *BoltJobstoreTestSuite) TestLevelFilteredJobHistory() {
-	jobOptions := jobstore.JobHistoryFilterOptions{
+	jobOptions := jobstore.JobHistoryQuery{
 		ExcludeExecutionLevel: true,
 	}
-	execOptions := jobstore.JobHistoryFilterOptions{
+	execOptions := jobstore.JobHistoryQuery{
 		ExcludeJobLevel: true,
 	}
 
-	history, err := s.store.GetJobHistory(s.ctx, "110", jobOptions)
+	jobHistoryQueryResponse, err := s.store.GetJobHistory(s.ctx, "110", jobOptions)
 	s.Require().NoError(err, "failed to get job history")
-	s.Require().Equal(4, len(history))
-	s.Require().Equal(models.JobStateTypePending, history[1].JobState.New)
+	s.Require().Equal(4, len(jobHistoryQueryResponse.JobHistory))
+	s.Require().Equal(models.JobStateTypePending, jobHistoryQueryResponse.JobHistory[1].JobState.New)
 
-	count := lo.Reduce(history, func(agg int, item models.JobHistory, _ int) int {
+	count := lo.Reduce(jobHistoryQueryResponse.JobHistory, func(agg int, item models.JobHistory, _ int) int {
 		if item.Type == models.JobHistoryTypeJobLevel {
 			return agg + 1
 		}
@@ -305,12 +345,12 @@ func (s *BoltJobstoreTestSuite) TestLevelFilteredJobHistory() {
 	}, 0)
 	s.Require().Equal(count, 4)
 
-	history, err = s.store.GetJobHistory(s.ctx, "110", execOptions)
+	jobHistoryQueryResponse, err = s.store.GetJobHistory(s.ctx, "110", execOptions)
 	s.Require().NoError(err, "failed to get job history")
-	s.Require().Equal(4, len(history))
-	s.Require().Equal(models.ExecutionStateAskForBid, history[1].ExecutionState.New)
+	s.Require().Equal(4, len(jobHistoryQueryResponse.JobHistory))
+	s.Require().Equal(models.ExecutionStateAskForBid, jobHistoryQueryResponse.JobHistory[1].ExecutionState.New)
 
-	count = lo.Reduce(history, func(agg int, item models.JobHistory, _ int) int {
+	count = lo.Reduce(jobHistoryQueryResponse.JobHistory, func(agg int, item models.JobHistory, _ int) int {
 		if item.Type == models.JobHistoryTypeExecutionLevel {
 			return agg + 1
 		}
@@ -640,7 +680,6 @@ func (s *BoltJobstoreTestSuite) TestGetExecutions() {
 	s.NotNil(state)
 	s.Equal(2, len(state))
 	s.Equal(state[0].GetModifyTime().After(state[1].GetModifyTime()), true)
-
 }
 
 func (s *BoltJobstoreTestSuite) TestInProgressJobs() {
@@ -778,7 +817,6 @@ func (s *BoltJobstoreTestSuite) TestEvents() {
 }
 
 func (s *BoltJobstoreTestSuite) TestEvaluations() {
-
 	eval := models.Evaluation{
 		ID:    "e1",
 		JobID: "10",