add evaluation watcher and enable atomic enqueues (#3998)

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()

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)
 	})

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.