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

Issue:
No bid rejection for transient reasons. E.g. no available capacity
To simplify our scheduling logic and minimize the edge cases and decision branches, I would like us to move to a state where bid rejections by compute nodes are only for terminal reasons and not for transient reasons, such as no available capacity. Compute nodes already queue jobs locally if they don't have immediate capacity, they can share both the available and queued capacity with the orchestrator, and the orchestrator can make a decision on ranking nodes based on those capacities and can filter compute nodes with a large queue and retry later. This enables higher priority jobs to be scheduled in compute nodes with large queue of pending jobs, but get the chance to run sooner if we use priority queues in the executor_buffer

Even if the Orchestrator's view of the network's utilization is stale, we have solutions that can mitigation that including:
1. Use Sparrow scheduling where the orchestrator will select multiple compute nodes, and they only accept when the job reaches the head of the of the queue and ready to be executed https://github.com/bacalhau-project/bacalhau/issues/3856
2. Add a Housekeeping task that tracks executions taking too long in negotiation phase and enqueue an evaluation. The scheduler might decide to pick another node if there are better ranked ones or keep the same node otherwise


Discussion / hints:

Fix patch:
diff --git a/cmd/cli/config/auto.go b/cmd/cli/config/auto.go
index 52298c3f3a..9ecf3bf407 100644
--- a/cmd/cli/config/auto.go
+++ b/cmd/cli/config/auto.go
@@ -126,9 +126,6 @@ func autoConfig(ctx context.Context, settings *autoSettings) error {
 	if err := setResources(types.NodeComputeCapacityDefaultJobResourceLimits, settings.DefaultPercentage, physicalResources); err != nil {
 		return err
 	}
-	if err := setResources(types.NodeComputeCapacityQueueResourceLimits, settings.QueuePercentage, physicalResources); err != nil {
-		return err
-	}
 
 	return nil
 }
diff --git a/cmd/cli/serve/util.go b/cmd/cli/serve/util.go
index d3650094cf..f5b14bd057 100644
--- a/cmd/cli/serve/util.go
+++ b/cmd/cli/serve/util.go
@@ -37,10 +37,9 @@ func GetComputeConfig(ctx context.Context, createExecutionStore bool) (node.Comp
 	}
 
 	totalResources, totalErr := cfg.Capacity.TotalResourceLimits.ToResources()
-	queueResources, queueErr := cfg.Capacity.QueueResourceLimits.ToResources()
 	jobResources, jobErr := cfg.Capacity.JobResourceLimits.ToResources()
 	defaultResources, defaultErr := cfg.Capacity.DefaultJobResourceLimits.ToResources()
