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

Issue:
The requester node attempts to schedule work on disconnected nodes resulting in the job never running
Due to changes here: 
- https://github.com/bacalhau-project/bacalhau/pull/3768/files#diff-16bad1556b6ffdd42e5c57a235f78726a613fc46759a999fb23703ce93f92ecbR161-R163
- https://github.com/bacalhau-project/bacalhau/pull/3768/files#diff-3034b567889a6a7f1ea37dcf47d446e1fbf1a2928894a4fb7d9d2cfbafc857ecR93
- https://github.com/bacalhau-project/bacalhau/pull/3768/files#diff-6552869d10da40039d76eca490cf89bb20a2e4eb860c892ac1569b7a0eb88bebR110

Current proposal is to:
- make auto-approve work via a fix for: https://github.com/bacalhau-project/bacalhau/issues/3783
- modify the aforementioned code to only schedule on connected nodes.

cc @rossjones & @wdbaruni to weigh in on how the new event system introduced in https://github.com/bacalhau-project/bacalhau/pull/3772 can be used to force scheduling of executions when offline compute nodes come online again.


Discussion / hints:
What is the proposal here? I believe the default option now is to auto-approve nodes, and only schedule on approved and connected nodes. Is any of that still missing?
Yeah the only scheduling on connected and approved nodes is missing. Currently we schedule on disconnected node for some job types and ignore their approval state for other job types. frankly it's a bit of a mess:
- we need to address this TODO: https://github.com/bacalhau-project/bacalhau/blob/main/pkg/orchestrator/scheduler/batch_service_job.go#L158
- modify this section: https://github.com/bacalhau-project/bacalhau/blob/main/pkg/orchestrator/scheduler/daemon_job.go#L93
- modify this section: https://github.com/bacalhau-project/bacalhau/blob/main/pkg/orchestrator/scheduler/ops_job.go#L110

Or rather than modify, allow these aspects of scheduling to be configured.

Further we need to ensure that worked scheduled on an offline node runs when the node comes back online which we will need https://github.com/bacalhau-project/bacalhau/pull/3772 to do. e.g. the orchestrator could listen for connected events and create an evaluation to execute the work.
Another point to consider:
How can we allow users to define different scheduling heuristics for compute nodes. e.g. nodes in a data center ought to have a more strict requirement on connectedness than nodes that are expected to go offline for longer periods of time (e.g. submarine compute nodes)
@wdbaruni previously suggested adding another (a third) timeout in future which allows nodes to be offline for that long before being considered dead. 

Fix patch:
diff --git a/pkg/node/requester.go b/pkg/node/requester.go
index cf1a88b655..99ba1211dd 100644
--- a/pkg/node/requester.go
+++ b/pkg/node/requester.go
@@ -99,12 +99,6 @@ func NewRequesterNode(
 		}),
 	)
 
