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

Issue:
Atomic enqueue of evaluations
In many places we update the state of jobs or executions, then create an evaluation, and then enqueue the evaluation to the broker. All of these operations are not atomic and not in a single transaction. This can result in a lot of edge cases and potentially dangling or stale jobs and executions due to partial update of their state, but failure to trigger their re-evaluation.

We need to look into ways to do these operations atomically in a single transaction. This is might be easier to update the job state and create an evaluation in a single transaction as they are part of the same datastore, but the evaluation broker has its own in-memory state, and we need to figure out a way to implement state watchers where we don't need to explicitly enqueue evaluations, but allow the broker to consume state events and enqueue an evaluation whenever one is created


Discussion / hints:

Fix patch:
diff --git a/pkg/jobstore/boltdb/store.go b/pkg/jobstore/boltdb/store.go
index 796218f477..481bc7bd99 100644
--- a/pkg/jobstore/boltdb/store.go
+++ b/pkg/jobstore/boltdb/store.go
@@ -3,11 +3,9 @@ package boltjobstore
 import (
 	"bytes"
 	"context"
-	"encoding/json"
 	"fmt"
 	"sort"
 	"strings"
-	"sync"
 	"time"
 
 	"github.com/benbjohnson/clock"
@@ -42,11 +40,10 @@ const (
 var SpecKey = []byte("spec")
 
 type BoltJobStore struct {
-	database    *bolt.DB
-	clock       clock.Clock
-	marshaller  marshaller.Marshaller
-	watchers    []*jobstore.Watcher
-	watcherLock sync.Mutex
+	database        *bolt.DB
+	clock           clock.Clock
+	marshaller      marshaller.Marshaller
+	watchersManager *jobstore.WatchersManager
 
 	inProgressIndex  *Index
 	namespacesIndex  *Index
@@ -91,10 +88,10 @@ func NewBoltJobStore(dbPath string, options ...Option) (*BoltJobStore, error) {
 	}
 
 	store := &BoltJobStore{
-		database:   db,
-		clock:      clock.New(),
-		marshaller: marshaller.NewJSONMarshaller(),
-		watchers:   make([]*jobstore.Watcher, 0), //nolint:gomnd
+		database:        db,
+		clock:           clock.New(),
+		marshaller:      marshaller.NewJSONMarshaller(),
+		watchersManager: jobstore.NewWatchersManager(),
 	}
 
 	for _, opt := range options {
@@ -147,26 +144,13 @@ func (b *BoltJobStore) BeginTx(ctx context.Context) (jobstore.TxContext, error)
 
 func (b *BoltJobStore) Watch(ctx context.Context,
 	types jobstore.StoreWatcherType,
-	events jobstore.StoreEventType) chan jobstore.WatchEvent {
-	w := jobstore.NewWatcher(types, events)
-
-	b.watcherLock.Lock() // keep the watchers lock as narrow as possible
-	b.watchers = append(b.watchers, w)
-	b.watcherLock.Unlock()
-
-	return w.Channel()
+	events jobstore.StoreEventType,
+	options ...jobstore.WatcherOption) *jobstore.Watcher {
+	return b.watchersManager.NewWatcher(ctx, types, events, options...)
 }
 
-func (b *BoltJobStore) triggerEvent(t jobstore.StoreWatcherType, e jobstore.StoreEventType, object interface{}) {
-	data, _ := json.Marshal(object)
-
-	for _, w := range b.watchers {
-		if !w.IsWatchingEvent(e) || !w.IsWatchingType(t) {
-			return
-		}
-
-		_ = w.WriteEvent(t, e, data, false) // Do not block
-	}
+func (b *BoltJobStore) triggerEvent(t jobstore.StoreWatcherType, e jobstore.StoreEventType, object any) {
+	b.watchersManager.Write(t, e, object)
 }
 
 // GetJob retrieves the Job identified by the id string. If the job isn't found it will
@@ -1274,9 +1258,7 @@ func (b *BoltJobStore) deleteEvaluation(tx *bolt.Tx, id string) error {
 }
 
 func (b *BoltJobStore) Close(ctx context.Context) error {
-	for _, w := range b.watchers {
-		w.Close()
-	}
+	b.watchersManager.Close()
 
 	log.Ctx(ctx).Debug().Msg("closing bolt-backed job store")
 	return b.database.Close()
diff --git a/pkg/jobstore/mocks.go b/pkg/jobstore/mocks.go
index e787b872ea..3570d1b842 100644
--- a/pkg/jobstore/mocks.go
+++ b/pkg/jobstore/mocks.go
@@ -362,15 +362,20 @@ func (mr *MockStoreMockRecorder) UpdateJobState(ctx, request interface{}) *gomoc
 }
 
 // Watch mocks base method.
-func (m *MockStore) Watch(ctx context.Context, types StoreWatcherType, events StoreEventType) chan WatchEvent {
+func (m *MockStore) Watch(ctx context.Context, types StoreWatcherType, events StoreEventType, options ...WatcherOption) *Watcher {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "Watch", ctx, types, events)
-	ret0, _ := ret[0].(chan WatchEvent)
+	varargs := []interface{}{ctx, types, events}
+	for _, a := range options {
+		varargs = append(varargs, a)
+	}
+	ret := m.ctrl.Call(m, "Watch", varargs...)
+	ret0, _ := ret[0].(*Watcher)
 	return ret0
 }
 
 // Watch indicates an expected call of Watch.
-func (mr *MockStoreMockRecorder) Watch(ctx, types, events interface{}) *gomock.Call {
+func (mr *MockStoreMockRecorder) Watch(ctx, types, events interface{}, options ...interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Watch", reflect.TypeOf((*MockStore)(nil).Watch), ctx, types, events)
+	varargs := append([]interface{}{ctx, types, events}, options...)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Watch", reflect.TypeOf((*MockStore)(nil).Watch), varargs...)
 }
diff --git a/pkg/jobstore/types.go b/pkg/jobstore/types.go
index 76afab62cb..2e62dce45b 100644
--- a/pkg/jobstore/types.go
+++ b/pkg/jobstore/types.go
@@ -55,7 +55,7 @@ type Store interface {
 	// will contain a timestamp, but also the StoreWatcherType and
 	// StoreEventType that triggered the event. A json encoded `[]byte`
 	// of the related object will also be included in the [WatchEvent].
-	Watch(ctx context.Context, types StoreWatcherType, events StoreEventType) chan WatchEvent
+	Watch(ctx context.Context, types StoreWatcherType, events StoreEventType, options ...WatcherOption) *Watcher
 
 	// GetJob returns a job, identified by the id parameter, or an error if
 	// it does not exist.
diff --git a/pkg/jobstore/watcher.go b/pkg/jobstore/watcher.go
index 2a8da6ad56..b9a966ccf2 100644
--- a/pkg/jobstore/watcher.go
+++ b/pkg/jobstore/watcher.go
@@ -1,5 +1,17 @@
 package jobstore
 
+import (
+	"context"
+	"sync"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/rs/zerolog/log"
+)
+
+const defaultWatchChannelSize = 64
+const maxBlockingTimeToLog = 20 * time.Millisecond
+
 type StoreWatcherType int
 
 const (
@@ -40,45 +52,82 @@ func (s StoreEventType) String() string {
 	return "Unknown event"
 }
 
-const DefaultWatchChannelSize = 64
-
 // WatchEvent is the message passed through the watcher whenever a
 // specific event occurs on a specific type, as requested when creating
 // the watcher.
 type WatchEvent struct {
 	Kind      StoreWatcherType
 	Event     StoreEventType
-	Object    []byte
+	Object    any
 	Timestamp int64
 }
 
-func NewWatchEvent(kind StoreWatcherType, event StoreEventType, object []byte) WatchEvent {
-	return WatchEvent{
+func NewWatchEvent(kind StoreWatcherType, event StoreEventType, object any) *WatchEvent {
+	return &WatchEvent{
 		Kind:   kind,
 		Event:  event,
-		Object: append([]byte(nil), object...),
+		Object: object,
 		// TODO(ross): Add a timestamp from the actual time of the event (rather than the
 		// time at which this event was created).
 	}
 }
 
+type WatcherOption func(*Watcher)
+
+func WithChannelSize(size int) WatcherOption {
+	return func(w *Watcher) {
+		w.channel = make(chan *WatchEvent, size)
+	}
+}
+
+type FullChannelBehavior int
+
+const (
+	WatcherDrop FullChannelBehavior = iota
+	WatcherDropOldest
+	WatcherBlock
+)
+
+func WithFullChannelBehavior(behavior FullChannelBehavior) WatcherOption {
+	return func(w *Watcher) {
+		w.fullChannelBehavior = behavior
+	}
+}
+
 // Watcher is used by the jobstore to keep a record of parties interested in events happening
 // in the jobstore.  This allows for watching of job and execution types (or both), and for
 // create, update and delete events (or any combination).
 type Watcher struct {
-	types       StoreWatcherType // a bitmask of types being watched
-	events      StoreEventType   // a bitmask of events being watched
-	channelSize int
-	channel     chan WatchEvent
-}
-
-func NewWatcher(types StoreWatcherType, events StoreEventType) *Watcher {
-	return &Watcher{
-		types:       types,
-		events:      events,
-		channelSize: DefaultWatchChannelSize,
-		channel:     make(chan WatchEvent, DefaultWatchChannelSize),
+	types               StoreWatcherType // a bitmask of types being watched
+	events              StoreEventType   // a bitmask of events being watched
+	fullChannelBehavior FullChannelBehavior
+	channel             chan *WatchEvent
+	closed              bool
+	mu                  sync.Mutex
+}
+
+func NewWatcher(ctx context.Context, types StoreWatcherType, events StoreEventType, options ...WatcherOption) *Watcher {
+	w := &Watcher{
+		types:               types,
+		events:              events,
+		fullChannelBehavior: WatcherDrop,
+	}
+
+	for _, option := range options {
+		option(w)
+	}
+
+	if w.channel == nil {
+		w.channel = make(chan *WatchEvent, defaultWatchChannelSize)
 	}
+
+	// Close the worker when the context is done
+	go func() {
+		<-ctx.Done()
+		w.Close()
+	}()
+
+	return w
 }
 
 func (w *Watcher) IsWatchingType(kind StoreWatcherType) bool {
@@ -89,25 +138,105 @@ func (w *Watcher) IsWatchingEvent(event StoreEventType) bool {
 	return w.events&event > 0
 }
 
-func (w *Watcher) Channel() chan WatchEvent {
+func (w *Watcher) Channel() chan *WatchEvent {
 	return w.channel
 }
 
-func (w *Watcher) WriteEvent(kind StoreWatcherType, event StoreEventType, object []byte, allowBlock bool) bool {
-	// If we don't want to block, return a fail when the channel is currently full.
-	// By default, we'll block and wait for a space in the channel
-	if len(w.channel) == w.channelSize && !allowBlock {
-		return false
+func (w *Watcher) write(kind StoreWatcherType, event StoreEventType, object any) bool {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	startTime := time.Now()
+
+	newEvent := NewWatchEvent(kind, event, object)
+
+	switch w.fullChannelBehavior {
+	case WatcherBlock:
+		w.channel <- newEvent
+		if time.Since(startTime) > maxBlockingTimeToLog {
+			log.Debug().Msgf("Watcher blocked for %v writing event %s:%s", time.Since(startTime), kind, event)
+		}
+		return true
+	case WatcherDrop:
+		select {
+		case w.channel <- newEvent:
+		default:
+			// Channel is full, drop the new event
+			log.Debug().Msgf("Watcher queue is full, dropping new event %s:%s", kind, event)
+			return false
+		}
+	case WatcherDropOldest:
+		select {
+		case w.channel <- newEvent:
+		default:
+			// Channel is full, drop the oldest event and try again
+			<-w.channel
+			w.channel <- newEvent
+			log.Debug().Msgf("Watcher queue is full, dropping oldest event and adding new event %s:%s", kind, event)
+			return true
+		}
 	}
+	return false
+}
 
-	w.channel <- WatchEvent{
-		Kind:   kind,
-		Event:  event,
-		Object: object,
+func (w *Watcher) Close() {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	if !w.closed {
+		close(w.channel)
+		w.closed = true
 	}
-	return true
 }
 
-func (w *Watcher) Close() {
-	close(w.channel)
+// WatchersManager is a helper type that can be used by the different
+// jobstore implementations to manage watchers. It allows for the creation
+// of new watchers, the writing of events to all interested watchers and
+// the cleanup of closed watchers.
+type WatchersManager struct {
+	watchers    map[string]*Watcher
+	watcherLock sync.Mutex
+}
+
+func NewWatchersManager() *WatchersManager {
+	return &WatchersManager{
+		watchers: make(map[string]*Watcher),
+	}
+}
+
+// NewWatcher creates a new Watcher managed by the WatchersManager
+func (w *WatchersManager) NewWatcher(
+	ctx context.Context, types StoreWatcherType, events StoreEventType, options ...WatcherOption) *Watcher {
+	watcher := NewWatcher(ctx, types, events, options...)
+	watcherID := uuid.NewString()
+
+	w.watcherLock.Lock()
+	defer w.watcherLock.Unlock()
+	w.watchers[watcherID] = watcher
+	return watcher
+}
+
+// Write writes an event to all interested watchers
+// If a watcher is closed, it is removed from the list of watchers
+func (w *WatchersManager) Write(kind StoreWatcherType, event StoreEventType, object any) {
+	closedWatchers := make([]string, 0)
+	for id, watcher := range w.watchers {
+		if watcher.closed {
+			closedWatchers = append(closedWatchers, id)
+			continue
+		}
+		if watcher.IsWatchingType(kind) && watcher.IsWatchingEvent(event) {
+			watcher.write(kind, event, object)
+		}
+	}
+	for _, id := range closedWatchers {
+		delete(w.watchers, id)
+	}
+}
+
+// Close closes all watchers
+func (w *WatchersManager) Close() {
+	w.watcherLock.Lock()
+	defer w.watcherLock.Unlock()
+	for _, watcher := range w.watchers {
+		watcher.Close()
+	}
 }
diff --git a/pkg/node/requester.go b/pkg/node/requester.go
index a15f726a1d..5c6c7cdaf9 100644
--- a/pkg/node/requester.go
+++ b/pkg/node/requester.go
@@ -2,6 +2,7 @@ package node
 
 import (
 	"context"
+	"fmt"
 
 	"github.com/rs/zerolog/log"
 
@@ -112,6 +113,13 @@ func NewRequesterNode(
 	}
 	evalBroker.SetEnabled(true)
 
+	// evaluations watcher
+	evaluationsWatcher := evaluation.NewWatcher(jobStore, evalBroker)
+	if err = evaluationsWatcher.Backfill(ctx); err != nil {
+		return nil, fmt.Errorf("failed to backfill evaluations: %w", err)
+	}
+	evaluationsWatcher.Start(ctx)
+
 	// planners that execute the proposed plan by the scheduler
 	// order of the planners is important as they are executed in order
 	planners := planner.NewChain(
@@ -213,7 +221,6 @@ func NewRequesterNode(
 
 	endpoint := requester.NewBaseEndpoint(&requester.BaseEndpointParams{
 		ID:                         nodeID,
-		EvaluationBroker:           evalBroker,
 		EventEmitter:               eventEmitter,
 		ComputeEndpoint:            computeProxy,
 		Store:                      jobStore,
@@ -246,7 +253,6 @@ func NewRequesterNode(
 
 	endpointV2 := orchestrator.NewBaseEndpoint(&orchestrator.BaseEndpointParams{
 		ID:                nodeID,
-		EvaluationBroker:  evalBroker,
 		Store:             jobStore,
 		EventEmitter:      eventEmitter,
 		ComputeProxy:      computeProxy,
@@ -256,10 +262,9 @@ func NewRequesterNode(
 	})
 
 	housekeeping, err := orchestrator.NewHousekeeping(orchestrator.HousekeepingParams{
-		EvaluationBroker: evalBroker,
-		JobStore:         jobStore,
-		Interval:         requesterConfig.HousekeepingBackgroundTaskInterval,
-		TimeoutBuffer:    requesterConfig.HousekeepingTimeoutBuffer,
+		JobStore:      jobStore,
+		Interval:      requesterConfig.HousekeepingBackgroundTaskInterval,
+		TimeoutBuffer: requesterConfig.HousekeepingTimeoutBuffer,
 	})
 	if err != nil {
 		return nil, err
diff --git a/pkg/orchestrator/endpoint.go b/pkg/orchestrator/endpoint.go
index 1aff9e5385..345292b116 100644
--- a/pkg/orchestrator/endpoint.go
+++ b/pkg/orchestrator/endpoint.go
@@ -22,7 +22,6 @@ import (
 
 type BaseEndpointParams struct {
 	ID                string
-	EvaluationBroker  EvaluationBroker
 	Store             jobstore.Store
 	EventEmitter      EventEmitter
 	ComputeProxy      compute.Endpoint
@@ -33,7 +32,6 @@ type BaseEndpointParams struct {
 
 type BaseEndpoint struct {
 	id                string
-	evaluationBroker  EvaluationBroker
 	store             jobstore.Store
 	eventEmitter      EventEmitter
 	computeProxy      compute.Endpoint
@@ -45,7 +43,6 @@ type BaseEndpoint struct {
 func NewBaseEndpoint(params *BaseEndpointParams) *BaseEndpoint {
 	return &BaseEndpoint{
 		id:                params.ID,
-		evaluationBroker:  params.EvaluationBroker,
 		store:             params.Store,
 		eventEmitter:      params.EventEmitter,
 		computeProxy:      params.ComputeProxy,
@@ -125,9 +122,6 @@ func (e *BaseEndpoint) SubmitJob(ctx context.Context, request *SubmitJobRequest)
 		return nil, err
 	}
 
-	if err := e.evaluationBroker.Enqueue(eval); err != nil {
-		return nil, err
-	}
 	e.eventEmitter.EmitJobCreated(ctx, *job)
 	return &SubmitJobResponse{
 		JobID:        job.ID,
@@ -186,11 +180,6 @@ func (e *BaseEndpoint) StopJob(ctx context.Context, request *StopJobRequest) (St
 			log.Ctx(ctx).Error().Err(err).Msgf("failed to save evaluation for stop job %s", request.JobID)
 			return StopJobResponse{}, err
 		}
-
-		err = e.evaluationBroker.Enqueue(eval)
-		if err != nil {
-			return StopJobResponse{}, err
-		}
 		evalID = eval.ID
 	}
 	e.eventEmitter.EmitEventSilently(ctx, model.JobEvent{
diff --git a/pkg/orchestrator/evaluation/watcher.go b/pkg/orchestrator/evaluation/watcher.go
new file mode 100644
index 0000000000..ebe32b482e
--- /dev/null
+++ b/pkg/orchestrator/evaluation/watcher.go
@@ -0,0 +1,94 @@
+package evaluation
+
+import (
+	"context"
+	"sync"
+
+	"github.com/bacalhau-project/bacalhau/pkg/jobstore"
+	"github.com/bacalhau-project/bacalhau/pkg/models"
+	"github.com/bacalhau-project/bacalhau/pkg/orchestrator"
+	"github.com/rs/zerolog/log"
+)
+
+// Watcher is the integration between the jobstore and the evaluation broker.
+// It watches for changes in the jobstore and enqueues evaluations in the broker
+// It also handles populating the evaluation broker with non-terminal evaluations during startup
+type Watcher struct {
+	// store is the jobstore to watch
+	store jobstore.Store
+	// broker is the evaluation broker to enqueue evaluations
+	broker orchestrator.EvaluationBroker
+
+	watching  bool
+	startOnce sync.Once
+	stopOnce  sync.Once
+	stopChan  chan struct{}
+}
+
+// NewWatcher creates a new Watcher
+func NewWatcher(store jobstore.Store, broker orchestrator.EvaluationBroker) *Watcher {
+	return &Watcher{
+		store:    store,
+		broker:   broker,
+		stopChan: make(chan struct{}),
+	}
+}
+
+// IsWatching returns true if the watcher is currently watching
+func (w *Watcher) IsWatching() bool {
+	return w.watching
+}
+
+// Start starts the watcher in a goroutine
+func (w *Watcher) Start(ctx context.Context) {
+	w.startOnce.Do(func() {
+		go w.watchAndEnqueue(ctx)
+	})
+}
+
+// Backfill populates the broker with non-terminal evaluations
+func (w *Watcher) Backfill(ctx context.Context) error {
+	// TODO: Implement Backfill
+	return nil
+}
+
+func (w *Watcher) watchAndEnqueue(ctx context.Context) {
+	watcher := w.store.Watch(ctx, jobstore.EvaluationWatcher, jobstore.CreateEvent)
+	defer watcher.Close()
+
+	w.watching = true
+	defer func() {
+		w.watching = false
+	}()
+
+	for {
+		select {
+		case evt := <-watcher.Channel():
+			if evt == nil {
+				log.Ctx(ctx).Debug().Msg("Watcher channel closed, stopping eval watcher")
+				return
+			}
+			eval, ok := evt.Object.(models.Evaluation)
+			if !ok {
+				log.Error().Msgf("Received unexpected object type in eval watcher: %T", evt.Object)
+				continue
+			}
+			if err := w.broker.Enqueue(&eval); err != nil {
+				log.Error().Err(err).Msgf("Failed to enqueue %s", eval.String())
+			}
+		case <-ctx.Done():
+			log.Ctx(ctx).Debug().Msg("Context cancelled, stopping eval watcher")
+			return
+		case <-w.stopChan:
+			log.Ctx(ctx).Debug().Msg("Stop channel closed, stopping eval watcher")
+			return
+		}
+	}
+}
+
+// Stop stops the watcher
+func (w *Watcher) Stop() {
+	w.stopOnce.Do(func() {
+		close(w.stopChan)
+	})
+}
diff --git a/pkg/orchestrator/housekeeping.go b/pkg/orchestrator/housekeeping.go
index 07126a6c23..e7535413d1 100644
--- a/pkg/orchestrator/housekeeping.go
+++ b/pkg/orchestrator/housekeeping.go
@@ -20,8 +20,7 @@ const (
 )
 
 type HousekeepingParams struct {
-	EvaluationBroker EvaluationBroker
-	JobStore         jobstore.Store
+	JobStore jobstore.Store
 	// Interval is the interval at which housekeeping tasks are run
 	Interval time.Duration
 	// Workers is the maximum number of parallel workers for housekeeping tasks
@@ -36,10 +35,9 @@ type HousekeepingParams struct {
 }
 
 type Housekeeping struct {
-	evaluationBroker EvaluationBroker
-	jobStore         jobstore.Store
-	interval         time.Duration
-	timeoutBuffer    time.Duration
+	jobStore      jobstore.Store
+	interval      time.Duration
+	timeoutBuffer time.Duration
 
 	workersSem chan struct{}
 	waitGroup  sync.WaitGroup
@@ -61,7 +59,6 @@ func NewHousekeeping(params HousekeepingParams) (*Housekeeping, error) {
 
 	// validate params
 	err := errors.Join(
-		validate.IsNotNil(params.EvaluationBroker, "evaluation broker cannot be nil"),
 		validate.IsNotNil(params.JobStore, "job store cannot be nil"),
 		validate.IsGreaterThanZero(params.Interval, "interval must be greater than zero"),
 		validate.IsGreaterThanZero(params.Workers, "workers must be greater than zero"),
@@ -72,13 +69,12 @@ func NewHousekeeping(params HousekeepingParams) (*Housekeeping, error) {
 	}
 
 	h := &Housekeeping{
-		evaluationBroker: params.EvaluationBroker,
-		jobStore:         params.JobStore,
-		interval:         params.Interval,
-		timeoutBuffer:    params.TimeoutBuffer,
-		workersSem:       make(chan struct{}, params.Workers),
-		stopChan:         make(chan struct{}),
-		clock:            params.Clock,
+		jobStore:      params.JobStore,
+		interval:      params.Interval,
+		timeoutBuffer: params.TimeoutBuffer,
+		workersSem:    make(chan struct{}, params.Workers),
+		stopChan:      make(chan struct{}),
+		clock:         params.Clock,
 	}
 
 	return h, nil
@@ -231,11 +227,6 @@ func (h *Housekeeping) handleTimeoutExecutions(ctx context.Context, execution *m
 		return fmt.Errorf("failed to create evaluation %+v: %w", eval, err)
 	}
 
-	err = h.evaluationBroker.Enqueue(eval)
-	if err != nil {
-		return fmt.Errorf("failed to enqueue evaluation %+v: %w", eval, err)
-	}
-
 	log.Ctx(ctx).Debug().Msgf("enqueued evaluation for timed-out execution %+v", eval)
 	return nil
 }
diff --git a/pkg/requester/endpoint.go b/pkg/requester/endpoint.go
index 961ff5f92e..b3cbbc9d26 100644
--- a/pkg/requester/endpoint.go
+++ b/pkg/requester/endpoint.go
@@ -23,7 +23,6 @@ import (
 
 type BaseEndpointParams struct {
 	ID                         string
-	EvaluationBroker           orchestrator.EvaluationBroker
 	Store                      jobstore.Store
 	EventEmitter               orchestrator.EventEmitter
 	ComputeEndpoint            compute.Endpoint
@@ -35,13 +34,12 @@ type BaseEndpointParams struct {
 
 // BaseEndpoint base implementation of requester Endpoint
 type BaseEndpoint struct {
-	id               string
-	evaluationBroker orchestrator.EvaluationBroker
-	store            jobstore.Store
-	eventEmitter     orchestrator.EventEmitter
-	computesvc       compute.Endpoint
-	transforms       []jobtransform.Transformer
-	postTransforms   []jobtransform.PostTransformer
+	id             string
+	store          jobstore.Store
+	eventEmitter   orchestrator.EventEmitter
+	computesvc     compute.Endpoint
+	transforms     []jobtransform.Transformer
+	postTransforms []jobtransform.PostTransformer
 }
 
 func NewBaseEndpoint(params *BaseEndpointParams) *BaseEndpoint {
@@ -59,13 +57,12 @@ func NewBaseEndpoint(params *BaseEndpointParams) *BaseEndpoint {
 	}
 
 	return &BaseEndpoint{
-		id:               params.ID,
-		evaluationBroker: params.EvaluationBroker,
-		computesvc:       params.ComputeEndpoint,
-		store:            params.Store,
-		transforms:       transforms,
-		postTransforms:   postTransforms,
-		eventEmitter:     params.EventEmitter,
+		id:             params.ID,
+		computesvc:     params.ComputeEndpoint,
+		store:          params.Store,
+		transforms:     transforms,
+		postTransforms: postTransforms,
+		eventEmitter:   params.EventEmitter,
 	}
 }
 
@@ -152,11 +149,6 @@ func (e *BaseEndpoint) SubmitJob(ctx context.Context, data model.JobCreatePayloa
 		return nil, err
 	}
 
-	err = e.evaluationBroker.Enqueue(eval)
-	if err != nil {
-		return nil, err
-	}
-
 	e.eventEmitter.EmitJobCreated(ctx, *job)
 	return legacyJob, nil
 }
@@ -210,11 +202,6 @@ func (e *BaseEndpoint) CancelJob(ctx context.Context, request CancelJobRequest)
 			log.Ctx(ctx).Error().Err(err).Msgf("failed to save evaluation for cancel job %s", request.JobID)
 			return CancelJobResult{}, err
 		}
-
-		err = e.evaluationBroker.Enqueue(eval)
-		if err != nil {
-			return CancelJobResult{}, err
-		}
 	}
 	e.eventEmitter.EmitEventSilently(ctx, model.JobEvent{
 		JobID:     request.JobID,
@@ -375,11 +362,6 @@ func (e *BaseEndpoint) enqueueEvaluation(ctx context.Context, jobID, operation s
 		log.Ctx(ctx).Error().Err(err).Msgf("[%s] failed to create/save evaluation for job %s", operation, jobID)
 		return
 	}
-
-	err = e.evaluationBroker.Enqueue(eval)
-	if err != nil {
-		log.Ctx(ctx).Error().Err(err).Msgf("[%s] failed to enqueue evaluation for job %s", operation, jobID)
-	}
 }
 
 // Compile-time interface check:


Test patch:
diff --git a/pkg/jobstore/boltdb/store_test.go b/pkg/jobstore/boltdb/store_test.go
index 81ff9fac3f..4d7f412d61 100644
--- a/pkg/jobstore/boltdb/store_test.go
+++ b/pkg/jobstore/boltdb/store_test.go
@@ -4,7 +4,6 @@ package boltjobstore
 
 import (
 	"context"
-	"encoding/json"
 	"os"
 	"path/filepath"
 	"testing"
@@ -579,7 +578,7 @@ func (s *BoltJobstoreTestSuite) TestShortIDs() {
 }
 
 func (s *BoltJobstoreTestSuite) TestEvents() {
-	ch := s.store.Watch(s.ctx,
+	watcher := s.store.Watch(s.ctx,
 		jobstore.JobWatcher|jobstore.ExecutionWatcher,
 		jobstore.CreateEvent|jobstore.UpdateEvent|jobstore.DeleteEvent,
 	)
@@ -596,14 +595,13 @@ func (s *BoltJobstoreTestSuite) TestEvents() {
 		s.Require().NoError(err)
 
 		// Read an event, it should be a jobcreate
-		ev := <-ch
+		ev := <-watcher.Channel()
 		s.Require().Equal(ev.Event, jobstore.CreateEvent)
 		s.Require().Equal(ev.Kind, jobstore.JobWatcher)
 
-		var decodedJob models.Job
-		err = json.Unmarshal(ev.Object, &decodedJob)
-		s.Require().NoError(err)
-		s.Require().Equal(decodedJob.ID, job.ID)
+		expectedJob, ok := ev.Object.(models.Job)
+		s.Require().True(ok, "expected object to be a job")
+		s.Require().Equal(expectedJob.ID, job.ID)
 	})
 
 	s.Run("execution create event", func() {
@@ -615,7 +613,7 @@ func (s *BoltJobstoreTestSuite) TestEvents() {
 		s.Require().NoError(err)
 
 		// Read an event, it should be a ExecutionForJob Create
-		ev := <-ch
+		ev := <-watcher.Channel()
 		s.Require().Equal(ev.Event, jobstore.CreateEvent)
 		s.Require().Equal(ev.Kind, jobstore.ExecutionWatcher)
 	})
@@ -630,7 +628,7 @@ func (s *BoltJobstoreTestSuite) TestEvents() {
 			Event: models.Event{Message: "event test"},
 		}
 		_ = s.store.UpdateJobState(s.ctx, request)
-		ev := <-ch
+		ev := <-watcher.Channel()
 		s.Require().Equal(ev.Event, jobstore.UpdateEvent)
 		s.Require().Equal(ev.Kind, jobstore.JobWatcher)
 	})
@@ -646,19 +644,18 @@ func (s *BoltJobstoreTestSuite) TestEvents() {
 			NewValues: execution,
 			Event:     models.Event{Message: "event test"},
 		})
-		ev := <-ch
+		ev := <-watcher.Channel()
 		s.Require().Equal(ev.Event, jobstore.UpdateEvent)
 		s.Require().Equal(ev.Kind, jobstore.ExecutionWatcher)
 
-		var decodedExecution models.Execution
-		err := json.Unmarshal(ev.Object, &decodedExecution)
-		s.Require().NoError(err)
-		s.Require().Equal(decodedExecution.ID, execution.ID)
+		expectedExec, ok := ev.Object.(models.Execution)
+		s.Require().True(ok, "expected object to be an execution")
+		s.Require().Equal(expectedExec.ID, execution.ID)
 	})
 
 	s.Run("delete job event", func() {
 		_ = s.store.DeleteJob(s.ctx, job.ID)
-		ev := <-ch
+		ev := <-watcher.Channel()
 		s.Require().Equal(ev.Event, jobstore.DeleteEvent)
 		s.Require().Equal(ev.Kind, jobstore.JobWatcher)
 	})
diff --git a/pkg/jobstore/watcher_test.go b/pkg/jobstore/watcher_test.go
index bb07e7c25c..7f542900f3 100644
--- a/pkg/jobstore/watcher_test.go
+++ b/pkg/jobstore/watcher_test.go
@@ -1,12 +1,14 @@
 //go:build unit || !integration
 
-package jobstore_test
+package jobstore
 
 import (
 	"context"
+	"sync"
 	"testing"
+	"time"
 
-	"github.com/bacalhau-project/bacalhau/pkg/jobstore"
+	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
 )
 
@@ -26,8 +28,8 @@ func (s *WatcherTestSuite) SetupTest() {
 func (s *WatcherTestSuite) TestCreateWatcher() {
 	testcases := []struct {
 		name                   string
-		types                  jobstore.StoreWatcherType
-		events                 jobstore.StoreEventType
+		types                  StoreWatcherType
+		events                 StoreEventType
 		expected_watching_job  bool
 		expected_watching_exec bool
 		expected_watching_eval bool
@@ -37,8 +39,8 @@ func (s *WatcherTestSuite) TestCreateWatcher() {
 	}{
 		{
 			name:                   "All the things",
-			types:                  jobstore.JobWatcher | jobstore.ExecutionWatcher | jobstore.EvaluationWatcher,
-			events:                 jobstore.CreateEvent | jobstore.UpdateEvent | jobstore.DeleteEvent,
+			types:                  JobWatcher | ExecutionWatcher | EvaluationWatcher,
+			events:                 CreateEvent | UpdateEvent | DeleteEvent,
 			expected_watching_job:  true,
 			expected_watching_exec: true,
 			expected_watching_eval: true,
@@ -48,8 +50,8 @@ func (s *WatcherTestSuite) TestCreateWatcher() {
 		},
 		{
 			name:                   "Job creation events",
-			types:                  jobstore.JobWatcher,
-			events:                 jobstore.CreateEvent,
+			types:                  JobWatcher,
+			events:                 CreateEvent,
 			expected_watching_eval: false,
 			expected_watching_job:  true,
 			expected_watching_exec: false,
@@ -59,8 +61,8 @@ func (s *WatcherTestSuite) TestCreateWatcher() {
 		},
 		{
 			name:                   "Job and Execution creation events",
-			types:                  jobstore.JobWatcher | jobstore.ExecutionWatcher,
-			events:                 jobstore.CreateEvent,
+			types:                  JobWatcher | ExecutionWatcher,
+			events:                 CreateEvent,
 			expected_watching_job:  true,
 			expected_watching_exec: true,
 			expected_watching_eval: false,
@@ -70,8 +72,8 @@ func (s *WatcherTestSuite) TestCreateWatcher() {
 		},
 		{
 			name:                   "Execution deletion events",
-			types:                  jobstore.ExecutionWatcher,
-			events:                 jobstore.DeleteEvent,
+			types:                  ExecutionWatcher,
+			events:                 DeleteEvent,
 			expected_watching_job:  false,
 			expected_watching_exec: true,
 			expected_watching_eval: false,
@@ -81,8 +83,8 @@ func (s *WatcherTestSuite) TestCreateWatcher() {
 		},
 		{
 			name:                   "Updates for job and executions",
-			types:                  jobstore.ExecutionWatcher | jobstore.JobWatcher,
-			events:                 jobstore.UpdateEvent,
+			types:                  ExecutionWatcher | JobWatcher,
+			events:                 UpdateEvent,
 			expected_watching_job:  true,
 			expected_watching_exec: true,
 			expected_watching_eval: false,
@@ -94,18 +96,18 @@ func (s *WatcherTestSuite) TestCreateWatcher() {
 
 	for _, tc := range testcases {
 		s.Run(tc.name, func() {
-			w := jobstore.NewWatcher(tc.types, tc.events)
+			w := NewWatcher(context.Background(), tc.types, tc.events)
 
-			watchingJob := w.IsWatchingType(jobstore.JobWatcher)
-			watchingExec := w.IsWatchingType(jobstore.ExecutionWatcher)
-			watchingEval := w.IsWatchingType(jobstore.EvaluationWatcher)
+			watchingJob := w.IsWatchingType(JobWatcher)
+			watchingExec := w.IsWatchingType(ExecutionWatcher)
+			watchingEval := w.IsWatchingType(EvaluationWatcher)
 			s.Equal(tc.expected_watching_job, watchingJob, "expectation around watching job not met")
 			s.Equal(tc.expected_watching_exec, watchingExec, "expectation around watching exec not met")
 			s.Equal(tc.expected_watching_eval, watchingEval, "expectation around watching evaluation not met")
 
-			watchingCreate := w.IsWatchingEvent(jobstore.CreateEvent)
-			watchingUpdate := w.IsWatchingEvent(jobstore.UpdateEvent)
-			watchingDelete := w.IsWatchingEvent(jobstore.DeleteEvent)
+			watchingCreate := w.IsWatchingEvent(CreateEvent)
+			watchingUpdate := w.IsWatchingEvent(UpdateEvent)
+			watchingDelete := w.IsWatchingEvent(DeleteEvent)
 			s.Equal(tc.expected_create, watchingCreate, "expectation around watching create events not met ")
 			s.Equal(tc.expected_update, watchingUpdate, "expectation around watching update events not met ")
 			s.Equal(tc.expected_delete, watchingDelete, "expectation around watching delete  events not met ")
@@ -117,39 +119,39 @@ func (s *WatcherTestSuite) TestCreateWatcher() {
 func (s *WatcherTestSuite) TestGetEvents() {
 	testcases := []struct {
 		name                string
-		types               jobstore.StoreWatcherType
-		events              jobstore.StoreEventType
-		write_kind          jobstore.StoreWatcherType
-		write_event         jobstore.StoreEventType
+		types               StoreWatcherType
+		events              StoreEventType
+		write_kind          StoreWatcherType
+		write_event         StoreEventType
 		expected_watchevent bool
-		expected_object     []byte
+		expected_object     any
 	}{
 		{
 			name:                "all the things",
-			types:               jobstore.JobWatcher | jobstore.ExecutionWatcher | jobstore.EvaluationWatcher,
-			events:              jobstore.CreateEvent | jobstore.UpdateEvent | jobstore.DeleteEvent,
-			write_kind:          jobstore.JobWatcher,
-			write_event:         jobstore.CreateEvent,
+			types:               JobWatcher | ExecutionWatcher | EvaluationWatcher,
+			events:              CreateEvent | UpdateEvent | DeleteEvent,
+			write_kind:          JobWatcher,
+			write_event:         CreateEvent,
 			expected_watchevent: true,
-			expected_object:     []byte("all the things"),
+			expected_object:     "all the things",
 		},
 		{
 			name:                "create an evaluation",
-			types:               jobstore.JobWatcher | jobstore.ExecutionWatcher | jobstore.EvaluationWatcher,
-			events:              jobstore.CreateEvent | jobstore.UpdateEvent | jobstore.DeleteEvent,
-			write_kind:          jobstore.EvaluationWatcher,
-			write_event:         jobstore.CreateEvent,
+			types:               JobWatcher | ExecutionWatcher | EvaluationWatcher,
+			events:              CreateEvent | UpdateEvent | DeleteEvent,
+			write_kind:          EvaluationWatcher,
+			write_event:         CreateEvent,
 			expected_watchevent: true,
-			expected_object:     []byte("create an evaluation"),
+			expected_object:     "create an evaluation",
 		},
 	}
 
 	for _, tc := range testcases {
 		s.Run(tc.name, func() {
-			w := jobstore.NewWatcher(tc.types, tc.events)
+			w := NewWatcher(context.Background(), tc.types, tc.events, WithFullChannelBehavior(WatcherBlock))
 			ch := w.Channel()
 
-			resp := w.WriteEvent(tc.write_kind, tc.write_event, []byte(tc.name), true)
+			resp := w.write(tc.write_kind, tc.write_event, tc.name)
 			s.Equal(tc.expected_watchevent, resp)
 
 			msg := <-ch
@@ -161,3 +163,185 @@ func (s *WatcherTestSuite) TestGetEvents() {
 		})
 	}
 }
+
+func (s *WatcherTestSuite) TestWriteEventWithFullChannel() {
+	testcases := []struct {
+		name                string
+		fullChannelBehavior FullChannelBehavior
+		firstEvent          any
+		secondEvent         any
+		expectedEvent       any
+		expectedSecondResp  bool
+		shouldBlock         bool
+	}{
+		{
+			name:                "Block",
+			fullChannelBehavior: WatcherBlock,
+			firstEvent:          "first",
+			secondEvent:         "second",
+			expectedEvent:       "first", // The first event is not dropped
+			expectedSecondResp:  true,
+			shouldBlock:         true,
+		},
+		{
+			name:                "Drop",
+			fullChannelBehavior: WatcherDrop,
+			firstEvent:          "first",
+			secondEvent:         "second",
+			expectedEvent:       "first", // The second event is dropped
+			expectedSecondResp:  false,
+		},
+		{
+			name:                "DropOldest",
+			fullChannelBehavior: WatcherDropOldest,
+			firstEvent:          "first",
+			secondEvent:         "second",
+			expectedEvent:       "second", // The first event is dropped
+			expectedSecondResp:  true,
+		},
+	}
+
+	for _, tc := range testcases {
+		s.Run(tc.name, func() {
+			w := NewWatcher(
+				context.Background(),
+				JobWatcher,
+				CreateEvent,
+				WithChannelSize(1),
+				WithFullChannelBehavior(tc.fullChannelBehavior))
+
+			wg := sync.WaitGroup{}
+			wg.Add(1)
+
+			go func() {
+				w.write(JobWatcher, CreateEvent, tc.firstEvent)
+
+				doneSecondEvent := make(chan bool)
+				go func() {
+					resp := w.write(JobWatcher, CreateEvent, tc.secondEvent)
+					doneSecondEvent <- true
+					s.Equal(tc.expectedSecondResp, resp, "resp was not as expected")
+				}()
+
+				timer := time.NewTimer(50 * time.Millisecond)
+				select {
+				case <-doneSecondEvent:
+					timer.Stop()
+					if tc.fullChannelBehavior == WatcherBlock {
+						s.Failf("WriteEvent did not block when it should have", "fullChannelBehavior: %v", tc.fullChannelBehavior)
+					}
+				case <-timer.C:
+					if tc.fullChannelBehavior != WatcherBlock {
+						s.Failf("WriteEvent blocked when it should not have", "fullChannelBehavior: %v", tc.fullChannelBehavior)
+					}
+				}
+				wg.Done()
+			}()
+
+			wg.Wait()
+			msg := <-w.Channel()
+			s.Equal(tc.expectedEvent, msg.Object, "msg was not as expected")
+		})
+	}
+}
+
+func (s *WatcherTestSuite) TestCloseWhenContextDone() {
+	// Create a context that we can cancel
+	ctx, cancel := context.WithCancel(s.ctx)
+
+	// Create a watcher with this context
+	w := NewWatcher(ctx, JobWatcher, CreateEvent)
+
+	// Ensure the watcher is not closed to start with
+	s.Require().False(w.closed, "Watcher should not be closed at start")
+
+	// Cancel the context
+	cancel()
+
+	// Now the watcher should be closed
+	s.Eventually(func() bool {
+		return w.closed
+	}, 100*time.Millisecond, 10*time.Millisecond, "Watcher should be closed after context is cancelled")
+}
+
+type WatchersManagerTestSuite struct {
+	suite.Suite
+	ctx context.Context
+}
+
+func TestWatchersManagerTestSuite(t *testing.T) {
+	suite.Run(t, new(WatchersManagerTestSuite))
+}
+
+func (s *WatchersManagerTestSuite) SetupTest() {
+	s.ctx = context.Background()
+}
+
+func (s *WatchersManagerTestSuite) TestNewWatcher() {
+	manager := NewWatchersManager()
+	watcher := manager.NewWatcher(s.ctx, JobWatcher, CreateEvent)
+
+	s.NotNil(watcher, "Watcher should not be nil")
+	s.True(watcher.IsWatchingType(JobWatcher), "Watcher should be watching JobWatcher type")
+	s.True(watcher.IsWatchingEvent(CreateEvent), "Watcher should be watching CreateEvent")
+}
+
+func (s *WatchersManagerTestSuite) TestWrite() {
+	manager := NewWatchersManager()
+	watcher := manager.NewWatcher(s.ctx, JobWatcher, CreateEvent)
+
+	manager.Write(JobWatcher, CreateEvent, "Test object")
+	verifyWatcher(s.T(), watcher, JobWatcher, CreateEvent, "Test object")
+}
+
+func (s *WatchersManagerTestSuite) TestClose() {
+	manager := NewWatchersManager()
+	watcher := manager.NewWatcher(s.ctx, JobWatcher, CreateEvent)
+
+	manager.Close()
+	s.True(watcher.closed, "Watcher should be closed")
+}
+
+func (s *WatchersManagerTestSuite) TestWriteToMultipleWatchers() {
+	manager := NewWatchersManager()
+
+	// Create three watchers
+	watcher1 := manager.NewWatcher(s.ctx, JobWatcher, CreateEvent)
+	watcher2 := manager.NewWatcher(s.ctx, JobWatcher, CreateEvent)
+	watcher3 := manager.NewWatcher(s.ctx, ExecutionWatcher, UpdateEvent)
+	watcher4 := manager.NewWatcher(s.ctx, EvaluationWatcher, DeleteEvent)
+
+	// Write an event to JobWatcher with CreateEvent
+	manager.Write(JobWatcher, CreateEvent, "Test object 1")
+
+	// Write an event to ExecutionWatcher with UpdateEvent
+	manager.Write(ExecutionWatcher, UpdateEvent, "Test object 2")
+
+	// Verify watcher1 and watcher2 received "Test object 1" and nothing else
+	verifyWatcher(s.T(), watcher1, JobWatcher, CreateEvent, "Test object 1")
+	verifyWatcher(s.T(), watcher2, JobWatcher, CreateEvent, "Test object 1")
+
+	// Verify watcher3 received "Test object 2" and nothing else
+	verifyWatcher(s.T(), watcher3, ExecutionWatcher, UpdateEvent, "Test object 2")
+
+	// Verify watcher4 received nothing
+	verifyWatcher(s.T(), watcher4, EvaluationWatcher, DeleteEvent, "")
+}
+
+// verifyWatcher is a helper function to verify a watcher received a specific event
+func verifyWatcher(t *testing.T, watcher *Watcher, kind StoreWatcherType, event StoreEventType, expectedObject any) {
+	select {
+	case receivedEvent := <-watcher.Channel():
+		if expectedObject == "" {
+			require.Fail(t, "Watcher received an event when it should not have")
+		} else {
+			require.Equal(t, kind, receivedEvent.Kind, "Event kind should match")
+			require.Equal(t, event, receivedEvent.Event, "Event type should match")
+			require.Equal(t, expectedObject, receivedEvent.Object, "Event object should match")
+		}
+	case <-time.After(50 * time.Millisecond):
+		if expectedObject != "" {
+			require.Fail(t, "Timeout waiting for event")
+		}
+	}
+}
diff --git a/pkg/orchestrator/endpoint_test.go b/pkg/orchestrator/endpoint_test.go
index a4ec8f351a..3fa21e2bb6 100644
--- a/pkg/orchestrator/endpoint_test.go
+++ b/pkg/orchestrator/endpoint_test.go
@@ -71,9 +71,6 @@ func (s *EndpointSuite) TestInlinePinnerTransformInSubmit() {
 	})
 	s.Require().NoError(err)
 
-	evalBroker := orchestrator.NewMockEvaluationBroker(ctrl)
-	evalBroker.EXPECT().Enqueue(gomock.Any()).Return(nil)
-
 	tmpFile := filepath.Join(s.T().TempDir(), "test.db")
 	jobstore, err := boltjobstore.NewBoltJobStore(tmpFile)
 	defer func() {
@@ -83,10 +80,9 @@ func (s *EndpointSuite) TestInlinePinnerTransformInSubmit() {
 	s.Require().NoError(err)
 
 	endpoint := orchestrator.NewBaseEndpoint(&orchestrator.BaseEndpointParams{
-		ID:               "test_endpoint",
-		EvaluationBroker: evalBroker,
-		Store:            jobstore,
-		EventEmitter:     eventEmitter,
+		ID:           "test_endpoint",
+		Store:        jobstore,
+		EventEmitter: eventEmitter,
 		JobTransformer: transformer.ChainedTransformer[*models.Job]{
 			transformer.JobFn(transformer.IDGenerator),
 			transformer.NewInlineStoragePinner(storageProviders),
diff --git a/pkg/orchestrator/evaluation/watcher_test.go b/pkg/orchestrator/evaluation/watcher_test.go
new file mode 100644
index 0000000000..653310b658
--- /dev/null
+++ b/pkg/orchestrator/evaluation/watcher_test.go
@@ -0,0 +1,130 @@
+//go:build unit || !integration
+
+package evaluation_test
+
+import (
+	"context"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/suite"
+
+	"github.com/bacalhau-project/bacalhau/pkg/jobstore"
+	boltjobstore "github.com/bacalhau-project/bacalhau/pkg/jobstore/boltdb"
+	"github.com/bacalhau-project/bacalhau/pkg/models"
+	"github.com/bacalhau-project/bacalhau/pkg/orchestrator/evaluation"
+	"github.com/bacalhau-project/bacalhau/pkg/test/mock"
+)
+
+type WatcherTestSuite struct {
+	suite.Suite
+	store   jobstore.Store
+	broker  *evaluation.InMemoryBroker
+	watcher *evaluation.Watcher
+}
+
+func (s *WatcherTestSuite) SetupTest() {
+	// Open the Bolt database on the temporary file.
+	var err error
+	s.store, err = boltjobstore.NewBoltJobStore(filepath.Join(s.T().TempDir(), "evaluation-watcher.db"))
+	s.Require().NoError(err)
+
+	s.broker, err = evaluation.NewInMemoryBroker(evaluation.InMemoryBrokerParams{})
+	s.Require().NoError(err)
+	s.broker.SetEnabled(true)
+
+	s.watcher = evaluation.NewWatcher(s.store, s.broker)
+}
+
+func (s *WatcherTestSuite) TearDownTest() {
+	_ = s.store.Close(context.Background())
+}
+
+func (s *WatcherTestSuite) TestEnqueueEval() {
+	// start the watcher
+	s.watcher.Start(context.Background())
+	s.Eventually(func() bool {
+		return s.watcher.IsWatching()
+	}, 100*time.Millisecond, 10*time.Millisecond)
+
+	eval, err := s.createEvaluation()
+
+	dequeuedEval, _, err := s.broker.Dequeue([]string{eval.Type}, 100*time.Millisecond)
+	s.Require().NoError(err)
+	s.Require().NotNil(dequeuedEval, "evaluation should be enqueued")
+	s.Require().Equal(eval.ID, dequeuedEval.ID)
+
+	// no more evaluations
+	dequeuedEval, _, err = s.broker.Dequeue([]string{eval.Type}, 10*time.Millisecond)
+	s.Require().NoError(err)
+	s.Require().Nil(dequeuedEval, "evaluation should not be enqueued")
+}
+
+// TestStoppedWatcher tests that the watcher stops watching when the context is canceled
+func (s *WatcherTestSuite) TestStoppedWatcher_ContextDone() {
+	// start the watcher
+	ctx, cancel := context.WithCancel(context.Background())
+	s.watcher.Start(ctx)
+	s.Eventually(func() bool {
+		return s.watcher.IsWatching()
+	}, 100*time.Millisecond, 10*time.Millisecond)
+
+	cancel()
+	s.Eventually(func() bool {
+		return !s.watcher.IsWatching()
+	}, 100*time.Millisecond, 10*time.Millisecond)
+
+	// Create an evaluation
+	eval, err := s.createEvaluation()
+	s.Require().NoError(err)
+
+	// no evaluations should be enqueued
+	dequeuedEval, _, err := s.broker.Dequeue([]string{eval.Type}, 10*time.Millisecond)
+	s.Require().NoError(err)
+	s.Require().Nil(dequeuedEval, "evaluation should not be enqueued")
+}
+
+// TestStoppedWatcher_StopCalled tests that the watcher stops watching when Stop is called
+func (s *WatcherTestSuite) TestStoppedWatcher_StopCalled() {
+	// start the watcher
+	s.watcher.Start(context.Background())
+	s.Eventually(func() bool {
+		return s.watcher.IsWatching()
+	}, 100*time.Millisecond, 10*time.Millisecond)
+
+	s.watcher.Stop()
+	s.Eventually(func() bool {
+		return !s.watcher.IsWatching()
+	}, 100*time.Millisecond, 10*time.Millisecond)
+
+	// Create an evaluation
+	eval, err := s.createEvaluation()
+	s.Require().NoError(err)
+
+	// no evaluations should be enqueued
+	dequeuedEval, _, err := s.broker.Dequeue([]string{eval.Type}, 10*time.Millisecond)
+	s.Require().NoError(err)
+	s.Require().Nil(dequeuedEval, "evaluation should not be enqueued")
+}
+
+func (s *WatcherTestSuite) createEvaluation() (*models.Evaluation, error) {
+	// Create a job
+	job := mock.Job()
+	eval := mock.Eval()
+	eval.JobID = job.ID
+	eval.Type = job.Type
+
+	// Create job
+	err := s.store.CreateJob(context.Background(), *job, models.Event{})
+	s.Require().NoError(err)
+
+	// Create an evaluation
+	err = s.store.CreateEvaluation(context.Background(), *eval)
+	s.Require().NoError(err)
+	return eval, err
+}
+
+func TestWatcherTestSuite(t *testing.T) {
+	suite.Run(t, new(WatcherTestSuite))
+}
diff --git a/pkg/orchestrator/housekeeping_test.go b/pkg/orchestrator/housekeeping_test.go
index 63ff87ef1a..67094d2054 100644
--- a/pkg/orchestrator/housekeeping_test.go
+++ b/pkg/orchestrator/housekeeping_test.go
@@ -25,26 +25,23 @@ const (
 
 type HousekeepingTestSuite struct {
 	suite.Suite
-	ctrl                 *gomock.Controller
-	clock                *clock.Mock
-	mockJobStore         *jobstore.MockStore
-	mockEvaluationBroker *MockEvaluationBroker
-	housekeeping         *Housekeeping
+	ctrl         *gomock.Controller
+	clock        *clock.Mock
+	mockJobStore *jobstore.MockStore
+	housekeeping *Housekeeping
 }
 
 func (s *HousekeepingTestSuite) SetupTest() {
 	s.ctrl = gomock.NewController(s.T())
 	s.clock = clock.NewMock()
 	s.mockJobStore = jobstore.NewMockStore(s.ctrl)
-	s.mockEvaluationBroker = NewMockEvaluationBroker(s.ctrl)
 
 	h, _ := NewHousekeeping(HousekeepingParams{
-		EvaluationBroker: s.mockEvaluationBroker,
-		JobStore:         s.mockJobStore,
-		Interval:         200 * time.Millisecond,
-		Workers:          1,
-		TimeoutBuffer:    timeoutBuffer,
-		Clock:            s.clock,
+		JobStore:      s.mockJobStore,
+		Interval:      200 * time.Millisecond,
+		Workers:       1,
+		TimeoutBuffer: timeoutBuffer,
+		Clock:         s.clock,
 	})
 
 	// we only want to freeze time to have more deterministic tests.
@@ -353,10 +350,6 @@ func (s *HousekeepingTestSuite) assertEvaluationEnqueued(job models.Job) {
 	s.mockJobStore.EXPECT().CreateEvaluation(gomock.Any(), gomock.Any()).Do(func(ctx context.Context, eval models.Evaluation) {
 		matchEvaluation(&eval)
 	}).Return(nil)
-
-	s.mockEvaluationBroker.EXPECT().Enqueue(gomock.Any()).Do(func(eval *models.Evaluation) {
-		matchEvaluation(eval)
-	}).Return(nil)
 }
 
 // mockJob creates a mock job for testing. It takes list of ModifyTime for executions