-	if err := errors.Join(totalErr, queueErr, jobErr, defaultErr); err != nil {
+	if err := errors.Join(totalErr, jobErr, defaultErr); err != nil {
 		return node.ComputeConfig{}, err
 	}
 
@@ -56,7 +55,6 @@ func GetComputeConfig(ctx context.Context, createExecutionStore bool) (node.Comp
 
 	return node.NewComputeConfigWith(node.ComputeConfigParams{
 		TotalResourceLimits:                   *totalResources,
-		QueueResourceLimits:                   *queueResources,
 		JobResourceLimits:                     *jobResources,
 		DefaultJobResourceLimits:              *defaultResources,
 		IgnorePhysicalResourceLimits:          cfg.Capacity.IgnorePhysicalResourceLimits,
diff --git a/docs/docs/dev/cli-reference/all-flags.md b/docs/docs/dev/cli-reference/all-flags.md
index 236a3b39d2..56886e3174 100644
--- a/docs/docs/dev/cli-reference/all-flags.md
+++ b/docs/docs/dev/cli-reference/all-flags.md
@@ -416,11 +416,6 @@ Config File:
                    disk: 568 GB
                    gpu: "0"
                    memory: 52 GB
-               queueresourcelimits:
-                   cpu: 15000m
-                   disk: 1.1 TB
-                   gpu: "0"
-                   memory: 103 GB
                totalresourcelimits:
                    cpu: 7500m
                    disk: 568 GB
@@ -450,11 +445,6 @@ Config File:
                    disk: 568 GB
                    gpu: "0"
                    memory: 52 GB
-               queueresourcelimits:
-                   cpu: 50000m
-                   disk: 3.8 TB
-                   gpu: "0"
-                   memory: 344 GB
                totalresourcelimits:
                    cpu: 7500m
                    disk: 568 GB
diff --git a/docs/docs/dev/cli-reference/cli/config/auto-resources.md b/docs/docs/dev/cli-reference/cli/config/auto-resources.md
index 2ecc618cac..4ae2316d64 100644
--- a/docs/docs/dev/cli-reference/cli/config/auto-resources.md
+++ b/docs/docs/dev/cli-reference/cli/config/auto-resources.md
@@ -60,11 +60,6 @@ bacalhau config auto-resources [flags]
                    disk: 568 GB
                    gpu: "0"
                    memory: 52 GB
-               queueresourcelimits:
-                   cpu: 15000m
-                   disk: 1.1 TB
-                   gpu: "0"
-                   memory: 103 GB
                totalresourcelimits:
                    cpu: 7500m
                    disk: 568 GB
@@ -96,11 +91,6 @@ bacalhau config auto-resources [flags]
                    disk: 568 GB
                    gpu: "0"
                    memory: 52 GB
-               queueresourcelimits:
-                   cpu: 50000m
-                   disk: 3.8 TB
-                   gpu: "0"
-                   memory: 344 GB
                totalresourcelimits:
                    cpu: 7500m
                    disk: 568 GB
@@ -132,11 +122,6 @@ bacalhau config auto-resources [flags]
                    disk: 190 GB
                    gpu: "0"
                    memory: 17 GB
-               queueresourcelimits:
-                   cpu: 15000m
-                   disk: 1.1 TB
-                   gpu: "0"
-                   memory: 103 GB
                totalresourcelimits:
                    cpu: 2500m
                    disk: 190 GB
diff --git a/pkg/bidstrategy/resource/capacity_available_strategy.go b/pkg/bidstrategy/resource/capacity_available_strategy.go
deleted file mode 100644
index 9ef7cdcf54..0000000000
--- a/pkg/bidstrategy/resource/capacity_available_strategy.go
+++ /dev/null
@@ -1,50 +0,0 @@
-package resource
-
-import (
-	"context"
-	"fmt"
-
-	"github.com/bacalhau-project/bacalhau/pkg/bidstrategy"
-	"github.com/bacalhau-project/bacalhau/pkg/compute/capacity"
-	"github.com/bacalhau-project/bacalhau/pkg/models"
-)
-
-type AvailableCapacityStrategyParams struct {
-	RunningCapacityTracker  capacity.Tracker
-	EnqueuedCapacityTracker capacity.Tracker
-}
-
-type AvailableCapacityStrategy struct {
-	runningCapacityTracker  capacity.Tracker
-	enqueuedCapacityTracker capacity.Tracker
-}
-
-func NewAvailableCapacityStrategy(params AvailableCapacityStrategyParams) *AvailableCapacityStrategy {
-	s := &AvailableCapacityStrategy{
-		runningCapacityTracker:  params.RunningCapacityTracker,
-		enqueuedCapacityTracker: params.EnqueuedCapacityTracker,
-	}
-	return s
-}
-
-func (s *AvailableCapacityStrategy) ShouldBidBasedOnUsage(
-	ctx context.Context, request bidstrategy.BidStrategyRequest, usage models.Resources) (bidstrategy.BidStrategyResponse, error) {
-	runningCapacity := s.runningCapacityTracker.GetAvailableCapacity(ctx)
-	enqueuedCapacity := s.enqueuedCapacityTracker.GetAvailableCapacity(ctx)
-	totalCapacity := runningCapacity.Add(enqueuedCapacity)
-	if usage.LessThanEq(*totalCapacity) {
-		return bidstrategy.BidStrategyResponse{
-			ShouldBid:  true,
-			ShouldWait: false,
-			Reason:     "",
-		}, nil
-	}
-	return bidstrategy.BidStrategyResponse{
-		ShouldBid:  false,
-		ShouldWait: false,
-		Reason:     fmt.Sprintf("insufficient capacity - requested: %s, available: %s", usage.String(), totalCapacity.String()),
-	}, nil
-}
-
-// compile-time interface check
-var _ bidstrategy.ResourceBidStrategy = (*AvailableCapacityStrategy)(nil)
diff --git a/pkg/compute/bidder.go b/pkg/compute/bidder.go
index 5c9455f718..3e6218af5e 100644
--- a/pkg/compute/bidder.go
+++ b/pkg/compute/bidder.go
@@ -274,14 +274,16 @@ func (b Bidder) runSemanticBidding(
 		strategyType := reflect.TypeOf(s).String()
 		resp, err := s.ShouldBid(ctx, request)
 
-		log.Ctx(ctx).WithLevel(logger.ErrOrDebug(err)).
-			Err(err).
-			Str("Job", request.Job.ID).
-			Str("Strategy", strategyType).
-			Bool("Bid", resp.ShouldBid).
-			Bool("Wait", resp.ShouldWait).
-			Str("Reason", resp.Reason).
-			Send()
+		if err != nil || !resp.ShouldBid {
+			log.Ctx(ctx).WithLevel(logger.ErrOrDebug(err)).
+				Err(err).
+				Str("Job", request.Job.ID).
+				Str("Strategy", strategyType).
+				Bool("Bid", resp.ShouldBid).
+				Bool("Wait", resp.ShouldWait).
+				Str("Reason", resp.Reason).
+				Send()
+		}
 
 		if err != nil {
 			// NB: failure here results in a callback to OnComputeFailure
@@ -335,14 +337,16 @@ func (b Bidder) runResourceBidding(
 		strategyType := reflect.TypeOf(s).String()
 		resp, err := s.ShouldBidBasedOnUsage(ctx, request, *resourceUsage)
 
-		log.Ctx(ctx).WithLevel(logger.ErrOrDebug(err)).
-			Err(err).
-			Str("Job", request.Job.ID).
-			Str("Strategy", strategyType).
-			Bool("Bid", resp.ShouldBid).
-			Bool("Wait", resp.ShouldWait).
-			Str("Reason", resp.Reason).
-			Send()
+		if err != nil || !resp.ShouldBid {
+			log.Ctx(ctx).WithLevel(logger.ErrOrDebug(err)).
+				Err(err).
+				Str("Job", request.Job.ID).
+				Str("Strategy", strategyType).
+				Bool("Bid", resp.ShouldBid).
+				Bool("Wait", resp.ShouldWait).
+				Str("Reason", resp.Reason).
+				Send()
+		}
 
 		if err != nil {
 			// NB: failure here results in a callback to OnComputeFailure
diff --git a/pkg/compute/capacity/types.go b/pkg/compute/capacity/types.go
index 3f510bc2e7..530e104177 100644
--- a/pkg/compute/capacity/types.go
+++ b/pkg/compute/capacity/types.go
@@ -25,6 +25,17 @@ type Tracker interface {
 	Remove(ctx context.Context, usage models.Resources)
 }
 
+// UsageTracker keeps track of the current resource usage of the compute node.
+// Useful when tracking jobs in the queue pending and haven't started yet.
+type UsageTracker interface {
+	// Add adds the given resource usage to the tracker.
+	Add(ctx context.Context, usage models.Resources)
+	// Remove removes the given resource usage from the tracker.
+	Remove(ctx context.Context, usage models.Resources)
+	// GetUsedCapacity returns the current resource usage of the tracker
+	GetUsedCapacity(ctx context.Context) models.Resources
+}
+
 // UsageCalculator calculates the resource usage of a job.
 // Can also be used to populate the resource usage of a job with default values if not defined
 type UsageCalculator interface {
diff --git a/pkg/compute/capacity/usage_tracker.go b/pkg/compute/capacity/usage_tracker.go
new file mode 100644
index 0000000000..e47f29b115
--- /dev/null
+++ b/pkg/compute/capacity/usage_tracker.go
@@ -0,0 +1,38 @@
+package capacity
+
+import (
+	"context"
+	"sync"
+
+	"github.com/bacalhau-project/bacalhau/pkg/models"
+)
+
+// LocalUsageTracker keeps track of the resources used regardless of the total capacity.
+// It is useful when tracking jobs in the queue pending and haven't started yet.
+type LocalUsageTracker struct {
+	usedCapacity models.Resources
+	mu           sync.Mutex
+}
+
+func NewLocalUsageTracker() *LocalUsageTracker {
+	return &LocalUsageTracker{}
+}
+
+func (t *LocalUsageTracker) Add(ctx context.Context, usage models.Resources) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	t.usedCapacity = *t.usedCapacity.Add(usage)
+}
+
+func (t *LocalUsageTracker) Remove(ctx context.Context, usage models.Resources) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	t.usedCapacity = *t.usedCapacity.Sub(usage)
+}
+
+func (t *LocalUsageTracker) GetUsedCapacity(ctx context.Context) models.Resources {
+	return t.usedCapacity
+}
+
+// compile-time check that LocalUsageTracker implements Tracker
+var _ UsageTracker = (*LocalUsageTracker)(nil)
diff --git a/pkg/compute/executor_buffer.go b/pkg/compute/executor_buffer.go
index 0f6e99f411..aae0530e1c 100644
--- a/pkg/compute/executor_buffer.go
+++ b/pkg/compute/executor_buffer.go
@@ -31,7 +31,7 @@ type ExecutorBufferParams struct {
 	DelegateExecutor           Executor
 	Callback                   Callback
 	RunningCapacityTracker     capacity.Tracker
-	EnqueuedCapacityTracker    capacity.Tracker
+	EnqueuedUsageTracker       capacity.UsageTracker
 	DefaultJobExecutionTimeout time.Duration
 }
 
@@ -44,7 +44,7 @@ type ExecutorBufferParams struct {
 type ExecutorBuffer struct {
 	ID                         string
 	runningCapacity            capacity.Tracker
-	enqueuedCapacity           capacity.Tracker
+	enqueuedCapacity           capacity.UsageTracker
 	delegateService            Executor
 	callback                   Callback
 	running                    map[string]*bufferTask
@@ -61,7 +61,7 @@ func NewExecutorBuffer(params ExecutorBufferParams) *ExecutorBuffer {
 	r := &ExecutorBuffer{
 		ID:                         params.ID,
 		runningCapacity:            params.RunningCapacityTracker,
-		enqueuedCapacity:           params.EnqueuedCapacityTracker,
+		enqueuedCapacity:           params.EnqueuedUsageTracker,
 		delegateService:            params.DelegateExecutor,
 		callback:                   params.Callback,
 		running:                    make(map[string]*bufferTask),
@@ -108,18 +108,7 @@ func (s *ExecutorBuffer) Run(ctx context.Context, localExecutionState store.Loca
 		err = models.NewBaseError("execution %s already running", execution.ID)
 		return err
 	}
-	if added := s.enqueuedCapacity.AddIfHasCapacity(ctx, *execution.TotalAllocatedResources()); added == nil {
-		err = models.NewBaseError("not enough capacity to enqueue job").WithRetryable()
-		return err
-	} else {
-		// Update the execution to include all the resources that have
-		// actually been allocated. Effectively this is picking the GPU(s)
-		// that the job will use. Note that this is not persisted here, as
-		// it was based on current usage information which would change
-		// under a restart, so it will only persist if the job starts
-		execution.AllocateResources(execution.Job.Task().Name, *added)
-	}
-
+	s.enqueuedCapacity.Add(ctx, *execution.TotalAllocatedResources())
 	s.queuedTasks.Enqueue(newBufferTask(localExecutionState), int64(execution.Job.Priority))
 	s.deque()
 	return err
@@ -182,9 +171,9 @@ func (s *ExecutorBuffer) deque() {
 	for i := 0; i < max; i++ {
 		qitem := s.queuedTasks.DequeueWhere(func(task *bufferTask) bool {
 			// If we don't have enough resources to run this task, then we will skip it
-			queued := task.localExecutionState.Execution.TotalAllocatedResources()
-			added := s.runningCapacity.AddIfHasCapacity(ctx, *queued)
-			if added == nil {
+			queuedResources := task.localExecutionState.Execution.TotalAllocatedResources()
+			allocatedResources := s.runningCapacity.AddIfHasCapacity(ctx, *queuedResources)
+			if allocatedResources == nil {
 				return false
 			}
 
@@ -192,11 +181,11 @@ func (s *ExecutorBuffer) deque() {
 			// actually been allocated
 			task.localExecutionState.Execution.AllocateResources(
 				task.localExecutionState.Execution.Job.Task().Name,
-				*added,
+				*allocatedResources,
 			)
 
-			// Claim the resources now so that we don't count allocated resources
-			s.enqueuedCapacity.Remove(ctx, *queued)
+			// Claim the resources now so that we don't count queued resources
+			s.enqueuedCapacity.Remove(ctx, *queuedResources)
 			return true
 		})
 
diff --git a/pkg/compute/management_client.go b/pkg/compute/management_client.go
index 11bbb990e1..6be4427e08 100644
--- a/pkg/compute/management_client.go
+++ b/pkg/compute/management_client.go
@@ -16,14 +16,15 @@ import (
 )
 
 type ManagementClientParams struct {
-	NodeID               string
-	LabelsProvider       models.LabelsProvider
-	ManagementProxy      ManagementEndpoint
-	NodeInfoDecorator    models.NodeInfoDecorator
-	ResourceTracker      capacity.Tracker
-	RegistrationFilePath string
-	HeartbeatClient      *heartbeat.HeartbeatClient
-	ControlPlaneSettings types.ComputeControlPlaneConfig
+	NodeID                   string
+	LabelsProvider           models.LabelsProvider
+	ManagementProxy          ManagementEndpoint
+	NodeInfoDecorator        models.NodeInfoDecorator
+	AvailableCapacityTracker capacity.Tracker
+	QueueUsageTracker        capacity.UsageTracker
+	RegistrationFilePath     string
+	HeartbeatClient          *heartbeat.HeartbeatClient
+	ControlPlaneSettings     types.ComputeControlPlaneConfig
 }
 
 // ManagementClient is used to call management functions with
@@ -31,28 +32,30 @@ type ManagementClientParams struct {
 // it will periodically send an update to the requester node with
 // the latest node info for this node.
 type ManagementClient struct {
-	done              chan struct{}
-	labelsProvider    models.LabelsProvider
-	managementProxy   ManagementEndpoint
-	nodeID            string
-	nodeInfoDecorator models.NodeInfoDecorator
-	resourceTracker   capacity.Tracker
-	registrationFile  *RegistrationFile
-	heartbeatClient   *heartbeat.HeartbeatClient
-	settings          types.ComputeControlPlaneConfig
+	done                     chan struct{}
+	labelsProvider           models.LabelsProvider
+	managementProxy          ManagementEndpoint
+	nodeID                   string
+	nodeInfoDecorator        models.NodeInfoDecorator
+	availableCapacityTracker capacity.Tracker
+	queueUsageTracker        capacity.UsageTracker
+	registrationFile         *RegistrationFile
+	heartbeatClient          *heartbeat.HeartbeatClient
+	settings                 types.ComputeControlPlaneConfig
 }
 
 func NewManagementClient(params *ManagementClientParams) *ManagementClient {
 	return &ManagementClient{
-		done:              make(chan struct{}, 1),
-		labelsProvider:    params.LabelsProvider,
-		managementProxy:   params.ManagementProxy,
-		nodeID:            params.NodeID,
-		nodeInfoDecorator: params.NodeInfoDecorator,
-		registrationFile:  NewRegistrationFile(params.RegistrationFilePath),
-		resourceTracker:   params.ResourceTracker,
-		heartbeatClient:   params.HeartbeatClient,
-		settings:          params.ControlPlaneSettings,
+		done:                     make(chan struct{}, 1),
+		labelsProvider:           params.LabelsProvider,
+		managementProxy:          params.ManagementProxy,
+		nodeID:                   params.NodeID,
+		nodeInfoDecorator:        params.NodeInfoDecorator,
+		registrationFile:         NewRegistrationFile(params.RegistrationFilePath),
+		availableCapacityTracker: params.AvailableCapacityTracker,
+		queueUsageTracker:        params.QueueUsageTracker,
+		heartbeatClient:          params.HeartbeatClient,
+		settings:                 params.ControlPlaneSettings,
 	}
 }
 
@@ -121,13 +124,14 @@ func (m *ManagementClient) deliverInfo(ctx context.Context) {
 }
 
 func (m *ManagementClient) updateResources(ctx context.Context) {
-	log.Ctx(ctx).Debug().Msg("Sending updated resources")
+	request := requests.UpdateResourcesRequest{
+		NodeID:            m.nodeID,
+		AvailableCapacity: m.availableCapacityTracker.GetAvailableCapacity(ctx),
+		QueueUsedCapacity: m.queueUsageTracker.GetUsedCapacity(ctx),
+	}
+	log.Ctx(ctx).Debug().Msgf("Sending updated resources: %+v", request)
 
-	resources := m.resourceTracker.GetAvailableCapacity(ctx)
-	_, err := m.managementProxy.UpdateResources(ctx, requests.UpdateResourcesRequest{
-		NodeID:    m.nodeID,
-		Resources: resources,
-	})
+	_, err := m.managementProxy.UpdateResources(ctx, request)
 	if err != nil {
 		log.Ctx(ctx).Warn().Err(err).Msg("failed to send resource update to requester node")
 	}
diff --git a/pkg/compute/node_info_decorator.go b/pkg/compute/node_info_decorator.go
index fe848c482b..cd6c066d18 100644
--- a/pkg/compute/node_info_decorator.go
+++ b/pkg/compute/node_info_decorator.go
@@ -11,31 +11,34 @@ import (
 )
 
 type NodeInfoDecoratorParams struct {
-	Executors          executor.ExecutorProvider
-	Publisher          publisher.PublisherProvider
-	Storages           storage.StorageProvider
-	CapacityTracker    capacity.Tracker
-	ExecutorBuffer     *ExecutorBuffer
-	MaxJobRequirements models.Resources
+	Executors              executor.ExecutorProvider
+	Publisher              publisher.PublisherProvider
+	Storages               storage.StorageProvider
+	RunningCapacityTracker capacity.Tracker
+	QueueCapacityTracker   capacity.UsageTracker
+	ExecutorBuffer         *ExecutorBuffer
+	MaxJobRequirements     models.Resources
 }
 
 type NodeInfoDecorator struct {
-	executors          executor.ExecutorProvider
-	publishers         publisher.PublisherProvider
-	storages           storage.StorageProvider
-	capacityTracker    capacity.Tracker
-	executorBuffer     *ExecutorBuffer
-	maxJobRequirements models.Resources
+	executors              executor.ExecutorProvider
+	publishers             publisher.PublisherProvider
+	storages               storage.StorageProvider
+	runningCapacityTracker capacity.Tracker
+	queueCapacityTracker   capacity.UsageTracker
+	executorBuffer         *ExecutorBuffer
+	maxJobRequirements     models.Resources
 }
 
 func NewNodeInfoDecorator(params NodeInfoDecoratorParams) *NodeInfoDecorator {
 	return &NodeInfoDecorator{
-		executors:          params.Executors,
-		publishers:         params.Publisher,
-		storages:           params.Storages,
-		capacityTracker:    params.CapacityTracker,
-		executorBuffer:     params.ExecutorBuffer,
-		maxJobRequirements: params.MaxJobRequirements,
+		executors:              params.Executors,
+		publishers:             params.Publisher,
+		storages:               params.Storages,
+		runningCapacityTracker: params.RunningCapacityTracker,
+		queueCapacityTracker:   params.QueueCapacityTracker,
+		executorBuffer:         params.ExecutorBuffer,
+		maxJobRequirements:     params.MaxJobRequirements,
 	}
 }
 
@@ -45,8 +48,9 @@ func (n *NodeInfoDecorator) DecorateNodeInfo(ctx context.Context, nodeInfo model
 		ExecutionEngines:   n.executors.Keys(ctx),
 		Publishers:         n.publishers.Keys(ctx),
 		StorageSources:     n.storages.Keys(ctx),
-		MaxCapacity:        n.capacityTracker.GetMaxCapacity(ctx),
-		AvailableCapacity:  n.capacityTracker.GetAvailableCapacity(ctx),
+		MaxCapacity:        n.runningCapacityTracker.GetMaxCapacity(ctx),
+		AvailableCapacity:  n.runningCapacityTracker.GetAvailableCapacity(ctx),
+		QueueUsedCapacity:  n.queueCapacityTracker.GetUsedCapacity(ctx),
 		MaxJobRequirements: n.maxJobRequirements,
 		RunningExecutions:  len(n.executorBuffer.RunningExecutions()),
 		EnqueuedExecutions: n.executorBuffer.EnqueuedExecutionsCount(),
diff --git a/pkg/config/configenv/dev.go b/pkg/config/configenv/dev.go
index 723d6ded4e..10da50ac14 100644
--- a/pkg/config/configenv/dev.go
+++ b/pkg/config/configenv/dev.go
@@ -110,12 +110,6 @@ var DevelopmentComputeConfig = types.ComputeConfig{
 			Disk:   "",
 			GPU:    "",
 		},
-		QueueResourceLimits: models.ResourcesConfig{
-			CPU:    "",
-			Memory: "",
-			Disk:   "",
-			GPU:    "",
-		},
 	},
 	ExecutionStore: types.JobStoreConfig{
 		Type: types.BoltDB,
@@ -135,7 +129,6 @@ var DevelopmentComputeConfig = types.ComputeConfig{
 		ProbeHTTP:           "",
 		ProbeExec:           "",
 	},
-	Queue: types.QueueConfig{},
 	Logging: types.LoggingConfig{
 		LogRunningExecutionsInterval: types.Duration(10 * time.Second),
 	},
diff --git a/pkg/config/configenv/local.go b/pkg/config/configenv/local.go
index 3846b98814..eed25b4377 100644
--- a/pkg/config/configenv/local.go
+++ b/pkg/config/configenv/local.go
@@ -101,12 +101,6 @@ var LocalComputeConfig = types.ComputeConfig{
 			Disk:   "",
 			GPU:    "",
 		},
-		QueueResourceLimits: models.ResourcesConfig{
-			CPU:    "",
-			Memory: "",
-			Disk:   "",
-			GPU:    "",
-		},
 	},
 	ExecutionStore: types.JobStoreConfig{
 		Type: types.BoltDB,
@@ -126,7 +120,6 @@ var LocalComputeConfig = types.ComputeConfig{
 		ProbeHTTP:           "",
 		ProbeExec:           "",
 	},
-	Queue: types.QueueConfig{},
 	Logging: types.LoggingConfig{
 		LogRunningExecutionsInterval: types.Duration(10 * time.Second),
 	},
diff --git a/pkg/config/configenv/production.go b/pkg/config/configenv/production.go
index c3dbc25efb..9707b5c0fc 100644
--- a/pkg/config/configenv/production.go
+++ b/pkg/config/configenv/production.go
@@ -118,12 +118,6 @@ var ProductionComputeConfig = types.ComputeConfig{
 			Disk:   "",
 			GPU:    "",
 		},
-		QueueResourceLimits: models.ResourcesConfig{
-			CPU:    "",
-			Memory: "",
-			Disk:   "",
-			GPU:    "",
-		},
 	},
 	ExecutionStore: types.JobStoreConfig{
 		Type: types.BoltDB,
@@ -143,7 +137,6 @@ var ProductionComputeConfig = types.ComputeConfig{
 		ProbeHTTP:           "",
 		ProbeExec:           "",
 	},
-	Queue: types.QueueConfig{},
 	Logging: types.LoggingConfig{
 		LogRunningExecutionsInterval: types.Duration(10 * time.Second),
 	},
diff --git a/pkg/config/configenv/staging.go b/pkg/config/configenv/staging.go
index c2d2e03282..1651678bbb 100644
--- a/pkg/config/configenv/staging.go
+++ b/pkg/config/configenv/staging.go
@@ -116,12 +116,6 @@ var StagingComputeConfig = types.ComputeConfig{
 			Disk:   "",
 			GPU:    "",
 		},
-		QueueResourceLimits: models.ResourcesConfig{
-			CPU:    "",
-			Memory: "",
-			Disk:   "",
-			GPU:    "",
-		},
 	},
 	ExecutionStore: types.JobStoreConfig{
 		Type: types.BoltDB,
@@ -141,7 +135,6 @@ var StagingComputeConfig = types.ComputeConfig{
 		ProbeHTTP:           "",
 		ProbeExec:           "",
 	},
-	Queue: types.QueueConfig{},
 	Logging: types.LoggingConfig{
 		LogRunningExecutionsInterval: types.Duration(10 * time.Second),
 	},
diff --git a/pkg/config/types/compute.go b/pkg/config/types/compute.go
index 5e6dde5603..5d69d2818b 100644
--- a/pkg/config/types/compute.go
+++ b/pkg/config/types/compute.go
@@ -10,7 +10,6 @@ type ComputeConfig struct {
 	ExecutionStore       JobStoreConfig            `yaml:"ExecutionStore"`
 	JobTimeouts          JobTimeoutConfig          `yaml:"JobTimeouts"`
 	JobSelection         model.JobSelectionPolicy  `yaml:"JobSelection"`
-	Queue                QueueConfig               `yaml:"Queue"`
 	Logging              LoggingConfig             `yaml:"Logging"`
 	ManifestCache        DockerCacheConfig         `yaml:"ManifestCache"`
 	LogStreamConfig      LogStreamConfig           `yaml:"LogStream"`
@@ -25,7 +24,6 @@ type CapacityConfig struct {
 	// Per job amount of resource the system can be using at one time.
 	JobResourceLimits        models.ResourcesConfig `yaml:"JobResourceLimits"`
 	DefaultJobResourceLimits models.ResourcesConfig `yaml:"DefaultJobResourceLimits"`
-	QueueResourceLimits      models.ResourcesConfig `yaml:"QueueResourceLimits"`
 }
 
 type JobTimeoutConfig struct {
@@ -45,9 +43,6 @@ type JobTimeoutConfig struct {
 	DefaultJobExecutionTimeout Duration `yaml:"DefaultJobExecutionTimeout"`
 }
 
-type QueueConfig struct {
-}
-
 type LoggingConfig struct {
 	// logging running executions
 	LogRunningExecutionsInterval Duration `yaml:"LogRunningExecutionsInterval"`
diff --git a/pkg/config/types/generated_constants.go b/pkg/config/types/generated_constants.go
index ef397a06c3..bd21b2e53b 100644
--- a/pkg/config/types/generated_constants.go
+++ b/pkg/config/types/generated_constants.go
@@ -62,11 +62,6 @@ const NodeComputeCapacityDefaultJobResourceLimitsCPU = "Node.Compute.Capacity.De
 const NodeComputeCapacityDefaultJobResourceLimitsMemory = "Node.Compute.Capacity.DefaultJobResourceLimits.Memory"
 const NodeComputeCapacityDefaultJobResourceLimitsDisk = "Node.Compute.Capacity.DefaultJobResourceLimits.Disk"
 const NodeComputeCapacityDefaultJobResourceLimitsGPU = "Node.Compute.Capacity.DefaultJobResourceLimits.GPU"
-const NodeComputeCapacityQueueResourceLimits = "Node.Compute.Capacity.QueueResourceLimits"
-const NodeComputeCapacityQueueResourceLimitsCPU = "Node.Compute.Capacity.QueueResourceLimits.CPU"
-const NodeComputeCapacityQueueResourceLimitsMemory = "Node.Compute.Capacity.QueueResourceLimits.Memory"
-const NodeComputeCapacityQueueResourceLimitsDisk = "Node.Compute.Capacity.QueueResourceLimits.Disk"
-const NodeComputeCapacityQueueResourceLimitsGPU = "Node.Compute.Capacity.QueueResourceLimits.GPU"
 const NodeComputeExecutionStore = "Node.Compute.ExecutionStore"
 const NodeComputeExecutionStoreType = "Node.Compute.ExecutionStore.Type"
 const NodeComputeExecutionStorePath = "Node.Compute.ExecutionStore.Path"
@@ -82,7 +77,6 @@ const NodeComputeJobSelectionRejectStatelessJobs = "Node.Compute.JobSelection.Re
 const NodeComputeJobSelectionAcceptNetworkedJobs = "Node.Compute.JobSelection.AcceptNetworkedJobs"
 const NodeComputeJobSelectionProbeHTTP = "Node.Compute.JobSelection.ProbeHTTP"
 const NodeComputeJobSelectionProbeExec = "Node.Compute.JobSelection.ProbeExec"
-const NodeComputeQueue = "Node.Compute.Queue"
 const NodeComputeLogging = "Node.Compute.Logging"
 const NodeComputeLoggingLogRunningExecutionsInterval = "Node.Compute.Logging.LogRunningExecutionsInterval"
 const NodeComputeManifestCache = "Node.Compute.ManifestCache"
diff --git a/pkg/config/types/generated_viper_defaults.go b/pkg/config/types/generated_viper_defaults.go
index 75217fd072..1d14a8b439 100644
--- a/pkg/config/types/generated_viper_defaults.go
+++ b/pkg/config/types/generated_viper_defaults.go
@@ -84,11 +84,6 @@ func SetDefaults(cfg BacalhauConfig, opts ...SetOption) {
 	p.Viper.SetDefault(NodeComputeCapacityDefaultJobResourceLimitsMemory, cfg.Node.Compute.Capacity.DefaultJobResourceLimits.Memory)
 	p.Viper.SetDefault(NodeComputeCapacityDefaultJobResourceLimitsDisk, cfg.Node.Compute.Capacity.DefaultJobResourceLimits.Disk)
 	p.Viper.SetDefault(NodeComputeCapacityDefaultJobResourceLimitsGPU, cfg.Node.Compute.Capacity.DefaultJobResourceLimits.GPU)
-	p.Viper.SetDefault(NodeComputeCapacityQueueResourceLimits, cfg.Node.Compute.Capacity.QueueResourceLimits)
-	p.Viper.SetDefault(NodeComputeCapacityQueueResourceLimitsCPU, cfg.Node.Compute.Capacity.QueueResourceLimits.CPU)
-	p.Viper.SetDefault(NodeComputeCapacityQueueResourceLimitsMemory, cfg.Node.Compute.Capacity.QueueResourceLimits.Memory)
-	p.Viper.SetDefault(NodeComputeCapacityQueueResourceLimitsDisk, cfg.Node.Compute.Capacity.QueueResourceLimits.Disk)
-	p.Viper.SetDefault(NodeComputeCapacityQueueResourceLimitsGPU, cfg.Node.Compute.Capacity.QueueResourceLimits.GPU)
 	p.Viper.SetDefault(NodeComputeExecutionStore, cfg.Node.Compute.ExecutionStore)
 	p.Viper.SetDefault(NodeComputeExecutionStoreType, cfg.Node.Compute.ExecutionStore.Type)
 	p.Viper.SetDefault(NodeComputeExecutionStorePath, cfg.Node.Compute.ExecutionStore.Path)
@@ -104,7 +99,6 @@ func SetDefaults(cfg BacalhauConfig, opts ...SetOption) {
 	p.Viper.SetDefault(NodeComputeJobSelectionAcceptNetworkedJobs, cfg.Node.Compute.JobSelection.AcceptNetworkedJobs)
 	p.Viper.SetDefault(NodeComputeJobSelectionProbeHTTP, cfg.Node.Compute.JobSelection.ProbeHTTP)
 	p.Viper.SetDefault(NodeComputeJobSelectionProbeExec, cfg.Node.Compute.JobSelection.ProbeExec)
-	p.Viper.SetDefault(NodeComputeQueue, cfg.Node.Compute.Queue)
 	p.Viper.SetDefault(NodeComputeLogging, cfg.Node.Compute.Logging)
 	p.Viper.SetDefault(NodeComputeLoggingLogRunningExecutionsInterval, cfg.Node.Compute.Logging.LogRunningExecutionsInterval.AsTimeDuration())
 	p.Viper.SetDefault(NodeComputeManifestCache, cfg.Node.Compute.ManifestCache)
@@ -282,11 +276,6 @@ func Set(cfg BacalhauConfig, opts ...SetOption) {
 	p.Viper.Set(NodeComputeCapacityDefaultJobResourceLimitsMemory, cfg.Node.Compute.Capacity.DefaultJobResourceLimits.Memory)
 	p.Viper.Set(NodeComputeCapacityDefaultJobResourceLimitsDisk, cfg.Node.Compute.Capacity.DefaultJobResourceLimits.Disk)
 	p.Viper.Set(NodeComputeCapacityDefaultJobResourceLimitsGPU, cfg.Node.Compute.Capacity.DefaultJobResourceLimits.GPU)
-	p.Viper.Set(NodeComputeCapacityQueueResourceLimits, cfg.Node.Compute.Capacity.QueueResourceLimits)
-	p.Viper.Set(NodeComputeCapacityQueueResourceLimitsCPU, cfg.Node.Compute.Capacity.QueueResourceLimits.CPU)
-	p.Viper.Set(NodeComputeCapacityQueueResourceLimitsMemory, cfg.Node.Compute.Capacity.QueueResourceLimits.Memory)
-	p.Viper.Set(NodeComputeCapacityQueueResourceLimitsDisk, cfg.Node.Compute.Capacity.QueueResourceLimits.Disk)
-	p.Viper.Set(NodeComputeCapacityQueueResourceLimitsGPU, cfg.Node.Compute.Capacity.QueueResourceLimits.GPU)
 	p.Viper.Set(NodeComputeExecutionStore, cfg.Node.Compute.ExecutionStore)
 	p.Viper.Set(NodeComputeExecutionStoreType, cfg.Node.Compute.ExecutionStore.Type)
 	p.Viper.Set(NodeComputeExecutionStorePath, cfg.Node.Compute.ExecutionStore.Path)
@@ -302,7 +291,6 @@ func Set(cfg BacalhauConfig, opts ...SetOption) {
 	p.Viper.Set(NodeComputeJobSelectionAcceptNetworkedJobs, cfg.Node.Compute.JobSelection.AcceptNetworkedJobs)
 	p.Viper.Set(NodeComputeJobSelectionProbeHTTP, cfg.Node.Compute.JobSelection.ProbeHTTP)
 	p.Viper.Set(NodeComputeJobSelectionProbeExec, cfg.Node.Compute.JobSelection.ProbeExec)
-	p.Viper.Set(NodeComputeQueue, cfg.Node.Compute.Queue)
 	p.Viper.Set(NodeComputeLogging, cfg.Node.Compute.Logging)
 	p.Viper.Set(NodeComputeLoggingLogRunningExecutionsInterval, cfg.Node.Compute.Logging.LogRunningExecutionsInterval.AsTimeDuration())
 	p.Viper.Set(NodeComputeManifestCache, cfg.Node.Compute.ManifestCache)
diff --git a/pkg/models/node_info.go b/pkg/models/node_info.go
index 793ef499ca..b61e45f133 100644
--- a/pkg/models/node_info.go
+++ b/pkg/models/node_info.go
@@ -109,6 +109,7 @@ type ComputeNodeInfo struct {
 	Publishers         []string  `json:"Publishers"`
 	StorageSources     []string  `json:"StorageSources"`
 	MaxCapacity        Resources `json:"MaxCapacity"`
+	QueueUsedCapacity  Resources `json:"QueueCapacity"`
 	AvailableCapacity  Resources `json:"AvailableCapacity"`
 	MaxJobRequirements Resources `json:"MaxJobRequirements"`
 	RunningExecutions  int       `json:"RunningExecutions"`
diff --git a/pkg/models/requests/management.go b/pkg/models/requests/management.go
index 153f30a919..1a3c9d98a7 100644
--- a/pkg/models/requests/management.go
+++ b/pkg/models/requests/management.go
@@ -20,8 +20,9 @@ type UpdateInfoResponse struct {
 }
 
 type UpdateResourcesRequest struct {
-	NodeID    string
-	Resources models.Resources
+	NodeID            string
+	AvailableCapacity models.Resources
+	QueueUsedCapacity models.Resources
 }
 
 type UpdateResourcesResponse struct{}
diff --git a/pkg/node/compute.go b/pkg/node/compute.go
index 0a2fe69032..8ec7769dd8 100644
--- a/pkg/node/compute.go
+++ b/pkg/node/compute.go
@@ -67,9 +67,7 @@ func NewComputeNode(
 	runningCapacityTracker := capacity.NewLocalTracker(capacity.LocalTrackerParams{
 		MaxCapacity: config.TotalResourceLimits,
 	})
-	enqueuedCapacityTracker := capacity.NewLocalTracker(capacity.LocalTrackerParams{
-		MaxCapacity: config.QueueResourceLimits,
-	})
+	enqueuedUsageTracker := capacity.NewLocalUsageTracker()
 
 	resultsPath, err := compute.NewResultsPath()
 	if err != nil {
@@ -92,7 +90,7 @@ func NewComputeNode(
 		DelegateExecutor:           baseExecutor,
 		Callback:                   computeCallback,
 		RunningCapacityTracker:     runningCapacityTracker,
-		EnqueuedCapacityTracker:    enqueuedCapacityTracker,
+		EnqueuedUsageTracker:       enqueuedUsageTracker,
 		DefaultJobExecutionTimeout: config.DefaultJobExecutionTimeout,
 	})
 	runningInfoProvider := sensors.NewRunningExecutionsInfoProvider(sensors.RunningExecutionsInfoProviderParams{
@@ -133,12 +131,13 @@ func NewComputeNode(
 
 	// node info
 	nodeInfoDecorator := compute.NewNodeInfoDecorator(compute.NodeInfoDecoratorParams{
-		Executors:          executors,
-		Publisher:          publishers,
-		Storages:           storages,
-		CapacityTracker:    runningCapacityTracker,
-		ExecutorBuffer:     bufferRunner,
-		MaxJobRequirements: config.JobResourceLimits,
+		Executors:              executors,
+		Publisher:              publishers,
+		Storages:               storages,
+		RunningCapacityTracker: runningCapacityTracker,
+		QueueCapacityTracker:   enqueuedUsageTracker,
+		ExecutorBuffer:         bufferRunner,
+		MaxJobRequirements:     config.JobResourceLimits,
 	})
 
 	bidder := NewBidder(config,
@@ -146,7 +145,6 @@ func NewComputeNode(
 		storages,
 		executors,
 		runningCapacityTracker,
-		enqueuedCapacityTracker,
 		nodeID,
 		executionStore,
 		computeCallback,
@@ -205,14 +203,15 @@ func NewComputeNode(
 		// with the requester node, and then if successful will send regular node
 		// info updates.
 		managementClient = compute.NewManagementClient(&compute.ManagementClientParams{
-			NodeID:               nodeID,
-			LabelsProvider:       labelsProvider,
-			ManagementProxy:      managementProxy,
-			NodeInfoDecorator:    nodeInfoDecorator,
-			RegistrationFilePath: regFilename,
-			ResourceTracker:      runningCapacityTracker,
-			HeartbeatClient:      heartbeatClient,
-			ControlPlaneSettings: config.ControlPlaneSettings,
+			NodeID:                   nodeID,
+			LabelsProvider:           labelsProvider,
+			ManagementProxy:          managementProxy,
+			NodeInfoDecorator:        nodeInfoDecorator,
+			RegistrationFilePath:     regFilename,
+			AvailableCapacityTracker: runningCapacityTracker,
+			QueueUsageTracker:        enqueuedUsageTracker,
+			HeartbeatClient:          heartbeatClient,
+			ControlPlaneSettings:     config.ControlPlaneSettings,
 		})
 		if err := managementClient.RegisterNode(ctx); err != nil {
 			return nil, fmt.Errorf("failed to register node with requester: %s", err)
@@ -256,7 +255,6 @@ func NewBidder(
 	storages storage.StorageProvider,
 	executors executor.ExecutorProvider,
 	runningCapacityTracker capacity.Tracker,
-	enqueuedCapacityTracker capacity.Tracker,
 	nodeID string,
 	executionStore store.ExecutionStore,
 	computeCallback compute.Callback,
@@ -303,10 +301,6 @@ func NewBidder(
 			resource.NewMaxCapacityStrategy(resource.MaxCapacityStrategyParams{
 				MaxJobRequirements: config.JobResourceLimits,
 			}),
-			resource.NewAvailableCapacityStrategy(resource.AvailableCapacityStrategyParams{
-				RunningCapacityTracker:  runningCapacityTracker,
-				EnqueuedCapacityTracker: enqueuedCapacityTracker,
-			}),
 			executor_util.NewExecutorSpecificBidStrategy(executors),
 		}
 	} else {
diff --git a/pkg/node/config_compute.go b/pkg/node/config_compute.go
index 25e739180f..44a0ed908d 100644
--- a/pkg/node/config_compute.go
+++ b/pkg/node/config_compute.go
@@ -44,7 +44,6 @@ type JobSelectionPolicy struct {
 type ComputeConfigParams struct {
 	// Capacity config
 	TotalResourceLimits          models.Resources
-	QueueResourceLimits          models.Resources
 	JobResourceLimits            models.Resources
 	DefaultJobResourceLimits     models.Resources
 	PhysicalResourcesProvider    capacity.Provider
@@ -82,7 +81,6 @@ type ComputeConfigParams struct {
 type ComputeConfig struct {
 	// Capacity config
 	TotalResourceLimits          models.Resources
-	QueueResourceLimits          models.Resources
 	JobResourceLimits            models.Resources
 	DefaultJobResourceLimits     models.Resources
 	IgnorePhysicalResourceLimits bool
@@ -189,19 +187,12 @@ func NewComputeConfigWith(params ComputeConfigParams) (ComputeConfig, error) {
 		Merge(DefaultComputeConfig.JobResourceLimits).
 		Merge(*totalResourceLimits)
 
-	// by default set the queue size to the total resource limits, which allows the node overcommit to double of the total resource limits.
-	// i.e. total resource limits can be busy in running state, and enqueue up to the total resource limits in the queue.
-	if params.QueueResourceLimits.IsZero() {
-		params.QueueResourceLimits = *totalResourceLimits
-	}
-
 	// populate default job resource limits with default values and job resource limits if not set
 	defaultJobResourceLimits := params.DefaultJobResourceLimits.
 		Merge(DefaultComputeConfig.DefaultJobResourceLimits)
 
 	config := ComputeConfig{
 		TotalResourceLimits:          *totalResourceLimits,
-		QueueResourceLimits:          params.QueueResourceLimits,
 		JobResourceLimits:            *jobResourceLimits,
 		DefaultJobResourceLimits:     *defaultJobResourceLimits,
 		IgnorePhysicalResourceLimits: params.IgnorePhysicalResourceLimits,
@@ -247,12 +238,6 @@ func validateConfig(config ComputeConfig, physicalResources models.Resources) er
 				config.JobResourceLimits, config.TotalResourceLimits))
 	}
 
-	if !config.JobResourceLimits.LessThanEq(config.QueueResourceLimits) {
-		err = errors.Join(err,
-			fmt.Errorf("job resource limits %+v exceed queue size limits %+v, which will prevent processing the job",
-				config.JobResourceLimits, config.QueueResourceLimits))
-	}
-
 	if !config.DefaultJobResourceLimits.LessThanEq(config.JobResourceLimits) {
 		err = errors.Join(err,
 			fmt.Errorf("default job resource limits %+v exceed job resource limits %+v",
diff --git a/pkg/node/manager/node_manager.go b/pkg/node/manager/node_manager.go
index bb6079c310..78d26b5968 100644
--- a/pkg/node/manager/node_manager.go
+++ b/pkg/node/manager/node_manager.go
@@ -20,13 +20,18 @@ const (
 	resourceMapLockCount = 32
 )
 
+type trackedResources struct {
+	availableCapacity models.Resources
+	queueUsedCapacity models.Resources
+}
+
 // NodeManager is responsible for managing compute nodes and their
 // membership within the cluster through the entire lifecycle. It
 // also provides operations for querying and managing compute
 // node information.
 type NodeManager struct {
 	store                routing.NodeInfoStore
-	resourceMap          *concurrency.StripedMap[models.Resources]
+	resourceMap          *concurrency.StripedMap[trackedResources]
 	heartbeats           *heartbeat.HeartbeatServer
 	defaultApprovalState models.NodeMembershipState
 }
@@ -41,7 +46,7 @@ type NodeManagerParams struct {
 // to the structure.
 func NewNodeManager(params NodeManagerParams) *NodeManager {
 	return &NodeManager{
-		resourceMap:          concurrency.NewStripedMap[models.Resources](resourceMapLockCount),
+		resourceMap:          concurrency.NewStripedMap[trackedResources](resourceMapLockCount),
 		store:                params.NodeInfo,
 		heartbeats:           params.Heartbeats,
 		defaultApprovalState: params.DefaultApprovalState,
@@ -154,11 +159,14 @@ func (n *NodeManager) UpdateResources(ctx context.Context,
 		return &requests.UpdateResourcesResponse{}, nil
 	}
 
-	log.Ctx(ctx).Debug().Msg("updating resources availability for node")
+	log.Ctx(ctx).Debug().Msgf("updating node resources availability: %+v", request)
 
 	// Update the resources for the node in the stripedmap. This is a thread-safe operation as locking
 	// is handled by the stripedmap on a per-bucket basis.
-	n.resourceMap.Put(request.NodeID, request.Resources)
+	n.resourceMap.Put(request.NodeID, trackedResources{
+		availableCapacity: request.AvailableCapacity,
+		queueUsedCapacity: request.QueueUsedCapacity,
+	})
 	return &requests.UpdateResourcesResponse{}, nil
 }
 
@@ -174,7 +182,8 @@ func (n *NodeManager) Add(ctx context.Context, nodeInfo models.NodeState) error
 func (n *NodeManager) addToInfo(ctx context.Context, state *models.NodeState) {
 	resources, found := n.resourceMap.Get(state.Info.NodeID)
 	if found && state.Info.ComputeNodeInfo != nil {
-		state.Info.ComputeNodeInfo.AvailableCapacity = resources
+		state.Info.ComputeNodeInfo.AvailableCapacity = resources.availableCapacity
+		state.Info.ComputeNodeInfo.QueueUsedCapacity = resources.queueUsedCapacity
 	}
 
 	if n.heartbeats != nil {
diff --git a/pkg/node/requester.go b/pkg/node/requester.go
index 85a77c08b2..a15f726a1d 100644
--- a/pkg/node/requester.go
+++ b/pkg/node/requester.go
@@ -93,6 +93,7 @@ func NewRequesterNode(
 		ranking.NewMaxUsageNodeRanker(),
 		ranking.NewMinVersionNodeRanker(ranking.MinVersionNodeRankerParams{MinVersion: requesterConfig.MinBacalhauVersion}),
 		ranking.NewPreviousExecutionsNodeRanker(ranking.PreviousExecutionsNodeRankerParams{JobStore: jobStore}),
+		ranking.NewAvailableCapacityNodeRanker(),
 		// arbitrary rankers
 		ranking.NewRandomNodeRanker(ranking.RandomNodeRankerParams{
 			RandomnessRange: requesterConfig.NodeRankRandomnessRange,
diff --git a/pkg/orchestrator/selection/ranking/available_capacity.go b/pkg/orchestrator/selection/ranking/available_capacity.go
new file mode 100644
index 0000000000..8c8a8de547
--- /dev/null
+++ b/pkg/orchestrator/selection/ranking/available_capacity.go
@@ -0,0 +1,185 @@
+package ranking
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/rs/zerolog/log"
+
+	"github.com/bacalhau-project/bacalhau/pkg/lib/math"
+	"github.com/bacalhau-project/bacalhau/pkg/models"
+	"github.com/bacalhau-project/bacalhau/pkg/orchestrator"
+)
+
+// Constants to normalize resource values to a comparable scale.
+const (
+	// Scaling factors for different resource types to normalize them to a comparable scale.
+	memoryScale = 1e6 // Scale memory to megabytes
+	diskScale   = 1e9 // Scale disk to gigabytes
+	gpuScale    = 1   // No scaling needed for GPU as it's already in units
+	cpuScale    = 1   // No scaling needed for CPU as it's already in cores
+
+	// Weights for different resource types. These values are chosen to reflect
+	// the relative importance of each resource type in the context of node capacity.
+	// CPU is considered the most critical resource, followed by memory, disk, and GPU.
+	defaultCPUWeight    = 0.4 // CPU has the highest weight as it directly affects computation speed.
+	defaultMemoryWeight = 0.3 // Memory is crucial for handling large datasets and multitasking.
+	defaultDiskWeight   = 0.2 // Disk space is important but less critical than CPU and memory for most tasks.
+	defaultGPUWeight    = 0.1 // GPU is given the lowest weight as not all tasks require GPU processing.
+
+	// Rank range constants
+	maxAvailableCapacityRank = 80
+	maxQueueCapacityRank     = 20
+)
+
+// resourceWeights struct to hold resource weights.
+type resourceWeights struct {
+	cpuWeight    float64
+	memoryWeight float64
+	diskWeight   float64
+	gpuWeight    float64
+}
+
+// defaultResourceWeights returns the default resource weights.
+var defaultResourceWeights = resourceWeights{
+	cpuWeight:    defaultCPUWeight,
+	memoryWeight: defaultMemoryWeight,
+	diskWeight:   defaultDiskWeight,
+	gpuWeight:    defaultGPUWeight,
+}
+
+// AvailableCapacityNodeRanker ranks nodes based on their available capacity and queue used capacity.
+type AvailableCapacityNodeRanker struct{}
+
+// NewAvailableCapacityNodeRanker creates a new instance of AvailableCapacityNodeRanker.
+func NewAvailableCapacityNodeRanker() *AvailableCapacityNodeRanker {
+	return &AvailableCapacityNodeRanker{}
+}
+
+// dynamicWeights calculates the weights for resources based on the job requirements.
+func dynamicWeights(jobRequirements *models.Resources) resourceWeights {
+	// Normalize the resource values
+	normalizedCPU := jobRequirements.CPU / cpuScale
+	normalizedMemory := float64(jobRequirements.Memory) / memoryScale
+	normalizedDisk := float64(jobRequirements.Disk) / diskScale
+	normalizedGPU := float64(jobRequirements.GPU) / gpuScale
+
+	// Calculate the total normalized resource value
+	total := normalizedCPU + normalizedMemory + normalizedDisk + normalizedGPU
+	if total == 0 {
+		// Return default weights if job requirements are all zero
+		return defaultResourceWeights
+	}
+
+	// Calculate and return dynamic weights based on normalized resource values
+	return resourceWeights{
+		cpuWeight:    normalizedCPU / total,
+		memoryWeight: normalizedMemory / total,
+		diskWeight:   normalizedDisk / total,
+		gpuWeight:    normalizedGPU / total,
+	}
+}
+
+func weightedCapacity(resources models.Resources, weights resourceWeights) float64 {
+	normalizedCPU := resources.CPU / cpuScale
+	normalizedMemory := float64(resources.Memory) / memoryScale
+	normalizedDisk := float64(resources.Disk) / diskScale
+	normalizedGPU := float64(resources.GPU) / gpuScale
+
+	return (normalizedCPU * weights.cpuWeight) +
+		(normalizedMemory * weights.memoryWeight) +
+		(normalizedDisk * weights.diskWeight) +
+		(normalizedGPU * weights.gpuWeight)
+}
+
+// calculateWeightedCapacities calculates the weighted capacities for each node and determines the maximum values
+func (s *AvailableCapacityNodeRanker) calculateWeightedCapacities(nodes []models.NodeInfo, weights resourceWeights) (
+	map[string]float64, map[string]float64, float64, float64) {
+	var maxWeightedAvailableCapacity, maxQueueUsedCapacity float64
+	weightedAvailableCapacities := make(map[string]float64, len(nodes))
+	weightedQueueCapacities := make(map[string]float64, len(nodes))
+
+	for _, node := range nodes {
+		weightedAvailableCapacity := weightedCapacity(node.ComputeNodeInfo.AvailableCapacity, weights)
+		weightedQueueUsedCapacity := weightedCapacity(node.ComputeNodeInfo.QueueUsedCapacity, weights)
+
+		weightedAvailableCapacities[node.NodeID] = weightedAvailableCapacity
+		weightedQueueCapacities[node.NodeID] = weightedQueueUsedCapacity
+
+		if weightedAvailableCapacity > maxWeightedAvailableCapacity {
+			maxWeightedAvailableCapacity = weightedAvailableCapacity
+		}
+		if weightedQueueUsedCapacity > maxQueueUsedCapacity {
+			maxQueueUsedCapacity = weightedQueueUsedCapacity
+		}
+	}
+
+	return weightedAvailableCapacities, weightedQueueCapacities, maxWeightedAvailableCapacity, maxQueueUsedCapacity
+}
+
+// rankNodesBasedOnCapacities ranks nodes based on normalized weighted capacities
+func (s *AvailableCapacityNodeRanker) rankNodesBasedOnCapacities(ctx context.Context, nodes []models.NodeInfo,
+	wAvailableCapacities, wQueueCapacities map[string]float64, maxAvailableCapacity, maxQueueCapacity float64) (
+	[]orchestrator.NodeRank, error) {
+	ranks := make([]orchestrator.NodeRank, len(nodes))
+
+	for i, node := range nodes {
+		weightedAvailableCapacity := wAvailableCapacities[node.NodeID]
+		weightedQueueUsedCapacity := wQueueCapacities[node.NodeID]
+
+		// Calculate the ratios of available and queue capacities
+		availableRatio := 0.0
+		queueRatio := 0.0
+
+		if maxAvailableCapacity > 0 {
+			availableRatio = weightedAvailableCapacity / maxAvailableCapacity
+		}
+		if maxQueueCapacity > 0 {
+			queueRatio = weightedQueueUsedCapacity / maxQueueCapacity
+		}
+
+		// Normalize the ratios to the rank range
+		normalizedAvailableRank := availableRatio * float64(maxAvailableCapacityRank)
+		normalizedQueueRank := (1 - queueRatio) * float64(maxQueueCapacityRank)
+
+		// Calculate the final rank, higher available capacity and lower queue used capacity should give a higher rank
+		rank := normalizedAvailableRank + normalizedQueueRank
+
+		// Ensure the rank is within the desired range
+		rank = math.Max(rank, float64(orchestrator.RankPossible))
+		rank = math.Min(rank, maxAvailableCapacityRank+maxQueueCapacityRank)
+
+		// Assign rank and reason to the node
+		ranks[i] = orchestrator.NodeRank{
+			NodeInfo: node,
+			Rank:     int(rank),
+			Reason: fmt.Sprintf(
+				"Ranked based on available capacity %s and queue capacity %s",
+				node.ComputeNodeInfo.AvailableCapacity.String(), node.ComputeNodeInfo.QueueUsedCapacity.String()),
+			Retryable: true,
+		}
+		log.Ctx(ctx).Trace().Object("Rank", ranks[i]).Msg("Ranked node")
+	}
+
+	return ranks, nil
+}
+
+// RankNodes ranks nodes based on their available capacity and queue used capacity.
+// Nodes with more available capacity are ranked higher, and nodes with more queue capacity are ranked lower.
+func (s *AvailableCapacityNodeRanker) RankNodes(
+	ctx context.Context, job models.Job, nodes []models.NodeInfo) ([]orchestrator.NodeRank, error) {
+	// Get dynamic weights based on job requirements
+	jobResources, err := job.Task().ResourcesConfig.ToResources()
+	if err != nil {
+		return nil, fmt.Errorf("failed to get job resources: %w", err)
+	}
+	weights := dynamicWeights(jobResources)
+
+	// Calculate weighted capacities for each node and determine the maximum values
+	wAvailableCapacities, wQueueCapacities, maxAvailableCapacity, maxQueueCapacity :=
+		s.calculateWeightedCapacities(nodes, weights)
+
+	// Rank nodes based on normalized weighted capacities
+	return s.rankNodesBasedOnCapacities(
+		ctx, nodes, wAvailableCapacities, wQueueCapacities, maxAvailableCapacity, maxQueueCapacity)
+}


Test patch:
diff --git a/pkg/compute/capacity/usag_tracker_test.go b/pkg/compute/capacity/usag_tracker_test.go
new file mode 100644
index 0000000000..73200ebc63
--- /dev/null
+++ b/pkg/compute/capacity/usag_tracker_test.go
@@ -0,0 +1,41 @@
+//go:build unit || !integration
+
+package capacity
+
+import (
+	"context"
+	"testing"
+
+	"github.com/bacalhau-project/bacalhau/pkg/models"
+	"github.com/stretchr/testify/suite"
+)
+
+type LocalUsageTrackerTestSuite struct {
+	suite.Suite
+	tracker *LocalUsageTracker
+}
+
+func (s *LocalUsageTrackerTestSuite) SetupTest() {
+	s.tracker = NewLocalUsageTracker()
+}
+
+func (s *LocalUsageTrackerTestSuite) TestAddAndRemove() {
+	ctx := context.Background()
+	usage := models.Resources{CPU: 2, Memory: 1024, Disk: 10000, GPU: 1}
+
+	// Test Add
+	s.tracker.Add(ctx, usage)
+
+	usedCapacity := s.tracker.GetUsedCapacity(ctx)
+	s.Require().Equal(usage, usedCapacity)
+
+	// Test Remove
+	s.tracker.Remove(ctx, usage)
+
+	usedCapacity = s.tracker.GetUsedCapacity(ctx)
+	s.Require().True(usedCapacity.IsZero())
+}
+
+func TestLocalUsageTrackerTestSuite(t *testing.T) {
+	suite.Run(t, new(LocalUsageTrackerTestSuite))
+}
diff --git a/pkg/config/configenv/test.go b/pkg/config/configenv/test.go
index 67fde0a196..090fc5b88c 100644
--- a/pkg/config/configenv/test.go
+++ b/pkg/config/configenv/test.go
@@ -105,12 +105,6 @@ var TestingComputeConfig = types.ComputeConfig{
 			Disk:   "",
 			GPU:    "",
 		},
-		QueueResourceLimits: models.ResourcesConfig{
-			CPU:    "",
-			Memory: "",
-			Disk:   "",
-			GPU:    "",
-		},
 	},
 	ExecutionStore: types.JobStoreConfig{
 		Type: types.BoltDB,
@@ -130,7 +124,6 @@ var TestingComputeConfig = types.ComputeConfig{
 		ProbeHTTP:           "",
 		ProbeExec:           "",
 	},
-	Queue: types.QueueConfig{},
 	Logging: types.LoggingConfig{
 		LogRunningExecutionsInterval: types.Duration(10 * time.Second),
 	},
diff --git a/pkg/orchestrator/selection/ranking/available_capacity_test.go b/pkg/orchestrator/selection/ranking/available_capacity_test.go
new file mode 100644
index 0000000000..2d91ce425a
--- /dev/null
+++ b/pkg/orchestrator/selection/ranking/available_capacity_test.go
@@ -0,0 +1,292 @@
+//go:build unit || !integration
+
+package ranking
+
+import (
+	"context"
+	"sort"
+	"testing"
+
+	"github.com/bacalhau-project/bacalhau/pkg/models"
+	"github.com/bacalhau-project/bacalhau/pkg/test/mock"
+	"github.com/stretchr/testify/suite"
+)
+
+type AvailableCapacityNodeRankerSuite struct {
+	suite.Suite
+	ranker *AvailableCapacityNodeRanker
+}
+
+func (suite *AvailableCapacityNodeRankerSuite) SetupTest() {
+	suite.ranker = NewAvailableCapacityNodeRanker()
+}
+
+type nodeScenario struct {
+	nodeID            string
+	availableCapacity models.Resources
+	queueUsedCapacity models.Resources
+}
+
+type rankNodesTestCase struct {
+	name         string
+	nodes        []nodeScenario
+	jobResources models.ResourcesConfig
+	expected     []string
+	equalCheck   bool
+}
+
+func (suite *AvailableCapacityNodeRankerSuite) TestRankNodes() {
+	testCases := []rankNodesTestCase{
+		{
+			name: "Only available capacity",
+			nodes: []nodeScenario{
+				{
+					nodeID:            "node1",
+					availableCapacity: models.Resources{CPU: 4, Memory: 16000, Disk: 100000, GPU: 1},
+				},
+				{
+					nodeID:            "node2",
+					availableCapacity: models.Resources{CPU: 6, Memory: 24000, Disk: 150000, GPU: 2},
+				},
+				{
+					nodeID:            "node3",
+					availableCapacity: models.Resources{CPU: 2, Memory: 8000, Disk: 50000, GPU: 1},
+				},
+			},
+			expected: []string{"node2", "node1", "node3"},
+		},
+		{
+			name: "Only queued capacity",
+			nodes: []nodeScenario{
+				{
+					nodeID:            "node1",
+					queueUsedCapacity: models.Resources{CPU: 2, Memory: 8000, Disk: 40000, GPU: 1},
+				},
+				{
+					nodeID:            "node2",
+					queueUsedCapacity: models.Resources{CPU: 1, Memory: 4000, Disk: 20000, GPU: 0},
+				},
+				{
+					nodeID:            "node3",
+					queueUsedCapacity: models.Resources{CPU: 0, Memory: 0, Disk: 0, GPU: 0},
+				},
+			},
+			expected: []string{"node3", "node2", "node1"},
+		},
+		{
+			name: "Both available and queued capacities",
+			nodes: []nodeScenario{
+				{
+					nodeID:            "node1",
+					availableCapacity: models.Resources{CPU: 4, Memory: 16000, Disk: 100000, GPU: 1},
+					queueUsedCapacity: models.Resources{CPU: 2, Memory: 8000, Disk: 40000, GPU: 1},
+				},
+				{
+					nodeID:            "node2",
+					availableCapacity: models.Resources{CPU: 6, Memory: 24000, Disk: 150000, GPU: 2},
+					queueUsedCapacity: models.Resources{CPU: 1, Memory: 4000, Disk: 20000, GPU: 0},
+				},
+				{
+					nodeID:            "node3",
+					availableCapacity: models.Resources{CPU: 2, Memory: 8000, Disk: 50000, GPU: 1},
+					queueUsedCapacity: models.Resources{CPU: 0, Memory: 0, Disk: 0, GPU: 0},
+				},
+			},
+			expected: []string{"node2", "node1", "node3"},
+		},
+		{
+			name: "Equal capacities",
+			nodes: []nodeScenario{
+				{
+					nodeID:            "node1",
+					availableCapacity: models.Resources{CPU: 4, Memory: 16000, Disk: 100000, GPU: 1},
+					queueUsedCapacity: models.Resources{CPU: 2, Memory: 8000, Disk: 40000, GPU: 1},
+				},
+				{
+					nodeID:            "node2",
+					availableCapacity: models.Resources{CPU: 4, Memory: 16000, Disk: 100000, GPU: 1},
+					queueUsedCapacity: models.Resources{CPU: 2, Memory: 8000, Disk: 40000, GPU: 1},
+				},
+				{
+					nodeID:            "node3",
+					availableCapacity: models.Resources{CPU: 4, Memory: 16000, Disk: 100000, GPU: 1},
+					queueUsedCapacity: models.Resources{CPU: 2, Memory: 8000, Disk: 40000, GPU: 1},
+				},
+			},
+			equalCheck: true,
+		},
+		{
+			name: "Zero capacities",
+			nodes: []nodeScenario{
+				{
+					nodeID: "node1",
+				},
+				{
+					nodeID: "node2",
+				},
+				{
+					nodeID: "node3",
+				},
+			},
+			equalCheck: true,
+		},
+		{
+			name: "One node with zero capacities",
+			nodes: []nodeScenario{
+				{
+					nodeID:            "node1",
+					availableCapacity: models.Resources{CPU: 4, Memory: 16000, Disk: 100000, GPU: 1},
+					queueUsedCapacity: models.Resources{CPU: 1, Memory: 4000, Disk: 20000, GPU: 0},
+				},
+				{
+					nodeID:            "node2",
+					availableCapacity: models.Resources{CPU: 0, Memory: 0, Disk: 0, GPU: 0},
+					queueUsedCapacity: models.Resources{CPU: 0, Memory: 0, Disk: 0, GPU: 0},
+				},
+				{
+					nodeID:            "node3",
+					availableCapacity: models.Resources{CPU: 2, Memory: 8000, Disk: 50000, GPU: 1},
+					queueUsedCapacity: models.Resources{CPU: 0, Memory: 0, Disk: 0, GPU: 0},
+				},
+			},
+			expected: []string{"node1", "node3", "node2"},
+		},
+		{
+			name: "High CPU weight",
+			nodes: []nodeScenario{
+				{
+					nodeID:            "node1",
+					availableCapacity: models.Resources{CPU: 8, Memory: 16000, Disk: 100000, GPU: 1},
+					queueUsedCapacity: models.Resources{CPU: 2, Memory: 4000, Disk: 50000, GPU: 0},
+				},
+				{
+					nodeID:            "node2",
+					availableCapacity: models.Resources{CPU: 4, Memory: 24000, Disk: 150000, GPU: 2},
+					queueUsedCapacity: models.Resources{CPU: 1, Memory: 2000, Disk: 30000, GPU: 1},
+				},
+				{
+					nodeID:            "node3",
+					availableCapacity: models.Resources{CPU: 2, Memory: 32000, Disk: 200000, GPU: 4},
+					queueUsedCapacity: models.Resources{CPU: 0, Memory: 1000, Disk: 10000, GPU: 2},
+				},
+			},
+			jobResources: models.ResourcesConfig{
+				CPU: "1",
+			},
+			expected: []string{"node1", "node2", "node3"},
+		},
+		{
+			name: "High Memory weight",
+			nodes: []nodeScenario{
+				{
+					nodeID:            "node1",
+					availableCapacity: models.Resources{CPU: 4, Memory: 32000, Disk: 100000, GPU: 1},
+					queueUsedCapacity: models.Resources{CPU: 1, Memory: 8000, Disk: 50000, GPU: 0},
+				},
+				{
+					nodeID:            "node2",
+					availableCapacity: models.Resources{CPU: 8, Memory: 16000, Disk: 150000, GPU: 2},
+					queueUsedCapacity: models.Resources{CPU: 2, Memory: 4000, Disk: 30000, GPU: 1},
+				},
+				{
+					nodeID:            "node3",
+					availableCapacity: models.Resources{CPU: 2, Memory: 8000, Disk: 200000, GPU: 4},
+					queueUsedCapacity: models.Resources{CPU: 0, Memory: 2000, Disk: 10000, GPU: 2},
+				},
+			},
+			jobResources: models.ResourcesConfig{
+				Memory: "10mb",
+			},
+			expected: []string{"node1", "node2", "node3"},
+		},
+		{
+			name: "High Disk weight",
+			nodes: []nodeScenario{
+				{
+					nodeID:            "node1",
+					availableCapacity: models.Resources{CPU: 4, Memory: 16000, Disk: 200000, GPU: 1},
+					queueUsedCapacity: models.Resources{CPU: 1, Memory: 4000, Disk: 100000, GPU: 0},
+				},
+				{
+					nodeID:            "node2",
+					availableCapacity: models.Resources{CPU: 8, Memory: 32000, Disk: 100000, GPU: 2},
+					queueUsedCapacity: models.Resources{CPU: 2, Memory: 8000, Disk: 50000, GPU: 1},
+				},
+				{
+					nodeID:            "node3",
+					availableCapacity: models.Resources{CPU: 2, Memory: 16000, Disk: 150000, GPU: 4},
+					queueUsedCapacity: models.Resources{CPU: 0, Memory: 4000, Disk: 20000, GPU: 2},
+				},
+			},
+			jobResources: models.ResourcesConfig{
+				Disk: "10mb",
+			},
+			expected: []string{"node1", "node3", "node2"},
+		},
+		{
+			name: "High GPU weight",
+			nodes: []nodeScenario{
+				{
+					nodeID:            "node1",
+					availableCapacity: models.Resources{CPU: 2, Memory: 8000, Disk: 100000, GPU: 4},
+					queueUsedCapacity: models.Resources{CPU: 1, Memory: 4000, Disk: 50000, GPU: 1},
+				},
+				{
+					nodeID:            "node2",
+					availableCapacity: models.Resources{CPU: 8, Memory: 32000, Disk: 150000, GPU: 2},
+					queueUsedCapacity: models.Resources{CPU: 2, Memory: 8000, Disk: 30000, GPU: 0},
+				},
+				{
+					nodeID:            "node3",
+					availableCapacity: models.Resources{CPU: 4, Memory: 16000, Disk: 200000, GPU: 1},
+					queueUsedCapacity: models.Resources{CPU: 0, Memory: 2000, Disk: 10000, GPU: 0},
+				},
+			},
+			jobResources: models.ResourcesConfig{
+				GPU: "1",
+			},
+			expected: []string{"node1", "node2", "node3"},
+		},
+	}
+
+	for _, tc := range testCases {
+		suite.Run(tc.name, func() {
+			nodes := make([]models.NodeInfo, len(tc.nodes))
+			for i, ns := range tc.nodes {
+				nodes[i] = models.NodeInfo{
+					NodeID: ns.nodeID,
+					ComputeNodeInfo: &models.ComputeNodeInfo{
+						AvailableCapacity: ns.availableCapacity,
+						QueueUsedCapacity: ns.queueUsedCapacity,
+					},
+				}
+			}
+
+			job := mock.Job()
+			job.Task().ResourcesConfig = &tc.jobResources
+			ranks, err := suite.ranker.RankNodes(context.Background(), *job, nodes)
+			suite.NoError(err)
+
+			// sort nodes by rank
+			sort.Slice(ranks, func(i, j int) bool {
+				return ranks[i].Rank > ranks[j].Rank
+			})
+
+			rankedNodeIDs := make([]string, len(ranks))
+			for i, r := range ranks {
+				rankedNodeIDs[i] = r.NodeInfo.ID()
+			}
+			if tc.equalCheck {
+				for i := 1; i < len(ranks); i++ {
+					suite.Equal(ranks[0].Rank, ranks[i].Rank, "Ranks are not equal")
+				}
+			} else {
+				suite.Equal(tc.expected, rankedNodeIDs)
+			}
+		})
+	}
+}
+
+func TestAvailableCapacityNodeRankerSuite(t *testing.T) {
+	suite.Run(t, new(AvailableCapacityNodeRankerSuite))
+}
diff --git a/pkg/test/compute/resourcelimits_test.go b/pkg/test/compute/resourcelimits_test.go
index 407f4b6097..80958ac924 100644
--- a/pkg/test/compute/resourcelimits_test.go
+++ b/pkg/test/compute/resourcelimits_test.go
@@ -10,6 +10,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/bacalhau-project/bacalhau/pkg/config/types"
 	"github.com/c2h5oh/datasize"
 	"github.com/davecgh/go-spew/spew"
 	"github.com/stretchr/testify/require"
@@ -299,6 +300,9 @@ func (suite *ComputeNodeResourceLimitsSuite) TestParallelGPU() {
 			GPUs:   make([]models.GPU, 1),
 		},
 		IgnorePhysicalResourceLimits: true, // we need to pretend that we have GPUs on each node
+		ControlPlaneSettings: types.ComputeControlPlaneConfig{
+			ResourceUpdateFrequency: types.Duration(50 * time.Millisecond),
+		},
 	})
 	suite.Require().NoError(err)
 	stack := teststack.Setup(ctx,
@@ -334,9 +338,10 @@ func (suite *ComputeNodeResourceLimitsSuite) TestParallelGPU() {
 			jobIds = append(jobIds, submittedJob.Metadata.ID)
 
 			// sleep a bit here to simulate jobs being sumbmitted over time
+			// and to give time for compute nodes to accept and run the jobs
 			// this needs to be less than the time the job lasts
 			// so we are running jobs in parallel
-			time.Sleep((10 + time.Duration(rand.Intn(10))) * time.Millisecond)
+			time.Sleep(200 * time.Millisecond)
 		}
 	}