-	// node selector
-	nodeSelector := selector.NewNodeSelector(selector.NodeSelectorParams{
-		NodeDiscoverer: nodeInfoStore,
-		NodeRanker:     nodeRankerChain,
-	})
-
 	// evaluation broker
 	evalBroker, err := evaluation.NewInMemoryBroker(evaluation.InMemoryBrokerParams{
 		VisibilityTimeout:    requesterConfig.EvalBrokerVisibilityTimeout,
@@ -151,26 +145,24 @@ func NewRequesterNode(
 		retryStrategy = retryStrategyChain
 	}
 
+	// node selector
+	nodeSelector := selector.NewNodeSelector(
+		nodeInfoStore,
+		nodeRankerChain,
+		// selector constraints: require nodes be online and approved to schedule
+		orchestrator.NodeSelectionConstraints{
+			RequireConnected: true,
+			RequireApproval:  true,
+		},
+	)
+
 	// scheduler provider
-	batchServiceJobScheduler := scheduler.NewBatchServiceJobScheduler(scheduler.BatchServiceJobSchedulerParams{
-		JobStore:      jobStore,
-		Planner:       planners,
-		NodeSelector:  nodeSelector,
-		RetryStrategy: retryStrategy,
-	})
+	batchServiceJobScheduler := scheduler.NewBatchServiceJobScheduler(jobStore, planners, nodeSelector, retryStrategy)
 	schedulerProvider := orchestrator.NewMappedSchedulerProvider(map[string]orchestrator.Scheduler{
 		models.JobTypeBatch:   batchServiceJobScheduler,
 		models.JobTypeService: batchServiceJobScheduler,
-		models.JobTypeOps: scheduler.NewOpsJobScheduler(scheduler.OpsJobSchedulerParams{
-			JobStore:     jobStore,
-			Planner:      planners,
-			NodeSelector: nodeSelector,
-		}),
-		models.JobTypeDaemon: scheduler.NewDaemonJobScheduler(scheduler.DaemonJobSchedulerParams{
-			JobStore:     jobStore,
-			Planner:      planners,
-			NodeSelector: nodeSelector,
-		}),
+		models.JobTypeOps:     scheduler.NewOpsJobScheduler(jobStore, planners, nodeSelector),
+		models.JobTypeDaemon:  scheduler.NewDaemonJobScheduler(jobStore, planners, nodeSelector),
 	})
 
 	workers := make([]*orchestrator.Worker, 0, requesterConfig.WorkerCount)
diff --git a/pkg/orchestrator/interfaces.go b/pkg/orchestrator/interfaces.go
index e137456b71..4484f15554 100644
--- a/pkg/orchestrator/interfaces.go
+++ b/pkg/orchestrator/interfaces.go
@@ -106,11 +106,11 @@ type NodeSelector interface {
 	AllNodes(ctx context.Context) ([]models.NodeInfo, error)
 
 	// AllMatchingNodes returns all nodes that match the job constrains and selection criteria.
-	AllMatchingNodes(ctx context.Context, job *models.Job, constraints *NodeSelectionConstraints) ([]models.NodeInfo, error)
+	AllMatchingNodes(ctx context.Context, job *models.Job) ([]models.NodeInfo, error)
 
 	// TopMatchingNodes return the top ranked desiredCount number of nodes that match job constraints
 	// ordered in descending order based on their rank, or error if not enough nodes match.
-	TopMatchingNodes(ctx context.Context, job *models.Job, desiredCount int, constraints *NodeSelectionConstraints) ([]models.NodeInfo, error)
+	TopMatchingNodes(ctx context.Context, job *models.Job, desiredCount int) ([]models.NodeInfo, error)
 }
 
 type RetryStrategy interface {
diff --git a/pkg/orchestrator/mocks.go b/pkg/orchestrator/mocks.go
index 04356a87a2..4fe4c58be8 100644
--- a/pkg/orchestrator/mocks.go
+++ b/pkg/orchestrator/mocks.go
@@ -15,6 +15,7 @@ import (
 	time "time"
 
 	models "github.com/bacalhau-project/bacalhau/pkg/models"
+	routing "github.com/bacalhau-project/bacalhau/pkg/routing"
 	gomock "go.uber.org/mock/gomock"
 )
 
@@ -291,19 +292,24 @@ func (m *MockNodeDiscoverer) EXPECT() *MockNodeDiscovererMockRecorder {
 	return m.recorder
 }
 
-// ListNodes mocks base method.
-func (m *MockNodeDiscoverer) ListNodes(ctx context.Context) ([]models.NodeInfo, error) {
+// List mocks base method.
+func (m *MockNodeDiscoverer) List(ctx context.Context, filter ...routing.NodeStateFilter) ([]models.NodeState, error) {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "ListNodes", ctx)
-	ret0, _ := ret[0].([]models.NodeInfo)
+	varargs := []any{ctx}
+	for _, a := range filter {
+		varargs = append(varargs, a)
+	}
+	ret := m.ctrl.Call(m, "List", varargs...)
+	ret0, _ := ret[0].([]models.NodeState)
 	ret1, _ := ret[1].(error)
 	return ret0, ret1
 }
 
-// ListNodes indicates an expected call of ListNodes.
-func (mr *MockNodeDiscovererMockRecorder) ListNodes(ctx any) *gomock.Call {
+// List indicates an expected call of List.
+func (mr *MockNodeDiscovererMockRecorder) List(ctx any, filter ...any) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListNodes", reflect.TypeOf((*MockNodeDiscoverer)(nil).ListNodes), ctx)
+	varargs := append([]any{ctx}, filter...)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "List", reflect.TypeOf((*MockNodeDiscoverer)(nil).List), varargs...)
 }
 
 // MockNodeRanker is a mock of NodeRanker interface.
@@ -368,18 +374,18 @@ func (m *MockNodeSelector) EXPECT() *MockNodeSelectorMockRecorder {
 }
 
 // AllMatchingNodes mocks base method.
-func (m *MockNodeSelector) AllMatchingNodes(ctx context.Context, job *models.Job, constraints *NodeSelectionConstraints) ([]models.NodeInfo, error) {
+func (m *MockNodeSelector) AllMatchingNodes(ctx context.Context, job *models.Job) ([]models.NodeInfo, error) {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "AllMatchingNodes", ctx, job, constraints)
+	ret := m.ctrl.Call(m, "AllMatchingNodes", ctx, job)
 	ret0, _ := ret[0].([]models.NodeInfo)
 	ret1, _ := ret[1].(error)
 	return ret0, ret1
 }
 
 // AllMatchingNodes indicates an expected call of AllMatchingNodes.
-func (mr *MockNodeSelectorMockRecorder) AllMatchingNodes(ctx, job, constraints any) *gomock.Call {
+func (mr *MockNodeSelectorMockRecorder) AllMatchingNodes(ctx, job any) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AllMatchingNodes", reflect.TypeOf((*MockNodeSelector)(nil).AllMatchingNodes), ctx, job, constraints)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AllMatchingNodes", reflect.TypeOf((*MockNodeSelector)(nil).AllMatchingNodes), ctx, job)
 }
 
 // AllNodes mocks base method.
@@ -398,18 +404,18 @@ func (mr *MockNodeSelectorMockRecorder) AllNodes(ctx any) *gomock.Call {
 }
 
 // TopMatchingNodes mocks base method.
-func (m *MockNodeSelector) TopMatchingNodes(ctx context.Context, job *models.Job, desiredCount int, constraints *NodeSelectionConstraints) ([]models.NodeInfo, error) {
+func (m *MockNodeSelector) TopMatchingNodes(ctx context.Context, job *models.Job, desiredCount int) ([]models.NodeInfo, error) {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "TopMatchingNodes", ctx, job, desiredCount, constraints)
+	ret := m.ctrl.Call(m, "TopMatchingNodes", ctx, job, desiredCount)
 	ret0, _ := ret[0].([]models.NodeInfo)
 	ret1, _ := ret[1].(error)
 	return ret0, ret1
 }
 
 // TopMatchingNodes indicates an expected call of TopMatchingNodes.
-func (mr *MockNodeSelectorMockRecorder) TopMatchingNodes(ctx, job, desiredCount, constraints any) *gomock.Call {
+func (mr *MockNodeSelectorMockRecorder) TopMatchingNodes(ctx, job, desiredCount any) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "TopMatchingNodes", reflect.TypeOf((*MockNodeSelector)(nil).TopMatchingNodes), ctx, job, desiredCount, constraints)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "TopMatchingNodes", reflect.TypeOf((*MockNodeSelector)(nil).TopMatchingNodes), ctx, job, desiredCount)
 }
 
 // MockRetryStrategy is a mock of RetryStrategy interface.
diff --git a/pkg/orchestrator/scheduler/batch_service_job.go b/pkg/orchestrator/scheduler/batch_service_job.go
index bc4010cc0d..110714bde1 100644
--- a/pkg/orchestrator/scheduler/batch_service_job.go
+++ b/pkg/orchestrator/scheduler/batch_service_job.go
@@ -20,23 +20,21 @@ import (
 type BatchServiceJobScheduler struct {
 	jobStore      jobstore.Store
 	planner       orchestrator.Planner
-	nodeSelector  orchestrator.NodeSelector
+	selector      orchestrator.NodeSelector
 	retryStrategy orchestrator.RetryStrategy
 }
 
-type BatchServiceJobSchedulerParams struct {
-	JobStore      jobstore.Store
-	Planner       orchestrator.Planner
-	NodeSelector  orchestrator.NodeSelector
-	RetryStrategy orchestrator.RetryStrategy
-}
-
-func NewBatchServiceJobScheduler(params BatchServiceJobSchedulerParams) *BatchServiceJobScheduler {
+func NewBatchServiceJobScheduler(
+	store jobstore.Store,
+	planner orchestrator.Planner,
+	selector orchestrator.NodeSelector,
+	strategy orchestrator.RetryStrategy,
+) *BatchServiceJobScheduler {
 	return &BatchServiceJobScheduler{
-		jobStore:      params.JobStore,
-		planner:       params.Planner,
-		nodeSelector:  params.NodeSelector,
-		retryStrategy: params.RetryStrategy,
+		jobStore:      store,
+		planner:       planner,
+		selector:      selector,
+		retryStrategy: strategy,
 	}
 }
 
@@ -69,7 +67,7 @@ func (b *BatchServiceJobScheduler) Process(ctx context.Context, evaluation *mode
 	}
 
 	// Retrieve the info for all the nodes that have executions for this job
-	nodeInfos, err := existingNodeInfos(ctx, b.nodeSelector, nonTerminalExecs)
+	nodeInfos, err := existingNodeInfos(ctx, b.selector, nonTerminalExecs)
 	if err != nil {
 		return err
 	}
@@ -155,16 +153,7 @@ func (b *BatchServiceJobScheduler) createMissingExecs(
 // placeExecs places the executions
 func (b *BatchServiceJobScheduler) placeExecs(ctx context.Context, execs execSet, job *models.Job) error {
 	if len(execs) > 0 {
-		// TODO: Remove the options once we are ready to enforce that only connected/approved nodes can be used
-		selectedNodes, err := b.nodeSelector.TopMatchingNodes(
-			ctx,
-			job,
-			len(execs),
-			&orchestrator.NodeSelectionConstraints{
-				RequireApproval:  false,
-				RequireConnected: false,
-			},
-		)
+		selectedNodes, err := b.selector.TopMatchingNodes(ctx, job, len(execs))
 		if err != nil {
 			return err
 		}
@@ -178,7 +167,7 @@ func (b *BatchServiceJobScheduler) placeExecs(ctx context.Context, execs execSet
 }
 
 func (b *BatchServiceJobScheduler) handleFailure(nonTerminalExecs execSet, failed execSet, plan *models.Plan, err error) {
-	// TODO: allow scheduling retries in a later time if don't find nodes instead of failing the job
+	// TODO(walid): allow scheduling retries in a later time if don't find nodes instead of failing the job
 	// mark all non-terminal executions as failed
 	nonTerminalExecs.markStopped(plan.Event, plan)
 
diff --git a/pkg/orchestrator/scheduler/daemon_job.go b/pkg/orchestrator/scheduler/daemon_job.go
index da0cc3ed75..bbd1cd300b 100644
--- a/pkg/orchestrator/scheduler/daemon_job.go
+++ b/pkg/orchestrator/scheduler/daemon_job.go
@@ -20,17 +20,15 @@ type DaemonJobScheduler struct {
 	nodeSelector orchestrator.NodeSelector
 }
 
-type DaemonJobSchedulerParams struct {
-	JobStore     jobstore.Store
-	Planner      orchestrator.Planner
-	NodeSelector orchestrator.NodeSelector
-}
-
-func NewDaemonJobScheduler(params DaemonJobSchedulerParams) *DaemonJobScheduler {
+func NewDaemonJobScheduler(
+	store jobstore.Store,
+	planner orchestrator.Planner,
+	selector orchestrator.NodeSelector,
+) *DaemonJobScheduler {
 	return &DaemonJobScheduler{
-		jobStore:     params.JobStore,
-		planner:      params.Planner,
-		nodeSelector: params.NodeSelector,
+		jobStore:     store,
+		planner:      planner,
+		nodeSelector: selector,
 	}
 }
 
@@ -86,12 +84,7 @@ func (b *DaemonJobScheduler) createMissingExecs(
 	ctx context.Context, job *models.Job, plan *models.Plan, existingExecs execSet) (execSet, error) {
 	newExecs := execSet{}
 
-	// Require approval when selecting nodes, but do not require them to be connected.
-	nodes, err := b.nodeSelector.AllMatchingNodes(
-		ctx,
-		job,
-		&orchestrator.NodeSelectionConstraints{RequireApproval: true, RequireConnected: false},
-	)
+	nodes, err := b.nodeSelector.AllMatchingNodes(ctx, job)
 	if err != nil {
 		return newExecs, err
 	}
diff --git a/pkg/orchestrator/scheduler/ops_job.go b/pkg/orchestrator/scheduler/ops_job.go
index 57d3a27b9b..cf462e2743 100644
--- a/pkg/orchestrator/scheduler/ops_job.go
+++ b/pkg/orchestrator/scheduler/ops_job.go
@@ -16,22 +16,20 @@ import (
 
 // OpsJobScheduler is a scheduler for batch jobs that run until completion
 type OpsJobScheduler struct {
-	jobStore     jobstore.Store
-	planner      orchestrator.Planner
-	nodeSelector orchestrator.NodeSelector
+	jobStore jobstore.Store
+	planner  orchestrator.Planner
+	selector orchestrator.NodeSelector
 }
 
-type OpsJobSchedulerParams struct {
-	JobStore     jobstore.Store
-	Planner      orchestrator.Planner
-	NodeSelector orchestrator.NodeSelector
-}
-
-func NewOpsJobScheduler(params OpsJobSchedulerParams) *OpsJobScheduler {
+func NewOpsJobScheduler(
+	store jobstore.Store,
+	planner orchestrator.Planner,
+	selector orchestrator.NodeSelector,
+) *OpsJobScheduler {
 	return &OpsJobScheduler{
-		jobStore:     params.JobStore,
-		planner:      params.Planner,
-		nodeSelector: params.NodeSelector,
+		jobStore: store,
+		planner:  planner,
+		selector: selector,
 	}
 }
 
@@ -64,7 +62,7 @@ func (b *OpsJobScheduler) Process(ctx context.Context, evaluation *models.Evalua
 	}
 
 	// Retrieve the info for all the nodes that have executions for this job
-	nodeInfos, err := existingNodeInfos(ctx, b.nodeSelector, nonTerminalExecs)
+	nodeInfos, err := existingNodeInfos(ctx, b.selector, nonTerminalExecs)
 	if err != nil {
 		return err
 	}
@@ -103,12 +101,7 @@ func (b *OpsJobScheduler) Process(ctx context.Context, evaluation *models.Evalua
 func (b *OpsJobScheduler) createMissingExecs(
 	ctx context.Context, job *models.Job, plan *models.Plan) (execSet, error) {
 	newExecs := execSet{}
-	nodes, err := b.nodeSelector.AllMatchingNodes(
-		ctx,
-		job, &orchestrator.NodeSelectionConstraints{
-			RequireApproval:  true,
-			RequireConnected: false,
-		})
+	nodes, err := b.selector.AllMatchingNodes(ctx, job)
 	if err != nil {
 		return newExecs, err
 	}
diff --git a/pkg/orchestrator/selection/selector/node_selector.go b/pkg/orchestrator/selection/selector/node_selector.go
index 3dd3d4cf27..a5db941f93 100644
--- a/pkg/orchestrator/selection/selector/node_selector.go
+++ b/pkg/orchestrator/selection/selector/node_selector.go
@@ -15,25 +15,26 @@ import (
 	"github.com/bacalhau-project/bacalhau/pkg/util/generic"
 )
 
-type NodeSelectorParams struct {
-	NodeDiscoverer orchestrator.NodeDiscoverer
-	NodeRanker     orchestrator.NodeRanker
-}
-
 type NodeSelector struct {
-	nodeDiscoverer orchestrator.NodeDiscoverer
-	nodeRanker     orchestrator.NodeRanker
+	discoverer  orchestrator.NodeDiscoverer
+	ranker      orchestrator.NodeRanker
+	constraints orchestrator.NodeSelectionConstraints
 }
 
-func NewNodeSelector(params NodeSelectorParams) *NodeSelector {
+func NewNodeSelector(
+	discoverer orchestrator.NodeDiscoverer,
+	ranker orchestrator.NodeRanker,
+	constraints orchestrator.NodeSelectionConstraints,
+) *NodeSelector {
 	return &NodeSelector{
-		nodeDiscoverer: params.NodeDiscoverer,
-		nodeRanker:     params.NodeRanker,
+		discoverer:  discoverer,
+		ranker:      ranker,
+		constraints: constraints,
 	}
 }
 
 func (n NodeSelector) AllNodes(ctx context.Context) ([]models.NodeInfo, error) {
-	nodeStates, err := n.nodeDiscoverer.List(ctx)
+	nodeStates, err := n.discoverer.List(ctx)
 	if err != nil {
 		return nil, fmt.Errorf("failed to list discovered nodes: %w", err)
 	}
@@ -45,10 +46,8 @@ func (n NodeSelector) AllNodes(ctx context.Context) ([]models.NodeInfo, error) {
 	return nodeInfos, nil
 }
 
-func (n NodeSelector) AllMatchingNodes(ctx context.Context,
-	job *models.Job,
-	constraints *orchestrator.NodeSelectionConstraints) ([]models.NodeInfo, error) {
-	filteredNodes, _, err := n.rankAndFilterNodes(ctx, job, constraints)
+func (n NodeSelector) AllMatchingNodes(ctx context.Context, job *models.Job) ([]models.NodeInfo, error) {
+	filteredNodes, _, err := n.rankAndFilterNodes(ctx, job)
 	if err != nil {
 		return nil, err
 	}
@@ -57,10 +56,12 @@ func (n NodeSelector) AllMatchingNodes(ctx context.Context,
 	return nodeInfos, nil
 }
 
-func (n NodeSelector) TopMatchingNodes(ctx context.Context,
-	job *models.Job, desiredCount int,
-	constraints *orchestrator.NodeSelectionConstraints) ([]models.NodeInfo, error) {
-	possibleNodes, rejectedNodes, err := n.rankAndFilterNodes(ctx, job, constraints)
+func (n NodeSelector) TopMatchingNodes(
+	ctx context.Context,
+	job *models.Job,
+	desiredCount int,
+) ([]models.NodeInfo, error) {
+	possibleNodes, rejectedNodes, err := n.rankAndFilterNodes(ctx, job)
 	if err != nil {
 		return nil, err
 	}
@@ -80,10 +81,11 @@ func (n NodeSelector) TopMatchingNodes(ctx context.Context,
 	return selectedInfos, nil
 }
 
-func (n NodeSelector) rankAndFilterNodes(ctx context.Context,
+func (n NodeSelector) rankAndFilterNodes(
+	ctx context.Context,
 	job *models.Job,
-	constraints *orchestrator.NodeSelectionConstraints) (selected, rejected []orchestrator.NodeRank, err error) {
-	listed, err := n.nodeDiscoverer.List(ctx)
+) (selected, rejected []orchestrator.NodeRank, err error) {
+	listed, err := n.discoverer.List(ctx)
 	if err != nil {
 		return nil, nil, err
 	}
@@ -97,11 +99,11 @@ func (n NodeSelector) rankAndFilterNodes(ctx context.Context,
 			return false
 		}
 
-		if constraints.RequireApproval && nodeState.Membership != models.NodeMembership.APPROVED {
+		if n.constraints.RequireApproval && nodeState.Membership != models.NodeMembership.APPROVED {
 			return false
 		}
 
-		if constraints.RequireConnected && nodeState.Connection != models.NodeStates.CONNECTED {
+		if n.constraints.RequireConnected && nodeState.Connection != models.NodeStates.CONNECTED {
 			return false
 		}
 
@@ -118,7 +120,7 @@ func (n NodeSelector) rankAndFilterNodes(ctx context.Context,
 		nodeInfos = append(nodeInfos, ns.Info)
 	}
 
-	rankedNodes, err := n.nodeRanker.RankNodes(ctx, *job, nodeInfos)
+	rankedNodes, err := n.ranker.RankNodes(ctx, *job, nodeInfos)
 	if err != nil {
 		return nil, nil, err
 	}


Test patch:
diff --git a/pkg/orchestrator/scheduler/batch_job_test.go b/pkg/orchestrator/scheduler/batch_job_test.go
index bc34c88148..bb0c15b4c3 100644
--- a/pkg/orchestrator/scheduler/batch_job_test.go
+++ b/pkg/orchestrator/scheduler/batch_job_test.go
@@ -49,12 +49,12 @@ func (s *BatchJobSchedulerTestSuite) SetupTest() {
 	s.nodeSelector = orchestrator.NewMockNodeSelector(ctrl)
 	s.retryStrategy = retry.NewFixedStrategy(retry.FixedStrategyParams{ShouldRetry: true})
 
-	s.scheduler = NewBatchServiceJobScheduler(BatchServiceJobSchedulerParams{
-		JobStore:      s.jobStore,
-		Planner:       s.planner,
-		NodeSelector:  s.nodeSelector,
-		RetryStrategy: s.retryStrategy,
-	})
+	s.scheduler = NewBatchServiceJobScheduler(
+		s.jobStore,
+		s.planner,
+		s.nodeSelector,
+		s.retryStrategy,
+	)
 }
 
 func TestBatchSchedulerTestSuite(t *testing.T) {
@@ -304,19 +304,13 @@ func (s *BatchJobSchedulerTestSuite) TestProcess_ShouldMarkJobAsFailed_NoRetry()
 }
 
 func (s *BatchJobSchedulerTestSuite) mockNodeSelection(job *models.Job, nodeInfos []models.NodeInfo, desiredCount int) {
-	constraints := &orchestrator.NodeSelectionConstraints{
-		RequireApproval:  false,
-		RequireConnected: false,
-	}
-
 	if len(nodeInfos) < desiredCount {
-		s.nodeSelector.EXPECT().TopMatchingNodes(gomock.Any(), job, desiredCount, constraints).Return(nil, orchestrator.ErrNotEnoughNodes{})
+		s.nodeSelector.EXPECT().TopMatchingNodes(gomock.Any(), job, desiredCount).Return(nil, orchestrator.ErrNotEnoughNodes{})
 	} else {
 		s.nodeSelector.EXPECT().TopMatchingNodes(
 			gomock.Any(),
 			job,
 			desiredCount,
-			constraints,
 		).Return(nodeInfos, nil)
 	}
 }
diff --git a/pkg/orchestrator/scheduler/daemon_job_test.go b/pkg/orchestrator/scheduler/daemon_job_test.go
index 5f74a87077..d90d48bf4e 100644
--- a/pkg/orchestrator/scheduler/daemon_job_test.go
+++ b/pkg/orchestrator/scheduler/daemon_job_test.go
@@ -30,11 +30,11 @@ func (s *DaemonJobSchedulerTestSuite) SetupTest() {
 	s.planner = orchestrator.NewMockPlanner(ctrl)
 	s.nodeSelector = orchestrator.NewMockNodeSelector(ctrl)
 
-	s.scheduler = NewDaemonJobScheduler(DaemonJobSchedulerParams{
-		JobStore:     s.jobStore,
-		Planner:      s.planner,
-		NodeSelector: s.nodeSelector,
-	})
+	s.scheduler = NewDaemonJobScheduler(
+		s.jobStore,
+		s.planner,
+		s.nodeSelector,
+	)
 }
 
 func TestDaemonJobSchedulerTestSuite(t *testing.T) {
@@ -53,7 +53,7 @@ func (s *DaemonJobSchedulerTestSuite) TestProcess_ShouldCreateNewExecutions() {
 		*fakeNodeInfo(s.T(), nodeIDs[1]),
 		*fakeNodeInfo(s.T(), nodeIDs[2]),
 	}
-	s.nodeSelector.EXPECT().AllMatchingNodes(gomock.Any(), gomock.Any(), gomock.Any()).Return(nodeInfos, nil)
+	s.nodeSelector.EXPECT().AllMatchingNodes(gomock.Any(), gomock.Any()).Return(nodeInfos, nil)
 
 	matcher := NewPlanMatcher(s.T(), PlanMatcherParams{
 		Evaluation:               evaluation,
@@ -78,7 +78,7 @@ func (s *DaemonJobSchedulerTestSuite) TestProcess_ShouldNOTMarkJobAsCompleted()
 	executions[1].ComputeState = models.NewExecutionState(models.ExecutionStateCompleted) // Simulate a completed execution
 	s.jobStore.EXPECT().GetJob(gomock.Any(), job.ID).Return(*job, nil)
 	s.jobStore.EXPECT().GetExecutions(gomock.Any(), jobstore.GetExecutionsOptions{JobID: job.ID}).Return(executions, nil)
-	s.nodeSelector.EXPECT().AllMatchingNodes(gomock.Any(), gomock.Any(), gomock.Any()).Return([]models.NodeInfo{}, nil)
+	s.nodeSelector.EXPECT().AllMatchingNodes(gomock.Any(), gomock.Any()).Return([]models.NodeInfo{}, nil)
 
 	// Noop plan
 	matcher := NewPlanMatcher(s.T(), PlanMatcherParams{
@@ -101,7 +101,7 @@ func (s *DaemonJobSchedulerTestSuite) TestProcess_ShouldMarkLostExecutionsOnUnhe
 		*fakeNodeInfo(s.T(), executions[1].NodeID),
 	}
 	s.nodeSelector.EXPECT().AllNodes(gomock.Any()).Return(nodeInfos, nil)
-	s.nodeSelector.EXPECT().AllMatchingNodes(gomock.Any(), job, gomock.Any()).Return(nodeInfos, nil)
+	s.nodeSelector.EXPECT().AllMatchingNodes(gomock.Any(), job).Return(nodeInfos, nil)
 
 	matcher := NewPlanMatcher(s.T(), PlanMatcherParams{
 		Evaluation:         evaluation,
@@ -130,7 +130,7 @@ func (s *DaemonJobSchedulerTestSuite) TestProcess_ShouldNOTMarkJobAsFailed() {
 		*fakeNodeInfo(s.T(), executions[1].NodeID),
 	}
 	s.nodeSelector.EXPECT().AllNodes(gomock.Any()).Return(nodeInfos, nil)
-	s.nodeSelector.EXPECT().AllMatchingNodes(gomock.Any(), job, gomock.Any()).Return(nodeInfos, nil)
+	s.nodeSelector.EXPECT().AllMatchingNodes(gomock.Any(), job).Return(nodeInfos, nil)
 
 	matcher := NewPlanMatcher(s.T(), PlanMatcherParams{
 		Evaluation: evaluation,
@@ -166,7 +166,7 @@ func (s *DaemonJobSchedulerTestSuite) TestProcessFail_NoMatchingNodes() {
 	executions := []models.Execution{} // no executions yet
 	s.jobStore.EXPECT().GetJob(gomock.Any(), job.ID).Return(*job, nil)
 	s.jobStore.EXPECT().GetExecutions(gomock.Any(), jobstore.GetExecutionsOptions{JobID: job.ID}).Return(executions, nil)
-	s.nodeSelector.EXPECT().AllMatchingNodes(gomock.Any(), job, gomock.Any()).Return([]models.NodeInfo{}, nil)
+	s.nodeSelector.EXPECT().AllMatchingNodes(gomock.Any(), job).Return([]models.NodeInfo{}, nil)
 
 	// Noop plan
 	matcher := NewPlanMatcher(s.T(), PlanMatcherParams{
diff --git a/pkg/orchestrator/scheduler/ops_job_test.go b/pkg/orchestrator/scheduler/ops_job_test.go
index c9813071c3..77aea26e08 100644
--- a/pkg/orchestrator/scheduler/ops_job_test.go
+++ b/pkg/orchestrator/scheduler/ops_job_test.go
@@ -30,11 +30,11 @@ func (s *OpsJobSchedulerTestSuite) SetupTest() {
 	s.planner = orchestrator.NewMockPlanner(ctrl)
 	s.nodeSelector = orchestrator.NewMockNodeSelector(ctrl)
 
-	s.scheduler = NewOpsJobScheduler(OpsJobSchedulerParams{
-		JobStore:     s.jobStore,
-		Planner:      s.planner,
-		NodeSelector: s.nodeSelector,
-	})
+	s.scheduler = NewOpsJobScheduler(
+		s.jobStore,
+		s.planner,
+		s.nodeSelector,
+	)
 }
 
 func TestOpsJobSchedulerTestSuite(t *testing.T) {
@@ -168,7 +168,7 @@ func (s *OpsJobSchedulerTestSuite) TestProcessFail_NoMatchingNodes() {
 }
 
 func (s *OpsJobSchedulerTestSuite) mockNodeSelection(job *models.Job, nodeInfos []models.NodeInfo) {
-	s.nodeSelector.EXPECT().AllMatchingNodes(gomock.Any(), job, gomock.Any()).Return(nodeInfos, nil)
+	s.nodeSelector.EXPECT().AllMatchingNodes(gomock.Any(), job).Return(nodeInfos, nil)
 }
 
 func mockOpsJob() (*models.Job, []models.Execution, *models.Evaluation) {
diff --git a/pkg/orchestrator/scheduler/service_job_test.go b/pkg/orchestrator/scheduler/service_job_test.go
index 9fcac97b15..c5789b8b05 100644
--- a/pkg/orchestrator/scheduler/service_job_test.go
+++ b/pkg/orchestrator/scheduler/service_job_test.go
@@ -41,12 +41,12 @@ func (s *ServiceJobSchedulerTestSuite) SetupTest() {
 	s.nodeSelector = orchestrator.NewMockNodeSelector(ctrl)
 	s.retryStrategy = retry.NewFixedStrategy(retry.FixedStrategyParams{ShouldRetry: true})
 
-	s.scheduler = NewBatchServiceJobScheduler(BatchServiceJobSchedulerParams{
-		JobStore:      s.jobStore,
-		Planner:       s.planner,
-		NodeSelector:  s.nodeSelector,
-		RetryStrategy: s.retryStrategy,
-	})
+	s.scheduler = NewBatchServiceJobScheduler(
+		s.jobStore,
+		s.planner,
+		s.nodeSelector,
+		s.retryStrategy,
+	)
 }
 
 func TestServiceSchedulerTestSuite(t *testing.T) {
@@ -322,14 +322,10 @@ func (s *ServiceJobSchedulerTestSuite) TestProcess_ShouldMarkJobAsFailed_NoRetry
 }
 
 func (s *ServiceJobSchedulerTestSuite) mockNodeSelection(job *models.Job, nodeInfos []models.NodeInfo, desiredCount int) {
-	constraints := &orchestrator.NodeSelectionConstraints{
-		RequireApproval:  false,
-		RequireConnected: false,
-	}
 	if len(nodeInfos) < desiredCount {
-		s.nodeSelector.EXPECT().TopMatchingNodes(gomock.Any(), job, desiredCount, constraints).Return(nil, orchestrator.ErrNotEnoughNodes{})
+		s.nodeSelector.EXPECT().TopMatchingNodes(gomock.Any(), job, desiredCount).Return(nil, orchestrator.ErrNotEnoughNodes{})
 	} else {
-		s.nodeSelector.EXPECT().TopMatchingNodes(gomock.Any(), job, desiredCount, constraints).Return(nodeInfos, nil)
+		s.nodeSelector.EXPECT().TopMatchingNodes(gomock.Any(), job, desiredCount).Return(nodeInfos, nil)
 	}
 }
 
diff --git a/pkg/test/requester/node_selection_test.go b/pkg/test/requester/node_selection_test.go
index bdb291b16a..82a0e8e0e3 100644
--- a/pkg/test/requester/node_selection_test.go
+++ b/pkg/test/requester/node_selection_test.go
@@ -63,13 +63,10 @@ func (s *NodeSelectionSuite) SetupSuite() {
 			},
 		},
 	}
-	requesterConfig, err := node.NewRequesterConfigWith(
-		node.RequesterConfigParams{
-			NodeRankRandomnessRange: 0,
-			OverAskForBidsFactor:    1,
-		},
-	)
+	requesterConfig, err := node.NewRequesterConfigWithDefaults()
 	s.Require().NoError(err)
+	requesterConfig.OverAskForBidsFactor = 1
+
 	stack := teststack.Setup(ctx,
 		s.T(),
 		devstack.WithNumberOfRequesterOnlyNodes(1),
diff --git a/pkg/test/requester/retries_test.go b/pkg/test/requester/retries_test.go
index 58a5970cb0..849b342ca7 100644
--- a/pkg/test/requester/retries_test.go
+++ b/pkg/test/requester/retries_test.go
@@ -132,13 +132,10 @@ func (s *RetriesSuite) SetupSuite() {
 	}
 	ctx := context.Background()
 
-	requesterConfig, err := node.NewRequesterConfigWith(
-		node.RequesterConfigParams{
-			NodeRankRandomnessRange: 0,
-			OverAskForBidsFactor:    1,
-		},
-	)
+	requesterConfig, err := node.NewRequesterConfigWithDefaults()
 	s.Require().NoError(err)
+	requesterConfig.OverAskForBidsFactor = 1
+
 	stack := teststack.Setup(ctx, s.T(),
 		devstack.WithNumberOfRequesterOnlyNodes(1),
 		devstack.WithNumberOfComputeOnlyNodes(len(nodeOverrides)-1),